crates/knut-thund/tests/sdp_drop_coverage.rs
//! **The full SDP drop list — every knob Native actually runs.**
//!
//! [`SparkBackend::lowering_report`] enumerates precisely what the SDP-over-Connect
//! lowering DROPS: watermark, window, a `ProcessingTime` trigger, a non-append
//! output mode (`Update`/`Complete`), CDC apply-changes (4.1) / SCD-Type2 history
//! (4.2), and data-quality expectations. `iceberg_spark_gap.rs` proved the first
//! wave end to end (watermark late-row eviction, window close, SCD2 history). THIS
//! file closes the rest of the list so every reported drop has a matching, real,
//! row-level proof that the **native** DataFusion backend executes it:
//!
//! * `Update` output mode — emits only the CHANGED aggregate rows per trigger
//! (an upsert delta), distinct from `Complete`'s full-table emission.
//! * `ProcessingTime` trigger — the native backend runs it (one trigger per
//! arriving micro-batch; the wall-clock `interval_ms` is advisory, honestly,
//! since the in-process synthetic source has no real-time clock).
//! * CDC `apply_as_deletes` predicate — a delete identified by a user predicate,
//! not the `DELETE` change type, removes the row.
//! * CDC `UPDATE_BEFORE` pre-image — the pre-image of an in-place update carries
//! no net change and is dropped (not counted as a version or a latest row).
//! * `Warn` expectation — a violating row is KEPT and the violation recorded
//! (the third `ON VIOLATION` policy alongside `Drop`/`Fail`).
//!
//! Every test asserts BOTH halves of the honest gap: (a) the Spark lowering report
//! names the drop, and (b) the native backend runs it with a row-level assertion.
//! RED-when-broken: neutering the report fails half (a); a native backend that
//! ignored the knob fails half (b). Neither can fake the other.
//!
//! Gated on `--features native` (the DataFusion tree runs the native half); the CI
//! `--features native` lane picks this file up automatically (see the crate
//! `Cargo.toml` note on `native`-gated test files).
#![cfg(feature = "native")]
use knut_thund::backend::{native::NativeBackend, spark::SparkBackend};
use knut_thund::ir::{
CdcSpec, Dataset, Expectation, Flow, FlowKind, OnViolation, OutputMode, OutputType, Pipeline,
ScdType, SourceSpec, Trigger,
};
use knut_thund::{ExecBackend, RunHandle};
use datafusion::arrow::array::{Int64Array, RecordBatch, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;
// ── corpus builders ─────────────────────────────────────────────────────────
/// A single-column `(k)` key batch — the input to a `GROUP BY k` running count.
fn k_batch(ks: &[i64]) -> RecordBatch {
let schema = Arc::new(Schema::new(vec![Field::new("k", DataType::Int64, false)]));
RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(ks.to_vec()))]).unwrap()
}
/// 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()
}
/// A changelog batch `(id, name, seq, _change_type, gone)` — the extra `gone`
/// flag lets a user predicate (`gone = 1`) mark a delete without the `DELETE`
/// change type, exercising `CdcSpec::apply_as_deletes`.
fn change_batch_gone(rows: &[(i64, &str, i64, &str, i64)]) -> 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),
Field::new("gone", DataType::Int64, 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<_>>(),
)),
Arc::new(Int64Array::from(
rows.iter().map(|r| r.4).collect::<Vec<_>>(),
)),
],
)
.unwrap()
}
/// A streaming `GROUP BY k` running-count pipeline over a Kafka source named
/// `events`, parameterised by output mode + trigger.
fn agg_stream_pipeline(output_mode: OutputMode, trigger: Trigger) -> Pipeline {
let mut flow = Flow::streaming(
"f",
"out",
SourceSpec::Kafka {
bootstrap: "localhost:9092".into(),
topic: "events".into(),
format: "json".into(),
},
)
.with_query("SELECT k, COUNT(*) AS n FROM events GROUP BY k");
if let FlowKind::Streaming {
output_mode: om,
trigger: tg,
..
} = &mut flow.kind
{
*om = output_mode;
*tg = trigger;
}
Pipeline::new("agg_stream")
.with_dataset(Dataset::new("out", OutputType::Table).incremental())
.with_flow(flow)
}
/// The pair `(k, n)` for every row of an `out` snapshot, sorted.
fn kn_pairs(batches: &[RecordBatch]) -> Vec<(i64, i64)> {
let mut out = Vec::new();
for b in batches {
let ks = b
.column(b.schema().index_of("k").unwrap())
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
let ns = b
.column(b.schema().index_of("n").unwrap())
.as_any()
.downcast_ref::<Int64Array>()
.unwrap();
for r in 0..b.num_rows() {
out.push((ks.value(r), ns.value(r)));
}
}
out.sort();
out
}
fn rows(batches: &[RecordBatch]) -> usize {
batches.iter().map(|b| b.num_rows()).sum()
}
// ── Update output mode: changed rows only (SDP drops it; native runs it) ──────
/// **`OutputMode::Update` — SDP drops it, Native runs it as an upsert delta.**
///
/// SDP streaming tables are append-only, so the report drops `Update`. The native
/// backend runs it: each trigger emits exactly the aggregate rows whose value
/// CHANGED (or are new) — the running `GROUP BY` upsert delta. Contrasted below
/// with `Complete` (whole table each trigger) so the two modes provably differ,
/// not alias. HONEST: this is an upsert delta (changed/new rows re-emitted), not a
/// retraction stream — no negative rows are produced, since the native run/event
/// surface carries no retract marker.
#[test]
fn update_mode_emits_only_changed_rows_and_sdp_drops_it() {
let p = agg_stream_pipeline(OutputMode::Update, Trigger::Continuous);
// (a) The Spark report drops the Update output mode.
let report = SparkBackend::new("sc://localhost:15002").lowering_report(&p);
let drop = report.flow("f").expect("f has dropped knobs");
assert!(
drop.dropped
.iter()
.any(|d| d.contains("output_mode") && d.contains("Update")),
"report drops the Update output mode: {:?}",
drop.dropped
);
// (b) Native runs it. Two micro-batches: k=1 changes 1→2 on trigger 1; k=2 is
// unchanged. t0 -> {(1,1),(2,1)}; t1 -> {(1,2),(2,1)}.
let run = NativeBackend::new()
.with_stream_input("events", vec![vec![k_batch(&[1, 2])], vec![k_batch(&[1])]])
.run(&p)
.expect("native runs the Update streaming aggregate");
// Final materialised table is the full running result.
assert_eq!(
kn_pairs(run.output("out").unwrap()),
vec![(1, 2), (2, 1)],
"final table is the running count"
);
// The trigger-1 increment carries ONLY the changed group (1, 2) — the upsert
// delta, not the whole table.
let incs = run.increments("out");
assert_eq!(incs.len(), 2, "one increment per micro-batch trigger");
assert_eq!(
kn_pairs(&incs[1]),
vec![(1, 2)],
"Update emits only the changed group on trigger 1"
);
}
/// **`OutputMode::Complete` — SDP drops it, Native emits the whole table each
/// trigger.** The behavioural foil to `Update`: over the SAME two micro-batches,
/// Complete's trigger-1 increment is the FULL table (both groups), where Update's
/// was the single changed group. That the two increments differ is the proof both
/// modes genuinely execute distinct semantics on the native engine.
#[test]
fn complete_mode_emits_full_table_each_trigger_distinct_from_update() {
let p = agg_stream_pipeline(OutputMode::Complete, Trigger::Continuous);
let report = SparkBackend::new("sc://localhost:15002").lowering_report(&p);
assert!(
report
.flow("f")
.unwrap()
.dropped
.iter()
.any(|d| d.contains("output_mode") && d.contains("Complete")),
"report drops the Complete output mode"
);
let run = NativeBackend::new()
.with_stream_input("events", vec![vec![k_batch(&[1, 2])], vec![k_batch(&[1])]])
.run(&p)
.expect("native runs the Complete streaming aggregate");
let incs = run.increments("out");
// Complete re-emits the whole table each trigger: trigger 1 = both groups.
assert_eq!(
kn_pairs(&incs[1]),
vec![(1, 2), (2, 1)],
"Complete emits the full table on trigger 1"
);
// And it is strictly more than Update's single changed row — the modes differ.
assert!(
rows(&incs[1]) > 1,
"Complete's increment is the whole table, not just the delta"
);
}
// ── ProcessingTime trigger (SDP drops it; native runs it per micro-batch) ─────
/// **`Trigger::ProcessingTime` — SDP drops it, Native runs it.** SDP streaming
/// tables take no explicit processing-time trigger over Connect, so the report
/// drops it. The native backend runs the flow, firing one trigger per arriving
/// micro-batch. HONEST: the wall-clock `interval_ms` is advisory here — the
/// in-process synthetic source has no real-time clock, so cadence follows the
/// micro-batch arrivals, not the interval. The RESULT is still the correct running
/// aggregate.
#[test]
fn processing_time_trigger_runs_native_and_sdp_drops_it() {
let p = agg_stream_pipeline(
OutputMode::Update,
Trigger::ProcessingTime { interval_ms: 5_000 },
);
let report = SparkBackend::new("sc://localhost:15002").lowering_report(&p);
assert!(
report
.flow("f")
.unwrap()
.dropped
.iter()
.any(|d| d.contains("ProcessingTime")),
"report drops the ProcessingTime trigger: {:?}",
report.flow("f").unwrap().dropped
);
let run = NativeBackend::new()
.with_stream_input("events", vec![vec![k_batch(&[1, 2])], vec![k_batch(&[1])]])
.run(&p)
.expect("native runs the ProcessingTime-triggered stream");
// One trigger per arriving micro-batch (the honest cadence), correct result.
assert_eq!(
run.trigger_count("out"),
2,
"one trigger fired per micro-batch"
);
assert_eq!(
kn_pairs(run.output("out").unwrap()),
vec![(1, 2), (2, 1)],
"running aggregate is correct"
);
}
// ── CDC apply_as_deletes predicate (a delete not flagged by the change type) ──
fn cdc_pipeline_with(cdc: CdcSpec) -> Pipeline {
let flow = Flow {
name: "f_cdc".into(),
target: "dim".into(),
reads: vec!["changes".into()],
kind: FlowKind::Cdc { cdc },
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)
}
fn ids(batches: &[RecordBatch]) -> Vec<i64> {
let mut out: Vec<i64> = batches
.iter()
.flat_map(|b| {
let idx = b.schema().index_of("id").unwrap();
b.column(idx)
.as_any()
.downcast_ref::<Int64Array>()
.unwrap()
.values()
.to_vec()
})
.collect();
out.sort_unstable();
out
}
/// **CDC `apply_as_deletes` predicate — Native removes a row a user predicate,
/// not the change type, marks as deleted.** Every changelog row here is an
/// `INSERT`; the delete is identified by `gone = 1`. The native merge honours the
/// predicate and drops id=2, keeping id=1. (The 4.1 Spark report drops the whole
/// CDC flow — the other half of the gap.)
#[test]
fn cdc_apply_as_deletes_predicate_removes_row_native() {
let cdc = CdcSpec {
keys: vec!["id".into()],
sequence_by: "seq".into(),
apply_as_deletes: Some("gone = 1".into()),
scd_type: ScdType::Type1,
};
let p = cdc_pipeline_with(cdc);
// 4.1 Spark cannot lower a CDC flow at all — the report says so.
let report = SparkBackend::new("sc://localhost:15002").lowering_report(&p);
assert!(
report
.flow("f_cdc")
.unwrap()
.dropped
.iter()
.any(|d| d.contains("CDC apply-changes")),
"4.1 report drops the CDC flow"
);
// Native: id=2's latest row carries gone=1 → deleted by the predicate.
let run = NativeBackend::new()
.with_stream_input(
"changes",
vec![vec![change_batch_gone(&[
(1, "a", 1, "INSERT", 0),
(2, "b", 2, "INSERT", 0),
(2, "b", 3, "INSERT", 1),
])]],
)
.run(&p)
.expect("native runs the predicate-delete CDC merge");
assert_eq!(
ids(run.output("dim").unwrap()),
vec![1],
"predicate `gone = 1` deleted id=2"
);
}
/// **CDC `UPDATE_BEFORE` pre-image is dropped.** The pre-image row of an in-place
/// update carries no net change; the native merge filters `_change_type =
/// 'UPDATE_BEFORE'` before ranking, so id=1's latest value is the `UPDATE_AFTER`
/// ("a2"), never the stale pre-image ("a"). One row, name "a2".
#[test]
fn cdc_update_before_preimage_is_dropped_native() {
let cdc = CdcSpec {
keys: vec!["id".into()],
sequence_by: "seq".into(),
apply_as_deletes: None,
scd_type: ScdType::Type1,
};
let p = cdc_pipeline_with(cdc);
let run = NativeBackend::new()
.with_stream_input(
"changes",
vec![vec![change_batch(&[
(1, "a", 1, "INSERT"),
(1, "a", 2, "UPDATE_BEFORE"), // pre-image — must be dropped
(1, "a2", 3, "UPDATE_AFTER"),
])]],
)
.run(&p)
.expect("native runs the update merge");
let out = run.output("dim").unwrap();
assert_eq!(
rows(out),
1,
"one surviving row for id=1 (pre-image did not become a version)"
);
let names: Vec<String> = out
.iter()
.flat_map(|b| {
let idx = b.schema().index_of("name").unwrap();
let a = b
.column(idx)
.as_any()
.downcast_ref::<StringArray>()
.unwrap();
(0..b.num_rows())
.map(|r| a.value(r).to_string())
.collect::<Vec<_>>()
})
.collect();
assert_eq!(
names,
vec!["a2".to_string()],
"latest value wins; the UPDATE_BEFORE pre-image is gone"
);
}
// ── Warn expectation (the third ON VIOLATION policy; SDP drops expectations) ──
/// **`OnViolation::Warn` — Native keeps the violating row and records the
/// violation.** SDP drops expectations entirely (not upstreamed in OSS 4.1), so
/// the report names the drop. On the native backend the `Warn` policy (unlike
/// `Drop`/`Fail`, both already proven) KEEPS the offending row and emits a WARN
/// event — the data-quality gate observes without gating.
#[test]
fn warn_expectation_keeps_rows_and_records_native() {
let batch = {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("amount", DataType::Int64, false),
]));
RecordBatch::try_new(
schema,
vec![
Arc::new(Int64Array::from(vec![1, 2, 3])),
Arc::new(Int64Array::from(vec![10, -5, 20])), // id=2 violates amount > 0
],
)
.unwrap()
};
let p = Pipeline::new("warn")
.with_dataset(Dataset::new("checked", OutputType::Table))
.with_flow(
Flow::batch("f", "checked", ["orders"])
.with_query("SELECT * FROM orders")
.expect(Expectation::new("amount_positive", "amount > 0").on(OnViolation::Warn)),
);
// SDP drops expectations — the report says so.
let report = SparkBackend::new("sc://localhost:15002").lowering_report(&p);
assert!(
report
.flow("f")
.unwrap()
.dropped
.iter()
.any(|d| d.contains("expectation")),
"report drops the expectation"
);
// Native: all three rows are KEPT (warn, not drop), and a WARN event fires.
let mut run = NativeBackend::new()
.with_input("orders", vec![batch])
.run(&p)
.expect("native runs the warn-expectation batch flow");
assert_eq!(
run.row_count("checked"),
3,
"Warn keeps the violating row (all 3 rows survive)"
);
let events = run.poll_events().unwrap();
assert!(
events
.iter()
.any(|e| e.message.contains("WARN") && e.message.contains("amount_positive")),
"a WARN violation event is recorded: {:?}",
events.iter().map(|e| &e.message).collect::<Vec<_>>()
);
}
// ── the whole drop list is reported for one maximal pipeline ──────────────────
/// A single streaming flow that trips EVERY streaming drop at once — watermark,
/// window, ProcessingTime trigger, non-append (Update) mode, and an expectation —
/// so the report enumerates the full list on one flow. The row-level native
/// execution of each is proven by the focused tests above and in
/// `iceberg_spark_gap.rs`; this asserts the report is COMPLETE (nothing silently
/// un-flagged).
#[test]
fn report_enumerates_the_full_streaming_drop_list() {
use knut_thund::ir::{Watermark, WindowSpec};
let mut flow = Flow::streaming(
"f",
"out",
SourceSpec::Kafka {
bootstrap: "b".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 })
.expect(Expectation::new("n_pos", "n >= 0"));
if let FlowKind::Streaming {
output_mode,
trigger,
..
} = &mut flow.kind
{
*output_mode = OutputMode::Update;
*trigger = Trigger::ProcessingTime { interval_ms: 1000 };
}
let p = Pipeline::new("maximal")
.with_dataset(Dataset::new("out", OutputType::Table).incremental())
.with_flow(flow);
let dropped = SparkBackend::new("sc://localhost:15002")
.lowering_report(&p)
.all_dropped();
for needle in [
"watermark",
"window",
"ProcessingTime",
"output_mode",
"expectation",
] {
assert!(
dropped.iter().any(|d| d.contains(needle)),
"the report names `{needle}` as dropped; got {dropped:?}"
);
}
}