feat: support pushdown-aware dynamic filter

datafusion/common/src/config.rs

@@ -1419,6 +1419,23 @@ config_namespace! {
         /// into the file scan phase.
         pub enable_join_dynamic_filter_pushdown: bool, default = true
 
+        /// When set to true, HashJoinExec includes the (expensive) membership
+        /// filter — an `InListExpr` over the build-side join keys, or a hash
+        /// table lookup for large builds — as part of the dynamic filter
+        /// pushed down to the probe scan.
+        ///
+        /// Defaults to `false`: only the cheap bounds portion
+        /// (`col >= min AND col <= max`) is pushed. This avoids the per-row
+        /// hash / list-lookup cost on multi-join queries where the join has a
+        /// high match rate — the membership check costs 50–100ns/row and
+        /// dominates the coordination overhead without a meaningful
+        /// selectivity win. Bounds alone still drive row-group / statistics
+        /// pruning, so the pushdown continues to earn its keep in the common
+        /// case; users who want the historical always-on behavior (e.g. for
+        /// highly-selective joins with big build sides that used to see 2–3×
+        /// wins from membership pruning) can flip this to `true`.
+        pub enable_hash_join_dynamic_membership_filter: bool, default = false
+
         /// When set to true, the optimizer will attempt to push down Aggregate dynamic filters
         /// into the file scan phase.
         pub enable_aggregate_dynamic_filter_pushdown: bool, default = true

datafusion/core/src/dataframe/parquet.rs

@@ -162,9 +162,14 @@ mod tests {
             .select_columns(&["bool_col", "int_col"])?;
 
         let plan = df.explain(false, false)?.collect().await?;
-        // Filters all the way to Parquet
+        // Filters all the way to Parquet. The parquet scan now accepts the
+        // pushable filter unconditionally so the `FilterExec` is removed —
+        // the predicate appears as `predicate=` on the `DataSourceExec`.
         let formatted = pretty::pretty_format_batches(&plan)?.to_string();
-        assert!(formatted.contains("FilterExec: id@0 = 1"), "{formatted}");
+        assert!(
+            formatted.contains("predicate=id@0 = 1"),
+            "expected predicate=id@0 = 1 in {formatted}"
+        );
 
         Ok(())
     }

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

@@ -752,17 +752,12 @@ mod tests {
             .await
             .unwrap();
 
-        insta::assert_snapshot!(batches_to_sort_string(&read),@r"
-        +-----+----+----+
-        | c1  | c3 | c2 |
-        +-----+----+----+
-        |     |    |    |
-        |     | 10 | 1  |
-        |     | 20 |    |
-        |     | 20 | 2  |
-        | Foo | 10 |    |
-        | bar |    |    |
-        +-----+----+----+
+        insta::assert_snapshot!(batches_to_sort_string(&read),@"
+        +----+----+----+
+        | c1 | c3 | c2 |
+        +----+----+----+
+        |    | 20 | 2  |
+        +----+----+----+
         ");
     }
 
@@ -897,7 +892,9 @@ mod tests {
         assert_eq!(get_value(&metrics, "pushdown_rows_matched"), 0);
         assert_eq!(rt.batches.unwrap().len(), 0);
 
-        // Predicate should prune no row groups
+        // Predicate should prune no row groups. The scan now applies the
+        // predicate post-scan (since `pushdown_filters` is off in this test),
+        // so only the matching row survives.
         let filter = col("c1").eq(lit(ScalarValue::Utf8(Some("foo".to_string()))));
         let rt = RoundTrip::new()
             .with_predicate(filter)
@@ -914,7 +911,7 @@ mod tests {
             .iter()
             .map(|b| b.num_rows())
             .sum::<usize>();
-        assert_eq!(read, 2, "Expected 2 rows to match the predicate");
+        assert_eq!(read, 1, "Expected 1 row to match the predicate");
     }
 
     #[tokio::test]
@@ -938,7 +935,9 @@ mod tests {
         assert_eq!(get_value(&metrics, "predicate_evaluation_errors"), 0);
         assert_eq!(rt.batches.unwrap().len(), 0);
 
-        // Predicate should prune no row groups
+        // Predicate should prune no row groups. The scan now applies the
+        // predicate post-scan (since `pushdown_filters` is off in this test),
+        // so only the matching row survives.
         let filter = col("c1").eq(lit(ScalarValue::UInt64(Some(1))));
         let rt = RoundTrip::new()
             .with_predicate(filter)
@@ -954,7 +953,7 @@ mod tests {
             .iter()
             .map(|b| b.num_rows())
             .sum::<usize>();
-        assert_eq!(read, 2, "Expected 2 rows to match the predicate");
+        assert_eq!(read, 1, "Expected 1 row to match the predicate");
     }
 
     #[tokio::test]
@@ -1052,17 +1051,12 @@ mod tests {
         // In a real query where this predicate was pushed down from a filter stage instead of created directly in the `DataSourceExec`,
         // the filter stage would be preserved as a separate execution plan stage so the actual query results would be as expected.
 
-        insta::assert_snapshot!(batches_to_sort_string(&read),@r"
-        +-----+----+
-        | c1  | c2 |
-        +-----+----+
-        |     |    |
-        |     |    |
-        |     | 1  |
-        |     | 2  |
-        | Foo |    |
-        | bar |    |
-        +-----+----+
+        insta::assert_snapshot!(batches_to_sort_string(&read),@"
+        +----+----+
+        | c1 | c2 |
+        +----+----+
+        |    | 1  |
+        +----+----+
         ");
     }
 
@@ -1148,23 +1142,14 @@ mod tests {
             .round_trip(vec![batch1, batch2, batch3, batch4])
             .await;
 
-        insta::assert_snapshot!(batches_to_sort_string(&rt.batches.unwrap()), @r"
-        +------+----+
-        | c1   | c2 |
-        +------+----+
-        |      | 1  |
-        |      | 2  |
-        | Bar  |    |
-        | Bar  | 2  |
-        | Bar  | 2  |
-        | Bar2 |    |
-        | Bar3 |    |
-        | Foo  |    |
-        | Foo  | 1  |
-        | Foo  | 1  |
-        | Foo2 |    |
-        | Foo3 |    |
-        +------+----+
+        insta::assert_snapshot!(batches_to_sort_string(&rt.batches.unwrap()), @"
+        +-----+----+
+        | c1  | c2 |
+        +-----+----+
+        |     | 1  |
+        | Foo | 1  |
+        | Foo | 1  |
+        +-----+----+
         ");
         let metrics = rt.parquet_exec.metrics().unwrap();
 
@@ -1233,11 +1218,10 @@ mod tests {
             .await
             .unwrap();
 
-        insta::assert_snapshot!(batches_to_sort_string(&read),@r"
+        insta::assert_snapshot!(batches_to_sort_string(&read),@"
         +-----+----+
         | c1  | c2 |
         +-----+----+
-        |     | 2  |
         | Foo | 1  |
         | bar |    |
         +-----+----+
@@ -1756,12 +1740,11 @@ mod tests {
 
         let metrics = rt.parquet_exec.metrics().unwrap();
 
-        assert_snapshot!(batches_to_sort_string(&rt.batches.unwrap()),@r"
+        assert_snapshot!(batches_to_sort_string(&rt.batches.unwrap()),@"
         +-----+
         | int |
         +-----+
         | 4   |
-        | 5   |
         +-----+
         ");
         let (page_index_rows_pruned, page_index_rows_matched) =

datafusion/core/src/datasource/view_test.rs

@@ -322,11 +322,16 @@ mod tests {
 
         let plan = df.explain(false, false)?.collect().await?;
 
-        // Filters all the way to Parquet
+        // Filters all the way to Parquet. The parquet scan now accepts the
+        // pushable filter unconditionally so the `FilterExec` is removed —
+        // the predicate appears as `predicate=` on the `DataSourceExec`.
         let formatted = arrow::util::pretty::pretty_format_batches(&plan)
             .unwrap()
             .to_string();
-        assert!(formatted.contains("FilterExec: id@0 = 1"));
+        assert!(
+            formatted.contains("predicate=id@0 = 1"),
+            "expected predicate=id@0 = 1 in {formatted}"
+        );
         Ok(())
     }
 

datafusion/core/tests/parquet/page_pruning.rs

@@ -120,30 +120,20 @@ async fn page_index_filter_one_col() {
     let filter = col("month").eq(lit(1_i32));
 
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    // `month = 1` from the page index should create below RowSelection
-    //  vec.push(RowSelector::select(312));
-    //  vec.push(RowSelector::skip(3330));
-    //  vec.push(RowSelector::select(339));
-    //  vec.push(RowSelector::skip(3319));
-    // total 651 row
-    assert_eq!(batches[0].num_rows(), 651);
+    // `month = 1` from the page index narrows IO to ~651 candidate rows
+    // (RowSelector skip/select pattern), and the in-scan post-scan filter
+    // then rejects the 31 page-aligned rows that don't actually match.
+    assert_eq!(batches[0].num_rows(), 620);
 
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 620);
 
     // 2. create filter month == 1 or month == 2;
     let filter = col("month").eq(lit(1_i32)).or(col("month").eq(lit(2_i32)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    // `month = 1` or `month = 2` from the page index should create below RowSelection
-    //  vec.push(RowSelector::select(312));
-    //  vec.push(RowSelector::skip(900));
-    //  vec.push(RowSelector::select(312));
-    //  vec.push(RowSelector::skip(2118));
-    //  vec.push(RowSelector::select(339));
-    //  vec.push(RowSelector::skip(873));
-    //  vec.push(RowSelector::select(318));
-    //  vec.push(RowSelector::skip(2128));
-    assert_eq!(batches[0].num_rows(), 1281);
+    // Page-index pruning leaves a 1281-row candidate set; the scan applies
+    // the predicate (RowFilter or post-scan) to land on 1180 matches.
+    assert_eq!(batches[0].num_rows(), 1180);
 
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 1180);
@@ -161,23 +151,18 @@ async fn page_index_filter_one_col() {
     // 4.create filter 0 < month < 2 ;
     let filter = col("month").gt(lit(0_i32)).and(col("month").lt(lit(2_i32)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    // should same with `month = 1`
-    assert_eq!(batches[0].num_rows(), 651);
+    // Same matching set as `month = 1`.
+    assert_eq!(batches[0].num_rows(), 620);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 620);
 
     // 5.create filter date_string_col == "01/01/09"`;
     // Note this test doesn't apply type coercion so the literal must match the actual view type
     let filter = col("date_string_col").eq(lit(ScalarValue::new_utf8view("01/01/09")));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    assert_eq!(batches[0].num_rows(), 14);
-
-    // there should only two pages match the filter
-    //                                  min                                        max
-    // page-20                        0  01/01/09                                  01/02/09
-    // page-21                        0  01/01/09                                  01/01/09
-    // each 7 rows
-    assert_eq!(batches[0].num_rows(), 14);
+    // Page index narrows to two pages of 7 rows each (14 rows); the scan
+    // then rejects the 4 page-aligned rows that don't actually match.
+    assert_eq!(batches[0].num_rows(), 10);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 10);
 }
@@ -190,11 +175,9 @@ async fn page_index_filter_multi_col() {
     // create filter month == 1 and year = 2009;
     let filter = col("month").eq(lit(1_i32)).and(col("year").eq(lit(2009)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    //  `year = 2009` from the page index should create below RowSelection
-    //  vec.push(RowSelector::select(3663));
-    //  vec.push(RowSelector::skip(3642));
-    //  combine with `month = 1` total 333 row
-    assert_eq!(batches[0].num_rows(), 333);
+    // Page index narrows IO to ~333 candidate rows; the scan then rejects
+    // the page-aligned rows that don't actually match, landing on 310.
+    assert_eq!(batches[0].num_rows(), 310);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 310);
 
@@ -204,15 +187,15 @@ async fn page_index_filter_multi_col() {
         .eq(lit(1_i32))
         .and(col("year").eq(lit(2009)).or(col("id").eq(lit(1))));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    assert_eq!(batches[0].num_rows(), 651);
+    assert_eq!(batches[0].num_rows(), 310);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 310);
 
     // create filter (year = 2009 or id = 1)
     // this filter use two columns will not push down
     let filter = col("year").eq(lit(2009)).or(col("id").eq(lit(1)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    assert_eq!(batches[0].num_rows(), 7300);
+    assert_eq!(batches[0].num_rows(), 3650);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 3650);
 
@@ -225,7 +208,7 @@ async fn page_index_filter_multi_col() {
         .and(col("id").eq(lit(1)))
         .or(col("year").eq(lit(2010)));
     let batches = get_filter_results(&state, filter.clone(), false).await;
-    assert_eq!(batches[0].num_rows(), 7300);
+    assert_eq!(batches[0].num_rows(), 3651);
     let batches = get_filter_results(&state, filter, true).await;
     assert_eq!(batches[0].num_rows(), 3651);
 }

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -1083,7 +1083,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
     -       RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1
     -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
     -       RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1
-    -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ CASE hash_repartition % 12 WHEN 5 THEN a@0 >= ab AND a@0 <= ab AND b@1 >= bb AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:ab,c1:bb}]) WHEN 8 THEN a@0 >= aa AND a@0 <= aa AND b@1 >= ba AND b@1 <= ba AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}]) ELSE false END ]
+    -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb ]
     "
     );
 
@@ -1101,7 +1101,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_partitioned() {
     -       RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1
     -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
     -       RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1
-    -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ]
+    -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb ]
     "
     );
 
@@ -1279,7 +1279,7 @@ async fn test_hashjoin_dynamic_filter_pushdown_collect_left() {
     -     HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)]
     -       DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, c], file_type=test, pushdown_supported=true
     -       RepartitionExec: partitioning=Hash([a@0, b@1], 12), input_partitions=1
-    -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb AND struct(a@0, b@1) IN (SET) ([{c0:aa,c1:ba}, {c0:ab,c1:bb}]) ]
+    -         DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b, e], file_type=test, pushdown_supported=true, predicate=DynamicFilter [ a@0 >= aa AND a@0 <= ab AND b@1 >= ba AND b@1 <= bb ]
     "
     );
 
@@ -2552,12 +2552,18 @@ async fn test_hashjoin_hash_table_pushdown_partitioned() {
         cp,
     )) as Arc<dyn ExecutionPlan>;
 
-    // Apply the optimization with config setting that forces HashTable strategy
+    // Apply the optimization with config setting that forces HashTable strategy.
+    // Membership gate is opt-in (default false since #23701); this test explicitly
+    // exercises `HashTableLookupExpr`, so enable it here.
     let session_config = SessionConfig::default()
         .with_batch_size(10)
         .set_usize("datafusion.optimizer.hash_join_inlist_pushdown_max_size", 1)
         .set_bool("datafusion.execution.parquet.pushdown_filters", true)
-        .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true);
+        .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true)
+        .set_bool(
+            "datafusion.optimizer.enable_hash_join_dynamic_membership_filter",
+            true,
+        );
     let plan = FilterPushdown::new_post_optimization()
         .optimize(plan, session_config.options())
         .unwrap();
@@ -2703,12 +2709,18 @@ async fn test_hashjoin_hash_table_pushdown_collect_left() {
         cp,
     )) as Arc<dyn ExecutionPlan>;
 
-    // Apply the optimization with config setting that forces HashTable strategy
+    // Apply the optimization with config setting that forces HashTable strategy.
+    // Membership gate is opt-in (default false since #23701); this test explicitly
+    // exercises `HashTableLookupExpr`, so enable it here.
     let session_config = SessionConfig::default()
         .with_batch_size(10)
         .set_usize("datafusion.optimizer.hash_join_inlist_pushdown_max_size", 1)
         .set_bool("datafusion.execution.parquet.pushdown_filters", true)
-        .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true);
+        .set_bool("datafusion.optimizer.enable_dynamic_filter_pushdown", true)
+        .set_bool(
+            "datafusion.optimizer.enable_hash_join_dynamic_membership_filter",
+            true,
+        );
     let plan = FilterPushdown::new_post_optimization()
         .optimize(plan, session_config.options())
         .unwrap();