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.
105 lines
3.1 KiB
Rust
105 lines
3.1 KiB
Rust
#![cfg(feature = "watch")]
|
|
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
use datastore::{Client, Row};
|
|
|
|
#[derive(Debug, PartialEq, Row, Serialize, Deserialize)]
|
|
struct MyRow {
|
|
num: u32,
|
|
}
|
|
|
|
async fn create_table(client: &Client) {
|
|
client
|
|
.query(
|
|
"
|
|
CREATE TABLE test(num UInt32)
|
|
ENGINE = MergeTree
|
|
ORDER BY num
|
|
",
|
|
)
|
|
.execute()
|
|
.await
|
|
.unwrap();
|
|
}
|
|
|
|
async fn insert_into_table(client: &Client, rows: &[MyRow]) {
|
|
let mut insert = client.insert("test").unwrap();
|
|
for row in rows {
|
|
insert.write(row).await.unwrap();
|
|
}
|
|
insert.end().await.unwrap();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn changes() {
|
|
let client = prepare_database!();
|
|
|
|
create_table(&client).await;
|
|
|
|
let mut cursor1 = client
|
|
.watch("SELECT ?fields FROM test ORDER BY num")
|
|
.limit(1)
|
|
.fetch::<MyRow>()
|
|
.unwrap();
|
|
|
|
let mut cursor2 = client
|
|
.watch("SELECT sum(num) as num FROM test")
|
|
.fetch::<MyRow>()
|
|
.unwrap();
|
|
|
|
// Insert first batch.
|
|
insert_into_table(&client, &[MyRow { num: 1 }, MyRow { num: 2 }]).await;
|
|
assert_eq!(cursor1.next().await.unwrap(), Some((1, MyRow { num: 1 })));
|
|
assert_eq!(cursor1.next().await.unwrap(), Some((1, MyRow { num: 2 })));
|
|
assert_eq!(cursor2.next().await.unwrap(), Some((1, MyRow { num: 3 })));
|
|
|
|
// Insert second batch.
|
|
insert_into_table(&client, &[MyRow { num: 3 }, MyRow { num: 4 }]).await;
|
|
assert_eq!(cursor1.next().await.unwrap(), Some((2, MyRow { num: 1 })));
|
|
assert_eq!(cursor1.next().await.unwrap(), Some((2, MyRow { num: 2 })));
|
|
assert_eq!(cursor1.next().await.unwrap(), Some((2, MyRow { num: 3 })));
|
|
assert_eq!(cursor1.next().await.unwrap(), Some((2, MyRow { num: 4 })));
|
|
assert_eq!(cursor2.next().await.unwrap(), Some((2, MyRow { num: 10 })));
|
|
|
|
// Insert third batch.
|
|
insert_into_table(&client, &[MyRow { num: 5 }, MyRow { num: 6 }]).await;
|
|
assert_eq!(cursor1.next().await.unwrap(), None);
|
|
assert_eq!(cursor2.next().await.unwrap(), Some((3, MyRow { num: 21 })));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn events() {
|
|
let client = prepare_database!();
|
|
|
|
create_table(&client).await;
|
|
|
|
let mut cursor1 = client
|
|
.watch("SELECT num FROM test ORDER BY num")
|
|
.limit(1)
|
|
.only_events()
|
|
.fetch()
|
|
.unwrap();
|
|
|
|
let mut cursor2 = client
|
|
.watch("SELECT sum(num) as num FROM test")
|
|
.only_events()
|
|
.fetch()
|
|
.unwrap();
|
|
|
|
// Insert first batch.
|
|
insert_into_table(&client, &[MyRow { num: 1 }, MyRow { num: 2 }]).await;
|
|
assert_eq!(cursor1.next().await.unwrap(), Some(1));
|
|
assert_eq!(cursor2.next().await.unwrap(), Some(1));
|
|
|
|
// Insert second batch.
|
|
insert_into_table(&client, &[MyRow { num: 3 }, MyRow { num: 4 }]).await;
|
|
assert_eq!(cursor1.next().await.unwrap(), Some(2));
|
|
assert_eq!(cursor2.next().await.unwrap(), Some(2));
|
|
|
|
// Insert third batch.
|
|
insert_into_table(&client, &[MyRow { num: 5 }, MyRow { num: 6 }]).await;
|
|
assert_eq!(cursor1.next().await.unwrap(), None);
|
|
assert_eq!(cursor2.next().await.unwrap(), Some(3));
|
|
}
|