You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/01 10:37:51 UTC
(datafusion) branch main updated: fix: avoid double-counting build_time in HashJoinExec elapsed_compute (#24142)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 739974b275 fix: avoid double-counting build_time in HashJoinExec elapsed_compute (#24142)
739974b275 is described below
commit 739974b275d2cd1fe65001287506b6983f80db3e
Author: ulrichurriola-parada-ops <ul...@pigment.com>
AuthorDate: Tue Sep 1 10:08:24 2026 +0000
fix: avoid double-counting build_time in HashJoinExec elapsed_compute (#24142)
## Rationale for this change
`HashJoinExec`'s `elapsed_compute` metric (visible via `EXPLAIN
ANALYZE`, `.metrics()`, and any tooling built on top of them) overstates
the actual compute time for the join, sometimes substantially (observed
a +47% higher than `build_time + join_time` combined, even though
`elapsed_compute` is documented/expected to equal exactly that sum).
This misleads anyone using the metric to diagnose slow joins or to
attribute CPU time within a query plan.
## What changes are included in this PR?
`BuildProbeJoinMetrics` derives `Clone` and previously updated
`elapsed_compute` from its own `Drop` impl. `HashJoinExec::execute()`
clones this struct into the build-side future (`collect_left_input`)
while the original is kept by `HashJoinStream`, so two instances exist
for the lifetime of a join, and `Drop` fired once per instance —
double-counting `build_time` whenever the build side yields control
(e.g. returns `Pending`) at least once before completing, which is the
common case for anything other than an already-materialized,
single-batch build side.
This PR moves the `elapsed_compute` update into a new
`ElapsedComputeFinalizer`, held behind an `Arc` on
`BuildProbeJoinMetrics`. Cloning the outer struct now only bumps the
`Arc`'s reference count instead of creating a second independent `Drop`,
so the update runs exactly once — when the last clone is dropped — using
the fully-accumulated final values of `build_time` and `join_time`.
## Are these changes tested?
Yes. Added
`build_probe_join_metrics_elapsed_compute_not_double_counted_on_clone`,
which reproduces the clone/drop sequence directly (without needing a
full hash join) and asserts `elapsed_compute == build_time + join_time`
exactly.
Ran the full existing `joins::` test module (`cargo test -p
datafusion-physical-plan --lib joins::`): 1023 passed, 0 failed, no
regressions.
## Are there any user-facing changes?
`HashJoinExec`'s `elapsed_compute` metric will report lower (correct)
values than before for any join whose build side isn't already fully
materialized in a single poll — most visible in `EXPLAIN ANALYZE` output
and any plan-visualization tooling built on DataFusion's metrics. No
public API changes; `BuildProbeJoinMetrics` is a private (`pub(crate)`)
type.
---------
Co-authored-by: Siew Kam Onn <ko...@gmail.com>
---
datafusion/physical-plan/src/joins/utils.rs | 75 +++++++++++++++++++++++++++--
1 file changed, 72 insertions(+), 3 deletions(-)
diff --git a/datafusion/physical-plan/src/joins/utils.rs b/datafusion/physical-plan/src/joins/utils.rs
index 4702493f4e..4f98b649e6 100644
--- a/datafusion/physical-plan/src/joins/utils.rs
+++ b/datafusion/physical-plan/src/joins/utils.rs
@@ -1777,10 +1777,28 @@ pub(crate) struct BuildProbeJoinMetrics {
/// Average number of build-side join-key matches per matched probe row before
/// applying any join filter
pub(crate) avg_fanout: metrics::RatioMetrics,
+ /// Ensures the `elapsed_compute` update below happens exactly once, no
+ /// matter how many clones of `BuildProbeJoinMetrics` exist (see
+ /// `ElapsedComputeFinalizer` for why this is needed).
+ _elapsed_compute_finalizer: Arc<ElapsedComputeFinalizer>,
+}
+
+// Upon being dropped, this will update the "elapsed compute" part of the metrics.
+// Wrapped in an `Arc` so that it is only dropped once (we previously had a bug
+// where it was cloned, causing over-estimated metrics).
+struct ElapsedComputeFinalizer {
+ baseline: BaselineMetrics,
+ build_time: metrics::Time,
+ join_time: metrics::Time,
+}
+
+// Define Debug impl for ElapsedComputeFinalizer directly to avoid duplicates fields
+impl Debug for ElapsedComputeFinalizer {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("ElapsedComputeFinalizer").finish()
+ }
}
-// This Drop implementation updates the elapsed compute part of the metrics.
-//
// Why is this in a Drop?
// - We keep track of build_time and join_time separately, but baseline metrics have
// a total elapsed_compute time. Instead of remembering to update both the metrics
@@ -1791,7 +1809,7 @@ pub(crate) struct BuildProbeJoinMetrics {
// - The elapsed_compute `Time` is represented by an `Arc<AtomicUsize>`. So even when
// this `BuildProbeJoinMetrics` is dropped, the elapsed_compute is usable through the
// Arc reference.
-impl Drop for BuildProbeJoinMetrics {
+impl Drop for ElapsedComputeFinalizer {
fn drop(&mut self) {
self.baseline.elapsed_compute().add(&self.build_time);
self.baseline.elapsed_compute().add(&self.join_time);
@@ -1833,6 +1851,12 @@ impl BuildProbeJoinMetrics {
.with_type(MetricType::Summary)
.ratio_metrics("avg_fanout", partition);
+ let elapsed_compute_finalizer = Arc::new(ElapsedComputeFinalizer {
+ baseline: baseline.clone(),
+ build_time: build_time.clone(),
+ join_time: join_time.clone(),
+ });
+
Self {
baseline,
build_time,
@@ -1844,6 +1868,7 @@ impl BuildProbeJoinMetrics {
input_rows,
probe_hit_rate,
avg_fanout,
+ _elapsed_compute_finalizer: elapsed_compute_finalizer,
}
}
}
@@ -2572,8 +2597,10 @@ pub fn compare_join_arrays(
mod tests {
use std::collections::HashMap;
use std::pin::Pin;
+ use std::time::Duration;
use super::*;
+ use crate::metrics::MetricValue;
use arrow::datatypes::{DataType, Fields};
use arrow::error::{ArrowError, Result as ArrowResult};
@@ -2752,6 +2779,48 @@ mod tests {
check_join_set_is_valid(&left, &right, on)
}
+ #[test]
+ fn build_probe_join_metrics_elapsed_compute_not_double_counted_on_clone() {
+ // `BuildProbeJoinMetrics` is cloned into the build-side future
+ // (`collect_left_input`) while the original stays with the
+ // `HashJoinStream` (see `HashJoinExec::execute`). Reproduce that
+ // shape here without spinning up a full hash join: accrue build_time
+ // before the clone is dropped (mirroring a build side that yields
+ // control at least once before completing), then accrue the rest of
+ // build_time plus join_time before the original is dropped.
+ let metrics_set = ExecutionPlanMetricsSet::new();
+ let join_metrics = BuildProbeJoinMetrics::new(0, &metrics_set);
+
+ let build_side_clone = join_metrics.clone();
+ join_metrics
+ .build_time
+ .add_duration(Duration::from_millis(100));
+ drop(build_side_clone);
+
+ join_metrics
+ .build_time
+ .add_duration(Duration::from_millis(50));
+ join_metrics
+ .join_time
+ .add_duration(Duration::from_millis(20));
+ drop(join_metrics);
+
+ let elapsed_compute = metrics_set
+ .clone_inner()
+ .iter()
+ .find_map(|m| match m.value() {
+ MetricValue::ElapsedCompute(time) => Some(time.value()),
+ _ => None,
+ })
+ .expect("elapsed_compute metric should be present");
+ let expected = Duration::from_millis(100 + 50 + 20).as_nanos() as usize;
+ assert_eq!(
+ elapsed_compute, expected,
+ "elapsed_compute should equal build_time + join_time exactly once, \
+ not once per BuildProbeJoinMetrics clone"
+ );
+ }
+
#[test]
fn check_valid() -> Result<()> {
let left = vec![Column::new("a", 0), Column::new("b1", 1)];
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org