From 515d79975aa581fadd062ed9a6b9a3336873f15a Mon Sep 17 00:00:00 2001 From: Ariel Miculas Date: Tue, 21 Jul 2026 23:50:21 +0300 Subject: [PATCH 1/2] fix: output_bytes metric in hash aggregation --- .../src/metrics/baseline.rs | 23 +- .../physical-expr-common/src/metrics/mod.rs | 4 +- .../aggregates/aggregate_hash_table/common.rs | 42 +++- .../aggregate_hash_table/final_table.rs | 4 +- .../partial_reduce_table.rs | 4 +- .../aggregate_hash_table/partial_table.rs | 4 +- .../aggregate_hash_table/single_table.rs | 4 +- .../src/aggregates/hash_stream.rs | 235 +++++++++++++++++- .../src/aggregates/partial_reduce_stream.rs | 11 +- .../src/aggregates/single_stream.rs | 11 +- 10 files changed, 306 insertions(+), 36 deletions(-) diff --git a/datafusion/physical-expr-common/src/metrics/baseline.rs b/datafusion/physical-expr-common/src/metrics/baseline.rs index 52ad4aac9fd98..d6f1d4e268bc4 100644 --- a/datafusion/physical-expr-common/src/metrics/baseline.rs +++ b/datafusion/physical-expr-common/src/metrics/baseline.rs @@ -20,7 +20,10 @@ use std::{borrow::Cow, collections::BTreeMap, sync::Arc, task::Poll}; use arrow::record_batch::RecordBatch; -use datafusion_common::{Result, utils::memory::get_record_batch_memory_size}; +use datafusion_common::{ + Result, + utils::memory::{RecordBatchMemoryCounter, get_record_batch_memory_size}, +}; use super::{ Count, ExecutionPlanMetricsSet, Metric, MetricBuilder, MetricsSet, Time, Timestamp, @@ -312,6 +315,24 @@ impl SplitMetrics { } } +#[derive(Debug, Default)] +pub struct RecordBatchMemoryMetrics(RecordBatchMemoryCounter); + +impl RecordBatchMemoryMetrics { + pub fn new() -> Self { + Self::default() + } + + /// Similar to RecordBatch.record_output, but deduplicating accross batches to avoid + /// output_size inflation due to shared memory + pub fn record_output(&mut self, batch: &RecordBatch, bm: &BaselineMetrics) { + bm.record_output(batch.num_rows()); + let n_bytes = self.0.count_batch(batch); + bm.output_bytes.add(n_bytes); + bm.output_batches.add(1); + } +} + /// Trait for things that produce output rows as a result of execution. pub trait RecordOutput { /// Record that some number of output rows have been produced diff --git a/datafusion/physical-expr-common/src/metrics/mod.rs b/datafusion/physical-expr-common/src/metrics/mod.rs index d6048a0fcd338..c5c2624dbc592 100644 --- a/datafusion/physical-expr-common/src/metrics/mod.rs +++ b/datafusion/physical-expr-common/src/metrics/mod.rs @@ -37,7 +37,9 @@ use std::{ // public exports -pub use baseline::{BaselineMetrics, RecordOutput, SpillMetrics, SplitMetrics}; +pub use baseline::{ + BaselineMetrics, RecordBatchMemoryMetrics, RecordOutput, SpillMetrics, SplitMetrics, +}; pub use builder::MetricBuilder; pub use custom::CustomMetricValue; pub use elapsed_compute::{ElapsedComputeFuture, ElapsedComputeFutureExt}; diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs index eaf39929ced62..7b8748dd7027a 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs @@ -33,6 +33,7 @@ use crate::aggregates::order::GroupOrdering; use crate::aggregates::{ AggregateExec, PhysicalGroupBy, aggregate_expressions, evaluate_group_by, }; +use crate::metrics::{BaselineMetrics, RecordBatchMemoryMetrics}; /// Marker for raw rows -> partial state aggregation. pub(in crate::aggregates) struct PartialMarker; @@ -219,6 +220,7 @@ impl AggregateHashTable { pub(super) fn next_output_batch_inner( &mut self, materialize_accumulator_fn: MaterializeAccumulatorFn, + bm: &BaselineMetrics, ) -> Result> { let output_schema = Arc::clone(&self.output_schema); let batch_size = self.batch_size; @@ -253,7 +255,7 @@ impl AggregateHashTable { } }; - let batch = output.next_batch(batch_size); + let batch = output.next_batch(batch_size, bm); if output.is_exhausted() { self.state = AggregateHashTableState::Done; } else { @@ -431,14 +433,26 @@ pub(super) enum AggregateHashTableState { pub(super) struct MaterializedAggregateOutput { batch: RecordBatch, offset: usize, + /// Deduplicates buffer bytes across the slices sliced from `batch`, since + /// they share the same underlying buffers and must only be counted once + /// in `output_bytes`. + record_batch_metrics: RecordBatchMemoryMetrics, } impl MaterializedAggregateOutput { pub(super) fn new(batch: RecordBatch) -> Self { - Self { batch, offset: 0 } + Self { + batch, + offset: 0, + record_batch_metrics: RecordBatchMemoryMetrics::new(), + } } - pub(super) fn next_batch(&mut self, batch_size: usize) -> Option { + pub(super) fn next_batch( + &mut self, + batch_size: usize, + bm: &BaselineMetrics, + ) -> Option { debug_assert!(batch_size > 0); if self.is_exhausted() { return None; @@ -447,6 +461,7 @@ impl MaterializedAggregateOutput { let length = batch_size.min(self.batch.num_rows() - self.offset); let batch = self.batch.slice(self.offset, length); self.offset += length; + self.record_batch_metrics.record_output(&batch, bm); Some(batch) } @@ -633,11 +648,22 @@ mod tests { vec![Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5]))], )?; let mut output = MaterializedAggregateOutput::new(batch); - - assert_eq!(int32_values(&output.next_batch(2).unwrap(), 0), vec![1, 2]); - assert_eq!(int32_values(&output.next_batch(2).unwrap(), 0), vec![3, 4]); - assert_eq!(int32_values(&output.next_batch(2).unwrap(), 0), vec![5]); - assert!(output.next_batch(2).is_none()); + let metrics_set = crate::metrics::ExecutionPlanMetricsSet::new(); + let bm = BaselineMetrics::new(&metrics_set, 0); + + assert_eq!( + int32_values(&output.next_batch(2, &bm).unwrap(), 0), + vec![1, 2] + ); + assert_eq!( + int32_values(&output.next_batch(2, &bm).unwrap(), 0), + vec![3, 4] + ); + assert_eq!( + int32_values(&output.next_batch(2, &bm).unwrap(), 0), + vec![5] + ); + assert!(output.next_batch(2, &bm).is_none()); assert!(output.is_exhausted()); Ok(()) diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/final_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/final_table.rs index 522cc9066b14b..f005405a89bfe 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/final_table.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/final_table.rs @@ -20,6 +20,7 @@ use arrow::record_batch::RecordBatch; use datafusion_common::Result; use crate::aggregates::AggregateExec; +use crate::metrics::BaselineMetrics; use super::common::{AggregateHashTable, FinalMarker, HashAggregateAccumulator}; @@ -54,8 +55,9 @@ impl AggregateHashTable { /// exhausted, and an internal error if polled in the `Building` state. pub(in crate::aggregates) fn next_output_batch( &mut self, + bm: &BaselineMetrics, ) -> Result> { - self.next_output_batch_inner(HashAggregateAccumulator::evaluate_to_columns) + self.next_output_batch_inner(HashAggregateAccumulator::evaluate_to_columns, bm) } /// Final aggregation consumes partial aggregate states and merges them into diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_reduce_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_reduce_table.rs index d8e92c5928b8a..a54faeb39acfc 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_reduce_table.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_reduce_table.rs @@ -20,6 +20,7 @@ use arrow::record_batch::RecordBatch; use datafusion_common::Result; use crate::aggregates::AggregateExec; +use crate::metrics::BaselineMetrics; use super::common::{AggregateHashTable, HashAggregateAccumulator, PartialReduceMarker}; @@ -48,8 +49,9 @@ impl AggregateHashTable { /// exhausted, and an internal error if polled in the `Building` state. pub(in crate::aggregates) fn next_output_batch( &mut self, + bm: &BaselineMetrics, ) -> Result> { - self.next_output_batch_inner(HashAggregateAccumulator::state) + self.next_output_batch_inner(HashAggregateAccumulator::state, bm) } /// Partial-reduce aggregation consumes partial aggregate states and merges diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs index ffac42feaa3b3..10d39d405343a 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs @@ -27,6 +27,7 @@ use datafusion_common::{Result, assert_eq_or_internal_err}; use crate::aggregates::group_values::new_group_values; use crate::aggregates::order::GroupOrdering; use crate::aggregates::{AggregateExec, group_id_array, max_duplicate_ordinal}; +use crate::metrics::BaselineMetrics; use super::common::{ AggregateHashTable, AggregateHashTableBuffer, AggregateHashTableState, @@ -64,8 +65,9 @@ impl AggregateHashTable { /// exhausted, and an internal error if polled in the `Building` state. pub(in crate::aggregates) fn next_output_batch( &mut self, + bm: &BaselineMetrics, ) -> Result> { - self.next_output_batch_inner(HashAggregateAccumulator::state) + self.next_output_batch_inner(HashAggregateAccumulator::state, bm) } pub(in crate::aggregates) fn can_skip_aggregation(&self) -> bool { diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/single_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/single_table.rs index 5dcb735d083c4..1dea1f5dfb2da 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/single_table.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/single_table.rs @@ -20,6 +20,7 @@ use arrow::record_batch::RecordBatch; use datafusion_common::Result; use crate::aggregates::AggregateExec; +use crate::metrics::BaselineMetrics; use super::common::{AggregateHashTable, HashAggregateAccumulator, SingleMarker}; @@ -54,8 +55,9 @@ impl AggregateHashTable { /// exhausted, and an internal error if polled in the `Building` state. pub(in crate::aggregates) fn next_output_batch( &mut self, + bm: &BaselineMetrics, ) -> Result> { - self.next_output_batch_inner(HashAggregateAccumulator::evaluate_to_columns) + self.next_output_batch_inner(HashAggregateAccumulator::evaluate_to_columns, bm) } /// Single aggregation consumes raw input rows and updates the table's diff --git a/datafusion/physical-plan/src/aggregates/hash_stream.rs b/datafusion/physical-plan/src/aggregates/hash_stream.rs index 62b92965030ae..c79ae6e4893b0 100644 --- a/datafusion/physical-plan/src/aggregates/hash_stream.rs +++ b/datafusion/physical-plan/src/aggregates/hash_stream.rs @@ -532,7 +532,9 @@ impl PartialHashAggregateStream { let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); let timer = elapsed_compute.timer(); - let result = original_state.hash_table_mut().next_output_batch(); + let result = original_state + .hash_table_mut() + .next_output_batch(&self.baseline_metrics); timer.done(); match result { @@ -560,10 +562,7 @@ impl PartialHashAggregateStream { original_state }; - ControlFlow::Break(( - Poll::Ready(Some(Ok(batch.record_output(&self.baseline_metrics)))), - next_state, - )) + ControlFlow::Break((Poll::Ready(Some(Ok(batch))), next_state)) } Ok(None) => { let _ = self.reservation.try_resize(0); @@ -894,7 +893,9 @@ impl FinalHashAggregateStream { let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); let timer = elapsed_compute.timer(); - let result = original_state.hash_table_mut().next_output_batch(); + let result = original_state + .hash_table_mut() + .next_output_batch(&self.baseline_metrics); timer.done(); match result { @@ -909,10 +910,7 @@ impl FinalHashAggregateStream { original_state }; - ControlFlow::Break(( - Poll::Ready(Some(Ok(batch.record_output(&self.baseline_metrics)))), - next_state, - )) + ControlFlow::Break((Poll::Ready(Some(Ok(batch))), next_state)) } Ok(None) => { let _ = self.reservation.try_resize(0); @@ -1020,6 +1018,223 @@ mod tests { use datafusion_physical_expr::expressions::col; use futures::StreamExt; + use crate::metrics::MetricValue; + + fn task_ctx_with_batch_size(batch_size: u64) -> Result> { + let runtime = RuntimeEnvBuilder::default().build_arc()?; + let mut task_ctx = TaskContext::default().with_runtime(runtime); + let mut session_config = task_ctx.session_config().clone(); + session_config = session_config.set( + "datafusion.execution.batch_size", + &datafusion_common::ScalarValue::UInt64(Some(batch_size)), + ); + task_ctx = task_ctx.with_session_config(session_config); + Ok(Arc::new(task_ctx)) + } + + fn output_bytes_metric(agg: &AggregateExec) -> usize { + agg.metrics() + .unwrap() + .sum(|m| matches!(m.value(), MetricValue::OutputBytes(_))) + .map(|v| v.as_usize()) + .unwrap() + } + + // Regression test: the hash table materializes its emitted output once, + // then slices it into `batch_size`-sized chunks across successive + // `poll_next` calls (see `next_batch` in aggregate_hash_table/common.rs). + // Since those slices share the same underlying buffers, `output_bytes` + // must not count those buffers once per slice. + #[tokio::test] + async fn test_partial_hash_stream_output_bytes_metric_deduplicates_across_slices() + -> Result<()> { + async fn run(batch_size: u64) -> Result<(usize, usize)> { + let schema = Arc::new(Schema::new(vec![ + Field::new("group_col", DataType::Int32, false), + Field::new("value_col", DataType::Int64, false), + ])); + + // 30 distinct groups, so the hash table's final emit is one 30-row batch. + let group_ids: Vec = (0..30).collect(); + let values: Vec = vec![1; 30]; + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int32Array::from(group_ids)), + Arc::new(Int64Array::from(values)), + ], + )?; + let input_partitions = vec![vec![batch]]; + + let group_expr = vec![(col("group_col", &schema)?, "group_col".to_string())]; + let aggr_expr = vec![Arc::new( + AggregateExprBuilder::new(count_udaf(), vec![col("value_col", &schema)?]) + .schema(Arc::clone(&schema)) + .alias("count_value") + .build()?, + )]; + + let exec = + TestMemoryExec::try_new(&input_partitions, Arc::clone(&schema), None)?; + let exec = Arc::new(TestMemoryExec::update_cache(&Arc::new(exec))); + + let aggregate_exec = AggregateExec::try_new( + AggregateMode::Partial, + PhysicalGroupBy::new_single(group_expr), + aggr_expr, + vec![None], + exec, + Arc::clone(&schema), + )?; + + let task_ctx = task_ctx_with_batch_size(batch_size)?; + let mut stream = + PartialHashAggregateStream::new(&aggregate_exec, &task_ctx, 0)?; + let mut num_batches = 0; + while let Some(result) = stream.next().await { + result?; + num_batches += 1; + } + + Ok((num_batches, output_bytes_metric(&aggregate_exec))) + } + + // batch_size = 100: the 30-row emit fits in one output batch, no slicing. + let (num_batches_whole, output_bytes_whole) = run(100).await?; + assert_eq!(num_batches_whole, 1); + + // batch_size = 10: the same 30-row emit is sliced into 3 output + // batches, all sharing buffers with the same original batch. + let (num_batches_split, output_bytes_split) = run(10).await?; + assert_eq!(num_batches_split, 3); + + // The emitted batch is materialized once regardless of batch_size — + // batch_size only controls how that one batch is later sliced across + // poll_next calls. So with correct deduplication, output_bytes must + // be exactly the same whether or not the batch gets split. + assert_eq!( + output_bytes_split, output_bytes_whole, + "output_bytes should be deduplicated across output slices" + ); + + Ok(()) + } + + // Same regression as above, but for the final stage: the final hash + // table also materializes its emitted output once and slices it across + // polls, so its output_bytes must be deduplicated the same way. + #[tokio::test] + async fn test_final_hash_stream_output_bytes_metric_deduplicates_across_slices() + -> Result<()> { + async fn run(batch_size: u64) -> Result<(usize, usize)> { + let schema = Arc::new(Schema::new(vec![ + Field::new("group_col", DataType::Int32, false), + Field::new("value_col", DataType::Int64, false), + ])); + + // 30 distinct groups, so the final stage's emit is one 30-row batch. + let group_ids: Vec = (0..30).collect(); + let values: Vec = vec![1; 30]; + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int32Array::from(group_ids)), + Arc::new(Int64Array::from(values)), + ], + )?; + let input_partitions = vec![vec![batch]]; + + let group_by = PhysicalGroupBy::new_single(vec![( + col("group_col", &schema)?, + "group_col".to_string(), + )]); + let aggr_expr = vec![Arc::new( + AggregateExprBuilder::new(count_udaf(), vec![col("value_col", &schema)?]) + .schema(Arc::clone(&schema)) + .alias("count_value") + .build()?, + )]; + + let exec = + TestMemoryExec::try_new(&input_partitions, Arc::clone(&schema), None)?; + let exec = Arc::new(TestMemoryExec::update_cache(&Arc::new(exec))); + + let partial_aggregate_exec = AggregateExec::try_new( + AggregateMode::Partial, + group_by.clone(), + aggr_expr.clone(), + vec![None], + exec, + Arc::clone(&schema), + )?; + + // Generate the partial-state input for the final stage under + // test. Use a large batch_size here so the partial stage returns + // one unsliced batch — the split we're testing happens in the + // *final* stage below, not here. + let gen_task_ctx = task_ctx_with_batch_size(1024)?; + let mut partial_stream = PartialHashAggregateStream::new( + &partial_aggregate_exec, + &gen_task_ctx, + 0, + )?; + let mut partial_batches = Vec::new(); + while let Some(result) = partial_stream.next().await { + partial_batches.push(result?); + } + assert_eq!( + partial_batches.len(), + 1, + "expected one unsliced partial batch" + ); + let partial_schema = partial_aggregate_exec.schema(); + + let final_input = TestMemoryExec::try_new( + &[partial_batches], + Arc::clone(&partial_schema), + None, + )?; + let final_input = + Arc::new(TestMemoryExec::update_cache(&Arc::new(final_input))); + + let final_aggregate_exec = AggregateExec::try_new( + AggregateMode::Final, + group_by.as_final(), + aggr_expr, + vec![None], + final_input, + Arc::clone(&schema), + )?; + + let task_ctx = task_ctx_with_batch_size(batch_size)?; + let mut stream = + FinalHashAggregateStream::new(&final_aggregate_exec, &task_ctx, 0)?; + let mut num_batches = 0; + while let Some(result) = stream.next().await { + result?; + num_batches += 1; + } + + Ok((num_batches, output_bytes_metric(&final_aggregate_exec))) + } + + // batch_size = 100: the 30-row emit fits in one output batch, no slicing. + let (num_batches_whole, output_bytes_whole) = run(100).await?; + assert_eq!(num_batches_whole, 1); + + // batch_size = 10: the same 30-row emit is sliced into 3 output + // batches, all sharing buffers with the same original batch. + let (num_batches_split, output_bytes_split) = run(10).await?; + assert_eq!(num_batches_split, 3); + + assert_eq!( + output_bytes_split, output_bytes_whole, + "output_bytes should be deduplicated across output slices" + ); + + Ok(()) + } + #[tokio::test] async fn test_partial_hash_stream_double_emission_race_condition_bug() -> Result<()> { // Fix for https://github.com/apache/datafusion/issues/18701 diff --git a/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs b/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs index 2f4535e66f4ef..f538e01909cfe 100644 --- a/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs +++ b/datafusion/physical-plan/src/aggregates/partial_reduce_stream.rs @@ -35,7 +35,7 @@ use futures::stream::{Stream, StreamExt}; use super::AggregateExec; use super::aggregate_hash_table::{AggregateHashTable, PartialReduceMarker}; -use crate::metrics::{BaselineMetrics, RecordOutput, SpillMetrics}; +use crate::metrics::{BaselineMetrics, SpillMetrics}; use crate::stream::EmptyRecordBatchStream; use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream}; @@ -276,7 +276,9 @@ impl PartialReduceHashAggregateStream { let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); let timer = elapsed_compute.timer(); - let result = original_state.hash_table_mut().next_output_batch(); + let result = original_state + .hash_table_mut() + .next_output_batch(&self.baseline_metrics); timer.done(); match result { @@ -291,10 +293,7 @@ impl PartialReduceHashAggregateStream { original_state }; - ControlFlow::Break(( - Poll::Ready(Some(Ok(batch.record_output(&self.baseline_metrics)))), - next_state, - )) + ControlFlow::Break((Poll::Ready(Some(Ok(batch))), next_state)) } Ok(None) => { let _ = self.reservation.try_resize(0); diff --git a/datafusion/physical-plan/src/aggregates/single_stream.rs b/datafusion/physical-plan/src/aggregates/single_stream.rs index 886ffdd3a0b99..e86d568d732a2 100644 --- a/datafusion/physical-plan/src/aggregates/single_stream.rs +++ b/datafusion/physical-plan/src/aggregates/single_stream.rs @@ -35,7 +35,7 @@ use futures::stream::{Stream, StreamExt}; use super::AggregateExec; use super::aggregate_hash_table::{AggregateHashTable, SingleMarker}; -use crate::metrics::{BaselineMetrics, RecordOutput}; +use crate::metrics::BaselineMetrics; use crate::stream::EmptyRecordBatchStream; use crate::{InputOrderMode, RecordBatchStream, SendableRecordBatchStream}; @@ -264,7 +264,9 @@ impl SingleHashAggregateStream { let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); let timer = elapsed_compute.timer(); - let result = original_state.hash_table_mut().next_output_batch(); + let result = original_state + .hash_table_mut() + .next_output_batch(&self.baseline_metrics); timer.done(); match result { @@ -285,10 +287,7 @@ impl SingleHashAggregateStream { original_state }; - ControlFlow::Break(( - Poll::Ready(Some(Ok(batch.record_output(&self.baseline_metrics)))), - next_state, - )) + ControlFlow::Break((Poll::Ready(Some(Ok(batch))), next_state)) } Ok(None) => { let _ = self.reservation.try_resize(0); From ab748ff01ef744935615d28ae988556689fead9b Mon Sep 17 00:00:00 2001 From: Ariel Miculas Date: Wed, 22 Jul 2026 11:33:38 +0300 Subject: [PATCH 2/2] chore: fix typo in comment --- datafusion/physical-expr-common/src/metrics/baseline.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/datafusion/physical-expr-common/src/metrics/baseline.rs b/datafusion/physical-expr-common/src/metrics/baseline.rs index d6f1d4e268bc4..a27be1934690e 100644 --- a/datafusion/physical-expr-common/src/metrics/baseline.rs +++ b/datafusion/physical-expr-common/src/metrics/baseline.rs @@ -323,7 +323,7 @@ impl RecordBatchMemoryMetrics { Self::default() } - /// Similar to RecordBatch.record_output, but deduplicating accross batches to avoid + /// Similar to RecordBatch.record_output, but deduplicating across batches to avoid /// output_size inflation due to shared memory pub fn record_output(&mut self, batch: &RecordBatch, bm: &BaselineMetrics) { bm.record_output(batch.num_rows());