knut-thund 0.1.7

//! 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,
            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>,
        }

        /// One value of a group key (Int64 / Utf8 / null), ordered for a stable
        /// `BTreeMap` group order.
        #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
        enum KeyVal {
            I(i64),
            S(String),
            Null,
        }

        /// A window identity: `[start, end)` + the close point.
        #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
        struct WinKey {
            start: i64,
            end: i64,
            close_end: i64,
        }

        /// Per-`(window, group)` accumulator partials, aligned to `AggPlan.aggs`.
        /// Every partial is an `i64` (COUNT/SUM/MIN/MAX over Int64 → Int64), so
        /// the state serialises to a fixed Int64-per-agg schema.
        type Partials = Vec<i64>;
        /// The open operator state: window → group key → partials.
        type OpenState = BTreeMap<WinKey, BTreeMap<Vec<KeyVal>, Partials>>;

        impl AggPlan {
            /// Detect a supported incremental-aggregation shape for `query` over a
            /// source of `src_schema`, or `None` (→ recompute-per-window fallback).
            pub(in crate::backend::native) async fn detect(
                ctx: &SessionContext,
                query: &str,
                source_name: &str,
                src_schema: &Arc<Schema>,
            ) -> Option<AggPlan> {
                // Register an empty source so the planner resolves the columns.
                let mem = MemTable::try_new(src_schema.clone(), vec![vec![]]).ok()?;
                let _ = ctx.deregister_table(source_name);
                ctx.register_table(source_name, Arc::new(mem)).ok()?;
                let df = ctx.sql(query).await.ok()?;
                let out_schema = Arc::new(df.schema().as_arrow().clone());
                let plan = df.logical_plan().clone();

                let aggregate = find_aggregate(&plan)?;
                // No WHERE / no non-scan input: the aggregate must sit directly on
                // a plain table scan (a Filter/Join/Projection under it → bail).
                if !is_plain_scan(&aggregate.input) {
                    return None;
                }
                // The top must be a PASS-THROUGH projection over the aggregate —
                // only column refs / bare aggregate exprs (optionally aliased). A
                // COMPUTING projection (e.g. `SUM(x)+1`, `CAST(...)`) would make the
                // operator emit the raw aggregate instead of the computed value, so
                // it must fall back to recompute (correctness over coverage).
                if !is_passthrough_top(&plan) {
                    return None;
                }

                // Group columns: each must be a bare column of Int64/Utf8.
                let mut groups = Vec::new();
                for g in &aggregate.group_expr {
                    let Expr::Column(c) = g else { return None };
                    let idx = src_schema.index_of(&c.name).ok()?;
                    let utf8 = match src_schema.field(idx).data_type() {
                        DataType::Int64 => false,
                        DataType::Utf8 => true,
                        _ => return None,
                    };
                    groups.push(GroupCol {
                        idx,
                        name: c.name.clone(),
                        utf8,
                    });
                }

                // Aggregates: COUNT(*)/COUNT(col)/SUM/MIN/MAX over Int64.
                let mut aggs = Vec::new();
                for a in &aggregate.aggr_expr {
                    let inner = match a {
                        Expr::Alias(al) => al.expr.as_ref(),
                        other => other,
                    };
                    let Expr::AggregateFunction(af) = inner else {
                        return None;
                    };
                    if af.params.distinct || af.params.filter.is_some() {
                        return None;
                    }
                    let name = af.func.name().to_ascii_lowercase();
                    let args = &af.params.args;
                    let col_idx = |e: &Expr| -> Option<usize> {
                        if let Expr::Column(c) = e {
                            let i = src_schema.index_of(&c.name).ok()?;
                            matches!(src_schema.field(i).data_type(), DataType::Int64).then_some(i)
                        } else {
                            None
                        }
                    };
                    let agg = match name.as_str() {
                        // `COUNT(*)` lowers to `count(Int64(1))` — a literal arg —
                        // and counts every row; `COUNT(col)` counts non-null rows.
                        "count" if args.is_empty() => AggFn::Count,
                        "count" if args.len() == 1 => match &args[0] {
                            Expr::Literal(..) => AggFn::Count,
                            Expr::Column(_) => AggFn::CountCol(col_idx(&args[0])?),
                            _ => return None,
                        },
                        "sum" if args.len() == 1 => AggFn::Sum(col_idx(&args[0])?),
                        "min" if args.len() == 1 => AggFn::Min(col_idx(&args[0])?),
                        "max" if args.len() == 1 => AggFn::Max(col_idx(&args[0])?),
                        "avg" if args.len() == 1 => AggFn::Avg(col_idx(&args[0])?),
                        _ => return None,
                    };
                    aggs.push(agg);
                }
                if aggs.is_empty() {
                    return None;
                }
                // Flat partial-slot offsets (AVG takes two slots, the rest one).
                let mut offsets = Vec::with_capacity(aggs.len());
                let mut width = 0usize;
                for a in &aggs {
                    offsets.push(width);
                    width += a.width();
                }

                // Output schema must be exactly [group cols (names+types), aggs
                // (Int64)] in that order — this both guards a reordering/computing
                // projection and fixes the emit column mapping.
                if out_schema.fields().len() != groups.len() + aggs.len() {
                    return None;
                }
                for (i, g) in groups.iter().enumerate() {
                    let f = out_schema.field(i);
                    if f.name() != &g.name {
                        return None;
                    }
                    let ok = if g.utf8 {
                        f.data_type() == &DataType::Utf8
                    } else {
                        f.data_type() == &DataType::Int64
                    };
                    if !ok {
                        return None;
                    }
                }
                for (k, a) in aggs.iter().enumerate() {
                    if out_schema.field(groups.len() + k).data_type() != &a.out_type() {
                        return None;
                    }
                }

                Some(AggPlan {
                    groups,
                    aggs,
                    offsets,
                    width,
                    out_schema,
                })
            }
        }

        /// Descend single-input wrappers to the `Aggregate` node, or `None`.
        fn find_aggregate(p: &LogicalPlan) -> Option<&datafusion::logical_expr::Aggregate> {
            match p {
                LogicalPlan::Aggregate(a) => Some(a),
                LogicalPlan::Projection(pr) => find_aggregate(&pr.input),
                LogicalPlan::SubqueryAlias(s) => find_aggregate(&s.input),
                _ => None,
            }
        }

        /// True if `p` is a plain table scan (no pushed filter) — possibly under a
        /// `SubqueryAlias`. A `WHERE` surfaces as `TableScan.filters` or a `Filter`
        /// node, both of which fail this check → the caller falls back.
        fn is_plain_scan(p: &LogicalPlan) -> bool {
            match p {
                LogicalPlan::TableScan(ts) => ts.filters.is_empty(),
                LogicalPlan::SubqueryAlias(s) => is_plain_scan(&s.input),
                _ => false,
            }
        }

        /// Peel `Expr::Alias` layers to the underlying expression.
        fn strip_alias(e: &Expr) -> &Expr {
            match e {
                Expr::Alias(a) => strip_alias(&a.expr),
                other => other,
            }
        }

        /// True if the top of the plan is a pure PASS-THROUGH over the aggregate:
        /// the `Aggregate` node itself, or a `Projection` whose every expression is
        /// (after alias-stripping) a bare column ref or a bare aggregate call — NOT
        /// a computed expression. A computing projection (`SUM(x)+1`, a `CAST`, a
        /// scalar function) would emit the raw aggregate value, so it fails here and
        /// the caller falls back to recompute-per-window.
        fn is_passthrough_top(p: &LogicalPlan) -> bool {
            match p {
                LogicalPlan::Aggregate(_) => true,
                LogicalPlan::SubqueryAlias(s) => is_passthrough_top(&s.input),
                LogicalPlan::Projection(pr) => {
                    pr.expr.iter().all(|e| {
                        matches!(strip_alias(e), Expr::Column(_) | Expr::AggregateFunction(_))
                    }) && is_passthrough_top(&pr.input)
                }
                _ => false,
            }
        }

        /// The windows an event time belongs to, per the (Tumbling/Hopping) spec.
        fn windows_for(spec: &WindowSpec, et: i64) -> Vec<WinKey> {
            match spec {
                WindowSpec::Tumbling { size_ms } => {
                    let size = *size_ms as i64;
                    if size <= 0 {
                        return Vec::new();
                    }
                    let s = et.div_euclid(size) * size;
                    vec![WinKey {
                        start: s,
                        end: s + size,
                        close_end: s + size,
                    }]
                }
                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 out = Vec::new();
                    let mut s = et.div_euclid(slide) * slide;
                    while s > et - size {
                        out.push(WinKey {
                            start: s,
                            end: s + size,
                            close_end: s + size,
                        });
                        s -= slide;
                    }
                    out
                }
                // Session windows merge across arrivals — not incrementalised here.
                WindowSpec::Session { .. } => Vec::new(),
            }
        }

        /// Read one row's group key from the source columns.
        fn group_key(groups: &[GroupCol], batch: &RecordBatch, row: usize) -> Result<Vec<KeyVal>> {
            let mut k = Vec::with_capacity(groups.len());
            for g in groups {
                let col = batch.column(g.idx);
                if g.utf8 {
                    let a = col
                        .as_any()
                        .downcast_ref::<StringArray>()
                        .ok_or_else(|| df_err("agg group", format!("`{}` not Utf8", g.name)))?;
                    k.push(if a.is_null(row) {
                        KeyVal::Null
                    } else {
                        KeyVal::S(a.value(row).to_string())
                    });
                } else {
                    let a = col
                        .as_any()
                        .downcast_ref::<Int64Array>()
                        .ok_or_else(|| df_err("agg group", format!("`{}` not Int64", g.name)))?;
                    k.push(if a.is_null(row) {
                        KeyVal::Null
                    } else {
                        KeyVal::I(a.value(row))
                    });
                }
            }
            Ok(k)
        }

        /// The identity partials for the aggregate list (before any row folds in),
        /// laid out flat by `AggFn::width` (AVG contributes `[sum=0, count=0]`).
        fn init_partials(aggs: &[AggFn]) -> Partials {
            let mut p = Vec::new();
            for a in aggs {
                match a {
                    AggFn::Count | AggFn::CountCol(_) | AggFn::Sum(_) => p.push(0),
                    AggFn::Min(_) => p.push(i64::MAX),
                    AggFn::Max(_) => p.push(i64::MIN),
                    AggFn::Avg(_) => {
                        p.push(0); // running sum
                        p.push(0); // running non-null count
                    }
                }
            }
            p
        }

        /// Fold one source row into a group's partials, using each aggregate's flat
        /// slot `offset`.
        fn fold_row(
            aggs: &[AggFn],
            offsets: &[usize],
            p: &mut Partials,
            batch: &RecordBatch,
            row: usize,
        ) -> Result<()> {
            let int_at = |c: usize| -> Result<&Int64Array> {
                batch
                    .column(c)
                    .as_any()
                    .downcast_ref::<Int64Array>()
                    .ok_or_else(|| df_err("agg", "aggregate column not Int64"))
            };
            for (j, a) in aggs.iter().enumerate() {
                let off = offsets[j];
                match a {
                    AggFn::Count => p[off] += 1,
                    AggFn::CountCol(c) => {
                        if !batch.column(*c).is_null(row) {
                            p[off] += 1;
                        }
                    }
                    AggFn::Sum(c) | AggFn::Min(c) | AggFn::Max(c) => {
                        let col = int_at(*c)?;
                        if col.is_null(row) {
                            continue; // DataFusion ignores nulls in SUM/MIN/MAX
                        }
                        let v = col.value(row);
                        match a {
                            AggFn::Sum(_) => p[off] += v,
                            AggFn::Min(_) => p[off] = p[off].min(v),
                            AggFn::Max(_) => p[off] = p[off].max(v),
                            _ => unreachable!(),
                        }
                    }
                    AggFn::Avg(c) => {
                        let col = int_at(*c)?;
                        if col.is_null(row) {
                            continue; // AVG ignores nulls (both sum and count)
                        }
                        p[off] += col.value(row); // sum
                        p[off + 1] += 1; // count
                    }
                }
            }
            Ok(())
        }

        /// Total open accumulator entries (the bounded working-set metric).
        fn open_size(open: &OpenState) -> usize {
            open.values().map(|g| g.len()).sum()
        }

        /// Build the emitted `RecordBatch` for one closed window: schema
        /// `[window_start, window_end] ++ out_schema`, one row per group.
        fn emit_window(
            plan: &AggPlan,
            out_tagged: &Arc<Schema>,
            wk: &WinKey,
            groups: &BTreeMap<Vec<KeyVal>, Partials>,
        ) -> Result<RecordBatch> {
            let entries: Vec<(&Vec<KeyVal>, &Partials)> = groups.iter().collect();
            let n = entries.len();
            let mut ws = Vec::with_capacity(n);
            let mut we = Vec::with_capacity(n);
            // group-column builders
            let mut gi: Vec<Vec<i64>> = vec![Vec::with_capacity(n); plan.groups.len()];
            let mut gs: Vec<Vec<Option<String>>> = vec![Vec::with_capacity(n); plan.groups.len()];
            for (key, _) in &entries {
                ws.push(wk.start);
                we.push(wk.close_end);
                for (ci, kv) in key.iter().enumerate() {
                    if plan.groups[ci].utf8 {
                        gs[ci].push(match kv {
                            KeyVal::S(s) => Some(s.clone()),
                            _ => None,
                        });
                    } else {
                        gi[ci].push(match kv {
                            KeyVal::I(v) => *v,
                            _ => 0,
                        });
                    }
                }
            }
            let mut cols: Vec<ArrayRef> = Vec::with_capacity(out_tagged.fields().len());
            cols.push(Arc::new(Int64Array::from(ws)));
            cols.push(Arc::new(Int64Array::from(we)));
            for (ci, g) in plan.groups.iter().enumerate() {
                if g.utf8 {
                    cols.push(Arc::new(StringArray::from(gs[ci].clone())));
                } else {
                    cols.push(Arc::new(Int64Array::from(gi[ci].clone())));
                }
            }
            // One column per aggregate, built from its flat partial slot(s): AVG
            // divides its (sum, count) to Float64 here (null for an all-null group,
            // matching SQL `AVG`); every other aggregate emits its Int64 partial.
            for (j, a) in plan.aggs.iter().enumerate() {
                let off = plan.offsets[j];
                match a {
                    AggFn::Avg(_) => {
                        let vals: Vec<Option<f64>> = entries
                            .iter()
                            .map(|(_, p)| {
                                let cnt = p[off + 1];
                                (cnt != 0).then(|| p[off] as f64 / cnt as f64)
                            })
                            .collect();
                        cols.push(Arc::new(Float64Array::from(vals)));
                    }
                    _ => {
                        let vals: Vec<i64> = entries.iter().map(|(_, p)| p[off]).collect();
                        cols.push(Arc::new(Int64Array::from(vals)));
                    }
                }
            }
            RecordBatch::try_new(out_tagged.clone(), cols).map_err(|e| df_err("agg emit", e))
        }

        /// The fixed operator-state schema for checkpointing: window identity +
        /// group columns (typed) + one Int64 per aggregate PARTIAL SLOT (`width`,
        /// so AVG contributes two: its running sum and count).
        fn state_schema(plan: &AggPlan) -> Arc<Schema> {
            let mut fields = vec![
                Field::new("__ws", DataType::Int64, false),
                Field::new("__we", DataType::Int64, false),
                Field::new("__wc", DataType::Int64, false),
            ];
            for (i, g) in plan.groups.iter().enumerate() {
                let ty = if g.utf8 {
                    DataType::Utf8
                } else {
                    DataType::Int64
                };
                fields.push(Field::new(format!("__g{i}"), ty, true));
            }
            for i in 0..plan.width {
                fields.push(Field::new(format!("__a{i}"), DataType::Int64, false));
            }
            Arc::new(Schema::new(fields))
        }

        /// Serialise the open state to one `RecordBatch` (row per window,group).
        fn state_to_batch(plan: &AggPlan, open: &OpenState) -> Result<Vec<RecordBatch>> {
            let schema = state_schema(plan);
            let n = open_size(open);
            let mut ws = Vec::with_capacity(n);
            let mut we = Vec::with_capacity(n);
            let mut wc = Vec::with_capacity(n);
            let mut gi: Vec<Vec<i64>> = vec![Vec::with_capacity(n); plan.groups.len()];
            let mut gs: Vec<Vec<Option<String>>> = vec![Vec::with_capacity(n); plan.groups.len()];
            let mut acols: Vec<Vec<i64>> = vec![Vec::with_capacity(n); plan.width];
            for (wk, gmap) in open {
                for (key, partials) in gmap {
                    ws.push(wk.start);
                    we.push(wk.end);
                    wc.push(wk.close_end);
                    for (ci, kv) in key.iter().enumerate() {
                        if plan.groups[ci].utf8 {
                            gs[ci].push(match kv {
                                KeyVal::S(s) => Some(s.clone()),
                                _ => None,
                            });
                        } else {
                            gi[ci].push(match kv {
                                KeyVal::I(v) => *v,
                                _ => 0,
                            });
                        }
                    }
                    for (j, v) in partials.iter().enumerate() {
                        acols[j].push(*v);
                    }
                }
            }
            if n == 0 {
                return Ok(Vec::new());
            }
            let mut cols: Vec<ArrayRef> = vec![
                Arc::new(Int64Array::from(ws)),
                Arc::new(Int64Array::from(we)),
                Arc::new(Int64Array::from(wc)),
            ];
            for (ci, g) in plan.groups.iter().enumerate() {
                if g.utf8 {
                    cols.push(Arc::new(StringArray::from(gs[ci].clone())));
                } else {
                    cols.push(Arc::new(Int64Array::from(gi[ci].clone())));
                }
            }
            for j in 0..plan.width {
                cols.push(Arc::new(Int64Array::from(acols[j].clone())));
            }
            Ok(vec![
                RecordBatch::try_new(schema, cols).map_err(|e| df_err("agg state", e))?,
            ])
        }

        /// Rebuild the open state from checkpointed `batches`.
        fn state_from_batches(plan: &AggPlan, batches: &[RecordBatch]) -> Result<OpenState> {
            let mut open: OpenState = BTreeMap::new();
            let g0 = 3usize; // first group column index in the state schema
            for b in batches {
                let ws = int_col(b, 0)?;
                let we = int_col(b, 1)?;
                let wc = int_col(b, 2)?;
                for r in 0..b.num_rows() {
                    let wk = WinKey {
                        start: ws.value(r),
                        end: we.value(r),
                        close_end: wc.value(r),
                    };
                    let mut key = Vec::with_capacity(plan.groups.len());
                    for (ci, g) in plan.groups.iter().enumerate() {
                        let col = b.column(g0 + ci);
                        if g.utf8 {
                            let a = col
                                .as_any()
                                .downcast_ref::<StringArray>()
                                .ok_or_else(|| df_err("agg restore", "group not Utf8"))?;
                            key.push(if a.is_null(r) {
                                KeyVal::Null
                            } else {
                                KeyVal::S(a.value(r).to_string())
                            });
                        } else {
                            let a = col
                                .as_any()
                                .downcast_ref::<Int64Array>()
                                .ok_or_else(|| df_err("agg restore", "group not Int64"))?;
                            key.push(if a.is_null(r) {
                                KeyVal::Null
                            } else {
                                KeyVal::I(a.value(r))
                            });
                        }
                    }
                    let a0 = g0 + plan.groups.len();
                    let mut partials = Vec::with_capacity(plan.width);
                    for j in 0..plan.width {
                        partials.push(int_col(b, a0 + j)?.value(r));
                    }
                    open.entry(wk).or_default().insert(key, partials);
                }
            }
            Ok(open)
        }

        fn int_col(b: &RecordBatch, i: usize) -> Result<&Int64Array> {
            b.column(i)
                .as_any()
                .downcast_ref::<Int64Array>()
                .ok_or_else(|| df_err("agg state", format!("column {i} not Int64")))
        }

        /// Drive a windowed streaming flow with the incremental-aggregation
        /// operator + the phase 3-4 checkpoint/recovery substrate.
        #[allow(clippy::too_many_arguments)]
        pub(in crate::backend::native) async fn run_windowed_incremental_agg(
            ctx: &SessionContext,
            flow: &Flow,
            _query: &str,
            source_name: &str,
            src_schema: &Arc<Schema>,
            micro_batches: &[Vec<RecordBatch>],
            trigger: &Trigger,
            window: &WindowSpec,
            watermark: &Watermark,
            state_root: &str,
            pipeline_name: &str,
            max_triggers: Option<usize>,
            plan: AggPlan,
            run: &mut super::NativeRun,
        ) -> Result<bool> {
            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 out_tagged = tagged_schema(&plan.out_schema);

            let mut tracker =
                WatermarkTracker::new(watermark.allowed_lateness_ms, watermark.idle_timeout_ms);
            let mut open: OpenState = BTreeMap::new();
            let mut emitted: BTreeSet<(i64, i64)> = BTreeSet::new();
            let mut all_emitted: Vec<RecordBatch> = Vec::new();
            let mut open_series: Vec<usize> = Vec::new();
            let mut closed_count = 0usize;
            let mut late_total = 0u64;

            // ── Checkpoint store + crash recovery (phases 3-4, agg state) ──
            let store =
                checkpoint::ParquetStateStore::open(state_root, pipeline_name, &flow.target)?;
            let mut start_offset = 0usize;
            if let Some(m) = store.latest().await? {
                let st = store.load_operator_state(&m).await?;
                open = state_from_batches(&plan, &st)?;
                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);
                functional_status(
                    "knut-thund/thund_recovery",
                    "restore",
                    true,
                    &format!(
                        "{}: resumed incremental-agg flow from epoch {} at offset {}",
                        flow.name, m.epoch, m.offset
                    ),
                );
            }

            run.events.push_back(event(
                Some(&flow.target),
                Some(RunPhase::Running),
                format!(
                    "incremental-agg flow `{}` ({}): {} trigger(s), window {:?}, {} group col(s), {} agg(s), start_offset {}",
                    flow.name, flow.kind.label(), triggers.len(), window, plan.groups.len(), plan.aggs.len(), 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;
                }
                if let Some(mx) = max_triggers {
                    if processed_this_run >= mx {
                        break;
                    }
                }
                processed_this_run += 1;
                let arrived: usize = group.iter().map(|b| b.num_rows()).sum();
                let now_ms = i as i64;

                // Watermark: advance W, drop late rows before they update any acc.
                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;
                }

                // Fold each kept row into its window(s)' group accumulators — no
                // buffering of raw rows, no SQL.
                let et_idx = src_schema
                    .index_of(&col)
                    .map_err(|e| df_err("agg event-time column", e))?;
                for b in &kept {
                    let et = b
                        .column(et_idx)
                        .as_any()
                        .downcast_ref::<Int64Array>()
                        .ok_or_else(|| df_err("agg", "event-time column not Int64"))?;
                    for r in 0..b.num_rows() {
                        if et.is_null(r) {
                            continue;
                        }
                        let key = group_key(&plan.groups, b, r)?;
                        for wk in windows_for(window, et.value(r)) {
                            // A row for an already-closed window is late (dropped
                            // above), so any window we see here is still open.
                            let gmap = open.entry(wk).or_default();
                            let p = gmap
                                .entry(key.clone())
                                .or_insert_with(|| init_partials(&plan.aggs));
                            fold_row(&plan.aggs, &plan.offsets, p, b, r)?;
                        }
                    }
                }

                // Close windows whose close point passed the watermark; emit from
                // accumulators (one row per group) and evict.
                let mut increment: Vec<RecordBatch> = Vec::new();
                let mut closed_this = 0usize;
                if let Some(w) = w {
                    let ready: Vec<WinKey> = open
                        .keys()
                        .copied()
                        .filter(|wk| w >= wk.close_end + lateness)
                        .collect();
                    for wk in ready {
                        let tag = (wk.start, wk.close_end);
                        let gmap = open.remove(&wk).unwrap_or_default();
                        if !emitted.insert(tag) {
                            continue; // already closed once (recovery guard)
                        }
                        let batch = emit_window(&plan, &out_tagged, &wk, &gmap)?;
                        increment.push(batch.clone());
                        all_emitted.push(batch);
                        closed_count += 1;
                        closed_this += 1;
                        run.events.push_back(event(
                            Some(&flow.target),
                            None,
                            format!("incremental-agg flow `{}`: window [{}, {}) closed & evicted at watermark {w}", flow.name, wk.start, wk.close_end),
                        ));
                    }
                }

                open_series.push(open_size(&open));
                run.increments
                    .entry(flow.target.clone())
                    .or_default()
                    .push(increment);

                // Barrier @ trigger boundary: snapshot accumulator state as epoch i.
                let state = state_to_batch(&plan, &open)?;
                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, &state).await?;
                checkpoints += 1;

                run.events.push_back(event(
                    Some(&flow.target),
                    None,
                    format!("incremental-agg flow `{}` trigger {i}: +{arrived} arrived, watermark {:?}, {closed_this} window(s) closed, {} acc(s) open; checkpoint epoch {i} committed", flow.name, w, open_size(&open)),
                ));
            }

            // End-of-stream flush (only for a fully-consumed stream — a capped
            // run leaves its still-open windows checkpointed for the resume).
            let mut flushed = 0usize;
            if max_triggers.is_none() {
                let remaining: Vec<WinKey> = open.keys().copied().collect();
                for wk in remaining {
                    let tag = (wk.start, wk.close_end);
                    let gmap = open.remove(&wk).unwrap_or_default();
                    if !emitted.insert(tag) {
                        continue;
                    }
                    all_emitted.push(emit_window(&plan, &out_tagged, &wk, &gmap)?);
                    closed_count += 1;
                    flushed += 1;
                }
                open.clear();
                open_series.push(0);
            }

            let final_watermark = tracker.watermark();
            let emitted_rows: usize = all_emitted.iter().map(|b| b.num_rows()).sum();
            materialize(ctx, &flow.target, out_tagged, 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.incremental_agg.insert(flow.target.clone(), true);
            run.events.push_back(event(
                Some(&flow.target),
                None,
                format!(
                    "incremental-agg flow `{}` materialised `{}` ({emitted_rows} row(s), {closed_count} window(s) closed [{flushed} flushed], {late_total} late row(s) dropped, watermark {:?}, {checkpoints} checkpoint(s))",
                    flow.name, flow.target, final_watermark
                ),
            ));

            functional_status(
                "knut-thund/thund_incremental",
                "agg",
                true,
                &format!(
                    "{}: {closed_count} window(s) via maintained accumulators (no per-window SQL), {emitted_rows} row(s)",
                    flow.name
                ),
            );
            functional_status(
                "knut-thund/thund_window",
                "close",
                true,
                &format!(
                    "{}: {closed_count} window(s) closed & evicted (incremental agg)",
                    flow.name
                ),
            );
            if checkpoints > 0 {
                functional_status(
                    "knut-thund/thund_checkpoint",
                    "commit",
                    true,
                    &format!(
                        "{}: {checkpoints} epoch(s) committed (incremental agg)",
                        flow.name
                    ),
                );
            }

            super::apply_expectations(ctx, flow, run).await?;
            Ok(true)
        }
    }

    #[cfg(test)]
    mod wm_tests {
        use super::WatermarkTracker;

        /// Single input: W = max_event_time − allowed_lateness, and it advances
        /// monotonically (a later trigger with a smaller max never lowers W).
        #[test]
        fn single_input_advances_monotonically_with_lateness() {
            let mut t = WatermarkTracker::new(50, 0);
            t.observe_max("s", Some(110), 0);
            assert_eq!(t.advance(0), Some(60), "110 - 50");
            // A trigger whose max (100) would yield W=50 must NOT lower it.
            t.observe_max("s", Some(100), 1);
            assert_eq!(t.advance(1), Some(60), "monotonic: stays at 60");
            // A higher max advances it.
            t.observe_max("s", Some(300), 2);
            assert_eq!(t.advance(2), Some(250), "300 - 50");
        }

        /// Combined watermark is the MIN across inputs (slowest gates progress).
        #[test]
        fn min_across_inputs_gates_progress() {
            let mut t = WatermarkTracker::new(0, 0);
            t.observe_max("fast", Some(1000), 0);
            t.observe_max("slow", Some(200), 0);
            assert_eq!(t.advance(0), Some(200), "the slow input gates W");
        }

        /// An input silent past `idle_timeout_ms` drops out of the min, so the
        /// fast input's progress is no longer held back by the quiet one.
        #[test]
        fn idle_input_stops_gating() {
            let mut t = WatermarkTracker::new(0, 10);
            t.observe_max("fast", Some(1000), 0);
            t.observe_max("slow", Some(200), 0);
            // now_ms=5: neither idle yet → slow still gates.
            assert_eq!(t.advance(5), Some(200));
            // fast keeps producing; slow stays silent past the 10ms timeout.
            t.observe_max("fast", Some(2000), 20);
            // now_ms=20: slow last active at 0 → 20 > 10 → idle → drops out.
            assert_eq!(t.advance(20), Some(2000), "idle slow input no longer gates");
        }

        /// `idle_timeout_ms == 0` means never idle: a quiet input keeps gating.
        #[test]
        fn zero_idle_timeout_never_idles() {
            let mut t = WatermarkTracker::new(0, 0);
            t.observe_max("a", Some(500), 0);
            t.observe_max("b", Some(100), 0);
            assert_eq!(
                t.advance(1_000_000),
                Some(100),
                "b never idles, still gates"
            );
        }
    }
}

#[cfg(all(test, feature = "native"))]
mod native_tests {
    use super::*;
    use crate::ir::{Dataset, Expectation, Flow, OnViolation, OutputType, Pipeline};
    use datafusion::arrow::array::{Int64Array, RecordBatch};
    use datafusion::arrow::datatypes::{DataType, Field, Schema};
    use std::sync::Arc;

    /// A tiny `(id, amount)` input batch: three rows, one with a negative amount.
    fn orders_batch() -> RecordBatch {
        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Int64, false),
            Field::new("amount", DataType::Int64, false),
        ]));
        RecordBatch::try_new(
            schema,
            vec![
                Arc::new(Int64Array::from(vec![1, 2, 3])),
                Arc::new(Int64Array::from(vec![10, -5, 20])),
            ],
        )
        .unwrap()
    }

    /// End to end: seed `raw_orders`, run a batch flow that filters it, then a
    /// second flow that reads the first flow's output — proving the DAG
    /// materialises in-process and downstream flows see upstream results.
    #[test]
    fn batch_dag_runs_end_to_end_on_datafusion() {
        let p = Pipeline::new("orders")
            .with_dataset(Dataset::new("clean_orders", OutputType::MaterializedView))
            .with_dataset(Dataset::new("order_summary", OutputType::MaterializedView))
            .with_flow(
                Flow::batch("f_clean", "clean_orders", ["raw_orders"])
                    .with_query("SELECT id, amount FROM raw_orders WHERE amount > 0"),
            )
            .with_flow(
                Flow::batch("f_sum", "order_summary", ["clean_orders"])
                    .with_query("SELECT count(*) AS n, sum(amount) AS total FROM clean_orders"),
            );

        let backend = NativeBackend::new().with_input("raw_orders", vec![orders_batch()]);
        let mut run = backend.run(&p).expect("pipeline runs");

        // clean_orders keeps the two positive rows.
        assert_eq!(
            run.row_count("clean_orders"),
            2,
            "filtered to positive amounts"
        );
        // order_summary aggregates them: n=2, total=30.
        let summary = run.output("order_summary").expect("summary produced");
        let n = summary[0]
            .column(0)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap()
            .value(0);
        let total = summary[0]
            .column(1)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap()
            .value(0);
        assert_eq!(
            (n, total),
            (2, 30),
            "aggregate over the produced upstream table"
        );

        let events = run.poll_events().unwrap();
        assert!(
            events.iter().any(|e| e.message.contains("complete")),
            "run emits a completion event: {events:?}"
        );
        // Drained once, then empty.
        assert!(run.poll_events().unwrap().is_empty());
    }

    /// A DROP expectation removes violating rows from the materialised output.
    #[test]
    fn drop_expectation_filters_violating_rows() {
        let p = Pipeline::new("orders")
            .with_dataset(Dataset::new("clean_orders", OutputType::MaterializedView))
            .with_flow(
                Flow::batch("f_clean", "clean_orders", ["raw_orders"])
                    .with_query("SELECT id, amount FROM raw_orders")
                    .expect(
                        Expectation::new("amount_positive", "amount > 0").on(OnViolation::Drop),
                    ),
            );
        let backend = NativeBackend::new().with_input("raw_orders", vec![orders_batch()]);
        let mut run = backend.run(&p).expect("runs");
        assert_eq!(
            run.row_count("clean_orders"),
            2,
            "the negative-amount row was dropped"
        );
        assert!(
            run.poll_events()
                .unwrap()
                .iter()
                .any(|e| e.message.contains("DROP") && e.message.contains("dropped 1"))
        );
    }

    /// A FAIL expectation aborts the whole run with a real error.
    #[test]
    fn fail_expectation_aborts_the_run() {
        let p = Pipeline::new("orders")
            .with_dataset(Dataset::new("clean_orders", OutputType::MaterializedView))
            .with_flow(
                Flow::batch("f_clean", "clean_orders", ["raw_orders"])
                    .with_query("SELECT id, amount FROM raw_orders")
                    .expect(
                        Expectation::new("amount_positive", "amount > 0").on(OnViolation::Fail),
                    ),
            );
        let backend = NativeBackend::new().with_input("raw_orders", vec![orders_batch()]);
        let err = backend.run(&p).unwrap_err();
        assert!(matches!(err, crate::error::ThundError::Backend(m) if m.contains("FAILED")));
    }

    /// A streaming flow over a live (Kafka) source is honestly deferred, not
    /// faked — the run still completes for the rest of the graph.
    #[test]
    fn streaming_kafka_flow_is_deferred() {
        use crate::ir::SourceSpec;
        let p = Pipeline::new("clicks")
            .with_dataset(Dataset::new("click_counts", OutputType::Table))
            .with_flow(
                Flow::streaming(
                    "f_clicks",
                    "click_counts",
                    SourceSpec::Kafka {
                        bootstrap: "localhost:9092".into(),
                        topic: "clicks".into(),
                        format: "json".into(),
                    },
                )
                .with_query("SELECT * FROM clicks"),
            );
        let backend = NativeBackend::new();
        let mut run = backend
            .run(&p)
            .expect("run completes despite deferred streaming flow");
        assert_eq!(
            run.row_count("click_counts"),
            0,
            "nothing materialised for the live source"
        );
        assert!(
            run.poll_events()
                .unwrap()
                .iter()
                .any(|e| e.message.contains("deferred"))
        );
    }

    /// A single-column `(id)` streaming batch.
    fn id_batch(ids: &[i64]) -> RecordBatch {
        let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)]));
        RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(ids.to_vec()))]).unwrap()
    }

    /// Phase 0 parity: the default mode is recompute and advertises no
    /// incremental state; `Incremental` advertises it only WITH a `state_root`.
    #[test]
    fn default_streaming_mode_is_recompute_and_incremental_state_capability() {
        let b = NativeBackend::new();
        assert_eq!(
            b.streaming_mode,
            StreamingMode::Recompute,
            "default is recompute"
        );
        assert!(
            !b.capabilities().incremental_state,
            "recompute advertises no incremental state"
        );

        // Incremental requested but no state_root → still advertises false (degrades).
        let inc = NativeBackend::new().with_streaming_mode(StreamingMode::Incremental);
        assert!(
            !inc.capabilities().incremental_state,
            "no state_root → advertises false"
        );

        // Incremental WITH a state_root → the capability is advertised true.
        let inc_root = NativeBackend::new()
            .with_streaming_mode(StreamingMode::Incremental)
            .with_state_root("file:///tmp/thund-state");
        assert!(
            inc_root.capabilities().incremental_state,
            "mode + state_root → advertised"
        );
    }

    /// Phase 0: `Incremental` with no `state_root` degrades to the recompute
    /// path — byte-identical result — and logs the degrade honestly.
    #[test]
    fn incremental_without_state_root_degrades_to_recompute() {
        use crate::ir::{FlowKind, OutputMode, SourceSpec, Trigger};
        let mk = || {
            let mut flow = Flow::streaming(
                "f",
                "out",
                SourceSpec::Kafka {
                    bootstrap: "b".into(),
                    topic: "events".into(),
                    format: "json".into(),
                },
            )
            .with_query("SELECT id FROM events");
            if let FlowKind::Streaming {
                trigger,
                output_mode,
                ..
            } = &mut flow.kind
            {
                *trigger = Trigger::AvailableNow;
                *output_mode = OutputMode::Append;
            }
            Pipeline::new("p")
                .with_dataset(Dataset::new("out", OutputType::Table))
                .with_flow(flow)
        };
        let streams = || vec![vec![id_batch(&[1, 2])], vec![id_batch(&[3])]];

        let recompute = NativeBackend::new()
            .with_stream_input("events", streams())
            .run(&mk())
            .expect("recompute runs");
        let mut incremental = NativeBackend::new()
            .with_streaming_mode(StreamingMode::Incremental)
            .with_stream_input("events", streams())
            .run(&mk())
            .expect("incremental (degraded) runs");

        assert_eq!(recompute.row_count("out"), 3);
        assert_eq!(
            incremental.row_count("out"),
            recompute.row_count("out"),
            "incremental without a state_root equals the recompute path"
        );
        assert!(
            incremental
                .poll_events()
                .unwrap()
                .iter()
                .any(|e| e.message.contains("degrading to recompute")),
            "the degrade is logged honestly"
        );
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Graph-sink (FalkorDB) tests — the executable rel2graph path.
//
// The SEAM the coordinator asked for: given a graph `OutputType::Sink` dataset +
// `knut.*` props → the expected `UNWIND … MERGE` ops, asserted at the
// `GraphMutation` / graphar-falkordb-call level WITHOUT a live database. A live
// FalkorDB round-trip is additionally provided, honestly `#[ignore]`d (it needs
// a Valkey/FalkorDB container).
// ═══════════════════════════════════════════════════════════════════════════
#[cfg(all(test, feature = "native", feature = "rel2graph"))]
mod graph_sink_tests {
    use super::graph_sink::{GraphElem, edge_edge_set, ext, graph_elem, node_vertex_set};
    use super::*;
    use crate::ir::{Dataset, Flow, OutputType, Pipeline};
    use crate::rel2graph::prop;
    use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray};
    use datafusion::arrow::datatypes::{DataType, Field, Schema};
    use std::sync::Arc;

    /// A `node__User` graph-sink dataset (the shape `to_thund_pipeline` emits).
    fn node_dataset() -> Dataset {
        Dataset::new("node__User", OutputType::Sink)
            .with_format("falkordb-node")
            .with_properties([
                (prop::NODE_LABEL, "User"),
                (prop::NODE_KEY, "user_id"),
                (prop::NODE_PROPS, "name,age"),
                (prop::NODE_TABLE, "users"),
                (prop::SINK, "falkordb"),
            ])
    }

    /// An `edge__User__WORKS_AT__Company` graph-sink dataset.
    fn edge_dataset() -> Dataset {
        Dataset::new("edge__User__WORKS_AT__Company", OutputType::Sink)
            .with_format("falkordb-edge")
            .with_properties([
                (prop::EDGE_REL, "WORKS_AT"),
                (prop::EDGE_FROM, "User"),
                (prop::EDGE_TO, "Company"),
                (prop::EDGE_FROM_KEY, "user_id"),
                (prop::EDGE_TO_KEY, "company_id"),
                (prop::EDGE_TABLE, "employments"),
                (prop::SINK, "falkordb"),
            ])
    }

    fn users_batch() -> RecordBatch {
        RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("user_id", DataType::Int64, false),
                Field::new("name", DataType::Utf8, false),
            ])),
            vec![
                Arc::new(Int64Array::from(vec![1, 2])),
                Arc::new(StringArray::from(vec!["alice", "bob"])),
            ],
        )
        .unwrap()
    }

    /// The edge projection's materialised batch: `[from_key, to_key, ...props]`.
    fn employments_batch() -> RecordBatch {
        RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("user_id", DataType::Int64, false),
                Field::new("company_id", DataType::Int64, false),
                Field::new("since", DataType::Utf8, false),
            ])),
            vec![
                Arc::new(Int64Array::from(vec![1, 2])),
                Arc::new(Int64Array::from(vec![10, 10])),
                Arc::new(StringArray::from(vec!["2021", "2022"])),
            ],
        )
        .unwrap()
    }

    /// Graph-sink classification: a `falkordb-node`/`-edge` Sink carrying the
    /// matching `knut.*` prop is a graph element; a plain Table (or a Sink with an
    /// unrelated format) is not — so the graph write never hijacks a file sink.
    #[test]
    fn classifies_graph_sink_datasets() {
        assert_eq!(graph_elem(&node_dataset()), Some(GraphElem::Node));
        assert_eq!(graph_elem(&edge_dataset()), Some(GraphElem::Edge));
        // A normal table is not a graph sink.
        assert_eq!(graph_elem(&Dataset::new("t", OutputType::Table)), None);
        // A Sink with a plain file format is not a graph sink.
        assert_eq!(
            graph_elem(&Dataset::new("f", OutputType::Sink).with_format("parquet")),
            None
        );
        // A `falkordb-node` format WITHOUT the node prop is not classified (the
        // prop is required, not just the format).
        assert_eq!(
            graph_elem(&Dataset::new("x", OutputType::Sink).with_format("falkordb-node")),
            None
        );
    }

    /// A node sink maps to a `VertexSet` MERGE-keyed on `knut.node.key` — the same
    /// idempotent upsert (`MERGE (n:User {user_id: ...}) SET n += r`) the PySpark
    /// `_write_nodes` projection emits. The batch passes through verbatim.
    #[test]
    fn node_sink_maps_to_vertex_set() {
        let vs = node_vertex_set(&node_dataset(), users_batch()).unwrap();
        assert_eq!(vs.label, "User");
        assert_eq!(vs.id_column, "user_id", "MERGE key = knut.node.key");
        assert_eq!(vs.batch.num_rows(), 2);
        // The id column plus every prop column ride along for `SET n += r`.
        let schema = vs.batch.schema();
        let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(cols, vec!["user_id", "name"]);
    }

    /// An edge sink maps to an `EdgeSet`: endpoints MATCHed on `from_key`/`to_key`
    /// (`src_id_column`/`dst_id_column`), and the batch's `from_key`/`to_key`
    /// columns renamed to the `src`/`dst` carriers bifröst's MERGE expects — the
    /// exact shape of the PySpark `from_key AS src, to_key AS dst` +
    /// `MERGE (a:User {user_id: r.src}) MERGE (b:Company {company_id: r.dst})
    /// MERGE (a)-[e:WORKS_AT]->(b) SET e += r.props` projection.
    #[test]
    fn edge_sink_maps_to_edge_set_with_renamed_endpoints() {
        let es = edge_edge_set(&edge_dataset(), employments_batch()).unwrap();
        assert_eq!(es.src_label, "User");
        assert_eq!(es.edge_type, "WORKS_AT");
        assert_eq!(es.dst_label, "Company");
        assert_eq!(
            es.src_id_column, "user_id",
            "endpoint MATCH prop = from_key"
        );
        assert_eq!(
            es.dst_id_column, "company_id",
            "endpoint MATCH prop = to_key"
        );
        // from_key -> src, to_key -> dst, `since` stays an edge property.
        let schema = es.batch.schema();
        let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(cols, vec!["src", "dst", "since"]);
        assert_eq!(es.batch.num_rows(), 2);
    }

    /// **Self-referential / renamed-FK edge.** When the FK read column differs
    /// from the node merge key — `Employee-REPORTS_TO->Employee` reads the
    /// endpoint from `manager_id` but MERGEs both endpoints on `employee_id` —
    /// the sink must rename the *read columns* (`from_col`/`to_col`) to `src`/
    /// `dst` while the MATCH properties stay the *merge keys* (`from_key`/
    /// `to_key`). The old single-key model read `employee_id AS dst`, collapsing
    /// every edge into a self-loop; this pins the split at the executable seam.
    #[test]
    fn edge_sink_self_ref_reads_fk_column_merges_on_node_key() {
        let ds = Dataset::new("edge__Employee__REPORTS_TO__Employee", OutputType::Sink)
            .with_format("falkordb-edge")
            .with_properties([
                (prop::EDGE_REL, "REPORTS_TO"),
                (prop::EDGE_FROM, "Employee"),
                (prop::EDGE_TO, "Employee"),
                // Merge keys — both endpoints identified by the Employee node key.
                (prop::EDGE_FROM_KEY, "employee_id"),
                (prop::EDGE_TO_KEY, "employee_id"),
                // Read columns — the value comes from employee_id → manager_id.
                (prop::EDGE_FROM_COL, "employee_id"),
                (prop::EDGE_TO_COL, "manager_id"),
                (prop::EDGE_TABLE, "erp.hr.employees"),
                (prop::SINK, "falkordb"),
            ]);
        // The projection flow projects the FK columns [employee_id, manager_id].
        let batch = RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("employee_id", DataType::Int64, false),
                Field::new("manager_id", DataType::Int64, true),
            ])),
            vec![
                Arc::new(Int64Array::from(vec![4, 5])),
                Arc::new(Int64Array::from(vec![1, 2])),
            ],
        )
        .unwrap();

        let es = edge_edge_set(&ds, batch).unwrap();
        // MATCH props are the node MERGE KEY on BOTH ends (not the FK column).
        assert_eq!(es.src_id_column, "employee_id");
        assert_eq!(es.dst_id_column, "employee_id");
        // The renamed batch carries the VALUES read from employee_id / manager_id.
        let schema = es.batch.schema();
        let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(cols, vec!["src", "dst"], "from_col→src, to_col→dst");
        let src = es
            .batch
            .column(0)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        let dst = es
            .batch
            .column(1)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        assert_eq!(src.values(), &[4, 5], "src = employee_id (the reporter)");
        assert_eq!(
            dst.values(),
            &[1, 2],
            "dst = manager_id (NOT employee_id: no self-loop)"
        );
        // The regression guard: src and dst are genuinely different columns.
        assert_ne!(
            src.values(),
            dst.values(),
            "REPORTS_TO is not an identity self-loop"
        );
    }

    /// A missing endpoint key column fails LOUDLY (a projection/schema mismatch is
    /// never a silent drop), mirroring the file-sink loud-gap law.
    #[test]
    fn edge_sink_missing_endpoint_column_errors() {
        // A batch that lacks `company_id` (the to_key).
        let bad = RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("user_id", DataType::Int64, false),
                Field::new("since", DataType::Utf8, false),
            ])),
            vec![
                Arc::new(Int64Array::from(vec![1])),
                Arc::new(StringArray::from(vec!["2021"])),
            ],
        )
        .unwrap();
        let err = edge_edge_set(&edge_dataset(), bad).unwrap_err();
        assert!(
            err.to_string().contains("company_id") && err.to_string().contains("NOT FOUND"),
            "loud on the missing to_key column, got: {err}"
        );
    }

    // ── Enhancement 1: composite + hashed node keys (open-Q #3) ──────────────

    /// Clone a dataset and add/override the given `knut.*` properties (the
    /// builder `with_properties` REPLACES the map, so tests that layer an
    /// enrichment prop onto a base dataset insert directly).
    fn with_ext<'a>(
        base: &Dataset,
        extra: impl IntoIterator<Item = (&'a str, &'a str)>,
    ) -> Dataset {
        let mut ds = base.clone();
        for (k, v) in extra {
            ds.properties.insert(k.to_string(), v.to_string());
        }
        ds
    }

    /// A COMPOSITE node key (`knut.node.keys`) makes the sink MERGE on a
    /// synthesised surrogate column keyed on the joined value of ALL named
    /// columns — instead of the single `knut.node.key`. The individual key
    /// columns stay in the batch as properties (still queryable), and the
    /// surrogate column carries the canonical `col1␟col2` join.
    #[test]
    fn node_composite_key_synthesizes_raw_surrogate() {
        let ds = with_ext(&node_dataset(), [(ext::NODE_KEYS, "user_id,name")]);
        let vs = node_vertex_set(&ds, users_batch()).unwrap();
        assert_eq!(
            vs.id_column, "knut_key",
            "MERGE key is the synthesised surrogate"
        );
        let schema = vs.batch.schema();
        let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(
            cols,
            vec!["user_id", "name", "knut_key"],
            "source key columns retained as properties; surrogate appended"
        );
        let key_idx = schema.index_of("knut_key").unwrap();
        let keys = vs
            .batch
            .column(key_idx)
            .as_any()
            .downcast_ref::<datafusion::arrow::array::StringArray>()
            .unwrap();
        // users_batch: (1, "alice"), (2, "bob") → "1␟alice", "2␟bob".
        assert_eq!(keys.value(0), "1\u{1f}alice");
        assert_eq!(keys.value(1), "2\u{1f}bob");
    }

    /// The OPTIONAL hashed surrogate (`knut.node.key_hash`) reduces the composite
    /// value to a compact, deterministic 64-bit FNV-1a hex key. Distinct rows →
    /// distinct keys; the same input always hashes the same (stable identity).
    #[test]
    fn node_hashed_composite_key_is_deterministic() {
        let ds = with_ext(
            &node_dataset(),
            [
                (ext::NODE_KEYS, "user_id,name"),
                (ext::NODE_KEY_HASH, "true"),
                (ext::NODE_KEY_COL, "uid"),
            ],
        );
        let vs = node_vertex_set(&ds, users_batch()).unwrap();
        assert_eq!(vs.id_column, "uid", "custom surrogate column name honoured");
        let idx = vs.batch.schema().index_of("uid").unwrap();
        let keys = vs
            .batch
            .column(idx)
            .as_any()
            .downcast_ref::<datafusion::arrow::array::StringArray>()
            .unwrap();
        assert_eq!(keys.value(0).len(), 16, "16-hex 64-bit FNV-1a surrogate");
        assert_ne!(
            keys.value(0),
            keys.value(1),
            "distinct rows → distinct keys"
        );
        // Deterministic: a second projection of the same input yields the same key.
        let again = node_vertex_set(&ds, users_batch()).unwrap();
        let again_idx = again.batch.schema().index_of("uid").unwrap();
        let again_keys = again
            .batch
            .column(again_idx)
            .as_any()
            .downcast_ref::<datafusion::arrow::array::StringArray>()
            .unwrap();
        assert_eq!(
            keys.value(0),
            again_keys.value(0),
            "hash is stable across runs"
        );
    }

    // ── Enhancement 2: parallel / property-keyed edges (open-Q #4) ───────────

    /// `knut.edge.keys` folds edge PROPERTY columns into the relationship MERGE
    /// key so parallel edges survive; unset ⇒ empty (collapse, historical). A
    /// key naming the `src`/`dst` carrier, or a column absent from the output,
    /// fails loudly (a projection bug, never a silent drop).
    #[test]
    fn edge_property_keyed_merge_key() {
        // Unset → collapse (empty merge key), historical behaviour.
        let plain = edge_edge_set(&edge_dataset(), employments_batch()).unwrap();
        assert!(
            plain.edge_key_props.is_empty(),
            "no `knut.edge.keys` ⇒ one edge per pair"
        );

        // `since` folded in → parallel edges survive.
        let ds = with_ext(&edge_dataset(), [(ext::EDGE_KEYS, "since")]);
        let es = edge_edge_set(&ds, employments_batch()).unwrap();
        assert_eq!(es.edge_key_props, vec!["since".to_string()]);
        // `since` is still an edge property column (SET e += r.props writes it).
        let schema = es.batch.schema();
        let cols: Vec<&str> = schema.fields().iter().map(|f| f.name().as_str()).collect();
        assert_eq!(cols, vec!["src", "dst", "since"]);

        // Naming the endpoint carrier is loud.
        let bad_ep = with_ext(&edge_dataset(), [(ext::EDGE_KEYS, "src")]);
        assert!(edge_edge_set(&bad_ep, employments_batch()).is_err());
        // Naming a non-existent column is loud.
        let bad_col = with_ext(&edge_dataset(), [(ext::EDGE_KEYS, "nope")]);
        assert!(edge_edge_set(&bad_col, employments_batch()).is_err());
    }

    // ── Enhancement 3: typed property projection (open-Q #7) ─────────────────

    /// `knut.node.cast` typed-projects (casts) named property columns before the
    /// load, replacing the Arrow passthrough. A `user_id` arriving as Utf8 is
    /// cast to Int64; an unknown type token or column fails loudly.
    #[test]
    fn typed_property_projection_casts_columns() {
        // A batch whose `user_id` is a STRING that must become an Int64 key.
        let batch = RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("user_id", DataType::Utf8, false),
                Field::new("name", DataType::Utf8, false),
            ])),
            vec![
                Arc::new(StringArray::from(vec!["1", "2"])),
                Arc::new(StringArray::from(vec!["alice", "bob"])),
            ],
        )
        .unwrap();
        let ds = with_ext(&node_dataset(), [(ext::NODE_CAST, "user_id:int64")]);
        let vs = node_vertex_set(&ds, batch.clone()).unwrap();
        let idx = vs.batch.schema().index_of("user_id").unwrap();
        assert_eq!(
            vs.batch.schema().field(idx).data_type(),
            &DataType::Int64,
            "typed projection cast user_id Utf8 → Int64"
        );
        let ids = vs
            .batch
            .column(idx)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        assert_eq!(ids.values(), &[1, 2]);

        // Unknown type token → loud.
        let bad_ty = with_ext(&node_dataset(), [(ext::NODE_CAST, "user_id:blob")]);
        assert!(node_vertex_set(&bad_ty, batch.clone()).is_err());
        // Cast an unknown column → loud.
        let bad_col = with_ext(&node_dataset(), [(ext::NODE_CAST, "ghost:int64")]);
        assert!(node_vertex_set(&bad_col, batch).is_err());
    }

    /// A minimal rel2graph-shaped pipeline: two graph Sink datasets fed by
    /// projection flows over seeded inputs. Nodes then edges, `falkordb-node` /
    /// `falkordb-edge` formats carrying the `knut.*` props.
    fn graph_pipeline() -> Pipeline {
        Pipeline::new("rel2graph")
            .with_dataset(node_dataset())
            .with_dataset(edge_dataset())
            .with_flow(
                Flow::batch("flow_node", "node__User", ["users"])
                    .with_projection(["user_id", "name"]),
            )
            .with_flow(
                Flow::batch(
                    "flow_edge",
                    "edge__User__WORKS_AT__Company",
                    ["employments"],
                )
                .with_projection(["user_id", "company_id", "since"]),
            )
    }

    /// END-TO-END CONSTRUCTION (no DB): the NativeBackend runs the projection
    /// flows, materialises both graph Sink datasets, and — with no FalkorDB target
    /// — CONSTRUCTS the graph mutation and records each element on the run handle
    /// as `written = false` (honest construction-only, never a silent fake). This
    /// is "bifröst mapping → thund → NativeBackend" proven executable up to the
    /// graph write, at the seam, with no container.
    #[test]
    fn native_backend_constructs_graph_sinks_without_a_target() {
        let run = NativeBackend::new()
            .with_input("users", vec![users_batch()])
            .with_input("employments", vec![employments_batch()])
            .run(&graph_pipeline())
            .expect("the rel2graph pipeline runs on DataFusion");

        let node = run
            .graph_sink("node__User")
            .expect("the node graph sink was resolved");
        assert_eq!(node.kind, GraphElem::Node);
        assert_eq!(node.element, "User");
        assert_eq!(node.rows, 2);
        assert!(!node.written, "no target → constructed only");

        let edge = run
            .graph_sink("edge__User__WORKS_AT__Company")
            .expect("the edge graph sink was resolved");
        assert_eq!(edge.kind, GraphElem::Edge);
        assert_eq!(edge.element, "WORKS_AT");
        assert_eq!(edge.rows, 2);
        assert!(!edge.written);
    }

    /// LIVE round-trip (needs a FalkorDB/Valkey at `redis://127.0.0.1:6379`).
    /// `#[ignore]`d honestly — it drives the real bifröst FalkorDbSink and MERGEs
    /// the graph. Run with:
    ///   `cargo test -p knut-thund --features falkordb graph_sink -- --ignored`
    #[test]
    #[ignore = "needs a live FalkorDB/Valkey container"]
    fn native_backend_loads_into_live_falkordb() {
        let run = NativeBackend::new()
            .with_input("users", vec![users_batch()])
            .with_input("employments", vec![employments_batch()])
            .with_falkordb_sink("redis://127.0.0.1:6379", "thund_graph_sink_test")
            .run(&graph_pipeline())
            .expect("the rel2graph pipeline runs and MERGEs into FalkorDB");

        assert!(
            run.graph_sink("node__User").is_some_and(|g| g.written),
            "nodes MERGEd into the live graph"
        );
        assert!(
            run.graph_sink("edge__User__WORKS_AT__Company")
                .is_some_and(|g| g.written),
            "edges MERGEd into the live graph"
        );
    }
}