Crate datum_sql

Expand description

§datum-sql

datum-sql is Datum’s SQL front end: DataFusion parses, optimizes, and builds physical expressions; Datum lowers the supported physical plan into ordinary Datum stream stages and materializes it with Datum sinks.

The crate uses Arrow RecordBatch values as the SQL data unit. Append-only queries stay as Source<RecordBatch>. Updating streams use Source<ChangelogBatch>, where row-level ChangeOp metadata stays outside Arrow row data so projections and filters cannot accidentally drop it.

§Install

cargo add datum-sql

Enable connector adapters only when you need them:

cargo add datum-sql --features mq,cdc

§What Works

Register in-memory Datum sources as SQL tables.
Register append-only, committable, and changelog/updating sources.
Parse and plan SQL with DataFusion 54 over Arrow 58.
Lower scans, projection, filter, coalesce, limit, windowed aggregation, and supported inner equi-joins into Datum stages.
Run bounded append-only queries and collect Arrow batches.
Build continuous SqlEvent<RecordBatch> streams carrying data, watermarks, and future barrier markers outside row data.
Assign event-time watermarks from an Arrow timestamp column with bounded out-of-orderness.
Register append-only and changelog-aware SQL sinks.
Execute INSERT INTO <registered_sink> SELECT ... with at-least-once source-position commit ordering for committable sources.
Opt into Kafka JSON table/sink adapters with the mq feature and PostgreSQL CDC changelog adapters with the cdc feature.

§Quick Start

use std::sync::Arc;

use arrow::array::{Int64Array, StringArray};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::record_batch::RecordBatch;
use datum::Source;
use datum_sql::DatumSqlContext;

let schema = Arc::new(Schema::new(vec![
    Field::new("city", DataType::Utf8, false),
    Field::new("temp", DataType::Int64, false),
]));
let batch = RecordBatch::try_new(
    Arc::clone(&schema),
    vec![
        Arc::new(StringArray::from(vec!["sf", "nyc", "sea"])),
        Arc::new(Int64Array::from(vec![67, 74, 58])),
    ],
)?;

let context = DatumSqlContext::new();
context.register_source("weather", schema, Source::from_iter([batch]))?;

let output = context
    .execute("SELECT city, temp FROM weather WHERE temp >= 70")
    .await?;

assert_eq!(output[0].num_rows(), 1);

Run the checked example from the repository:

cargo run -p datum-sql --example projection_filter

§Windowed Streaming Query

Event-time queries register the timestamp column at the table boundary. The continuous lowering returns a Datum source of SqlEvent<RecordBatch> values, so watermarks stay outside user row data.

use std::sync::Arc;
use std::time::Duration;

use arrow::array::{Int64Array, TimestampNanosecondArray};
use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
use arrow::record_batch::RecordBatch;
use datum::Source;
use datum_sql::{DatumSqlContext, EventTimeConfig};

let schema = Arc::new(Schema::new(vec![
    Field::new("event_time", DataType::Timestamp(TimeUnit::Nanosecond, None), false),
    Field::new("auction", DataType::Int64, false),
    Field::new("price", DataType::Int64, false),
]));
let batch = RecordBatch::try_new(
    Arc::clone(&schema),
    vec![
        Arc::new(TimestampNanosecondArray::from(vec![1_000_000_000_i64])),
        Arc::new(Int64Array::from(vec![42])),
        Arc::new(Int64Array::from(vec![900])),
    ],
)?;

let context = DatumSqlContext::new();
context.register_source_with_event_time(
    "bids",
    schema,
    Source::from_iter([batch]),
    EventTimeConfig::bounded_out_of_orderness("event_time", Duration::from_secs(1)),
)?;

let source = context
    .streaming_source(
        "SELECT date_bin(INTERVAL '10 seconds', event_time) AS window_start, \
                auction, count(*) AS bid_count \
         FROM bids \
         GROUP BY date_bin(INTERVAL '10 seconds', event_time), auction",
    )
    .await?;

§Continuous Queries And Sinks

For streaming use, call streaming_source to get a Datum source of SqlEvent<RecordBatch> values, or call execute_streaming to materialize a continuous query. SqlEvent::Data carries user rows; SqlEvent::Watermark and SqlEvent::Barrier carry stream control metadata outside Arrow columns.

let context = DatumSqlContext::new();
context.register_source("input", schema, Source::from_iter([batch]))?;
context.register_append_sink("out", |_batch| Ok(()))?;

let handle = context
    .execute_streaming("INSERT INTO out SELECT id FROM input")
    .await?;
handle.wait()?;

§Feature Flags

datum-sql has no default features.

mq enables Kafka JSON source and sink adapters built on datum-mq.
cdc enables PostgreSQL CDC changelog table adapters built on datum-cdc.

§Known Limitations

SQL coverage is intentionally narrow. Unsupported DataFusion physical nodes return DataFusionError::NotImplemented instead of falling back to DataFusion execution.
Streaming joins support append-only inner equi-joins. Outer joins, semi/anti joins, non-equi joins, and changelog joins are not implemented yet.
Windowed aggregation supports fixed tumbling and hopping windows over event-time input, with a conservative aggregate set. Session windows, grouping sets, aggregate FILTER, distinct aggregates, and processing-time windows are deferred.
Changelog-aware sinks are explicit. Append-only sinks reject updating streams; there is no implicit adapter that drops retractions.
Sink execution is at-least-once. Exactly-once requires checkpoint barriers, durable operator state, and transactional sink recovery work that is still design-only.
Kafka support is JSON-row oriented today. Avro, Protobuf, schema registry integration, and long-lived producer batching are future work.
CDC decoding covers the current datum-cdc pgoutput text-value shape. Initial snapshots, DDL/schema history, and binary pgoutput decoding are not provided by datum-sql.
The SQL benchmark record is frozen in roadmap/benchmarks/sql.md. Current durable verdicts include a q3 all-metric win, q4/q7 wall wins, corrected q8 wall wins after WP-SQL-F1, and q5/ingest misses with named follow-up levers; cite that record for exact numbers and caveats.

§Design References

The design record and go/no-go notes live in roadmap/M11-datum-sql.md. Benchmark plans and results live under roadmap/benchmarks/.

Re-exports§

pub use connect::BatchEnvelope;
pub use connect::EnvelopedBatch;
pub use connect::EnvelopedRecordBatch;

Modules§

connect: Connector adapters for datum-sql.

Structs§

ChangelogBatch: A typed changelog envelope over a plain Arrow RecordBatch.
CommittableRecordBatch: Append-only SQL batch plus source metadata and a deferred commit handle.
ContinuousQueryHandle: Materialized handle for a streaming SQL query.
DatumSqlContext: SQL planning context that registers Datum sources and executes supported DataFusion physical plans on Datum.
EventTimeConfig: Event-time declaration for a registered SQL table.
SourceCommit: A committable source position.
SqlBarrier: Future checkpoint/alignment marker carried outside SQL row data.
SqlExecutionMetrics: Per-query counters populated by SQL lowering and native Datum stages.
StreamingJoinConfig: Streaming join lowering policy.
StreamingJoinMetrics: Counters maintained by streaming join stages.
StreamingJoinStateLimits: Hard limits for unbounded streaming equi-joins.
StreamingJoinWindow: Event-time interval join settings.
Watermark: Event-time progress in nanoseconds since the Unix epoch.
WindowedAggregationMetrics: Counters maintained by a windowed aggregation stage.

Enums§

ChangeOp: Row-level operation for SQL streams that can revise prior output.
SqlEvent: In-band SQL stream envelope for data and control records.
SqlSourcePosition: Type-erased source position carried by SQL sink pipelines.
StreamingJoinMode: Supported streaming join state classes.
WatermarkStrategy: Strategy for producing watermarks from an event-time column.

Traits§

SqlEventPayload: Payloads that expose the Arrow batch used to read event-time columns.

Functions§

assign_event_time_watermarks: Assigns event-time watermarks to a source.
data_events: Wraps every payload from a source as SqlEvent::Data.
map_sql_event_data: Applies a function to the data branch of a SQL event stream.

Crate datum_sql

Crate datum_sql Copy item path

§datum-sql

§Install

§What Works

§Quick Start

§Windowed Streaming Query

§Continuous Queries And Sinks

§Feature Flags

§Known Limitations

§Design References

Re-exports§

Modules§

Structs§

Enums§

Traits§

Functions§

Crate datum_sql