fix(hash-join): measure only the work of a probe row without a match

A row that a dynamic filter removes is a probe row without a match. Its
saving is the work that such a row gets: the evaluation and the hashes of
the join keys and the hash table lookup. The check of the candidates
(`equal_rows_arr`) and the output indices are work for the matches only.
While the filter is on, most probe rows that reach the join are matches,
thus this work made the measured saving too large.

PR: #25681

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

Author: adriangb

datafusion/physical-expr/src/filter_stats.rs

@@ -170,9 +170,13 @@ impl FilterCost {
 /// probe row and looks them up in its hash table. A row that the dynamic
 /// filter of the join removes in the scan does not get this work, thus this
 /// is the saving of a removed row. The work that depends on a match (the
-/// output of a matched row) is not in it: the filter does not remove
-/// matched rows. The producer of a dynamic filter measures it and the
-/// consumers of the filter read it (see
+/// check of the candidates of the lookup and the output of a matched row)
+/// is not in it: the filter removes only rows without a match. While the
+/// filter is on, most rows that reach the producer are matches, thus work
+/// that only matches get would make the saving too large (TPC-DS SF1 Q31:
+/// 6 ns for each probe row with the check, 0.1 to 1.2 ns without it). The
+/// producer of a dynamic filter measures it and the consumers of the filter
+/// read it (see
 /// [`DynamicFilterPhysicalExpr::removed_row_work`]).
 ///
 /// Lock-free.

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

@@ -389,8 +389,10 @@ pub(super) struct HashJoinStream {
     /// 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.
+    /// before the join does not get this work. The check of the candidates
+    /// of the lookup and the output are not in it: only matches get them,
+    /// and while the filter is on, most probe rows are matches. Empty
+    /// without dynamic filter pushdown.
     removed_row_work: Vec<Arc<RemovedRowWork>>,
     /// Scratch space for probe indices during hash lookup
     probe_indices_buffer: Vec<u32>,
@@ -425,6 +427,18 @@ impl RecordBatchStream for HashJoinStream {
     }
 }
 
+/// Records the time since `start` in `works` (see the `removed_row_work`
+/// field of [`HashJoinStream`]), without rows: the rows of a probe batch are
+/// recorded once, when the batch arrives.
+fn record_lookup_work(works: &[Arc<RemovedRowWork>], start: Option<Instant>) {
+    if let Some(start) = start {
+        let nanos = duration_nanos(start.elapsed());
+        for work in works {
+            work.record(0, nanos);
+        }
+    }
+}
+
 /// Executes lookups by hash against JoinHashMap and resolves potential
 /// hash collisions.
 /// Returns build/probe indices satisfying the equality condition, along with
@@ -494,7 +508,27 @@ pub(super) fn lookup_join_hashmap(
         probe_indices_buffer,
         build_indices_buffer,
     );
+    let (build_indices, probe_indices) = equal_candidates(
+        build_side_values,
+        probe_side_values,
+        null_equality,
+        probe_indices_buffer,
+        build_indices_buffer,
+    )?;
+    Ok((build_indices, probe_indices, next_offset))
+}
 
+/// Keeps the candidates of a hash table lookup (the indices in
+/// `probe_indices_buffer` and `build_indices_buffer`) whose join keys are
+/// equal, and returns their indices. Only the probe rows with a candidate
+/// (a match or a hash collision) get this work.
+fn equal_candidates(
+    build_side_values: &[ArrayRef],
+    probe_side_values: &[ArrayRef],
+    null_equality: NullEquality,
+    probe_indices_buffer: &mut Vec<u32>,
+    build_indices_buffer: &mut Vec<u64>,
+) -> Result<(UInt64Array, UInt32Array)> {
     let build_indices_unfiltered: UInt64Array =
         std::mem::take(build_indices_buffer).into();
     let probe_indices_unfiltered: UInt32Array =
@@ -514,7 +548,7 @@ pub(super) fn lookup_join_hashmap(
     *build_indices_buffer = build_indices_unfiltered.into_parts().1.into();
     *probe_indices_buffer = probe_indices_unfiltered.into_parts().1.into();
 
-    Ok((build_indices, probe_indices, next_offset))
+    Ok((build_indices, probe_indices))
 }
 
 /// Counts the number of distinct elements in the input array.
@@ -916,22 +950,33 @@ impl HashJoinStream {
             return Ok(StatefulStreamResult::Continue);
         }
 
-        // get the matched by join keys indices
+        // get the matched by join keys indices. The lookup of the hashes is
+        // the work that every probe row gets, and a dynamic filter saves it
+        // for each row that it removes: its time goes to `removed_row_work`.
+        // The check and the output of the candidates is work for the matches
+        // only, which a dynamic filter does not remove.
         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(
-                map.as_ref(),
-                build_side.left_data.values(),
-                &state.values,
-                self.null_equality,
-                &self.hashes_buffer,
-                state.valid_keys.as_ref(),
-                self.batch_size,
-                state.offset,
-                &mut self.probe_indices_buffer,
-                &mut self.build_indices_buffer,
-            )?,
+            Map::HashMap(map) => {
+                let next_offset = map.get_matched_indices_with_limit_offset(
+                    &self.hashes_buffer,
+                    state.valid_keys.as_ref(),
+                    self.batch_size,
+                    state.offset,
+                    &mut self.probe_indices_buffer,
+                    &mut self.build_indices_buffer,
+                );
+                record_lookup_work(&self.removed_row_work, work_start);
+                let (left_indices, right_indices) = equal_candidates(
+                    build_side.left_data.values(),
+                    &state.values,
+                    self.null_equality,
+                    &mut self.probe_indices_buffer,
+                    &mut self.build_indices_buffer,
+                )?;
+                (left_indices, right_indices, next_offset)
+            }
             Map::ArrayMap(array_map) => {
                 let next_offset = array_map.get_matched_indices_with_limit_offset(
                     &state.values,
@@ -940,19 +985,14 @@ impl HashJoinStream {
                     &mut self.probe_indices_buffer,
                     &mut self.build_indices_buffer,
                 )?;
+                record_lookup_work(&self.removed_row_work, work_start);
                 (
                     UInt64Array::from(self.build_indices_buffer.clone()),
                     UInt32Array::from(self.probe_indices_buffer.clone()),
                     next_offset,
                 )
             }
         };
-
-        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