crates/knut-thund/tests/iceberg_spark_gap.rs
//! **The Iceberg↔Spark streaming GAP — proven honestly, end to end.**
//!
//! The central claim of Þund's two-backend design is that the *one* Arrow IR can
//! carry streaming knobs Spark Declarative Pipelines (SDP) cannot express, that
//! the Spark path **drops them honestly** (a machine-readable capability report,
//! not a silent degrade), and that the **native** DataFusion backend actually
//! RUNS what SDP drops. This test drives that whole loop over ONE IR value:
//!
//! 1. Spark `check` *accepts* a watermark+window streaming pipeline — the
//! silent-acceptance gap: SDP takes it and runs it on Spark's defaults,
//! losing the explicit event-time semantics.
//! 2. `SparkBackend::lowering_report` names *exactly* which knobs are dropped,
//! per flow (watermark, window, non-append output mode, CDC, expectations).
//! 3. The SAME IR runs on the native backend and the dropped knobs take
//! effect: the watermark drops a late row, the windows close and evict.
//!
//! RED-when-broken: neutering the report makes step 2 assert-fail; a native
//! backend that ignored the watermark/window makes step 3 assert-fail. Neither
//! path can fake the other.
//!
//! Gated on `--features native` (the DataFusion tree runs step 3); the report /
//! lowering half (steps 1-2) compiles on the default build too.
#![cfg(feature = "native")]
use knut_thund::backend::{native::NativeBackend, native::StreamingMode, spark::SparkBackend};
use knut_thund::ir::{
CdcSpec, Dataset, Flow, FlowKind, OutputMode, OutputType, Pipeline, ScdType, SourceSpec,
Trigger, Watermark, WindowSpec,
};
use knut_thund::{ExecBackend, RunHandle};
use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;
// ── shared corpus ───────────────────────────────────────────────────────────
/// An `(id, et)` event-time batch (epoch-ms in `et`).
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 that 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.
fn wm_micro_batches() -> Vec<Vec<RecordBatch>> {
vec![
vec![evt_batch(&[1, 2], &[100, 110])],
vec![evt_batch(&[3, 4], &[200, 210])],
vec![evt_batch(&[5, 6], &[120, 300])], // id=5 is late once W=250
]
}
/// The gap pipeline: a streaming flow with an event-time watermark AND a
/// tumbling window — the two knobs SDP-over-Connect cannot express — plus a
/// per-window COUNT. Append output, per-trigger cadence.
fn gap_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 COUNT(*) AS n FROM events")
.with_watermark(Watermark {
event_time_column: "et".into(),
allowed_lateness_ms: 50,
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("gap_stream")
.with_dataset(Dataset::new("events_out", OutputType::Table).incremental())
.with_flow(flow)
}
/// **Step 1+2: Spark ACCEPTS the streaming pipeline but the report names every
/// dropped knob.** This is the silent-acceptance gap made honest.
#[test]
fn spark_accepts_but_report_names_the_dropped_streaming_knobs() {
let p = gap_pipeline();
let spark = SparkBackend::new("sc://localhost:15002");
// Spark's SDP takes a watermark+window streaming table without complaint —
// `check` does NOT gate on event-time/window (they aren't hard-refusals,
// they silently run on Spark defaults). That silent acceptance is the gap.
spark
.check(&p)
.expect("Spark accepts the streaming pipeline (it just can't honour the knobs)");
// The graph SHAPE still lowers faithfully (1 dataset, 1 flow, the query).
let sdp = p.to_sdp();
assert_eq!(sdp.datasets.len(), 1);
assert_eq!(sdp.flows.len(), 1);
assert_eq!(
sdp.flows[0].query.as_deref(),
Some("SELECT COUNT(*) AS n FROM events")
);
// But the lowering report is NOT lossless — it names the watermark + window
// as dropped on the `f_events` flow.
let report = spark.lowering_report(&p);
assert!(!report.is_lossless(), "watermark+window lowering is lossy");
let drop = report.flow("f_events").expect("f_events has dropped knobs");
assert!(
drop.dropped.iter().any(|d| d.contains("watermark")),
"the watermark drop is reported: {:?}",
drop.dropped
);
assert!(
drop.dropped.iter().any(|d| d.contains("window")),
"the window drop is reported: {:?}",
drop.dropped
);
}
/// **Step 3: the native backend RUNS the dropped knobs.** The same IR, executed
/// on DataFusion in incremental mode, drops the late row (the watermark SDP
/// dropped) and closes/evicts the windows (the window SDP dropped) — so what the
/// Spark path reports as lost, the native path actually computes.
#[test]
fn native_runs_what_sdp_drops_watermark_and_window() {
let p = gap_pipeline();
let state = tempfile::tempdir().unwrap();
let mut run = NativeBackend::new()
.with_streaming_mode(StreamingMode::Incremental)
.with_state_root(format!("file://{}", state.path().display()))
.with_stream_input("events", wm_micro_batches())
.run(&p)
.expect("native runs the streaming pipeline");
// The window knob SDP dropped: windows close + evict on the event-time
// watermark. Two [0,100)/[100,200)/[200,300)/[300,400)-style windows close.
assert!(
run.windows_closed("events_out") >= 1,
"native closes event-time windows"
);
// The watermark knob SDP dropped: the late id=5 (et=120, below W=250) is
// dropped and surfaced honestly.
let events = run.poll_events().unwrap();
assert!(
events
.iter()
.any(|e| e.message.contains("late row(s) dropped")),
"native drops the late row the watermark bounds: {:?}",
events.iter().map(|e| &e.message).collect::<Vec<_>>()
);
}
// ── one IR, both backends (item C — the ExecBackend seam) ────────────────────
/// A plain `(k, v)` batch.
fn kv_batch(ks: &[i64], vs: &[i64]) -> RecordBatch {
let schema = Arc::new(Schema::new(vec![
Field::new("k", DataType::Int64, false),
Field::new("v", DataType::Int64, false),
]));
RecordBatch::try_new(
schema,
vec![
Arc::new(Int64Array::from(ks.to_vec())),
Arc::new(Int64Array::from(vs.to_vec())),
],
)
.unwrap()
}
/// **The ExecBackend seam: ONE IR, both backends.** A batch medallion pipeline
/// is authored once and handed to both backends. The native backend genuinely
/// *executes* it on DataFusion (real output rows asserted); the Spark backend
/// lowers the SAME IR to a valid SDP `DataflowGraph` and accepts it in `check`
/// (a live cluster run needs a reachable server — honestly out of scope here,
/// but the lowering+validation is the runnable half). Both consume the identical
/// `Pipeline` value — the whole point of the pluggable backend.
#[test]
fn both_backends_consume_the_same_batch_ir() {
let p = Pipeline::new("rollup")
.with_dataset(Dataset::new("total", OutputType::MaterializedView))
.with_flow(
Flow::batch("agg", "total", ["orders"]).with_query("SELECT SUM(v) AS s FROM orders"),
);
// Native: really run it. 10+20+30 = 60.
let native_run = NativeBackend::new()
.with_input("orders", vec![kv_batch(&[1, 2, 3], &[10, 20, 30])])
.run(&p)
.expect("native executes the batch IR");
let out = native_run.output("total").expect("native produced `total`");
let s = out[0]
.column(0)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap()
.value(0);
assert_eq!(s, 60, "native computed SUM(v)");
// Spark: the SAME IR lowers to a valid SDP graph and passes `check`.
let spark = SparkBackend::new("sc://localhost:15002");
spark.check(&p).expect("spark accepts the same batch IR");
let sdp = p.to_sdp();
sdp.validate().expect("lowered SDP graph is valid");
assert_eq!(sdp.datasets.len(), 1);
assert_eq!(
sdp.flows[0].query.as_deref(),
Some("SELECT SUM(v) AS s FROM orders")
);
// And this batch IR is perfectly lossless onto SDP (no dropped knobs) —
// the gap is a *streaming* phenomenon, not a batch one.
assert!(
spark.lowering_report(&p).is_lossless(),
"a batch pipeline lowers losslessly"
);
}
// ── CDC dimension: 4.1 drops CDC entirely; 4.2 keeps Type1 but drops Type2 ────
/// A changelog batch `(id, name, seq, _change_type)`.
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),
]));
RecordBatch::try_new(
schema,
vec![
Arc::new(Int64Array::from(
rows.iter().map(|r| r.0).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.1).collect::<Vec<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.2).collect::<Vec<_>>(),
)),
Arc::new(StringArray::from(
rows.iter().map(|r| r.3).collect::<Vec<_>>(),
)),
],
)
.unwrap()
}
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)
}
/// A 4.1 Spark target cannot lower CDC at all — `check` refuses it AND the
/// report records the whole flow as dropped. A 4.2 target accepts Type1 but the
/// report still drops the Type2 history semantics. The native backend runs the
/// full SCD Type2 merge either way.
#[test]
fn cdc_gap_spark_41_drops_all_42_drops_type2_history() {
let p2 = cdc_pipeline(ScdType::Type2);
// 4.1: `check` hard-refuses CDC, and the report says the flow is dropped.
let spark41 = SparkBackend::new("sc://localhost:15002");
assert!(spark41.check(&p2).is_err(), "4.1 refuses CDC in check");
let r41 = spark41.lowering_report(&p2);
assert!(
r41.flow("f_cdc")
.unwrap()
.dropped
.iter()
.any(|d| d.contains("CDC apply-changes")),
"4.1 reports the whole CDC flow dropped: {:?}",
r41.all_dropped()
);
// 4.2: `check` accepts CDC, but the report still drops the Type2 history.
let spark42 = SparkBackend::new("sc://localhost:15002").spark_4_2();
spark42.check(&p2).expect("4.2 accepts a CDC flow");
let r42 = spark42.lowering_report(&p2);
assert!(
r42.flow("f_cdc")
.unwrap()
.dropped
.iter()
.any(|d| d.contains("SCD Type2")),
"4.2 reports SCD Type2 history dropped: {:?}",
r42.all_dropped()
);
// A Type1 CDC flow lowers losslessly onto 4.2.
assert!(
spark42
.lowering_report(&cdc_pipeline(ScdType::Type1))
.is_lossless(),
"Type1 CDC is lossless on a 4.2 target"
);
// The native backend RUNS the full SCD Type2 merge SDP could not.
let run = NativeBackend::new()
.with_stream_input(
"changes",
vec![
vec![change_batch(&[(1, "a", 1, "INSERT")])],
vec![change_batch(&[(1, "a2", 3, "UPDATE_AFTER")])],
],
)
.run(&p2)
.expect("native runs the SCD2 CDC merge");
let out = run.output("dim").expect("merged dim produced");
let total: usize = out.iter().map(|b| b.num_rows()).sum();
assert_eq!(
total, 2,
"native keeps both versions (SCD2 history) — the knob SDP dropped"
);
}