[Vedin]

Summary

Adds datafusion.execution.recheck_dynamic_filter_before_read (default false). When it is on, a Parquet scan whose predicate holds a dynamic filter that can still change re-checks the file pruner once, right before the file's first data read. A file range claimed early is then skipped without fetching its row group if the threshold published meanwhile excludes it. A range is claimed early by file_stream_open_ahead, or by a partition that drew an empty byte-range sliver.

Why: ORDER BY key LIMIT k over many single-row-group files. Each partition claims and opens its next file, and evaluates prune_file, before the TopK heap holds k rows. So those files pass pruning against the empty lit(true) filter, and their row groups are fetched and decoded even though the early-stop check then drops them after the first batch.

Measured on SPCS through Rustice (SF1 lineitem, 10 files, LIMIT 10):

Setting	Ranges matched	Rows decoded	Bytes scanned
defaults (5 partitions, open-ahead on)	7–8 / 14	0.6–1.2 M	50 MB
open-ahead off	3 / 14	0.6 M	28 MB
1 partition, open-ahead off	1 / 10	0.6 M	5.6 MB

Mechanism

• EarlyStoppingStream runs its existing FilePruner::should_prune check once before the first inner poll. The pruning predicate is rebuilt only when a watched filter changed; otherwise the check is an atomic load. A file stopped this way counts like an early-stopped file in files_ranges_pruned_statistics.
• The check is attached only to ranges with a row group left to fetch and no RowGroupPruner. Multi-row-group files are already checked before each row group.
• New metric files_ranges_pruned_before_read. It is registered only when the option is on and the predicate holds a dynamic filter, so default EXPLAIN ANALYZE output is unchanged.
• Plumbing: FileSource::with_execution_options(&ExecutionOptions) -> Option<Arc<dyn FileSource>>, default None. FileScanConfig::open_with_args calls it with the task's session config, so sources built with default options get the setting too. ParquetSource clones itself only when the value differs.
• Soundness: every dynamic filter producer only tightens (TopK, aggregate MIN/MAX bounds; hash-join filters are complete before the probe side runs). So a file pruned before its first read would also have been pruned by the existing first-batch check. This adds no new way to drop rows.

With the option off, behaviour is unchanged: no wrapper, metric or clone.

Tests

• Gated-store integration tests (core/tests/parquet/topk_recheck_before_read.rs): metadata reads return at once and data reads yield first.
  • 1 partition, 6 files: option off fetches 2 row groups, option on fetches 1.
  • 5 partitions, 10 files → 14 ranges including empty slivers: off fetches 9 row groups, on fetches 5.
  • No dynamic filter: identical metrics, bytes and results.
  • These fail with the check made a no-op.
• Opener unit test: a stopped ahead file makes 0 data reads.
• Fuzz test (core/tests/fuzz_cases/topk_recheck_before_read_fuzz.rs, extended_tests). It runs on a multi-thread runtime over a delayed store, comparing TopK with the option on vs off vs a full-sort oracle, every column by row id. Covered: one/two keys, ASC/DESC, NULLS FIRST/LAST, ties, Int64/Float64/Utf8/Decimal128/Date keys, files with and without statistics, constant-key files, batch sizes 1..8192. Seed and config are logged (TOPK_RECHECK_FUZZ_SEED, TOPK_RECHECK_FUZZ_CASES). Run on 3 × 256 seeded cases after the last change, and on ~1,900 cases before it: all pass.
• cargo test -p datafusion-datasource-parquet, parquet_integration (topk_recheck), information_schema.slt settings rows, configs.md. cargo clippy --no-deps -D warnings on the changed crates. cargo fmt.

Found while fuzzing, not changed here: NaN rows. Parquet min/max exclude NaN, so constant_value_from_stats treats {1.0, NaN} as the constant 1.0, and max-side pruning can drop NaN rows. NaN cases are therefore written only to files without statistics, and a separate fix will follow.

🤖 Generated with Claude Code