feat(hash-join): measure the work that a dynamic filter saves for each removed row

The hash join records, in the `RemovedRowWork` of each dynamic filter
that it produces, the rows of each probe batch and the time of the work
that it does for every probe row, match or no match: the evaluation and
the hashes of the join keys and the hash table lookup. A row that the
filter removes before the join does not get this work. The work for a
matched row (the output) is not in it: the filter does not remove
matched rows.

PR: #25681

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: adriangb

datafusion/physical-plan/src/joins/hash_join/exec.rs

@@ -1836,6 +1836,21 @@ impl ExecutionPlan for HashJoinExec {
 
         let batch_size = context.session_config().batch_size();
 
+        // The join measures the work that its dynamic filters save for each
+        // row that they remove (see `RemovedRowWork`).
+        let removed_row_work = self
+            .dynamic_filter
+            .as_ref()
+            .filter(|_| enable_dynamic_filter_pushdown)
+            .map(|df| {
+                df.membership
+                    .iter()
+                    .chain(df.bounds.iter())
+                    .map(|filter| Arc::clone(filter.removed_row_work()))
+                    .collect()
+            })
+            .unwrap_or_default();
+
         // we have the batches and the hash map with their keys. We can how create a stream
         // over the right that uses this information to issue new batches.
         let right_stream = self.right.execute(partition, context)?;
@@ -1855,27 +1870,30 @@ impl ExecutionPlan for HashJoinExec {
             .map(|(_, right_expr)| Arc::clone(right_expr))
             .collect::<Vec<_>>();
 
-        Ok(Box::pin(HashJoinStream::new(
-            partition,
-            self.schema(),
-            on_right,
-            self.filter.clone(),
-            self.join_type,
-            right_stream,
-            self.random_state.random_state().clone(),
-            join_metrics,
-            column_indices_after_projection,
-            self.null_equality,
-            HashJoinStreamState::WaitBuildSide,
-            BuildSide::Initial(BuildSideInitialState { left_fut }),
-            batch_size,
-            vec![],
-            self.right.output_ordering().is_some(),
-            build_accumulator,
-            self.mode,
-            null_aware,
-            self.fetch,
-        )))
+        Ok(Box::pin(
+            HashJoinStream::new(
+                partition,
+                self.schema(),
+                on_right,
+                self.filter.clone(),
+                self.join_type,
+                right_stream,
+                self.random_state.random_state().clone(),
+                join_metrics,
+                column_indices_after_projection,
+                self.null_equality,
+                HashJoinStreamState::WaitBuildSide,
+                BuildSide::Initial(BuildSideInitialState { left_fut }),
+                batch_size,
+                vec![],
+                self.right.output_ordering().is_some(),
+                build_accumulator,
+                self.mode,
+                null_aware,
+                self.fetch,
+            )
+            .with_removed_row_work(removed_row_work),
+        ))
     }
 
     fn metrics(&self) -> Option<MetricsSet> {
@@ -8750,6 +8768,87 @@ mod tests {
         Ok(())
     }
 
+    /// The join measures the work that its dynamic filter saves for each
+    /// row that the filter removes: its work for each probe row.
+    #[tokio::test]
+    async fn test_dynamic_filter_measures_removed_row_work() -> Result<()> {
+        let task_ctx = Arc::new(TaskContext::default());
+        let left = build_table(
+            ("a1", &vec![1, 2, 3]),
+            ("b1", &vec![4, 5, 6]),
+            ("c1", &vec![7, 8, 9]),
+        );
+        // `MIN_OBSERVED_ROWS` probe rows.
+        let rows = datafusion_physical_expr::filter_stats::MIN_OBSERVED_ROWS as i32;
+        let values: Vec<i32> = (0..rows).collect();
+        let right = build_table(("a2", &values), ("b2", &values), ("c2", &values));
+        let on = vec![(
+            Arc::new(Column::new_with_schema("a1", &left.schema())?) as _,
+            Arc::new(Column::new_with_schema("a2", &right.schema())?) as _,
+        )];
+        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
+        let consumer: Arc<dyn PhysicalExpr> = Arc::clone(&dynamic_filter) as _;
+        // The consumer does not apply the filter here: the join sees all
+        // probe rows.
+        let right = Arc::new(FilterExecBuilder::new(consumer, right).build()?);
+        let mut join = HashJoinExec::try_new(
+            left,
+            right,
+            on,
+            None,
+            &JoinType::Inner,
+            None,
+            PartitionMode::CollectLeft,
+            NullEquality::NullEqualsNothing,
+            false,
+        )?;
+        join.dynamic_filter = Some(HashJoinExecDynamicFilter::new(
+            Some(Arc::clone(&dynamic_filter)),
+            None,
+        ));
+        let work = Arc::clone(dynamic_filter.removed_row_work());
+        assert_eq!(work.ns_per_row(), None);
+        let batches = common::collect(join.execute(0, task_ctx)?).await?;
+        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
+        // The filter removed all but 3 probe rows, thus the join saw fewer
+        // than `MIN_OBSERVED_ROWS` rows: no measurement yet.
+        assert_eq!(work.ns_per_row(), None);
+
+        // Without a consumer that filters, the join sees all probe rows.
+        let task_ctx = Arc::new(TaskContext::default());
+        let left = build_table(
+            ("a1", &vec![1, 2, 3]),
+            ("b1", &vec![4, 5, 6]),
+            ("c1", &vec![7, 8, 9]),
+        );
+        let right = build_table(("a2", &values), ("b2", &values), ("c2", &values));
+        let on = vec![(
+            Arc::new(Column::new_with_schema("a1", &left.schema())?) as _,
+            Arc::new(Column::new_with_schema("a2", &right.schema())?) as _,
+        )];
+        let dynamic_filter = HashJoinExec::create_dynamic_filter(&on);
+        let mut join = HashJoinExec::try_new(
+            left,
+            right,
+            on,
+            None,
+            &JoinType::Inner,
+            None,
+            PartitionMode::CollectLeft,
+            NullEquality::NullEqualsNothing,
+            false,
+        )?;
+        join.dynamic_filter = Some(HashJoinExecDynamicFilter::new(
+            Some(Arc::clone(&dynamic_filter)),
+            None,
+        ));
+        let batches = common::collect(join.execute(0, task_ctx)?).await?;
+        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 3);
+        let measured = dynamic_filter.removed_row_work().ns_per_row();
+        assert!(measured.is_some_and(|ns| ns > 0.0), "{measured:?}");
+        Ok(())
+    }
+
     /// The [`PartitionMode::Partitioned`] counterpart of
     /// [`test_null_equal_dynamic_filter_keeps_probe_nulls_for_build_logical_null`].
     ///

datafusion/physical-plan/src/joins/hash_join/stream.rs

@@ -49,10 +49,12 @@ use arrow::array::{Array, ArrayRef, UInt32Array, UInt64Array};
 use arrow::buffer::{BooleanBuffer, NullBuffer};
 use arrow::datatypes::{Schema, SchemaRef};
 use arrow::record_batch::RecordBatch;
+use datafusion_common::instant::Instant;
 use datafusion_common::{
     JoinSide, JoinType, NullEquality, Result, internal_datafusion_err, internal_err,
 };
 use datafusion_physical_expr::PhysicalExprRef;
+use datafusion_physical_expr::filter_stats::{RemovedRowWork, duration_nanos};
 
 use datafusion_common::hash_utils::RandomState;
 use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
@@ -382,6 +384,14 @@ pub(super) struct HashJoinStream {
     batch_size: usize,
     /// Scratch space for computing hashes
     hashes_buffer: Vec<u64>,
+    /// The work measurements of the dynamic filters that this join produces
+    /// (see [`RemovedRowWork`]). The join records the rows of each probe
+    /// batch and the time of the work that it does for a probe row whether
+    /// or not the row matches: the evaluation and the hashes of the join
+    /// keys and the hash table lookup. A row that the dynamic filter removes
+    /// before the join does not get this work. Empty without dynamic filter
+    /// pushdown.
+    removed_row_work: Vec<Arc<RemovedRowWork>>,
     /// Scratch space for probe indices during hash lookup
     probe_indices_buffer: Vec<u32>,
     /// Scratch space for build indices during hash lookup
@@ -585,6 +595,7 @@ impl HashJoinStream {
             null_mark_hashes_buffer: Vec::new(),
             null_mark_probe_indices_buffer: Vec::new(),
             null_mark_build_indices_buffer: Vec::new(),
+            removed_row_work: vec![],
             right_side_ordered,
             build_report: BuildReportHandle::new(partition, mode, build_accumulator),
             mode,
@@ -593,6 +604,24 @@ impl HashJoinStream {
         }
     }
 
+    /// Records the work for each probe row in `works`, see
+    /// the `removed_row_work` field.
+    pub(super) fn with_removed_row_work(
+        mut self,
+        works: Vec<Arc<RemovedRowWork>>,
+    ) -> Self {
+        self.removed_row_work = works;
+        self
+    }
+
+    /// Records `rows` probe rows and `nanos` nanoseconds of work for each
+    /// row in the dynamic filters of this join.
+    fn record_removed_row_work(&self, rows: usize, nanos: u64) {
+        for work in &self.removed_row_work {
+            work.record(rows as u64, nanos);
+        }
+    }
+
     /// Returns the next state after the build side has been fully collected
     /// and any required build-side coordination has completed.
     fn state_after_build_ready(
@@ -779,6 +808,7 @@ impl HashJoinStream {
                 self.state = HashJoinStreamState::ExhaustedProbeSide;
             }
             Some(Ok(batch)) => {
+                let work_start = (!self.removed_row_work.is_empty()).then(Instant::now);
                 // Precalculate hash values for fetched batch
                 let keys_values = evaluate_expressions_to_arrays(&self.on_right, &batch)?;
 
@@ -797,6 +827,12 @@ impl HashJoinStream {
                     None
                 };
 
+                if let Some(work_start) = work_start {
+                    self.record_removed_row_work(
+                        batch.num_rows(),
+                        duration_nanos(work_start.elapsed()),
+                    );
+                }
                 self.join_metrics.input_batches.add(1);
                 self.join_metrics.input_rows.add(batch.num_rows());
 
@@ -881,6 +917,7 @@ impl HashJoinStream {
         }
 
         // get the matched by join keys indices
+        let work_start = (!self.removed_row_work.is_empty()).then(Instant::now);
         let (left_indices, right_indices, next_offset) = match build_side.left_data.map()
         {
             Map::HashMap(map) => lookup_join_hashmap(
@@ -911,6 +948,11 @@ impl HashJoinStream {
             }
         };
 
+        if let Some(work_start) = work_start {
+            for work in &self.removed_row_work {
+                work.record(0, duration_nanos(work_start.elapsed()));
+            }
+        }
         let matched_probe_rows = state.count_new_matched_probe_rows(&right_indices);
 
         self.join_metrics