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.
125 lines
3.1 KiB
Rust
125 lines
3.1 KiB
Rust
use std::{
|
|
future::Future,
|
|
mem,
|
|
time::{Duration, Instant},
|
|
};
|
|
|
|
use bytes::Bytes;
|
|
use criterion::{black_box, criterion_group, criterion_main, Criterion, Throughput};
|
|
use http_body_util::Empty;
|
|
use hyper::{body::Incoming, Request, Response};
|
|
use serde::Serialize;
|
|
|
|
use datastore::{error::Result, Client, Compression, Row};
|
|
|
|
mod common;
|
|
|
|
async fn serve(request: Request<Incoming>) -> Response<Empty<Bytes>> {
|
|
common::skip_incoming(request).await;
|
|
Response::new(Empty::new())
|
|
}
|
|
|
|
#[derive(Row, Serialize)]
|
|
struct SomeRow {
|
|
a: u64,
|
|
b: i64,
|
|
c: i32,
|
|
d: u32,
|
|
e: u64,
|
|
f: u32,
|
|
g: u64,
|
|
h: i64,
|
|
}
|
|
|
|
impl SomeRow {
|
|
fn sample() -> Self {
|
|
black_box(Self {
|
|
a: 42,
|
|
b: 42,
|
|
c: 42,
|
|
d: 42,
|
|
e: 42,
|
|
f: 42,
|
|
g: 42,
|
|
h: 42,
|
|
})
|
|
}
|
|
}
|
|
|
|
async fn run_insert(client: Client, iters: u64) -> Result<Duration> {
|
|
let start = Instant::now();
|
|
let mut insert = client.insert("table")?;
|
|
|
|
for _ in 0..iters {
|
|
insert.write(&SomeRow::sample()).await?;
|
|
}
|
|
|
|
insert.end().await?;
|
|
Ok(start.elapsed())
|
|
}
|
|
|
|
#[cfg(feature = "inserter")]
|
|
async fn run_inserter<const WITH_PERIOD: bool>(client: Client, iters: u64) -> Result<Duration> {
|
|
let start = Instant::now();
|
|
let mut inserter = client.inserter("table")?.with_max_rows(iters);
|
|
|
|
if WITH_PERIOD {
|
|
// Just to measure overhead, not to actually use it.
|
|
inserter = inserter.with_period(Some(Duration::from_secs(1000)));
|
|
}
|
|
|
|
for _ in 0..iters {
|
|
inserter.write(&SomeRow::sample())?;
|
|
inserter.commit().await?;
|
|
}
|
|
|
|
inserter.end().await?;
|
|
Ok(start.elapsed())
|
|
}
|
|
|
|
fn run<F>(c: &mut Criterion, name: &str, port: u16, f: impl Fn(Client, u64) -> F)
|
|
where
|
|
F: Future<Output = Result<Duration>> + Send + 'static,
|
|
{
|
|
let addr = format!("127.0.0.1:{port}").parse().unwrap();
|
|
let _server = common::start_server(addr, serve);
|
|
let runner = common::start_runner();
|
|
|
|
let mut group = c.benchmark_group(name);
|
|
group.throughput(Throughput::Bytes(mem::size_of::<SomeRow>() as u64));
|
|
group.bench_function("no compression", |b| {
|
|
b.iter_custom(|iters| {
|
|
let client = Client::default()
|
|
.with_url(format!("http://{addr}"))
|
|
.with_compression(Compression::None);
|
|
runner.run((f)(client, iters))
|
|
})
|
|
});
|
|
#[cfg(feature = "lz4")]
|
|
group.bench_function("lz4", |b| {
|
|
b.iter_custom(|iters| {
|
|
let client = Client::default()
|
|
.with_url(format!("http://{addr}"))
|
|
.with_compression(Compression::Lz4);
|
|
runner.run((f)(client, iters))
|
|
})
|
|
});
|
|
group.finish();
|
|
}
|
|
|
|
fn insert(c: &mut Criterion) {
|
|
run(c, "insert", 6543, run_insert);
|
|
}
|
|
|
|
#[cfg(feature = "inserter")]
|
|
fn inserter(c: &mut Criterion) {
|
|
run(c, "inserter", 6544, run_inserter::<false>);
|
|
run(c, "inserter-period", 6545, run_inserter::<true>);
|
|
}
|
|
|
|
#[cfg(not(feature = "inserter"))]
|
|
criterion_group!(benches, insert);
|
|
#[cfg(feature = "inserter")]
|
|
criterion_group!(benches, insert, inserter);
|
|
criterion_main!(benches);
|