knut-thund 0.1.4

//! Integration proof for the native backend's **micro-batch streaming** layer.
//!
//! Drives a *synthetic bounded-but-streamed* source (a range/memory source fed
//! in N micro-batches via `NativeBackend::with_stream_input`) through the REAL
//! DataFusion engine — no live Kafka/socket, no heavy run. Asserts:
//!   * incremental batches are produced in order (one increment per trigger),
//!   * the final materialised result equals the batch path over the same data,
//!   * `AvailableNow` collapses to a single bounded pass,
//!   * drop / fail data-quality expectations behave on the streaming result.
//!
//! Gated on `--features native` (the DataFusion tree); a no-op otherwise.
#![cfg(feature = "native")]

use knut_thund::backend::{
    ExecBackend, RunHandle,
    native::{NativeBackend, StreamingMode},
};
use knut_thund::ir::{
    CdcSpec, Dataset, Expectation, Flow, FlowKind, OnViolation, OutputMode, OutputType, Pipeline,
    ScdType, SourceSpec, Trigger, Watermark, WindowSpec,
};

use datafusion::arrow::array::{Array, Int64Array, RecordBatch, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;

/// An `(id, v)` batch from parallel id/value slices.
fn batch(ids: &[i64], vs: &[i64]) -> RecordBatch {
    let schema = Arc::new(Schema::new(vec![
        Field::new("id", DataType::Int64, false),
        Field::new("v", DataType::Int64, false),
    ]));
    RecordBatch::try_new(
        schema,
        vec![
            Arc::new(Int64Array::from(ids.to_vec())),
            Arc::new(Int64Array::from(vs.to_vec())),
        ],
    )
    .unwrap()
}

/// Three micro-batches of the `events` source.
fn micro_batches() -> Vec<Vec<RecordBatch>> {
    vec![
        vec![batch(&[1, 2], &[10, -5])],
        vec![batch(&[3, 4], &[20, 0])],
        vec![batch(&[5], &[30])],
    ]
}

/// All the same rows, concatenated — for the batch-path comparison.
fn all_rows() -> RecordBatch {
    batch(&[1, 2, 3, 4, 5], &[10, -5, 20, 0, 30])
}

/// A streaming Kafka flow over the seeded `events` topic, running `query` with
/// `output_mode` and `trigger`.
fn stream_pipeline(query: &str, output_mode: OutputMode, trigger: Trigger) -> Pipeline {
    let mut flow = Flow::streaming(
        "f_events",
        "events_out",
        SourceSpec::Kafka {
            bootstrap: "localhost:9092".into(),
            topic: "events".into(),
            format: "json".into(),
        },
    )
    .with_query(query);
    if let FlowKind::Streaming {
        output_mode: om,
        trigger: tr,
        ..
    } = &mut flow.kind
    {
        *om = output_mode;
        *tr = trigger;
    }
    Pipeline::new("stream")
        .with_dataset(Dataset::new("events_out", OutputType::Table).incremental())
        .with_flow(flow)
}

/// The batch-path result of `query` over all rows seen at once.
fn batch_result(query: &str) -> Vec<i64> {
    let p = Pipeline::new("batch")
        .with_dataset(Dataset::new("events_out", OutputType::MaterializedView))
        .with_flow(Flow::batch("f_batch", "events_out", ["events"]).with_query(query));
    let run = NativeBackend::new()
        .with_input("events", vec![all_rows()])
        .run(&p)
        .expect("batch runs");
    sorted_ids(run.output("events_out").expect("batch output"))
}

/// Sorted `id` column across a slice of batches.
fn sorted_ids(batches: &[RecordBatch]) -> Vec<i64> {
    let mut v: Vec<i64> = batches
        .iter()
        .flat_map(|b| {
            b.column(0)
                .as_any()
                .downcast_ref::<Int64Array>()
                .unwrap()
                .values()
                .to_vec()
        })
        .collect();
    v.sort_unstable();
    v
}

/// Append mode over a row-wise filter: each trigger emits exactly the newly
/// arrived rows that pass, in order; the concatenated increments and the final
/// materialised table both equal the batch path.
#[test]
fn append_mode_emits_new_rows_per_trigger_and_matches_batch() {
    let query = "SELECT id, v FROM events WHERE v > 0";
    let p = stream_pipeline(query, OutputMode::Append, Trigger::Continuous);
    let run = NativeBackend::new()
        .with_stream_input("events", micro_batches())
        .run(&p)
        .expect("streaming run completes");

    // One increment per micro-batch, in order.
    assert_eq!(
        run.trigger_count("events_out"),
        3,
        "one increment per micro-batch"
    );
    let incs = run.increments("events_out");
    // Trigger 0: rows (1,10),(2,-5) -> only id 1 passes v>0.
    assert_eq!(sorted_ids(&incs[0]), vec![1]);
    // Trigger 1: rows (3,20),(4,0) -> only id 3 passes.
    assert_eq!(sorted_ids(&incs[1]), vec![3]);
    // Trigger 2: row (5,30) -> id 5 passes.
    assert_eq!(sorted_ids(&incs[2]), vec![5]);

    // Concatenated increments == final materialised == batch path.
    let concat: Vec<i64> = {
        let mut all: Vec<i64> = incs.iter().flat_map(|i| sorted_ids(i)).collect();
        all.sort_unstable();
        all
    };
    let final_rows = sorted_ids(run.output("events_out").expect("final output"));
    assert_eq!(concat, final_rows, "increments reconstruct the final table");
    assert_eq!(
        final_rows,
        batch_result(query),
        "final == batch path over same data"
    );
}

/// Complete mode over an aggregation: each trigger emits the full running
/// aggregate; the final table equals the batch path.
#[test]
fn complete_mode_emits_full_aggregate_each_trigger_and_matches_batch() {
    let query = "SELECT count(*) AS n, sum(v) AS total FROM events";
    let p = stream_pipeline(query, OutputMode::Complete, Trigger::Continuous);
    let run = NativeBackend::new()
        .with_stream_input("events", micro_batches())
        .run(&p)
        .expect("streaming run completes");

    assert_eq!(
        run.trigger_count("events_out"),
        3,
        "one increment per micro-batch"
    );
    let incs = run.increments("events_out");
    // Complete mode: every increment is the full result (a single aggregate row).
    for inc in incs {
        let rows: usize = inc.iter().map(|b| b.num_rows()).sum();
        assert_eq!(rows, 1, "complete-mode aggregate emits one row per trigger");
    }
    // Running counts grow monotonically: n = 2, 4, 5.
    let n = |inc: &Vec<RecordBatch>| {
        inc[0]
            .column(0)
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap()
            .value(0)
    };
    assert_eq!(n(&incs[0]), 2);
    assert_eq!(n(&incs[1]), 4);
    assert_eq!(n(&incs[2]), 5);

    // Final aggregate: n=5, total=10+(-5)+20+0+30 = 55.
    let out = run.output("events_out").expect("final output");
    let total = out[0]
        .column(1)
        .as_any()
        .downcast_ref::<Int64Array>()
        .unwrap()
        .value(0);
    assert_eq!((n(&out.to_vec()), total), (5, 55));
}

/// `AvailableNow` collapses all currently-available data into ONE bounded pass.
#[test]
fn available_now_is_a_single_bounded_pass() {
    let query = "SELECT id, v FROM events WHERE v > 0";
    let p = stream_pipeline(query, OutputMode::Append, Trigger::AvailableNow);
    let run = NativeBackend::new()
        .with_stream_input("events", micro_batches())
        .run(&p)
        .expect("streaming run completes");

    assert_eq!(
        run.trigger_count("events_out"),
        1,
        "AvailableNow => single pass"
    );
    let final_rows = sorted_ids(run.output("events_out").expect("final output"));
    assert_eq!(
        final_rows,
        batch_result(query),
        "single pass still equals batch path"
    );
    // The single increment holds every passing row.
    assert_eq!(sorted_ids(&run.increments("events_out")[0]), vec![1, 3, 5]);
}

/// A DROP expectation removes violating rows from the streaming result, just
/// like the batch path.
#[test]
fn drop_expectation_filters_streaming_result() {
    let query = "SELECT id, v FROM events";
    let mut p = stream_pipeline(query, OutputMode::Append, Trigger::AvailableNow);
    p.flows[0]
        .expectations
        .push(Expectation::new("v_positive", "v > 0").on(OnViolation::Drop));
    let mut run = NativeBackend::new()
        .with_stream_input("events", micro_batches())
        .run(&p)
        .expect("streaming run completes");
    // Rows with v>0: ids 1,3,5 (ids 2 (-5) and 4 (0) dropped).
    assert_eq!(
        sorted_ids(run.output("events_out").expect("final output")),
        vec![1, 3, 5]
    );
    assert!(
        run.poll_events()
            .unwrap()
            .iter()
            .any(|e| e.message.contains("DROP"))
    );
}

/// A FAIL expectation aborts the streaming run with a real error.
#[test]
fn fail_expectation_aborts_streaming_run() {
    let query = "SELECT id, v FROM events";
    let mut p = stream_pipeline(query, OutputMode::Append, Trigger::AvailableNow);
    p.flows[0]
        .expectations
        .push(Expectation::new("v_positive", "v > 0").on(OnViolation::Fail));
    let err = NativeBackend::new()
        .with_stream_input("events", micro_batches())
        .run(&p)
        .unwrap_err();
    assert!(matches!(err, knut_thund::ThundError::Backend(m) if m.contains("FAILED")));
}

/// A downstream BATCH flow reads the materialised streaming table by name —
/// proving the streaming result registers into the shared in-process catalog.
#[test]
fn downstream_batch_flow_reads_streaming_result() {
    let mut flow = Flow::streaming(
        "f_events",
        "events_out",
        SourceSpec::Kafka {
            bootstrap: "localhost:9092".into(),
            topic: "events".into(),
            format: "json".into(),
        },
    )
    .with_query("SELECT id, v FROM events WHERE v > 0");
    if let FlowKind::Streaming { trigger, .. } = &mut flow.kind {
        *trigger = Trigger::AvailableNow;
    }
    let p = Pipeline::new("stream_then_batch")
        .with_dataset(Dataset::new("events_out", OutputType::Table).incremental())
        .with_dataset(Dataset::new("summary", OutputType::MaterializedView))
        .with_flow(flow)
        .with_flow(
            Flow::batch("f_sum", "summary", ["events_out"])
                .with_query("SELECT count(*) AS n, sum(v) AS total FROM events_out"),
        );
    let run = NativeBackend::new()
        .with_stream_input("events", micro_batches())
        .run(&p)
        .expect("mixed streaming+batch run completes");
    // Positive rows: v = 10, 20, 30 -> n=3, total=60.
    let out = run.output("summary").expect("summary produced");
    let n = out[0]
        .column(0)
        .as_any()
        .downcast_ref::<Int64Array>()
        .unwrap()
        .value(0);
    let total = out[0]
        .column(1)
        .as_any()
        .downcast_ref::<Int64Array>()
        .unwrap()
        .value(0);
    assert_eq!((n, total), (3, 60));
}

/// With NO seeded stream (and no live infra), a Kafka flow is still deferred
/// honestly — micro-batch streaming is opt-in via `with_stream_input`.
#[test]
fn unseeded_kafka_flow_still_defers() {
    let p = stream_pipeline(
        "SELECT * FROM events",
        OutputMode::Append,
        Trigger::Continuous,
    );
    let mut run = NativeBackend::new()
        .run(&p)
        .expect("run completes despite deferral");
    assert_eq!(
        run.row_count("events_out"),
        0,
        "nothing materialised for an unseeded live source"
    );
    assert!(
        run.poll_events()
            .unwrap()
            .iter()
            .any(|e| e.message.contains("deferred"))
    );
}

// ── Incremental mode: event-time watermark + late-row drop (phase 1) ────────
//
// The incremental streaming path advances an event-time watermark
// `W = max(event_time) − allowed_lateness` (monotonic) and drops rows whose
// event time falls below `W` before the query sees them, counting them as
// `late_rows_dropped`. Recompute (the default) keeps every row — so the two
// paths diverge exactly by the dropped late rows, the whole point of the mode.

/// An `(id, et)` batch: `id` identity, `et` the event-time column (epoch ms).
fn evt_batch(ids: &[i64], ets: &[i64]) -> RecordBatch {
    let schema = Arc::new(Schema::new(vec![
        Field::new("id", DataType::Int64, false),
        Field::new("et", DataType::Int64, false),
    ]));
    RecordBatch::try_new(
        schema,
        vec![
            Arc::new(Int64Array::from(ids.to_vec())),
            Arc::new(Int64Array::from(ets.to_vec())),
        ],
    )
    .unwrap()
}

/// Three triggers whose event times force exactly one late row: after the
/// watermark reaches 250 (trigger 2, max 300, lateness 50), the id=5 row at
/// et=120 is late and dropped; id=6 at et=300 is on time.
fn wm_micro_batches() -> Vec<Vec<RecordBatch>> {
    vec![
        vec![evt_batch(&[1, 2], &[100, 110])], // W -> 60
        vec![evt_batch(&[3, 4], &[200, 210])], // W -> 160
        vec![evt_batch(&[5, 6], &[120, 300])], // W -> 250; id=5 (et 120) is late
    ]
}

/// A streaming flow over `events` with an event-time watermark on `et`
/// (allowed lateness 50ms), append output, per-trigger cadence.
fn wm_pipeline() -> Pipeline {
    let mut flow = Flow::streaming(
        "f_events",
        "events_out",
        SourceSpec::Kafka {
            bootstrap: "localhost:9092".into(),
            topic: "events".into(),
            format: "json".into(),
        },
    )
    .with_query("SELECT id, et FROM events")
    .with_watermark(Watermark {
        event_time_column: "et".into(),
        allowed_lateness_ms: 50,
        idle_timeout_ms: 0,
    });
    if let FlowKind::Streaming {
        output_mode,
        trigger,
        ..
    } = &mut flow.kind
    {
        *output_mode = OutputMode::Append;
        *trigger = Trigger::Continuous;
    }
    Pipeline::new("wm_stream")
        .with_dataset(Dataset::new("events_out", OutputType::Table).incremental())
        .with_flow(flow)
}

/// Incremental mode (with a `state_root`) drops the one late row and counts it;
/// the recompute default keeps every row — the paths differ by exactly the drop.
#[test]
fn incremental_watermark_drops_and_counts_late_rows() {
    // Recompute default: every row survives (watermark is advisory).
    let recompute = NativeBackend::new()
        .with_stream_input("events", wm_micro_batches())
        .run(&wm_pipeline())
        .expect("recompute runs");
    assert_eq!(
        recompute.row_count("events_out"),
        6,
        "recompute keeps all rows"
    );
    assert_eq!(
        sorted_ids(recompute.output("events_out").unwrap()),
        vec![1, 2, 3, 4, 5, 6]
    );

    // Incremental: the late id=5 (et=120, below the 250 watermark) is dropped.
    let wm_state = tempfile::tempdir().unwrap();
    let mut incremental = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", wm_state.path().display()))
        .with_stream_input("events", wm_micro_batches())
        .run(&wm_pipeline())
        .expect("incremental runs");
    assert_eq!(
        incremental.row_count("events_out"),
        5,
        "one late row dropped"
    );
    assert_eq!(
        sorted_ids(incremental.output("events_out").unwrap()),
        vec![1, 2, 3, 4, 6],
        "id=5 (the late row) is gone; every on-time row survives"
    );

    // The per-trigger increments reflect the drop: trigger 2 emits only id=6.
    let incs = incremental.increments("events_out");
    assert_eq!(incs.len(), 3, "one increment per trigger");
    assert_eq!(sorted_ids(&incs[0]), vec![1, 2]);
    assert_eq!(sorted_ids(&incs[1]), vec![3, 4]);
    assert_eq!(
        sorted_ids(&incs[2]),
        vec![6],
        "id=5 dropped as late in trigger 2"
    );

    // The drop is reported honestly on a RunEvent (count + watermark).
    let events = incremental.poll_events().unwrap();
    assert!(
        events
            .iter()
            .any(|e| e.message.contains("1 late row(s) dropped")),
        "the late-drop count is surfaced: {events:?}"
    );
}

/// With NO late rows, the incremental path equals the recompute path (and the
/// batch path) over the same data — the green correctness invariant.
#[test]
fn incremental_without_late_rows_equals_recompute() {
    // Event times increase but stay within the 50ms allowed lateness of the
    // running max, so no row ever falls below the watermark → nothing is late.
    let on_time = || {
        vec![
            vec![evt_batch(&[1, 2], &[100, 110])], // W -> 60
            vec![evt_batch(&[3, 4], &[150, 160])], // W -> 110; 150,160 on time
            vec![evt_batch(&[5, 6], &[200, 210])], // W -> 160; 200,210 on time
        ]
    };
    let recompute = NativeBackend::new()
        .with_stream_input("events", on_time())
        .run(&wm_pipeline())
        .expect("recompute runs");
    let wm_state = tempfile::tempdir().unwrap();
    let incremental = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", wm_state.path().display()))
        .with_stream_input("events", on_time())
        .run(&wm_pipeline())
        .expect("incremental runs");
    assert_eq!(
        sorted_ids(incremental.output("events_out").unwrap()),
        sorted_ids(recompute.output("events_out").unwrap()),
        "no late rows => incremental == recompute"
    );
    assert_eq!(incremental.row_count("events_out"), 6);
}

// ── Incremental mode: windowed close / emit-once / evict (phase 2) ──────────
//
// A windowed streaming flow in incremental mode assigns kept rows to event-time
// windows and, when the watermark `W ≥ window_end + allowed_lateness`, CLOSES
// each window, emits its aggregate exactly once, and EVICTS its buffered rows.
// Closed windows leave the working set — so the buffered (open) state stays
// bounded, the property the recompute path (ever-growing accumulation) lacks.

/// A windowed streaming flow over `events` with a per-window `COUNT(*)`, an
/// event-time watermark on `et`, and the given window spec / allowed lateness.
fn windowed_pipeline(window: WindowSpec, lateness_ms: u64) -> Pipeline {
    let mut flow = Flow::streaming(
        "f_events",
        "win_out",
        SourceSpec::Kafka {
            bootstrap: "localhost:9092".into(),
            topic: "events".into(),
            format: "json".into(),
        },
    )
    .with_query("SELECT COUNT(*) AS n FROM events")
    .with_watermark(Watermark {
        event_time_column: "et".into(),
        allowed_lateness_ms: lateness_ms,
        idle_timeout_ms: 0,
    })
    .with_window(window);
    if let FlowKind::Streaming {
        output_mode,
        trigger,
        ..
    } = &mut flow.kind
    {
        *output_mode = OutputMode::Append;
        *trigger = Trigger::Continuous;
    }
    Pipeline::new("win_stream")
        .with_dataset(Dataset::new("win_out", OutputType::Table).incremental())
        .with_flow(flow)
}

/// `(window_start, window_end, count)` triples from a windowed result, sorted by
/// window start. The emitted schema is `[window_start, window_end, n]`.
fn window_counts(batches: &[RecordBatch]) -> Vec<(i64, i64, i64)> {
    let mut out = Vec::new();
    for b in batches {
        let ws = b.column(0).as_any().downcast_ref::<Int64Array>().unwrap();
        let we = b.column(1).as_any().downcast_ref::<Int64Array>().unwrap();
        let n = b.column(2).as_any().downcast_ref::<Int64Array>().unwrap();
        for r in 0..b.num_rows() {
            out.push((ws.value(r), we.value(r), n.value(r)));
        }
    }
    out.sort_unstable();
    out
}

/// Run a windowed pipeline in incremental mode over `mb`. Each run gets a fresh
/// (empty) `state_root` so the barrier checkpoints it writes never leak into
/// another run's crash-recovery (a populated state_root auto-resumes — phase 4).
fn run_windowed(
    mb: Vec<Vec<RecordBatch>>,
    window: WindowSpec,
    lateness_ms: u64,
    _tag: &str,
) -> knut_thund::backend::native::NativeRun {
    let state = tempfile::tempdir().expect("temp state root");
    NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", state.path().display()))
        .with_stream_input("events", mb)
        .run(&windowed_pipeline(window, lateness_ms))
        .expect("windowed incremental runs")
}

/// A tumbling window closes exactly once at the watermark that passes
/// `end + lateness`, emits its count, and evicts its rows — so the buffered
/// open state stays bounded well below the total rows seen.
#[test]
fn windowed_tumbling_closes_once_and_evicts() {
    // Tumbling 100ms windows, allowed lateness 50ms. Window [s, s+100) closes
    // when W = max_et - 50 >= s + 100 + 50, i.e. max_et >= s + 200.
    let mb = vec![
        vec![evt_batch(&[1, 2], &[10, 20])],   // window [0,100)    W=-30
        vec![evt_batch(&[3, 4], &[110, 120])], // window [100,200)  W=70
        vec![evt_batch(&[5, 6], &[210, 220])], // W=170 -> [0,100) closes
        vec![evt_batch(&[7, 8], &[310, 320])], // W=270 -> [100,200) closes
    ];
    let mut run = run_windowed(mb, WindowSpec::Tumbling { size_ms: 100 }, 50, "tumble");

    // Four windows all close-and-emit exactly once (two on the watermark
    // mid-stream, two flushed at end-of-stream), each counting its 2 rows.
    assert_eq!(
        run.windows_closed("win_out"),
        4,
        "four windows close exactly once"
    );
    assert_eq!(
        window_counts(run.output("win_out").unwrap()),
        vec![(0, 100, 2), (100, 200, 2), (200, 300, 2), (300, 400, 2)],
        "each window aggregates exactly its own rows"
    );

    // Closes happen at the RIGHT watermark, surfaced honestly on RunEvents.
    let events = run.poll_events().unwrap();
    let msgs: Vec<&str> = events.iter().map(|e| e.message.as_str()).collect();
    assert!(
        msgs.iter()
            .any(|m| m.contains("window [0, 100) closed & evicted at watermark 170")),
        "window [0,100) closes when W first reaches 170: {msgs:?}"
    );
    assert!(
        msgs.iter()
            .any(|m| m.contains("window [100, 200) closed & evicted at watermark 270")),
        "window [100,200) closes when W first reaches 270: {msgs:?}"
    );

    // Bounded memory: 8 rows flowed through, but the incremental-agg operator
    // (COUNT(*) is a supported shape — phase 5) holds only per-open-window
    // accumulators, not the raw rows. At most two windows are open at once, so
    // the open working set peaks at 2 accumulators and shrinks back to 0 once
    // all windows close — far below the 8 rows seen.
    assert!(
        run.used_incremental_agg("win_out"),
        "COUNT(*) uses the incremental-agg operator"
    );
    assert_eq!(
        run.peak_open_rows("win_out"),
        2,
        "open accumulators bounded far below 8 rows"
    );
    let series = run.window_open_series("win_out");
    assert_eq!(
        *series.last().unwrap(),
        0,
        "all state evicted by end-of-stream"
    );
    assert!(
        run.peak_open_rows("win_out") < 8,
        "the evicting open state stays below the total rows seen: {series:?}"
    );
}

/// After a window closes, a late arrival for it is dropped by the watermark gate
/// and can NEVER re-open the window — its count stays final.
#[test]
fn windowed_late_after_close_is_dropped_not_reopened() {
    // Same as above, but trigger 3 also carries a late row (et=50) for the
    // already-closed [0,100) window. W=270 there, so et=50 < W is dropped.
    let mb = vec![
        vec![evt_batch(&[1, 2], &[10, 20])],
        vec![evt_batch(&[3, 4], &[110, 120])],
        vec![evt_batch(&[5, 6], &[210, 220])], // [0,100) closes at W=170
        vec![evt_batch(&[7, 8, 9], &[310, 320, 50])], // id=9 is late for [0,100)
    ];
    let mut run = run_windowed(mb, WindowSpec::Tumbling { size_ms: 100 }, 50, "late");

    // Still exactly four windows, and [0,100) still counts 2 — the late row did
    // not re-open it or inflate its aggregate.
    assert_eq!(
        run.windows_closed("win_out"),
        4,
        "no window re-opens for a late row"
    );
    let counts = window_counts(run.output("win_out").unwrap());
    assert_eq!(
        counts,
        vec![(0, 100, 2), (100, 200, 2), (200, 300, 2), (300, 400, 2)],
        "the closed window's count is final; late row absorbed nowhere"
    );
    assert_eq!(
        counts.iter().filter(|(s, _, _)| *s == 0).count(),
        1,
        "the closed window is emitted once, not re-emitted"
    );

    // The drop is reported honestly.
    let events = run.poll_events().unwrap();
    assert!(
        events
            .iter()
            .any(|e| e.message.contains("1 late row(s) dropped")),
        "the late arrival is counted as dropped: {:?}",
        events.iter().map(|e| &e.message).collect::<Vec<_>>()
    );
}

/// Hopping (overlapping) windows: each row participates in several slide
/// windows; every window still closes-and-emits exactly once.
#[test]
fn windowed_hopping_closes_each_overlapping_window_once() {
    // size 200ms, slide 100ms (lateness 0). Each et lands in two windows.
    let mb = vec![
        vec![evt_batch(&[1], &[100])],
        vec![evt_batch(&[2], &[200])],
        vec![evt_batch(&[3], &[300])],
        vec![evt_batch(&[4], &[400])],
    ];
    let run = run_windowed(
        mb,
        WindowSpec::Hopping {
            size_ms: 200,
            slide_ms: 100,
        },
        0,
        "hop",
    );

    assert_eq!(
        run.windows_closed("win_out"),
        5,
        "five overlapping windows close once each"
    );
    assert_eq!(
        window_counts(run.output("win_out").unwrap()),
        vec![
            (0, 200, 1),   // et 100
            (100, 300, 2), // et 100, 200
            (200, 400, 2), // et 200, 300
            (300, 500, 2), // et 300, 400
            (400, 600, 1), // et 400
        ],
        "overlapping windows each aggregate their covered rows once"
    );
}

/// Session windows: activity within `gap_ms` merges into one session; a session
/// closes a gap after its last event and emits once.
#[test]
fn windowed_session_closes_on_inactivity_gap() {
    // gap 50ms, lateness 0. Events 10,40 form one session; 100,130 another
    // (100 is 60ms after 40 > gap, so a new session starts).
    let mb = vec![
        vec![evt_batch(&[1], &[10])],
        vec![evt_batch(&[2], &[40])],
        vec![evt_batch(&[3], &[100])], // W=100 >= (40+50) -> session [10,40] closes
        vec![evt_batch(&[4], &[130])],
    ];
    let run = run_windowed(mb, WindowSpec::Session { gap_ms: 50 }, 0, "session");

    assert_eq!(
        run.windows_closed("win_out"),
        2,
        "two sessions close once each"
    );
    assert_eq!(
        window_counts(run.output("win_out").unwrap()),
        vec![(10, 90, 2), (100, 180, 2)],
        "each session aggregates its own activity span (end = last event + gap)"
    );
}

// ── Checkpoint state store + crash recovery (phases 3-4) ────────────────────
//
// The incremental path injects a BARRIER at every trigger boundary and commits
// an epoch (operator state → Parquet, then a JSON manifest = the atomic commit
// point) under `state_root`. On restart the newest COMMITTED epoch is restored —
// operator state + watermark + source offset — and the run RESUMES without
// re-emitting a window that closed pre-crash and without losing an on-time
// event. `with_max_triggers` is the restart boundary: run once capped (a
// simulated crash), run again uncapped over the same source + state_root.

/// The 4-trigger tumbling stream reused across the recovery tests. A single
/// uninterrupted run closes windows [0,100),[100,200),[200,300),[300,400).
fn recovery_micro_batches() -> Vec<Vec<RecordBatch>> {
    vec![
        vec![evt_batch(&[1, 2], &[10, 20])],   // window [0,100)   W=-30
        vec![evt_batch(&[3, 4], &[110, 120])], // window [100,200) W=70
        vec![evt_batch(&[5, 6], &[210, 220])], // W=170 -> [0,100) closes
        vec![evt_batch(&[7, 8], &[310, 320])], // W=270 -> [100,200) closes
    ]
}

/// Phase 3: a windowed incremental run commits one barrier/epoch checkpoint per
/// trigger under its `state_root` (durable operator state + manifest).
#[test]
fn checkpoint_commits_one_epoch_per_trigger() {
    let state = tempfile::tempdir().unwrap();
    let uri = format!("file://{}", state.path().display());
    let run = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri)
        .with_stream_input("events", recovery_micro_batches())
        .run(&windowed_pipeline(
            WindowSpec::Tumbling { size_ms: 100 },
            50,
        ))
        .expect("windowed incremental runs");
    assert_eq!(
        run.checkpoints_committed("win_out"),
        4,
        "one epoch committed per trigger"
    );
    assert_eq!(
        run.resumed_from_epoch("win_out"),
        None,
        "a cold run resumes from nothing"
    );
    assert_eq!(run.windows_closed("win_out"), 4);
}

/// Phase 4: crash mid-stream (cap at 2 triggers), then resume from the
/// checkpoint over the full stream. The resumed run restores state from the
/// committed epoch and closes EVERY window exactly once — the combined result
/// equals a single uninterrupted run, with no double-emit and no lost event.
#[test]
fn recovery_resumes_from_checkpoint_without_double_emit() {
    let state = tempfile::tempdir().unwrap();
    let uri = format!("file://{}", state.path().display());
    let win = || windowed_pipeline(WindowSpec::Tumbling { size_ms: 100 }, 50);

    // The oracle: one uninterrupted run.
    let oracle_state = tempfile::tempdir().unwrap();
    let oracle = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", oracle_state.path().display()))
        .with_stream_input("events", recovery_micro_batches())
        .run(&win())
        .expect("oracle runs");
    let oracle_counts = window_counts(oracle.output("win_out").unwrap());

    // Run 1: process only the first 2 triggers, then "crash". No window has
    // closed yet (W=70 < 150), so nothing is emitted — but state is checkpointed.
    let run1 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri.clone())
        .with_max_triggers(Some(2))
        .with_stream_input("events", recovery_micro_batches())
        .run(&win())
        .expect("capped run1");
    assert_eq!(
        run1.checkpoints_committed("win_out"),
        2,
        "2 epochs committed before the crash"
    );
    assert_eq!(run1.windows_closed("win_out"), 0, "no window closed yet");
    assert_eq!(
        run1.resumed_from_epoch("win_out"),
        None,
        "run1 is the cold start"
    );

    // Run 2: same state_root, full stream, uncapped → RECOVER + resume.
    let run2 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri.clone())
        .with_stream_input("events", recovery_micro_batches())
        .run(&win())
        .expect("resumed run2");
    assert_eq!(
        run2.resumed_from_epoch("win_out"),
        Some(1),
        "resumed from the committed epoch 1"
    );
    assert_eq!(
        run2.windows_closed("win_out"),
        4,
        "all four windows close once in the resumed run"
    );
    assert_eq!(
        window_counts(run2.output("win_out").unwrap()),
        oracle_counts,
        "resumed result equals a single uninterrupted run — no double-emit, no loss"
    );
}

/// `(window_start, window_end, count)` triples over EVERY per-trigger increment
/// of a windowed run, in emission order (not the materialised `output()`).
/// The increment stream is the durable output a resumed run appends to.
fn increment_counts(
    run: &knut_thund::backend::native::NativeRun,
    name: &str,
) -> Vec<(i64, i64, i64)> {
    let mut out = Vec::new();
    for trig in run.increments(name) {
        out.extend(window_counts(trig));
    }
    out
}

/// Phase 4, the sharper guarantee: a window that closes-and-emits BEFORE the
/// crash must NOT be re-emitted after recovery, and no on-time window may be
/// lost — so the emissions of run-1 and run-2 are DISJOINT and their union
/// equals a single uninterrupted run. The existing `..without_double_emit`
/// test is contrived so nothing closes pre-crash (W never reaches a close
/// point in its 2 triggers); this one caps the crash at 3 triggers so window
/// [0,100) has already closed + evicted in run-1, exercising the recovery
/// `emitted`-set replay-guard on a window that is gone from the operator state.
#[test]
fn recovery_incremental_agg_window_closed_before_crash_not_reemitted() {
    let win = || windowed_pipeline(WindowSpec::Tumbling { size_ms: 100 }, 50);

    // Oracle: one uninterrupted run over the full stream.
    let oracle_state = tempfile::tempdir().unwrap();
    let oracle = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", oracle_state.path().display()))
        .with_stream_input("events", recovery_micro_batches())
        .run(&win())
        .expect("oracle runs");
    assert!(
        oracle.used_incremental_agg("win_out"),
        "COUNT(*) drives the incremental-agg operator"
    );
    let oracle_counts = window_counts(oracle.output("win_out").unwrap());
    assert_eq!(
        oracle_counts,
        vec![(0, 100, 2), (100, 200, 2), (200, 300, 2), (300, 400, 2)],
        "the oracle closes all four windows once each"
    );

    // Run 1: cap at 3 triggers. Trigger 2 advances W=170 >= 100+50, so window
    // [0,100) CLOSES + EVICTS inside run-1 (emitted, then gone from state), and
    // its epoch-2 checkpoint records it in the `emitted_windows` set.
    let state = tempfile::tempdir().unwrap();
    let uri = format!("file://{}", state.path().display());
    let run1 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri.clone())
        .with_max_triggers(Some(3))
        .with_stream_input("events", recovery_micro_batches())
        .run(&win())
        .expect("capped run1");
    assert_eq!(
        run1.resumed_from_epoch("win_out"),
        None,
        "run1 is the cold start"
    );
    assert_eq!(
        run1.checkpoints_committed("win_out"),
        3,
        "3 epochs committed before the crash"
    );
    assert_eq!(
        run1.windows_closed("win_out"),
        1,
        "window [0,100) closed pre-crash"
    );
    let run1_emitted = window_counts(run1.output("win_out").unwrap());
    assert_eq!(
        run1_emitted,
        vec![(0, 100, 2)],
        "run1 emitted exactly the pre-crash window"
    );

    // Run 2: same state_root, full stream, uncapped -> RECOVER from epoch 2.
    let run2 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri)
        .with_stream_input("events", recovery_micro_batches())
        .run(&win())
        .expect("resumed run2");
    assert_eq!(
        run2.resumed_from_epoch("win_out"),
        Some(2),
        "resumed from committed epoch 2"
    );
    let run2_emitted = window_counts(run2.output("win_out").unwrap());

    // The pre-crash window is NOT re-emitted by run-2 (the replay-guard held).
    assert!(
        !run2_emitted.iter().any(|(s, _, _)| *s == 0),
        "the already-closed window [0,100) is never re-emitted after recovery: {run2_emitted:?}"
    );

    // Exactly-once across the crash: run-1 and run-2 emissions are DISJOINT
    // (no double-emit) and their union equals the uninterrupted oracle (no
    // loss). This is the true cross-crash contract; `output()` alone is
    // per-run, so the guarantee lives in the union of the two runs' emissions.
    for w in &run1_emitted {
        assert!(
            !run2_emitted.contains(w),
            "no window emitted by BOTH runs (double-emit): {w:?}"
        );
    }
    let mut union = run1_emitted.clone();
    union.extend(run2_emitted.clone());
    union.sort_unstable();
    assert_eq!(
        union, oracle_counts,
        "run1 ∪ run2 emissions == one uninterrupted run (no loss, no dup)"
    );

    // The same disjoint-union guarantee holds on the per-trigger INCREMENT
    // stream (what a durable sink appends), independent of the flush-only
    // `output()` rows: run-2's increments never carry the pre-crash window.
    let run2_incr = increment_counts(&run2, "win_out");
    assert!(
        !run2_incr.iter().any(|(s, _, _)| *s == 0),
        "run2's increment stream never re-appends the pre-crash window: {run2_incr:?}"
    );
}

/// Phase 4 (non-windowed): the watermark/late-drop path also checkpoints +
/// recovers. Crash after 2 triggers, resume; the final table equals a single
/// run over the same rows (late rows still dropped exactly once).
#[test]
fn recovery_resumes_non_windowed_watermark_path() {
    let state = tempfile::tempdir().unwrap();
    let uri = format!("file://{}", state.path().display());

    let oracle_state = tempfile::tempdir().unwrap();
    let oracle = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", oracle_state.path().display()))
        .with_stream_input("events", wm_micro_batches())
        .run(&wm_pipeline())
        .expect("oracle runs");
    assert_eq!(
        oracle.row_count("events_out"),
        5,
        "one late row dropped in the oracle"
    );

    // Crash after 2 triggers.
    let run1 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri.clone())
        .with_max_triggers(Some(2))
        .with_stream_input("events", wm_micro_batches())
        .run(&wm_pipeline())
        .expect("capped run1");
    assert_eq!(run1.checkpoints_committed("events_out"), 2);

    // Resume.
    let run2 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri)
        .with_stream_input("events", wm_micro_batches())
        .run(&wm_pipeline())
        .expect("resumed run2");
    assert_eq!(
        run2.resumed_from_epoch("events_out"),
        Some(1),
        "resumed from committed epoch 1"
    );
    assert_eq!(
        sorted_ids(run2.output("events_out").unwrap()),
        sorted_ids(oracle.output("events_out").unwrap()),
        "resumed non-windowed result equals a single run (late row dropped once)"
    );
}

// ── Incremental aggregation operator (phase 5) ──────────────────────────────
//
// StreamingMode::Incremental drives a supported windowed aggregate
// (COUNT/SUM/MIN/MAX over Int64, GROUP BY Int64/Utf8, Tumbling/Hopping) with
// MAINTAINED ACCUMULATORS — updated per arriving row, emitted on window close,
// no per-window SQL re-run and no buffering of raw rows. Proven equivalent to
// the recompute oracle (the pre-existing COUNT window tests assert byte-identical
// results through this operator) and, here, to an independent ground truth over
// a grouped SUM/MIN/MAX corpus. Unsupported shapes fall back to recompute.

/// An `(id, et, g, x)` batch — event time `et`, group key `g`, value `x`.
fn evt4(rows: &[(i64, i64, i64, i64)]) -> RecordBatch {
    let schema = Arc::new(Schema::new(vec![
        Field::new("id", DataType::Int64, false),
        Field::new("et", DataType::Int64, false),
        Field::new("g", DataType::Int64, false),
        Field::new("x", DataType::Int64, false),
    ]));
    RecordBatch::try_new(
        schema,
        vec![
            Arc::new(Int64Array::from(
                rows.iter().map(|r| r.0).collect::<Vec<_>>(),
            )),
            Arc::new(Int64Array::from(
                rows.iter().map(|r| r.1).collect::<Vec<_>>(),
            )),
            Arc::new(Int64Array::from(
                rows.iter().map(|r| r.2).collect::<Vec<_>>(),
            )),
            Arc::new(Int64Array::from(
                rows.iter().map(|r| r.3).collect::<Vec<_>>(),
            )),
        ],
    )
    .unwrap()
}

/// Ground-truth tumbling-window group aggregates over flat `(et, g, x)` rows:
/// `(window_start, window_end, g, count, sum, min, max)`, sorted.
fn ground_truth_tumbling(
    rows: &[(i64, i64, i64, i64)],
    size: i64,
) -> Vec<(i64, i64, i64, i64, i64, i64, i64)> {
    use std::collections::BTreeMap;
    let mut acc: BTreeMap<(i64, i64), (i64, i64, i64, i64)> = BTreeMap::new();
    for (_, et, g, x) in rows {
        let s = et.div_euclid(size) * size;
        let e = acc.entry((s, *g)).or_insert((0, 0, i64::MAX, i64::MIN));
        e.0 += 1;
        e.1 += x;
        e.2 = e.2.min(*x);
        e.3 = e.3.max(*x);
    }
    let mut out: Vec<_> = acc
        .into_iter()
        .map(|((s, g), (n, sum, mn, mx))| (s, s + size, g, n, sum, mn, mx))
        .collect();
    out.sort_unstable();
    out
}

/// A windowed aggregate pipeline over `events` running `query`, watermark on
/// `et` (lateness 0), tumbling 100ms, append output.
fn agg_pipeline(query: &str) -> Pipeline {
    let mut flow = Flow::streaming(
        "f_events",
        "win_out",
        SourceSpec::Kafka {
            bootstrap: "localhost:9092".into(),
            topic: "events".into(),
            format: "json".into(),
        },
    )
    .with_query(query)
    // Lateness 100ms (>= a window) keeps W below the current window's event
    // times, so no on-time row of the ascending corpus is dropped as late — the
    // ground truth then counts every row.
    .with_watermark(Watermark {
        event_time_column: "et".into(),
        allowed_lateness_ms: 100,
        idle_timeout_ms: 0,
    })
    .with_window(WindowSpec::Tumbling { size_ms: 100 });
    if let FlowKind::Streaming {
        output_mode,
        trigger,
        ..
    } = &mut flow.kind
    {
        *output_mode = OutputMode::Append;
        *trigger = Trigger::Continuous;
    }
    Pipeline::new("agg_stream")
        .with_dataset(Dataset::new("win_out", OutputType::Table).incremental())
        .with_flow(flow)
}

/// A non-decreasing-event-time grouped corpus (no late rows), spanning four
/// tumbling windows across three groups.
fn agg_corpus() -> Vec<(i64, i64, i64, i64)> {
    vec![
        (1, 10, 1, 10),
        (2, 20, 1, 20),
        (3, 30, 2, 5),
        (4, 110, 1, 7),
        (5, 120, 2, 8),
        (6, 210, 3, 100),
        (7, 220, 1, 4),
        (8, 310, 2, 50),
        (9, 330, 3, 60),
    ]
}

fn corpus_micro_batches() -> Vec<Vec<RecordBatch>> {
    let c = agg_corpus();
    vec![
        vec![evt4(&c[0..3])],
        vec![evt4(&c[3..5])],
        vec![evt4(&c[5..7])],
        vec![evt4(&c[7..9])],
    ]
}

/// Phase 5: a grouped `COUNT/SUM/MIN/MAX` windowed aggregate runs on maintained
/// accumulators (no per-window SQL) and matches the independent ground truth.
#[test]
fn incremental_agg_grouped_matches_ground_truth() {
    let state = tempfile::tempdir().unwrap();
    let query =
        "SELECT g, COUNT(*) AS n, SUM(x) AS s, MIN(x) AS mn, MAX(x) AS mx FROM events GROUP BY g";
    let run = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", state.path().display()))
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(query))
        .expect("incremental-agg runs");

    assert!(
        run.used_incremental_agg("win_out"),
        "supported shape uses the incremental-agg operator"
    );
    assert_eq!(
        run.windows_closed("win_out"),
        4,
        "four tumbling windows close once each"
    );

    // Emitted schema: [window_start, window_end, g, n, s, mn, mx].
    let out = run.output("win_out").expect("agg output");
    let mut actual: Vec<(i64, i64, i64, i64, i64, i64, i64)> = Vec::new();
    for b in out {
        let c = |i: usize| b.column(i).as_any().downcast_ref::<Int64Array>().unwrap();
        let (ws, we, g, n, s, mn, mx) = (c(0), c(1), c(2), c(3), c(4), c(5), c(6));
        for r in 0..b.num_rows() {
            actual.push((
                ws.value(r),
                we.value(r),
                g.value(r),
                n.value(r),
                s.value(r),
                mn.value(r),
                mx.value(r),
            ));
        }
    }
    actual.sort_unstable();
    assert_eq!(
        actual,
        ground_truth_tumbling(&agg_corpus(), 100),
        "incremental accumulators equal the ground-truth window aggregates"
    );
}

/// Phase 5: the incremental-agg operator survives a crash + recovery exactly
/// like the buffered path — a resumed run's accumulator state restores from the
/// checkpoint and the final aggregates equal a single uninterrupted run.
#[test]
fn incremental_agg_recovers_across_a_crash() {
    let query = "SELECT g, SUM(x) AS s FROM events GROUP BY g";
    let oracle_state = tempfile::tempdir().unwrap();
    let oracle = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", oracle_state.path().display()))
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(query))
        .expect("oracle runs");
    assert!(oracle.used_incremental_agg("win_out"));

    // One state dir, kept alive across both runs (crash then resume).
    let state = tempfile::tempdir().unwrap();
    let uri = format!("file://{}", state.path().display());
    let _run1 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri.clone())
        .with_max_triggers(Some(2))
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(query))
        .expect("capped run1");
    let run2 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri)
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(query))
        .expect("resumed run2");
    assert_eq!(
        run2.resumed_from_epoch("win_out"),
        Some(1),
        "agg operator resumed from the checkpoint"
    );

    let pairs = |run: &knut_thund::backend::native::NativeRun| {
        let mut v: Vec<(i64, i64, i64, i64)> = Vec::new();
        for b in run.output("win_out").unwrap() {
            let c = |i: usize| b.column(i).as_any().downcast_ref::<Int64Array>().unwrap();
            let (ws, g, s) = (c(0), c(2), c(3));
            for r in 0..b.num_rows() {
                v.push((ws.value(r), 0, g.value(r), s.value(r)));
            }
        }
        v.sort_unstable();
        v
    };
    assert_eq!(
        pairs(&run2),
        pairs(&oracle),
        "recovered agg result equals a single run"
    );
}

/// Ground-truth tumbling-window group AVG over flat `(et, g, x)` rows:
/// `(window_start, window_end, g, count, avg_f64, sum)`, sorted. Independent of
/// the operator's internal (sum, count) partials — this IS the definition.
fn ground_truth_avg(
    rows: &[(i64, i64, i64, i64)],
    size: i64,
) -> Vec<(i64, i64, i64, i64, f64, i64)> {
    use std::collections::BTreeMap;
    let mut acc: BTreeMap<(i64, i64), (i64, i64)> = BTreeMap::new(); // (sum, count)
    for (_, et, g, x) in rows {
        let s = et.div_euclid(size) * size;
        let e = acc.entry((s, *g)).or_insert((0, 0));
        e.0 += x;
        e.1 += 1;
    }
    let mut out: Vec<_> = acc
        .into_iter()
        .map(|((s, g), (sum, n))| (s, s + size, g, n, sum as f64 / n as f64, sum))
        .collect();
    out.sort_by(|l, r| l.partial_cmp(r).unwrap());
    out
}

/// Phase 5 (additive shape): a MIXED windowed aggregate `COUNT(*)/AVG/SUM` runs on
/// maintained accumulators — AVG kept EXACTLY as flat (sum, count) i64 partials and
/// divided to Float64 only on emit — and matches the independent ground truth. This
/// exercises multi-slot partial offsets (AVG occupies two) and a mixed Int64/Float64
/// emit schema.
#[test]
fn incremental_agg_avg_matches_ground_truth() {
    use datafusion::arrow::array::Float64Array;
    let state = tempfile::tempdir().unwrap();
    let query = "SELECT g, COUNT(*) AS n, AVG(x) AS a, SUM(x) AS s FROM events GROUP BY g";
    let run = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", state.path().display()))
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(query))
        .expect("incremental-agg with AVG runs");

    assert!(
        run.used_incremental_agg("win_out"),
        "AVG is now a supported incremental shape"
    );
    assert_eq!(
        run.windows_closed("win_out"),
        4,
        "four tumbling windows close once each"
    );

    // Emitted schema: [window_start, window_end, g, n, a(Float64), s].
    let out = run.output("win_out").expect("agg output");
    assert_eq!(
        out[0].schema().field(4).data_type(),
        &DataType::Float64,
        "the AVG column is emitted as Float64"
    );
    let mut actual: Vec<(i64, i64, i64, i64, f64, i64)> = Vec::new();
    for b in out {
        let c = |i: usize| b.column(i).as_any().downcast_ref::<Int64Array>().unwrap();
        let a = b.column(4).as_any().downcast_ref::<Float64Array>().unwrap();
        let (ws, we, g, n, s) = (c(0), c(1), c(2), c(3), c(5));
        for r in 0..b.num_rows() {
            actual.push((
                ws.value(r),
                we.value(r),
                g.value(r),
                n.value(r),
                a.value(r),
                s.value(r),
            ));
        }
    }
    actual.sort_by(|l, r| l.partial_cmp(r).unwrap());
    assert_eq!(
        actual,
        ground_truth_avg(&agg_corpus(), 100),
        "incremental AVG accumulators equal the ground-truth window averages"
    );
}

/// Phase 5 (additive shape): the AVG accumulator (two i64 partials per group)
/// checkpoints + recovers across a simulated crash exactly like the single-slot
/// aggregates — a resumed run's averages equal a single uninterrupted run.
#[test]
fn incremental_agg_avg_recovers_across_a_crash() {
    use datafusion::arrow::array::Float64Array;
    let query = "SELECT g, AVG(x) AS a FROM events GROUP BY g";
    let oracle_state = tempfile::tempdir().unwrap();
    let oracle = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", oracle_state.path().display()))
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(query))
        .expect("oracle runs");
    assert!(oracle.used_incremental_agg("win_out"));

    let state = tempfile::tempdir().unwrap();
    let uri = format!("file://{}", state.path().display());
    let _run1 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri.clone())
        .with_max_triggers(Some(2))
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(query))
        .expect("capped run1");
    let run2 = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(uri)
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(query))
        .expect("resumed run2");
    assert_eq!(
        run2.resumed_from_epoch("win_out"),
        Some(1),
        "AVG operator resumed from checkpoint"
    );

    let avgs = |run: &knut_thund::backend::native::NativeRun| {
        let mut v: Vec<(i64, i64, f64)> = Vec::new();
        for b in run.output("win_out").unwrap() {
            let ws = b.column(0).as_any().downcast_ref::<Int64Array>().unwrap();
            let g = b.column(2).as_any().downcast_ref::<Int64Array>().unwrap();
            let a = b.column(3).as_any().downcast_ref::<Float64Array>().unwrap();
            for r in 0..b.num_rows() {
                v.push((ws.value(r), g.value(r), a.value(r)));
            }
        }
        v.sort_by(|l, r| l.partial_cmp(r).unwrap());
        v
    };
    assert_eq!(
        avgs(&run2),
        avgs(&oracle),
        "recovered AVG state equals a single run"
    );
}

/// Phase 5 fallback: an unsupported shape (a COMPUTING projection over the
/// aggregate, and a WHERE filter) does NOT use the incremental-agg operator — it
/// degrades to the recompute-per-window path, which still closes every window
/// (nothing regresses, and no computed value is silently mis-emitted).
#[test]
fn incremental_agg_falls_back_for_unsupported_shapes() {
    // A COMPUTING projection (`SUM(x)+1`) over the aggregate → recompute. The
    // incremental operator emits the raw aggregate, so a computed projection MUST
    // fall back or it would silently return SUM(x), not SUM(x)+1.
    let s1 = tempfile::tempdir().unwrap();
    let computed = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", s1.path().display()))
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(
            "SELECT g, SUM(x) + 1 AS s1 FROM events GROUP BY g",
        ))
        .expect("computed projection runs via recompute-per-window");
    assert!(
        !computed.used_incremental_agg("win_out"),
        "computing projection falls back"
    );
    assert_eq!(
        computed.windows_closed("win_out"),
        4,
        "fallback path still closes every window"
    );

    // A WHERE filter → recompute-per-window.
    let s2 = tempfile::tempdir().unwrap();
    let filtered = NativeBackend::new()
        .with_streaming_mode(StreamingMode::Incremental)
        .with_state_root(format!("file://{}", s2.path().display()))
        .with_stream_input("events", corpus_micro_batches())
        .run(&agg_pipeline(
            "SELECT COUNT(*) AS n FROM events WHERE x > 6",
        ))
        .expect("filtered runs via recompute-per-window");
    assert!(
        !filtered.used_incremental_agg("win_out"),
        "WHERE falls back"
    );
}

// ── CDC apply-changes / SCD merge ───────────────────────────────────────────
//
// The native backend's CDC merge operator folds an upstream changelog — carrying
// skade & knut-bifrost's shared `_change_type` vocabulary (INSERT / UPDATE_AFTER
// / UPDATE_BEFORE / DELETE) — into the target, key-matched and ordered by the
// flow's `sequence_by`. These prove SCD type 1 (overwrite) and type 2 (history).

/// A changelog batch: parallel `(id, name, seq, _change_type)` slices.
fn change_batch(rows: &[(i64, &str, i64, &str)]) -> RecordBatch {
    let schema = Arc::new(Schema::new(vec![
        Field::new("id", DataType::Int64, false),
        Field::new("name", DataType::Utf8, false),
        Field::new("seq", DataType::Int64, false),
        Field::new("_change_type", DataType::Utf8, false),
    ]));
    let ids: Vec<i64> = rows.iter().map(|r| r.0).collect();
    let names: Vec<&str> = rows.iter().map(|r| r.1).collect();
    let seqs: Vec<i64> = rows.iter().map(|r| r.2).collect();
    let cts: Vec<&str> = rows.iter().map(|r| r.3).collect();
    RecordBatch::try_new(
        schema,
        vec![
            Arc::new(Int64Array::from(ids)),
            Arc::new(StringArray::from(names)),
            Arc::new(Int64Array::from(seqs)),
            Arc::new(StringArray::from(cts)),
        ],
    )
    .unwrap()
}

/// The insert/update/delete changelog seeded as three micro-batches:
///  * t0: INSERT id=1 "a", INSERT id=2 "b"
///  * t1: UPDATE_AFTER id=1 "a2", INSERT id=3 "c"
///  * t2: DELETE id=2
fn cdc_micro_batches() -> Vec<Vec<RecordBatch>> {
    vec![
        vec![change_batch(&[
            (1, "a", 1, "INSERT"),
            (2, "b", 2, "INSERT"),
        ])],
        vec![change_batch(&[
            (1, "a2", 3, "UPDATE_AFTER"),
            (3, "c", 4, "INSERT"),
        ])],
        vec![change_batch(&[(2, "b", 5, "DELETE")])],
    ]
}

/// A CDC apply-changes pipeline over the seeded `changes` changelog, keyed by
/// `id` and sequenced by `seq`, with the given SCD type.
fn cdc_pipeline(scd_type: ScdType) -> Pipeline {
    let flow = Flow {
        name: "f_cdc".into(),
        target: "dim".into(),
        reads: vec!["changes".into()],
        kind: FlowKind::Cdc {
            cdc: CdcSpec {
                keys: vec!["id".into()],
                sequence_by: "seq".into(),
                apply_as_deletes: None,
                scd_type,
            },
        },
        query: Some("SELECT * FROM changes".into()),
        projection: Vec::new(),
        filter: None,
        expectations: Vec::new(),
    };
    Pipeline::new("cdc")
        .with_dataset(Dataset::new("dim", OutputType::Table).incremental())
        .with_flow(flow)
}

/// Column `name` from a slice of batches as `i64` values (panics if absent/typed
/// otherwise).
fn i64_col(batches: &[RecordBatch], name: &str) -> Vec<i64> {
    batches
        .iter()
        .flat_map(|b| {
            let idx = b.schema().index_of(name).expect("column present");
            b.column(idx)
                .as_any()
                .downcast_ref::<Int64Array>()
                .expect("i64 column")
                .values()
                .to_vec()
        })
        .collect()
}

/// `(id, name)` pairs from the merged target, sorted for a stable assertion.
fn id_name_pairs(batches: &[RecordBatch]) -> Vec<(i64, String)> {
    let mut out = Vec::new();
    for b in batches {
        let ids = b
            .column(b.schema().index_of("id").unwrap())
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        let names = b
            .column(b.schema().index_of("name").unwrap())
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        for r in 0..b.num_rows() {
            out.push((ids.value(r), names.value(r).to_string()));
        }
    }
    out.sort();
    out
}

/// SCD type 1: the merged target is the latest non-deleted row per key. Over the
/// insert/update/delete changelog: id=1 updated to "a2", id=2 deleted, id=3 new.
#[test]
fn cdc_scd_type1_upserts_and_deletes() {
    let run = NativeBackend::new()
        .with_stream_input("changes", cdc_micro_batches())
        .run(&cdc_pipeline(ScdType::Type1))
        .expect("cdc run completes");

    let out = run.output("dim").expect("merged target produced");
    // id=1 overwritten to "a2", id=3 inserted, id=2 removed by the DELETE.
    assert_eq!(
        id_name_pairs(out),
        vec![(1, "a2".to_string()), (3, "c".to_string())],
        "latest non-deleted row per key"
    );

    // One increment per micro-batch, reflecting the merged snapshot as it grows.
    assert_eq!(
        run.trigger_count("dim"),
        3,
        "one merge per changelog micro-batch"
    );
    let incs = run.increments("dim");
    // t0: ids {1,2}; t1: id=1 updated + id=3 added -> {1,2,3}; t2: id=2 deleted -> {1,3}.
    let mut t0 = i64_col(&incs[0], "id");
    t0.sort_unstable();
    assert_eq!(t0, vec![1, 2]);
    let mut t2 = i64_col(&incs[2], "id");
    t2.sort_unstable();
    assert_eq!(t2, vec![1, 3], "the DELETE removed id=2 from the target");
}

/// SCD type 2: every version is kept with derived `__start_at` / `__end_at`
/// bounds. id=1 has two versions (a closed, a2 open); id=2's single version is
/// closed by its DELETE (no open version); id=3 has one open version.
#[test]
fn cdc_scd_type2_keeps_version_history() {
    let run = NativeBackend::new()
        .with_stream_input("changes", cdc_micro_batches())
        .run(&cdc_pipeline(ScdType::Type2))
        .expect("cdc run completes");

    let out = run.output("dim").expect("merged target produced");
    let out: Vec<RecordBatch> = out.to_vec();

    // 4 emitted versions: id=1 ×2, id=2 ×1 (closed), id=3 ×1.
    let total: usize = out.iter().map(|b| b.num_rows()).sum();
    assert_eq!(
        total, 4,
        "one row per non-delete change (history preserved)"
    );

    // Gather (id, name, start_at, end_at|None) tuples.
    let mut versions: Vec<(i64, String, i64, Option<i64>)> = Vec::new();
    for b in &out {
        let ids = b
            .column(b.schema().index_of("id").unwrap())
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        let names = b
            .column(b.schema().index_of("name").unwrap())
            .as_any()
            .downcast_ref::<StringArray>()
            .unwrap();
        let starts = b
            .column(b.schema().index_of("__start_at").unwrap())
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        let ends = b
            .column(b.schema().index_of("__end_at").unwrap())
            .as_any()
            .downcast_ref::<Int64Array>()
            .unwrap();
        for r in 0..b.num_rows() {
            let end = if ends.is_null(r) {
                None
            } else {
                Some(ends.value(r))
            };
            versions.push((
                ids.value(r),
                names.value(r).to_string(),
                starts.value(r),
                end,
            ));
        }
    }
    versions.sort();

    // id=1: "a" valid [1,3), then "a2" open [3,∞).
    assert!(
        versions.contains(&(1, "a".into(), 1, Some(3))),
        "id=1 first version closed at seq 3"
    );
    assert!(
        versions.contains(&(1, "a2".into(), 3, None)),
        "id=1 current version open"
    );
    // id=2: single version closed by the DELETE at seq 5, no open version.
    assert!(
        versions.contains(&(2, "b".into(), 2, Some(5))),
        "id=2 closed by its delete"
    );
    assert!(
        !versions.iter().any(|v| v.0 == 2 && v.3.is_none()),
        "id=2 has no open version after delete"
    );
    // id=3: current open version.
    assert!(
        versions.contains(&(3, "c".into(), 4, None)),
        "id=3 current version open"
    );
}

/// A CDC flow with no seeded changelog (and no registered source) surfaces the
/// planning failure honestly rather than faking a merge.
#[test]
fn cdc_flow_without_a_changelog_errors() {
    let err = NativeBackend::new()
        .run(&cdc_pipeline(ScdType::Type1))
        .unwrap_err();
    assert!(
        matches!(err, knut_thund::ThundError::Backend(_)),
        "unseeded CDC changelog fails as a backend error, got {err:?}"
    );
}