crates/knut-thund/tests/streaming_microbatch.rs
//! 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:?}"
);
}