feat: disable join dynamic filter pushdown by default

When a hash join's build-side dynamic filter contains a `hash_lookup` term,
evaluating it on every probe-side row inside the scan duplicates the work
the join's own probe is about to do. On TPC-H Q17 this doubles end-to-end
query time (the equivalent of running the hash probe twice), and similar
~20-100% regressions show up across TPC-H Q3/Q5/Q8/Q9/Q14/Q18 and many
TPC-DS queries that join a small dim table to a large fact table.

Flip the default so users don't pay this cost on the common shape; the
config is still available per-query when the build-side filter is
selective enough to make scan-level pruning worthwhile (e.g. small
dimension table that prunes most of a fact table's row groups / pages).

The reset block at the end of dynamic_filter_pushdown_config.slt is
switched from `SET ... = true` to `RESET ...` so the test ends in the
configured default regardless of what that default is.

Author: rajesh-cmyk

datafusion/common/src/config.rs

@@ -1126,7 +1126,15 @@ config_namespace! {
 
         /// When set to true, the optimizer will attempt to push down Join dynamic filters
         /// into the file scan phase.
-        pub enable_join_dynamic_filter_pushdown: bool, default = true
+        ///
+        /// Disabled by default: when a join's build-side dynamic filter contains a
+        /// `hash_lookup` term, evaluating it on every probe-side row inside the scan
+        /// duplicates the work the join's own probe is about to do, which on benchmarks
+        /// like TPC-H Q17 doubles the query time (the equivalent of running the probe
+        /// twice). Re-enable per-query when the build-side filter is selective enough
+        /// to make scan-level pruning worthwhile (e.g. a small dimension table that
+        /// prunes most of a large fact table's row groups / pages).
+        pub enable_join_dynamic_filter_pushdown: bool, default = false
 
         /// When set to true, the optimizer will attempt to push down Aggregate dynamic filters
         /// into the file scan phase.

datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt

@@ -157,7 +157,7 @@ logical_plan
 physical_plan
 01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(id@0, id@0)], projection=[id@2, data@3, info@1]
 02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/join_right.parquet]]}, projection=[id, info], file_type=parquet
-03)--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 ]
+03)--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
 
 # Disable Join dynamic filter pushdown
 statement ok
@@ -857,16 +857,16 @@ DROP TABLE agg_parquet;
 
 # Config reset
 statement ok
-SET datafusion.optimizer.enable_topk_dynamic_filter_pushdown = true;
+RESET datafusion.optimizer.enable_topk_dynamic_filter_pushdown;
 
 statement ok
-SET datafusion.optimizer.enable_join_dynamic_filter_pushdown = true;
+RESET datafusion.optimizer.enable_join_dynamic_filter_pushdown;
 
 statement ok
-SET datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = true;
+RESET datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown;
 
 statement ok
-SET datafusion.optimizer.enable_dynamic_filter_pushdown = true;
+RESET datafusion.optimizer.enable_dynamic_filter_pushdown;
 
 statement ok
 RESET datafusion.execution.parquet.max_row_group_size;

datafusion/sqllogictest/test_files/explain_tree.slt

@@ -339,8 +339,8 @@ physical_plan
 17)-----------------------------│          files: 1         ││ partition_count(in->out): │
 18)-----------------------------│      format: parquet      ││           1 -> 4          │
 19)-----------------------------│                           ││                           │
-20)-----------------------------│         predicate:        ││    partitioning_scheme:   │
-21)-----------------------------│  DynamicFilter [ empty ]  ││     RoundRobinBatch(4)    │
+20)-----------------------------│                           ││    partitioning_scheme:   │
+21)-----------------------------│                           ││     RoundRobinBatch(4)    │
 22)-----------------------------└───────────────────────────┘└─────────────┬─────────────┘
 23)----------------------------------------------------------┌─────────────┴─────────────┐
 24)----------------------------------------------------------│       DataSourceExec      │

datafusion/sqllogictest/test_files/information_schema.slt

@@ -301,7 +301,7 @@ datafusion.optimizer.default_filter_selectivity 20
 datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown true
 datafusion.optimizer.enable_distinct_aggregation_soft_limit true
 datafusion.optimizer.enable_dynamic_filter_pushdown true
-datafusion.optimizer.enable_join_dynamic_filter_pushdown true
+datafusion.optimizer.enable_join_dynamic_filter_pushdown false
 datafusion.optimizer.enable_leaf_expression_pushdown true
 datafusion.optimizer.enable_piecewise_merge_join false
 datafusion.optimizer.enable_round_robin_repartition true
@@ -451,7 +451,7 @@ datafusion.optimizer.default_filter_selectivity 20 The default filter selectivit
 datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown true When set to true, the optimizer will attempt to push down Aggregate dynamic filters into the file scan phase.
 datafusion.optimizer.enable_distinct_aggregation_soft_limit true When set to true, the optimizer will push a limit operation into grouped aggregations which have no aggregate expressions, as a soft limit, emitting groups once the limit is reached, before all rows in the group are read.
 datafusion.optimizer.enable_dynamic_filter_pushdown true When set to true attempts to push down dynamic filters generated by operators (TopK, Join & Aggregate) into the file scan phase. For example, for a query such as `SELECT * FROM t ORDER BY timestamp DESC LIMIT 10`, the optimizer will attempt to push down the current top 10 timestamps that the TopK operator references into the file scans. This means that if we already have 10 timestamps in the year 2025 any files that only have timestamps in the year 2024 can be skipped / pruned at various stages in the scan. The config will suppress `enable_join_dynamic_filter_pushdown`, `enable_topk_dynamic_filter_pushdown` & `enable_aggregate_dynamic_filter_pushdown` So if you disable `enable_topk_dynamic_filter_pushdown`, then enable `enable_dynamic_filter_pushdown`, the `enable_topk_dynamic_filter_pushdown` will be overridden.
-datafusion.optimizer.enable_join_dynamic_filter_pushdown true When set to true, the optimizer will attempt to push down Join dynamic filters into the file scan phase.
+datafusion.optimizer.enable_join_dynamic_filter_pushdown false When set to true, the optimizer will attempt to push down Join dynamic filters into the file scan phase. Disabled by default: when a join's build-side dynamic filter contains a `hash_lookup` term, evaluating it on every probe-side row inside the scan duplicates the work the join's own probe is about to do, which on benchmarks like TPC-H Q17 doubles the query time (the equivalent of running the probe twice). Re-enable per-query when the build-side filter is selective enough to make scan-level pruning worthwhile (e.g. a small dimension table that prunes most of a large fact table's row groups / pages).
 datafusion.optimizer.enable_leaf_expression_pushdown true When set to true, the optimizer will extract leaf expressions (such as `get_field`) from filter/sort/join nodes into projections closer to the leaf table scans, and push those projections down towards the leaf nodes.
 datafusion.optimizer.enable_piecewise_merge_join false When set to true, piecewise merge join is enabled. PiecewiseMergeJoin is currently experimental. Physical planner will opt for PiecewiseMergeJoin when there is only one range filter.
 datafusion.optimizer.enable_round_robin_repartition true When set to true, the physical plan optimizer will try to add round robin repartitioning to increase parallelism to leverage more CPU cores
@@ -895,7 +895,7 @@ show functions
 statement ok
 reset datafusion.catalog.information_schema;
 
-# The SLT runner sets `target_partitions` to 4 instead of using the default, so 
+# The SLT runner sets `target_partitions` to 4 instead of using the default, so
 # reset it explicitly.
 statement ok
 set datafusion.execution.target_partitions = 4;

datafusion/sqllogictest/test_files/joins.slt

@@ -2865,9 +2865,8 @@ physical_plan
 02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(t2_id@0, t1_id@0)]
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
 04)----SortExec: expr=[t1_id@0 ASC NULLS LAST], preserve_partitioning=[true]
-05)------FilterExec: DynamicFilter [ empty ]
-06)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 query IT rowsort
 SELECT t1_id, t1_name FROM left_semi_anti_join_table_t1 t1 WHERE t1_id IN (SELECT t2_id FROM left_semi_anti_join_table_t2 t2) ORDER BY t1_id
@@ -2901,9 +2900,8 @@ physical_plan
 02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(t2_id@0, t1_id@0)]
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
 04)----SortExec: expr=[t1_id@0 ASC NULLS LAST], preserve_partitioning=[true]
-05)------FilterExec: DynamicFilter [ empty ]
-06)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 query IT
 SELECT t1_id, t1_name FROM left_semi_anti_join_table_t1 t1 LEFT SEMI JOIN left_semi_anti_join_table_t2 t2 ON (t1_id = t2_id) ORDER BY t1_id
@@ -2958,9 +2956,8 @@ physical_plan
 02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(t2_id@0, t1_id@0)]
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
 04)----SortExec: expr=[t1_id@0 ASC NULLS LAST], preserve_partitioning=[true]
-05)------FilterExec: DynamicFilter [ empty ]
-06)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 query IT rowsort
 SELECT t1_id, t1_name FROM left_semi_anti_join_table_t1 t1 WHERE t1_id IN (SELECT t2_id FROM left_semi_anti_join_table_t2 t2) ORDER BY t1_id
@@ -2994,9 +2991,8 @@ physical_plan
 02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(t2_id@0, t1_id@0)]
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
 04)----SortExec: expr=[t1_id@0 ASC NULLS LAST], preserve_partitioning=[true]
-05)------FilterExec: DynamicFilter [ empty ]
-06)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 query IT
 SELECT t1_id, t1_name FROM left_semi_anti_join_table_t1 t1 LEFT SEMI JOIN left_semi_anti_join_table_t2 t2 ON (t1_id = t2_id) ORDER BY t1_id
@@ -3052,9 +3048,8 @@ physical_plan
 02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(t2_id@0, t1_id@0)], filter=t2_name@1 != t1_name@0
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
 04)----SortExec: expr=[t1_id@0 ASC NULLS LAST], preserve_partitioning=[true]
-05)------FilterExec: DynamicFilter [ empty ]
-06)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 query ITI rowsort
 SELECT t1_id, t1_name, t1_int FROM right_semi_anti_join_table_t1 t1 WHERE EXISTS (SELECT * FROM right_semi_anti_join_table_t2 t2 where t2.t2_id = t1.t1_id and t2.t2_name <> t1.t1_name) ORDER BY t1_id
@@ -3069,9 +3064,8 @@ physical_plan
 02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(t2_id@0, t1_id@0)], filter=t2_name@0 != t1_name@1
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
 04)----SortExec: expr=[t1_id@0 ASC NULLS LAST], preserve_partitioning=[true]
-05)------FilterExec: DynamicFilter [ empty ]
-06)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 query ITI rowsort
 SELECT t1_id, t1_name, t1_int FROM right_semi_anti_join_table_t2 t2 RIGHT SEMI JOIN right_semi_anti_join_table_t1 t1 on (t2.t2_id = t1.t1_id and t2.t2_name <> t1.t1_name) ORDER BY t1_id
@@ -3124,9 +3118,8 @@ physical_plan
 02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(t2_id@0, t1_id@0)], filter=t2_name@1 != t1_name@0
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
 04)----SortExec: expr=[t1_id@0 ASC NULLS LAST], preserve_partitioning=[true]
-05)------FilterExec: DynamicFilter [ empty ]
-06)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 query ITI rowsort
 SELECT t1_id, t1_name, t1_int FROM right_semi_anti_join_table_t1 t1 WHERE EXISTS (SELECT * FROM right_semi_anti_join_table_t2 t2 where t2.t2_id = t1.t1_id and t2.t2_name <> t1.t1_name) ORDER BY t1_id
@@ -3141,9 +3134,8 @@ physical_plan
 02)--HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(t2_id@0, t1_id@0)], filter=t2_name@0 != t1_name@1
 03)----DataSourceExec: partitions=1, partition_sizes=[1]
 04)----SortExec: expr=[t1_id@0 ASC NULLS LAST], preserve_partitioning=[true]
-05)------FilterExec: DynamicFilter [ empty ]
-06)--------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
-07)----------DataSourceExec: partitions=1, partition_sizes=[1]
+05)------RepartitionExec: partitioning=RoundRobinBatch(2), input_partitions=1
+06)--------DataSourceExec: partitions=1, partition_sizes=[1]
 
 query ITI rowsort
 SELECT t1_id, t1_name, t1_int FROM right_semi_anti_join_table_t2 t2 RIGHT SEMI JOIN right_semi_anti_join_table_t1 t1 on (t2.t2_id = t1.t1_id and t2.t2_name <> t1.t1_name) ORDER BY t1_id
@@ -4149,8 +4141,7 @@ physical_plan
 01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(b@1, y@1)], filter=a@0 < x@1
 02)--DataSourceExec: partitions=1, partition_sizes=[0]
 03)--SortExec: expr=[x@0 ASC NULLS LAST], preserve_partitioning=[false]
-04)----FilterExec: DynamicFilter [ empty ]
-05)------DataSourceExec: partitions=1, partition_sizes=[0]
+04)----DataSourceExec: partitions=1, partition_sizes=[0]
 
 # Test full join with limit
 statement ok
@@ -4452,8 +4443,7 @@ physical_plan
 04)------FilterExec: b@1 > 3, projection=[a@0]
 05)--------DataSourceExec: partitions=2, partition_sizes=[1, 1]
 06)----SortExec: expr=[c@2 DESC], preserve_partitioning=[true]
-07)------FilterExec: DynamicFilter [ empty ]
-08)--------DataSourceExec: partitions=2, partition_sizes=[1, 1]
+07)------DataSourceExec: partitions=2, partition_sizes=[1, 1]
 
 query TT
 explain select * from test where a in (select a from test where b > 3) order by c desc nulls last;
@@ -4473,8 +4463,7 @@ physical_plan
 04)------FilterExec: b@1 > 3, projection=[a@0]
 05)--------DataSourceExec: partitions=2, partition_sizes=[1, 1]
 06)----SortExec: expr=[c@2 DESC NULLS LAST], preserve_partitioning=[true]
-07)------FilterExec: DynamicFilter [ empty ]
-08)--------DataSourceExec: partitions=2, partition_sizes=[1, 1]
+07)------DataSourceExec: partitions=2, partition_sizes=[1, 1]
 
 query III
 select * from test where a in (select a from test where b > 3) order by c desc nulls first;