crates/knut-thund/tests/native_union_append.rs
//! Integration proof for the native backend's **multi-flow fan-in (union
//! append)** — SDP's canonical "one streaming table, N append flows" pattern.
//!
//! The IR always permitted more than one [`Flow`] to write the same target
//! dataset (unique flow names, shared targets; `validate()` + `topo_order()`
//! both accept it). Before this, the executor's `materialize` deregistered and
//! re-registered the target per flow, so the **last** flow to run silently
//! clobbered every earlier flow's rows. Now the executor unions them: each
//! append flow reads its own independently-authored source and their rows are
//! concatenated into one asset, which downstream flows and the file sink see in
//! full.
//!
//! Asserts, on the REAL DataFusion engine (no live infra):
//! * two batch flows into one target union (not clobber) — exact rows;
//! * three-way fan-in accumulates all three;
//! * a downstream flow reading the fan-in target sees every append flow's rows;
//! * a schema-mismatched append flow fails the run **loudly** (loud-gap law);
//! * single-flow-per-target is unchanged (no union path taken).
//!
//! Gated on `--features native` (the DataFusion tree); a no-op otherwise.
#![cfg(feature = "native")]
use knut_thund::backend::{ExecBackend, native::NativeBackend};
use knut_thund::ir::{Dataset, Flow, OutputType, Pipeline};
use datafusion::arrow::array::{Array, Int64Array, RecordBatch, StringArray};
use datafusion::arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;
/// An `(region, amount)` batch from parallel slices — a typed "sales feed".
fn feed(regions: &[&str], amounts: &[i64]) -> RecordBatch {
let schema = Arc::new(Schema::new(vec![
Field::new("region", DataType::Utf8, false),
Field::new("amount", DataType::Int64, false),
]));
RecordBatch::try_new(
schema,
vec![
Arc::new(StringArray::from(regions.to_vec())),
Arc::new(Int64Array::from(amounts.to_vec())),
],
)
.unwrap()
}
/// Collect a `(region, amount)` result as sorted pairs (union/GROUP-BY order is
/// unspecified) so assertions are deterministic.
fn sorted_pairs(batches: &[RecordBatch]) -> Vec<(String, i64)> {
let mut out: Vec<(String, i64)> = Vec::new();
for b in batches {
let region = b
.column_by_name("region")
.expect("region column")
.as_any()
.downcast_ref::<StringArray>()
.expect("region is Utf8");
let amount = b
.column_by_name("amount")
.expect("amount column")
.as_any()
.downcast_ref::<Int64Array>()
.expect("amount is Int64");
for i in 0..b.num_rows() {
out.push((region.value(i).to_string(), amount.value(i)));
}
}
out.sort();
out
}
/// Two batch flows, each reading its OWN in-memory source, both writing the one
/// `all_sales` table. Their rows must UNION — the fan-in the executor now does.
#[test]
fn two_append_flows_union_into_one_target() {
let out = NativeBackend::new()
.with_input("east", vec![feed(&["US", "US"], &[10, 20])])
.with_input("west", vec![feed(&["EU", "JP"], &[30, 40])])
.run(
&Pipeline::new("fan_in")
.with_dataset(Dataset::new("all_sales", OutputType::Table))
.with_flow(
Flow::batch("from_east", "all_sales", ["east"])
.with_query("SELECT region, amount FROM east"),
)
.with_flow(
Flow::batch("from_west", "all_sales", ["west"])
.with_query("SELECT region, amount FROM west"),
),
)
.expect("fan-in run completes");
let got = sorted_pairs(out.output("all_sales").expect("all_sales produced"));
let expected = vec![
("EU".to_string(), 30),
("JP".to_string(), 40),
("US".to_string(), 10),
("US".to_string(), 20),
];
// RED-when-broken: without the union splice, `all_sales` holds ONLY the last
// flow's rows (the two `west` rows), so this equality fails.
assert_eq!(
got, expected,
"both append flows' rows are present (union, not clobber)"
);
assert_eq!(out.row_count("all_sales"), 4, "4 unioned rows total");
knut_thund::functional_status(
"knut-thund/native_union_append",
"two_flows",
got == expected && out.row_count("all_sales") == 4,
"east(2) ∪ west(2) -> all_sales(4) via multi-flow fan-in",
);
}
/// Three append flows accumulate — the union is not limited to two.
#[test]
fn three_way_fan_in_accumulates_all() {
let out = NativeBackend::new()
.with_input("a", vec![feed(&["US"], &[1])])
.with_input("b", vec![feed(&["EU"], &[2])])
.with_input("c", vec![feed(&["JP"], &[3])])
.run(
&Pipeline::new("three_way")
.with_dataset(Dataset::new("merged", OutputType::Table))
.with_flow(Flow::batch("f_a", "merged", ["a"]).with_query("SELECT * FROM a"))
.with_flow(Flow::batch("f_b", "merged", ["b"]).with_query("SELECT * FROM b"))
.with_flow(Flow::batch("f_c", "merged", ["c"]).with_query("SELECT * FROM c")),
)
.expect("three-way fan-in completes");
let got = sorted_pairs(out.output("merged").expect("merged produced"));
let expected = vec![
("EU".to_string(), 2),
("JP".to_string(), 3),
("US".to_string(), 1),
];
assert_eq!(got, expected, "all three append flows contribute");
assert_eq!(out.row_count("merged"), 3);
knut_thund::functional_status(
"knut-thund/native_union_append",
"three_flows",
got == expected,
"a ∪ b ∪ c -> merged(3)",
);
}
/// A DOWNSTREAM flow reading the fan-in target sees every append flow's rows —
/// the union is real in the catalog, not just in the returned handle.
#[test]
fn downstream_flow_sees_full_union() {
let out = NativeBackend::new()
.with_input("east", vec![feed(&["US", "US"], &[10, 20])])
.with_input("west", vec![feed(&["EU"], &[30])])
.run(
&Pipeline::new("fan_in_then_rollup")
.with_dataset(Dataset::new("all_sales", OutputType::Table))
.with_dataset(Dataset::new("total", OutputType::MaterializedView))
.with_flow(
Flow::batch("from_east", "all_sales", ["east"])
.with_query("SELECT * FROM east"),
)
.with_flow(
Flow::batch("from_west", "all_sales", ["west"])
.with_query("SELECT * FROM west"),
)
.with_flow(
Flow::batch("rollup", "total", ["all_sales"])
.with_query("SELECT 'ALL' AS region, SUM(amount) AS amount FROM all_sales"),
),
)
.expect("fan-in + downstream rollup completes");
// SUM over the FULL union (10+20+30) — clobber would sum only west's 30.
let got = sorted_pairs(out.output("total").expect("total produced"));
assert_eq!(
got,
vec![("ALL".to_string(), 60)],
"downstream sees union: 10+20+30 = 60"
);
knut_thund::functional_status(
"knut-thund/native_union_append",
"downstream",
got == vec![("ALL".to_string(), 60)],
"rollup over the fan-in union sums all append flows (60)",
);
}
/// Loud-gap law: an append flow whose output schema differs from the accumulated
/// one fails the run rather than silently dropping rows or mixing schemas.
#[test]
fn schema_mismatched_append_flow_fails_loudly() {
let wide_schema = Arc::new(Schema::new(vec![
Field::new("region", DataType::Utf8, false),
Field::new("amount", DataType::Int64, false),
Field::new("extra", DataType::Int64, false),
]));
let wide = RecordBatch::try_new(
wide_schema,
vec![
Arc::new(StringArray::from(vec!["EU"])),
Arc::new(Int64Array::from(vec![9])),
Arc::new(Int64Array::from(vec![99])),
],
)
.unwrap();
let res = NativeBackend::new()
.with_input("east", vec![feed(&["US"], &[10])])
.with_input("west", vec![wide])
.run(
&Pipeline::new("mismatch")
.with_dataset(Dataset::new("all_sales", OutputType::Table))
.with_flow(
Flow::batch("from_east", "all_sales", ["east"])
.with_query("SELECT * FROM east"),
)
.with_flow(
Flow::batch("from_west", "all_sales", ["west"])
.with_query("SELECT * FROM west"),
),
);
assert!(
res.is_err(),
"schema-mismatched fan-in must fail, not silently drop/mix"
);
let msg = format!("{}", res.unwrap_err());
assert!(
msg.contains("union-append"),
"error names the union-append gap: {msg}"
);
knut_thund::functional_status(
"knut-thund/native_union_append",
"schema_mismatch_loud",
msg.contains("union-append"),
"an incompatible append flow fails the run loudly",
);
}
/// Additive guard: a single flow per target takes NO union path and is unchanged.
#[test]
fn single_flow_per_target_unchanged() {
let out = NativeBackend::new()
.with_input("east", vec![feed(&["US", "EU"], &[10, 20])])
.run(
&Pipeline::new("single")
.with_dataset(Dataset::new("all_sales", OutputType::Table))
.with_flow(
Flow::batch("only", "all_sales", ["east"]).with_query("SELECT * FROM east"),
),
)
.expect("single-flow run completes");
let got = sorted_pairs(out.output("all_sales").expect("produced"));
assert_eq!(got, vec![("EU".to_string(), 20), ("US".to_string(), 10)]);
assert_eq!(out.row_count("all_sales"), 2);
knut_thund::functional_status(
"knut-thund/native_union_append",
"single_flow_unchanged",
out.row_count("all_sales") == 2,
"one flow per target: no union path, byte-for-byte unchanged",
);
}