knut-thund 0.1.1

//! 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;

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

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
    }

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

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

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

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.
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 DataFusion-backed execution engine (feature `native`).
#[cfg(feature = "native")]
mod exec {
    use super::{event, NativeBackend, NativeRun};
    use crate::error::{Result, ThundError};
    use crate::ir::{
        CdcSpec, Flow, FlowKind, OnViolation, OutputMode, Pipeline, ScdType, SourceSpec, Trigger,
    };
    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 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)?;
        let mut ran = 0usize;
        for target in &order {
            for f in pipeline.flows.iter().filter(|f| &f.target == target) {
                match run_flow(&ctx, f, &backend.seed_streams, &mut run).await {
                    Ok(true) => ran += 1,
                    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);
                    }
                }
            }
        }

        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,
        };
        let path = path.strip_prefix("file://").unwrap_or(&path).to_string();
        let res = match format.as_str() {
            "parquet" => ctx
                .register_parquet(&name, &path, ParquetReadOptions::default())
                .await,
            "csv" => ctx.register_csv(&name, &path, CsvReadOptions::default()).await,
            other => {
                events.push_back(event(
                    Some(&name),
                    None,
                    format!("source `{name}`: format `{other}` not registerable (need parquet/csv)"),
                ));
                return;
            }
        };
        match res {
            Ok(()) => events.push_back(event(
                Some(&name),
                None,
                format!("registered bounded source `{name}` <- {path}"),
            )),
            Err(e) => events.push_back(event(
                Some(&name),
                None,
                format!("source `{name}` not registered ({e}); flow will defer"),
            )),
        }
    }

    /// 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.
    async fn run_flow(
        ctx: &SessionContext,
        flow: &Flow,
        streams: &[(String, Vec<Vec<RecordBatch>>)],
        run: &mut NativeRun,
    ) -> Result<bool> {
        let Some(query) = flow.query.as_deref() else {
            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 { .. }) {
                if let Some(name) = stream_source_name(source) {
                    if let Some((_, micro_batches)) = streams.iter().find(|(n, _)| n == &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 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 query.
        let df = match ctx.sql(query).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)),
        };
        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)
    }

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

    /// 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)
    }
}

#[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")));
    }
}