crates/nornir-parquet-delta/tests/mirror.rs
//! **The gates.** A partially-filled mirror must be readable by the ORDINARY parquet
//! reader for the groups it holds, must refuse the groups it does not, and must refuse
//! itself entirely once the source has moved.
//!
//! The fixture is a real parquet file written by the stock arrow writer — a mirror that
//! only round-trips its own bytes proves nothing about whether `parquet` will accept a
//! file with holes in it, and that acceptance is the entire premise.
use std::sync::Arc;
use arrow_array::{Int64Array, RecordBatch, StringArray};
use arrow_schema::{DataType, Field, Schema};
use nornir_parquet_delta::{footer_range, plan, tail_probe, Error, Mirror, Need, Source};
use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
use parquet::arrow::ProjectionMask;
use parquet::file::properties::WriterProperties;
const GROUPS: usize = 8;
const ROWS_PER_GROUP: usize = 500;
/// A three-column table with `GROUPS` row groups: `id` (i64), `payload` (a fat string, so
/// the column sizes differ by an order of magnitude and a projection is visibly cheaper),
/// `tag` (a small string).
fn write_fixture(path: &std::path::Path) {
let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("payload", DataType::Utf8, false),
Field::new("tag", DataType::Utf8, false),
]));
let props = WriterProperties::builder().set_max_row_group_row_count(Some(ROWS_PER_GROUP)).build();
let file = std::fs::File::create(path).unwrap();
let mut w = parquet::arrow::ArrowWriter::try_new(file, Arc::clone(&schema), Some(props)).unwrap();
let n = GROUPS * ROWS_PER_GROUP;
let ids: Vec<i64> = (0..n as i64).collect();
let payload: Vec<String> = (0..n).map(|i| format!("{:0>200}", i * 7919)).collect();
let tags: Vec<String> = (0..n).map(|i| format!("t{}", i % 13)).collect();
let batch = RecordBatch::try_new(
schema,
vec![
Arc::new(Int64Array::from(ids)),
Arc::new(StringArray::from(payload)),
Arc::new(StringArray::from(tags)),
],
)
.unwrap();
w.write(&batch).unwrap();
w.close().unwrap();
}
/// The two-step footer handshake, played out over a local file standing in for a server.
fn source_of(path: &std::path::Path, version: &str) -> Source {
use std::os::unix::fs::FileExt;
let f = std::fs::File::open(path).unwrap();
let len = f.metadata().unwrap().len();
let mut tail = [0u8; 8];
f.read_exact_at(&mut tail, tail_probe(len).offset).unwrap();
let fr = footer_range(len, &tail).unwrap();
let mut footer = vec![0u8; fr.len as usize];
f.read_exact_at(&mut footer, fr.offset).unwrap();
Source { len, footer, version: version.to_string() }
}
/// Serve a plan's pieces out of the source file — the transport this crate does not own.
fn deliver(mirror: &mut Mirror, src_path: &std::path::Path, need: &Need) -> u64 {
use std::os::unix::fs::FileExt;
let f = std::fs::File::open(src_path).unwrap();
let p = mirror.plan(need).unwrap();
let fetched = p.fetch_bytes;
for piece in p.pieces {
let mut buf = vec![0u8; piece.range.len as usize];
f.read_exact_at(&mut buf, piece.range.offset).unwrap();
mirror.accept(piece, &buf).unwrap();
}
mirror.save().unwrap();
fetched
}
fn col_idx(m: &Mirror, name: &str) -> usize {
Need::columns_by_path(m.metadata(), |p| p == name)[0]
}
/// GATE 1 — the stock reader reads a mirror with holes in it, and gets the SAME rows the
/// source has for the groups the mirror holds.
#[test]
fn the_ordinary_parquet_reader_reads_a_partially_filled_mirror() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("source.parquet");
write_fixture(&src);
let s = source_of(&src, "v1");
let mut m = Mirror::create(dir.path().join("mirror.parquet"), &s).unwrap();
assert_eq!(m.metadata().num_row_groups(), GROUPS);
let want_groups = vec![2usize, 5];
let id = col_idx(&m, "id");
let tag = col_idx(&m, "tag");
let need = Need::new(want_groups.clone(), vec![id, tag]);
assert!(!m.holds(&need).unwrap(), "a fresh mirror holds nothing");
let fetched = deliver(&mut m, &src, &need);
assert!(fetched > 0, "the plan for two groups of two columns cannot be free");
assert!(m.holds(&need).unwrap(), "after delivery the mirror holds the need");
assert!(m.plan(&need).unwrap().satisfied());
// Read the mirror with the stock reader, through the stock builder, and compare
// against the same read over the source.
fn ids_of<R: parquet::file::reader::ChunkReader + 'static>(
b: ParquetRecordBatchReaderBuilder<R>,
leaves: [usize; 2],
groups: Vec<usize>,
) -> Vec<i64> {
let mask = ProjectionMask::leaves(b.parquet_schema(), leaves);
let it = b.with_projection(mask).with_row_groups(groups).build().unwrap();
let mut out = Vec::new();
for batch in it {
let batch = batch.unwrap();
let ids =
batch.column_by_name("id").unwrap().as_any().downcast_ref::<Int64Array>().unwrap();
out.extend(ids.values().iter().copied());
}
out
}
let mirrored = ids_of(
ParquetRecordBatchReaderBuilder::try_new(m.reader()).expect("stock reader opens the mirror"),
[id, tag],
want_groups.clone(),
);
let truth = ids_of(
ParquetRecordBatchReaderBuilder::try_new(std::fs::File::open(&src).unwrap()).unwrap(),
[id, tag],
want_groups.clone(),
);
assert_eq!(mirrored.len(), 2 * ROWS_PER_GROUP);
assert_eq!(mirrored, truth, "a mirrored group must read back the source's own rows");
}
/// GATE 2 (RED FIRST) — reading a group the mirror does NOT hold is a named error, not a
/// page of zeros. Proven red by asking for group 0, which is exactly the read the sparse
/// hole would otherwise answer.
#[test]
fn a_group_the_mirror_does_not_hold_is_an_error_and_never_zeros() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("source.parquet");
write_fixture(&src);
let s = source_of(&src, "v1");
let mut m = Mirror::create(dir.path().join("mirror.parquet"), &s).unwrap();
let id = col_idx(&m, "id");
deliver(&mut m, &src, &Need::new(vec![2], vec![id]));
let b = ParquetRecordBatchReaderBuilder::try_new(m.reader()).unwrap();
let mask = ProjectionMask::leaves(b.parquet_schema(), [id]);
let mut it = b.with_projection(mask).with_row_groups(vec![0]).build().unwrap();
let first = it.next().expect("the reader yields something for a group it was given");
let err = first.expect_err("a hole must not decode into rows");
let text = format!("{err:?}");
assert!(text.contains("mirror hole"), "the failure must name itself: {text}");
// And the column it does NOT hold of a group it DOES hold, which is the subtler half.
let payload = col_idx(&m, "payload");
assert!(!m.holds(&Need::new(vec![2], vec![payload])).unwrap());
let b = ParquetRecordBatchReaderBuilder::try_new(m.reader()).unwrap();
let mask = ProjectionMask::leaves(b.parquet_schema(), [payload]);
let mut it = b.with_projection(mask).with_row_groups(vec![2]).build().unwrap();
let err = it.next().unwrap().expect_err("an unheld column of a held group is still a hole");
assert!(format!("{err:?}").contains("mirror hole"));
}
/// GATE 3 (RED FIRST) — a moved source invalidates the whole mirror. Proven red by
/// opening with the token it was created under, which must succeed.
#[test]
fn a_changed_source_token_invalidates_the_mirror() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("source.parquet");
write_fixture(&src);
let path = dir.path().join("mirror.parquet");
let mut m = Mirror::create(&path, &source_of(&src, "cut-2026-09-01")).unwrap();
let id = col_idx(&m, "id");
deliver(&mut m, &src, &Need::new(vec![1], vec![id]));
drop(m);
let same = Mirror::open(&path, "cut-2026-09-01").expect("the same token opens it");
assert!(same.holds(&Need::new(vec![1], vec![id])).unwrap(), "the bitmap survived the round trip");
drop(same);
match Mirror::open(&path, "cut-2026-09-02") {
Err(Error::StaleSource { held, expected }) => {
assert_eq!(held, "cut-2026-09-01");
assert_eq!(expected, "cut-2026-09-02");
}
other => panic!("a moved source must be StaleSource, got {other:?}", other = other.map(|_| "Ok")),
}
// …and `open_or_create` starts a fresh one rather than reading the old offsets.
let fresh = Mirror::open_or_create(&path, &source_of(&src, "cut-2026-09-02")).unwrap();
assert_eq!(fresh.presence().count(), 0, "a stale mirror is discarded, not repaired");
assert_eq!(fresh.version(), "cut-2026-09-02");
}
/// The file is SPARSE: it is as long as the source and occupies a fraction of it.
///
/// Reported, not assumed — on a filesystem without sparse support this is the test that
/// says so out loud instead of the design quietly costing full size.
#[test]
fn the_mirror_file_is_sparse() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("source.parquet");
write_fixture(&src);
let s = source_of(&src, "v1");
let apparent = s.len;
let mut m = Mirror::create(dir.path().join("mirror.parquet"), &s).unwrap();
let id = col_idx(&m, "id");
deliver(&mut m, &src, &Need::new(vec![0], vec![id]));
assert_eq!(m.apparent_bytes(), apparent);
let on_disk = m.on_disk_bytes().unwrap();
println!("apparent {apparent} B, on disk {on_disk} B");
assert!(
on_disk < apparent,
"a mirror holding one column of one row group of {GROUPS} occupies {on_disk} B of {apparent} B \
— the filesystem is not giving sparse files, and the whole storage claim is void",
);
}
/// A body of the wrong length is refused. Writing it would put every later offset in that
/// chunk's neighbourhood out of phase with the footer, undetectably.
#[test]
fn a_chunk_body_of_the_wrong_length_is_refused() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("source.parquet");
write_fixture(&src);
let s = source_of(&src, "v1");
let mut m = Mirror::create(dir.path().join("mirror.parquet"), &s).unwrap();
let id = col_idx(&m, "id");
let p = m.plan(&Need::new(vec![0], vec![id])).unwrap();
let piece = p.pieces[0];
let short = vec![0u8; piece.range.len as usize - 1];
match m.accept(piece, &short) {
Err(Error::WrongLength { want, got, .. }) => {
assert_eq!(want, piece.range.len);
assert_eq!(got, piece.range.len - 1);
}
other => panic!("a short body must be WrongLength, got {:?}", other.map(|()| "Ok")),
}
assert_eq!(m.presence().count(), 0, "a refused chunk is not marked held");
}
/// The plan is a function of the SET, and it says what a cold query costs before any
/// mirror exists.
#[test]
fn a_cold_plan_prices_a_query_without_a_mirror() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("source.parquet");
write_fixture(&src);
let f = std::fs::File::open(&src).unwrap();
let meta = parquet::file::metadata::ParquetMetaDataReader::new().parse_and_finish(&f).unwrap();
let payload = Need::columns_by_path(&meta, |p| p == "payload");
let small = Need::columns_by_path(&meta, |p| p == "id" || p == "tag");
let groups = Need::row_groups_where(&meta, |g| g.num_rows() > 0);
assert_eq!(groups.len(), GROUPS);
let fat = plan(&meta, &Need::new(groups.clone(), payload), None).unwrap();
let thin = plan(&meta, &Need::new(groups.clone(), small), None).unwrap();
assert!(
fat.fetch_bytes > thin.fetch_bytes * 4,
"the projection has to be worth something: payload {} vs id+tag {}",
fat.fetch_bytes,
thin.fetch_bytes,
);
assert_eq!(fat.held_bytes, 0, "no mirror ⇒ nothing held");
assert!(!fat.satisfied());
// Order-independence: the same set, built backwards, is the same plan.
let mut backwards = groups.clone();
backwards.reverse();
let again = plan(&meta, &Need::new(backwards, Need::columns_by_path(&meta, |p| p == "payload")), None).unwrap();
assert_eq!(again, fat, "a plan is a function of the set, not of the caller's order");
// Coalescing: one row group's chunks are contiguous, so a whole-table single-column
// plan cannot need more ranges than it has pieces.
assert!(fat.coalesced().len() <= fat.pieces.len());
}
/// A need naming a row group or column that does not exist is an error, not an empty
/// plan — an empty plan is shaped exactly like a satisfied one.
#[test]
fn a_need_outside_the_source_is_an_error_and_not_an_empty_plan() {
let dir = tempfile::tempdir().unwrap();
let src = dir.path().join("source.parquet");
write_fixture(&src);
let s = source_of(&src, "v1");
let m = Mirror::create(dir.path().join("mirror.parquet"), &s).unwrap();
assert!(matches!(
m.plan(&Need::new(vec![GROUPS], vec![0])),
Err(Error::OutOfRange { what: "row group", .. })
));
assert!(matches!(
m.plan(&Need::new(vec![0], vec![99])),
Err(Error::OutOfRange { what: "column", .. })
));
assert!(matches!(m.holds(&Need::new(vec![GROUPS], vec![0])), Err(Error::OutOfRange { .. })));
}
/// **The held version is readable WITHOUT a source, and it is the mirror's own — not a
/// hopeful echo of whatever the caller asks for.**
///
/// `Mirror::held_version` exists so an OFFLINE consumer can open the file it already
/// holds instead of being told it has nothing (korp lane MT, 2026-09-03: a thin client
/// started while its server was down could reach 43.8 MB of correctly mirrored ground
/// only through this call). That is a real loosening of the freshness rule, so it has to
/// be exactly as loose as it claims and no looser:
///
/// * it returns the token the sidecar RECORDS, so a caller can see what it is assuming;
/// * opening on that token gives back the presence the mirror really had;
/// * and `Mirror::open` on a DIFFERENT token still refuses, so the loosening is confined
/// to the caller that asked for it.
///
/// **RED FIRST, driven on this build:** make `held_version` return
/// `expect_version`-shaped nonsense (e.g. `Ok(String::new())`) and the open below fails
/// as `StaleSource`; delete the `st.version != expect_version` arm and the last assertion
/// — that a wrong token is still refused — fails instead. Neither half is green alone.
#[test]
fn the_held_version_is_readable_offline_and_is_the_mirrors_own() {
let dir = tempfile::tempdir().expect("tmp");
let src_path = dir.path().join("source.parquet");
write_fixture(&src_path);
let src = source_of(&src_path, "cut-v1");
let mirror_path = dir.path().join("mirror.parquet");
let mut m = Mirror::create(&mirror_path, &src).expect("create");
let need = Need::new(vec![0], vec![col_idx(&m, "id")]);
let moved = deliver(&mut m, &src_path, &need);
m.save().expect("save");
let held_before = m.presence().count();
assert!(moved > 0 && held_before > 0, "nothing was filled — the reopen proves nothing");
drop(m);
// NO SOURCE. Only the file and its sidecar.
let token = Mirror::held_version(&mirror_path).expect("the sidecar names its version");
assert_eq!(token, "cut-v1", "the token must be the mirror's own, not a guess");
let reopened = Mirror::open(&mirror_path, &token).expect("open on its own token");
assert_eq!(
reopened.presence().count(),
held_before,
"an offline open must give back the ground it really holds",
);
assert!(reopened.holds(&need).expect("holds"), "…and it must answer for it");
// The loosening is CONFINED: a wrong token is still refused.
let e = match Mirror::open(&mirror_path, "cut-v2") {
Err(e) => e,
Ok(_) => panic!("a moved cut must still refuse to open"),
};
assert!(
matches!(e, nornir_parquet_delta::Error::StaleSource { .. }),
"a different token must still be StaleSource, not opened: {e:?}",
);
}