nordisk/knut
crates/knut-thund/src/backend/native.rs
//! The **native Arrow/DataFusion** execution backend (feature `native`).
//!
//! This is the Arroyo-shaped half of Þund: rather than ship a pipeline off to
//! a cluster, it runs it in-process as Arrow `RecordBatch` streams. The design
//! (per the landscape research) is:
//!
//! - Reuse DataFusion's planner + **physical operators + expression
//! evaluator** rather than writing query kernels from scratch — wrap them in
//! *streaming* operator versions for unbounded flows (exactly Arroyo's
//! strategy on DataFusion).
//! - Carry DataFusion's orthogonal `Boundedness` × `EmissionType` so the
//! planner mechanically rejects unrunnable streaming plans up front (the
//! `SanityCheckPlan` analogue surfaced by [`super::ExecBackend::check`]).
//! - Event-time watermarks (min-across-inputs, idle-partition handling, an
//! explicit allowed-lateness knob — the thing Arroyo lacks), windowing, and
//! barrier-based checkpointing as **first-class** additions on top of
//! DataFusion, which deliberately omits them.
//! - State **off-heap on object storage** (incremental Parquet on S3, *not*
//! RocksDB — the Arroyo/RisingWave consensus), so workers stay stateless and
//! rescale cheaply.
//! - Transport: Arrow IPC over the wire (and Flight for inter-node shuffle),
//! zero-copy, reusing knut's existing Arrow/Flight stack.
//!
//! # What runs today
//!
//! With the `native` feature the **batch** dataflow runs genuinely end to end
//! on DataFusion: the pipeline's datasets are materialised in topological
//! order — each flow's `query` is planned + executed by DataFusion, its result
//! registered in-process (a `MemTable`) under the target dataset name so
//! downstream flows read it, and data-quality [`crate::ir::Expectation`]s are
//! evaluated (warn / drop-row / fail-update) against the produced rows. Real
//! [`RunEvent`]s carry the per-flow lifecycle and row counts, and the produced
//! batches are retrievable from [`NativeRun`]`::output` (feature `native`).
//!
//! The **streaming** layer runs unbounded (Kafka-style) sources as **micro-batch
//! streaming** on DataFusion: seed a synthetic source with
//! [`NativeBackend::with_stream_input`] (N micro-batches) and each trigger pulls
//! the next micro-batch, re-runs the flow's IR→DataFusion lowering over the
//! accumulated state, and emits an incremental result honoring the streaming
//! IR's [`crate::ir::Trigger`] (per-batch cadence vs `AvailableNow` single pass)
//! and [`crate::ir::OutputMode`] (Complete = full table each trigger; Append /
//! Update = the delta since the previous trigger). The final materialised table
//! equals the batch path over the same rows. A streaming flow with **no**
//! in-process data (no seed, no live Kafka/CDC infra), a CDC apply-changes/SCD
//! merge, or a query DataFusion cannot plan is still reported honestly as
//! *deferred* on its [`RunEvent`] rather than silently faked.
//!
//! Not yet covered (documented non-goals of this micro-batch pass): event-time
//! watermark-based late-row dropping / window closing, barrier checkpoints, and
//! exactly-once — the recompute-over-accumulated strategy is at-least-once and
//! recomputes per trigger rather than maintaining incremental operator state.
//!
//! Everything here is gated on the `native` feature so the IR + builder + Spark
//! backend compile without the DataFusion tree.
use super::{Capabilities, ExecBackend, RunEvent, RunHandle, RunPhase};
use crate::error::Result;
#[cfg(not(feature = "native"))]
use crate::error::ThundError;
use crate::ir::Pipeline;
#[cfg(feature = "native")]
use datafusion::arrow::array::RecordBatch;
/// How the native backend drives **unbounded (streaming/CDC)** flows.
///
/// Additive mode selector (design `thund-streaming-state.md`, phase 0). The
/// default is [`StreamingMode::Recompute`] — today's behaviour, byte-for-byte
/// unchanged — so every existing caller keeps the recompute path with no edit.
/// The mode is read only inside the streaming dispatch; batch flows ignore it.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum StreamingMode {
/// Re-run the flow's query over the accumulated micro-batch state each
/// trigger (`run_streaming_flow`). At-least-once, no operator state,
/// unbounded working set. **DEFAULT and fallback — never removed.**
#[default]
Recompute,
/// Event-time watermark tracker + (future) windowed/keyed operator state +
/// barrier checkpoints under `state_root`. Opt-in. Degrades honestly to
/// [`StreamingMode::Recompute`] when it cannot run (no `state_root`).
Incremental,
}
/// The native Arrow/DataFusion backend.
///
/// Construct with [`NativeBackend::new`]; seed input tables with
/// [`NativeBackend::with_input`] (feature `native`) and configure the
/// object-store state root with [`NativeBackend::with_state_root`].
#[derive(Debug, Default, Clone)]
pub struct NativeBackend {
/// Object-store URI under which incremental checkpoints/state are written
/// (`s3://…`, `file://…`). `None` = in-memory only (dev).
pub state_root: Option<String>,
/// How unbounded flows are driven. Defaults to [`StreamingMode::Recompute`]
/// (today's behaviour); [`StreamingMode::Incremental`] opts into the
/// event-time / stateful path when a `state_root` is set.
pub streaming_mode: StreamingMode,
/// Cap on the number of streaming triggers processed this run (`None` =
/// all — the default). A bounded run: the engine consumes the first `N`
/// micro-batches, checkpoints each, and stops. Additive and only consulted
/// on the incremental path; it is the **restart boundary** the crash-recovery
/// path (design phase 4) resumes across — run once capped, run again
/// uncapped over the same source and the checkpoint resumes from the
/// committed offset with no double-emit and no lost late events.
pub max_triggers: Option<usize>,
/// In-memory input tables seeded for a run, keyed by the table name the
/// flow queries reference. Lets a pipeline run end to end without external
/// files (the batch equivalent of a bounded source). Only present with the
/// `native` feature (it holds Arrow `RecordBatch`es).
#[cfg(feature = "native")]
seed_inputs: Vec<(String, Vec<RecordBatch>)>,
/// In-memory **streaming** sources seeded for a run, keyed by the table name
/// the flow's query reads (a Kafka flow's topic). Each source is delivered
/// as a sequence of micro-batches (`Vec<Vec<RecordBatch>>`) — one inner
/// `Vec<RecordBatch>` per trigger's worth of arriving rows. This is the
/// synthetic (bounded-but-streamed) equivalent of an unbounded source, so a
/// streaming pipeline runs end to end in-process without live Kafka/CDC.
#[cfg(feature = "native")]
seed_streams: Vec<(String, Vec<Vec<RecordBatch>>)>,
/// On-disk bounded file sources registered for a run as `(table, path,
/// format)` — the on-disk sibling of [`NativeBackend::with_input`]. Each is
/// registered under its explicit `table` name (parquet/csv) before any flow
/// runs, so a plain **batch** flow whose query reads that name scans the
/// local file directly — the natural "point thund at `orders.csv`, run SQL,
/// get a table back" path, no in-memory pre-load and no live infra. Only
/// present with the `native` feature.
/// Each entry is `(table, path, format, partition_cols)`. An empty
/// `partition_cols` (the [`with_file_input`](NativeBackend::with_file_input)
/// default) registers a plain single-file/directory source; a non-empty one
/// (via [`with_partitioned_file_input`](NativeBackend::with_partitioned_file_input))
/// registers a **Hive-partitioned directory** whose partition columns are
/// recovered from the `col=val/…` path segments — the read-side sibling of
/// the partitioned file sink.
#[cfg(feature = "native")]
seed_file_inputs: Vec<(String, String, String, Vec<String>)>,
/// On-disk file **sinks** requested for a run as `(dataset, path, format)` —
/// the write-side sibling of [`NativeBackend::with_file_input`]. After a
/// dataset materialises, its rows are written to a single local
/// `parquet`/`csv` file at `path`, completing the on-disk round trip a laptop
/// consumer wants: read `orders.csv`, run a `GROUP BY`, and land the result
/// as `rollup.parquet` on disk — no external system, no catalog. Additive and
/// only present with the `native` feature; the in-memory `NativeRun::output`
/// path is unchanged whether or not a sink is declared.
#[cfg(feature = "native")]
seed_file_outputs: Vec<(String, String, String)>,
/// FalkorDB graph-sink target as `(redis_url, graph_name)`. When set, any
/// dataset that is a **graph sink** — an [`OutputType::Sink`] with a
/// `falkordb-node` / `falkordb-edge` format carrying the rel2graph
/// `knut.node.*` / `knut.edge.*` properties — is written to FalkorDB after
/// the dataflow materialises it, as an idempotent `UNWIND … MERGE` load
/// (nodes-before-edges, both endpoints MERGEd, keyed on the merge key). The
/// write is driven through the EXISTING
/// [`knut_bifrost::sink::FalkorDbSink`] (built on `graphar-falkordb`); this
/// backend reimplements no Cypher — it only maps the thund IR + Arrow
/// batches onto the bifröst `GraphMutation` contract. `None` (the default)
/// still *constructs* the mutation from a graph-sink dataset and records it
/// on [`NativeRun::graph_sink`], but writes nothing (honest construction-only
/// — never a silent fake). Present only with `native` + `rel2graph`.
#[cfg(all(feature = "native", feature = "rel2graph"))]
falkordb_sink: Option<(String, String)>,
}
impl NativeBackend {
/// A new in-memory native backend (no durable state).
pub fn new() -> Self {
Self::default()
}
/// Set the object-store state root, returning `self`.
pub fn with_state_root(mut self, uri: impl Into<String>) -> Self {
self.state_root = Some(uri.into());
self
}
/// Select the [`StreamingMode`] for unbounded flows, returning `self`.
/// Defaults to [`StreamingMode::Recompute`]; passing
/// [`StreamingMode::Incremental`] opts into the event-time / stateful path
/// (which degrades to recompute if no `state_root` is set).
pub fn with_streaming_mode(mut self, mode: StreamingMode) -> Self {
self.streaming_mode = mode;
self
}
/// Cap the number of streaming triggers this run consumes (design phase 4's
/// restart boundary), returning `self`. `None` (default) processes the whole
/// seeded stream. Only the incremental path honours it; it lets a run stop
/// after a committed epoch so a subsequent run recovers from the checkpoint.
pub fn with_max_triggers(mut self, max: Option<usize>) -> Self {
self.max_triggers = max;
self
}
/// Seed an in-memory input table under `name`, readable by any flow whose
/// query references it. The batch equivalent of a bounded source — this is
/// how a pipeline runs end to end in-process without external files.
#[cfg(feature = "native")]
pub fn with_input(mut self, name: impl Into<String>, batches: Vec<RecordBatch>) -> Self {
self.seed_inputs.push((name.into(), batches));
self
}
/// Seed a synthetic **streaming** source under `name`, delivered as
/// `micro_batches` — each inner `Vec<RecordBatch>` is one micro-batch (one
/// trigger's worth of arriving rows). This lets a streaming / unbounded
/// (Kafka-style) flow run end to end in-process as micro-batch streaming
/// without live Kafka/CDC infra: the backend pulls the next micro-batch each
/// trigger, re-runs the IR→DataFusion lowering over the accumulated state,
/// and emits incremental results per the flow's [`crate::ir::OutputMode`].
/// `name` must be the table the flow's query reads (a Kafka flow's topic).
#[cfg(feature = "native")]
pub fn with_stream_input(
mut self,
name: impl Into<String>,
micro_batches: Vec<Vec<RecordBatch>>,
) -> Self {
self.seed_streams.push((name.into(), micro_batches));
self
}
/// Register an **on-disk** bounded file source under `name`, readable by any
/// batch flow whose query references it. The on-disk sibling of
/// [`with_input`](Self::with_input): instead of handing Arrow
/// `RecordBatch`es in memory, point thund at a local `parquet`/`csv` file and
/// it scans it directly. `format` is `"parquet"` or `"csv"`; `path` may carry
/// a `file://` prefix (stripped) or be a plain filesystem path. This is the
/// natural laptop-consumer entry — "read `orders.csv`, run a `SELECT … GROUP
/// BY`, get a table back" — with no external service and no in-memory
/// pre-load. Additive: the default build and every in-memory path are
/// unchanged.
#[cfg(feature = "native")]
pub fn with_file_input(
mut self,
name: impl Into<String>,
path: impl Into<String>,
format: impl Into<String>,
) -> Self {
self.seed_file_inputs
.push((name.into(), path.into(), format.into(), Vec::new()));
self
}
/// Register an on-disk **Hive-partitioned directory** as a bounded file
/// source under `name`, recovering the declared `partition_cols` from the
/// `col=val/…` path segments. The read-side sibling of
/// [`with_file_output`](Self::with_file_output)'s partitioned sink: where the
/// partitioned sink writes `region=US/…/part.parquet`, this reads that
/// directory back with `region` reconstructed as a real column (from the
/// path, not the data files), closing the partitioned round trip through
/// thund's own source path instead of a hand-rolled directory walk. `path` is
/// the directory ROOT (the `col=val` subdirs live under it); `format` is
/// `"parquet"` or `"csv"`. Partition values are recovered as `Utf8` strings
/// (standard Hive semantics — the value is a path segment). Additive: with an
/// empty `partition_cols` this is exactly [`with_file_input`](Self::with_file_input).
#[cfg(feature = "native")]
pub fn with_partitioned_file_input(
mut self,
name: impl Into<String>,
path: impl Into<String>,
format: impl Into<String>,
partition_cols: impl IntoIterator<Item = impl Into<String>>,
) -> Self {
self.seed_file_inputs.push((
name.into(),
path.into(),
format.into(),
partition_cols.into_iter().map(Into::into).collect(),
));
self
}
/// Write dataset `name`'s materialised rows to a single on-disk file after
/// the run, returning `self`. The write-side sibling of
/// [`with_file_input`](Self::with_file_input): where that points a flow at a
/// file to *read*, this lands a produced dataset as a `parquet`/`csv` file to
/// *keep*. `format` is `"parquet"` or `"csv"`; `path` may carry a `file://`
/// prefix (stripped) or be a plain filesystem path, and its parent directory
/// is created if missing. Completes the laptop round trip — read `orders.csv`,
/// run a `SELECT … GROUP BY`, write `rollup.parquet` — with no external
/// system and no catalog. The written file (path + row count + format) is
/// reported back via [`NativeRun::sink`]. Additive: declaring a sink does not
/// change the in-memory [`NativeRun::output`] path.
#[cfg(feature = "native")]
pub fn with_file_output(
mut self,
name: impl Into<String>,
path: impl Into<String>,
format: impl Into<String>,
) -> Self {
self.seed_file_outputs
.push((name.into(), path.into(), format.into()));
self
}
/// Set the FalkorDB **graph-sink** target as `(redis_url, graph_name)`,
/// returning `self`. With a target set, any graph-sink dataset — an
/// [`OutputType::Sink`] with a `falkordb-node` / `falkordb-edge` format
/// carrying the rel2graph `knut.node.*` / `knut.edge.*` properties (see
/// [`crate::rel2graph::to_thund_pipeline`]) — is written to FalkorDB after
/// the dataflow runs, as an idempotent `UNWIND … MERGE` graph load. This is
/// the executable half of "bifröst mapping → thund → NativeBackend →
/// FalkorDB": the write goes through the existing
/// [`knut_bifrost::sink::FalkorDbSink`] (nodes-before-edges, both-endpoints
/// MERGE, keyed on the merge key), reusing `graphar-falkordb` — no Cypher is
/// reimplemented here. Without a target the mutation is still *constructed*
/// and recorded on [`NativeRun::graph_sink`] (construction-only). Present
/// only with `native` + `rel2graph`.
#[cfg(all(feature = "native", feature = "rel2graph"))]
pub fn with_falkordb_sink(mut self, url: impl Into<String>, graph: impl Into<String>) -> Self {
self.falkordb_sink = Some((url.into(), graph.into()));
self
}
}
impl ExecBackend for NativeBackend {
type Run = NativeRun;
fn capabilities(&self) -> Capabilities {
// The native backend is the full-featured one: everything the IR can
// express, it (eventually) runs.
Capabilities {
name: "native".into(),
batch: true,
streaming: true,
event_time: true,
cdc: true,
expectations: true,
// Advertised true only when the incremental engine is actually
// selectable: `Incremental` mode AND a durable `state_root`.
// Otherwise the backend runs recompute, so it advertises false.
incremental_state: self.streaming_mode == StreamingMode::Incremental
&& self.state_root.is_some(),
// HONEST: false, and this one is an implementation gap rather than a
// design limit. The IR carries transforms verbatim, but the file
// sink maps every `partition_cols` entry to a literal Hive path
// segment (`register_*` with `table_partition_cols`, see
// `register_named_file`), so `months(event_ts)` would be looked up
// as a *column of that name* and fail at write time. Flipping this
// to `true` means teaching the sink to evaluate the transform
// expression per batch and key the partition on the result — at
// which point native gains what SDP structurally cannot have.
partition_transforms: false,
// The executable graph-sink write (`falkordb-node`/`falkordb-edge`
// Sink → `UNWIND … MERGE` through `FalkorDbSink`) exists only when
// BOTH the DataFusion engine and the rel2graph projection are
// compiled in — the exact `all(feature = "native", feature =
// "rel2graph")` gate `graph_sink::write_graph_sinks` lives behind.
// Without them there is no code to run a graph load, so `check`
// refuses one honestly rather than materialising datasets it cannot
// sink.
graph_sinks: cfg!(all(feature = "native", feature = "rel2graph")),
output_modes: vec!["append".into(), "update".into(), "complete".into()],
}
}
fn run(&self, pipeline: &Pipeline) -> Result<Self::Run> {
self.check(pipeline)?;
pipeline.validate()?;
#[cfg(feature = "native")]
{
exec::run_pipeline(self, pipeline)
}
#[cfg(not(feature = "native"))]
{
// Without the DataFusion tree there is no engine to run on. This is
// an honest capability gap (not a scaffold TODO): the IR is fully
// validated above, but execution needs `--features native`.
crate::functional_status(
"knut-thund/backend_native",
"execute",
false,
"native feature disabled: DataFusion engine not compiled in",
);
Err(ThundError::Backend(
"native backend requires the `native` feature (DataFusion engine not compiled in)"
.into(),
))
}
}
}
/// A handle to a native-backend run.
///
/// The batch dataflow runs to completion eagerly, so all [`RunEvent`]s are
/// queued when the handle is returned; [`RunHandle::poll_events`] drains them
/// once, then reports the run as over. The produced dataset batches are
/// retrievable with [`NativeRun`]`::output` (feature `native`).
#[derive(Debug, Default)]
pub struct NativeRun {
/// Lifecycle + metrics events, in emission order. Drained by `poll_events`.
events: std::collections::VecDeque<RunEvent>,
/// The materialised output of each dataset, keyed by name. Present with the
/// `native` feature. For a streaming dataset this is the final full result
/// over all micro-batches seen (equal to the batch path over the same data).
#[cfg(feature = "native")]
outputs: std::collections::BTreeMap<String, Vec<RecordBatch>>,
/// Per-trigger incremental output batches for each streaming dataset, in
/// trigger (micro-batch) order. Empty for a batch dataset.
#[cfg(feature = "native")]
increments: std::collections::BTreeMap<String, Vec<Vec<RecordBatch>>>,
/// Windows closed-and-evicted per streaming dataset (incremental windowed
/// path only — design phase 2). Each closes exactly once.
#[cfg(feature = "native")]
windows_closed: std::collections::BTreeMap<String, usize>,
/// The open-window state size (rows still buffered because their window is
/// not yet closed) recorded after each trigger, in trigger order. The
/// bounded-memory property lives here: on eviction the series drops, unlike
/// the recompute path's monotonically growing working set.
#[cfg(feature = "native")]
window_open_series: std::collections::BTreeMap<String, Vec<usize>>,
/// Barrier/epoch checkpoints committed per streaming dataset (incremental
/// path with a `state_root` — design phase 3). One per trigger processed.
#[cfg(feature = "native")]
checkpoints_committed: std::collections::BTreeMap<String, usize>,
/// For a run that RESUMED from a checkpoint (design phase 4), the epoch it
/// restored from, per dataset. Absent for a cold (from-scratch) run.
#[cfg(feature = "native")]
resumed_from_epoch: std::collections::BTreeMap<String, u64>,
/// Whether this run used the incremental **aggregation** operator (design
/// phase 5 — maintained accumulators, no per-window SQL re-run) for a
/// dataset, vs the recompute-per-window fallback.
#[cfg(feature = "native")]
incremental_agg: std::collections::BTreeMap<String, bool>,
/// On-disk file sinks written for a dataset (requested via
/// [`NativeBackend::with_file_output`]), keyed by dataset name. Records what
/// landed on disk — the path, format, and row count — so the write side is
/// observable from the return value (no need to re-read the file to know it
/// was written).
#[cfg(feature = "native")]
sinks: std::collections::BTreeMap<String, SinkWrite>,
/// Scan **pushdowns** proven for a flow that carried a typed
/// [`crate::ir::Flow::projection`] / [`crate::ir::Flow::filter`], keyed by
/// flow name. Records the columns the scan was pruned to, the pushed filter
/// predicate, and the optimized physical-plan text — so a caller (and the
/// test matrix) can assert the projection/filter reached the *source* (only
/// the listed columns/rows are read) rather than being applied after a full
/// scan. Empty for flows with no typed pushdown node.
#[cfg(feature = "native")]
pushdowns: std::collections::BTreeMap<String, Pushdown>,
/// Graph-sink writes constructed for a run, keyed by dataset name — one per
/// materialised `falkordb-node` / `falkordb-edge` graph-sink dataset. Records
/// the element (node label / edge rel type), whether it is a node or edge,
/// the row count, and whether the rows were actually MERGEd into a live
/// FalkorDB (`written = true`, needs a [`NativeBackend::with_falkordb_sink`]
/// target) or only constructed (`written = false`, no live target). Present
/// only with `native` + `rel2graph`.
#[cfg(all(feature = "native", feature = "rel2graph"))]
graph_sinks: std::collections::BTreeMap<String, GraphSinkWrite>,
}
/// What a single graph-sink dataset resolved to (features `native` +
/// `rel2graph`): the mapping of a `falkordb-node` / `falkordb-edge`
/// [`OutputType::Sink`] dataset onto a `graphar-falkordb` / `knut-bifrost`
/// `UNWIND … MERGE` load. Returned from [`NativeRun::graph_sink`] so a caller
/// (and the seam test) can assert — from the run handle — exactly which graph
/// element was written, how many rows, and whether it hit a live FalkorDB.
#[cfg(all(feature = "native", feature = "rel2graph"))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GraphSinkWrite {
/// The node label (node sink) or edge relationship type (edge sink).
pub element: String,
/// Whether this dataset is a node sink or an edge sink.
pub kind: graph_sink::GraphElem,
/// Rows in the materialised batch handed to the MERGE load.
pub rows: usize,
/// `true` when the rows were MERGEd into a live FalkorDB (a
/// [`NativeBackend::with_falkordb_sink`] target was set and the load
/// succeeded); `false` for construction-only (no live target).
pub written: bool,
}
/// Proof that a flow's typed [`crate::ir::Flow::projection`] /
/// [`crate::ir::Flow::filter`] was lowered to a **source scan pushdown**
/// (feature `native`). Returned from [`NativeRun::pushdown`] so a caller can
/// assert column pruning / filter pushdown happened at the scan.
#[cfg(feature = "native")]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Pushdown {
/// The columns the scan was pruned to (the projection pushed into the
/// source), in output order. Empty when the flow declared no projection
/// (all columns read).
pub projected_columns: Vec<String>,
/// The filter predicate pushed toward the scan, if any.
pub filter: Option<String>,
/// The optimized physical-plan text (`displayable(...).indent`). For a file
/// source this carries the scan node's `projection=[…]` (and `predicate=…`)
/// — the on-disk proof that pruning/filtering reached the source, not a
/// post-scan `ProjectionExec`/`FilterExec`.
pub physical_plan: String,
}
/// What a single on-disk file sink wrote (feature `native`): the resolved file
/// path, the format it was encoded in, and how many rows landed. Returned from
/// [`NativeRun::sink`] so a caller can assert the write from the run handle.
#[cfg(feature = "native")]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SinkWrite {
/// The path that was written (with any `file://` prefix stripped). A single
/// file for an unpartitioned sink; the **directory** root of a Hive layout
/// (`col=val/…/part.parquet`) when `partition_cols` is non-empty.
pub path: String,
/// The encoding used: `"parquet"` or `"csv"`.
pub format: String,
/// Total rows written across the dataset's materialised batches.
pub rows: usize,
/// The partition columns the output was partitioned by (empty for a plain
/// single-file sink). When non-empty, `path` is a directory and the rows are
/// laid out Hive-style, one subdirectory per distinct partition-value tuple —
/// honoring the dataset's declared [`crate::ir::Dataset::partition_cols`].
pub partition_cols: Vec<String>,
}
impl NativeRun {
/// The materialised batches for dataset `name`, if it was produced.
#[cfg(feature = "native")]
pub fn output(&self, name: &str) -> Option<&[RecordBatch]> {
self.outputs.get(name).map(|v| v.as_slice())
}
/// Total rows materialised for dataset `name` (0 if not produced).
#[cfg(feature = "native")]
pub fn row_count(&self, name: &str) -> usize {
self.outputs
.get(name)
.map(|b| b.iter().map(|rb| rb.num_rows()).sum())
.unwrap_or(0)
}
/// The per-trigger incremental output batches emitted for streaming dataset
/// `name`, in micro-batch order. Empty slice for a batch dataset or one that
/// produced nothing.
#[cfg(feature = "native")]
pub fn increments(&self, name: &str) -> &[Vec<RecordBatch>] {
self.increments
.get(name)
.map(|v| v.as_slice())
.unwrap_or(&[])
}
/// Number of streaming triggers (micro-batches) that emitted for `name`.
#[cfg(feature = "native")]
pub fn trigger_count(&self, name: &str) -> usize {
self.increments.get(name).map(|v| v.len()).unwrap_or(0)
}
/// Windows that closed-and-evicted for streaming dataset `name` (incremental
/// windowed path, design phase 2). Each window closes exactly once; `0` for
/// a non-windowed or recompute run.
#[cfg(feature = "native")]
pub fn windows_closed(&self, name: &str) -> usize {
self.windows_closed.get(name).copied().unwrap_or(0)
}
/// The open-window state size (buffered rows whose window has not yet
/// closed) recorded after each trigger, in trigger order. Bounded-memory
/// proof: this series drops when a window is evicted, in contrast to the
/// recompute path whose accumulated working set only grows. Empty for a
/// non-windowed or recompute run.
#[cfg(feature = "native")]
pub fn window_open_series(&self, name: &str) -> &[usize] {
self.window_open_series
.get(name)
.map(|v| v.as_slice())
.unwrap_or(&[])
}
/// The peak open-window state size across the run for `name` (the high-water
/// mark of buffered, not-yet-closed rows). Stays bounded on the incremental
/// windowed path; `0` for a non-windowed or recompute run.
#[cfg(feature = "native")]
pub fn peak_open_rows(&self, name: &str) -> usize {
self.window_open_series
.get(name)
.and_then(|v| v.iter().copied().max())
.unwrap_or(0)
}
/// Barrier/epoch checkpoints committed for streaming dataset `name` (design
/// phase 3) — one per trigger processed on the incremental path with a
/// `state_root`. `0` for a recompute run or one without a state root.
#[cfg(feature = "native")]
pub fn checkpoints_committed(&self, name: &str) -> usize {
self.checkpoints_committed.get(name).copied().unwrap_or(0)
}
/// If this run RESUMED dataset `name` from a durable checkpoint (design
/// phase 4), the committed epoch it restored from; `None` for a cold run.
#[cfg(feature = "native")]
pub fn resumed_from_epoch(&self, name: &str) -> Option<u64> {
self.resumed_from_epoch.get(name).copied()
}
/// Whether dataset `name` was produced by the incremental **aggregation**
/// operator (design phase 5 — maintained accumulators, no per-window SQL
/// re-run). `false` when it used the recompute-per-window fallback.
#[cfg(feature = "native")]
pub fn used_incremental_agg(&self, name: &str) -> bool {
self.incremental_agg.get(name).copied().unwrap_or(false)
}
/// The on-disk file sink written for dataset `name`, if one was requested via
/// [`NativeBackend::with_file_output`] and its dataset materialised. Carries
/// the resolved path, format, and row count — so the write is observable from
/// the run handle without re-reading the file. `None` when no sink was
/// declared for `name` (or its dataset produced nothing).
#[cfg(feature = "native")]
pub fn sink(&self, name: &str) -> Option<&SinkWrite> {
self.sinks.get(name)
}
/// The scan [`Pushdown`] proven for flow `flow_name`, if it carried a typed
/// [`crate::ir::Flow::projection`] / [`crate::ir::Flow::filter`] node. `None`
/// for a flow with only an opaque query (or no pushdown). Lets a caller
/// assert — from the run handle — that column pruning / filter pushdown
/// reached the source scan.
#[cfg(feature = "native")]
pub fn pushdown(&self, flow_name: &str) -> Option<&Pushdown> {
self.pushdowns.get(flow_name)
}
/// The [`GraphSinkWrite`] resolved for graph-sink dataset `name`, if it was a
/// `falkordb-node` / `falkordb-edge` sink that materialised. Carries the
/// element (node label / edge rel), row count, and whether it was MERGEd into
/// a live FalkorDB — so the graph load is observable from the run handle
/// without querying the database. Present only with `native` + `rel2graph`.
#[cfg(all(feature = "native", feature = "rel2graph"))]
pub fn graph_sink(&self, name: &str) -> Option<&GraphSinkWrite> {
self.graph_sinks.get(name)
}
}
impl RunHandle for NativeRun {
fn poll_events(&mut self) -> Result<Vec<RunEvent>> {
Ok(self.events.drain(..).collect())
}
fn cancel(&mut self) -> Result<()> {
Ok(())
}
}
/// Build one [`RunEvent`] with an RFC3339 `now` timestamp.
///
/// Only called from the `native` execution paths; without that feature the
/// helper (and its `RunPhase` import) stay compiled but unused, hence the
/// conditional allow rather than a hard `cfg` that would strip the import.
#[cfg_attr(not(feature = "native"), allow(dead_code))]
fn event(element: Option<&str>, phase: Option<RunPhase>, message: impl Into<String>) -> RunEvent {
RunEvent {
timestamp: Some(chrono::Utc::now().to_rfc3339()),
element: element.map(str::to_string),
message: message.into(),
phase,
}
}
/// The **FalkorDB graph-sink** write path (features `native` + `rel2graph`).
///
/// This is the executable second projection of the rel2graph two-projections
/// invariant: where [`crate::rel2graph::to_thund_pipeline`] emits graph node/edge
/// datasets as [`OutputType::Sink`]s carrying `knut.node.*` / `knut.edge.*`
/// properties (labels, merge keys, edge directions), THIS module turns a
/// materialised such dataset into a real `UNWIND … MERGE` graph load. It maps the
/// thund IR + the dataset's Arrow batches onto the `knut-bifrost`
/// [`GraphMutation`] contract and hands it to the EXISTING
/// [`knut_bifrost::sink::FalkorDbSink`] (built on `graphar-falkordb`), which
/// MERGEs vertices-before-edges, both endpoints, keyed on the merge key — exactly
/// the idempotent Cypher the PySpark projection emits (`knut_bifrost::codegen`).
/// No Cypher is reimplemented here.
///
/// The `Dataset → GraphMutation` mapping is a **pure** function
/// ([`node_vertex_set`] / [`edge_edge_set`] / [`graph_elem`]) so the seam — given
/// a Sink dataset + `knut.*` props → the expected vertex/edge sets — is testable
/// without a live database.
#[cfg(all(feature = "native", feature = "rel2graph"))]
pub mod graph_sink {
use std::sync::Arc;
use datafusion::arrow::array::{Array, ArrayRef, RecordBatch, StringArray};
use datafusion::arrow::compute::{cast, concat_batches};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use knut_bifrost::{EdgeSet, GraphMutation, VertexSet};
use super::{GraphSinkWrite, NativeBackend, NativeRun, event};
use crate::error::{Result, ThundError};
use crate::ir::{Dataset, OutputType, Pipeline};
use crate::rel2graph::prop;
/// **Optional** sink-side enrichment property keys, layered ON TOP of the
/// [`crate::rel2graph::prop`] keys the projection emits. Every one is
/// additive: absent ⇒ the historical single-key / passthrough behaviour, so
/// the default rel2graph projection (and the `two_projections_agree`
/// invariant) is untouched. Present ⇒ the graph sink honours the richer
/// contract. Defined HERE (the sink), not in `rel2graph`, because they are a
/// backend capability a caller opts a Sink dataset into — the projection
/// never has to emit them.
pub mod ext {
/// Composite node MERGE key: comma-joined source columns whose combined
/// value identifies the node. When set it OVERRIDES the single
/// [`super::prop::NODE_KEY`]; the sink synthesises a deterministic
/// surrogate key column (see [`NODE_KEY_COL`] / [`NODE_KEY_HASH`]) and
/// MERGEs on it, while the source columns ride along as properties.
pub const NODE_KEYS: &str = "knut.node.keys";
/// Name of the synthesised composite-surrogate key column (default
/// `knut_key`). Must not already exist in the node batch.
pub const NODE_KEY_COL: &str = "knut.node.key_col";
/// `true`/`fnv`/`hash`/`1` ⇒ the composite surrogate is a 64-bit FNV-1a
/// HASH (hex) of the joined key values, rather than the raw joined
/// string. An optional hashed surrogate key (standards open-Q #3).
pub const NODE_KEY_HASH: &str = "knut.node.key_hash";
/// Parallel / property-keyed edges: comma-joined edge PROPERTY columns
/// folded into the relationship MERGE key, so distinct property
/// combinations survive as parallel edges (standards open-Q #4). Names
/// must be property columns (not the endpoint keys).
pub const EDGE_KEYS: &str = "knut.edge.keys";
/// Typed projection of node properties: `col:type,col:type,…`
/// (standards open-Q #7). Each named column is CAST to the target Arrow
/// type before the load. Types: `int64`/`long`, `int32`/`int`,
/// `float64`/`double`, `float32`/`float`, `utf8`/`string`/`str`,
/// `bool`/`boolean`, `date32`.
pub const NODE_CAST: &str = "knut.node.cast";
/// Typed projection of edge properties — same grammar as [`NODE_CAST`].
pub const EDGE_CAST: &str = "knut.edge.cast";
}
fn backend_err(ctx: &str, e: impl std::fmt::Display) -> ThundError {
ThundError::Backend(format!("{ctx}: {e}"))
}
/// Which graph element a graph-sink dataset targets.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GraphElem {
/// A node/vertex sink (`falkordb-node`, `knut.node.*` props).
Node,
/// An edge/relationship sink (`falkordb-edge`, `knut.edge.*` props).
Edge,
}
/// Classify a dataset as a FalkorDB graph-sink element, or `None` if it is a
/// plain (file/table) dataset. A graph sink is an [`OutputType::Sink`] whose
/// `format` is `falkordb-node` / `falkordb-edge` AND which carries the
/// matching rel2graph property (`knut.node.label` / `knut.edge.rel`). Both
/// conditions are required so a hand-authored Sink with an unrelated format is
/// never mistaken for a graph load.
pub fn graph_elem(ds: &Dataset) -> Option<GraphElem> {
if ds.output_type != OutputType::Sink {
return None;
}
let fmt = ds.format.as_deref()?;
if fmt == "falkordb-node" && ds.properties.contains_key(prop::NODE_LABEL) {
Some(GraphElem::Node)
} else if fmt == "falkordb-edge" && ds.properties.contains_key(prop::EDGE_REL) {
Some(GraphElem::Edge)
} else {
None
}
}
/// Read a required `knut.*` property off a graph-sink dataset, or fail loudly
/// (a graph sink missing its own target property is a projection bug, not a
/// silent no-op).
fn require_prop<'a>(ds: &'a Dataset, key: &str) -> Result<&'a String> {
ds.properties.get(key).ok_or_else(|| {
ThundError::Backend(format!(
"graph sink `{}`: missing required property `{key}`",
ds.name
))
})
}
/// Map a **node** graph-sink dataset + its materialised batch onto a
/// [`VertexSet`]: `label = knut.node.label`, `id_column = knut.node.key`
/// (the MERGE key), batch verbatim. The bifröst sink then MERGEs
/// `(n:label {id_column: r.id_column}) SET n += r` — the same idempotent
/// upsert the PySpark `_write_nodes` projection emits.
///
/// Two optional enrichments layer on top (both absent ⇒ historical
/// behaviour): [`ext::NODE_CAST`] typed-casts named property columns before
/// the load, and [`ext::NODE_KEYS`] switches the MERGE key from the single
/// [`prop::NODE_KEY`] column to a synthesised composite/hashed surrogate.
pub fn node_vertex_set(ds: &Dataset, batch: RecordBatch) -> Result<VertexSet> {
let label = require_prop(ds, prop::NODE_LABEL)?.clone();
let batch = maybe_cast(ds, ext::NODE_CAST, batch)?;
let (id_column, batch) = node_merge_key(ds, batch)?;
Ok(VertexSet {
label,
id_column,
batch,
})
}
/// Resolve the node MERGE key + the batch to load under it.
///
/// * No [`ext::NODE_KEYS`] ⇒ the single [`prop::NODE_KEY`] column, batch
/// verbatim (historical behaviour).
/// * [`ext::NODE_KEYS`] present ⇒ a COMPOSITE key: a deterministic surrogate
/// column (raw joined value, or a 64-bit FNV-1a hash when
/// [`ext::NODE_KEY_HASH`] is truthy) is appended to the batch and becomes
/// the MERGE key. The individual key columns stay in the batch as
/// properties, so they remain queryable after the load.
fn node_merge_key(ds: &Dataset, batch: RecordBatch) -> Result<(String, RecordBatch)> {
let Some(keys_csv) = ds.properties.get(ext::NODE_KEYS) else {
return Ok((require_prop(ds, prop::NODE_KEY)?.clone(), batch));
};
let cols = parse_csv(keys_csv);
if cols.is_empty() {
return Err(ThundError::Backend(format!(
"node graph sink `{}`: `{}` is empty (composite key needs ≥1 column)",
ds.name,
ext::NODE_KEYS,
)));
}
let key_col = ds
.properties
.get(ext::NODE_KEY_COL)
.cloned()
.unwrap_or_else(|| "knut_key".to_string());
if batch.schema().index_of(&key_col).is_ok() {
return Err(ThundError::Backend(format!(
"node graph sink `{}`: surrogate key column `{key_col}` already exists \
(set `{}` to a free name)",
ds.name,
ext::NODE_KEY_COL,
)));
}
let hashed = matches!(
ds.properties.get(ext::NODE_KEY_HASH).map(String::as_str),
Some("true" | "fnv" | "hash" | "1")
);
let batch = append_surrogate_key(ds, batch, &cols, &key_col, hashed)?;
Ok((key_col, batch))
}
/// Map an **edge** graph-sink dataset + its materialised batch onto an
/// [`EdgeSet`]. Two distinct roles per endpoint: the VALUE is read from the
/// FK column (`from_col`/`to_col`, `knut.edge.from_col`/`knut.edge.to_col`)
/// and the MATCH property is the node's merge key (`from_key`/`to_key`,
/// `knut.edge.from_key`/`knut.edge.to_key`). bifröst's `EdgeSet` carries the
/// endpoint VALUES in `src`/`dst` columns — so the `from_col` column is
/// renamed to `src` and `to_col` to `dst`, every other column becoming an
/// edge property, while `src_id_column`/`dst_id_column` take the merge keys.
/// These coincide for a plain FK (`order_id` read AND MERGEd) but DIFFER for
/// a self-ref / renamed FK (`manager_id` read, MERGEd on `employee_id`).
/// Byte-for-byte the PySpark projection's `from_col AS src`, `to_col AS dst`,
/// then `MERGE (a:from {from_key: r.src}) MERGE (b:to {to_key: r.dst})
/// MERGE (a)-[e:rel]->(b) SET e += r.props`.
///
/// Two optional enrichments layer on top (both absent ⇒ historical
/// behaviour): [`ext::EDGE_CAST`] typed-casts named property columns, and
/// [`ext::EDGE_KEYS`] folds named property columns into the relationship
/// MERGE key so distinct property combinations survive as PARALLEL edges
/// (instead of collapsing to one edge per endpoint pair).
pub fn edge_edge_set(ds: &Dataset, batch: RecordBatch) -> Result<EdgeSet> {
// MERGE property names — the node keys the endpoints are matched by.
let from_key = require_prop(ds, prop::EDGE_FROM_KEY)?.clone();
let to_key = require_prop(ds, prop::EDGE_TO_KEY)?.clone();
// VALUE columns — the FK columns the endpoint values are read from. For a
// self-ref / renamed FK these differ from the merge keys (`manager_id`
// read, MERGEd on `employee_id`). Fall back to the merge key when a
// pre-`from_col` spec omits them (older projections, hand-authored specs)
// — historically identical whenever the FK column == the node key.
let from_col = ds
.properties
.get(prop::EDGE_FROM_COL)
.cloned()
.unwrap_or_else(|| from_key.clone());
let to_col = ds
.properties
.get(prop::EDGE_TO_COL)
.cloned()
.unwrap_or_else(|| to_key.clone());
let batch = maybe_cast(ds, ext::EDGE_CAST, batch)?;
// Rename the VALUE columns to the `src`/`dst` carriers bifröst's EdgeSet
// requires — the endpoint values come from the FK columns.
let batch = rename_endpoint_columns(ds, &batch, &from_col, &to_col)?;
let edge_key_props = edge_merge_props(ds, &batch)?;
Ok(EdgeSet {
src_label: require_prop(ds, prop::EDGE_FROM)?.clone(),
edge_type: require_prop(ds, prop::EDGE_REL)?.clone(),
dst_label: require_prop(ds, prop::EDGE_TO)?.clone(),
// The MATCH property names are the endpoint node MERGE KEYS — the
// property each endpoint node is identified by, which for a self-ref
// is the node key (`employee_id`), never the FK read column.
src_id_column: from_key,
dst_id_column: to_key,
edge_key_props,
replace_outgoing: false,
batch,
})
}
/// The edge property columns that participate in the relationship MERGE key
/// ([`ext::EDGE_KEYS`]). Empty when unset (one edge per endpoint pair). Each
/// named column must be a PROPERTY column present in the renamed edge batch
/// — never the `src`/`dst` endpoint carriers, and never absent (a merge key
/// naming a column that will not exist is a projection bug, failed loudly).
fn edge_merge_props(ds: &Dataset, renamed: &RecordBatch) -> Result<Vec<String>> {
let Some(keys_csv) = ds.properties.get(ext::EDGE_KEYS) else {
return Ok(Vec::new());
};
let schema = renamed.schema();
let mut out = Vec::new();
for col in parse_csv(keys_csv) {
if col == "src" || col == "dst" {
return Err(ThundError::Backend(format!(
"edge graph sink `{}`: `{}` names endpoint carrier `{col}` — \
the merge key folds PROPERTY columns, not the endpoints",
ds.name,
ext::EDGE_KEYS,
)));
}
if schema.index_of(&col).is_err() {
return Err(ThundError::Backend(format!(
"edge graph sink `{}`: merge-key property `{col}` (`{}`) is not an \
output column",
ds.name,
ext::EDGE_KEYS,
)));
}
out.push(col);
}
Ok(out)
}
/// Rename the edge batch's `from_key` column to `src` and `to_key` to `dst`
/// (the carrier names bifröst's [`EdgeSet`] requires), leaving every other
/// column — the edge properties — untouched. Fails loudly if either endpoint
/// key column is absent (a projection/schema mismatch, not a silent drop).
fn rename_endpoint_columns(
ds: &Dataset,
batch: &RecordBatch,
from_key: &str,
to_key: &str,
) -> Result<RecordBatch> {
let schema = batch.schema();
let (mut seen_src, mut seen_dst) = (false, false);
let mut fields: Vec<Field> = Vec::with_capacity(schema.fields().len());
for f in schema.fields() {
let name = f.name().as_str();
let new_name = if name == from_key {
seen_src = true;
"src"
} else if name == to_key {
seen_dst = true;
"dst"
} else {
name
};
fields.push(Field::new(new_name, f.data_type().clone(), f.is_nullable()));
}
if !seen_src || !seen_dst {
return Err(ThundError::Backend(format!(
"edge graph sink `{}`: endpoint key column(s) not in output schema \
(from_key `{from_key}`{}, to_key `{to_key}`{})",
ds.name,
if seen_src { "" } else { " NOT FOUND" },
if seen_dst { "" } else { " NOT FOUND" },
)));
}
RecordBatch::try_new(Arc::new(Schema::new(fields)), batch.columns().to_vec())
.map_err(|e| backend_err(&format!("edge graph sink `{}` rename", ds.name), e))
}
/// Split a comma-joined property list into trimmed, non-empty names.
fn parse_csv(csv: &str) -> Vec<String> {
csv.split(',')
.map(str::trim)
.filter(|s| !s.is_empty())
.map(str::to_string)
.collect()
}
/// Apply an optional typed-projection cast spec (`col:type,…`) named by
/// `prop_key` to `batch`, or return it unchanged when the property is absent.
/// Typed projection replaces the Arrow-passthrough default (open-Q #7).
fn maybe_cast(ds: &Dataset, prop_key: &str, batch: RecordBatch) -> Result<RecordBatch> {
let Some(spec) = ds.properties.get(prop_key) else {
return Ok(batch);
};
cast_batch(ds, prop_key, spec, batch)
}
/// Parse a `col:type,col:type,…` spec and CAST each named column of `batch`
/// to its target Arrow type. Loud on an unknown type token, a malformed
/// entry, a column not in the batch, or a cast Arrow rejects — a typed
/// projection that cannot be honoured is a contract error, never a silent
/// passthrough.
fn cast_batch(
ds: &Dataset,
prop_key: &str,
spec: &str,
batch: RecordBatch,
) -> Result<RecordBatch> {
let schema = batch.schema();
let mut columns: Vec<ArrayRef> = batch.columns().to_vec();
let mut fields: Vec<Field> = schema.fields().iter().map(|f| f.as_ref().clone()).collect();
for entry in spec.split(',').map(str::trim).filter(|s| !s.is_empty()) {
let (col, ty) = entry.split_once(':').ok_or_else(|| {
ThundError::Backend(format!(
"graph sink `{}`: malformed `{prop_key}` entry `{entry}` (want `col:type`)",
ds.name,
))
})?;
let (col, ty) = (col.trim(), ty.trim());
let idx = schema.index_of(col).map_err(|_| {
ThundError::Backend(format!(
"graph sink `{}`: `{prop_key}` casts unknown column `{col}`",
ds.name,
))
})?;
let target = parse_type_token(ds, prop_key, ty)?;
let casted = cast(columns[idx].as_ref(), &target).map_err(|e| {
backend_err(
&format!("graph sink `{}`: cast column `{col}` to `{ty}`", ds.name),
e,
)
})?;
fields[idx] = Field::new(col, target, fields[idx].is_nullable());
columns[idx] = casted;
}
RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)
.map_err(|e| backend_err(&format!("graph sink `{}` typed projection", ds.name), e))
}
/// Map a cast type token to an Arrow [`DataType`]. A bounded, explicit set —
/// unknown tokens fail loudly rather than defaulting.
fn parse_type_token(ds: &Dataset, prop_key: &str, ty: &str) -> Result<DataType> {
Ok(match ty.to_ascii_lowercase().as_str() {
"int64" | "long" | "bigint" => DataType::Int64,
"int32" | "int" | "integer" => DataType::Int32,
"float64" | "double" | "f64" => DataType::Float64,
"float32" | "float" | "f32" => DataType::Float32,
"utf8" | "string" | "str" | "text" => DataType::Utf8,
"bool" | "boolean" => DataType::Boolean,
"date32" | "date" => DataType::Date32,
other => {
return Err(ThundError::Backend(format!(
"graph sink `{}`: `{prop_key}` has unknown type `{other}`",
ds.name,
)));
}
})
}
/// Append the synthesised composite-surrogate MERGE key column to `batch`.
///
/// The key columns are each cast to `Utf8` and joined per row with a unit
/// separator (`U+001F`) into a canonical string; nulls render as the empty
/// string. When `hashed`, that string is reduced to a 64-bit FNV-1a hash
/// rendered as lowercase hex (a compact surrogate); otherwise the raw joined
/// string is the key. Deterministic and pure, so identity is stable across
/// runs. Fails loudly if a named key column is absent.
fn append_surrogate_key(
ds: &Dataset,
batch: RecordBatch,
key_cols: &[String],
key_col: &str,
hashed: bool,
) -> Result<RecordBatch> {
let schema = batch.schema();
// Each key column, cast to Utf8 for canonical stringification.
let mut string_cols: Vec<StringArray> = Vec::with_capacity(key_cols.len());
for col in key_cols {
let idx = schema.index_of(col).map_err(|_| {
ThundError::Backend(format!(
"node graph sink `{}`: composite key column `{col}` (`{}`) not in output",
ds.name,
ext::NODE_KEYS,
))
})?;
let utf8 = cast(batch.column(idx).as_ref(), &DataType::Utf8)
.map_err(|e| backend_err(&format!("node graph sink `{}` key cast", ds.name), e))?;
let arr = utf8
.as_any()
.downcast_ref::<StringArray>()
.expect("cast to Utf8 yields StringArray")
.clone();
string_cols.push(arr);
}
let n = batch.num_rows();
let mut keys: Vec<String> = Vec::with_capacity(n);
for row in 0..n {
let mut parts: Vec<&str> = Vec::with_capacity(string_cols.len());
for arr in &string_cols {
parts.push(if arr.is_null(row) { "" } else { arr.value(row) });
}
let joined = parts.join("\u{1f}");
keys.push(if hashed {
format!("{:016x}", fnv1a_64(joined.as_bytes()))
} else {
joined
});
}
let key_array: ArrayRef = Arc::new(StringArray::from(keys));
let mut fields: Vec<Field> = schema.fields().iter().map(|f| f.as_ref().clone()).collect();
fields.push(Field::new(key_col, DataType::Utf8, false));
let mut columns: Vec<ArrayRef> = batch.columns().to_vec();
columns.push(key_array);
RecordBatch::try_new(Arc::new(Schema::new(fields)), columns)
.map_err(|e| backend_err(&format!("node graph sink `{}` surrogate key", ds.name), e))
}
/// 64-bit FNV-1a — a small, dependency-free, deterministic hash for the
/// optional hashed composite surrogate key.
fn fnv1a_64(bytes: &[u8]) -> u64 {
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
for &b in bytes {
hash ^= u64::from(b);
hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
}
hash
}
/// Concatenate a dataset's materialised batches into one [`RecordBatch`] for
/// the graph load. Returns `None` for a dataset with no rows (nothing to
/// load) so the caller can skip it honestly.
fn concat(ds_name: &str, batches: &[RecordBatch]) -> Result<Option<RecordBatch>> {
let Some(first) = batches.first() else {
return Ok(None);
};
let schema = first.schema();
let one = concat_batches(&schema, batches.iter())
.map_err(|e| backend_err(&format!("graph sink `{ds_name}` concat"), e))?;
if one.num_rows() == 0 {
Ok(None)
} else {
Ok(Some(one))
}
}
/// Drive the graph-sink write for every materialised `falkordb-node` /
/// `falkordb-edge` dataset in the pipeline.
///
/// Builds ONE [`GraphMutation`] over all node datasets (as vertices) and all
/// edge datasets (as edges), in pipeline (declaration) order — nodes before
/// edges. When a [`NativeBackend::with_falkordb_sink`] target is set the
/// mutation is applied through the existing bifröst [`FalkorDbSink`]
/// (idempotent `UNWIND … MERGE`, vertices strictly before edges); without a
/// target the mutation is still constructed and recorded (construction-only,
/// not faked). Every resolved dataset is recorded on [`NativeRun::graph_sink`].
pub(crate) async fn write_graph_sinks(
backend: &NativeBackend,
pipeline: &Pipeline,
run: &mut NativeRun,
) -> Result<()> {
// Collect graph-sink datasets that actually materialised, mapping each to
// its vertex/edge set. Pipeline order = rel2graph declaration order, so
// nodes precede edges here too.
let mut mutation = GraphMutation::default();
// (dataset name, element kind, element label/rel, rows)
let mut resolved: Vec<(String, GraphElem, String, usize)> = Vec::new();
for ds in &pipeline.datasets {
let Some(elem) = graph_elem(ds) else {
continue;
};
let batches = run.outputs.get(&ds.name).cloned().unwrap_or_default();
let Some(batch) = concat(&ds.name, &batches)? else {
run.events.push_back(event(
Some(&ds.name),
None,
format!(
"graph sink `{}`: dataset produced no rows — nothing to MERGE",
ds.name
),
));
continue;
};
let rows = batch.num_rows();
match elem {
GraphElem::Node => {
let vs = node_vertex_set(ds, batch)?;
resolved.push((ds.name.clone(), elem, vs.label.clone(), rows));
mutation.vertices.push(vs);
}
GraphElem::Edge => {
let es = edge_edge_set(ds, batch)?;
resolved.push((ds.name.clone(), elem, es.edge_type.clone(), rows));
mutation.edges.push(es);
}
}
}
if resolved.is_empty() {
return Ok(());
}
// Apply to a live FalkorDB when a target is configured; otherwise record
// the mutation as constructed-only (honest — never a silent fake).
let written = match backend.falkordb_sink.clone() {
Some((url, graph)) => {
use knut_bifrost::sink::{FalkorDbSink, GraphSink};
let mut sink = FalkorDbSink::connect(&url, graph.as_str())
.await
.map_err(|e| backend_err("connect FalkorDB graph sink", e))?;
let report = sink
.apply(&mutation)
.await
.map_err(|e| backend_err("apply graph mutation to FalkorDB", e))?;
run.events.push_back(event(
None,
None,
format!(
"graph sink → FalkorDB `{graph}`: MERGEd {} vertex + {} edge row(s) \
({} node-set(s), {} edge-set(s), idempotent UNWIND…MERGE)",
report.vertices_written,
report.edges_written,
mutation.vertices.len(),
mutation.edges.len(),
),
));
crate::functional_status(
"knut-thund/native_graph_sink",
"load",
true,
&format!(
"{}: MERGEd {} vertices + {} edges into FalkorDB `{graph}`",
pipeline.name, report.vertices_written, report.edges_written,
),
);
true
}
None => {
run.events.push_back(event(
None,
None,
format!(
"graph sink: constructed {} vertex-set(s) + {} edge-set(s) \
(UNWIND…MERGE ready) but no FalkorDB target — call \
`with_falkordb_sink(url, graph)` to load (construction-only)",
mutation.vertices.len(),
mutation.edges.len(),
),
));
crate::functional_status(
"knut-thund/native_graph_sink",
"construct",
true,
&format!(
"{}: constructed graph mutation ({} node-set(s), {} edge-set(s)); no live target",
pipeline.name,
mutation.vertices.len(),
mutation.edges.len(),
),
);
false
}
};
for (name, kind, element, rows) in resolved {
run.graph_sinks.insert(
name,
GraphSinkWrite {
element,
kind,
rows,
written,
},
);
}
Ok(())
}
}
/// The DataFusion-backed execution engine (feature `native`).
#[cfg(feature = "native")]
mod exec {
use super::{NativeBackend, NativeRun, Pushdown, SinkWrite, StreamingMode, event};
use crate::backend::checkpoint;
use crate::error::{Result, ThundError};
use crate::ir::{
CdcSpec, Flow, FlowKind, OnViolation, OutputMode, Pipeline, ScdType, SourceSpec, Trigger,
Watermark, WindowSpec, WriteMode,
};
use crate::{backend::RunPhase, functional_status};
use datafusion::arrow::array::{Array, Int64Array, RecordBatch};
use datafusion::arrow::datatypes::Schema;
use datafusion::datasource::MemTable;
use datafusion::prelude::{CsvReadOptions, ParquetReadOptions, SessionContext};
use std::collections::VecDeque;
use std::sync::Arc;
fn df_err(ctx: &str, e: impl std::fmt::Display) -> ThundError {
ThundError::Backend(format!("{ctx}: {e}"))
}
/// Run the pipeline's batch dataflow to completion on DataFusion.
///
/// Synchronous entry point (the [`super::ExecBackend`] trait is sync); it
/// spins a current-thread Tokio runtime and blocks on the async engine,
/// which is fine because the batch dataflow is bounded.
pub(super) fn run_pipeline(backend: &NativeBackend, pipeline: &Pipeline) -> Result<NativeRun> {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.map_err(|e| df_err("build tokio runtime", e))?;
rt.block_on(execute(backend, pipeline))
}
async fn execute(backend: &NativeBackend, pipeline: &Pipeline) -> Result<NativeRun> {
let ctx = SessionContext::new();
let mut run = NativeRun::default();
// 1. Seed in-memory input tables (bounded sources supplied by the caller).
for (name, batches) in &backend.seed_inputs {
let schema: Arc<Schema> = batches
.first()
.map(|b| b.schema())
.ok_or_else(|| df_err("seed input", format!("`{name}` has no batches")))?;
let mem = MemTable::try_new(schema, vec![batches.clone()])
.map_err(|e| df_err("seed MemTable", e))?;
ctx.register_table(name.as_str(), Arc::new(mem))
.map_err(|e| df_err("register seed", e))?;
run.events.push_back(event(
Some(name),
None,
format!(
"seeded input `{name}` ({} row(s))",
batches.iter().map(|b| b.num_rows()).sum::<usize>()
),
));
}
// 2. Register on-disk bounded file inputs the caller pointed at
// (`with_file_input`) under their explicit table names — the on-disk
// sibling of the in-memory seeds above, so a plain batch flow can scan
// a local `orders.csv`/`.parquet` directly.
for (name, path, format, partition_cols) in &backend.seed_file_inputs {
register_named_file(&ctx, name, path, format, partition_cols, &mut run.events).await;
}
// 3. Register file-backed bounded sources declared on streaming flows.
for f in &pipeline.flows {
if let FlowKind::Streaming { source, .. } = &f.kind {
register_file_source(&ctx, source, &mut run.events).await;
}
}
run.events.push_back(event(
None,
Some(RunPhase::Planning),
format!(
"native/DataFusion: running `{}` ({} dataset(s), {} flow(s))",
pipeline.name,
pipeline.datasets.len(),
pipeline.flows.len()
),
));
// 3. Execute flows in topological order of their targets so a flow's
// upstream datasets are already materialised when it runs.
let order = pipeline.topo_order().ok_or(ThundError::Cyclic)?;
// Record the selected streaming mode for the test matrix. The mode only
// governs unbounded flows; recompute is the default and the fallback.
functional_status(
"knut-thund/backend_native",
"streaming_mode",
true,
&format!(
"{}: streaming mode = {:?} (state_root {})",
pipeline.name,
backend.streaming_mode,
if backend.state_root.is_some() {
"set"
} else {
"unset"
}
),
);
let mut ran = 0usize;
for target in &order {
// Multi-flow **fan-in (union append)**: the IR permits more than one
// flow to write the same target — SDP's canonical "streaming table +
// N append flows" pattern, where each flow reads its own
// independently-authored source and their rows are UNIONED into one
// asset. `run_flow`/`materialize` register the freshest flow's rows
// under the target (clobbering); here, once a prior flow has already
// produced this target in this run, we splice the freshest flow's
// output onto the accumulated union and re-materialise, so downstream
// flows and the sink see every append flow's rows — not just the last.
// Single-flow-per-target (the common case) never enters the union path
// (`target_produced` stays false) and is byte-for-byte unchanged.
let mut target_produced = false;
for f in pipeline.flows.iter().filter(|f| &f.target == target) {
// Snapshot the union accumulated by prior flows for this target
// BEFORE this flow clobbers it inside `run_flow`.
let prior: Option<Vec<RecordBatch>> = if target_produced {
run.outputs.get(target).cloned()
} else {
None
};
match run_flow(
&ctx,
f,
&backend.seed_streams,
backend.streaming_mode,
backend.state_root.as_deref(),
&pipeline.name,
backend.max_triggers,
&mut run,
)
.await
{
Ok(true) => {
ran += 1;
if let Some(prior) = prior {
union_append(&ctx, target, prior, &f.name, &pipeline.name, &mut run)?;
}
target_produced = true;
}
Ok(false) => {}
Err(e) => {
// A genuine execution failure (planning/exec error on a
// batch flow, or a FAIL-UPDATE expectation): fail the
// whole run honestly.
run.events.push_back(event(
Some(&f.target),
Some(RunPhase::Failed),
format!("flow `{}` failed: {e}", f.name),
));
functional_status(
"knut-thund/backend_native",
"execute",
false,
&format!("{}: flow `{}` failed: {e}", pipeline.name, f.name),
);
return Err(e);
}
}
}
}
// 4. Write any declared on-disk file sinks (the write-side sibling of
// `with_file_input`): land each materialised dataset as a single local
// parquet/csv file, completing the read-file → SQL → write-file round
// trip a laptop consumer wants. A sink whose dataset produced no rows
// is reported honestly (nothing written) rather than emitting an empty
// file silently.
for (dataset, path, format) in &backend.seed_file_outputs {
// A sink dataset may declare `partition_cols` — honor them (a
// Hive-partitioned directory write) instead of silently dropping
// them. Unpartitioned (the default) still lands a single file. The
// dataset's declared `write_mode` (default `Append`) drives
// clear-vs-accumulate too, so an imperative `with_file_output` honors
// a `Dataset::overwrite()` declared in the IR.
let (partition_cols, write_mode) = pipeline
.datasets
.iter()
.find(|d| &d.name == dataset)
.map(|d| (d.partition_cols.clone(), d.write_mode))
.unwrap_or_default();
write_file_sink(
dataset,
path,
format,
&partition_cols,
write_mode,
&pipeline.name,
&mut run,
)
.await?;
}
// 4b. Write any file sinks declared in the IR itself — a dataset that
// carries a `path` (via `Dataset::with_path`) is a declarative file
// sink, so a pipeline authored purely as IR/RON lands its outputs
// with no imperative `with_file_output` call. The imperative sinks
// above take precedence: a dataset already named there is skipped
// here (its imperative path/format wins, no double write). Reuses the
// same `write_file_sink` (single file, or Hive dir when partitioned).
let imperative: std::collections::BTreeSet<&str> = backend
.seed_file_outputs
.iter()
.map(|(d, _, _)| d.as_str())
.collect();
for d in &pipeline.datasets {
let Some(path) = d.path.as_deref() else {
continue;
};
if imperative.contains(d.name.as_str()) {
continue;
}
// Format: the dataset's declared `format`, else inferred from the
// path extension (`.csv` → csv), else parquet.
let format = d
.format
.clone()
.unwrap_or_else(|| infer_file_format(path).to_string());
write_file_sink(
&d.name,
path,
&format,
&d.partition_cols,
d.write_mode,
&pipeline.name,
&mut run,
)
.await?;
}
// 4c. Graph sinks (features `native` + `rel2graph`): a dataset that is an
// `OutputType::Sink` with a `falkordb-node` / `falkordb-edge` format
// carrying the rel2graph `knut.node.*` / `knut.edge.*` props is a
// GRAPH sink, not a file/Parquet sink. Turn each materialised one into
// an idempotent `UNWIND … MERGE` graph load driven through the
// existing `knut_bifrost::sink::FalkorDbSink` (built on
// `graphar-falkordb`) — nodes-before-edges, both endpoints MERGEd,
// keyed on the merge key. This is the executable half of "bifröst
// mapping → thund → NativeBackend → FalkorDB". Reuses graphar-falkordb;
// no Cypher is reimplemented here. With no live target configured the
// mutation is constructed + recorded (honest construction-only), not
// faked.
#[cfg(feature = "rel2graph")]
super::graph_sink::write_graph_sinks(backend, pipeline, &mut run).await?;
run.events.push_back(event(
None,
Some(RunPhase::Completed),
format!(
"native/DataFusion: `{}` complete ({ran} flow(s) executed)",
pipeline.name
),
));
// The native execution backend genuinely ran the batch dataflow → GREEN.
functional_status(
"knut-thund/backend_native",
"execute",
true,
&format!("{}: {ran} flow(s) executed on DataFusion", pipeline.name),
);
Ok(run)
}
/// Register a bounded local file source (`parquet`/`csv`) so a flow query
/// can read it. Kafka/CDC/remote sources need infra and are left for the
/// streaming layer; unregistered names surface as a planning error later.
async fn register_file_source(
ctx: &SessionContext,
source: &SourceSpec,
events: &mut VecDeque<super::RunEvent>,
) {
let (name, path, format) = match source {
SourceSpec::Batch { uri, format } => {
(source_table_name(uri), uri.clone(), format.clone())
}
SourceSpec::FileDrop { dir, format } => {
(source_table_name(dir), dir.clone(), format.clone())
}
// Unbounded / changelog sources are not runnable without infra.
_ => return,
};
register_named_file(ctx, &name, &path, &format, &[], events).await;
}
/// Register a local `parquet`/`csv` file under an explicit `table` name so a
/// flow query can scan it. Shared by the streaming file-source path
/// ([`register_file_source`]) and the caller-supplied on-disk batch inputs
/// ([`NativeBackend::with_file_input`]) so both register files exactly once,
/// the same way. `path` may carry a `file://` prefix (stripped).
///
/// With a non-empty `partition_cols` the `path` is treated as the ROOT of a
/// **Hive-partitioned directory** (`col=val/…`) and each partition column is
/// recovered from the path segments as a `Utf8` string — the read side of the
/// partitioned file sink. An empty `partition_cols` (the common case)
/// registers a plain single file exactly as before.
async fn register_named_file(
ctx: &SessionContext,
table: &str,
path: &str,
format: &str,
partition_cols: &[String],
events: &mut VecDeque<super::RunEvent>,
) {
use datafusion::arrow::datatypes::DataType;
let path = path.strip_prefix("file://").unwrap_or(path).to_string();
// Hive partition values are path segments → recovered as Utf8 strings.
let part_cols: Vec<(String, DataType)> = partition_cols
.iter()
.map(|c| (c.clone(), DataType::Utf8))
.collect();
let res = match format {
"parquet" => {
let opts = ParquetReadOptions::default().table_partition_cols(part_cols.clone());
ctx.register_parquet(table, &path, opts).await
}
"csv" => {
let opts = CsvReadOptions::default().table_partition_cols(part_cols.clone());
ctx.register_csv(table, &path, opts).await
}
other => {
events.push_back(event(
Some(table),
None,
format!(
"source `{table}`: format `{other}` not registerable (need parquet/csv)"
),
));
return;
}
};
match res {
Ok(()) => events.push_back(event(
Some(table),
None,
if partition_cols.is_empty() {
format!("registered bounded source `{table}` <- {path}")
} else {
format!(
"registered partitioned source `{table}` <- {path} (partitioned by {partition_cols:?})"
)
},
)),
Err(e) => events.push_back(event(
Some(table),
None,
format!("source `{table}` not registered ({e}); flow will defer"),
)),
}
}
/// Infer a file format from a path's extension for a declarative sink whose
/// dataset did not name one: `.csv` → `csv`, everything else (incl. a
/// partition-directory root with no extension) → `parquet`, the native
/// default write format.
fn infer_file_format(path: &str) -> &'static str {
let ext = std::path::Path::new(path)
.extension()
.and_then(|e| e.to_str())
.unwrap_or("");
if ext.eq_ignore_ascii_case("csv") {
"csv"
} else {
"parquet"
}
}
/// On [`WriteMode::Overwrite`], clear an already-populated sink destination
/// before writing so a re-run holds exactly one generation's rows
/// (fresh-dir / fresh-file semantics). A partitioned directory sink otherwise
/// *accumulates* new part-files on every run (the append bug the
/// partitioned-sink problem entry flagged); a single file is truncated by the
/// writer anyway, but a stale directory left at the same path is also removed
/// so switching a sink's shape is clean. On [`WriteMode::Append`] (the
/// default) this is a no-op — the pre-existing accumulate behaviour, so
/// existing pipelines are byte-for-byte unchanged. A destination that does
/// not exist yet is left untouched (nothing to clear).
fn clear_sink_target(path: &str, mode: WriteMode) -> Result<()> {
if mode != WriteMode::Overwrite {
return Ok(());
}
let p = std::path::Path::new(path);
match std::fs::metadata(p) {
Ok(m) if m.is_dir() => {
std::fs::remove_dir_all(p).map_err(|e| df_err("overwrite: clear sink dir", e))?;
}
Ok(_) => {
std::fs::remove_file(p).map_err(|e| df_err("overwrite: clear sink file", e))?;
}
// Nothing at the destination yet — a fresh write, nothing to clear.
Err(_) => {}
}
Ok(())
}
/// Write a materialised dataset's rows on-disk — the write side of
/// [`register_named_file`] / `with_file_input`. With no `partition_cols` this
/// writes a **single** `parquet`/`csv` file; with `partition_cols` it writes a
/// Hive-partitioned **directory** (`col=val/…`) via [`write_partitioned_sink`]
/// — honoring the dataset's declared partitioning instead of silently
/// ignoring it. Reads the produced batches from `run.outputs`; a dataset that
/// produced no rows is reported honestly (nothing written) rather than
/// dropping an empty file. Records the write in `run.sinks` and emits a
/// `native_file_sink` matrix cell so the write is observable from outside.
/// Creates the parent/target directory if absent so a consumer may name
/// `out/rollup.parquet` (or a `region=…/` layout root) freely.
async fn write_file_sink(
dataset: &str,
path: &str,
format: &str,
partition_cols: &[String],
write_mode: WriteMode,
pipeline_name: &str,
run: &mut NativeRun,
) -> Result<()> {
let clean = path.strip_prefix("file://").unwrap_or(path).to_string();
let Some(batches) = run.outputs.get(dataset).cloned() else {
run.events.push_back(event(
Some(dataset),
None,
format!("sink `{dataset}`: dataset not materialised — nothing written to {clean}"),
));
functional_status(
"knut-thund/native_file_sink",
"write",
false,
&format!("{pipeline_name}: sink `{dataset}` had no materialised rows"),
);
return Ok(());
};
let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
// Declarative WRITE MODE: on `Overwrite`, clear an already-populated
// destination BEFORE writing so a re-run holds exactly one generation's
// rows (fresh-dir semantics). This is the fix for the append bug the
// partitioned-sink entry flagged: a Hive directory otherwise accumulates
// new part-files on every run. On `Append` (the default) this is a no-op
// — the pre-existing accumulate behaviour, so existing pipelines are
// byte-for-byte unchanged (L2-additive).
clear_sink_target(&clean, write_mode)?;
// Partitioned sink: the dataset declared partition columns, so write a
// Hive-partitioned DIRECTORY (`col=val/…/part.parquet`) rather than a
// single file. Additive — an empty `partition_cols` (the default) falls
// straight through to the single-file writer below, byte-for-byte.
if !partition_cols.is_empty() {
return write_partitioned_sink(
dataset,
&clean,
format,
partition_cols,
&batches,
rows,
pipeline_name,
run,
)
.await;
}
if let Some(parent) = std::path::Path::new(&clean).parent() {
if !parent.as_os_str().is_empty() {
std::fs::create_dir_all(parent).map_err(|e| df_err("sink parent dir", e))?;
}
}
let schema = batches
.first()
.map(|b| b.schema())
.unwrap_or_else(|| Arc::new(Schema::empty()));
match format {
"parquet" => {
use datafusion::parquet::arrow::ArrowWriter;
let file =
std::fs::File::create(&clean).map_err(|e| df_err("create sink parquet", e))?;
let mut w = ArrowWriter::try_new(file, schema, None)
.map_err(|e| df_err("sink parquet writer", e))?;
for b in &batches {
w.write(b).map_err(|e| df_err("write sink parquet", e))?;
}
w.close().map_err(|e| df_err("close sink parquet", e))?;
}
"csv" => {
use datafusion::arrow::csv::WriterBuilder;
let file =
std::fs::File::create(&clean).map_err(|e| df_err("create sink csv", e))?;
let mut w = WriterBuilder::new().with_header(true).build(file);
for b in &batches {
w.write(b).map_err(|e| df_err("write sink csv", e))?;
}
}
other => {
run.events.push_back(event(
Some(dataset),
None,
format!("sink `{dataset}`: format `{other}` not writable (need parquet/csv)"),
));
functional_status(
"knut-thund/native_file_sink",
"write",
false,
&format!("{pipeline_name}: sink `{dataset}` format `{other}` unsupported"),
);
return Ok(());
}
}
run.sinks.insert(
dataset.to_string(),
SinkWrite {
path: clean.clone(),
format: format.to_string(),
rows,
partition_cols: Vec::new(),
},
);
run.events.push_back(event(
Some(dataset),
None,
format!("sink `{dataset}`: wrote {rows} row(s) -> {clean} ({format})"),
));
functional_status(
"knut-thund/native_file_sink",
"write",
rows > 0,
&format!("{pipeline_name}: sink `{dataset}` wrote {rows} row(s) to {clean} ({format})"),
);
Ok(())
}
/// Write a materialised dataset to a Hive-partitioned **directory**
/// (`col=val/…/part.parquet`) when it declares `partition_cols`, via
/// DataFusion's partitioned writer — the write side of the
/// `table_partition_cols` read path, and what makes the previously-inert
/// [`crate::ir::Dataset::partition_cols`] field real. Every declared
/// partition column must exist in the produced schema; a missing one fails
/// **loudly** (mirrors the file-source loud-gap law) rather than dropping the
/// partitioning silently. Records the write (path = the directory root,
/// `partition_cols` populated) so it is observable from [`NativeRun::sink`].
async fn write_partitioned_sink(
dataset: &str,
dir: &str,
format: &str,
partition_cols: &[String],
batches: &[RecordBatch],
rows: usize,
pipeline_name: &str,
run: &mut NativeRun,
) -> Result<()> {
use datafusion::dataframe::DataFrameWriteOptions;
let schema = batches
.first()
.map(|b| b.schema())
.unwrap_or_else(|| Arc::new(Schema::empty()));
// Loud gap: a declared partition column absent from the output is an
// error, not a silent no-op (a consumer would otherwise get an
// unpartitioned dump under a name that promised partitioning).
for pc in partition_cols {
if schema.field_with_name(pc).is_err() {
let msg = format!(
"partitioned sink `{dataset}`: partition column `{pc}` not in output schema"
);
run.events
.push_back(event(Some(dataset), None, msg.clone()));
functional_status(
"knut-thund/native_file_sink",
"partitioned",
false,
&format!("{pipeline_name}: {msg}"),
);
return Err(df_err("partitioned sink", msg));
}
}
std::fs::create_dir_all(dir).map_err(|e| df_err("sink partition dir", e))?;
// A fresh context turns the already-materialised batches back into a
// DataFrame purely to drive the partitioned writer; it never touches the
// run's own catalog.
let ctx = SessionContext::new();
let df = ctx
.read_batches(batches.iter().cloned())
.map_err(|e| df_err("partitioned sink read_batches", e))?;
let opts = DataFrameWriteOptions::new().with_partition_by(partition_cols.to_vec());
match format {
"parquet" => {
df.write_parquet(dir, opts, None)
.await
.map_err(|e| df_err("write partitioned parquet", e))?;
}
"csv" => {
df.write_csv(dir, opts, None)
.await
.map_err(|e| df_err("write partitioned csv", e))?;
}
other => {
run.events.push_back(event(
Some(dataset),
None,
format!(
"partitioned sink `{dataset}`: format `{other}` not writable (need parquet/csv)"
),
));
functional_status(
"knut-thund/native_file_sink",
"partitioned",
false,
&format!("{pipeline_name}: sink `{dataset}` format `{other}` unsupported"),
);
return Ok(());
}
}
run.sinks.insert(
dataset.to_string(),
SinkWrite {
path: dir.to_string(),
format: format.to_string(),
rows,
partition_cols: partition_cols.to_vec(),
},
);
run.events.push_back(event(
Some(dataset),
None,
format!(
"sink `{dataset}`: wrote {rows} row(s) partitioned by {partition_cols:?} -> {dir}/ ({format})"
),
));
functional_status(
"knut-thund/native_file_sink",
"partitioned",
rows > 0,
&format!(
"{pipeline_name}: sink `{dataset}` wrote {rows} row(s) partitioned by {partition_cols:?} to {dir} ({format})"
),
);
Ok(())
}
/// A stable table name for a file/topic source URI: the last path segment
/// without its extension.
fn source_table_name(uri: &str) -> String {
uri.rsplit('/')
.next()
.unwrap_or(uri)
.split('.')
.next()
.unwrap_or(uri)
.to_string()
}
/// Execute one flow. Returns `Ok(true)` if it materialised its target,
/// `Ok(false)` if it was skipped/deferred (no query, or an unbounded
/// streaming source), and `Err` on a genuine execution failure.
#[allow(clippy::too_many_arguments)]
async fn run_flow(
ctx: &SessionContext,
flow: &Flow,
streams: &[(String, Vec<Vec<RecordBatch>>)],
mode: StreamingMode,
state_root: Option<&str>,
pipeline_name: &str,
max_triggers: Option<usize>,
run: &mut NativeRun,
) -> Result<bool> {
// A flow is runnable if it carries an opaque `query` OR a typed
// projection/filter node (which reads its single input directly). A
// flow with neither is a structure-only edge and is skipped.
let query: Option<&str> = flow.query.as_deref();
if query.is_none() && !flow.has_pushdown() {
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"flow `{}` has no query — structure-only edge, skipped",
flow.name
),
));
return Ok(false);
}
let streaming = matches!(flow.kind, FlowKind::Streaming { .. });
// Unbounded live (Kafka/CDC-source) streaming flows are driven as
// micro-batch streaming when the caller seeded a synthetic source (see
// `NativeBackend::with_stream_input`): pull each micro-batch, re-run the
// IR→DataFusion lowering over the accumulated state, emit incremental
// results per the flow's output mode. With no seeded stream (and no live
// Kafka/CDC infra in-process) there is nothing to pull, so the flow is
// reported honestly as *deferred* rather than faked.
if let FlowKind::Streaming { source, .. } = &flow.kind {
if matches!(source, SourceSpec::Kafka { .. } | SourceSpec::Cdc { .. }) {
// Streaming flows are driven by their query; a live source with
// no query is a structure-only edge.
let Some(query) = query else {
return Ok(false);
};
if let Some(name) = stream_source_name(source) {
if let Some((_, micro_batches)) = streams.iter().find(|(n, _)| n == &name) {
// Mode routing (additive): `Recompute` is the default and
// the fallback. `Incremental` runs the event-time path only
// with a `state_root`; without one it degrades honestly.
if mode == StreamingMode::Incremental {
if let Some(root) = state_root {
return run_streaming_flow_incremental(
ctx,
flow,
query,
&name,
micro_batches,
root,
pipeline_name,
max_triggers,
run,
)
.await;
}
}
if mode == StreamingMode::Incremental {
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"streaming flow `{}`: incremental mode requested but no state_root set — degrading to recompute",
flow.name
),
));
}
return run_streaming_flow(ctx, flow, query, &name, micro_batches, run)
.await;
}
}
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"flow `{}` ({}) deferred: live/unbounded source has no in-process data — seed one with `with_stream_input`, or run against live Kafka/CDC infra",
flow.name,
flow.kind.label()
),
));
return Ok(false);
}
}
// CDC apply-changes / SCD merge flows: fold the upstream changelog
// (INSERT / UPDATE_AFTER / UPDATE_BEFORE / DELETE — skade & knut-bifrost's
// shared `_change_type` vocabulary) into the target as an idempotent,
// key-matched merge, ordered by the flow's `sequence_by`.
if let FlowKind::Cdc { cdc } = &flow.kind {
// A CDC flow is driven by its changelog query; without one it is a
// structure-only edge.
let Some(query) = query else {
return Ok(false);
};
// A CDC flow reads its changelog from an upstream `reads` dataset.
// When that dataset was seeded as a micro-batch stream, apply the
// merge incrementally per trigger; otherwise run once over the
// bounded changelog already registered (a seeded input / upstream).
let stream = flow
.reads
.iter()
.find_map(|r| streams.iter().find(|(n, _)| n == r));
return run_cdc_flow(ctx, flow, query, cdc, stream, run).await;
}
run.events.push_back(event(
Some(&flow.target),
Some(RunPhase::Running),
format!(
"flow `{}` ({}): executing query",
flow.name,
flow.kind.label()
),
));
// Plan + execute. The base DataFrame is the opaque `query` if the flow
// carries one, else — for a typed projection/filter flow — a direct
// scan of its single input table, so the typed nodes LOWER to real
// DataFusion scan pushdown (column pruning + filter predicate) rather
// than a post-scan projection/filter over an opaque SQL string.
let base_df = match query {
Some(q) => {
match ctx.sql(q).await {
Ok(df) => df,
Err(e) if streaming => {
// A streaming query DataFusion can't yet plan (windowing /
// event-time SQL) is deferred, not a hard failure.
run.events.push_back(event(
Some(&flow.target),
None,
format!("flow `{}` deferred: DataFusion cannot plan this streaming query ({e})", flow.name),
));
return Ok(false);
}
Err(e) => return Err(df_err(&format!("plan flow `{}`", flow.name), e)),
}
}
None => {
// A typed projection/filter flow with no query reads exactly
// one input directly (the single-input scan requirement).
let [input] = flow.reads.as_slice() else {
return Err(df_err(
&format!("plan flow `{}`", flow.name),
"a projection/filter flow with no query must read exactly one input",
));
};
ctx.table(input.as_str()).await.map_err(|e| {
df_err(&format!("scan input `{input}` for flow `{}`", flow.name), e)
})?
}
};
// Apply the typed pushdown nodes (introspectable, planner-lowered):
// the filter first (it references all input columns), then the column
// projection. DataFusion's optimizer pushes both toward the source scan.
let mut df = base_df;
if let Some(pred) = &flow.filter {
let dfschema = df.schema().clone();
let expr = ctx.parse_sql_expr(pred, &dfschema).map_err(|e| {
df_err(
&format!("parse filter `{pred}` for flow `{}`", flow.name),
e,
)
})?;
df = df
.filter(expr)
.map_err(|e| df_err(&format!("apply filter for flow `{}`", flow.name), e))?;
}
if !flow.projection.is_empty() {
let cols: Vec<&str> = flow.projection.iter().map(String::as_str).collect();
df = df
.select_columns(&cols)
.map_err(|e| df_err(&format!("apply projection for flow `{}`", flow.name), e))?;
}
// Capture the pushdown proof: the optimized physical plan carries the
// scan node's `projection=[…]` / `predicate=…` for a file source, so a
// caller/test can assert the pruning/filter reached the source. Only for
// flows that declared a typed pushdown node; non-fatal if it can't plan.
if flow.has_pushdown() {
match df.clone().create_physical_plan().await {
Ok(plan) => {
let physical_plan = datafusion::physical_plan::displayable(plan.as_ref())
.indent(true)
.to_string();
functional_status(
"knut-thund/backend_native",
"scan_pushdown",
true,
&format!(
"{pipeline_name}: flow `{}` pushdown projection={:?} filter={:?}",
flow.name, flow.projection, flow.filter
),
);
run.pushdowns.insert(
flow.name.clone(),
Pushdown {
projected_columns: flow.projection.clone(),
filter: flow.filter.clone(),
physical_plan,
},
);
}
Err(e) => {
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"flow `{}`: could not capture pushdown plan ({e})",
flow.name
),
));
}
}
}
let schema: Arc<Schema> = Arc::new(df.schema().as_arrow().clone());
let batches = df
.collect()
.await
.map_err(|e| df_err(&format!("execute flow `{}`", flow.name), e))?;
let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
// Materialise the target so downstream flows can read it by name.
materialize(ctx, &flow.target, schema, batches, run)?;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"flow `{}` materialised `{}` ({rows} row(s))",
flow.name, flow.target
),
));
// Evaluate data-quality expectations against the produced rows.
apply_expectations(ctx, flow, run).await?;
Ok(true)
}
/// Drive a streaming (Kafka-style unbounded) flow as **micro-batch
/// streaming** over DataFusion.
///
/// Each trigger pulls the next seeded micro-batch into the source's
/// accumulated state, re-runs the flow's query (the exact IR→DataFusion
/// lowering the batch path uses) over that state, and emits an incremental
/// result per [`OutputMode`]. Honors the streaming IR's [`Trigger`]:
/// `Continuous`/`ProcessingTime` fire once per arriving micro-batch (the
/// synthetic source's cadence — no wall-clock sleeps), `AvailableNow`
/// collapses all currently-available data into one bounded pass. The final
/// materialised target is the full result over all rows seen, identical to
/// the batch path over the same data. Single-engine and sequential — no
/// fan-out — mirroring the batch path.
async fn run_streaming_flow(
ctx: &SessionContext,
flow: &Flow,
query: &str,
source_name: &str,
micro_batches: &[Vec<RecordBatch>],
run: &mut NativeRun,
) -> Result<bool> {
let (trigger, output_mode) = match &flow.kind {
FlowKind::Streaming {
trigger,
output_mode,
..
} => (trigger.clone(), *output_mode),
_ => {
return Err(df_err(
"streaming",
"run_streaming_flow on a non-streaming flow",
));
}
};
// Source schema comes from the first row that arrives on any micro-batch.
let Some(src_schema) = micro_batches.iter().flatten().next().map(|b| b.schema()) else {
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"streaming flow `{}`: source `{source_name}` seeded empty — nothing to run",
flow.name
),
));
return Ok(false);
};
// Group the arriving micro-batches into triggers.
let triggers: Vec<Vec<RecordBatch>> = match trigger {
Trigger::AvailableNow => vec![micro_batches.iter().flatten().cloned().collect()],
_ => micro_batches.to_vec(),
};
run.events.push_back(event(
Some(&flow.target),
Some(RunPhase::Running),
format!(
"streaming flow `{}` ({}): {} micro-batch trigger(s), {:?} output",
flow.name,
flow.kind.label(),
triggers.len(),
output_mode
),
));
let mut accumulated: Vec<RecordBatch> = Vec::new();
let mut prev_result: Vec<RecordBatch> = Vec::new();
let mut result_schema: Arc<Schema> = Arc::new(Schema::empty());
for (i, group) in triggers.into_iter().enumerate() {
let arrived: usize = group.iter().map(|b| b.num_rows()).sum();
accumulated.extend(group);
// Register the source over ALL rows seen so far (the stream's state)
// and re-run the flow's query — the same lowering the batch path uses.
let mem = MemTable::try_new(src_schema.clone(), vec![accumulated.clone()])
.map_err(|e| df_err("stream source MemTable", e))?;
let _ = ctx.deregister_table(source_name);
ctx.register_table(source_name, Arc::new(mem))
.map_err(|e| df_err("register stream source", e))?;
let df = ctx
.sql(query)
.await
.map_err(|e| df_err(&format!("plan streaming flow `{}`", flow.name), e))?;
result_schema = Arc::new(df.schema().as_arrow().clone());
let full = df
.collect()
.await
.map_err(|e| df_err(&format!("execute streaming flow `{}`", flow.name), e))?;
// Incremental emission per output mode:
// * Complete: the full result table every trigger.
// * Append / Update: the delta vs the previous trigger's result.
// For a row-wise (non-aggregating) stream this delta is exactly
// the newly-appended rows; for an aggregation it is the changed
// groups. (True watermark-closed Append is a documented non-goal.)
let increment = match output_mode {
OutputMode::Complete => full.clone(),
OutputMode::Append | OutputMode::Update => {
delta(ctx, &result_schema, &full, &prev_result).await?
}
};
let emitted: usize = increment.iter().map(|b| b.num_rows()).sum();
run.increments
.entry(flow.target.clone())
.or_default()
.push(increment);
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"streaming flow `{}` trigger {i}: +{arrived} arrived row(s) -> {emitted} emitted row(s)",
flow.name
),
));
prev_result = full;
}
// The final materialised target is the full result over ALL data seen —
// by construction identical to the batch path over the same rows.
let final_rows: usize = prev_result.iter().map(|b| b.num_rows()).sum();
let triggers_fired = run
.increments
.get(&flow.target)
.map(|v| v.len())
.unwrap_or(0);
materialize(ctx, &flow.target, result_schema, prev_result, run)?;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"streaming flow `{}` materialised `{}` ({final_rows} row(s) over {triggers_fired} trigger(s))",
flow.name, flow.target
),
));
// Data-quality expectations apply to the final materialised table,
// exactly as on the batch path.
apply_expectations(ctx, flow, run).await?;
Ok(true)
}
/// Event-time **watermark tracker** (design `thund-streaming-state.md`,
/// phase 1). Holds the maximum observed event time per input source and
/// combines them into a single monotonic watermark
///
/// ```text
/// W = min over non-idle inputs of ( max_event_time(input) − allowed_lateness )
/// ```
///
/// the **minimum** so the slowest input gates progress (Flink/Arroyo
/// consensus). An input silent for longer than `idle_timeout_ms` stops
/// gating (drops out of the `min`) so one quiet partition cannot freeze the
/// stream; `idle_timeout_ms == 0` means never idle. `W` advances
/// monotonically — a trigger that would lower it leaves it unchanged. Event
/// times are epoch integers (ms) read from the IR watermark's
/// `event_time_column`.
#[derive(Debug)]
pub(super) struct WatermarkTracker {
allowed_lateness_ms: i64,
idle_timeout_ms: i64,
inputs: std::collections::BTreeMap<String, InputProgress>,
watermark: Option<i64>,
late_rows_dropped: u64,
}
/// Per-input progress: the max event time seen and the processing-time of
/// the last row observed (for idle detection).
#[derive(Debug)]
struct InputProgress {
max_event_time: Option<i64>,
last_active_ms: i64,
}
impl WatermarkTracker {
/// A tracker with the IR watermark's allowed-lateness + idle timeout.
pub(super) fn new(allowed_lateness_ms: u64, idle_timeout_ms: u64) -> Self {
WatermarkTracker {
allowed_lateness_ms: allowed_lateness_ms as i64,
idle_timeout_ms: idle_timeout_ms as i64,
inputs: std::collections::BTreeMap::new(),
watermark: None,
late_rows_dropped: 0,
}
}
/// Rebuild a tracker from a committed checkpoint (design phase 4): the
/// same lateness/idle knobs, the persisted combined `W`, the per-input
/// maxima, and the late-drop counter — so a resumed run's watermark
/// continues exactly where the crash left it (monotonic, no regression).
/// `restored_at_ms` seeds each input's `last_active_ms` (idle detection
/// resumes from the recovery point).
pub(super) fn restore(
allowed_lateness_ms: u64,
idle_timeout_ms: u64,
watermark: Option<i64>,
input_maxima: &[(String, i64)],
late_rows_dropped: u64,
restored_at_ms: i64,
) -> Self {
let mut inputs = std::collections::BTreeMap::new();
for (name, mx) in input_maxima {
inputs.insert(
name.clone(),
InputProgress {
max_event_time: Some(*mx),
last_active_ms: restored_at_ms,
},
);
}
WatermarkTracker {
allowed_lateness_ms: allowed_lateness_ms as i64,
idle_timeout_ms: idle_timeout_ms as i64,
inputs,
watermark,
late_rows_dropped,
}
}
/// The per-input max observed event time, for persisting in a checkpoint
/// manifest (only inputs that have seen a row appear).
pub(super) fn input_maxima(&self) -> Vec<(String, i64)> {
self.inputs
.iter()
.filter_map(|(k, p)| p.max_event_time.map(|mx| (k.clone(), mx)))
.collect()
}
/// Record the maximum event time observed on `input` at processing-time
/// `now_ms`. `None` means the input produced no rows this trigger (its
/// stored max is unchanged, and — because no rows arrived — its
/// `last_active_ms` is NOT refreshed, so it can go idle).
pub(super) fn observe_max(
&mut self,
input: &str,
max_event_time: Option<i64>,
now_ms: i64,
) {
let e = self
.inputs
.entry(input.to_string())
.or_insert(InputProgress {
max_event_time: None,
last_active_ms: now_ms,
});
if let Some(mx) = max_event_time {
e.max_event_time = Some(e.max_event_time.map_or(mx, |m| m.max(mx)));
e.last_active_ms = now_ms;
}
}
/// Recompute the combined watermark at processing-time `now_ms` as the
/// min over non-idle inputs of `max_event_time − allowed_lateness`, then
/// clamp it monotonically. Returns the current watermark.
pub(super) fn advance(&mut self, now_ms: i64) -> Option<i64> {
let mut combined: Option<i64> = None;
for p in self.inputs.values() {
let idle = self.idle_timeout_ms > 0
&& now_ms.saturating_sub(p.last_active_ms) > self.idle_timeout_ms;
if idle {
continue;
}
if let Some(mx) = p.max_event_time {
let w = mx - self.allowed_lateness_ms;
combined = Some(combined.map_or(w, |c| c.min(w)));
}
}
if let Some(c) = combined {
self.watermark = Some(self.watermark.map_or(c, |prev| prev.max(c)));
}
self.watermark
}
/// The current combined watermark (`None` until the first advance).
pub(super) fn watermark(&self) -> Option<i64> {
self.watermark
}
/// Add `n` to the running late-drop counter.
pub(super) fn record_drops(&mut self, n: u64) {
self.late_rows_dropped += n;
}
/// Total rows dropped as late across the stream so far.
pub(super) fn late_rows_dropped(&self) -> u64 {
self.late_rows_dropped
}
}
/// The maximum non-null value of the `col` (`Int64`) event-time column over
/// `batches`, or `None` if the column is absent/empty/not `Int64`.
fn max_event_time_i64(batches: &[RecordBatch], col: &str) -> Option<i64> {
let mut mx: Option<i64> = None;
for b in batches {
let Ok(idx) = b.schema().index_of(col) else {
continue;
};
let Some(a) = b.column(idx).as_any().downcast_ref::<Int64Array>() else {
continue;
};
for r in 0..a.len() {
if !a.is_null(r) {
let v = a.value(r);
mx = Some(mx.map_or(v, |m| m.max(v)));
}
}
}
mx
}
/// Split `batches` on the watermark: keep rows whose `col` event time is
/// `>= w` (or NULL — treated as on-time), dropping late rows (`et < w`).
/// Returns the retained batches and the number of rows dropped.
fn drop_late_rows(
batches: &[RecordBatch],
col: &str,
w: i64,
) -> Result<(Vec<RecordBatch>, u64)> {
use datafusion::arrow::array::BooleanArray;
use datafusion::arrow::compute::filter_record_batch;
let mut kept = Vec::new();
let mut dropped = 0u64;
for b in batches {
let idx = b
.schema()
.index_of(col)
.map_err(|e| df_err("watermark event-time column", e))?;
let a = b
.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| {
df_err(
"watermark",
format!("event-time column `{col}` is not Int64"),
)
})?;
let mask: BooleanArray = (0..a.len())
.map(|r| Some(a.is_null(r) || a.value(r) >= w))
.collect();
let keep = mask.iter().filter(|v| matches!(v, Some(true))).count();
dropped += (b.num_rows() - keep) as u64;
let f = filter_record_batch(b, &mask).map_err(|e| df_err("drop late rows", e))?;
if f.num_rows() > 0 {
kept.push(f);
}
}
Ok((kept, dropped))
}
/// Drive a streaming flow in **incremental mode** (design phase 1): an
/// event-time [`WatermarkTracker`] advances per trigger and **late rows**
/// (`event_time < W`) are dropped before the query sees them and counted as
/// `late_rows_dropped`. Aside from the watermark gate this mirrors
/// [`run_streaming_flow`] exactly (recompute over the accumulated *kept*
/// rows), so with no watermark — or no late rows — the result equals the
/// recompute path over the same rows. Windowed state + checkpoints are
/// later phases; this one adds only the watermark + late-drop.
#[allow(clippy::too_many_arguments)]
async fn run_streaming_flow_incremental(
ctx: &SessionContext,
flow: &Flow,
query: &str,
source_name: &str,
micro_batches: &[Vec<RecordBatch>],
state_root: &str,
pipeline_name: &str,
max_triggers: Option<usize>,
run: &mut NativeRun,
) -> Result<bool> {
let (trigger, output_mode, watermark, window) = match &flow.kind {
FlowKind::Streaming {
trigger,
output_mode,
watermark,
window,
..
} => (
trigger.clone(),
*output_mode,
watermark.clone(),
window.clone(),
),
_ => {
return Err(df_err(
"streaming",
"run_streaming_flow_incremental on a non-streaming flow",
));
}
};
// Phase 2/3/4/5: a windowed flow WITH a watermark drives the
// window-close / emit-once / evict operator (bounded memory), now with
// barrier checkpoints + crash recovery + incremental aggregation. Without
// a watermark there is no event-time frontier to close on, so fall
// through to the phase-1 recompute-over-kept-rows path (checkpointed).
if let (Some(win), Some(wm)) = (window.as_ref(), watermark.as_ref()) {
return run_windowed_flow_incremental(
ctx,
flow,
query,
source_name,
micro_batches,
&trigger,
win,
wm,
state_root,
pipeline_name,
max_triggers,
run,
)
.await;
}
let Some(src_schema) = micro_batches.iter().flatten().next().map(|b| b.schema()) else {
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"incremental streaming flow `{}`: source `{source_name}` seeded empty — nothing to run",
flow.name
),
));
return Ok(false);
};
let triggers: Vec<Vec<RecordBatch>> = match trigger {
Trigger::AvailableNow => vec![micro_batches.iter().flatten().cloned().collect()],
_ => micro_batches.to_vec(),
};
let et_col = watermark.as_ref().map(|w| w.event_time_column.clone());
let (al_ms, idle_ms) = watermark
.as_ref()
.map(|w| (w.allowed_lateness_ms, w.idle_timeout_ms))
.unwrap_or((0, 0));
// ── Barrier/epoch checkpoint store + crash recovery (phases 3-4) ──
// The operator state for this non-windowed path is the accumulated kept
// rows (the phase-1 working set). On restart from a committed epoch we
// restore those rows + the watermark + the source offset, and resume.
let store = checkpoint::ParquetStateStore::open(state_root, pipeline_name, &flow.target)?;
let mut accumulated: Vec<RecordBatch> = Vec::new();
let mut prev_result: Vec<RecordBatch> = Vec::new();
let mut result_schema: Arc<Schema> = Arc::new(Schema::empty());
let mut start_offset = 0usize;
let mut tracker = watermark
.as_ref()
.map(|w| WatermarkTracker::new(w.allowed_lateness_ms, w.idle_timeout_ms));
if let Some(m) = store.latest().await? {
accumulated = store.load_operator_state(&m).await?;
start_offset = m.offset;
if watermark.is_some() {
tracker = Some(WatermarkTracker::restore(
al_ms,
idle_ms,
m.watermark,
&m.input_maxima,
m.late_rows_dropped,
m.offset as i64,
));
}
// Seed prev_result by re-running the query over the restored state so
// the first resumed trigger's delta is honest (no re-emit of rows
// already emitted pre-crash).
if !accumulated.is_empty() {
let (rs, full) = run_query_over(
ctx,
source_name,
&src_schema,
query,
&flow.name,
accumulated.clone(),
)
.await?;
result_schema = rs;
prev_result = full;
}
run.resumed_from_epoch.insert(flow.target.clone(), m.epoch);
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"incremental streaming flow `{}`: RECOVERED from epoch {} (offset {}, watermark {:?}) — resuming",
flow.name, m.epoch, m.offset, m.watermark
),
));
functional_status(
"knut-thund/thund_recovery",
"restore",
true,
&format!(
"{}: resumed from epoch {} at offset {} (watermark {:?})",
flow.name, m.epoch, m.offset, m.watermark
),
);
}
run.events.push_back(event(
Some(&flow.target),
Some(RunPhase::Running),
format!(
"incremental streaming flow `{}` ({}): {} trigger(s), {:?} output, watermark {}, start_offset {}",
flow.name,
flow.kind.label(),
triggers.len(),
output_mode,
et_col.as_deref().unwrap_or("<none>"),
start_offset,
),
));
let mut checkpoints = 0usize;
let mut processed_this_run = 0usize;
for (i, group) in triggers.into_iter().enumerate() {
if i < start_offset {
continue; // already consumed + committed before the restart
}
if let Some(mx) = max_triggers {
if processed_this_run >= mx {
break; // bounded run / restart boundary
}
}
processed_this_run += 1;
let arrived: usize = group.iter().map(|b| b.num_rows()).sum();
// Event-time watermark: observe this trigger's max event time,
// advance W, then drop rows below W before the operator sees them.
let mut kept = group;
let mut dropped_this = 0u64;
if let (Some(tr), Some(col)) = (tracker.as_mut(), et_col.as_ref()) {
let now_ms = i as i64; // logical processing clock: trigger index
let mx = max_event_time_i64(&kept, col);
tr.observe_max(source_name, mx, now_ms);
if let Some(w) = tr.advance(now_ms) {
let (retained, dropped) = drop_late_rows(&kept, col, w)?;
dropped_this = dropped;
tr.record_drops(dropped);
kept = retained;
}
}
let kept_rows: usize = kept.iter().map(|b| b.num_rows()).sum();
accumulated.extend(kept);
// Re-run the query over ALL kept rows so far (same lowering as the
// recompute path — kept rows only differ by the late-drop).
let (rs, full) = run_query_over(
ctx,
source_name,
&src_schema,
query,
&flow.name,
accumulated.clone(),
)
.await?;
result_schema = rs;
let increment = match output_mode {
OutputMode::Complete => full.clone(),
OutputMode::Append | OutputMode::Update => {
delta(ctx, &result_schema, &full, &prev_result).await?
}
};
let emitted: usize = increment.iter().map(|b| b.num_rows()).sum();
run.increments
.entry(flow.target.clone())
.or_default()
.push(increment);
// Barrier @ trigger boundary: snapshot the accumulated operator state
// + watermark + offset as epoch `i` (the commit point).
let manifest = checkpoint::CheckpointManifest {
epoch: i as u64,
status: String::new(),
watermark: tracker.as_ref().and_then(|t| t.watermark()),
input_maxima: tracker
.as_ref()
.map(|t| t.input_maxima())
.unwrap_or_default(),
offset: i + 1,
late_rows_dropped: tracker.as_ref().map(|t| t.late_rows_dropped()).unwrap_or(0),
windows_closed: 0,
emitted_windows: Vec::new(),
operator_files: Vec::new(),
};
store.commit(manifest, &accumulated).await?;
checkpoints += 1;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"incremental flow `{}` trigger {i}: +{arrived} arrived, {dropped_this} late-dropped, {kept_rows} kept -> {emitted} emitted row(s); checkpoint epoch {i} committed",
flow.name
),
));
prev_result = full;
}
let final_rows: usize = prev_result.iter().map(|b| b.num_rows()).sum();
let triggers_fired = run
.increments
.get(&flow.target)
.map(|v| v.len())
.unwrap_or(0);
let late_dropped = tracker.as_ref().map(|t| t.late_rows_dropped()).unwrap_or(0);
let final_watermark = tracker.as_ref().and_then(|t| t.watermark());
run.checkpoints_committed
.insert(flow.target.clone(), checkpoints);
materialize(ctx, &flow.target, result_schema, prev_result, run)?;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"incremental streaming flow `{}` materialised `{}` ({final_rows} row(s) over {triggers_fired} trigger(s), {late_dropped} late row(s) dropped, watermark {:?}, {checkpoints} checkpoint(s))",
flow.name, flow.target, final_watermark
),
));
// HONEST STATUS: the incremental watermark path ran and reported its
// late-drop count → GREEN.
functional_status(
"knut-thund/thund_watermark",
"late_drop",
true,
&format!(
"{}: {late_dropped} late row(s) dropped, watermark {:?}",
flow.name, final_watermark
),
);
// Phase 3: the barrier/epoch checkpoint store committed durable state.
if checkpoints > 0 {
functional_status(
"knut-thund/thund_checkpoint",
"commit",
true,
&format!(
"{}: {checkpoints} epoch(s) committed under state_root",
flow.name
),
);
}
apply_expectations(ctx, flow, run).await?;
Ok(true)
}
/// A closeable/open event-time window over the kept stream. `[start, end)`
/// is the event-time range whose rows belong to the window; `close_end` is
/// the event-time point the window is finalisable after — it **closes** once
/// the watermark `W ≥ close_end + allowed_lateness`. For Tumbling/Hopping
/// `close_end == end`; for Session `close_end` is the last event + gap while
/// `end` bounds the activity span, so a session closes a gap after its last
/// row. Tagged onto emitted rows as `(window_start, window_end)` = `(start,
/// close_end)`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct WindowSlice {
start: i64,
end: i64,
close_end: i64,
}
/// Sorted, de-duplicated non-null event-time values in `batches[col]`.
fn distinct_event_times(batches: &[RecordBatch], col: &str) -> Vec<i64> {
let mut set = std::collections::BTreeSet::new();
for b in batches {
let Ok(idx) = b.schema().index_of(col) else {
continue;
};
let Some(a) = b.column(idx).as_any().downcast_ref::<Int64Array>() else {
continue;
};
for r in 0..a.len() {
if !a.is_null(r) {
set.insert(a.value(r));
}
}
}
set.into_iter().collect()
}
/// Every window present among the currently-buffered `open` rows, per the
/// [`WindowSpec`]. Tumbling → one window per row; Hopping → each overlapping
/// slide window; Session → activity spans separated by `> gap_ms` idle
/// gaps (adjacent events within the gap merge into one session). Returned in
/// ascending `start` order, de-duplicated.
fn window_slices(spec: &WindowSpec, open: &[RecordBatch], col: &str) -> Vec<WindowSlice> {
let ets = distinct_event_times(open, col);
match spec {
WindowSpec::Tumbling { size_ms } => {
let size = *size_ms as i64;
if size <= 0 {
return Vec::new();
}
let mut set = std::collections::BTreeSet::new();
for et in ets {
let s = et.div_euclid(size) * size;
set.insert((s, s + size));
}
set.into_iter()
.map(|(start, end)| WindowSlice {
start,
end,
close_end: end,
})
.collect()
}
WindowSpec::Hopping { size_ms, slide_ms } => {
let size = *size_ms as i64;
let slide = *slide_ms as i64;
if size <= 0 || slide <= 0 {
return Vec::new();
}
let mut set = std::collections::BTreeSet::new();
for et in ets {
// Slide windows starting at multiples of `slide` that cover
// `et`: start `s` with `s <= et < s + size`.
let mut s = et.div_euclid(slide) * slide;
while s > et - size {
set.insert((s, s + size));
s -= slide;
}
}
set.into_iter()
.map(|(start, end)| WindowSlice {
start,
end,
close_end: end,
})
.collect()
}
WindowSpec::Session { gap_ms } => {
let gap = *gap_ms as i64;
let mut slices = Vec::new();
let mut it = ets.into_iter();
let Some(first) = it.next() else {
return slices;
};
let (mut smin, mut smax) = (first, first);
for et in it {
if et - smax <= gap {
smax = et;
} else {
slices.push(WindowSlice {
start: smin,
end: smax + 1,
close_end: smax + gap,
});
smin = et;
smax = et;
}
}
slices.push(WindowSlice {
start: smin,
end: smax + 1,
close_end: smax + gap,
});
slices
}
}
}
/// Rows of `batches` whose `col` event time is in `[lo, hi)` (NULL excluded).
fn filter_et_range(
batches: &[RecordBatch],
col: &str,
lo: i64,
hi: i64,
) -> Result<Vec<RecordBatch>> {
use datafusion::arrow::array::BooleanArray;
use datafusion::arrow::compute::filter_record_batch;
let mut out = Vec::new();
for b in batches {
let idx = b
.schema()
.index_of(col)
.map_err(|e| df_err("window event-time column", e))?;
let a = b
.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| {
df_err("window", format!("event-time column `{col}` is not Int64"))
})?;
let mask: BooleanArray = (0..a.len())
.map(|r| Some(!a.is_null(r) && a.value(r) >= lo && a.value(r) < hi))
.collect();
let f = filter_record_batch(b, &mask).map_err(|e| df_err("filter window rows", e))?;
if f.num_rows() > 0 {
out.push(f);
}
}
Ok(out)
}
/// Run the flow's `query` over exactly `rows` (the source registered under
/// `source_name`), returning the result schema and batches.
async fn run_query_over(
ctx: &SessionContext,
source_name: &str,
src_schema: &Arc<Schema>,
query: &str,
flow_name: &str,
rows: Vec<RecordBatch>,
) -> Result<(Arc<Schema>, Vec<RecordBatch>)> {
let mem = MemTable::try_new(src_schema.clone(), vec![rows])
.map_err(|e| df_err("window source MemTable", e))?;
let _ = ctx.deregister_table(source_name);
ctx.register_table(source_name, Arc::new(mem))
.map_err(|e| df_err("register window source", e))?;
let df = ctx
.sql(query)
.await
.map_err(|e| df_err(&format!("plan windowed flow `{flow_name}`"), e))?;
let schema = Arc::new(df.schema().as_arrow().clone());
let out = df
.collect()
.await
.map_err(|e| df_err(&format!("execute windowed flow `{flow_name}`"), e))?;
Ok((schema, out))
}
/// Prepend constant `window_start` / `window_end` `Int64` columns to each
/// batch so a closed window's aggregate carries its bounds.
fn tag_window(batches: &[RecordBatch], start: i64, end: i64) -> Result<Vec<RecordBatch>> {
use datafusion::arrow::array::ArrayRef;
use datafusion::arrow::datatypes::{DataType, Field};
let mut out = Vec::new();
for b in batches {
let n = b.num_rows();
let mut fields = vec![
Field::new("window_start", DataType::Int64, false),
Field::new("window_end", DataType::Int64, false),
];
for f in b.schema().fields() {
fields.push(f.as_ref().clone());
}
let schema = Arc::new(Schema::new(fields));
let mut cols: Vec<ArrayRef> = vec![
Arc::new(Int64Array::from(vec![start; n])),
Arc::new(Int64Array::from(vec![end; n])),
];
for c in b.columns() {
cols.push(c.clone());
}
out.push(RecordBatch::try_new(schema, cols).map_err(|e| df_err("tag window", e))?);
}
Ok(out)
}
/// Build the tagged output schema (`window_start`, `window_end`, then the
/// query's result columns) from a query result schema.
fn tagged_schema(result_schema: &Arc<Schema>) -> Arc<Schema> {
use datafusion::arrow::datatypes::{DataType, Field};
let mut fields = vec![
Field::new("window_start", DataType::Int64, false),
Field::new("window_end", DataType::Int64, false),
];
for f in result_schema.fields() {
fields.push(f.as_ref().clone());
}
Arc::new(Schema::new(fields))
}
/// Drive a **windowed** streaming flow in incremental mode (design phase 2):
/// event-time [`WindowSpec`] windows CLOSE and EMIT exactly once when the
/// watermark `W ≥ window.close_end + allowed_lateness`, then their buffered
/// state is **EVICTED**. Late rows (`event_time < W`) are dropped by the
/// phase-1 gate before windowing — and because a closed window's rows are
/// all strictly below `W`, a late arrival for a closed window is dropped and
/// can never re-open it. Windows still open at end-of-stream are flushed on
/// a final pass so the materialised result covers every on-time row. The
/// buffered (open-window) working set stays bounded — closed windows leave
/// memory — which the recompute path's ever-growing accumulation cannot do.
#[allow(clippy::too_many_arguments)]
#[allow(clippy::too_many_arguments)]
async fn run_windowed_flow_incremental(
ctx: &SessionContext,
flow: &Flow,
query: &str,
source_name: &str,
micro_batches: &[Vec<RecordBatch>],
trigger: &Trigger,
window: &WindowSpec,
watermark: &Watermark,
state_root: &str,
pipeline_name: &str,
max_triggers: Option<usize>,
run: &mut NativeRun,
) -> Result<bool> {
let Some(src_schema) = micro_batches.iter().flatten().next().map(|b| b.schema()) else {
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"windowed streaming flow `{}`: source `{source_name}` seeded empty — nothing to run",
flow.name
),
));
return Ok(false);
};
// Phase 5: when the query is a supported keyed/ungrouped aggregation over
// a Tumbling/Hopping window, drive the INCREMENTAL AGGREGATION operator
// (maintained accumulators, no per-window SQL re-run, raw rows never
// buffered). It shares the checkpoint/recovery substrate. An unsupported
// shape (Session window, WHERE, AVG/float, computed projection, …) falls
// through to the recompute-per-window path below, so nothing regresses.
let agg_windowable = !matches!(window, WindowSpec::Session { .. });
if agg_windowable {
if let Some(plan) = agg::AggPlan::detect(ctx, query, source_name, &src_schema).await {
return agg::run_windowed_incremental_agg(
ctx,
flow,
query,
source_name,
&src_schema,
micro_batches,
trigger,
window,
watermark,
state_root,
pipeline_name,
max_triggers,
plan,
run,
)
.await;
}
}
let triggers: Vec<Vec<RecordBatch>> = match trigger {
Trigger::AvailableNow => vec![micro_batches.iter().flatten().cloned().collect()],
_ => micro_batches.to_vec(),
};
let col = watermark.event_time_column.clone();
let lateness = watermark.allowed_lateness_ms as i64;
let mut tracker =
WatermarkTracker::new(watermark.allowed_lateness_ms, watermark.idle_timeout_ms);
// Buffered rows of not-yet-closed windows — the evictable operator
// state. Closed windows are removed from here (bounded memory).
let mut open: Vec<RecordBatch> = Vec::new();
// Every window emitted, so each closes exactly once (guards re-emit and
// the end-of-stream flush).
let mut emitted: std::collections::BTreeSet<(i64, i64)> = std::collections::BTreeSet::new();
let mut all_emitted: Vec<RecordBatch> = Vec::new();
let mut out_schema: Option<Arc<Schema>> = None;
let mut open_series: Vec<usize> = Vec::new();
let mut closed_count = 0usize;
let mut late_total = 0u64;
// ── Barrier/epoch checkpoint store + crash recovery (phases 3-4) ──
// The operator state is the buffered `open` rows; recovery restores them
// plus the watermark, the emitted-window set, the counters, and the
// source offset, then resumes — so a resumed run neither re-emits a
// window that closed pre-crash nor loses an on-time (non-late) event.
let store = checkpoint::ParquetStateStore::open(state_root, pipeline_name, &flow.target)?;
let mut start_offset = 0usize;
if let Some(m) = store.latest().await? {
open = store.load_operator_state(&m).await?;
start_offset = m.offset;
closed_count = m.windows_closed;
late_total = m.late_rows_dropped;
emitted = m.emitted_windows.iter().copied().collect();
tracker = WatermarkTracker::restore(
watermark.allowed_lateness_ms,
watermark.idle_timeout_ms,
m.watermark,
&m.input_maxima,
m.late_rows_dropped,
m.offset as i64,
);
run.resumed_from_epoch.insert(flow.target.clone(), m.epoch);
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"windowed streaming flow `{}`: RECOVERED from epoch {} (offset {}, {} window(s) already closed, watermark {:?}) — resuming",
flow.name, m.epoch, m.offset, m.windows_closed, m.watermark
),
));
functional_status(
"knut-thund/thund_recovery",
"restore",
true,
&format!(
"{}: resumed windowed flow from epoch {} at offset {} ({} closed)",
flow.name, m.epoch, m.offset, m.windows_closed
),
);
}
run.events.push_back(event(
Some(&flow.target),
Some(RunPhase::Running),
format!(
"windowed streaming flow `{}` ({}): {} trigger(s), window {:?}, watermark `{}` (lateness {}ms), start_offset {}",
flow.name,
flow.kind.label(),
triggers.len(),
window,
col,
lateness,
start_offset,
),
));
let mut checkpoints = 0usize;
let mut processed_this_run = 0usize;
for (i, group) in triggers.into_iter().enumerate() {
if i < start_offset {
continue; // already consumed + committed before the restart
}
if let Some(mx) = max_triggers {
if processed_this_run >= mx {
break; // bounded run / restart boundary
}
}
processed_this_run += 1;
let arrived: usize = group.iter().map(|b| b.num_rows()).sum();
let now_ms = i as i64; // logical processing clock: trigger index
// Watermark: observe this trigger's max event time, advance W, then
// drop late rows (< W) before they reach any window.
let mx = max_event_time_i64(&group, &col);
tracker.observe_max(source_name, mx, now_ms);
let w = tracker.advance(now_ms);
let mut kept = group;
if let Some(w) = w {
let (retained, dropped) = drop_late_rows(&kept, &col, w)?;
tracker.record_drops(dropped);
late_total += dropped;
kept = retained;
}
open.extend(kept);
// Close every window whose close point is at or below the watermark
// (W ≥ close_end + lateness), emit its aggregate once, and evict.
let mut increment: Vec<RecordBatch> = Vec::new();
let mut closed_this = 0usize;
if let Some(w) = w {
let slices = window_slices(window, &open, &col);
for s in &slices {
if w < s.close_end + lateness {
continue; // still open
}
let tag = (s.start, s.close_end);
if !emitted.insert(tag) {
continue; // already closed once
}
let rows = filter_et_range(&open, &col, s.start, s.end)?;
let (res_schema, res) =
run_query_over(ctx, source_name, &src_schema, query, &flow.name, rows)
.await?;
if out_schema.is_none() {
out_schema = Some(tagged_schema(&res_schema));
}
let tagged = tag_window(&res, s.start, s.close_end)?;
increment.extend(tagged.iter().cloned());
all_emitted.extend(tagged);
closed_count += 1;
closed_this += 1;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"windowed flow `{}`: window [{}, {}) closed & evicted at watermark {w}",
flow.name, s.start, s.close_end
),
));
}
// Evict rows whose every window has closed; keep a row if it
// still belongs to at least one open window.
let all = window_slices(window, &open, &col);
let open_ranges: Vec<(i64, i64)> = all
.iter()
.filter(|s| w < s.close_end + lateness)
.map(|s| (s.start, s.end))
.collect();
if open_ranges.is_empty() {
open.clear();
} else {
open = retain_rows_in_ranges(&open, &col, &open_ranges)?;
}
}
let open_rows: usize = open.iter().map(|b| b.num_rows()).sum();
open_series.push(open_rows);
run.increments
.entry(flow.target.clone())
.or_default()
.push(increment);
// Barrier @ trigger boundary: snapshot the post-evict open state +
// watermark + emitted set + offset as epoch `i` (the commit point).
let manifest = checkpoint::CheckpointManifest {
epoch: i as u64,
status: String::new(),
watermark: tracker.watermark(),
input_maxima: tracker.input_maxima(),
offset: i + 1,
late_rows_dropped: late_total,
windows_closed: closed_count,
emitted_windows: emitted.iter().copied().collect(),
operator_files: Vec::new(),
};
store.commit(manifest, &open).await?;
checkpoints += 1;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"windowed flow `{}` trigger {i}: +{arrived} arrived, watermark {:?}, {closed_this} window(s) closed, {open_rows} row(s) buffered open; checkpoint epoch {i} committed",
flow.name, w
),
));
}
// End-of-stream flush is a boundary of THIS run: emit every window still
// open so the materialised result covers every on-time row (parity with
// the batch path). A bounded/capped run (a simulated crash) does NOT
// flush — its still-open windows stay checkpointed for the resumed run to
// close. So flush only when the whole seeded stream was consumed.
let consumed_all = max_triggers.is_none();
let mut flushed = 0usize;
if consumed_all {
let final_slices = window_slices(window, &open, &col);
for s in &final_slices {
let tag = (s.start, s.close_end);
if !emitted.insert(tag) {
continue;
}
let rows = filter_et_range(&open, &col, s.start, s.end)?;
let (res_schema, res) =
run_query_over(ctx, source_name, &src_schema, query, &flow.name, rows).await?;
if out_schema.is_none() {
out_schema = Some(tagged_schema(&res_schema));
}
let tagged = tag_window(&res, s.start, s.close_end)?;
all_emitted.extend(tagged);
closed_count += 1;
flushed += 1;
}
open.clear();
open_series.push(0);
}
let final_watermark = tracker.watermark();
let result_schema = out_schema.unwrap_or_else(|| Arc::new(Schema::empty()));
let emitted_rows: usize = all_emitted.iter().map(|b| b.num_rows()).sum();
materialize(ctx, &flow.target, result_schema, all_emitted, run)?;
run.windows_closed.insert(flow.target.clone(), closed_count);
run.window_open_series
.insert(flow.target.clone(), open_series);
run.checkpoints_committed
.insert(flow.target.clone(), checkpoints);
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"windowed streaming flow `{}` materialised `{}` ({emitted_rows} row(s), {closed_count} window(s) closed [{flushed} flushed at end], {late_total} late row(s) dropped, watermark {:?}, {checkpoints} checkpoint(s))",
flow.name, flow.target, final_watermark
),
));
// HONEST STATUS: the windowed incremental path closed + evicted its
// windows on the event-time watermark → GREEN.
functional_status(
"knut-thund/thund_window",
"close",
true,
&format!(
"{}: {closed_count} window(s) closed & evicted, {late_total} late dropped, watermark {:?}",
flow.name, final_watermark
),
);
if checkpoints > 0 {
functional_status(
"knut-thund/thund_checkpoint",
"commit",
true,
&format!(
"{}: {checkpoints} epoch(s) committed under state_root",
flow.name
),
);
}
apply_expectations(ctx, flow, run).await?;
Ok(true)
}
/// Keep rows of `batches` whose `col` event time falls in any of `ranges`
/// (each `[lo, hi)`) — the union of the still-open windows.
fn retain_rows_in_ranges(
batches: &[RecordBatch],
col: &str,
ranges: &[(i64, i64)],
) -> Result<Vec<RecordBatch>> {
use datafusion::arrow::array::BooleanArray;
use datafusion::arrow::compute::filter_record_batch;
let mut out = Vec::new();
for b in batches {
let idx = b
.schema()
.index_of(col)
.map_err(|e| df_err("window event-time column", e))?;
let a = b
.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.ok_or_else(|| {
df_err("window", format!("event-time column `{col}` is not Int64"))
})?;
let mask: BooleanArray = (0..a.len())
.map(|r| {
if a.is_null(r) {
return Some(false);
}
let v = a.value(r);
Some(ranges.iter().any(|(lo, hi)| v >= *lo && v < *hi))
})
.collect();
let f = filter_record_batch(b, &mask).map_err(|e| df_err("retain open rows", e))?;
if f.num_rows() > 0 {
out.push(f);
}
}
Ok(out)
}
/// Rows in `cur` not already present in `prev` (multiplicity-aware), via SQL
/// `EXCEPT ALL` — the per-trigger increment for Append/Update output.
async fn delta(
ctx: &SessionContext,
schema: &Arc<Schema>,
cur: &[RecordBatch],
prev: &[RecordBatch],
) -> Result<Vec<RecordBatch>> {
// First trigger (nothing emitted yet): everything is new.
if prev.is_empty() {
return Ok(cur.to_vec());
}
let cur_mem = MemTable::try_new(schema.clone(), vec![cur.to_vec()])
.map_err(|e| df_err("delta cur MemTable", e))?;
let prev_mem = MemTable::try_new(schema.clone(), vec![prev.to_vec()])
.map_err(|e| df_err("delta prev MemTable", e))?;
let _ = ctx.deregister_table("__thund_stream_cur");
let _ = ctx.deregister_table("__thund_stream_prev");
ctx.register_table("__thund_stream_cur", Arc::new(cur_mem))
.map_err(|e| df_err("register delta cur", e))?;
ctx.register_table("__thund_stream_prev", Arc::new(prev_mem))
.map_err(|e| df_err("register delta prev", e))?;
let out = ctx
.sql("SELECT * FROM __thund_stream_cur EXCEPT ALL SELECT * FROM __thund_stream_prev")
.await
.map_err(|e| df_err("delta EXCEPT ALL", e))?
.collect()
.await
.map_err(|e| df_err("delta collect", e))?;
let _ = ctx.deregister_table("__thund_stream_cur");
let _ = ctx.deregister_table("__thund_stream_prev");
Ok(out)
}
/// The changelog metadata column carrying the per-row change kind. This is
/// skade / knut-bifrost's shared vocabulary (`SyncMode::CHANGE_TYPE_COLUMN`):
/// `INSERT` / `UPDATE_AFTER` upsert, `DELETE` removes, `UPDATE_BEFORE` is the
/// pre-image of an in-place update and carries no net change (dropped).
const CHANGE_TYPE_COLUMN: &str = "_change_type";
/// Execute a **CDC apply-changes / SCD merge** flow: fold the flow's
/// changelog (its `query` over an upstream `reads` source) into the target
/// as an idempotent, key-matched merge ordered by `cdc.sequence_by`.
///
/// When the changelog source was seeded as a micro-batch stream, the merge
/// runs incrementally — each trigger accumulates the changelog seen so far
/// and recomputes the merged target (apply-changes is a pure function of the
/// ordered changelog, so recomputation is correct and deterministic). With no
/// seeded stream it runs once over the bounded changelog.
async fn run_cdc_flow(
ctx: &SessionContext,
flow: &Flow,
query: &str,
cdc: &CdcSpec,
stream: Option<&(String, Vec<Vec<RecordBatch>>)>,
run: &mut NativeRun,
) -> Result<bool> {
run.events.push_back(event(
Some(&flow.target),
Some(RunPhase::Running),
format!(
"cdc flow `{}` (scd {:?}): applying changes keyed by [{}] ordered by `{}`",
flow.name,
cdc.scd_type,
cdc.keys.join(", "),
cdc.sequence_by
),
));
// ── micro-batch changelog: accumulate + re-merge each trigger ──
if let Some((src, micro_batches)) = stream {
let Some(src_schema) = micro_batches.iter().flatten().next().map(|b| b.schema()) else {
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"cdc flow `{}`: changelog source `{src}` seeded empty — nothing to apply",
flow.name
),
));
return Ok(false);
};
let mut accumulated: Vec<RecordBatch> = Vec::new();
let mut merged: Vec<RecordBatch> = Vec::new();
let mut merged_schema: Arc<Schema> = Arc::new(Schema::empty());
for (i, group) in micro_batches.iter().enumerate() {
let arrived: usize = group.iter().map(|b| b.num_rows()).sum();
accumulated.extend(group.iter().cloned());
// Register the changelog seen so far and run the flow's query
// (the same lowering the batch path uses) to produce the change
// rows, then apply the keyed merge over them.
let mem = MemTable::try_new(src_schema.clone(), vec![accumulated.clone()])
.map_err(|e| df_err("cdc source MemTable", e))?;
let _ = ctx.deregister_table(src.as_str());
ctx.register_table(src.as_str(), Arc::new(mem))
.map_err(|e| df_err("register cdc source", e))?;
let df = ctx
.sql(query)
.await
.map_err(|e| df_err(&format!("plan cdc flow `{}`", flow.name), e))?;
let cl_schema: Arc<Schema> = Arc::new(df.schema().as_arrow().clone());
let changelog = df
.collect()
.await
.map_err(|e| df_err(&format!("execute cdc flow `{}`", flow.name), e))?;
let (m, ms) = apply_cdc_merge(ctx, flow, cdc, &changelog, &cl_schema).await?;
let rows: usize = m.iter().map(|b| b.num_rows()).sum();
run.increments
.entry(flow.target.clone())
.or_default()
.push(m.clone());
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"cdc flow `{}` trigger {i}: +{arrived} change(s) -> {rows} row(s) in target",
flow.name
),
));
merged = m;
merged_schema = ms;
}
let final_rows: usize = merged.iter().map(|b| b.num_rows()).sum();
let triggers = run
.increments
.get(&flow.target)
.map(|v| v.len())
.unwrap_or(0);
materialize(ctx, &flow.target, merged_schema, merged, run)?;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"cdc flow `{}` materialised `{}` ({final_rows} row(s) over {triggers} trigger(s))",
flow.name, flow.target
),
));
apply_expectations(ctx, flow, run).await?;
return Ok(true);
}
// ── bounded changelog: run the query once over registered tables ──
let df = ctx
.sql(query)
.await
.map_err(|e| df_err(&format!("plan cdc flow `{}`", flow.name), e))?;
let cl_schema: Arc<Schema> = Arc::new(df.schema().as_arrow().clone());
let changelog = df
.collect()
.await
.map_err(|e| df_err(&format!("execute cdc flow `{}`", flow.name), e))?;
let (merged, merged_schema) =
apply_cdc_merge(ctx, flow, cdc, &changelog, &cl_schema).await?;
let rows: usize = merged.iter().map(|b| b.num_rows()).sum();
materialize(ctx, &flow.target, merged_schema, merged, run)?;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"cdc flow `{}` materialised `{}` ({rows} row(s))",
flow.name, flow.target
),
));
apply_expectations(ctx, flow, run).await?;
Ok(true)
}
/// Quote a SQL identifier for interpolation (double-quote, doubling any
/// embedded quote) so key/sequence/data column names are safe in the merge.
fn quote_ident(name: &str) -> String {
format!("\"{}\"", name.replace('"', "\"\""))
}
/// The keyed apply-changes merge itself, over one materialised `changelog`
/// (already the flow's query result, carrying the `_change_type` column).
///
/// Consumes skade / knut-bifrost's `_change_type` vocabulary: `UPDATE_BEFORE`
/// rows are dropped (pre-images), a row is a delete when its change type is
/// `DELETE` (or the optional `apply_as_deletes` predicate holds), and every
/// other row (`INSERT` / `UPDATE_AFTER`) is an upsert. For each key the row
/// with the greatest `sequence_by` wins.
///
/// * **SCD type 1** — the target is the latest non-deleted row per key
/// (history overwritten).
/// * **SCD type 2** — every version is kept with derived `__start_at` /
/// `__end_at` validity bounds (`__end_at = NULL` for the current version);
/// a delete closes the open version without opening a new one.
async fn apply_cdc_merge(
ctx: &SessionContext,
flow: &Flow,
cdc: &CdcSpec,
changelog: &[RecordBatch],
cl_schema: &Arc<Schema>,
) -> Result<(Vec<RecordBatch>, Arc<Schema>)> {
// The changelog contract: a `_change_type` column + the key/sequence
// columns must all be present, else the merge is ill-defined.
if cl_schema.index_of(CHANGE_TYPE_COLUMN).is_err() {
return Err(df_err(
"cdc merge",
format!(
"changelog for flow `{}` has no `{CHANGE_TYPE_COLUMN}` column (the changelog contract)",
flow.name
),
));
}
for k in &cdc.keys {
if cl_schema.index_of(k).is_err() {
return Err(df_err(
"cdc merge",
format!("key column `{k}` not in changelog for flow `{}`", flow.name),
));
}
}
if cl_schema.index_of(&cdc.sequence_by).is_err() {
return Err(df_err(
"cdc merge",
format!(
"sequence_by column `{}` not in changelog for flow `{}`",
cdc.sequence_by, flow.name
),
));
}
if cdc.keys.is_empty() {
return Err(df_err(
"cdc merge",
format!("cdc flow `{}` has no key columns", flow.name),
));
}
const CL: &str = "__thund_cdc_changelog";
let mem = MemTable::try_new(cl_schema.clone(), vec![changelog.to_vec()])
.map_err(|e| df_err("cdc changelog MemTable", e))?;
let _ = ctx.deregister_table(CL);
ctx.register_table(CL, Arc::new(mem))
.map_err(|e| df_err("register cdc changelog", e))?;
let ct = quote_ident(CHANGE_TYPE_COLUMN);
let seq = quote_ident(&cdc.sequence_by);
let keys_list = cdc
.keys
.iter()
.map(|k| quote_ident(k))
.collect::<Vec<_>>()
.join(", ");
// Target rows carry every changelog column except the `_change_type`
// metadata (stripped so the target renders like an ordinary table).
let data_list = cl_schema
.fields()
.iter()
.map(|f| f.name().as_str())
.filter(|n| *n != CHANGE_TYPE_COLUMN)
.map(quote_ident)
.collect::<Vec<_>>()
.join(", ");
// A row is a delete on the DELETE change type, plus any user predicate.
let is_delete = match &cdc.apply_as_deletes {
Some(pred) => format!("({ct} = 'DELETE') OR ({pred})"),
None => format!("{ct} = 'DELETE'"),
};
let sql = match cdc.scd_type {
// Latest non-deleted row per key.
ScdType::Type1 => format!(
"WITH __cl AS (SELECT * FROM {CL} WHERE {ct} <> 'UPDATE_BEFORE'), \
__ranked AS (SELECT *, ROW_NUMBER() OVER \
(PARTITION BY {keys_list} ORDER BY {seq} DESC) AS __thund_rn FROM __cl) \
SELECT {data_list} FROM __ranked WHERE __thund_rn = 1 AND NOT ({is_delete})"
),
// Full version history with derived validity bounds. LEAD closes each
// version at the next change's sequence (a delete closes but is not
// itself emitted as a version).
ScdType::Type2 => format!(
"WITH __cl AS (SELECT * FROM {CL} WHERE {ct} <> 'UPDATE_BEFORE'), \
__ver AS (SELECT *, {seq} AS __start_at, \
LEAD({seq}) OVER (PARTITION BY {keys_list} ORDER BY {seq}) AS __end_at FROM __cl) \
SELECT {data_list}, __start_at, __end_at FROM __ver WHERE NOT ({is_delete})"
),
};
let df = ctx
.sql(&sql)
.await
.map_err(|e| df_err(&format!("cdc merge sql for `{}`", flow.name), e))?;
let schema: Arc<Schema> = Arc::new(df.schema().as_arrow().clone());
let out = df
.collect()
.await
.map_err(|e| df_err(&format!("cdc merge collect for `{}`", flow.name), e))?;
let _ = ctx.deregister_table(CL);
Ok((out, schema))
}
/// The table name a streaming flow's query reads its source rows from — a
/// Kafka topic, a CDC source table, or the last path segment of a file URI.
fn stream_source_name(source: &SourceSpec) -> Option<String> {
match source {
SourceSpec::Kafka { topic, .. } => Some(topic.clone()),
SourceSpec::Cdc { source } => Some(source.clone()),
SourceSpec::FileDrop { dir, .. } => Some(source_table_name(dir)),
SourceSpec::Batch { uri, .. } => Some(source_table_name(uri)),
}
}
/// Register `batches` as the in-process table `name` (dropping any prior
/// binding) and stash them on the run for retrieval.
fn materialize(
ctx: &SessionContext,
name: &str,
schema: Arc<Schema>,
batches: Vec<datafusion::arrow::array::RecordBatch>,
run: &mut NativeRun,
) -> Result<()> {
let mem = MemTable::try_new(schema, vec![batches.clone()])
.map_err(|e| df_err("build MemTable", e))?;
let _ = ctx.deregister_table(name);
ctx.register_table(name, Arc::new(mem))
.map_err(|e| df_err("register table", e))?;
run.outputs.insert(name.to_string(), batches);
Ok(())
}
/// **Union-append fan-in**: splice a freshly-run flow's output onto the rows
/// already accumulated for `target` by earlier flows in this run, and
/// re-materialise the union so downstream flows and the sink see every
/// append flow's rows — the executor half of SDP's "one streaming table, N
/// append flows" pattern (the IR always permitted N flows → 1 target; this
/// makes them union instead of clobber).
///
/// `prior` is the union accumulated by earlier flows (snapshotted BEFORE
/// this flow's `run_flow`/`materialize` overwrote `run.outputs[target]` with
/// only its own rows); the freshest flow's rows are read back out of
/// `run.outputs[target]`. The union is `prior ++ fresh` (order-preserving:
/// declaration order of the flows), re-registered under `target` and stored
/// as the new accumulated output.
///
/// **Loud gap (L4):** an append flow whose output schema differs from the
/// accumulated one fails the run rather than silently dropping rows or
/// producing a mixed-schema table — a union is only meaningful over a
/// common schema.
fn union_append(
ctx: &SessionContext,
target: &str,
prior: Vec<RecordBatch>,
flow_name: &str,
pipeline_name: &str,
run: &mut NativeRun,
) -> Result<()> {
// The freshest flow's rows are whatever `run_flow` just registered.
let fresh = run.outputs.get(target).cloned().unwrap_or_default();
let prior_schema = prior.first().map(|b| b.schema());
let fresh_schema = fresh.first().map(|b| b.schema());
// Schema agreement is required for a meaningful union. Compare fields
// (names + types + nullability) — an empty side (a flow that produced no
// rows) carries no schema and unions vacuously.
if let (Some(ps), Some(fs)) = (&prior_schema, &fresh_schema) {
if ps.fields() != fs.fields() {
let msg = format!(
"union-append into `{target}`: flow `{flow_name}` output schema {:?} does not match the accumulated schema {:?}",
fs.fields(),
ps.fields()
);
run.events.push_back(event(Some(target), None, msg.clone()));
functional_status(
"knut-thund/native_union_append",
"fan_in",
false,
&format!("{pipeline_name}: {msg}"),
);
return Err(df_err("union-append", msg));
}
}
let schema = prior_schema
.or(fresh_schema)
.unwrap_or_else(|| Arc::new(Schema::empty()));
let prior_rows: usize = prior.iter().map(|b| b.num_rows()).sum();
let fresh_rows: usize = fresh.iter().map(|b| b.num_rows()).sum();
let mut unioned = prior;
unioned.extend(fresh);
let total: usize = prior_rows + fresh_rows;
materialize(ctx, target, schema, unioned, run)?;
run.events.push_back(event(
Some(target),
None,
format!(
"union-append: flow `{flow_name}` +{fresh_rows} row(s) onto `{target}` -> {total} row(s) total"
),
));
functional_status(
"knut-thund/native_union_append",
"fan_in",
true,
&format!(
"{pipeline_name}: `{target}` unioned flow `{flow_name}` ({prior_rows} + {fresh_rows} = {total} row(s))"
),
);
Ok(())
}
/// Evaluate each of a flow's expectations against its materialised target.
async fn apply_expectations(
ctx: &SessionContext,
flow: &Flow,
run: &mut NativeRun,
) -> Result<()> {
for exp in &flow.expectations {
let violations = count_violations(ctx, &flow.target, &exp.constraint).await?;
if violations == 0 {
run.events.push_back(event(
Some(&flow.target),
None,
format!("expectation `{}` passed (0 violations)", exp.name),
));
continue;
}
match exp.on_violation {
OnViolation::Warn => run.events.push_back(event(
Some(&flow.target),
None,
format!(
"expectation `{}` WARN: {violations} violating row(s) kept",
exp.name
),
)),
OnViolation::Drop => {
// Re-materialise keeping only rows that satisfy the constraint.
let keep_sql = format!(
"SELECT * FROM \"{}\" WHERE ({})",
flow.target, exp.constraint
);
let df = ctx
.sql(&keep_sql)
.await
.map_err(|e| df_err(&format!("expectation `{}` filter", exp.name), e))?;
let schema = Arc::new(df.schema().as_arrow().clone());
let kept = df
.collect()
.await
.map_err(|e| df_err(&format!("expectation `{}` collect", exp.name), e))?;
let kept_rows: usize = kept.iter().map(|b| b.num_rows()).sum();
materialize(ctx, &flow.target, schema, kept, run)?;
run.events.push_back(event(
Some(&flow.target),
None,
format!(
"expectation `{}` DROP: dropped {violations} row(s), {kept_rows} kept",
exp.name
),
));
}
OnViolation::Fail => {
return Err(ThundError::Backend(format!(
"expectation `{}` FAILED on `{}`: {violations} violating row(s)",
exp.name, flow.target
)));
}
}
}
Ok(())
}
/// Count rows in `table` that violate the boolean `constraint`.
async fn count_violations(ctx: &SessionContext, table: &str, constraint: &str) -> Result<i64> {
let sql = format!("SELECT count(*) AS c FROM \"{table}\" WHERE NOT ({constraint})");
let batches = ctx
.sql(&sql)
.await
.map_err(|e| df_err("expectation constraint", e))?
.collect()
.await
.map_err(|e| df_err("expectation count", e))?;
let n = batches
.first()
.and_then(|b| b.column(0).as_any().downcast_ref::<Int64Array>())
.filter(|a| !a.is_empty())
.map(|a| a.value(0))
.unwrap_or(0);
Ok(n)
}
/// **Incremental aggregation operator** (design `thund-streaming-state-design.md`
/// phase 5): maintains keyed/ungrouped accumulators updated as rows arrive
/// and emits a window's aggregate from those accumulators on close — **no
/// per-window SQL re-run**, and the raw rows are never buffered (only the
/// per-`(window, group)` accumulators, so the working set is bounded by the
/// key cardinality, not the row count).
///
/// Supported (equivalence-proven-vs-recompute) shape, else the caller falls
/// back to the recompute-per-window path (nothing regresses):
/// `SELECT [group cols,] <aggs> FROM <source> [GROUP BY group cols]` where
/// the aggregates are `COUNT(*)` / `COUNT(col)` / `SUM(col)` / `MIN(col)` /
/// `MAX(col)` over `Int64` columns, group columns are `Int64`/`Utf8`, no
/// `WHERE`, and the window is Tumbling or Hopping (Session merging is not
/// incrementalised here → falls back). All aggregate outputs are `Int64`,
/// so both the emitted result and the checkpointed accumulator state have a
/// fixed, lossless schema.
mod agg {
use super::{
WatermarkTracker, checkpoint, df_err, drop_late_rows, event, materialize,
max_event_time_i64, tagged_schema,
};
use crate::backend::RunPhase;
use crate::error::Result;
use crate::functional_status;
use crate::ir::{Flow, Trigger, Watermark, WindowSpec};
use datafusion::arrow::array::{
Array, ArrayRef, Float64Array, Int64Array, RecordBatch, StringArray,
};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use datafusion::datasource::MemTable;
use datafusion::logical_expr::{Expr, LogicalPlan};
use datafusion::prelude::SessionContext;
use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;
/// A group column: its source-schema index + name + type (Int64/Utf8).
#[derive(Debug, Clone)]
struct GroupCol {
idx: usize,
name: String,
utf8: bool,
}
/// A supported aggregate over an `Int64` source column (or `*`).
#[derive(Debug, Clone)]
enum AggFn {
/// `COUNT(*)` — count every kept row.
Count,
/// `COUNT(col)` — count non-null rows of the source column.
CountCol(usize),
/// `SUM(col)` over `Int64`.
Sum(usize),
/// `MIN(col)` over `Int64`.
Min(usize),
/// `MAX(col)` over `Int64`.
Max(usize),
/// `AVG(col)` over `Int64` — maintained EXACTLY as two i64 partials
/// (`sum`, non-null `count`), divided to `Float64` only on emit, so no
/// float accumulation error creeps in over a window's arrivals.
Avg(usize),
}
impl AggFn {
/// The number of `i64` partial slots this aggregate maintains: AVG keeps
/// two (sum + count); every other aggregate keeps one.
fn width(&self) -> usize {
match self {
AggFn::Avg(_) => 2,
_ => 1,
}
}
/// The Arrow type of this aggregate's emitted column — AVG emits
/// `Float64` (matching DataFusion's `avg` over an integer column);
/// COUNT/SUM/MIN/MAX emit `Int64`.
fn out_type(&self) -> DataType {
match self {
AggFn::Avg(_) => DataType::Float64,
_ => DataType::Int64,
}
}
}
/// A detected, incrementalisable aggregate query plan.
pub(in crate::backend::native) struct AggPlan {
groups: Vec<GroupCol>,
aggs: Vec<AggFn>,
/// Partial-slot start offset per aggregate (AVG occupies two slots, the
/// rest one), so `Partials` stays a flat `Vec<i64>` indexed by offset.
offsets: Vec<usize>,
/// Total `i64` partial slots across all aggregates (`sum` of widths).
width: usize,
/// The query's result schema (`[group cols, agg cols]`) — the emitted
/// window rows carry `[window_start, window_end]` then these.
out_schema: Arc<Schema>,
The file goes on; the console shows all of it.