ScopeDB SDK for Rust

scopedb-client is an async Rust client for ScopeDB. It supports statements, declarative catalog reads and conditional updates, direct NDJSON table appends, and bounded concurrent streaming writes.

SDK 0.4 requires ScopeDB 0.6.0 or later and uses its native Statement, declarative Catalog, and table ingest APIs. See the migration guide when upgrading from SDK 0.3.

Installation

Requires Rust 1.91.0 or later. From your application crate:

cargo add scopedb-client serde_json
cargo add tokio --features macros,rt-multi-thread

Quickstart

Set SCOPEDB_ENDPOINT and SCOPEDB_API_KEY using the connection guide. Save this program as src/main.rs and run cargo run:

use scopedb_client::Client;

#[tokio::main(flavor = "current_thread")]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::builder(std::env::var("SCOPEDB_ENDPOINT")?)
        .api_key(std::env::var("SCOPEDB_API_KEY")?)
        .build()?;

    let result = client.query("SELECT 1 AS ready").await?;
    println!("{:?}", result.into_objects()?);
    Ok(())
}

Documentation

• SDK guide
• API reference
• Runnable examples are documented in examples/README.md.
• Release history is recorded in the repository's CHANGELOG.md.
• Report an issue

This SDK sends ScopeQL statements to ScopeDB; the language is documented separately. Use these canonical entry points:

• Quickstart
• Query guide
• Language reference

Client configuration

Applications that own TLS, proxy, timeout, or pooling settings can pass a compatible HTTP client through .http_client(...). The SDK-owned HTTP client disables redirects and transport retries to avoid replaying writes; custom clients must use .redirect(reqwest::redirect::Policy::none()).retry(reqwest::retry::never()) for the same guarantee. Client::new(endpoint, http_client) remains available when authentication is already configured on that client. Use the reqwest version re-exported as scopedb_client::reqwest to avoid dependency-version mismatches.

Every table append or Catalog mutation HTTP request has a client timeout of 30 seconds by default. Its budget starts after local encoding, compression, and request preparation and covers one HTTP attempt through reading the complete response body. Configure it with Client::builder(endpoint).write_timeout(Duration::from_secs(60)) before build(). Client::new uses the same default. Streams inherit the client timeout unless an append stream explicitly sets attempt_timeout. Both settings require a finite, positive duration; the Rust SDK has no option to disable this deadline. Increase the duration when a write needs more time. Shorter timeouts on a custom HTTP client still apply. Admission, retry backoff, and an entire flush()/shutdown() barrier are outside this budget.

Append client timeouts after an HTTP attempt begins preserve the existing error kind and source and add Error::write_details() with reason: WriteErrorReason::Timeout and outcome: AppendState::Unknown; is_retryable() is false. The structured details, source, and HTTP diagnostics remain available on subsequent errors returned by the stream and batch failure events. A timeout does not roll back a write: reconcile a client-interrupted write before replaying it. A server HTTP 408 is still a server response; an explicitly rejected append follows the existing safe retry policy. Streams retry temporary rejected batches by default. Unknown outcomes require explicit .retry_unknown(true); this permits at-least-once delivery and can duplicate rows. Catalog interruptions use catalog_commit_outcome_unknown and are never replayed automatically.

use scopedb_client::{AppendState, WriteErrorReason};
# fn inspect(error: &scopedb_client::Error) {
if let Some(details) = error.write_details() {
    if details.reason == WriteErrorReason::Timeout {
        assert_eq!(details.outcome, AppendState::Unknown);
        assert!(!error.is_retryable());
        eprintln!("write may have committed; reconcile before replaying");
    }
}
# }

Statement JSON request bodies and all table append requests use zstd compression by default. The default HTTP client also negotiates compressed responses. Append limits are based on the uncompressed NDJSON body.

The runnable examples read authentication from SCOPEDB_API_KEY. For backward compatibility, they fall back to SCOPEDB_TOKEN when the API key variable is unset or empty. The builder marks the resulting authorization header as sensitive so standard header and request Debug formatting redacts the credential.

Run a statement

# async fn demo() -> Result<(), scopedb_client::Error> {
# let client = scopedb_client::Client::new("http://127.0.0.1:6543", scopedb_client::reqwest::Client::new())?;
let result = client.query("SELECT 1 AS ready").await?;
let rows = result.into_objects()?;
println!("{rows:?}");
# Ok(())
# }

raw_rows() exposes the string-or-null wire cells without parsing. to_values() borrows the result while into_values() consumes it and moves owned string cells. to_objects() and into_objects() key values by output column name; they return an error when names are duplicated, so use the value form for intentionally duplicated columns. first() converts only the first row for point lookups and aggregates.

Typed values support PartialEq, so applications can compare cells and result rows directly. Array, object, any, and binary values retain their encoded strings and compare those strings; floating-point values follow Rust's usual equality rules.

For lifecycle control, call client.statement(scopeql).submit() and retain the returned handle. snapshot() reads the latest cached status without a network request, refresh().await fetches one current status, wait() polls to a terminal result, and cancel() requests cancellation. The statement ID and initial snapshot are available immediately after submission. cancel() returns the observed status: a pending or running response still needs wait(); completion retains results and failure retains the original error. Cancellation is a distinct ErrorKind::StatementCancelled when waiting. Terminal snapshots are cached. Dropping a query, refresh, or wait future stops local waiting and does not cancel server execution; keep a handle and call cancel() explicitly.

If query() or execute() fails after submission, Error::statement_id() identifies the existing execution. Resume it with client.statement_handle(id).wait(); the error does not advertise retrying the entire query as safe. Temporary errors from a handle's refresh() or wait() can still be retried on that handle. A lost submission response, HTTP 408, or server/gateway 5xx leaves submission uncertain and is not classified as retryable. Use with_statement_id(id) before submission when you need to look up that execution after a lost response; a known ID alone does not confirm acceptance.

Browse the catalog

Catalog reads return complete declarative definitions and an opaque ETag. Select the smallest subtree you need. Apply replaces that entire subtree, including its children; omitted fields are cleared. Preview with dry_run, apply with if_match to reject concurrent changes, and opt into allow_destructive only after reviewing data loss. Mutations never retry automatically. An interrupted or malformed mutation response has code catalog_commit_outcome_unknown; read the current state before deciding whether to retry. Typed definitions reject unknown fields, including nested fields, to prevent silent loss during replacement. Use raw() to read and apply newer definitions as serde_json::Value; arbitrary-precision JSON numbers are retained.

use scopedb_client::CatalogOptions;

# async fn demo() -> Result<(), scopedb_client::Error> {
# let client = scopedb_client::Client::new("http://127.0.0.1:6543", scopedb_client::reqwest::Client::new())?;
let resource = client.catalog().table("scopedb", "public", "events");
let snapshot = resource.get().await?;
let mut definition = snapshot.definition;
definition.comment = Some("Application events".to_string());
let preview = resource.apply(&definition, CatalogOptions {
    dry_run: true,
    if_match: Some(snapshot.etag),
    ..Default::default()
}).await?;
println!("destructive = {}", preview.destructive);
resource.apply(&definition, CatalogOptions {
    if_match: Some(preview.etag),
    ..Default::default()
}).await?;
# Ok(())
# }

See examples/catalog.rs for read-only catalog discovery.

Writing rows

The destination table must already exist. Table uses scopedb and public when the database or schema is not specified.

Write with a stream

Use append_stream() to send Rust values. The SDK converts them to NDJSON and writes them in batches automatically.

# async fn demo() -> Result<(), scopedb_client::Error> {
# let client = scopedb_client::Client::new("http://127.0.0.1:6543", scopedb_client::reqwest::Client::new())?;
let stream = client.table("events").append_stream().build()?;

stream.send(&serde_json::json!({"id": 1, "name": "first"})).await?;
stream.send_all([
    serde_json::json!({"id": 2, "name": "second"}),
    serde_json::json!({"id": 3, "name": "third"}),
]).await?;

let report = stream.shutdown().await?;
println!("Written: {} rows", report.committed_rows);
# Ok(())
# }

send() adds one row to the SDK's pending writes; send_all() adds rows from an iterator. These calls do not wait for the rows to be written. Call shutdown() when you are finished to wait for writing to complete and close the stream. If send_all() fails partway, its error includes the accepted count and stats() retains cumulative counts; still settle that accepted prefix with shutdown().

To wait for pending writes while keeping the stream open, call flush():

# async fn demo() -> Result<(), scopedb_client::Error> {
# let client = scopedb_client::Client::new("http://127.0.0.1:6543", scopedb_client::reqwest::Client::new())?;
# let stream = client.table("events").append_stream().build()?;
let report = stream.flush().await?;
println!("Written: {} rows", report.committed_rows);
# stream.shutdown().await?;
# Ok(())
# }

A report covers activity since the previous barrier. With the default Stop policy, a failed flush() or shutdown() returns the complete error, and stats().last_report retains partial settlement counts. Earlier batches may have committed. Reports and failure events expose aggregate counts without identifying individual input records; use application-owned business keys to reconcile stored rows before selecting records to replay. stats().last_failure.error preserves the code, details, timeout reason, HTTP metadata, and source in Continue mode as well.

Dropping an enqueued flush or started shutdown future stops waiting, but settlement continues. Call shutdown() again to observe the final result. Dropping all producer handles drains accepted rows while the Tokio runtime remains alive; runtime termination cannot guarantee delivery.

Continue after a batch fails

To keep a log or telemetry sink running after batch failures, use AppendFailurePolicy::Continue and keep the circuit breaker enabled. The default Stop policy instead closes the stream after a batch failure. Continue mode is best effort: read flush()/shutdown() reports and stats() to account for failed, unknown, and locally dropped rows.

For producers running on multiple application replicas, start with one concurrent batch per stream and tune against the shared service's capacity. The default is four per stream: ten replicas with one stream each can issue up to 40 concurrent batches, while .max_concurrent_batches(1) reduces that total to ten. Account for additional streams in each replica as well.

use scopedb_client::AppendCircuitBreakerOptions;
use scopedb_client::AppendFailurePolicy;

# async fn demo() -> Result<(), scopedb_client::Error> {
# let client = scopedb_client::Client::new("http://127.0.0.1:6543", scopedb_client::reqwest::Client::new())?;
let stream = client
    .table("events")
    .append_stream()
    .failure_policy(AppendFailurePolicy::Continue)
    .circuit_breaker(Some(AppendCircuitBreakerOptions::default()))
    .max_concurrent_batches(1)
    .build()?;

if let Err(error) = stream.try_send(&serde_json::json!({"name": "request.completed"})) {
    eprintln!("telemetry row not admitted: {error}");
}
let report = stream.shutdown().await?;
println!("Write results: {report:?}");
# Ok(())
# }

The circuit breaker is enabled by default. It pauses new batch dispatch after five consecutive availability failures and waits 30 seconds before probing recovery. try_send() returns immediately if the circuit is open or local capacity is full; use it when collecting telemetry must not wait for capacity. send().await provides backpressure instead. Send diagnostics to a separate sink and monitor dropped rows, failed batches, and unknown outcomes. See examples/telemetry.rs for a complete example.

Append NDJSON

Use append() when your data is already newline-delimited JSON: one JSON object per line.

ScopeDB 0.6's NDJSON decoder accepts integer inputs only within the signed 64-bit range, including inputs for uint columns and nested JSON values. Larger integers are rejected by the server; the SDK does not silently narrow them. Query results still support the full u64 range.

# async fn demo() -> Result<(), scopedb_client::Error> {
# let client = scopedb_client::Client::new("http://127.0.0.1:6543", scopedb_client::reqwest::Client::new())?;
let table = client.table("events");
let ndjson = r#"{"id": 1, "name": "first"}
{"id": 2, "name": "second"}"#;

let result = table.append(ndjson).await?;
println!("Written: {} rows", result.num_rows_inserted);
# Ok(())
# }

More examples

Inspecting errors

Use the error's methods to read its message, HTTP status, and request ID:

# fn inspect(error: &scopedb_client::Error) {
eprintln!("{}", error.message());
eprintln!("status = {:?}", error.http_status());
eprintln!("request_id = {:?}", error.request_id());
eprintln!("retryable = {}", error.is_retryable());
eprintln!("retry_after = {:?}", error.retry_after());
# }

When a statement reaches the failed state, Error::statement_details() preserves the server-provided code, message, and optional code-specific JSON:

# fn inspect(error: &scopedb_client::Error) {
if let Some(details) = error.statement_details() {
    eprintln!("statement failed with {}: {}", details.code, details.message);
    eprintln!("details = {:?}", details.details);
}
# }

Table helper

# async fn demo() -> Result<(), scopedb_client::Error> {
# let client = scopedb_client::Client::new("http://127.0.0.1:6543", scopedb_client::reqwest::Client::new())?;
let table = client.table("events").with_schema("public");
println!("identifier = {}", table.identifier());

let snapshot = table.catalog().get().await?;
let description = snapshot.definition;
println!("etag = {}", snapshot.etag);
println!("columns = {}", description.columns.len());
# Ok(())
# }

Migrate to ScopeDB 0.6

• Replace with_exec_timeout with with_max_elapsed(Duration). with_max_parallelism is removed; with_max_estimated(Duration) independently limits estimated execution time and does not control parallelism. Durations must be positive. Elapsed time includes queueing; omitted limits use server defaults.
• Replace list_*, fetch_*, and iterate_* Catalog calls with client.catalog().get() or a database/schema/table/view selector followed by .get(). Reads return the complete definition and an etag, without pagination.
• Replace Table::describe() and table_schema() with table.catalog().get(), returning both definition and etag. Use distinct and retention_days, replacing distinct_on and data_retention_days. Typed definitions reject unsupported fields; .raw() preserves them. Table::drop() remains available; use Catalog delete() when a precondition or preview is needed.
• cancel() returns and caches the complete StatementStatus, including any finished result or original failure. Failed states require error; cancelled states have no server message. Error::code() and details() retain structured errors, including unknown codes. Progress includes compute_seconds.
• Table appends use /v1/catalog/databases/{db}/schemas/{schema}/tables/{table}/ingest. Rename AppendRowsResult::append_state and AppendErrorDetails::append_state to status, matching the native response. Missing or unreadable outcomes remain unknown. The default still retries only temporary rejections. The new .retry_unknown(true) option permits replaying transient unknown outcomes and may insert duplicates; client timeouts are never replayed.
• Replace last_status() with snapshot(), status() / fetch_once() with refresh(), and fetch() with wait(). Read progress from snapshot().map(StatementStatus::progress) and match StatementStatus::Finished to inspect cached results. wait() also returns cached results without another request.
• Replace the removed json_rows() alias with raw_rows(), which directly returns the slice. Replace stream builder aliases batch_bytes, max_pending_bytes, and max_in_flight_requests with target_batch_bytes, max_buffered_bytes, and max_concurrent_batches.
• Read stats().last_failure.error instead of separate message/status fields. Its complete Error retains diagnostics across barriers. AppendStreamStats no longer implements equality because errors carry sources.
• Table::identifier() always includes the database and schema defaults, so SQL queries and drop() select the same table as Append and Catalog, including when only the database is configured. The legacy u_int type spelling is removed; use uint.

Upgrading the server from 0.5 clears statement history and saved results. Previously saved statement handles cannot be resumed. The removed transform Ingest API requires the migration below.

Migrate from Ingest

Client::insert, Client::ingest_stream, IngestStream, IngestStreamBuilder, and IngestResult have been removed. Use Table::append for one NDJSON request or Table::append_stream to batch Rust values. This SDK version requires ScopeDB 0.6.0 or later.

Append accepts rows matching the destination columns, not a ScopeQL statement. Move field renaming, type conversion, filtering, and routing into the producer before calling send. For transformations that must run in ScopeQL, append the original event into an object column in a staging table, then execute a separate FROM ... SELECT ... INSERT INTO ... statement. Staging and transformation are separate commits; track processed batches before retrying the transformation.

# async fn demo() -> Result<(), scopedb_client::Error> {
# let client = scopedb_client::Client::new("http://127.0.0.1:6543", scopedb_client::reqwest::Client::new())?;
let source = serde_json::json!({"ts": "2026-03-13T12:00:00Z", "name": "ScopeDB"});
let stream = client.table("events").append_stream()
    .max_concurrent_batches(1)
    .max_retries(0)
    .build()?;
let sent = stream.send(&serde_json::json!({
    "occurred_at": source["ts"],
    "name": source["name"],
})).await;
let report = stream.shutdown().await;
sent?;
println!("Committed rows: {}", report?.committed_rows);
# Ok(())
# }

Review these behavior changes when migrating:

• Append batches allow up to 8 MiB of uncompressed NDJSON and 200,000 rows. Reduce an old ingest batch-byte target above 8 MiB.
• Append stream build() returns a Result; handle configuration errors with ? before sending records.
• flush() and shutdown() return an AppendDeliveryReport for activity since the previous report. Read committed_rows instead of num_rows_inserted, and check outcome when using the continue policy. Direct append() still returns num_rows_inserted.
• Append streams send up to four batches concurrently by default. Use .max_concurrent_batches(1) for sequential requests; queries still require ORDER BY for ordered results.
• Append streams retry temporary rejections, where no rows committed. Use .retry_unknown(true) only if duplicate rows are acceptable, or .max_retries(0) to disable all retries. Client timeouts are never retried. Direct append() makes one attempt.

Examples

The example guide in examples/README.md includes setup instructions and commands to run each example from the repository root.

The wire-level endpoint and payload reference is in docs/rust-http-api.md. Release history is in the repository's CHANGELOG.md.

License

Licensed under the Apache License, Version 2.0.