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.
161 lines
3.7 KiB
Rust
161 lines
3.7 KiB
Rust
use serde::{Deserialize, Serialize};
|
|
|
|
use datastore::{error::Result, sql, Client, Row};
|
|
|
|
#[derive(Debug, Row, Serialize, Deserialize)]
|
|
struct MyRow<'a> {
|
|
no: u32,
|
|
name: &'a str,
|
|
}
|
|
|
|
#[derive(Debug, Row, Serialize, Deserialize)]
|
|
struct MyRowOwned {
|
|
no: u32,
|
|
name: String,
|
|
}
|
|
|
|
async fn ddl(client: &Client) -> Result<()> {
|
|
client.query("DROP TABLE IF EXISTS some").execute().await?;
|
|
client
|
|
.query(
|
|
"
|
|
CREATE TABLE some(no UInt32, name LowCardinality(String))
|
|
ENGINE = MergeTree
|
|
ORDER BY no
|
|
",
|
|
)
|
|
.execute()
|
|
.await
|
|
}
|
|
|
|
async fn insert(client: &Client) -> Result<()> {
|
|
let mut insert = client.insert("some")?;
|
|
for i in 0..1000 {
|
|
insert.write(&MyRow { no: i, name: "foo" }).await?;
|
|
}
|
|
|
|
insert.end().await
|
|
}
|
|
|
|
// This is a very basic example of using the `inserter` feature.
|
|
// See `inserter.rs` for real-world patterns.
|
|
#[cfg(feature = "inserter")]
|
|
async fn inserter(client: &Client) -> Result<()> {
|
|
let mut inserter = client
|
|
.inserter("some")?
|
|
.with_max_rows(100_000)
|
|
.with_period(Some(std::time::Duration::from_secs(15)));
|
|
|
|
for i in 0..1000 {
|
|
inserter.write(&MyRow { no: i, name: "foo" })?;
|
|
inserter.commit().await?;
|
|
}
|
|
|
|
inserter.end().await?;
|
|
Ok(())
|
|
}
|
|
|
|
async fn fetch(client: &Client) -> Result<()> {
|
|
let mut cursor = client
|
|
.query("SELECT ?fields FROM some WHERE name = ? AND no BETWEEN ? AND ?")
|
|
.bind("foo")
|
|
.bind(500)
|
|
.bind(504)
|
|
.fetch::<MyRow<'_>>()?;
|
|
|
|
while let Some(row) = cursor.next().await? {
|
|
println!("{row:?}");
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn fetch_all(client: &Client) -> Result<()> {
|
|
let vec = client
|
|
.query("SELECT ?fields FROM ? WHERE no BETWEEN ? AND ?")
|
|
.bind(sql::Identifier("some"))
|
|
.bind(500)
|
|
.bind(504)
|
|
.fetch_all::<MyRowOwned>()
|
|
.await?;
|
|
|
|
println!("{vec:?}");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn delete(client: &Client) -> Result<()> {
|
|
client
|
|
.clone()
|
|
.with_option("mutations_sync", "1")
|
|
.query("ALTER TABLE some DELETE WHERE no >= ?")
|
|
.bind(500)
|
|
.execute()
|
|
.await?;
|
|
|
|
Ok(())
|
|
}
|
|
|
|
async fn select_count(client: &Client) -> Result<()> {
|
|
let count = client
|
|
.query("SELECT count() FROM some")
|
|
.fetch_one::<u64>()
|
|
.await?;
|
|
|
|
println!("count() = {count}");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[cfg(feature = "watch")]
|
|
async fn watch(client: &Client) -> Result<()> {
|
|
let mut cursor = client
|
|
.watch("SELECT max(no) no, argMax(name, some.no) name FROM some")
|
|
.fetch::<MyRow<'_>>()?;
|
|
|
|
let (version, row) = cursor.next().await?.unwrap();
|
|
println!("version={version}, row={row:?}");
|
|
|
|
let mut insert = client.insert("some")?;
|
|
let row = MyRow {
|
|
no: row.no + 1,
|
|
name: "bar",
|
|
};
|
|
insert.write(&row).await?;
|
|
insert.end().await?;
|
|
|
|
let (version, row) = cursor.next().await?.unwrap();
|
|
println!("version={version}, row={row:?}");
|
|
|
|
// Or you can request only events without data.
|
|
let mut cursor = client
|
|
// It's possible to specify a view name.
|
|
.watch("lv_f2ac5347c013c5b9a6c1aab7192dd97c2748daa0")
|
|
.limit(10)
|
|
.only_events()
|
|
.fetch()?;
|
|
|
|
println!("{:?}", cursor.next().await);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
let client = Client::default().with_url("http://localhost:8123");
|
|
|
|
ddl(&client).await?;
|
|
insert(&client).await?;
|
|
#[cfg(feature = "inserter")]
|
|
inserter(&client).await?;
|
|
select_count(&client).await?;
|
|
fetch(&client).await?;
|
|
fetch_all(&client).await?;
|
|
delete(&client).await?;
|
|
select_count(&client).await?;
|
|
#[cfg(feature = "watch")]
|
|
watch(&client).await?;
|
|
|
|
Ok(())
|
|
}
|