SQL

The datum-sql crate lets you put DataFusion SQL in front of Datum streams. DataFusion owns SQL parsing, optimization, and Arrow expression evaluation. Datum owns source materialization, backpressure, stream execution, and sink materialization.

The boundary is deliberate: building tables, queries, and sinks creates blueprints. Work starts only when you call execute, streaming_source, execute_streaming, or select_into.

Data Model

SQL rows travel in Arrow RecordBatch values, not one Datum element per row. Append-only queries use Source<RecordBatch>.

Updating streams use Source<ChangelogBatch>. A ChangelogBatch carries a plain Arrow payload plus row-level ChangeOp values:

Operation	Meaning
Insert	Adds a row.
Delete	Retracts a row.
UpdateDelete	Old row in an adjacent update pair.
UpdateInsert	New row in an adjacent update pair.

Watermarks and checkpoint barriers are also outside Arrow row data. Continuous queries use SqlEvent<T>:

Event	Meaning
Data(T)	A data batch, usually RecordBatch or ChangelogBatch.
Watermark(Watermark)	Event-time progress in nanoseconds.
Barrier(SqlBarrier)	Future checkpoint/alignment marker.

Register A Source

Use DatumSqlContext to register a Datum source as a SQL table.

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(())
# }

execute is for bounded append-only queries. It lowers the supported DataFusion physical plan into Datum stages, runs it, and collects the output batches.

Event Time

Register event time at the table boundary so scan projection cannot lose the timestamp column needed for watermark generation.

use std::time::Duration;
use datum_sql::EventTimeConfig;

# use std::sync::Arc;
# use arrow::array::TimestampNanosecondArray;
# use arrow::datatypes::{DataType, Field, Schema, TimeUnit};
# use arrow::record_batch::RecordBatch;
# use datum::Source;
# use datum_sql::DatumSqlContext;
# fn source() -> datafusion::common::Result<(Arc<Schema>, RecordBatch)> {
# let schema = Arc::new(Schema::new(vec![Field::new(
#     "event_time",
#     DataType::Timestamp(TimeUnit::Nanosecond, None),
#     false,
# )]));
# let batch = RecordBatch::try_new(
#     Arc::clone(&schema),
#     vec![Arc::new(TimestampNanosecondArray::from(vec![1_i64]))],
# )?;
# Ok((schema, batch))
# }
# fn register() -> datafusion::common::Result<()> {
# let (schema, batch) = source()?;
# let context = DatumSqlContext::new();
context.register_source_with_event_time(
    "events",
    schema,
    Source::from_iter([batch]),
    EventTimeConfig::bounded_out_of_orderness("event_time", Duration::from_secs(5)),
)?;
# Ok(())
# }

Continuous lowering emits SqlEvent::Watermark values as event time advances. Windowed aggregation and streaming joins use those watermarks to close windows, drop late rows, and evict retained state.

Continuous Queries

Use streaming_source when you want to wire the lowered query into your own Datum graph:

# use datum_sql::DatumSqlContext;
# async fn run(context: DatumSqlContext) -> datafusion::common::Result<()> {
let source = context
    .streaming_source("SELECT * FROM events")
    .await?;
# let _ = source;
# Ok(())
# }

Use execute_streaming when the SQL should run until completion, cancellation, or failure:

# use datum_sql::DatumSqlContext;
# async fn run(context: DatumSqlContext) -> datafusion::common::Result<()> {
let handle = context.execute_streaming("SELECT * FROM events").await?;
handle.cancel();
# Ok(())
# }

Sinks

Sinks are registered separately from DataFusion tables because they have side effects and acceptance rules.

# use datum_sql::DatumSqlContext;
# async fn run(context: DatumSqlContext) -> datafusion::common::Result<()> {
context.register_append_sink("out", |_batch| Ok(()))?;

let handle = context
    .execute_streaming("INSERT INTO out SELECT * FROM events")
    .await?;
handle.wait()?;
# Ok(())
# }

Append-only sinks reject updating streams before materialization. To consume retractions, register a changelog-aware sink with register_changelog_sink.

Committable sources use at-least-once ordering: write the transformed output batch first, then commit the source position only after the sink confirms the write. Exactly-once remains future work.

Connectors

The optional mq feature adds Kafka JSON adapters built on datum-mq. JsonRowFormat maps one JSON object per Kafka record into an Arrow schema, and committable sources carry Kafka offset ranges through CommittableRecordBatch.

The optional cdc feature adds PostgreSQL CDC adapters built on datum-cdc. CdcTableAdapter maps ChangeEvent values into typed ChangelogBatch values.

Neither connector stores offsets, schema revisions, watermarks, or changelog ops in hidden Arrow columns.

Supported SQL Shape

The supported physical-plan lowering is intentionally conservative:

Area	Current support
Scans	Datum append-only, committable, and changelog sources.
Unary operators	Projection, filter, coalesce, local/global limit.
Windows	Fixed tumbling and hopping event-time aggregation.
Joins	Append-only inner equi-joins with windowed retention or hard state limits.
Sinks	INSERT INTO <registered_sink> SELECT ....

Unsupported plan nodes return a DataFusion NotImplemented error. There is no fallback path that silently runs part of the query on DataFusion's executor.

Limitations

• Outer joins, semi/anti joins, non-equi joins, and changelog joins are not implemented.
• Session windows, processing-time windows, grouping sets, distinct aggregates, aggregate FILTER, and broad SQL function coverage are deferred.
• Kafka support is JSON-row oriented today; Avro, Protobuf, schema registry support, and long-lived producer batching are future work.
• CDC support follows the current datum-cdc pgoutput text-value shape. Initial snapshots and DDL/schema history are outside this crate.
• Sink execution is at-least-once. Exactly-once needs checkpoint barriers, durable state, and transactional sink recovery.
• The SQL benchmark record is frozen in roadmap/benchmarks/sql.md: q3 is a claimable all-metric win, q4/q7 keep wall wins, q8 keeps wall wins after WP-SQL-F1, and q5 plus the ingestion rows remain behind with named follow-up levers.

See Also

• Kafka for datum-mq source and sink behavior.
• CDC for datum-cdc PostgreSQL source behavior.
• Agent for supervised long-running jobs.