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;

# async fn run() -> datafusion::common::Result<()> {
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);
# Ok(())
# }

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};

# async fn run() -> datafusion::common::Result<()> {
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?;
# let _ = source;
# Ok(())
# }

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.

# use std::sync::Arc;
# use arrow::array::Int64Array;
# use arrow::datatypes::{DataType, Field, Schema};
# use arrow::record_batch::RecordBatch;
# use datum::Source;
# use datum_sql::DatumSqlContext;
# async fn run() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
# let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
# let batch = RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(Int64Array::from(vec![1]))])?;
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()?;
# Ok(())
# }

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/.