diff --git a/.claude/claude.md b/.claude/claude.md index 15dd058..7f80ce2 100644 --- a/.claude/claude.md +++ b/.claude/claude.md @@ -104,3 +104,15 @@ If a schema change causes `site/examples/` to fail, don't fix them. Instead repo ## Code - Use nanoparquet for reading/writing parquet files (R code). + +## Benchmarking + +When benchmarking performance work (e.g. the parquet validators): + +- Benchmark on a realistic large dataset: a ~10M-row parquet file generated with nanoparquet, not a toy fixture. Use the duckdb row group size, i.e. `num_rows_per_row_group = 122880` +- Always report both speed and peak memory (RSS), as a summary table. Use best-of-N timings (e.g. best of 3). +- Isolate component costs (e.g. decode vs. hashing/dedup) before optimising, so effort goes where the bottleneck actually is. +- Verify correctness in the same harness (known duplicates, including ones spanning row groups) and re-measure before/after; revert changes that regress. +- Prefer real measurements over guesses when weighing an alternative (e.g. a new dependency vs. hand-rolled): measure binary size, dependency count, build time, and runtime. +- **Find hotspots with `/usr/bin/sample`** (macOS, built-in, no sudo — unlike flamegraph's dtrace). Loop the operation so the process runs ~20–30s, `sample 1 -file out.txt`, then read the "Sort by top of stack" section (self-time) piped through `c++filt`. Build with `CARGO_PROFILE_RELEASE_DEBUG=2` for symbols. Ignore `__psynch_cvwait` — that's idle rayon workers, not work. +- **Judge changes with `criterion`** (`crates/data-dict-parquet/benches/uniqueness.rs`, `cargo bench`). Use a saved-baseline A/B — `--save-baseline after`, restore the old code, `--baseline after` — not sequential runs, which overstate gains on a noisy machine. Trust the p-values: only keep a change criterion calls significant, and confirm profiler-suggested tweaks actually help (some hurt — e.g. `collect()` vs. a pre-sized `Vec` defeats vectorisation). diff --git a/Cargo.lock b/Cargo.lock index 07122bf..10aab8c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -46,6 +46,12 @@ dependencies = [ "alloc-no-stdlib", ] +[[package]] +name = "allocator-api2" +version = "0.2.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -55,6 +61,12 @@ dependencies = [ "libc", ] +[[package]] +name = "anes" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299" + [[package]] name = "anstream" version = "1.0.0" @@ -178,6 +190,12 @@ version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "cast" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" + [[package]] name = "cc" version = "1.2.63" @@ -207,6 +225,33 @@ dependencies = [ "windows-link", ] +[[package]] +name = "ciborium" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42e69ffd6f0917f5c029256a24d0161db17cea3997d185db0d35926308770f0e" +dependencies = [ + "ciborium-io", + "ciborium-ll", + "serde", +] + +[[package]] +name = "ciborium-io" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05afea1e0a06c9be33d539b876f1ce3692f4afea2cb41f740e7743225ed1c757" + +[[package]] +name = "ciborium-ll" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "57663b653d948a338bfb3eeba9bb2fd5fcfaecb9e199e87e1eda4d9e8b240fd9" +dependencies = [ + "ciborium-io", + "half", +] + [[package]] name = "clap" version = "4.6.1" @@ -299,6 +344,67 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "criterion" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2b12d017a929603d80db1831cd3a24082f8137ce19c69e6447f54f5fc8d692f" +dependencies = [ + "anes", + "cast", + "ciborium", + "clap", + "criterion-plot", + "is-terminal", + "itertools", + "num-traits", + "once_cell", + "oorandom", + "plotters", + "rayon", + "regex", + "serde", + "serde_derive", + "serde_json", + "tinytemplate", + "walkdir", +] + +[[package]] +name = "criterion-plot" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b50826342786a51a89e2da3a28f1c32b06e387201bc2d19791f622c673706b1" +dependencies = [ + "cast", + "itertools", +] + +[[package]] +name = "crossbeam-deque" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dd111b7b7f7d55b72c0a6ae361660ee5853c9af73f70c3c2ef6858b950e2e51" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" + [[package]] name = "crunchy" version = "0.2.4" @@ -338,7 +444,10 @@ dependencies = [ name = "data-dict-parquet" version = "0.0.1" dependencies = [ + "criterion", + "hashbrown 0.15.5", "parquet", + "rayon", ] [[package]] @@ -352,6 +461,12 @@ dependencies = [ "syn", ] +[[package]] +name = "either" +version = "1.16.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91622ff5e7162018101f2fea40d6ebf4a78bbe5a49736a2020649edf9693679e" + [[package]] name = "encode_unicode" version = "1.0.0" @@ -504,6 +619,8 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ + "allocator-api2", + "equivalent", "foldhash 0.1.5", ] @@ -537,6 +654,12 @@ version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + [[package]] name = "iana-time-zone" version = "0.1.65" @@ -709,12 +832,32 @@ version = "3.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8bb03732005da905c88227371639bf1ad885cc712789c011c31c5fb3ab3ccf02" +[[package]] +name = "is-terminal" +version = "0.4.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" +dependencies = [ + "hermit-abi", + "libc", + "windows-sys", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" +[[package]] +name = "itertools" +version = "0.10.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0fd2260e829bddf4cb6ea802289de2f86d6a7a690192fbe91b3f46e0f2c8473" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -887,6 +1030,12 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "oorandom" +version = "11.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e" + [[package]] name = "ordered-float" version = "2.10.1" @@ -944,6 +1093,34 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +[[package]] +name = "plotters" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5aeb6f403d7a4911efb1e33402027fc44f29b5bf6def3effcc22d7bb75f2b747" +dependencies = [ + "num-traits", + "plotters-backend", + "plotters-svg", + "wasm-bindgen", + "web-sys", +] + +[[package]] +name = "plotters-backend" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df42e13c12958a16b3f7f4386b9ab1f3e7933914ecea48da7139435263a4172a" + +[[package]] +name = "plotters-svg" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "51bae2ac328883f7acdfea3d66a7c35751187f870bc81f94563733a154d7a670" +dependencies = [ + "plotters-backend", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -1047,6 +1224,26 @@ version = "6.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" +[[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + [[package]] name = "regex" version = "1.12.3" @@ -1095,6 +1292,15 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + [[package]] name = "semver" version = "1.0.28" @@ -1292,6 +1498,16 @@ dependencies = [ "zerovec", ] +[[package]] +name = "tinytemplate" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "be4d6b5f19ff7664e8c98d03e2139cb510db9b0a60b55f8e8709b689d939b6bc" +dependencies = [ + "serde", + "serde_json", +] + [[package]] name = "twox-hash" version = "1.6.3" @@ -1356,6 +1572,16 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -1459,6 +1685,25 @@ dependencies = [ "semver", ] +[[package]] +name = "web-sys" +version = "0.3.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e0871acf327f283dc6da28a1696cdc64fb355ba9f935d052021fa77f35cce69" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys", +] + [[package]] name = "windows-core" version = "0.62.2" diff --git a/crates/data-dict-parquet/Cargo.toml b/crates/data-dict-parquet/Cargo.toml index fc6470d..a14c9e1 100644 --- a/crates/data-dict-parquet/Cargo.toml +++ b/crates/data-dict-parquet/Cargo.toml @@ -14,3 +14,12 @@ parquet = { workspace = true, features = [ "zstd", "brotli" ] } +rayon = "1" +hashbrown = "0.15" + +[dev-dependencies] +criterion = "0.5" + +[[bench]] +name = "uniqueness" +harness = false diff --git a/crates/data-dict-parquet/benches/uniqueness.rs b/crates/data-dict-parquet/benches/uniqueness.rs new file mode 100644 index 0000000..75d386e --- /dev/null +++ b/crates/data-dict-parquet/benches/uniqueness.rs @@ -0,0 +1,124 @@ +//! Uniqueness benchmarks over a generated Parquet fixture. +//! +//! Row count is configurable with `DDP_BENCH_ROWS` (default 3,000,000). The +//! fixture is written once to the temp dir and reused across runs. + +use std::fs::File; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use criterion::{Criterion, criterion_group, criterion_main}; +use data_dict_parquet::{UniquenessCheck, uniqueness_stats}; +use parquet::basic::{Compression, Repetition, Type as PhysicalType}; +use parquet::data_type::{ByteArray, ByteArrayType, Int64Type}; +use parquet::file::properties::WriterProperties; +use parquet::file::writer::SerializedFileWriter; +use parquet::schema::types::Type; + +const ROWS_PER_GROUP: usize = 1_000_000; + +fn rows() -> usize { + std::env::var("DDP_BENCH_ROWS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(3_000_000) +} + +fn fixture(rows: usize) -> PathBuf { + let path = std::env::temp_dir().join(format!("ddp_bench_uniqueness_{rows}.parquet")); + if path.exists() { + return path; + } + let schema = Arc::new( + Type::group_type_builder("schema") + .with_fields(vec![ + Arc::new(field("id", PhysicalType::INT64)), + Arc::new(field("code", PhysicalType::BYTE_ARRAY)), + Arc::new(field("part_a", PhysicalType::BYTE_ARRAY)), + Arc::new(field("part_b", PhysicalType::BYTE_ARRAY)), + ]) + .build() + .unwrap(), + ); + let props = Arc::new( + WriterProperties::builder() + .set_compression(Compression::SNAPPY) + .build(), + ); + let mut writer = + SerializedFileWriter::new(File::create(&path).unwrap(), schema, props).unwrap(); + let mut written = 0; + while written < rows { + let n = ROWS_PER_GROUP.min(rows - written); + let ids: Vec = (0..n).map(|i| (written + i) as i64).collect(); + let code: Vec = ids + .iter() + .map(|&i| ByteArray::from(format!("CODE-{i:08}").into_bytes())) + .collect(); + let part_a: Vec = ids + .iter() + .map(|&i| ByteArray::from(format!("G{:04}", i % 1000).into_bytes())) + .collect(); + let part_b: Vec = ids + .iter() + .map(|&i| ByteArray::from(format!("R{i:08}").into_bytes())) + .collect(); + let mut group = writer.next_row_group().unwrap(); + let mut col = group.next_column().unwrap().unwrap(); + col.typed::() + .write_batch(&ids, None, None) + .unwrap(); + col.close().unwrap(); + for values in [&code, &part_a, &part_b] { + let mut col = group.next_column().unwrap().unwrap(); + col.typed::() + .write_batch(values, None, None) + .unwrap(); + col.close().unwrap(); + } + group.close().unwrap(); + written += n; + } + writer.close().unwrap(); + path +} + +fn field(name: &str, physical: PhysicalType) -> Type { + Type::primitive_type_builder(name, physical) + .with_repetition(Repetition::REQUIRED) + .build() + .unwrap() +} + +fn one(columns: &[&str]) -> Vec { + vec![UniquenessCheck { + columns: columns.iter().map(|s| s.to_string()).collect(), + }] +} + +fn bench(c: &mut Criterion) { + let path = fixture(rows()); + let run = |checks: &[UniquenessCheck]| uniqueness_stats(Path::new(&path), checks, 5).unwrap(); + + let mut group = c.benchmark_group("uniqueness"); + group + .sample_size(10) + .warm_up_time(Duration::from_millis(500)); + group.bench_function("id", |b| b.iter(|| run(&one(&["id"])))); + group.bench_function("code", |b| b.iter(|| run(&one(&["code"])))); + group.bench_function("pk", |b| b.iter(|| run(&one(&["part_a", "part_b"])))); + group.bench_function("all", |b| { + b.iter(|| { + run(&[ + one(&["id"]).remove(0), + one(&["code"]).remove(0), + one(&["part_a", "part_b"]).remove(0), + ]) + }) + }); + group.finish(); +} + +criterion_group!(benches, bench); +criterion_main!(benches); diff --git a/crates/data-dict-parquet/src/lib.rs b/crates/data-dict-parquet/src/lib.rs index fdc9bf9..80061da 100644 --- a/crates/data-dict-parquet/src/lib.rs +++ b/crates/data-dict-parquet/src/lib.rs @@ -2,7 +2,9 @@ mod metadata; mod scan; +mod uniqueness; pub use metadata::{ColumnMeta, ColumnTypeInfo, column_meta, column_type_info, column_types}; pub use parquet::errors::ParquetError; pub use scan::{ColumnNeeds, ColumnStats, column_stats}; +pub use uniqueness::{UniquenessCheck, UniquenessStats, uniqueness_stats}; diff --git a/crates/data-dict-parquet/src/metadata.rs b/crates/data-dict-parquet/src/metadata.rs index 6c544d2..45ecb08 100644 --- a/crates/data-dict-parquet/src/metadata.rs +++ b/crates/data-dict-parquet/src/metadata.rs @@ -21,6 +21,11 @@ pub struct ColumnMeta { /// Total nulls across all row groups, or `None` when any row group omits /// null-count statistics. Required Parquet fields always report `Some(0)`. pub null_count: Option, + /// Number of rows in the file. + pub row_count: usize, + /// Distinct values when a single row group's footer provides the count. + /// Multiple row-group counts cannot prove file-wide uniqueness. + pub distinct_count: Option, } /// Read the inexpensive, footer-only statistics for each top-level column. @@ -47,7 +52,22 @@ pub fn column_meta(path: &Path) -> Result, ParquetEr .map(|count| total + count as usize) }) }; - (field.name().to_string(), ColumnMeta { null_count }) + let distinct_count = match meta.row_groups() { + [row_group] => row_group + .column(idx) + .statistics() + .and_then(|statistics| statistics.distinct_count_opt()) + .map(|count| count as usize), + _ => None, + }; + ( + field.name().to_string(), + ColumnMeta { + null_count, + row_count: meta.file_metadata().num_rows() as usize, + distinct_count, + }, + ) }) .collect()) } diff --git a/crates/data-dict-parquet/src/uniqueness.rs b/crates/data-dict-parquet/src/uniqueness.rs new file mode 100644 index 0000000..729ae11 --- /dev/null +++ b/crates/data-dict-parquet/src/uniqueness.rs @@ -0,0 +1,512 @@ +use std::fs::File; +use std::hash::BuildHasher; +use std::path::Path; + +use hashbrown::{DefaultHashBuilder, HashSet, HashTable}; +use parquet::basic::Type as PhysicalType; +use parquet::column::reader::ColumnReader; +use parquet::data_type::ByteArray; +use parquet::file::reader::{FileReader, SerializedFileReader}; +use rayon::prelude::*; + +use crate::ParquetError; + +/// Rows decoded per `read_records` call. Large enough to amortise per-call +/// overhead, small enough that a batch of every scanned column stays in cache. +const BATCH_ROWS: usize = 8192; + +#[derive(Clone)] +pub struct UniquenessCheck { + pub columns: Vec, +} + +#[derive(Default)] +pub struct UniquenessStats { + pub duplicate_count: usize, + pub duplicate_rows: Vec, +} + +/// Validate uniqueness exactly by hashing each row's key in a streaming, +/// column-projected pass over the file. +/// +/// Memory is proportional to the number of distinct keys: a duplicate is a key +/// already seen. A single integer-like column is hashed as `i64` without +/// allocating; a single byte column is hashed by its value bytes directly +/// (a zero-copy slice into the decoded page); only a composite key pays to +/// build a length-framed key so its columns never collide across splits. Checks +/// are independent and run in parallel, each reading only the columns it needs. +pub fn uniqueness_stats( + path: &Path, + checks: &[UniquenessCheck], + sample_limit: usize, +) -> Result, ParquetError> { + checks + .par_iter() + .map(|check| check_uniqueness(path, check, sample_limit)) + .collect() +} + +fn check_uniqueness( + path: &Path, + check: &UniquenessCheck, + sample_limit: usize, +) -> Result { + let file = + File::open(path).map_err(|e| ParquetError::General(format!("Cannot open file: {e}")))?; + let reader = SerializedFileReader::new(file)?; + let descr = reader.metadata().file_metadata().schema_descr_ptr(); + + let columns = plan_columns(check, &descr)?; + let rows = reader.metadata().file_metadata().num_rows().max(0) as usize; + let mut dedup = Dedup::new(check, &columns, rows); + let mut stat = UniquenessStats::default(); + + let mut row_offset = 0usize; + let mut batches: Vec = Vec::with_capacity(columns.len()); + for group in 0..reader.num_row_groups() { + let row_group = reader.get_row_group(group)?; + let mut readers = columns + .iter() + .map(|column| row_group.get_column_reader(column.leaf)) + .collect::, _>>()?; + loop { + batches.clear(); + let mut rows = 0; + for (reader, column) in readers.iter_mut().zip(&columns) { + let batch = read_batch(reader, column)?; + rows = batch.len(); + batches.push(batch); + } + if rows == 0 { + break; + } + dedup.scan(&batches, row_offset, sample_limit, &mut stat); + row_offset += rows; + } + } + Ok(stat) +} + +/// A column scanned for one or more checks, identified by its leaf index. +struct PlannedColumn { + name: String, + leaf: usize, + physical: PhysicalType, + max_def: i16, +} + +impl PlannedColumn { + /// Whether the column's values are hashed as a single `i64` scalar (as + /// opposed to a variable-length byte string). + fn is_scalar(&self) -> bool { + !matches!( + self.physical, + PhysicalType::BYTE_ARRAY | PhysicalType::FIXED_LEN_BYTE_ARRAY | PhysicalType::INT96 + ) + } +} + +/// Resolve every column named by the check to its leaf position, de-duplicated +/// so a column repeated within a key is read only once. +fn plan_columns( + check: &UniquenessCheck, + descr: &parquet::schema::types::SchemaDescPtr, +) -> Result, ParquetError> { + let mut columns: Vec = Vec::new(); + for name in &check.columns { + if columns.iter().any(|c| &c.name == name) { + continue; + } + let leaf = (0..descr.num_columns()) + .find(|&i| descr.column(i).name() == name) + .ok_or_else(|| ParquetError::General(format!("Column not found: {name}")))?; + let column = descr.column(leaf); + columns.push(PlannedColumn { + name: name.clone(), + leaf, + physical: column.physical_type(), + max_def: column.max_def_level(), + }); + } + Ok(columns) +} + +/// One column's values for a batch of rows, decoded row-aligned (nulls included) +/// into either scalar `i64`s or byte strings. Byte values keep the [`ByteArray`] +/// handles the reader produced, so [`ColumnBatch::bytes`] is a zero-copy slice +/// into Parquet's decoded page (or dictionary) buffer rather than a re-copy. +/// +/// `null` is empty when the column is `REQUIRED` (it can't contain nulls), which +/// skips a per-batch mask allocation on the common path. +enum ColumnBatch { + Scalar { + values: Vec, + null: Vec, + }, + Bytes { + values: Vec, + null: Vec, + }, +} + +impl ColumnBatch { + fn len(&self) -> usize { + match self { + ColumnBatch::Scalar { values, .. } => values.len(), + ColumnBatch::Bytes { values, .. } => values.len(), + } + } + + fn is_null(&self, row: usize) -> bool { + match self { + ColumnBatch::Scalar { null, .. } | ColumnBatch::Bytes { null, .. } => { + !null.is_empty() && null[row] + } + } + } + + fn scalar(&self, row: usize) -> i64 { + match self { + ColumnBatch::Scalar { values, .. } => values[row], + ColumnBatch::Bytes { .. } => unreachable!("scalar read of a byte column"), + } + } + + fn bytes(&self, row: usize) -> &[u8] { + match self { + ColumnBatch::Bytes { values, .. } => values[row].data(), + ColumnBatch::Scalar { .. } => unreachable!("byte read of a scalar column"), + } + } +} + +/// Read up to [`BATCH_ROWS`] rows from one column, expanding nulls back into +/// their row positions so every column in a batch shares the same row indices. +fn read_batch( + reader: &mut ColumnReader, + column: &PlannedColumn, +) -> Result { + let max_def = column.max_def; + macro_rules! scalar { + ($variant:path, $map:expr) => {{ + let $variant(ref mut typed) = *reader else { + return Err(physical_mismatch(column)); + }; + let mut def = Vec::new(); + let mut raw = Vec::new(); + let (records, _, _) = typed.read_records(BATCH_ROWS, Some(&mut def), None, &mut raw)?; + let mut values = vec![0i64; records]; + let mut null = Vec::new(); + if max_def == 0 { + for (out, value) in values.iter_mut().zip(&raw) { + *out = $map(value); + } + } else { + null = vec![false; records]; + let mut cursor = 0; + for row in 0..records { + if def[row] == max_def { + values[row] = $map(&raw[cursor]); + cursor += 1; + } else { + null[row] = true; + } + } + } + Ok(ColumnBatch::Scalar { values, null }) + }}; + } + // The rare `FixedLenByteArray`/`Int96` cases must convert each value into an + // owned `ByteArray`; `BYTE_ARRAY` (below) skips even that, handing the decoded + // vector straight through. + macro_rules! bytes_converted { + ($variant:path, $conv:expr) => {{ + let $variant(ref mut typed) = *reader else { + return Err(physical_mismatch(column)); + }; + let mut def = Vec::new(); + let mut raw = Vec::new(); + let (records, _, _) = typed.read_records(BATCH_ROWS, Some(&mut def), None, &mut raw)?; + let values = raw.into_iter().map($conv).collect(); + Ok(expand_bytes(values, &def, max_def, records)) + }}; + } + + match column.physical { + PhysicalType::BOOLEAN => scalar!(ColumnReader::BoolColumnReader, |v: &bool| *v as i64), + PhysicalType::INT32 => scalar!(ColumnReader::Int32ColumnReader, |v: &i32| *v as i64), + PhysicalType::INT64 => scalar!(ColumnReader::Int64ColumnReader, |v: &i64| *v), + PhysicalType::FLOAT => scalar!(ColumnReader::FloatColumnReader, float_bits), + PhysicalType::DOUBLE => scalar!(ColumnReader::DoubleColumnReader, double_bits), + PhysicalType::BYTE_ARRAY => { + let ColumnReader::ByteArrayColumnReader(ref mut typed) = *reader else { + return Err(physical_mismatch(column)); + }; + let mut def = Vec::new(); + let mut raw = Vec::new(); + let (records, _, _) = typed.read_records(BATCH_ROWS, Some(&mut def), None, &mut raw)?; + Ok(expand_bytes(raw, &def, max_def, records)) + } + PhysicalType::FIXED_LEN_BYTE_ARRAY => { + bytes_converted!(ColumnReader::FixedLenByteArrayColumnReader, fixed_len_owned) + } + PhysicalType::INT96 => bytes_converted!(ColumnReader::Int96ColumnReader, int96_owned), + } +} + +/// Place non-null byte values back at their row positions, moving (never cloning) +/// each value. With no nulls the decoded vector is already row-aligned, so it is +/// used as-is. +fn expand_bytes(nonnull: Vec, def: &[i16], max_def: i16, records: usize) -> ColumnBatch { + if max_def == 0 { + return ColumnBatch::Bytes { + values: nonnull, + null: Vec::new(), + }; + } + let mut values = Vec::with_capacity(records); + let mut null = vec![false; records]; + let mut nonnull = nonnull.into_iter(); + for row in 0..records { + if def[row] == max_def { + values.push(nonnull.next().expect("a value for each defined row")); + } else { + null[row] = true; + values.push(ByteArray::default()); + } + } + ColumnBatch::Bytes { values, null } +} + +fn float_bits(value: &f32) -> i64 { + (if *value == 0.0 { 0 } else { value.to_bits() }) as i64 +} + +fn double_bits(value: &f64) -> i64 { + (if *value == 0.0 { 0 } else { value.to_bits() }) as i64 +} + +fn fixed_len_owned(value: parquet::data_type::FixedLenByteArray) -> ByteArray { + ByteArray::from(value) +} + +fn int96_owned(value: parquet::data_type::Int96) -> ByteArray { + // Identify the value by the raw bytes of its three words. Byte order only + // has to be consistent within a run, which it is. + let bytes: Vec = value.data().iter().flat_map(|w| w.to_le_bytes()).collect(); + ByteArray::from(bytes) +} + +fn physical_mismatch(column: &PlannedColumn) -> ParquetError { + ParquetError::General(format!( + "Column reader type does not match physical type {:?}", + column.physical + )) +} + +/// Per-check duplicate detector, with a fast path for each single-column shape: +/// a scalar column hashes bare `i64`s, a single byte column hashes the value +/// bytes directly, and only a composite key pays for length-framing (so its +/// columns never collide across different splits). +enum Dedup { + Scalar { + column: usize, + seen: HashSet, + null_seen: bool, + }, + SingleBytes { + column: usize, + seen: ByteKeys, + null_seen: bool, + }, + Bytes { + columns: Vec, + seen: ByteKeys, + key: Vec, + }, +} + +impl Dedup { + fn new(check: &UniquenessCheck, columns: &[PlannedColumn], rows: usize) -> Self { + let positions = check + .columns + .iter() + .map(|name| { + columns + .iter() + .position(|column| &column.name == name) + .expect("planned column is missing") + }) + .collect::>(); + // A uniqueness check expects mostly-distinct keys, so size for one entry + // per row up front. This both skips the incremental rehashes and, more + // importantly, avoids the transient 2x spike of growing a near-full + // table by doubling. + if let [only] = positions.as_slice() { + if columns[*only].is_scalar() { + return Dedup::Scalar { + column: *only, + seen: HashSet::with_capacity(rows), + null_seen: false, + }; + } + return Dedup::SingleBytes { + column: *only, + seen: ByteKeys::with_capacity(rows), + null_seen: false, + }; + } + Dedup::Bytes { + columns: positions, + seen: ByteKeys::with_capacity(rows), + key: Vec::new(), + } + } + + fn scan( + &mut self, + batches: &[ColumnBatch], + row_offset: usize, + sample_limit: usize, + stat: &mut UniquenessStats, + ) { + match self { + Dedup::Scalar { + column, + seen, + null_seen, + } => { + let batch = &batches[*column]; + for row in 0..batch.len() { + let duplicate = if batch.is_null(row) { + std::mem::replace(null_seen, true) + } else { + !seen.insert(batch.scalar(row)) + }; + if duplicate { + record(stat, row_offset + row + 1, sample_limit); + } + } + } + Dedup::SingleBytes { + column, + seen, + null_seen, + } => { + let batch = &batches[*column]; + for row in 0..batch.len() { + let duplicate = if batch.is_null(row) { + std::mem::replace(null_seen, true) + } else { + !seen.insert(batch.bytes(row)) + }; + if duplicate { + record(stat, row_offset + row + 1, sample_limit); + } + } + } + Dedup::Bytes { columns, seen, key } => { + let rows = batches[columns[0]].len(); + for row in 0..rows { + key.clear(); + for &column in columns.iter() { + let batch = &batches[column]; + if batch.is_null(row) { + key.push(0); + } else if let ColumnBatch::Scalar { .. } = batch { + key.push(1); + key.extend_from_slice(&batch.scalar(row).to_le_bytes()); + } else { + let value = batch.bytes(row); + key.push(2); + key.extend_from_slice(&(value.len() as u32).to_le_bytes()); + key.extend_from_slice(value); + } + } + if !seen.insert(key) { + record(stat, row_offset + row + 1, sample_limit); + } + } + } + } + } +} + +/// A set of byte-string keys packed into one arena so distinct keys cost an +/// amortised append, not an allocation each. Entries reference the arena by +/// `(offset, length)`, and a single hash probe both tests membership and, when +/// absent, positions the insertion. +#[derive(Default)] +struct ByteKeys { + arena: Arena, + table: HashTable, + hasher: DefaultHashBuilder, +} + +/// A key's location in the arena: `(chunk, offset within chunk, length)`. +type KeyRef = (u32, u32, u32); + +impl ByteKeys { + fn with_capacity(rows: usize) -> Self { + ByteKeys { + arena: Arena::default(), + table: HashTable::with_capacity(rows), + hasher: DefaultHashBuilder::default(), + } + } + + /// Insert `key`, returning `true` if it was new (i.e. not a duplicate). + fn insert(&mut self, key: &[u8]) -> bool { + let Self { + arena, + table, + hasher, + } = self; + let hash = hasher.hash_one(key); + if table.find(hash, |&r| arena.get(r) == key).is_some() { + return false; + } + let entry = arena.push(key); + table.insert_unique(hash, entry, |&r| hasher.hash_one(arena.get(r))); + true + } +} + +/// Append-only byte store backed by fixed-size chunks. Existing chunks are never +/// reallocated, so growth is incremental — there is no transient doubling spike +/// as with one growing `Vec`, and no need to estimate the total size up front. +#[derive(Default)] +struct Arena { + chunks: Vec>, +} + +impl Arena { + const CHUNK: usize = 4 * 1024 * 1024; + + fn push(&mut self, key: &[u8]) -> KeyRef { + let fits = self + .chunks + .last() + .is_some_and(|chunk| chunk.capacity() - chunk.len() >= key.len()); + if !fits { + self.chunks + .push(Vec::with_capacity(key.len().max(Self::CHUNK))); + } + let chunk = self.chunks.len() as u32 - 1; + let buffer = self.chunks.last_mut().unwrap(); + let offset = buffer.len() as u32; + buffer.extend_from_slice(key); + (chunk, offset, key.len() as u32) + } + + fn get(&self, (chunk, offset, len): KeyRef) -> &[u8] { + &self.chunks[chunk as usize][offset as usize..offset as usize + len as usize] + } +} + +fn record(stat: &mut UniquenessStats, row: usize, sample_limit: usize) { + stat.duplicate_count += 1; + if stat.duplicate_rows.len() < sample_limit { + stat.duplicate_rows.push(row); + } +} diff --git a/crates/data-dict/src/problem.rs b/crates/data-dict/src/problem.rs index 268ea25..7be335c 100644 --- a/crates/data-dict/src/problem.rs +++ b/crates/data-dict/src/problem.rs @@ -120,6 +120,12 @@ pub enum ProblemKind { /// `D01` — a `required` (or `primary_key`) column contains nulls. `rows` /// lists the first few offending row numbers (1-based); `count` is the total. NullsInRequired { count: usize, rows: Vec }, + /// `D02` — a unique column or composite primary key contains duplicates. + DuplicateValues { + columns: Vec, + count: usize, + rows: Vec, + }, } impl ProblemKind { @@ -134,6 +140,7 @@ impl ProblemKind { ProblemKind::MissingSource => "M04", ProblemKind::UnreadableSource => "M05", ProblemKind::NullsInRequired { .. } => "D01", + ProblemKind::DuplicateValues { .. } => "D02", _ => return None, }) } @@ -148,7 +155,9 @@ impl ProblemKind { | ProblemKind::ExtraInData { .. } | ProblemKind::MissingSource | ProblemKind::UnreadableSource => Level::Meta, - ProblemKind::NullsInRequired { .. } => Level::Data, + ProblemKind::NullsInRequired { .. } | ProblemKind::DuplicateValues { .. } => { + Level::Data + } _ => return None, }) } @@ -518,6 +527,15 @@ mod tests { .level(), Some(Level::Data) ); + assert_eq!( + ProblemKind::DuplicateValues { + columns: vec!["id".into()], + count: 1, + rows: vec![2], + } + .code(), + Some("D02") + ); assert_eq!(ProblemKind::Io.code(), None); assert_eq!(ProblemKind::Io.level(), None); } diff --git a/crates/data-dict/src/validate_data.rs b/crates/data-dict/src/validate_data.rs index 8a1d873..343ee87 100644 --- a/crates/data-dict/src/validate_data.rs +++ b/crates/data-dict/src/validate_data.rs @@ -6,7 +6,7 @@ use std::collections::HashMap; use std::path::Path; -use data_dict_parquet::{ColumnMeta, ColumnNeeds, ColumnStats}; +use data_dict_parquet::{ColumnMeta, ColumnNeeds, ColumnStats, UniquenessCheck, UniquenessStats}; use crate::model::{Column, Constraint, Table}; use crate::problem::{Problem, ProblemKind, ProblemSet, Severity}; @@ -90,6 +90,50 @@ fn value_issues( } } + let mut uniqueness = Vec::new(); + for col in table + .columns + .iter() + .filter(|col| col.has(Constraint::Unique) && present(&col.name.value)) + { + let Some(meta) = metadata.get(&col.name.value) else { + continue; + }; + match crate::validate_meta::validate_d02_unique_column(table, col, meta) { + CheckResult::Pass => {} + CheckResult::Inconclusive => uniqueness.push(UniquenessTarget::Column(col)), + CheckResult::Fail(problem) => out.push(*problem), + } + } + let primary_key = table + .columns + .iter() + .filter(|col| col.has(Constraint::PrimaryKey)) + .collect::>(); + if !primary_key.is_empty() && primary_key.iter().all(|col| present(&col.name.value)) { + uniqueness.push(UniquenessTarget::PrimaryKey(primary_key)); + } + if !uniqueness.is_empty() { + let checks = uniqueness + .iter() + .map(UniquenessTarget::check) + .collect::>(); + let results = data_dict_parquet::uniqueness_stats(parquet_path, &checks, SAMPLE_LIMIT)?; + for (target, stats) in uniqueness.iter().zip(&results) { + if stats.duplicate_count == 0 { + continue; + } + match target { + UniquenessTarget::Column(col) => { + out.push(duplicates_in_unique_column(table, col, stats)); + } + UniquenessTarget::PrimaryKey(columns) => { + out.push(duplicates_in_primary_key(table, columns, stats)); + } + } + } + } + Ok(()) } @@ -173,3 +217,85 @@ fn nulls_in_required_data(table: &Table, col: &Column, count: usize, rows: Vec Problem { + let count = stats.duplicate_count; + let detail = crate::problem::format_rows(&stats.duplicate_rows, count); + let plural = if count == 1 { "" } else { "s" }; + let constraint_span = col + .constraints + .iter() + .find(|constraint| constraint.value == Constraint::Unique) + .map_or_else( + || col.name.span.clone(), + |constraint| constraint.span.clone(), + ); + Problem { + code: Some("D02"), + severity: Severity::Error, + message: format!("has {count} repeated occurrence{plural} ({detail})"), + column: None, + expected: Some("A unique column must not contain duplicate values.".into()), + span: Some(constraint_span), + context: vec![table.name.span.clone(), col.name.span.clone()], + kind: ProblemKind::DuplicateValues { + columns: vec![col.name.value.clone()], + count, + rows: stats.duplicate_rows.clone(), + }, + } +} + +fn duplicates_in_primary_key( + table: &Table, + columns: &[&Column], + stats: &UniquenessStats, +) -> Problem { + let count = stats.duplicate_count; + let detail = crate::problem::format_rows(&stats.duplicate_rows, count); + let plural = if count == 1 { "" } else { "s" }; + let last = columns + .last() + .expect("a primary key has at least one column"); + let constraint_span = last + .constraints + .iter() + .find(|constraint| constraint.value == Constraint::PrimaryKey) + .map_or_else( + || last.name.span.clone(), + |constraint| constraint.span.clone(), + ); + Problem { + code: Some("D02"), + severity: Severity::Error, + message: format!("has {count} repeated occurrence{plural} ({detail})"), + column: None, + expected: Some("The primary key must uniquely identify every row.".into()), + span: Some(constraint_span), + context: std::iter::once(table.name.span.clone()) + .chain(columns.iter().map(|col| col.name.span.clone())) + .collect(), + kind: ProblemKind::DuplicateValues { + columns: columns.iter().map(|col| col.name.value.clone()).collect(), + count, + rows: stats.duplicate_rows.clone(), + }, + } +} + +enum UniquenessTarget<'a> { + Column(&'a Column), + PrimaryKey(Vec<&'a Column>), +} + +impl UniquenessTarget<'_> { + fn check(&self) -> UniquenessCheck { + let columns = match self { + UniquenessTarget::Column(col) => vec![col.name.value.clone()], + UniquenessTarget::PrimaryKey(columns) => { + columns.iter().map(|col| col.name.value.clone()).collect() + } + }; + UniquenessCheck { columns } + } +} diff --git a/crates/data-dict/src/validate_meta.rs b/crates/data-dict/src/validate_meta.rs index eb81397..24e8801 100644 --- a/crates/data-dict/src/validate_meta.rs +++ b/crates/data-dict/src/validate_meta.rs @@ -58,6 +58,36 @@ pub(crate) fn validate_d01_required_not_null( } } +/// Attempt the individual-column form of D02 from footer statistics. +pub(crate) fn validate_d02_unique_column( + table: &Table, + col: &Column, + meta: &ColumnMeta, +) -> CheckResult { + if !col.has(Constraint::Unique) { + return CheckResult::Pass; + } + let (Some(distinct), Some(nulls)) = (meta.distinct_count, meta.null_count) else { + return CheckResult::Inconclusive; + }; + // Parquet writers differ in how they populate distinct counts around nulls; + // scan nullable data rather than drawing an unsafe footer-only conclusion. + if nulls > 0 { + return CheckResult::Inconclusive; + } + if distinct == meta.row_count { + CheckResult::Pass + } else if distinct < meta.row_count { + CheckResult::Fail(Box::new(duplicates_meta( + table, + col, + meta.row_count - distinct, + ))) + } else { + CheckResult::Inconclusive + } +} + fn nulls_in_required_meta(table: &Table, col: &Column, count: usize) -> Problem { let plural = if count == 1 { "" } else { "s" }; let constraint_span = col @@ -88,6 +118,32 @@ fn nulls_in_required_meta(table: &Table, col: &Column, count: usize) -> Problem } } +fn duplicates_meta(table: &Table, col: &Column, count: usize) -> Problem { + let plural = if count == 1 { "" } else { "s" }; + let constraint_span = col + .constraints + .iter() + .find(|constraint| constraint.value == Constraint::Unique) + .map_or_else( + || col.name.span.clone(), + |constraint| constraint.span.clone(), + ); + Problem { + code: Some("D02"), + severity: Severity::Error, + message: format!("has {count} repeated occurrence{plural}"), + column: None, + expected: Some("A unique column must not contain duplicate values.".into()), + span: Some(constraint_span), + context: vec![table.name.span.clone(), col.name.span.clone()], + kind: ProblemKind::DuplicateValues { + columns: vec![col.name.value.clone()], + count, + rows: Vec::new(), + }, + } +} + fn validate_m01_column_types(table: &Table, actual: &[(String, String)], out: &mut ProblemSet) { for col in &table.columns { // Only described columns present in the data are type-checked; an absent diff --git a/crates/data-dict/tests/validate_data.rs b/crates/data-dict/tests/validate_data.rs index cde468e..561546f 100644 --- a/crates/data-dict/tests/validate_data.rs +++ b/crates/data-dict/tests/validate_data.rs @@ -14,7 +14,7 @@ use std::sync::Arc; use data_dict::{Problem, ProblemKind, ProblemSet, Status, validate_data, validate_meta}; use indoc::{formatdoc, indoc}; -use parquet::data_type::DoubleType; +use parquet::data_type::{ByteArray, ByteArrayType, DoubleType}; use parquet::file::properties::{EnabledStatistics, WriterProperties}; use parquet::file::writer::{SerializedColumnWriter, SerializedFileWriter}; use parquet::schema::parser::parse_message_type; @@ -97,6 +97,50 @@ fn write_double_with_null(col: &mut SerializedColumnWriter) { .unwrap(); } +fn build_composite_key(first: &[f64], second: &[f64]) -> PathBuf { + let dir = temp_dir(); + let parquet = dir.join("data.parquet"); + let schema = Arc::new( + parse_message_type("message schema { REQUIRED DOUBLE a; REQUIRED DOUBLE b; }").unwrap(), + ); + let file = File::create(&parquet).unwrap(); + let mut writer = + SerializedFileWriter::new(file, schema, Arc::new(WriterProperties::builder().build())) + .unwrap(); + let mut row_group = writer.next_row_group().unwrap(); + let mut a = row_group.next_column().unwrap().unwrap(); + a.typed::() + .write_batch(first, None, None) + .unwrap(); + a.close().unwrap(); + let mut b = row_group.next_column().unwrap().unwrap(); + b.typed::() + .write_batch(second, None, None) + .unwrap(); + b.close().unwrap(); + row_group.close().unwrap(); + writer.close().unwrap(); + + write_dict( + &dir, + indoc! {" + tables: + t: + source: + parquet: data.parquet + columns: + - name: a + type: number(id) + constraints: [primary_key] + examples: [1, 2] + - name: b + type: number(id) + constraints: [primary_key] + examples: [1, 2] + "}, + ) +} + /// The defining difference between the two levels: a `required` column with /// nulls is a *value* problem, so it is invisible to `validate-meta` (which /// reads only names and types) but caught by `validate-data` (which scans). @@ -252,3 +296,111 @@ fn primary_key_implies_required_for_nulls() { result.items ); } + +#[test] +fn duplicate_values_in_unique_column_reported() { + let result = check_column( + "REQUIRED DOUBLE id", + |col| { + col.typed::() + .write_batch(&[1.0, 1.0, 2.0], None, None) + .unwrap(); + }, + indoc! {" + - name: id + type: number(id) + constraints: [unique] + examples: [1, 2] + "}, + ); + + assert!(matches!( + result.items.as_slice(), + [Problem { + code: Some("D02"), + kind: ProblemKind::DuplicateValues { columns, count: 1, rows }, + .. + }] if columns == &["id"] && rows == &[2] + )); +} + +/// Write a single required string column whose values are split across the +/// given row groups, so the scan accumulates row offsets across group +/// boundaries and exercises the variable-length byte-key path. +fn build_string_groups(groups: &[&[&str]]) -> PathBuf { + let dir = temp_dir(); + let parquet = dir.join("data.parquet"); + let schema = Arc::new( + parse_message_type("message schema { REQUIRED BYTE_ARRAY code (UTF8); }").unwrap(), + ); + let file = File::create(&parquet).unwrap(); + let mut writer = + SerializedFileWriter::new(file, schema, Arc::new(WriterProperties::builder().build())) + .unwrap(); + for group in groups { + let values = group + .iter() + .map(|s| ByteArray::from(*s)) + .collect::>(); + let mut row_group = writer.next_row_group().unwrap(); + let mut col = row_group.next_column().unwrap().unwrap(); + col.typed::() + .write_batch(&values, None, None) + .unwrap(); + col.close().unwrap(); + row_group.close().unwrap(); + } + writer.close().unwrap(); + + write_dict( + &dir, + indoc! {" + tables: + t: + source: + parquet: data.parquet + columns: + - name: code + type: string + constraints: [unique] + examples: [a, b] + "}, + ) +} + +#[test] +fn duplicate_string_values_across_row_groups_reported() { + // No duplicates across two groups. + let unique = build_string_groups(&[&["a", "b"], &["c", "d"]]); + assert_eq!(validate_data(&unique, None).status(), Status::Ok); + + // "a" recurs in the second group, so the duplicate sits at row 4 — proving + // row numbers carry across the row-group boundary. + let duplicate = build_string_groups(&[&["a", "b"], &["c", "a"]]); + let result = validate_data(&duplicate, None); + assert!(matches!( + result.items.as_slice(), + [Problem { + code: Some("D02"), + kind: ProblemKind::DuplicateValues { columns, count: 1, rows }, + .. + }] if columns == &["code"] && rows == &[4] + )); +} + +#[test] +fn composite_primary_key_is_checked_collectively() { + let unique = build_composite_key(&[1.0, 1.0, 2.0], &[1.0, 2.0, 1.0]); + assert_eq!(validate_data(&unique, None).status(), Status::Ok); + + let duplicate = build_composite_key(&[1.0, 1.0, 2.0], &[1.0, 1.0, 2.0]); + let result = validate_data(&duplicate, None); + assert!(matches!( + result.items.as_slice(), + [Problem { + code: Some("D02"), + kind: ProblemKind::DuplicateValues { columns, count: 1, rows }, + .. + }] if columns == &["a", "b"] && rows == &[2] + )); +} diff --git a/site/validation.md b/site/validation.md index 8630086..3afdb2d 100644 --- a/site/validation.md +++ b/site/validation.md @@ -65,3 +65,4 @@ When validating the data's metadata against the dictionary, each column mismatch When validating the data's values against the dictionary, each column mismatch is one of: * **Nulls in a required column** (D01, error): a `required` or `primary_key` column contains nulls. +* **Duplicate values** (D02, error): a `unique` column contains duplicate values, or the combination of all `primary_key` columns does not uniquely identify every row.