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/08/31 09:13:58 UTC
(datafusion) branch gh-readonly-queue/main/pr-24689-180c461a977dda4d2d15f54a98c88092674367ad created (now e5f82fcc5e)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24689-180c461a977dda4d2d15f54a98c88092674367ad
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at e5f82fcc5e fix: record `join_time` and output metrics in PiecewiseMergeJoin (#24689)
This branch includes the following new commits:
new e5f82fcc5e fix: record `join_time` and output metrics in PiecewiseMergeJoin (#24689)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: record `join_time` and output metrics in PiecewiseMergeJoin (#24689)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24689-180c461a977dda4d2d15f54a98c88092674367ad
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit e5f82fcc5ec0ca17a0d71be29dab9fc57e3a470c
Author: Burak Şen <bu...@gmail.com>
AuthorDate: Mon Aug 31 09:13:34 2026 +0000
fix: record `join_time` and output metrics in PiecewiseMergeJoin (#24689)
## Which issue does this PR close?
- Closes N/A
## Rationale for this change
`EXPLAIN ANALYZE` on a `PiecewiseMergeJoin` always reports `join_time=0`
and, for classic joins, `output_rows=0` / `output_bytes=0.0 B`, so PWMJ
plans can't be profiled like other joins.
## What changes are included in this PR?
- Classic stream: time probe-phase work under `join_time` (started only
after the streamed input's poll returns `Ready`) and record output
metrics via `record_poll`
- Existence stream: same `join_time` coverage for the scan and final
pass
- `elapsed_compute` picks both up via the existing
`BuildProbeJoinMetrics` drop
## Are these changes tested?
Yes. Classic tests now assert `assert_join_metrics!`, the existence
final-pass test pins `join_time > 0`, and a new test verifies
`join_time` excludes streamed-input wait. `explain_analyze.slt` shows
the real `output_bytes`.
## Are there any user-facing changes?
No (PWMJ is experimental and off by default).
---
.../src/joins/piecewise_merge_join/classic_join.rs | 245 +++++++++++++++++++--
.../joins/piecewise_merge_join/existence_join.rs | 26 ++-
.../sqllogictest/test_files/explain_analyze.slt | 2 +-
3 files changed, 241 insertions(+), 32 deletions(-)
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
index ac1385223f..41da213beb 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/classic_join.rs
@@ -212,7 +212,9 @@ impl ClassicPWMJStream {
&mut self,
cx: &mut std::task::Context<'_>,
) -> Poll<Result<StatefulStreamResult<Option<RecordBatch>>>> {
- match ready!(self.streamed.poll_next_unpin(cx)) {
+ let next_batch = ready!(self.streamed.poll_next_unpin(cx));
+ let _join_timer = self.join_metrics.join_time.timer();
+ match next_batch {
None => {
// Release the streamed input pipeline's resources.
let streamed_schema = self.streamed.schema();
@@ -267,6 +269,7 @@ impl ClassicPWMJStream {
fn process_stream_batch(
&mut self,
) -> Result<StatefulStreamResult<Option<RecordBatch>>> {
+ let _join_timer = self.join_metrics.join_time.timer();
let buffered_side = self.buffered_side.try_as_ready_mut()?;
let stream_batch = self.state.try_as_process_stream_batch_mut()?;
@@ -313,6 +316,7 @@ impl ClassicPWMJStream {
fn process_unmatched_buffered_batch(
&mut self,
) -> Result<StatefulStreamResult<Option<RecordBatch>>> {
+ let _join_timer = self.join_metrics.join_time.timer();
// Return early for `JoinType::Right` and `JoinType::Inner`
if matches!(self.join_type, JoinType::Right | JoinType::Inner) {
self.state = PiecewiseMergeJoinStreamState::Completed;
@@ -425,7 +429,8 @@ impl Stream for ClassicPWMJStream {
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> Poll<Option<Self::Item>> {
- self.poll_next_impl(cx)
+ let poll = self.poll_next_impl(cx);
+ self.join_metrics.baseline.record_poll(poll)
}
}
@@ -657,19 +662,24 @@ fn create_unmatched_batch(
mod tests {
use super::*;
use crate::{
- ExecutionPlan, common,
+ DisplayAs, DisplayFormatType, ExecutionPlan, PlanProperties, common,
joins::PiecewiseMergeJoinExec,
- test::{TestMemoryExec, build_table_i32},
+ metrics::MetricsSet,
+ stream::RecordBatchStreamAdapter,
+ test::{TestMemoryExec, assert_join_metrics, build_table_i32},
};
use arrow::array::{Date32Array, Date64Array};
use arrow_schema::{DataType, Field};
+ use datafusion_common::instant::Instant;
use datafusion_common::test_util::batches_to_string;
+ use datafusion_common::tree_node::TreeNodeRecursion;
use datafusion_execution::TaskContext;
use datafusion_execution::config::SessionConfig;
use datafusion_physical_expr::{PhysicalExpr, expressions::Column};
use futures::TryStreamExt;
use insta::assert_snapshot;
use std::sync::Arc;
+ use std::time::Duration;
fn columns(schema: &Schema) -> Vec<String> {
schema.fields().iter().map(|f| f.name().clone()).collect()
@@ -751,7 +761,7 @@ mod tests {
on: (PhysicalExprRef, PhysicalExprRef),
operator: Operator,
join_type: JoinType,
- ) -> Result<(Vec<String>, Vec<RecordBatch>)> {
+ ) -> Result<(Vec<String>, Vec<RecordBatch>, MetricsSet)> {
join_collect_with_options(left, right, on, operator, join_type).await
}
@@ -761,14 +771,15 @@ mod tests {
on: (PhysicalExprRef, PhysicalExprRef),
operator: Operator,
join_type: JoinType,
- ) -> Result<(Vec<String>, Vec<RecordBatch>)> {
+ ) -> Result<(Vec<String>, Vec<RecordBatch>, MetricsSet)> {
let task_ctx = Arc::new(TaskContext::default());
let join = join(left, right, on, operator, join_type)?;
let columns = columns(&join.schema());
let stream = join.execute(0, task_ctx)?;
let batches = common::collect(stream).await?;
- Ok((columns, batches))
+ let metrics = join.metrics().expect("metrics should be available");
+ Ok((columns, batches, metrics))
}
#[tokio::test]
@@ -804,7 +815,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Lt, JoinType::Inner).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -819,6 +830,8 @@ mod tests {
| 3 | 1 | 9 | 10 | 2 | 70 |
+----+----+----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 6);
Ok(())
}
@@ -855,7 +868,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Lt, JoinType::Inner).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -870,6 +883,8 @@ mod tests {
| 3 | 1 | 9 | 20 | 2 | 80 |
+----+----+----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 6);
Ok(())
}
@@ -906,7 +921,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::GtEq, JoinType::Inner).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -923,6 +938,8 @@ mod tests {
| 3 | 4 | 9 | 10 | 3 | 70 |
+----+----+----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 8);
Ok(())
}
@@ -955,10 +972,12 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::LtEq, JoinType::Inner).await?;
// An empty join result produces no batches at all, not an empty batch.
assert!(batches.is_empty());
+
+ assert_join_metrics!(metrics, 0);
Ok(())
}
@@ -993,7 +1012,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::GtEq, JoinType::Full).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -1006,6 +1025,7 @@ mod tests {
+----+----+-----+----+----+-----+
");
+ assert_join_metrics!(metrics, 3);
Ok(())
}
@@ -1042,7 +1062,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Gt, JoinType::Left).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -1057,6 +1077,8 @@ mod tests {
| 1 | 1 | 7 | | | |
+----+----+----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 6);
Ok(())
}
@@ -1093,7 +1115,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Gt, JoinType::Right).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -1106,6 +1128,8 @@ mod tests {
| | | | 10 | 5 | 70 |
+----+----+----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 4);
Ok(())
}
@@ -1142,7 +1166,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Lt, JoinType::Right).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -1156,6 +1180,8 @@ mod tests {
| 3 | 1 | 9 | 10 | 2 | 70 |
+----+----+----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 5);
Ok(())
}
@@ -1192,7 +1218,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::LtEq, JoinType::Inner).await?;
// Expected grouping follows right.b1 descending (4, 3, 2)
@@ -1207,6 +1233,8 @@ mod tests {
| 3 | 2 | 9 | 30 | 2 | 90 |
+----+----+----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 5);
Ok(())
}
@@ -1243,7 +1271,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Gt, JoinType::Inner).await?;
// Grouped by right in ascending evaluation for > (1,2,3)
@@ -1257,6 +1285,8 @@ mod tests {
| 3 | 4 | 9 | 10 | 3 | 70 |
+----+----+----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 4);
Ok(())
}
@@ -1356,6 +1386,16 @@ mod tests {
| | | | 20 | 5 | 80 |
+----+----+----+----+----+----+
");
+
+ let metrics = join.metrics().expect("metrics should be available");
+ assert_join_metrics!(metrics, 2);
+ assert!(
+ metrics
+ .sum_by_name("join_time")
+ .expect("join_time metric")
+ .as_usize()
+ > 0
+ );
Ok(())
}
@@ -1374,7 +1414,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Lt, JoinType::Inner).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -1384,6 +1424,8 @@ mod tests {
| 42 | 5 | 999 | 30 | 7 | 90 |
+----+----+-----+----+----+----+
");
+
+ assert_join_metrics!(metrics, 1);
Ok(())
}
@@ -1406,11 +1448,13 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Gt, JoinType::Inner).await?;
// An empty join result produces no batches at all, not an empty batch.
assert!(batches.is_empty());
+
+ assert_join_metrics!(metrics, 0);
Ok(())
}
@@ -1447,7 +1491,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Lt, JoinType::Inner).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -1457,6 +1501,8 @@ mod tests {
| 1970-01-04 | 2022-04-23 | 1970-01-10 | 1970-01-31 | 2022-04-25 | 1970-04-01 |
+------------+------------+------------+------------+------------+------------+
");
+
+ assert_join_metrics!(metrics, 1);
Ok(())
}
@@ -1493,7 +1539,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Lt, JoinType::Inner).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -1503,6 +1549,8 @@ mod tests {
| 1970-01-01T00:00:00.003 | 2022-04-23T08:44:01 | 1970-01-01T00:00:00.009 | 1970-01-01T00:00:00.030 | 2022-04-25T16:17:21 | 1970-01-01T00:00:00.090 |
+-------------------------+---------------------+-------------------------+-------------------------+---------------------+-------------------------+
");
+
+ assert_join_metrics!(metrics, 1);
Ok(())
}
@@ -1537,7 +1585,7 @@ mod tests {
Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
);
- let (_, batches) =
+ let (_, batches, metrics) =
join_collect(left, right, on, Operator::Lt, JoinType::Right).await?;
assert_snapshot!(batches_to_string(&batches), @r"
@@ -1548,6 +1596,159 @@ mod tests {
| | | | 1970-01-01T00:00:00.010 | 2022-04-23T08:44:01 | 1970-01-01T00:00:00.080 |
+-------------------------+---------------------+-------------------------+-------------------------+---------------------+-------------------------+
");
+
+ assert_join_metrics!(metrics, 2);
Ok(())
}
+
+ /// Wraps an input plan and sleeps `delay` before yielding each of its batches, to
+ /// simulate a slow streamed input.
+ #[derive(Debug)]
+ struct DelayedExec {
+ input: Arc<dyn ExecutionPlan>,
+ delay: Duration,
+ }
+
+ impl DisplayAs for DelayedExec {
+ fn fmt_as(
+ &self,
+ _t: DisplayFormatType,
+ f: &mut std::fmt::Formatter,
+ ) -> std::fmt::Result {
+ write!(f, "DelayedExec")
+ }
+ }
+
+ impl ExecutionPlan for DelayedExec {
+ fn name(&self) -> &str {
+ "DelayedExec"
+ }
+
+ fn properties(&self) -> &Arc<PlanProperties> {
+ self.input.properties()
+ }
+
+ fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+ vec![&self.input]
+ }
+
+ fn apply_expressions(
+ &self,
+ _f: &mut dyn FnMut(&Arc<dyn PhysicalExpr>) -> Result<TreeNodeRecursion>,
+ ) -> Result<TreeNodeRecursion> {
+ Ok(TreeNodeRecursion::Continue)
+ }
+
+ fn with_new_children(
+ self: Arc<Self>,
+ children: Vec<Arc<dyn ExecutionPlan>>,
+ ) -> Result<Arc<dyn ExecutionPlan>> {
+ Ok(Arc::new(Self {
+ input: Arc::clone(&children[0]),
+ delay: self.delay,
+ }))
+ }
+
+ fn execute(
+ &self,
+ partition: usize,
+ context: Arc<TaskContext>,
+ ) -> Result<SendableRecordBatchStream> {
+ let stream = self.input.execute(partition, context)?;
+ let schema = stream.schema();
+ let delay = self.delay;
+ Ok(Box::pin(RecordBatchStreamAdapter::new(
+ schema,
+ stream.then(move |item| async move {
+ tokio::time::sleep(delay).await;
+ item
+ }),
+ )))
+ }
+ }
+
+ fn join_time_of(metrics: &MetricsSet) -> Duration {
+ Duration::from_nanos(
+ metrics
+ .sum_by_name("join_time")
+ .map(|m| m.as_usize())
+ .unwrap_or(0) as u64,
+ )
+ }
+
+ /// `join_time` must not include time spent waiting for the streamed input: the timer
+ /// in `fetch_stream_batch` starts only once the input's poll returns `Ready`.
+ ///
+ /// Retries with 4x the delay (up to 3 attempts) when the `join_time < delay` check
+ /// fails. This de-flakes the check without masking real bugs: a genuine exclusion bug
+ /// makes `join_time` absorb the injected waits, so it scales with the delay and fails
+ /// at every escalation level, while a fixed-size disturbance (e.g. the OS preempting
+ /// the thread while a `join_time` timer is running) cannot grow 4x with it.
+ /// Deterministic invariants (row count, wall-time lower bound) are asserted on every
+ /// run and never retried. Mirrors the sort-merge join's `check_join_time_excluded`.
+ #[tokio::test]
+ async fn join_time_excludes_streamed_input_wait() -> Result<()> {
+ let mut delay = Duration::from_millis(50);
+ for attempt in 0..3 {
+ // Same data as `join_inner_less_than`, with the streamed side split into
+ // three batches so each one incurs the injected delay.
+ let left = build_table(
+ ("a1", &vec![1, 2, 3]),
+ ("b1", &vec![3, 2, 1]),
+ ("c1", &vec![7, 8, 9]),
+ );
+
+ let streamed_schema = Schema::new(vec![
+ Field::new("a2", DataType::Int32, false),
+ Field::new("b1", DataType::Int32, false),
+ Field::new("c2", DataType::Int32, false),
+ ]);
+ let streamed_batches = vec![
+ build_table_i32(("a2", &vec![10]), ("b1", &vec![2]), ("c2", &vec![70])),
+ build_table_i32(("a2", &vec![20]), ("b1", &vec![3]), ("c2", &vec![80])),
+ build_table_i32(("a2", &vec![30]), ("b1", &vec![4]), ("c2", &vec![90])),
+ ];
+ let right_mem = TestMemoryExec::try_new_exec(
+ &[streamed_batches],
+ Arc::new(streamed_schema),
+ None,
+ )?;
+ let right = Arc::new(DelayedExec {
+ input: right_mem,
+ delay,
+ });
+
+ let on = (
+ Arc::new(Column::new_with_schema("b1", &left.schema())?) as _,
+ Arc::new(Column::new_with_schema("b1", &right.schema())?) as _,
+ );
+ let join = join(left, right, on, Operator::Lt, JoinType::Inner)?;
+
+ let start = Instant::now();
+ let batches =
+ common::collect(join.execute(0, Arc::new(TaskContext::default()))?)
+ .await?;
+ let wall = start.elapsed();
+
+ let rows: usize = batches.iter().map(|b| b.num_rows()).sum();
+ assert_eq!(rows, 6, "all streamed rows should find their matches");
+ assert!(
+ wall >= delay * 3,
+ "streamed delays should dominate wall time, got {wall:?}"
+ );
+
+ let join_time =
+ join_time_of(&join.metrics().expect("metrics should be available"));
+ if join_time < delay {
+ return Ok(());
+ }
+ assert!(
+ attempt < 2,
+ "join_time ({join_time:?}) should be well below the injected \
+ delay ({delay:?}) even after escalating retries; wall {wall:?}"
+ );
+ delay *= 4;
+ }
+ unreachable!()
+ }
}
diff --git a/datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs b/datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs
index 887367b22e..85eaf4b602 100644
--- a/datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs
+++ b/datafusion/physical-plan/src/joins/piecewise_merge_join/existence_join.rs
@@ -227,7 +227,10 @@ impl ExistencePWMJStream {
return Poll::Ready(Ok(StatefulStreamResult::Continue));
}
- match ready!(self.streamed.poll_next_unpin(cx)) {
+ let next_batch = ready!(self.streamed.poll_next_unpin(cx));
+ let join_time = self.join_metrics.join_time.clone();
+ let _join_timer = join_time.timer();
+ match next_batch {
None => self.finish_streamed_side()?,
Some(Ok(batch)) => {
let stream_values: ArrayRef = self
@@ -387,6 +390,7 @@ impl ExistencePWMJStream {
/// Emits the existence result by slicing at the watermark: the marked buffered rows for
/// `LeftSemi`, the unmarked ones for `LeftAnti`.
fn emit_matched(&mut self) -> Result<StatefulStreamResult<Option<RecordBatch>>> {
+ let _join_timer = self.join_metrics.join_time.timer();
if !self.emitted {
self.emitted = true;
@@ -483,7 +487,7 @@ mod tests {
use crate::{
ExecutionPlan, common,
joins::PiecewiseMergeJoinExec,
- test::{TestMemoryExec, build_table_i32},
+ test::{TestMemoryExec, assert_join_metrics, build_table_i32},
};
use arrow_schema::{DataType, Field, Schema};
use datafusion_common::test_util::batches_to_string;
@@ -636,7 +640,8 @@ mod tests {
/// The final pass pushes one slice of the buffered batch into a `BatchCoalescer`, so a
/// result wider than `batch_size` has to be drained over several polls. Also pins the
- /// `output_rows` metric, which `record_poll` in `poll_next` is what supplies.
+ /// `output_rows` metric, which `record_poll` in `poll_next` is what supplies, and
+ /// `join_time`, which the streamed-side scan and the final pass record.
#[tokio::test]
async fn final_pass_chunks_output_and_records_output_rows() -> Result<()> {
let left = build_table(
@@ -683,12 +688,15 @@ mod tests {
assert_eq!(batches.len(), 2, "expected the final pass to be chunked");
assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 4);
- let output_rows = join
- .metrics()
- .unwrap()
- .output_rows()
- .expect("output_rows metric");
- assert_eq!(output_rows, 4);
+ let metrics = join.metrics().expect("metrics should be available");
+ assert_join_metrics!(metrics, 4);
+ assert!(
+ metrics
+ .sum_by_name("join_time")
+ .expect("join_time metric")
+ .as_usize()
+ > 0
+ );
Ok(())
}
diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt
index e083e8be33..81b274d6c1 100644
--- a/datafusion/sqllogictest/test_files/explain_analyze.slt
+++ b/datafusion/sqllogictest/test_files/explain_analyze.slt
@@ -422,7 +422,7 @@ FROM t1
JOIN t2 ON t1.k > t2.k;
----
Plan with Metrics
-01)PiecewiseMergeJoin: operator=Gt, join_type=Inner, on=(k > k), metrics=[output_bytes=0.0 B, build_mem_used=144.0 B]
+01)PiecewiseMergeJoin: operator=Gt, join_type=Inner, on=(k > k), metrics=[output_bytes=128.0 KB, build_mem_used=144.0 B]
02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=16.0 B]
03)----SortExec: expr=[column1@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=16.0 B, spilled_bytes=0.0 B]
04)------DataSourceExec: partitions=1, partition_sizes=[1], metrics=[]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org