feat: adaptive filter selectivity tracking + post-scan placement

Adds a runtime filter-placement layer on top of the row-group-morsel
split introduced by the parent commit. Each Parquet predicate is
assigned a `FilterId` and flows through a state machine
(`SelectivityTracker`) that moves it between three placements:

- `RowFilter` — evaluated inside the arrow-rs decoder as an
  `ArrowPredicate`, enabling late-materialization savings when the
  filter columns are a small fraction of the projection.
- `PostScan` — evaluated against the decoded wide batch before the
  projector strips it; used when late materialization has little to
  save or when the filter is CPU-heavy.
- `Dropped` — optional filters (hash-join dynamic filters wrapped in
  `OptionalFilterPhysicalExpr`) are skipped mid-stream when their CI
  upper bound on bytes-saved-per-second falls below a minimum.

Initial placement uses a cheap byte-ratio heuristic
(`filter_compressed_bytes / projection_compressed_bytes`); subsequent
placements refine using Welford online stats reported from both the
row-filter path (`DatafusionArrowPredicate::evaluate`) and the
post-scan path (`apply_post_scan_filters_with_stats`). Placement is
re-evaluated per morsel, so stats from the prior morsel's scan feed
into the next morsel's decision.

Config knobs on `TableParquetOptions.execution.parquet`:
- `filter_pushdown_min_bytes_per_sec` (default 100 MB/s)
- `filter_collecting_byte_ratio_threshold` (default 0.20)
- `filter_confidence_z` (default 2.0 ≈ 97.5% one-sided CI)

The `reorder_filters` option is removed; the adaptive tracker
subsumes its role.

Notable trade-offs documented in PR discussion:
- The adaptive layer adds ~10 % aggregate ClickBench overhead vs the
  pure morsel-split base (PR #10). Most of it lives in
  `ParquetLazyMorsel::build_stream_now` under parallel load; single-
  thread shows no regression. Candidate fix is splitting adaptive
  state out of `LazyMorselShared` so non-adaptive queries get the
  same `Arc` allocation shape as PR #10.
- The `OptionalFilterPhysicalExpr` wrapper changes plan display
  output (`DynamicFilter [...]` → `Optional(DynamicFilter [...])`);
  several sqllogictest expected outputs and snapshot tests were
  updated accordingly.
- A selectivity-tracker microbench was added under
  `benches/selectivity_tracker.rs` so future iterations on the
  tracker can be measured independently of full ClickBench.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: adriangb

benchmarks/bench.sh

@@ -820,7 +820,7 @@ run_clickbench_partitioned() {
 run_clickbench_pushdown() {
     RESULTS_FILE="${RESULTS_DIR}/clickbench_pushdown.json"
     echo "RESULTS_FILE: ${RESULTS_FILE}"
-    echo "Running clickbench (partitioned, 100 files) benchmark with pushdown_filters=true, reorder_filters=true..."
+    echo "Running clickbench (partitioned, 100 files) benchmark with pushdown_filters=true..."
     debug_run $CARGO_COMMAND --bin dfbench -- clickbench --pushdown --iterations 5 --path "${DATA_DIR}/hits_partitioned" --queries-path "${SCRIPT_DIR}/queries/clickbench/queries" -o "${RESULTS_FILE}" ${QUERY_ARG} ${LATENCY_ARG}
 }
 

benchmarks/src/clickbench.rs

@@ -58,7 +58,6 @@ pub struct RunOpt {
     ///
     /// Specifically, it enables:
     /// * `pushdown_filters = true`
-    /// * `reorder_filters = true`
     #[arg(long = "pushdown")]
     pushdown: bool,
 
@@ -196,14 +195,12 @@ impl RunOpt {
             // Turn on Parquet filter pushdown if requested
             if self.pushdown {
                 parquet_options.pushdown_filters = true;
-                parquet_options.reorder_filters = true;
             }
 
             if self.sorted_by.is_some() {
                 // We should compare the dynamic topk optimization when data is sorted, so we make the
                 // assumption that filter pushdown is also enabled in this case.
                 parquet_options.pushdown_filters = true;
-                parquet_options.reorder_filters = true;
             }
         }
 

datafusion-examples/examples/data_io/json_shredding.rs

@@ -90,7 +90,10 @@ pub async fn json_shredding() -> Result<()> {
     store.put(&path, payload).await?;
 
     // Set up query execution
-    let mut cfg = SessionConfig::new();
+    let mut cfg = SessionConfig::default().set(
+        "datafusion.execution.parquet.filter_pushdown_min_bytes_per_sec",
+        &ScalarValue::Float64(Some(0.0)),
+    );
     cfg.options_mut().execution.parquet.pushdown_filters = true;
     let ctx = SessionContext::new_with_config(cfg);
     ctx.runtime_env().register_object_store(

datafusion/common/src/config.rs

@@ -877,11 +877,6 @@ config_namespace! {
         /// reduce the number of rows decoded. This optimization is sometimes called "late materialization".
         pub pushdown_filters: bool, default = false
 
-        /// (reading) If true, filter expressions evaluated during the parquet decoding operation
-        /// will be reordered heuristically to minimize the cost of evaluation. If false,
-        /// the filters are applied in the same order as written in the query
-        pub reorder_filters: bool, default = false
-
         /// (reading) Force the use of RowSelections for filter results, when
         /// pushdown_filters is enabled. If false, the reader will automatically
         /// choose between a RowSelection and a Bitmap based on the number and
@@ -919,6 +914,49 @@ config_namespace! {
         /// parquet reader setting. 0 means no caching.
         pub max_predicate_cache_size: Option<usize>, default = None
 
+        /// (reading) Minimum bytes/sec throughput for adaptive filter pushdown.
+        /// Filters that achieve at least this throughput (bytes_saved / eval_time)
+        /// are promoted to row filters.
+        /// f64::INFINITY = no filters promoted (feature disabled).
+        /// 0.0 = all filters pushed as row filters (no adaptive logic).
+        /// Default: 104,857,600 bytes/sec (100 MiB/sec), empirically chosen based on
+        /// TPC-H, TPC-DS, and ClickBench benchmarks on an m4 MacBook Pro.
+        /// The optimal value for this setting likely depends on the relative
+        /// cost of CPU vs. IO in your environment, and to some extent the shape
+        /// of your query.
+        ///
+        /// **Interaction with `pushdown_filters`:**
+        /// This option only takes effect when `pushdown_filters = true`.
+        /// When pushdown is disabled, all filters run post-scan and this
+        /// threshold is ignored.
+        pub filter_pushdown_min_bytes_per_sec: f64, default = 104_857_600.0
+
+        /// (reading) Byte-ratio threshold for applying filters one at a time
+        /// (iterative pruning; aka row-level) vs. all at once (post-scan).
+        /// The ratio is computed as: (extra filter bytes not in projection) / (projected bytes).
+        /// Filters whose extra columns consume a smaller fraction than this threshold are placed as row filters.
+        /// Ratio of filter column bytes to projection bytes that controls
+        /// initial filter placement. Computed as
+        /// `filter_compressed_bytes / projection_compressed_bytes`.
+        /// Filters below this ratio start as row-level filters (enabling late
+        /// materialization); those above start as post-scan filters.
+        /// Default: 0.20 — filters whose columns are less than 20% of the
+        /// projection bytes start at row-level.
+        ///
+        /// **Interaction with `pushdown_filters`:**
+        /// Only takes effect when `pushdown_filters = true`.
+        pub filter_collecting_byte_ratio_threshold: f64, default = 0.20
+
+        /// (reading) Z-score for confidence intervals on filter effectiveness.
+        /// Controls how much statistical evidence is required before promoting
+        /// or demoting a filter. Lower values = faster decisions with less
+        /// confidence. Higher values = more conservative, requiring more data.
+        /// Default: 2.0 (~95% confidence).
+        ///
+        /// **Interaction with `pushdown_filters`:**
+        /// Only takes effect when `pushdown_filters = true`.
+        pub filter_confidence_z: f64, default = 2.0
+
         // The following options affect writing to parquet files
         // and map to parquet::file::properties::WriterProperties
 

datafusion/common/src/file_options/parquet_writer.rs

@@ -199,7 +199,7 @@ impl ParquetOptions {
             skip_metadata: _,
             metadata_size_hint: _,
             pushdown_filters: _,
-            reorder_filters: _,
+
             force_filter_selections: _, // not used for writer props
             allow_single_file_parallelism: _,
             maximum_parallel_row_group_writers: _,
@@ -210,6 +210,9 @@ impl ParquetOptions {
             coerce_int96: _,     // not used for writer props
             skip_arrow_metadata: _,
             max_predicate_cache_size: _,
+            filter_pushdown_min_bytes_per_sec: _, // not used for writer props
+            filter_collecting_byte_ratio_threshold: _, // not used for writer props
+            filter_confidence_z: _,               // not used for writer props
         } = self;
 
         let mut builder = WriterProperties::builder()
@@ -470,7 +473,7 @@ mod tests {
             skip_metadata: defaults.skip_metadata,
             metadata_size_hint: defaults.metadata_size_hint,
             pushdown_filters: defaults.pushdown_filters,
-            reorder_filters: defaults.reorder_filters,
+
             force_filter_selections: defaults.force_filter_selections,
             allow_single_file_parallelism: defaults.allow_single_file_parallelism,
             maximum_parallel_row_group_writers: defaults
@@ -484,6 +487,10 @@ mod tests {
             coerce_int96: None,
             max_predicate_cache_size: defaults.max_predicate_cache_size,
             use_content_defined_chunking: defaults.use_content_defined_chunking.clone(),
+            filter_pushdown_min_bytes_per_sec: defaults.filter_pushdown_min_bytes_per_sec,
+            filter_collecting_byte_ratio_threshold: defaults
+                .filter_collecting_byte_ratio_threshold,
+            filter_confidence_z: defaults.filter_confidence_z,
         }
     }
 
@@ -585,7 +592,7 @@ mod tests {
                 skip_metadata: global_options_defaults.skip_metadata,
                 metadata_size_hint: global_options_defaults.metadata_size_hint,
                 pushdown_filters: global_options_defaults.pushdown_filters,
-                reorder_filters: global_options_defaults.reorder_filters,
+
                 force_filter_selections: global_options_defaults.force_filter_selections,
                 allow_single_file_parallelism: global_options_defaults
                     .allow_single_file_parallelism,
@@ -607,6 +614,11 @@ mod tests {
                         norm_level: c.norm_level,
                     }
                 }),
+                filter_pushdown_min_bytes_per_sec: global_options_defaults
+                    .filter_pushdown_min_bytes_per_sec,
+                filter_collecting_byte_ratio_threshold: global_options_defaults
+                    .filter_collecting_byte_ratio_threshold,
+                filter_confidence_z: global_options_defaults.filter_confidence_z,
             },
             column_specific_options,
             key_value_metadata,

datafusion/core/src/dataframe/parquet.rs

@@ -164,7 +164,14 @@ mod tests {
         let plan = df.explain(false, false)?.collect().await?;
         // Filters all the way to Parquet
         let formatted = pretty::pretty_format_batches(&plan)?.to_string();
-        assert!(formatted.contains("FilterExec: id@0 = 1"), "{formatted}");
+        let data_source_exec_row = formatted
+            .lines()
+            .find(|line| line.contains("DataSourceExec:"))
+            .unwrap();
+        assert!(
+            data_source_exec_row.contains("predicate=id@0 = 1"),
+            "{formatted}"
+        );
 
         Ok(())
     }

datafusion/core/src/datasource/physical_plan/parquet.rs

@@ -167,9 +167,11 @@ mod tests {
             }
 
             if self.pushdown_predicate {
+                let mut opts = source.table_parquet_options().clone();
+                opts.global.filter_pushdown_min_bytes_per_sec = 0.0;
                 source = source
-                    .with_pushdown_filters(true)
-                    .with_reorder_filters(true);
+                    .with_table_parquet_options(opts)
+                    .with_pushdown_filters(true);
             } else {
                 source = source.with_pushdown_filters(false);
             }

datafusion/core/src/test_util/parquet.rs

@@ -57,8 +57,6 @@ pub struct TestParquetFile {
 pub struct ParquetScanOptions {
     /// Enable pushdown filters
     pub pushdown_filters: bool,
-    /// enable reordering filters
-    pub reorder_filters: bool,
     /// enable page index
     pub enable_page_index: bool,
 }
@@ -68,8 +66,9 @@ impl ParquetScanOptions {
     pub fn config(&self) -> SessionConfig {
         let mut config = ConfigOptions::new();
         config.execution.parquet.pushdown_filters = self.pushdown_filters;
-        config.execution.parquet.reorder_filters = self.reorder_filters;
         config.execution.parquet.enable_page_index = self.enable_page_index;
+        // Disable adaptive filter selection for tests that expect deterministic pushdown
+        config.execution.parquet.filter_pushdown_min_bytes_per_sec = 0.0;
         config.into()
     }
 }

datafusion/core/tests/parquet/filter_pushdown.rs

@@ -444,7 +444,6 @@ impl<'a> TestCase<'a> {
             .read_with_options(
                 ParquetScanOptions {
                     pushdown_filters: false,
-                    reorder_filters: false,
                     enable_page_index: false,
                 },
                 filter,
@@ -455,7 +454,6 @@ impl<'a> TestCase<'a> {
             .read_with_options(
                 ParquetScanOptions {
                     pushdown_filters: true,
-                    reorder_filters: false,
                     enable_page_index: false,
                 },
                 filter,
@@ -464,43 +462,28 @@ impl<'a> TestCase<'a> {
 
         assert_eq!(no_pushdown, only_pushdown);
 
-        let pushdown_and_reordering = self
-            .read_with_options(
-                ParquetScanOptions {
-                    pushdown_filters: true,
-                    reorder_filters: true,
-                    enable_page_index: false,
-                },
-                filter,
-            )
-            .await;
-
-        assert_eq!(no_pushdown, pushdown_and_reordering);
-
         let page_index_only = self
             .read_with_options(
                 ParquetScanOptions {
                     pushdown_filters: false,
-                    reorder_filters: false,
                     enable_page_index: true,
                 },
                 filter,
             )
             .await;
         assert_eq!(no_pushdown, page_index_only);
 
-        let pushdown_reordering_and_page_index = self
+        let pushdown_and_page_index = self
             .read_with_options(
                 ParquetScanOptions {
                     pushdown_filters: true,
-                    reorder_filters: true,
                     enable_page_index: true,
                 },
                 filter,
             )
             .await;
 
-        assert_eq!(no_pushdown, pushdown_reordering_and_page_index);
+        assert_eq!(no_pushdown, pushdown_and_page_index);
     }
 
     /// Reads data from a test parquet file using the specified scan options
@@ -633,6 +616,11 @@ async fn predicate_cache_default() -> datafusion_common::Result<()> {
 async fn predicate_cache_pushdown_default() -> datafusion_common::Result<()> {
     let mut config = SessionConfig::new();
     config.options_mut().execution.parquet.pushdown_filters = true;
+    config
+        .options_mut()
+        .execution
+        .parquet
+        .filter_pushdown_min_bytes_per_sec = 0.0;
     let ctx = SessionContext::new_with_config(config);
     // The cache is on by default, and used when filter pushdown is enabled
     PredicateCacheTest {
@@ -647,6 +635,11 @@ async fn predicate_cache_pushdown_default() -> datafusion_common::Result<()> {
 async fn predicate_cache_stats_issue_19561() -> datafusion_common::Result<()> {
     let mut config = SessionConfig::new();
     config.options_mut().execution.parquet.pushdown_filters = true;
+    config
+        .options_mut()
+        .execution
+        .parquet
+        .filter_pushdown_min_bytes_per_sec = 0.0;
     // force to get multiple batches to trigger repeated metric compound bug
     config.options_mut().execution.batch_size = 1;
     let ctx = SessionContext::new_with_config(config);
@@ -664,6 +657,11 @@ async fn predicate_cache_pushdown_default_selections_only()
 -> datafusion_common::Result<()> {
     let mut config = SessionConfig::new();
     config.options_mut().execution.parquet.pushdown_filters = true;
+    config
+        .options_mut()
+        .execution
+        .parquet
+        .filter_pushdown_min_bytes_per_sec = 0.0;
     // forcing filter selections minimizes the number of rows read from the cache
     config
         .options_mut()