mirror of
https://github.com/hanzo-ds/rs.git
synced 2026-10-30 20:49:56 +00:00
zeekay
86788566d1
Cut from the upstream v0.13.3 release tag (not HEAD) so this is a pure
rename with no behaviour change. v0.13.3 is what our consumers already
resolve to: insights/rust's Cargo.lock pins clickhouse 0.13.3 and its
manifest already declares `datastore` at 0.13.x.
The load-bearing change is the HTTP auth header rename. The Datastore
server reads only X-Datastore-User / X-Datastore-Key; the upstream names
return `401 Code: 194 Authentication failed`. Every un-renamed HTTP
client is therefore unable to authenticate at all — this is a
correctness fix, not branding.
- crate `clickhouse` -> `datastore`, `clickhouse-derive` -> `datastore-derive`;
versions preserved exactly (0.13.3 / 0.2.0).
- derive macro emits `datastore::Row` in lockstep, so downstream
`#[derive(Row)]` keeps resolving.
- User-Agent reports `datastore-rs/{version}`.
- test-harness env vars CLICKHOUSE_* -> DATASTORE_*; compose uses
ghcr.io/hanzoai/datastore with /etc/datastore-server config paths.
- doc links point at docs.hanzo.ai/datastore.
- `publish = false`: the `datastore` name on crates.io belongs to an
unrelated crate, so consume this by git tag, never by version range.
CHANGELOG is truncated to a fork notice rather than find-replaced —
rewriting upstream release history would manufacture changes Datastore
never shipped. References to upstream issues keep their numbers.
138 lines
3.4 KiB
Rust
138 lines
3.4 KiB
Rust
use serde::Deserialize;
|
|
|
|
use datastore::{error::Error, Client, Compression, Row};
|
|
|
|
#[tokio::test]
|
|
async fn deferred() {
|
|
let client = prepare_database!();
|
|
max_execution_time(client, false).await;
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn wait_end_of_query() {
|
|
let client = prepare_database!();
|
|
max_execution_time(client, true).await;
|
|
}
|
|
|
|
async fn max_execution_time(mut client: Client, wait_end_of_query: bool) {
|
|
if wait_end_of_query {
|
|
client = client.with_option("wait_end_of_query", "1")
|
|
}
|
|
|
|
// TODO: check different `timeout_overflow_mode`
|
|
let mut cursor = client
|
|
.with_compression(Compression::None)
|
|
.with_option("max_execution_time", "0.1")
|
|
.query("SELECT toUInt8(65 + number % 5) FROM system.numbers LIMIT 100000000")
|
|
.fetch::<u8>()
|
|
.unwrap();
|
|
|
|
let mut i = 0u64;
|
|
|
|
let err = loop {
|
|
match cursor.next().await {
|
|
Ok(Some(no)) => {
|
|
// Check that we haven't parsed something extra.
|
|
assert_eq!(no, (65 + i % 5) as u8);
|
|
i += 1;
|
|
}
|
|
Ok(None) => panic!("DB exception hasn't been found"),
|
|
Err(err) => break err,
|
|
}
|
|
};
|
|
|
|
assert!(wait_end_of_query ^ (i != 0));
|
|
assert!(err.to_string().contains("TIMEOUT_EXCEEDED"));
|
|
}
|
|
|
|
#[cfg(feature = "lz4")]
|
|
#[tokio::test]
|
|
async fn deferred_lz4() {
|
|
let client = prepare_database!().with_compression(Compression::Lz4);
|
|
|
|
client
|
|
.query("CREATE TABLE test(no UInt32) ENGINE = MergeTree ORDER BY no")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
#[derive(serde::Serialize, datastore::Row)]
|
|
struct Row {
|
|
no: u32,
|
|
}
|
|
|
|
let part_count = 100;
|
|
let part_size = 100_000;
|
|
|
|
// Due to compression we need more complex test here: write a lot of big parts.
|
|
for i in 0..part_count {
|
|
let mut insert = client.insert("test").unwrap();
|
|
|
|
for j in 0..part_size {
|
|
let row = Row {
|
|
no: i * part_size + j,
|
|
};
|
|
|
|
insert.write(&row).await.unwrap();
|
|
}
|
|
|
|
insert.end().await.unwrap();
|
|
}
|
|
|
|
let mut cursor = client
|
|
.with_option("max_execution_time", "0.1")
|
|
.query("SELECT no FROM test")
|
|
.fetch::<u32>()
|
|
.unwrap();
|
|
|
|
let mut i = 0;
|
|
|
|
let err = loop {
|
|
match cursor.next().await {
|
|
Ok(Some(_)) => i += 1,
|
|
Ok(None) => panic!("DB exception hasn't been found"),
|
|
Err(err) => break err,
|
|
}
|
|
};
|
|
|
|
assert_ne!(i, 0); // we're interested only in errors during processing
|
|
assert!(err.to_string().contains("TIMEOUT_EXCEEDED"));
|
|
}
|
|
|
|
// See #185.
|
|
#[tokio::test]
|
|
async fn invalid_schema() {
|
|
#[derive(Debug, Row, Deserialize)]
|
|
#[allow(dead_code)]
|
|
struct MyRow {
|
|
no: u32,
|
|
dec: Option<String>, // valid schema: u64-based types
|
|
}
|
|
|
|
let client = prepare_database!();
|
|
|
|
client
|
|
.query(
|
|
"CREATE TABLE test(no UInt32, dec Nullable(Decimal64(4)))
|
|
ENGINE = MergeTree
|
|
ORDER BY no",
|
|
)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
client
|
|
.query("INSERT INTO test VALUES (1, 1.1), (2, 2.2), (3, 3.3)")
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
|
|
let err = client
|
|
.query("SELECT ?fields FROM test")
|
|
.fetch_all::<MyRow>()
|
|
.await
|
|
.unwrap_err();
|
|
|
|
assert!(matches!(err, Error::NotEnoughData));
|
|
}
|