fix: start optional filters as RowFilter when byte ratio is small

Optional/dynamic filters from hash-join build sides were unconditionally
placed as PostScan on first encounter, losing late-materialization
benefits even when filter columns were small relative to the projection.
With few file splits opened in parallel, the tracker rarely accumulated
enough samples to promote them mid-query.

Apply the same byte_ratio_threshold heuristic used for static filters.
The CI lower-bound promotion and CI upper-bound demotion paths still
apply, including Drop for ineffective optional filters.

Local TPC-DS sf1 (M-series, pushdown_filters=true):

| Query | Main  | Branch | Branch+Fix |
|-------|-------|--------|------------|
| Q24   |    72 |    452 |         70 |
| Q17   |   124 |    212 |        121 |
| Q25   |   182 |    379 |        203 |
| Q29   |   152 |    312 |        145 |
| Q7    |   224 |    297 |        220 |
| Q58   |   129 |    191 |        133 |
| Q64   | 28213 |    672 |        578 |
| Q9    |   228 |     96 |         87 |
| Q76   |   172 |    105 |        156 |

Q76 regresses slightly vs the no-fix branch (CASE/hash_lookup is CPU-
heavy at row level) but still beats main.

Also updates dynamic_filter_pushdown_config.slt to match the
Optional(DynamicFilter ...) display introduced earlier in the branch.

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

Author: adriangb

datafusion/datasource-parquet/src/selectivity.rs

@@ -540,10 +540,7 @@ impl SelectivityTrackerInner {
         post_scan: &mut Vec<(FilterId, Arc<dyn PhysicalExpr>)>,
         stats_map: &HashMap<FilterId, Mutex<SelectivityStats>>,
     ) {
-        if expr
-            .downcast_ref::<OptionalFilterPhysicalExpr>()
-            .is_none()
-        {
+        if expr.downcast_ref::<OptionalFilterPhysicalExpr>().is_none() {
             self.filter_states.insert(id, FilterState::PostScan);
             post_scan.push((id, Arc::clone(expr)));
             // Reset stats for this filter so it can be re-evaluated as a post-scan filter.
@@ -640,29 +637,9 @@ impl SelectivityTrackerInner {
             let state = self.filter_states.get(&id).copied();
 
             let Some(state) = state else {
-                // New filter: decide initial placement.
-
-                // Optional/dynamic filters (e.g. hash join pushdown) always start
-                // PostScan because their selectivity is unknown and they can be
-                // dropped if ineffective. This is especially important for
-                // single-file tables where there's no second file to adapt.
-                let is_optional = expr
-                    .downcast_ref::<OptionalFilterPhysicalExpr>()
-                    .is_some();
-                if is_optional {
-                    debug!(
-                        "FilterId {id}: New optional filter → Post-scan (conservative) — {expr}"
-                    );
-                    self.filter_states.insert(id, FilterState::PostScan);
-                    if !stats_map.contains_key(&id) {
-                        new_filter_ids.push(id);
-                    }
-                    post_scan_filters.push((id, expr));
-                    continue;
-                }
-
-                // Static filters: use filter_bytes / projection_bytes to decide
-                // initial placement. This ratio captures the I/O tradeoff:
+                // New filter: decide initial placement using the
+                // filter_bytes / projection_bytes ratio. This ratio captures
+                // the I/O tradeoff:
                 //
                 // - Low ratio (filter columns are small vs projection): row-filter
                 //   enables late materialization — the large non-filter portion of
@@ -757,9 +734,7 @@ impl SelectivityTrackerInner {
                         let stats = entry.lock();
                         if let Some(ub) = stats.confidence_upper_bound(confidence_z)
                             && ub < config.min_bytes_per_sec
-                            && expr
-                                .downcast_ref::<OptionalFilterPhysicalExpr>()
-                                .is_some()
+                            && expr.downcast_ref::<OptionalFilterPhysicalExpr>().is_some()
                         {
                             drop(stats);
                             debug!(

datafusion/physical-expr-common/src/physical_expr.rs

@@ -1036,11 +1036,7 @@ mod test {
         );
 
         // Inner expr is NOT detectable as optional
-        assert!(
-            inner
-                .downcast_ref::<OptionalFilterPhysicalExpr>()
-                .is_none()
-        );
+        assert!(inner.downcast_ref::<OptionalFilterPhysicalExpr>().is_none());
     }
 
     #[test]

datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt

@@ -337,8 +337,8 @@ physical_plan
 01)HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(id@0, id@0)]
 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet
 03)--SortExec: expr=[data@1 DESC], preserve_partitioning=[false]
-04)----FilterExec: DynamicFilter [ empty ]
-05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+04)----FilterExec: Optional(DynamicFilter [ empty ])
+05)------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=Optional(DynamicFilter [ empty ])
 
 statement count 0
 SET datafusion.execution.parquet.pushdown_filters = true;
@@ -361,7 +361,7 @@ physical_plan
 01)HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(id@0, id@0)]
 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id], file_type=parquet
 03)--SortExec: expr=[data@1 DESC], preserve_partitioning=[false]
-04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=DynamicFilter [ empty ]
+04)----DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_left.parquet]]}, projection=[id, data], file_type=parquet, predicate=Optional(DynamicFilter [ empty ])
 
 statement count 0
 RESET datafusion.execution.parquet.pushdown_filters;