diff --git a/docs/contributor/STATS_QUICK_REFERENCE.md b/docs/contributor/STATS_QUICK_REFERENCE.md index 5c37b4376..de4e4061b 100644 --- a/docs/contributor/STATS_QUICK_REFERENCE.md +++ b/docs/contributor/STATS_QUICK_REFERENCE.md @@ -200,11 +200,11 @@ cat file.stats.csv.data.jsonl | jq . | head | File | Purpose | |------|---------| -| `src/cmd/stats.rs` | Main implementation (~6,320 lines) | +| `src/cmd/stats.rs` | Main implementation (~6,910 lines) | | `src/config.rs` | CSV reader configuration | | `src/select.rs` | Column selection logic | | `src/util.rs` | Utility functions | -| `tests/test_stats.rs` | Comprehensive test suite (~8,338 lines) | +| `tests/test_stats.rs` | Comprehensive test suite (~9,319 lines) | | `Cargo.toml` | Dependencies (see `stats` and `csv` crates) | --- diff --git a/src/cmd/stats.rs b/src/cmd/stats.rs index 623f6ab57..9cad1f65e 100644 --- a/src/cmd/stats.rs +++ b/src/cmd/stats.rs @@ -1698,26 +1698,27 @@ pub fn run(argv: &[&str]) -> CliResult<()> { return fail_incorrectusage_clierror!("{format_error}"); } - // NEVER autoindex a special-format input (.gz/.zip/.parquet/.jsonl/...) in `stats`. + // INVARIANT (issue #4462): `rconfig` below is the one and only resolved Config for this + // run's input, and every stats path SHARES it - `sequential_stats` and `parallel_stats` + // both take it by reference, and each parallel worker gets a CLONE. // - // Such an input is read through a CONVERTED temp file, and every Config converts to its OWN - // temp path. `parallel_stats`' workers each build a FRESH Config (`args.rconfig()`), so they - // resolve a DIFFERENT temp file, find no index beside it, and panic in the `.expect()` on - // `indexed()`. Every chunk dies; because a panicking pool worker only unwinds its own thread, - // the run still exited 0 and wrote a stats file with headers and ZERO data rows. Reproducible - // on master with `qsv stats -E --cache-threshold -105 data.csv.gz`. + // This matters for special-format inputs (.gz/.zip/.parquet/.jsonl/...), which are read + // through a CONVERTED temp file: each Config resolves its own temp, but clones share the + // `Arc` holding that resolution, so all of them see the SAME temp - and therefore + // the same sibling autoindex. Rebuilding `args.rconfig()` anywhere downstream breaks this: + // it resolves a DIFFERENT temp with no index beside it, and the worker panics in the + // `.expect()` on `indexed()`. Because a panicking pool worker only unwinds its own thread, + // that failed silently - the run exited 0 with a headers-only, ZERO-row stats file + // (`qsv stats -E --cache-threshold -105 data.csv.gz`, issue #4446). // - // This lives HERE rather than in `Config::index_files` on purpose: commands that hand their - // workers the already-RESOLVED Config - `frequency` does exactly that, deliberately - keep a - // consistent temp path across threads, so their special-format autoindexed parallel path is - // safe and must not be disabled. `stats` is the caller that reconstructs Configs, so `stats` - // is where the skip belongs. + // #4445 papered over this by refusing to autoindex special-format inputs at all, which cost + // them the parallel path entirely. Sharing the Config is the real fix, and it restores the + // memcheck index fallback below - the one that fires precisely when the input is too large + // to process sequentially. // - // Zeroed here so BOTH routes are covered: QSV_AUTOINDEX_SIZE (applied in Config::new) and - // the negative --cache-threshold below. - if rconfig.is_special_format() { - rconfig.autoindex_size = 0; - } + // Grep guard: there must be NO `self.rconfig()` / `args.rconfig()` call downstream of this + // point on a compute path. `Config::resolve_converted` logs one line per conversion; more + // than one per input means this invariant was broken. // infer delimiter when we're getting input from stdin // as the stats engine needs to know the delimiter or it will panic @@ -1784,10 +1785,8 @@ pub fn run(argv: &[&str]) -> CliResult<()> { }; // Thread the inferred delimiter through args rather than mutating the - // process environment. rconfig() applies flag_delimiter to every Config - // it builds, so this reaches the stats engine exactly the way the - // QSV_DEFAULT_DELIMITER env var did - without an unsafe global mutation - // that is UB if any other thread reads the environment concurrently. + // process environment - an unsafe global mutation that is UB if any other + // thread reads the environment concurrently. // An explicit --delimiter always wins. if args.flag_delimiter.is_none() { args.flag_delimiter = Some(Delimiter(inferred)); @@ -1796,6 +1795,13 @@ pub fn run(argv: &[&str]) -> CliResult<()> { args.arg_input = Some(tempfile_path.to_string_lossy().to_string()); rconfig.path = Some(tempfile_path); + // `rconfig` was built from `args` BEFORE the delimiter was inferred, and it is now + // the single Config every compute path reads through (see the invariant above). + // Re-apply the delimiter here or a tab/semicolon-delimited stdin input is parsed as + // comma-delimited - silently wrong stats, not an error. This used to be carried by + // the compute paths rebuilding `args.rconfig()` themselves; they no longer do. + // `Config::delimiter(None)` is a no-op, so an unset --delimiter changes nothing. + rconfig = rconfig.delimiter(args.flag_delimiter); } else { // check if the input file exists if let Some(path) = rconfig.path.clone() @@ -2026,7 +2032,7 @@ pub fn run(argv: &[&str]) -> CliResult<()> { // check if flag_cache_threshold is a negative number, // if so, set the autoindex_size to absolute of the number - if args.flag_cache_threshold.is_negative() && !rconfig.is_special_format() { + if args.flag_cache_threshold.is_negative() { rconfig.autoindex_size = args.flag_cache_threshold.unsigned_abs() as u64; autoindex_set = true; } @@ -2073,19 +2079,14 @@ pub fn run(argv: &[&str]) -> CliResult<()> { // per-column memory regardless of sequential vs. parallel // Only propagate the original OOM error if NEITHER fallback engages. let mut index_succeeded = false; - // `!rconfig.is_special_format()` for the same reason the autoindex is - // skipped above, and it matters MORE here: this fallback fires precisely - // when the file is too large for sequential processing. It creates an - // index beside the RESOLVED TEMP file and then selects `parallel_stats`, - // whose workers rebuild a fresh Config, resolve a DIFFERENT temp, find no - // index, and panic - turning an out-of-memory condition into a - // headers-only stats file at exit 0. The DataSketches fallback below - // still engages for these inputs, so a large compressed file keeps a - // memory mitigation; it just stays sequential. - if indexed_result.is_none() - && !rconfig.is_stdin() - && !rconfig.is_special_format() - { + // Special-format inputs take this path too (issue #4462). The index is + // built beside the RESOLVED TEMP (`mem_path`), and `parallel_stats`' + // workers clone this very `rconfig`, so they resolve to that same temp + // and find the index. #4445 excluded them here - which was backwards, + // since this fallback fires precisely when the input is too large to + // process sequentially, leaving a large compressed file with no index + // escape hatch at all. + if indexed_result.is_none() && !rconfig.is_stdin() { log::info!( "File too large for sequential processing. Auto-creating index to \ enable parallel processing..." @@ -2152,7 +2153,7 @@ pub fn run(argv: &[&str]) -> CliResult<()> { // without an index, we need to count the number of records in the file // safety: we know util::count_rows() will not return an Err let capacity_hint = util::count_rows(&rconfig).unwrap(); - args.sequential_stats(&resolved_whitelist, capacity_hint) + args.sequential_stats(&resolved_whitelist, capacity_hint, &rconfig) }, Some(idx) => { // with an index, we get the rowcount instantaneously from the index @@ -2160,12 +2161,12 @@ pub fn run(argv: &[&str]) -> CliResult<()> { match args.flag_jobs { Some(num_jobs) => { if num_jobs == 1 { - args.sequential_stats(&resolved_whitelist, idx_count) + args.sequential_stats(&resolved_whitelist, idx_count, &rconfig) } else { - args.parallel_stats(&resolved_whitelist, idx_count) + args.parallel_stats(&resolved_whitelist, idx_count, &rconfig) } }, - _ => args.parallel_stats(&resolved_whitelist, idx_count), + _ => args.parallel_stats(&resolved_whitelist, idx_count, &rconfig), } }, }?; @@ -2289,7 +2290,9 @@ pub fn run(argv: &[&str]) -> CliResult<()> { // NOTE: the branch that used to stand here, keyed on rconfig.is_stdin(), was // unreachable - rconfig.path is repointed at the spill temp file far above this // point, so is_stdin() is always false by the time we get here. - if let Some(path) = rconfig.path + // `path` is cloned rather than moved out: `rconfig` is still needed inside this block to + // resolve the autoindex's real (possibly converted-temp) location. + if let Some(path) = rconfig.path.clone() && !input_was_stdin { // if we read from a file, copy the temp stats file to ".stats.csv" or @@ -2327,7 +2330,30 @@ pub fn run(argv: &[&str]) -> CliResult<()> { // looked for at `data.csv.idx` and survived the cleanup it asked for. The // removal only log::warn!s on failure, so the leak was silent. It happened to // work for `.csv` inputs by coincidence. - let index_file = util::idx_path(&path); + // + // The index lives beside the path that was actually INDEXED, which for a + // special-format input is the converted temp, not `path` (kept for stats-cache + // naming). Before #4462 these inputs never autoindexed, so `path` was always + // right; now they do, and using `path` would look for `data.csv.gz.idx`, + // never find it, and log a spurious "Could not remove index file" warning on + // every such run. + // + // `resolved_path()` is a cached read here, never a conversion. Two facts, + // in this order: `autoindex_set` is only ever set on the compute path (inside + // `if compute_stats`), and this cleanup block runs far BELOW the + // `rconfig.indexed()` call on that path - which has already populated the + // OnceLock by the time we get here. (The flag itself is set just BEFORE + // `indexed()`, not after; it is the cleanup block's position that guarantees + // the resolution, not the assignment's.) A cache-hit run leaves + // `autoindex_set` false and never reaches this, so the run whose whole purpose + // is to skip work does not decompress anything. + let index_file = util::idx_path( + &rconfig + .resolved_path() + .ok() + .flatten() + .unwrap_or_else(|| path.clone()), + ); log::debug!("deleting index file: {}", index_file.display()); if std::fs::remove_file(index_file.clone()).is_err() { // fails silently if it can't remove the index file @@ -2644,17 +2670,25 @@ impl Args { /// * CSV parsing errors are propagated as `CliError` /// * Date inference initialization errors are handled /// * File I/O errors are wrapped in appropriate error types + /// + /// # `rconfig` + /// + /// MUST be the caller's already-resolved Config, never a fresh `self.rconfig()`. A + /// special-format input converts to a temp file on first read and each Config resolves its + /// OWN temp, so rebuilding one here would convert the input a second time. See the + /// invariant comment in `run()`. fn sequential_stats( &self, whitelist: &str, capacity_hint: u64, + rconfig: &Config, ) -> CliResult<(csv::ByteRecord, Vec, u64)> { - let mut rdr = self.rconfig().reader()?; + let mut rdr = rconfig.reader()?; let full_headers = rdr.byte_headers()?.clone(); // Find weight column index and exclude it from selection let (weight_col_idx, sel, headers) = - self.process_headers_with_weight_exclusion(&full_headers)?; + self.process_headers_with_weight_exclusion(&full_headers, rconfig)?; init_date_inference(self.flag_infer_dates, &headers, whitelist)?; @@ -2737,11 +2771,12 @@ impl Args { &self, whitelist: &str, idx_count: u64, + rconfig: &Config, ) -> CliResult<(csv::ByteRecord, Vec, u64)> { // N.B. This method doesn't handle the case when the number of records // is zero correctly. So we use `sequential_stats` instead. if idx_count == 0 { - return self.sequential_stats(whitelist, 0); + return self.sequential_stats(whitelist, 0, rconfig); } // Retain freed jemalloc pages for the duration of this parallel run when it @@ -2752,12 +2787,12 @@ impl Args { util::retain_alloc_pages_for_aggregation(); } - let mut rdr = self.rconfig().reader()?; + let mut rdr = rconfig.reader()?; let full_headers = rdr.byte_headers()?.clone(); // Find weight column index and exclude it from selection let (weight_col_idx, sel, headers) = - self.process_headers_with_weight_exclusion(&full_headers)?; + self.process_headers_with_weight_exclusion(&full_headers, rconfig)?; init_date_inference(self.flag_infer_dates, &headers, whitelist)?; @@ -2785,7 +2820,7 @@ impl Args { let (chunking_mode_info, chunk_size) = if needs_memory_aware_chunking { // Sample records for memory estimation - let sample_records = util::sample_records(&self.rconfig(), 1000); + let sample_records = util::sample_records(rconfig, 1000); // Calculate memory-aware chunk size let chunk_size = calculate_memory_aware_chunk_size( @@ -2838,14 +2873,19 @@ impl Args { let args = Arc::new(self.clone()); for i in 0..nchunks { let (send, args, sel) = (send.clone(), Arc::clone(&args), sel.clone()); + // CLONE the resolved Config - never rebuild it with `args.rconfig()`. The clone + // shares the `Arc` holding a special-format input's converted temp, so + // every worker re-opens the index next to the SAME temp the parent indexed. A + // fresh Config would resolve a different temp with no index beside it and panic + // in the expect() below (issues #4446 / #4462). + let rconf = rconfig.clone(); let weight_idx: Option = weight_col_idx; pool.execute(move || { // The parent verified the index exists before chunking, but it can be // deleted or invalidated in between (TOCTOU) - notably by a concurrent // run cleaning up its autoindex. Fail loudly with actionable info like // the seek() below, rather than hitting undefined behavior. - let mut idx = args - .rconfig() + let mut idx = rconf .indexed() .expect("Failed to re-open index for parallel stats.") .expect("Index is no longer available for parallel stats."); @@ -3209,6 +3249,8 @@ impl Args { /// # Arguments /// /// * `full_headers` - The full CSV headers as a `ByteRecord` + /// * `rconfig` - the caller's already-resolved Config (see the invariant in `run()`); never + /// build a fresh one here /// /// # Returns /// @@ -3228,6 +3270,7 @@ impl Args { fn process_headers_with_weight_exclusion( &self, full_headers: &csv::ByteRecord, + rconfig: &Config, ) -> CliResult<(Option, Selection, csv::ByteRecord)> { if let Some(ref weight_col) = self.flag_weight { // Find weight column index in full headers @@ -3244,7 +3287,7 @@ impl Args { })?; // Create selection excluding weight column - let sel = self.rconfig().selection(full_headers)?; + let sel = rconfig.selection(full_headers)?; // Remove weight column index from selection if present let sel_vec: Vec = sel .iter() @@ -3268,7 +3311,7 @@ impl Args { Ok((Some(weight_idx), modified_sel, selected_headers)) } else { // No weight column specified, use normal selection - let sel = self.rconfig().selection(full_headers)?; + let sel = rconfig.selection(full_headers)?; let headers: csv::ByteRecord = sel.select(full_headers).collect(); Ok((None, sel, headers)) } diff --git a/src/config.rs b/src/config.rs index d73df733d..6bb3d518d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -626,6 +626,18 @@ impl Config { match util::convert_special_format(src, self.special_format, self.delimiter) { Ok(temp) => { let (_, delim, _) = get_delim_by_extension(&temp, self.delimiter); + // Logged INSIDE get_or_init, so it fires exactly once per Config + // family (a Config and all its clones share this OnceLock). That + // makes the line countable: more than one per input means some + // caller rebuilt a Config instead of cloning the resolved one, + // which silently doubles the conversion cost and puts any + // path-keyed artifact (an autoindex, most of all) beside a temp + // nobody else can see. + info!( + "converted special-format input {} to {}", + src.display(), + temp.display() + ); Ok((temp, delim)) }, Err(e) => Err(format!("Failed to convert special format: {e}")), @@ -664,10 +676,16 @@ impl Config { /// Whether this input is a special format (`.gz`/`.zip`/`.parquet`/`.jsonl`/...) that is read /// through a CONVERTED temp file rather than directly. /// - /// Callers that reconstruct a fresh `Config` per worker need this: each `Config` converts to - /// its OWN temp path, so anything keyed to that path - an autoindex, most notably - is - /// invisible to every other `Config` built from the same input. + /// No production caller remains: it existed so `stats` could refuse to autoindex these + /// inputs (#4445), and #4462 removed that refusal by having `stats` share one resolved + /// Config instead. Kept because the hazard it names is structural - a command that builds + /// a fresh `Config` per worker from a SPECIAL-FORMAT path resolves a DIFFERENT temp, so + /// anything keyed to that path (an autoindex, most notably) is invisible to every other + /// worker. Prefer one of the two fixes over branching on this: share the resolved `Config` + /// (`stats`, `frequency`), or resolve once and hand workers the RESOLVED PATH so the + /// Configs they build are never special-format (`moarstats`, since #4464). #[inline] + #[allow(dead_code)] pub const fn is_special_format(&self) -> bool { !matches!(self.special_format, SpecialFormat::Unknown) } @@ -1243,13 +1261,23 @@ mod tests { /// A special-format input must remain AUTOINDEXABLE at the Config level. /// - /// The `stats` parallel path cannot use such an index - it rebuilds a fresh Config per - /// worker, so each resolves a different converted temp file - and `stats` therefore skips - /// autoindexing these inputs itself. That skip must NOT live here: `frequency` deliberately - /// hands its workers the already-resolved Config (see the comments at its `indexed()` call - /// and in `parallel_ftables`), so its temp path is consistent across threads and its - /// special-format autoindexed parallel path is safe. Disabling autoindex in `index_files` - /// silently dropped `frequency` back to sequential processing on large compressed inputs. + /// Such an index is keyed to the CONVERTED TEMP file, so it is only usable by callers whose + /// threads all read through the SAME resolved Config. `frequency` (`parallel_ftables`), + /// `stats` (`parallel_stats`, since #4462) and the commands fixed in #4459 all hand their + /// workers a CLONE of the resolved Config - clones share the `Arc` holding the + /// resolution, so every thread sees one temp and one index. + /// + /// A caller that instead REBUILDS a Config per worker from the ORIGINAL path resolves a + /// different temp with no index beside it. No caller does that today: `moarstats` does + /// rebuild a Config per worker (`compute_outliers_and_kga`, `compute_all_bivariatestats`), + /// but both are handed `read_input_path` - already resolved by #4464 - so the Configs they + /// build are ordinary, not special-format. That is the second valid shape, and it is worth + /// preserving: passing either the resolved Config or the resolved path is fine, passing the + /// original path to a worker that rebuilds is not. + /// + /// The remedy for a future offender is to fix that caller - NOT to disable autoindexing + /// here, which would silently drop every well-behaved caller back to sequential processing + /// on large compressed inputs. #[test] fn special_format_input_is_still_autoindexable() { use std::io::Write; diff --git a/tests/test_stats.rs b/tests/test_stats.rs index c391006c3..fe2b26e99 100644 --- a/tests/test_stats.rs +++ b/tests/test_stats.rs @@ -8928,17 +8928,23 @@ fn stats_epoch_date_is_included_in_date_minmax() { } #[test] -fn stats_autoindex_is_skipped_for_special_format_inputs() { - // A special-format input (.gz/.zip/.parquet/.jsonl/...) is read through a CONVERTED temp - // file, and every Config instance converts to its OWN temp path. Autoindexing that temp - // file created an index only the parent Config could find: `stats`' parallel workers each - // build a fresh Config, resolved a DIFFERENT temp path, found no index beside it, and every - // worker panicked - yielding a stats file with headers and ZERO rows at exit code 0. +fn stats_parallelizes_special_format_inputs() { + // Issue #4462. A special-format input (.gz/.zip/.parquet/.jsonl/...) is read through a + // CONVERTED temp file, and every Config resolves its OWN temp. #4445 responded by refusing + // to autoindex these inputs at all, which cost them the parallel path entirely. + // + // #4462 shares ONE resolved Config across every stats path instead (clones share the + // `Arc` holding the resolution), so the parallel path is back: all workers see the + // same temp and the same sibling index. + // // A .zip fixture rather than .gz on purpose: zip support is non-optional in every build, // so this test also runs under `-F lite`, where the flate2 codec is absent. + // + // NOTE: results alone CANNOT distinguish sequential from parallel - they are identical by + // design, which is the whole point. The discriminating evidence is in the log. use std::io::Write; - let wrk = Workdir::new("stats_autoindex_special_format"); + let wrk = Workdir::new("stats_special_format_parallel"); let mut plain = String::from("a,b\n"); for i in 0..300 { @@ -8954,36 +8960,152 @@ fn stats_autoindex_is_skipped_for_special_format_inputs() { zw.write_all(plain.as_bytes()).unwrap(); zw.finish().unwrap(); + let log_name = format!( + "{}_rCURRENT.log", + wrk.qsv_bin().file_stem().unwrap().to_string_lossy() + ); + // -105 is negative (so |it| becomes the autoindex size, and 105 bytes is far below the - // fixture) and ends in 5 (so the autoindex is cleaned up afterwards) - let stats_of = |input: &str, out: &str| -> String { + // fixture) and ends in 5 (so the autoindex is cleaned up afterwards). + // Each run gets its OWN log dir - qsv appends to a single rCURRENT.log per directory, so a + // shared one would mix the two runs' lines and make the counts below meaningless. + let stats_of = |input: &str, out: &str, logdir: &str| -> (String, String) { + let log_dir = wrk.path(logdir); + std::fs::create_dir_all(&log_dir).unwrap(); let mut cmd = wrk.command("stats"); - cmd.arg("-E") + cmd.env("QSV_LOG_LEVEL", "info") + .env("QSV_LOG_DIR", &log_dir) + .arg("-E") .args(["--cache-threshold", "-105"]) .args(["--output", wrk.path(out).to_str().unwrap()]) .arg(input); wrk.assert_success(&mut cmd); - std::fs::read_to_string(wrk.path(out)).unwrap() + ( + std::fs::read_to_string(wrk.path(out)).unwrap(), + std::fs::read_to_string(log_dir.join(&log_name)).unwrap_or_default(), + ) }; - let zip_stats = stats_of("sf.zip", "zip.csv"); + let (zip_stats, zip_log) = stats_of("sf.zip", "zip.csv", "ziplog"); assert!( zip_stats.lines().count() > 1, - "a compressed input produced a stats file with no data rows - every parallel worker \ - failed to find the autoindex built against a different temp file" + "a compressed input produced a stats file with no data rows - the parallel workers failed \ + to find the autoindex" + ); + + // THE assertion for #4462: the parallel path actually ran. `parallel_stats` is the only + // producer of an "nchunks=" line; `sequential_stats` logs nothing of the sort. Verified by + // mutation - on master (which zeroes autoindex_size for special-format inputs) this run + // emits ZERO such lines. + assert!( + zip_log.contains("nchunks="), + "a special-format input must now use the parallel path; log had no nchunks= \ + line:\n{zip_log}" + ); + + // ...and it must convert the input EXACTLY ONCE. Each Config resolves its own temp, so a + // second line here means some stats path rebuilt `args.rconfig()` instead of sharing the + // resolved Config - which silently doubles the conversion cost AND puts the autoindex + // beside a temp the other threads cannot see. That is the #4446 crash shape. + let conversions = zip_log + .lines() + .filter(|l| l.contains("converted special-format input")) + .count(); + assert_eq!( + conversions, 1, + "the input must be converted exactly once, but was converted {conversions} \ + times:\n{zip_log}" ); + // The compressed run must not leave a warning behind either - notably the autoindex + // cleanup looking for `sf.zip.idx` (which never exists; the index lives beside the temp). + assert!( + !zip_log.contains("Could not remove index file"), + "autoindex cleanup looked for the index beside the compressed source, not the converted \ + temp:\n{zip_log}" + ); + + // The SEQUENTIAL path shares the resolved Config too. Without --cache-threshold or + // QSV_AUTOINDEX_SIZE there is no index, so this run goes through `sequential_stats` - and + // must still convert the input exactly once, not once for the setup Config and again for + // the one `sequential_stats` used to build its own reader. + { + std::fs::copy(wrk.path("sf.zip"), wrk.path("sf3.zip")).unwrap(); + let log_dir = wrk.path("seqlog"); + std::fs::create_dir_all(&log_dir).unwrap(); + let mut cmd = wrk.command("stats"); + cmd.env("QSV_LOG_LEVEL", "info") + .env("QSV_LOG_DIR", &log_dir) + // pinned, not inherited: this run's whole point is that NO index exists, and + // Workdir::command passes the ambient environment through. A QSV_AUTOINDEX_SIZE + // set on the CI runner would autoindex and silently make this the parallel path. + .env("QSV_AUTOINDEX_SIZE", "0") + .arg("-E") + .args(["--output", wrk.path("seq.csv").to_str().unwrap()]) + .arg("sf3.zip"); + wrk.assert_success(&mut cmd); + let seq_log = std::fs::read_to_string(log_dir.join(&log_name)).unwrap_or_default(); + assert!( + !seq_log.contains("nchunks="), + "with no index this run should NOT be parallel - the assertion below would then be \ + testing the wrong path:\n{seq_log}" + ); + let seq_conversions = seq_log + .lines() + .filter(|l| l.contains("converted special-format input")) + .count(); + assert_eq!( + seq_conversions, 1, + "the sequential path must reuse the resolved Config, converting exactly once, but \ + converted {seq_conversions} times:\n{seq_log}" + ); + assert_eq!( + std::fs::read_to_string(wrk.path("seq.csv")).unwrap(), + zip_stats, + "the sequential path must produce the same stats as the parallel one" + ); + } + + // The SECOND route into the autoindex - the QSV_AUTOINDEX_SIZE env var, applied in + // Config::new - was zeroed by a separate guard, so it needs its own run. Without + // --cache-threshold the index is left in place, hence a throwaway copy of the fixture. + { + std::fs::copy(wrk.path("sf.zip"), wrk.path("sf2.zip")).unwrap(); + let log_dir = wrk.path("envlog"); + std::fs::create_dir_all(&log_dir).unwrap(); + let mut cmd = wrk.command("stats"); + cmd.env("QSV_LOG_LEVEL", "info") + .env("QSV_LOG_DIR", &log_dir) + .env("QSV_AUTOINDEX_SIZE", "105") + .arg("-E") + .args(["--output", wrk.path("env.csv").to_str().unwrap()]) + .arg("sf2.zip"); + wrk.assert_success(&mut cmd); + let env_log = std::fs::read_to_string(log_dir.join(&log_name)).unwrap_or_default(); + assert!( + env_log.contains("nchunks="), + "QSV_AUTOINDEX_SIZE must also reach a special-format input's parallel path; log had \ + no nchunks= line:\n{env_log}" + ); + assert_eq!( + std::fs::read_to_string(wrk.path("env.csv")).unwrap(), + zip_stats, + "the QSV_AUTOINDEX_SIZE route must produce the same stats" + ); + } + // and the numbers must match the identical uncompressed input exactly - let plain_stats = stats_of("sf.csv", "plain.csv"); + let (plain_stats, _) = stats_of("sf.csv", "plain.csv", "plainlog"); assert_eq!( zip_stats, plain_stats, "stats for a .zip input must equal stats for the same data uncompressed" ); - // no autoindex is created beside a special-format input... + // The autoindex is built beside the CONVERTED TEMP, never beside the compressed source, + // so nothing is left in the work dir. assert!( !wrk.path("sf.zip.idx").exists(), - "a converted temp file must never be autoindexed" + "an index must never be written beside the compressed source" ); // ...while ordinary inputs still autoindex. -104 is negative (autoindex) but does not end @@ -9000,6 +9122,44 @@ fn stats_autoindex_is_skipped_for_special_format_inputs() { ); } +// `stats` infers the delimiter of a stdin input by peeking at its spill temp file, and stores +// the result in `args.flag_delimiter`. That used to reach the stats engine only because +// `sequential_stats`/`parallel_stats` rebuilt their own Config from `args`. #4462 made them +// share the ALREADY-BUILT `rconfig` instead - which was constructed BEFORE the inference - so +// the delimiter now has to be re-applied to it explicitly. Getting this wrong parses a TSV as +// a single comma-delimited column: wrong stats, exit code 0, no warning. +#[test] +fn stats_infers_stdin_delimiter() { + let wrk = Workdir::new("stats_stdin_delim"); + let mut tsv = String::from("a\tb\tc\n"); + for i in 0..50 { + tsv.push_str(&format!("{i}\t{}\tx\n", i * 2)); + } + wrk.create_from_string("in.tsv", &tsv); + + let mut cmd = wrk.command("stats"); + // QSV_DEFAULT_DELIMITER short-circuits the inference this test exists to cover, and + // Workdir::command inherits the ambient environment - so remove it explicitly. + cmd.env_remove("QSV_DEFAULT_DELIMITER").arg("-E").arg("-"); + cmd.stdin(std::process::Stdio::from( + std::fs::File::open(wrk.path("in.tsv")).unwrap(), + )); + let got: String = wrk.stdout(&mut cmd); + + // three data rows (one per column) means the tab delimiter was honored; a single row + // means the whole line was read as one comma-delimited field. + let fields: Vec<&str> = got + .lines() + .skip(1) + .map(|l| l.split(',').next().unwrap()) + .collect(); + assert_eq!( + fields, + vec!["a", "b", "c"], + "the tab delimiter inferred from stdin must reach the stats engine; got:\n{got}" + ); +} + // A blank line is counted by polars (util::count_rows) but SKIPPED by the csv reader that // actually feeds the compute pass. Seeding RECORD_COUNT from a separate pre-pass therefore // inflated every per-record denominator in to_record() - uniqueness_ratio, and the