# 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
```sh
cargo add datum-sql
```
Enable connector adapters only when you need them:
```sh
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
```rust
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:
```sh
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.
```rust
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.
```rust
# 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]))?;
.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`](../../roadmap/M11-datum-sql.md). Benchmark plans and
results live under [`roadmap/benchmarks/`](../../roadmap/benchmarks/).