feat: BufferExec waits for hash join dynamic filters before buffering

Adds ProducerKind to DynamicFilterPhysicalExpr to distinguish between
hash join (safe to wait), TopK, and Aggregate (would deadlock) producers.

BufferExec now walks its child subtree at execution time, collects any
HashJoin dynamic filters, and waits for them to complete before the
background task begins polling the input stream. This allows scans below
BufferExec to benefit from dynamic filters rather than being eagerly
read before the filter is populated.

A new session config option `hash_join_buffering_dynamic_filter_wait_ms`
controls the wait behavior (0 = disabled, usize::MAX = wait indefinitely,
any other value = timeout in ms). Defaults to usize::MAX.

Closes #20778

Author: LiaCastaneda

datafusion/common/src/config.rs

@@ -684,6 +684,17 @@ config_namespace! {
         ///
         /// Disabled by default, set to a number greater than 0 for enabling it.
         pub hash_join_buffering_capacity: usize, default = 0
+
+        /// How long (in milliseconds) the probe-side buffer should wait for hash join dynamic
+        /// filters to be fully populated before starting to buffer data.
+        ///
+        /// When `hash_join_buffering_capacity` is enabled and dynamic filters are present below
+        /// the buffer, waiting for the filter to complete allows the input scan to benefit from
+        /// the filter and skip reading unnecessary data.
+        ///
+        /// Set to 0 to disable waiting (buffer immediately, same as previous behavior).
+        /// Set to `usize::MAX` to wait indefinitely until the filter is complete.
+        pub hash_join_buffering_dynamic_filter_wait_ms: usize, default = usize::MAX
     }
 }
 

datafusion/core/tests/physical_optimizer/filter_pushdown.rs

@@ -5146,3 +5146,182 @@ async fn test_hashjoin_dynamic_filter_pushdown_left_semi_join() {
     "
     );
 }
+
+/// Helper: build and optimize a HashJoin plan with dynamic filter pushdown + buffering enabled.
+/// Returns the optimized plan ready for execution.
+fn build_buffered_hash_join_plan(
+    build_scan: Arc<dyn ExecutionPlan>,
+    probe_scan: Arc<dyn ExecutionPlan>,
+    build_schema: &SchemaRef,
+    probe_schema: &SchemaRef,
+    join_type: JoinType,
+    config: &ConfigOptions,
+) -> Arc<dyn ExecutionPlan> {
+    use datafusion_common::NullEquality;
+    use datafusion_physical_optimizer::hash_join_buffering::HashJoinBuffering;
+
+    let join = Arc::new(
+        HashJoinExec::try_new(
+            build_scan,
+            probe_scan,
+            vec![(
+                col("a", build_schema).unwrap(),
+                col("a", probe_schema).unwrap(),
+            )],
+            None,
+            &join_type,
+            None,
+            PartitionMode::CollectLeft,
+            NullEquality::NullEqualsNothing,
+            false,
+        )
+        .unwrap(),
+    ) as Arc<dyn ExecutionPlan>;
+
+    // Apply dynamic filter pushdown so the probe scan receives the dynamic filter
+    let plan = FilterPushdown::new_post_optimization()
+        .optimize(join, config)
+        .unwrap();
+
+    // Apply buffering so BufferExec wraps the probe side
+    HashJoinBuffering::new().optimize(plan, config).unwrap()
+}
+
+/// Test that BufferExec correctly waits for a HashJoin dynamic filter before buffering,
+/// and that the query produces correct results.
+#[tokio::test]
+async fn test_buffer_exec_waits_for_hash_join_dynamic_filter() {
+    let build_batches =
+        vec![record_batch!(("a", Utf8, ["aa", "ab"]), ("b", Int32, [1, 2])).unwrap()];
+    let build_schema = Arc::new(Schema::new(vec![
+        Field::new("a", DataType::Utf8, false),
+        Field::new("b", DataType::Int32, false),
+    ]));
+    let build_scan = TestScanBuilder::new(Arc::clone(&build_schema))
+        .with_support(true)
+        .with_batches(build_batches)
+        .build();
+
+    let probe_batches = vec![
+        record_batch!(
+            ("a", Utf8, ["aa", "ab", "ac"]),
+            ("c", Float64, [1.0, 2.0, 3.0])
+        )
+        .unwrap(),
+    ];
+    let probe_schema = Arc::new(Schema::new(vec![
+        Field::new("a", DataType::Utf8, false),
+        Field::new("c", DataType::Float64, false),
+    ]));
+    let probe_scan = TestScanBuilder::new(Arc::clone(&probe_schema))
+        .with_support(true)
+        .with_batches(probe_batches)
+        .build();
+
+    let mut config = ConfigOptions::default();
+    config.optimizer.enable_dynamic_filter_pushdown = true;
+    config.execution.hash_join_buffering_capacity = 1024 * 1024;
+    config.execution.hash_join_buffering_dynamic_filter_wait_ms = usize::MAX;
+
+    let plan = build_buffered_hash_join_plan(
+        build_scan,
+        probe_scan,
+        &build_schema,
+        &probe_schema,
+        JoinType::Inner,
+        &config,
+    );
+
+    let session_config = SessionConfig::from(config);
+    let session_ctx = SessionContext::new_with_config(session_config);
+    session_ctx.register_object_store(
+        ObjectStoreUrl::parse("test://").unwrap().as_ref(),
+        Arc::new(InMemory::new()),
+    );
+    let task_ctx = session_ctx.state().task_ctx();
+    let batches = collect(plan, task_ctx).await.unwrap();
+
+    // "ac" has no match on build side so only "aa" and "ab" rows appear
+    let result = format!("{}", pretty_format_batches(&batches).unwrap());
+    insta::assert_snapshot!(result, @r"
+    +----+---+----+-----+
+    | a  | b | a  | c   |
+    +----+---+----+-----+
+    | aa | 1 | aa | 1.0 |
+    | ab | 2 | ab | 2.0 |
+    +----+---+----+-----+
+    ");
+}
+
+/// Test that BufferExec + HashJoin with a TopK above does not deadlock.
+/// The TopK filter is above BufferExec, so it is NOT in the child subtree
+/// and will not be waited on.
+#[tokio::test]
+async fn test_buffer_exec_hash_join_with_topk_above_no_deadlock() {
+    let build_batches = vec![
+        record_batch!(("a", Utf8, ["aa", "ab", "ac"]), ("b", Int32, [1, 2, 3])).unwrap(),
+    ];
+    let build_schema = Arc::new(Schema::new(vec![
+        Field::new("a", DataType::Utf8, false),
+        Field::new("b", DataType::Int32, false),
+    ]));
+    let build_scan = TestScanBuilder::new(Arc::clone(&build_schema))
+        .with_support(true)
+        .with_batches(build_batches)
+        .build();
+
+    let probe_batches = vec![
+        record_batch!(
+            ("a", Utf8, ["aa", "ab", "ac", "ad"]),
+            ("c", Float64, [1.0, 2.0, 3.0, 4.0])
+        )
+        .unwrap(),
+    ];
+    let probe_schema = Arc::new(Schema::new(vec![
+        Field::new("a", DataType::Utf8, false),
+        Field::new("c", DataType::Float64, false),
+    ]));
+    let probe_scan = TestScanBuilder::new(Arc::clone(&probe_schema))
+        .with_support(true)
+        .with_batches(probe_batches)
+        .build();
+
+    let mut config = ConfigOptions::default();
+    config.optimizer.enable_dynamic_filter_pushdown = true;
+    config.execution.hash_join_buffering_capacity = 1024 * 1024;
+    config.execution.hash_join_buffering_dynamic_filter_wait_ms = usize::MAX;
+
+    let plan = build_buffered_hash_join_plan(
+        build_scan,
+        probe_scan,
+        &build_schema,
+        &probe_schema,
+        JoinType::Inner,
+        &config,
+    );
+
+    // Wrap with TopK (SortExec with fetch) above the join
+    let sort_expr =
+        PhysicalSortExpr::new(col("a", &plan.schema()).unwrap(), SortOptions::default());
+    let topk = Arc::new(
+        SortExec::new(LexOrdering::new(vec![sort_expr]).unwrap(), plan)
+            .with_fetch(Some(2)),
+    ) as Arc<dyn ExecutionPlan>;
+
+    let session_config = SessionConfig::from(config);
+    let session_ctx = SessionContext::new_with_config(session_config);
+    session_ctx.register_object_store(
+        ObjectStoreUrl::parse("test://").unwrap().as_ref(),
+        Arc::new(InMemory::new()),
+    );
+    let task_ctx = session_ctx.state().task_ctx();
+
+    // Should complete without deadlock
+    let batches =
+        tokio::time::timeout(std::time::Duration::from_secs(10), collect(topk, task_ctx))
+            .await
+            .expect("should not deadlock")
+            .unwrap();
+
+    assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 2);
+}

datafusion/datasource-parquet/src/opener.rs

@@ -1588,7 +1588,7 @@ mod test {
     use datafusion_expr::{col, lit};
     use datafusion_physical_expr::{
         PhysicalExpr,
-        expressions::{Column, DynamicFilterPhysicalExpr, Literal},
+        expressions::{Column, DynamicFilterPhysicalExpr, Literal, ProducerKind},
         planner::logical2physical,
         projection::ProjectionExprs,
     };
@@ -1922,6 +1922,7 @@ mod test {
         Arc::new(DynamicFilterPhysicalExpr::new(
             expr.children().into_iter().map(Arc::clone).collect(),
             expr,
+            ProducerKind::HashJoin,
         ))
     }
 

datafusion/physical-expr/src/expressions/dynamic_filters.rs

@@ -28,6 +28,26 @@ use datafusion_common::{
 use datafusion_expr::ColumnarValue;
 use datafusion_physical_expr_common::physical_expr::DynHash;
 
+/// Identifies which kind of operator produced a [`DynamicFilterPhysicalExpr`].
+///
+/// This is used by operators like [`BufferExec`] to decide whether it is safe
+/// to wait for a dynamic filter to complete before starting execution.
+/// Only [`ProducerKind::HashJoin`] filters are guaranteed to be fully populated
+/// before the probe side starts, making them safe to wait on.
+/// [`ProducerKind::TopK`] and [`ProducerKind::Aggregate`] filters are populated
+/// incrementally during execution and waiting on them would cause a deadlock.
+///
+/// [`BufferExec`]: datafusion_physical_plan::buffer::BufferExec
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ProducerKind {
+    /// Produced by a hash join build side. Safe to wait on — fully populated before probe starts.
+    HashJoin,
+    /// Produced by a TopK operator. Not safe to wait on — populated incrementally.
+    TopK,
+    /// Produced by an aggregate operator. Not safe to wait on — may never complete.
+    Aggregate,
+}
+
 /// State of a dynamic filter, tracking both updates and completion.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum FilterState {
@@ -55,7 +75,7 @@ impl FilterState {
 /// For more background, please also see the [Dynamic Filters: Passing Information Between Operators During Execution for 25x Faster Queries blog]
 ///
 /// [Dynamic Filters: Passing Information Between Operators During Execution for 25x Faster Queries blog]: https://datafusion.apache.org/blog/2025/09/10/dynamic-filters
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct DynamicFilterPhysicalExpr {
     /// The original children of this PhysicalExpr, if any.
     /// This is necessary because the dynamic filter may be initialized with a placeholder (e.g. `lit(true)`)
@@ -74,6 +94,8 @@ pub struct DynamicFilterPhysicalExpr {
     /// But this can have overhead in production, so it's only included in our tests.
     data_type: Arc<RwLock<Option<DataType>>>,
     nullable: Arc<RwLock<Option<bool>>>,
+    /// The kind of operator that produced this dynamic filter.
+    producer_kind: ProducerKind,
 }
 
 #[derive(Debug)]
@@ -168,6 +190,7 @@ impl DynamicFilterPhysicalExpr {
     pub fn new(
         children: Vec<Arc<dyn PhysicalExpr>>,
         inner: Arc<dyn PhysicalExpr>,
+        producer_kind: ProducerKind,
     ) -> Self {
         let (state_watch, _) = watch::channel(FilterState::InProgress { generation: 1 });
         Self {
@@ -177,6 +200,7 @@ impl DynamicFilterPhysicalExpr {
             state_watch,
             data_type: Arc::new(RwLock::new(None)),
             nullable: Arc::new(RwLock::new(None)),
+            producer_kind,
         }
     }
 
@@ -289,6 +313,11 @@ impl DynamicFilterPhysicalExpr {
         let _ = rx.wait_for(|state| state.generation() > current_gen).await;
     }
 
+    /// Returns the [`ProducerKind`] of this dynamic filter.
+    pub fn producer_kind(&self) -> ProducerKind {
+        self.producer_kind
+    }
+
     /// Wait asynchronously until this dynamic filter is marked as complete.
     ///
     /// This method returns immediately if the filter is already complete.
@@ -372,6 +401,7 @@ impl PhysicalExpr for DynamicFilterPhysicalExpr {
             state_watch: self.state_watch.clone(),
             data_type: Arc::clone(&self.data_type),
             nullable: Arc::clone(&self.nullable),
+            producer_kind: self.producer_kind,
         }))
     }
 
@@ -478,6 +508,7 @@ mod test {
         let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(
             vec![col("a", &table_schema).unwrap()],
             expr as Arc<dyn PhysicalExpr>,
+            ProducerKind::HashJoin,
         ));
         // Simulate two `ParquetSource` files with different filter schemas
         // Both of these should hit the same inner `PhysicalExpr` even after `update()` is called
@@ -632,6 +663,7 @@ mod test {
         let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(
             vec![],
             lit(42) as Arc<dyn PhysicalExpr>,
+            ProducerKind::HashJoin,
         ));
 
         // Mark as complete immediately
@@ -667,6 +699,7 @@ mod test {
         let dynamic_filter = Arc::new(DynamicFilterPhysicalExpr::new(
             vec![Arc::clone(&col_a), Arc::clone(&col_b)],
             expr as Arc<dyn PhysicalExpr>,
+            ProducerKind::HashJoin,
         ));
 
         // Clone the Arc (two references to the same DynamicFilterPhysicalExpr)
@@ -730,6 +763,7 @@ mod test {
         let filter = Arc::new(DynamicFilterPhysicalExpr::new(
             vec![],
             lit(true) as Arc<dyn PhysicalExpr>,
+            ProducerKind::HashJoin,
         ));
 
         // Initially, only one reference to the inner Arc exists

datafusion/physical-expr/src/expressions/mod.rs

@@ -45,7 +45,7 @@ pub use cast::{CastExpr, cast};
 pub use cast_column::CastColumnExpr;
 pub use column::{Column, col, with_new_schema};
 pub use datafusion_expr::utils::format_state_name;
-pub use dynamic_filters::DynamicFilterPhysicalExpr;
+pub use dynamic_filters::{DynamicFilterPhysicalExpr, ProducerKind};
 pub use in_list::{InListExpr, in_list};
 pub use is_not_null::{IsNotNullExpr, is_not_null};
 pub use is_null::{IsNullExpr, is_null};

datafusion/physical-plan/src/aggregates/mod.rs

@@ -52,7 +52,9 @@ use datafusion_execution::TaskContext;
 use datafusion_expr::{Accumulator, Aggregate};
 use datafusion_physical_expr::aggregate::AggregateFunctionExpr;
 use datafusion_physical_expr::equivalence::ProjectionMapping;
-use datafusion_physical_expr::expressions::{Column, DynamicFilterPhysicalExpr, lit};
+use datafusion_physical_expr::expressions::{
+    Column, DynamicFilterPhysicalExpr, ProducerKind, lit,
+};
 use datafusion_physical_expr::{
     ConstExpr, EquivalenceProperties, physical_exprs_contains,
 };
@@ -1167,7 +1169,11 @@ impl AggregateExec {
 
         if !aggr_dyn_filters.is_empty() {
             self.dynamic_filter = Some(Arc::new(AggrDynFilter {
-                filter: Arc::new(DynamicFilterPhysicalExpr::new(all_cols, lit(true))),
+                filter: Arc::new(DynamicFilterPhysicalExpr::new(
+                    all_cols,
+                    lit(true),
+                    ProducerKind::Aggregate,
+                )),
                 supported_accumulators_info: aggr_dyn_filters,
             }))
         }