Re-check dynamic filters before a Parquet file's first data read

datafusion/common/src/config.rs

@@ -1057,6 +1057,27 @@ config_namespace! {
         /// file only after the current one is fully consumed.
         pub file_stream_open_ahead: bool, default = false
 
+        /// When `true`, a Parquet scan whose predicate holds a dynamic filter
+        /// that can still change (for example the threshold pushed down by an
+        /// `ORDER BY ... LIMIT` TopK) re-checks the file's statistics against
+        /// the current filter right before reading the file's first data
+        /// pages. A file range that was opened early (by
+        /// `file_stream_open_ahead`, or while the filter was still empty) is
+        /// then skipped without fetching or decoding its row groups if the
+        /// filter has since been updated to exclude it. The check reuses the
+        /// scan's per-batch early-stop pruner, so the pruning predicate is
+        /// only rebuilt when the filter changed. Skipped ranges are counted by
+        /// the `files_ranges_pruned_before_read` metric, which is registered
+        /// only for scans with a dynamic filter while this option is enabled.
+        /// The check uses the file's `PartitionedFile` statistics, so it does
+        /// nothing for files without them, and it applies to every dynamic
+        /// filter that can still change (TopK thresholds and aggregate MIN/MAX
+        /// bounds; hash-join filters are complete before the probe side runs).
+        /// When `false` (the default) a single-row-group file is re-checked only
+        /// after its first batch; a file with several row groups is already
+        /// re-checked before each row group.
+        pub recheck_dynamic_filter_before_read: bool, default = false
+
         /// Aggregation ratio (number of distinct groups / number of input rows)
         /// threshold for skipping partial aggregation. If the value is greater
         /// then partial aggregation will skip aggregation for further input

datafusion/core/tests/fuzz_cases/mod.rs

@@ -26,6 +26,7 @@ mod sort_fuzz;
 #[expect(clippy::needless_pass_by_value)]
 mod sort_query_fuzz;
 mod topk_filter_pushdown;
+mod topk_recheck_before_read_fuzz;
 
 mod aggregation_fuzzer;
 #[expect(clippy::needless_pass_by_value)]