crates/knut-thund/tests/native_file_sink.rs
//! Integration proof for the native backend's **on-disk file sink** — the
//! write-side sibling of `with_file_input`. It closes the laptop round trip:
//! read a local `orders.csv`, run a batch `GROUP BY` on the REAL DataFusion
//! engine, and **land the result as a single `parquet`/`csv` file on disk** via
//! [`NativeBackend::with_file_output`]. No external system, no catalog.
//!
//! Before this, the native backend could read a file but only ever returned its
//! result *in memory* (`NativeRun::output`); a consumer had no way to keep the
//! result as a file they could hand to the next tool. Here the produced dataset
//! is written to disk and then **re-ingested** through `with_file_input` — a
//! full read → SQL → write → read-back loop that asserts the exact rows.
//!
//! 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, WriteMode};
use datafusion::arrow::array::{Array, Int64Array, RecordBatch, StringArray, StringViewArray};
use std::io::Write;
/// A rollup pipeline reading the registered `orders` table.
fn rollup_pipeline() -> Pipeline {
Pipeline::new("orders_rollup")
.with_dataset(Dataset::new("by_customer", OutputType::Sink))
.with_flow(
Flow::batch("agg", "by_customer", ["orders"])
.with_query("SELECT customer, SUM(amount) AS total FROM orders GROUP BY customer"),
)
}
/// A trivial pass-through pipeline: `SELECT *` over the registered `t` table.
/// Used to re-ingest a written sink file and read its rows back.
fn passthrough_pipeline() -> Pipeline {
Pipeline::new("readback")
.with_dataset(Dataset::new("t_out", OutputType::MaterializedView))
.with_flow(Flow::batch("scan", "t_out", ["t"]).with_query("SELECT * FROM t"))
}
/// Collect a `(customer, total)` result sorted by customer (GROUP BY order is
/// unspecified). Accepts `Utf8` (CSV) or `Utf8View` (parquet) string columns.
fn sorted_totals(batches: &[RecordBatch]) -> Vec<(String, i64)> {
let mut out: Vec<(String, i64)> = Vec::new();
for b in batches {
let col = b.column(0).as_any();
let cust: Vec<String> = if let Some(a) = col.downcast_ref::<StringArray>() {
(0..a.len()).map(|i| a.value(i).to_string()).collect()
} else if let Some(a) = col.downcast_ref::<StringViewArray>() {
(0..a.len()).map(|i| a.value(i).to_string()).collect()
} else {
panic!("customer column is neither Utf8 nor Utf8View");
};
let total = b
.column(1)
.as_any()
.downcast_ref::<Int64Array>()
.expect("total is Int64");
for (i, c) in cust.into_iter().enumerate() {
out.push((c, total.value(i)));
}
}
out.sort();
out
}
/// Write the shared `orders.csv` fixture (two customers, five rows).
fn write_orders_csv(path: &std::path::Path) {
let mut f = std::fs::File::create(path).expect("create orders.csv");
writeln!(f, "customer,amount").unwrap();
writeln!(f, "alice,10").unwrap();
writeln!(f, "bob,5").unwrap();
writeln!(f, "alice,20").unwrap();
writeln!(f, "bob,7").unwrap();
writeln!(f, "alice,3").unwrap();
}
const EXPECTED: [(&str, i64); 2] = [("alice", 33), ("bob", 12)];
fn expected_vec() -> Vec<(String, i64)> {
EXPECTED.iter().map(|(c, t)| (c.to_string(), *t)).collect()
}
/// CSV in → rollup → **parquet sink** on disk → re-ingest the parquet →
/// identical rows. The sink goes in a *nested* dir to prove `write_file_sink`
/// creates the parent directory. RED-when-broken (exact-row asserts on both the
/// `SinkWrite` report and the re-read file).
#[test]
fn parquet_sink_round_trip() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_orders_csv(&src);
// nested subdir that does NOT exist yet — exercises create_dir_all.
let sink = dir.path().join("out").join("rollup.parquet");
let run = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
.with_file_output("by_customer", sink.to_str().unwrap(), "parquet")
.run(&rollup_pipeline())
.expect("native run with a parquet file sink completes");
// (a) The in-memory output path is unchanged (additive).
assert_eq!(
sorted_totals(run.output("by_customer").expect("in-memory output")),
expected_vec()
);
// (b) The sink is reported from the run handle: 2 rows, parquet, real path.
let report = run.sink("by_customer").expect("a sink was written");
assert_eq!(report.rows, 2, "two rollup groups landed on disk");
assert_eq!(report.format, "parquet");
assert_eq!(report.path, sink.to_str().unwrap());
assert!(sink.exists(), "the parquet sink file exists on disk");
assert!(
std::fs::metadata(&sink).unwrap().len() > 0,
"sink is non-empty"
);
// (c) Re-ingest the written parquet through the SAME file-source path and
// assert the exact rows — a full read → SQL → write → read-back loop.
let back = NativeBackend::new()
.with_file_input("t", sink.to_str().unwrap(), "parquet")
.run(&passthrough_pipeline())
.expect("re-ingest the written parquet");
let round_trip = sorted_totals(back.output("t_out").expect("re-read output"));
assert_eq!(
round_trip,
expected_vec(),
"the rows written to the parquet sink survive a read-back byte-for-byte"
);
knut_thund::functional_status(
"knut-thund/native_file_sink",
"parquet_round_trip",
round_trip == expected_vec() && report.rows == 2,
"orders.csv -> rollup -> rollup.parquet (sink) -> re-ingest -> identical rows",
);
}
/// The same round trip, but the sink is a **CSV** file — proves the csv arm of
/// `write_file_sink` and that the header + rows re-parse cleanly.
#[test]
fn csv_sink_round_trip() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_orders_csv(&src);
let sink = dir.path().join("rollup.csv");
let run = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
.with_file_output("by_customer", sink.to_str().unwrap(), "csv")
.run(&rollup_pipeline())
.expect("native run with a csv file sink completes");
let report = run.sink("by_customer").expect("a csv sink was written");
assert_eq!(report.rows, 2);
assert_eq!(report.format, "csv");
assert!(sink.exists(), "the csv sink file exists on disk");
let back = NativeBackend::new()
.with_file_input("t", sink.to_str().unwrap(), "csv")
.run(&passthrough_pipeline())
.expect("re-ingest the written csv");
let round_trip = sorted_totals(back.output("t_out").expect("re-read output"));
assert_eq!(
round_trip,
expected_vec(),
"the rows written to the csv sink survive a read-back"
);
knut_thund::functional_status(
"knut-thund/native_file_sink",
"csv_round_trip",
round_trip == expected_vec(),
"orders.csv -> rollup -> rollup.csv (sink) -> re-ingest -> identical rows",
);
}
/// A sink pointed at a dataset the pipeline never produces writes **no file**
/// and reports no [`SinkWrite`] — honest, not a silent empty file. RED-when-
/// broken: if the sink fabricated an empty file this would fail on `!exists()`.
#[test]
fn sink_for_unproduced_dataset_writes_nothing() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_orders_csv(&src);
let sink = dir.path().join("ghost.parquet");
let run = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
// `does_not_exist` is never a dataset in the pipeline.
.with_file_output("does_not_exist", sink.to_str().unwrap(), "parquet")
.run(&rollup_pipeline())
.expect("run still completes; the unmatched sink is skipped");
assert!(
run.sink("does_not_exist").is_none(),
"no sink report for a missing dataset"
);
assert!(
!sink.exists(),
"no file is written for a dataset that produced nothing"
);
knut_thund::functional_status(
"knut-thund/native_file_sink",
"unproduced_dataset_no_file",
run.sink("does_not_exist").is_none() && !sink.exists(),
"a sink for a non-existent dataset writes nothing and reports nothing",
);
}
// ---------------------------------------------------------------------------
// Partitioned sink — honoring `Dataset.partition_cols` (Hive directory write).
// ---------------------------------------------------------------------------
/// A regional rollup whose SINK dataset declares `partition_cols = ["region"]`,
/// so the native backend must write a Hive-partitioned directory rather than a
/// single file.
fn regional_rollup_pipeline() -> Pipeline {
Pipeline::new("orders_by_region")
.with_dataset(Dataset::new("by_customer", OutputType::Sink).with_partition_cols(["region"]))
.with_flow(Flow::batch("agg", "by_customer", ["orders"]).with_query(
"SELECT region, customer, SUM(amount) AS total \
FROM orders GROUP BY region, customer",
))
}
/// Write the regional `orders.csv` fixture: `region,customer,amount`. Groups by
/// (region, customer): US/alice=30, US/carol=3, EU/bob=12 → 3 output rows across
/// two regions (US has 2 groups, EU has 1).
fn write_regional_orders_csv(path: &std::path::Path) {
let mut f = std::fs::File::create(path).expect("create orders.csv");
writeln!(f, "region,customer,amount").unwrap();
writeln!(f, "US,alice,10").unwrap();
writeln!(f, "EU,bob,5").unwrap();
writeln!(f, "US,alice,20").unwrap();
writeln!(f, "EU,bob,7").unwrap();
writeln!(f, "US,carol,3").unwrap();
}
/// Read a Hive-partitioned parquet directory back into `(region, customer,
/// total)` rows, reconstructing `region` from each `region=<val>` subdirectory
/// name (the partition column is encoded in the path, not the data files). Uses
/// the synchronous arrow parquet reader — no runtime needed.
fn read_hive_parquet(root: &std::path::Path) -> Vec<(String, String, i64)> {
use datafusion::parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
let mut out: Vec<(String, String, i64)> = Vec::new();
for part in std::fs::read_dir(root).expect("read sink root") {
let part = part.expect("dir entry");
if !part.path().is_dir() {
continue;
}
let dirname = part.file_name().to_string_lossy().to_string();
// Hive layout: `region=US`.
let region = dirname
.split_once('=')
.map(|(_, v)| v.to_string())
.unwrap_or_else(|| panic!("subdir `{dirname}` is not a `col=val` partition dir"));
for file in std::fs::read_dir(part.path()).expect("read partition dir") {
let file = file.expect("file entry");
if file.path().extension().and_then(|e| e.to_str()) != Some("parquet") {
continue;
}
let f = std::fs::File::open(file.path()).expect("open parquet");
let reader = ParquetRecordBatchReaderBuilder::try_new(f)
.expect("parquet reader builder")
.build()
.expect("build parquet reader");
for batch in reader {
let batch = batch.expect("read batch");
// The data files carry only the NON-partition columns
// (customer, total) — region lives in the path.
let cust_col = batch
.column_by_name("customer")
.expect("customer column present in data file");
let cust: Vec<String> =
if let Some(a) = cust_col.as_any().downcast_ref::<StringArray>() {
(0..a.len()).map(|i| a.value(i).to_string()).collect()
} else if let Some(a) = cust_col.as_any().downcast_ref::<StringViewArray>() {
(0..a.len()).map(|i| a.value(i).to_string()).collect()
} else {
panic!("customer column is neither Utf8 nor Utf8View");
};
let total = batch
.column_by_name("total")
.expect("total column present")
.as_any()
.downcast_ref::<Int64Array>()
.expect("total is Int64");
assert!(
batch.column_by_name("region").is_none(),
"region is the partition column — it must NOT be duplicated into the data file"
);
for (i, c) in cust.into_iter().enumerate() {
out.push((region.clone(), c, total.value(i)));
}
}
}
}
out.sort();
out
}
/// A sink dataset that declares `partition_cols = ["region"]` must land a
/// **Hive-partitioned directory** (`region=US/…`, `region=EU/…`), not a single
/// file. RED-when-broken: if `partition_cols` were ignored (the pre-fix
/// behaviour) a single `by_region` FILE would be written — `is_dir()` and the
/// per-region subdirectory assertions would both fail.
#[test]
fn partitioned_parquet_sink_writes_hive_directory() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_regional_orders_csv(&src);
// Directory ROOT (nested + not pre-existing — also exercises create_dir_all).
let sink = dir.path().join("out").join("by_region");
let run = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
.with_file_output("by_customer", sink.to_str().unwrap(), "parquet")
.run(®ional_rollup_pipeline())
.expect("native run with a partitioned parquet sink completes");
// (a) The sink report reflects the partitioning + row count.
let report = run
.sink("by_customer")
.expect("a partitioned sink was written");
assert_eq!(
report.partition_cols,
vec!["region".to_string()],
"partitioned by region"
);
assert_eq!(report.format, "parquet");
assert_eq!(report.rows, 3, "three (region, customer) groups landed");
assert_eq!(
report.path,
sink.to_str().unwrap(),
"the sink path is the directory root"
);
// (b) The output is a DIRECTORY with one `region=<val>` subdir per region.
assert!(
sink.is_dir(),
"a partitioned sink is a directory, not a single file"
);
let mut parts: Vec<String> = std::fs::read_dir(&sink)
.expect("read sink dir")
.filter_map(|e| e.ok())
.filter(|e| e.path().is_dir())
.map(|e| e.file_name().to_string_lossy().to_string())
.collect();
parts.sort();
assert_eq!(
parts,
vec!["region=EU".to_string(), "region=US".to_string()],
"one Hive subdirectory per distinct region"
);
// (c) Read the partitioned files back and reconstruct the full rows — the
// region comes from the path, proving a real Hive layout.
let got = read_hive_parquet(&sink);
let expected = vec![
("EU".to_string(), "bob".to_string(), 12),
("US".to_string(), "alice".to_string(), 30),
("US".to_string(), "carol".to_string(), 3),
];
assert_eq!(
got, expected,
"partitioned rows survive a read-back with region from the path"
);
knut_thund::functional_status(
"knut-thund/native_file_sink",
"partitioned_parquet",
got == expected && parts.len() == 2 && report.partition_cols == vec!["region".to_string()],
"orders.csv -> rollup -> region=US/ , region=EU/ (Hive parquet sink) -> read-back identical",
);
}
/// A declared partition column that is NOT in the produced schema must fail the
/// run **loudly** (loud-gap law) — never silently write an unpartitioned dump
/// under a name that promised partitioning. RED-when-broken: if the check were
/// dropped the run would succeed and a file/dir would appear.
#[test]
fn partition_col_absent_from_schema_fails_loudly() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_regional_orders_csv(&src);
let sink = dir.path().join("out").join("bad");
// The rollup produces (region, customer, total); `nope` is not a column.
let pipeline = Pipeline::new("bad_partition")
.with_dataset(Dataset::new("by_customer", OutputType::Sink).with_partition_cols(["nope"]))
.with_flow(Flow::batch("agg", "by_customer", ["orders"]).with_query(
"SELECT region, customer, SUM(amount) AS total FROM orders GROUP BY region, customer",
));
let res = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
.with_file_output("by_customer", sink.to_str().unwrap(), "parquet")
.run(&pipeline);
assert!(
res.is_err(),
"an unknown partition column must fail the run, not drop partitioning"
);
assert!(
!sink.exists(),
"no partitioned output is written when the partition column is invalid"
);
knut_thund::functional_status(
"knut-thund/native_file_sink",
"partition_col_loud_gap",
res.is_err() && !sink.exists(),
"a partition column absent from the output schema fails loudly (no silent unpartitioned dump)",
);
}
// ---------------------------------------------------------------------------
// Declarative file sink — a `Dataset` that names its own on-disk `path` in the
// IR, so a pipeline (or a `.ron` spec) lands its outputs with NO imperative
// `with_file_output` call. The imperative builder still wins when both are set.
// ---------------------------------------------------------------------------
/// A Sink dataset that declares its `path` in the IR is written after the run —
/// no `with_file_output`. RED-when-broken: if the declarative sink weren't
/// honored, `run.sink(...)` is `None` and the file never appears.
#[test]
fn declarative_sink_writes_file_without_imperative_output() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_orders_csv(&src);
// Nested + non-existent dir → also exercises create_dir_all on the IR path.
let sink = dir.path().join("out").join("rollup.parquet");
let pipeline = Pipeline::new("orders_rollup")
.with_dataset(
Dataset::new("by_customer", OutputType::Sink)
.with_path(sink.to_str().unwrap())
.with_format("parquet"),
)
.with_flow(
Flow::batch("agg", "by_customer", ["orders"])
.with_query("SELECT customer, SUM(amount) AS total FROM orders GROUP BY customer"),
);
let run = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
// No .with_file_output — the sink path comes from the IR dataset.
.run(&pipeline)
.expect("declarative-sink run completes");
let report = run
.sink("by_customer")
.expect("the IR-declared sink was written");
assert_eq!(report.rows, 2, "two rollup groups landed");
assert_eq!(report.format, "parquet");
assert_eq!(
report.path,
sink.to_str().unwrap(),
"wrote to the IR-declared path"
);
assert!(sink.exists(), "the declared parquet file exists on disk");
// Re-ingest the written file → identical rows (full read → SQL → declared
// write → read-back loop, driven entirely by the IR).
let back = NativeBackend::new()
.with_file_input("t", sink.to_str().unwrap(), "parquet")
.run(&passthrough_pipeline())
.expect("re-ingest the declaratively-written parquet");
let round_trip = sorted_totals(back.output("t_out").expect("re-read output"));
assert_eq!(round_trip, expected_vec());
knut_thund::functional_status(
"knut-thund/native_file_sink",
"declarative_sink",
sink.exists() && report.rows == 2 && round_trip == expected_vec(),
"an IR-declared Dataset.path lands the file with no imperative with_file_output",
);
}
/// A declarative sink with NO explicit `format` infers it from the path
/// extension (`.csv` → the csv writer). RED-when-broken: a wrong/absent
/// inference would write parquet bytes to a `.csv` name and the csv re-ingest
/// below would mis-parse (rows would not match).
#[test]
fn declarative_sink_infers_csv_from_extension() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_orders_csv(&src);
let sink = dir.path().join("rollup.csv");
let pipeline = Pipeline::new("orders_rollup")
.with_dataset(
// No .with_format — the `.csv` extension must select the csv writer.
Dataset::new("by_customer", OutputType::Sink).with_path(sink.to_str().unwrap()),
)
.with_flow(
Flow::batch("agg", "by_customer", ["orders"])
.with_query("SELECT customer, SUM(amount) AS total FROM orders GROUP BY customer"),
);
let run = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
.run(&pipeline)
.expect("declarative csv-by-extension run completes");
let report = run.sink("by_customer").expect("sink written");
assert_eq!(
report.format, "csv",
"the .csv extension inferred the csv writer"
);
assert!(sink.exists());
let back = NativeBackend::new()
.with_file_input("t", sink.to_str().unwrap(), "csv")
.run(&passthrough_pipeline())
.expect("re-ingest the inferred csv");
let round_trip = sorted_totals(back.output("t_out").expect("re-read"));
assert_eq!(round_trip, expected_vec());
knut_thund::functional_status(
"knut-thund/native_file_sink",
"declarative_sink_infer_csv",
report.format == "csv" && round_trip == expected_vec(),
"a declared sink path ending .csv infers the csv writer with no explicit format",
);
}
/// When a dataset both declares a `path` AND an imperative `with_file_output`
/// targets it, the imperative one WINS and the declared path is NOT also
/// written. RED-when-broken: if the declarative loop didn't skip
/// imperatively-covered datasets, BOTH files would exist (a double write) and
/// the `!declared.exists()` assert fails.
#[test]
fn imperative_sink_overrides_declarative_path() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_orders_csv(&src);
let declared = dir.path().join("declared.parquet");
let imperative = dir.path().join("imperative.parquet");
let pipeline = Pipeline::new("orders_rollup")
.with_dataset(
Dataset::new("by_customer", OutputType::Sink)
.with_path(declared.to_str().unwrap())
.with_format("parquet"),
)
.with_flow(
Flow::batch("agg", "by_customer", ["orders"])
.with_query("SELECT customer, SUM(amount) AS total FROM orders GROUP BY customer"),
);
let run = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
.with_file_output("by_customer", imperative.to_str().unwrap(), "parquet")
.run(&pipeline)
.expect("run with both an imperative and a declarative sink");
let report = run.sink("by_customer").expect("a sink was written");
assert_eq!(
report.path,
imperative.to_str().unwrap(),
"the imperative path wins"
);
assert!(imperative.exists(), "the imperative file was written");
assert!(
!declared.exists(),
"the declared path is NOT double-written"
);
knut_thund::functional_status(
"knut-thund/native_file_sink",
"imperative_overrides_declarative",
imperative.exists() && !declared.exists(),
"an imperative with_file_output wins over a dataset's declared path (no double write)",
);
}
/// A declarative sink honors `partition_cols` — a `Dataset` naming both a `path`
/// and partition columns lands a Hive directory, read back through the
/// partitioned source. RED-when-broken: without the declarative-partition wiring
/// a single file (or nothing) would be written and `is_dir()` / the read-back
/// row count would fail.
#[test]
fn declarative_partitioned_sink_writes_hive_directory() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_regional_orders_csv(&src);
let root = dir.path().join("out").join("by_region");
let pipeline = Pipeline::new("orders_by_region")
.with_dataset(
Dataset::new("by_customer", OutputType::Sink)
.with_path(root.to_str().unwrap())
.with_format("parquet")
.with_partition_cols(["region"]),
)
.with_flow(Flow::batch("agg", "by_customer", ["orders"]).with_query(
"SELECT region, customer, SUM(amount) AS total FROM orders GROUP BY region, customer",
));
let run = NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
.run(&pipeline)
.expect("declarative partitioned-sink run completes");
let report = run
.sink("by_customer")
.expect("a partitioned sink was written");
assert_eq!(report.partition_cols, vec!["region".to_string()]);
assert!(
root.is_dir(),
"a declarative partitioned sink is a Hive directory"
);
// Read it back through the partitioned source path.
let back = NativeBackend::new()
.with_partitioned_file_input("t", root.to_str().unwrap(), "parquet", ["region"])
.run(
&Pipeline::new("read_partitioned")
.with_dataset(Dataset::new("t_out", OutputType::MaterializedView))
.with_flow(
Flow::batch("scan", "t_out", ["t"])
.with_query("SELECT region, customer, total FROM t"),
),
)
.expect("read the declarative Hive dir back");
let rows: usize = back
.output("t_out")
.expect("read-back output")
.iter()
.map(|b| b.num_rows())
.sum();
assert_eq!(
rows, 3,
"three (region, customer) groups survive the declarative partitioned round trip"
);
knut_thund::functional_status(
"knut-thund/native_file_sink",
"declarative_partitioned_sink",
root.is_dir() && rows == 3,
"an IR-declared sink with partition_cols writes a Hive dir, read back through the partitioned source",
);
}
// ---------------------------------------------------------------------------
// Declarative write mode — Overwrite vs Append. Re-running a partitioned file
// sink WITHOUT a clear accumulates new part-files each run (the append bug the
// partitioned-sink problem entry flagged). `WriteMode::Overwrite` clears the
// target dir first, so a re-run holds ONE generation's rows; `WriteMode::Append`
// (the default) accumulates — doubled after two runs.
// ---------------------------------------------------------------------------
/// The regional rollup with an explicit sink [`WriteMode`] — same query and
/// partitioning as [`regional_rollup_pipeline`], but the sink dataset declares
/// whether a re-run overwrites or appends.
fn regional_rollup_pipeline_mode(mode: WriteMode) -> Pipeline {
Pipeline::new("orders_by_region")
.with_dataset(
Dataset::new("by_customer", OutputType::Sink)
.with_partition_cols(["region"])
.with_write_mode(mode),
)
.with_flow(Flow::batch("agg", "by_customer", ["orders"]).with_query(
"SELECT region, customer, SUM(amount) AS total \
FROM orders GROUP BY region, customer",
))
}
/// Run the regional partitioned sink into the same directory `runs` times,
/// driving the write mode from `pipeline` (each run re-reads the fixture).
fn run_partitioned_sink_times(
sink: &std::path::Path,
src: &std::path::Path,
pipeline: &Pipeline,
runs: usize,
) {
for _ in 0..runs {
NativeBackend::new()
.with_file_input("orders", src.to_str().unwrap(), "csv")
.with_file_output("by_customer", sink.to_str().unwrap(), "parquet")
.run(pipeline)
.expect("native run with a partitioned sink completes");
}
}
/// The three (region, customer) groups a single generation of the regional
/// rollup produces: EU/bob=12, US/alice=30, US/carol=3.
fn one_generation() -> Vec<(String, String, i64)> {
vec![
("EU".to_string(), "bob".to_string(), 12),
("US".to_string(), "alice".to_string(), 30),
("US".to_string(), "carol".to_string(), 3),
]
}
/// `WriteMode::Overwrite`: running the SAME partitioned sink twice into the same
/// directory holds exactly ONE generation's rows — the second run replaces the
/// first (fresh-dir semantics), it does not accumulate. RED-when-broken:
/// neutralize `clear_sink_target` (make the Overwrite arm a no-op) and the
/// directory keeps both generations' part-files → 6 rows, and this assert fails.
#[test]
fn write_mode_overwrite_holds_one_generation() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_regional_orders_csv(&src);
let sink = dir.path().join("out").join("by_region");
let pipeline = regional_rollup_pipeline_mode(WriteMode::Overwrite);
run_partitioned_sink_times(&sink, &src, &pipeline, 2);
let got = read_hive_parquet(&sink);
assert_eq!(
got,
one_generation(),
"Overwrite: a re-run replaces the prior contents — exactly one generation, no duplicates"
);
assert_eq!(got.len(), 3, "three groups, not doubled");
knut_thund::functional_status(
"knut-thund/native_file_sink",
"write_mode_overwrite",
got == one_generation(),
"re-running an Overwrite partitioned sink clears the dir first — one generation, not doubled",
);
}
/// `WriteMode::Append` (explicit): running the SAME partitioned sink twice
/// ACCUMULATES — the directory holds both generations' part-files, so every
/// group appears twice (6 rows total). This pins the pre-existing behaviour as
/// the default and is the counterpart the Overwrite test contrasts against.
/// RED-when-broken: if Overwrite semantics leaked into Append (an always-clear),
/// the count collapses to 3 and this assert fails.
#[test]
fn write_mode_append_accumulates_generations() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_regional_orders_csv(&src);
let sink = dir.path().join("out").join("by_region");
let pipeline = regional_rollup_pipeline_mode(WriteMode::Append);
run_partitioned_sink_times(&sink, &src, &pipeline, 2);
let got = read_hive_parquet(&sink);
assert_eq!(
got.len(),
6,
"Append: two runs accumulate — every group appears twice (doubled)"
);
// Every single-generation group is present exactly twice.
for row in one_generation() {
let n = got.iter().filter(|r| **r == row).count();
assert_eq!(
n, 2,
"group {row:?} appears once per run (two runs -> twice)"
);
}
knut_thund::functional_status(
"knut-thund/native_file_sink",
"write_mode_append",
got.len() == 6,
"re-running an Append partitioned sink accumulates — two generations, doubled rows",
);
}
/// The default write mode (no `.with_write_mode`) IS `Append` — the pre-existing
/// behaviour — so an unchanged pipeline keeps accumulating on re-run (L2). This
/// proves the additive default: existing specs deserialize/behave unchanged.
#[test]
fn write_mode_default_is_append() {
let dir = tempfile::tempdir().expect("tempdir");
let src = dir.path().join("orders.csv");
write_regional_orders_csv(&src);
let sink = dir.path().join("out").join("by_region");
// `regional_rollup_pipeline()` sets NO write mode → the default.
let pipeline = regional_rollup_pipeline();
run_partitioned_sink_times(&sink, &src, &pipeline, 2);
let got = read_hive_parquet(&sink);
assert_eq!(
got.len(),
6,
"the default write mode is Append — two runs accumulate (today's behaviour preserved)"
);
assert_eq!(
WriteMode::default(),
WriteMode::Append,
"the type default is Append"
);
knut_thund::functional_status(
"knut-thund/native_file_sink",
"write_mode_default_append",
got.len() == 6 && WriteMode::default() == WriteMode::Append,
"an unchanged pipeline (no write_mode) defaults to Append — additive, no behaviour change",
);
}