From 81ffb0c888f397e3e73c400f021aac63b138c1e4 Mon Sep 17 00:00:00 2001 From: Minh Nguyen Date: Wed, 7 Oct 2026 09:45:17 +0200 Subject: [PATCH 1/3] perf: compute output_bytes without building ArrayData Closes: #26071 `BaselineMetrics::record_output` called `get_record_batch_memory_size` for every output batch, which materializes `ArrayData` and inserts every buffer address into a hash set. Compute `output_bytes` as the sum of each column's `Array::get_array_memory_size` instead: it never builds `ArrayData` and is 10.8-13.9x faster on wide struct batches. `get_record_batch_memory_size` is unchanged and remains what memory reservations use, since those need exact deduplicated accounting. --- .../src/metrics/baseline.rs | 33 ++++++++++++++----- docs/source/user-guide/metrics.md | 12 +++---- 2 files changed, 31 insertions(+), 14 deletions(-) diff --git a/datafusion/physical-expr-common/src/metrics/baseline.rs b/datafusion/physical-expr-common/src/metrics/baseline.rs index 52ad4aac9fd98..bbe8cf244947f 100644 --- a/datafusion/physical-expr-common/src/metrics/baseline.rs +++ b/datafusion/physical-expr-common/src/metrics/baseline.rs @@ -20,7 +20,7 @@ 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; use super::{ Count, ExecutionPlanMetricsSet, Metric, MetricBuilder, MetricsSet, Time, Timestamp, @@ -62,9 +62,14 @@ pub struct BaselineMetrics { /// Memory usage of all output batches. /// - /// Note: This value may be overestimated. If multiple output `RecordBatch` - /// instances share underlying memory buffers, their sizes will be counted - /// multiple times. + /// Computed as the sum of each output column's `Array::get_array_memory_size` + /// so that recording this metric never builds `ArrayData`. Use + /// `datafusion_common::utils::memory::get_record_batch_memory_size` instead + /// when exact, buffer-deduplicated accounting is required. + /// + /// Note: This value may be overestimated. A buffer shared between two + /// columns, or between an array and its children, is counted once per + /// reference, and each array's own in-memory structure is included. /// Issue: output_bytes: Count, @@ -312,6 +317,20 @@ impl SplitMetrics { } } +/// Returns the total memory of `batch`'s arrays, computed without building +/// `ArrayData`. +/// +/// Unlike `datafusion_common::utils::memory::get_record_batch_memory_size`, +/// this does not deduplicate buffers shared between columns, and includes each +/// `Array`'s own in-memory structure. +fn output_bytes(batch: &RecordBatch) -> usize { + batch + .columns() + .iter() + .map(|column| column.get_array_memory_size()) + .sum() +} + /// 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 @@ -331,8 +350,7 @@ impl RecordOutput for usize { impl RecordOutput for RecordBatch { fn record_output(self, bm: &BaselineMetrics) -> Self { bm.record_output(self.num_rows()); - let n_bytes = get_record_batch_memory_size(&self); - bm.output_bytes.add(n_bytes); + bm.output_bytes.add(output_bytes(&self)); bm.output_batches.add(1); self } @@ -341,8 +359,7 @@ impl RecordOutput for RecordBatch { impl RecordOutput for &RecordBatch { fn record_output(self, bm: &BaselineMetrics) -> Self { bm.record_output(self.num_rows()); - let n_bytes = get_record_batch_memory_size(self); - bm.output_bytes.add(n_bytes); + bm.output_bytes.add(output_bytes(&self)); bm.output_batches.add(1); self } diff --git a/docs/source/user-guide/metrics.md b/docs/source/user-guide/metrics.md index 111df66ccc08a..8b2980c3645a9 100644 --- a/docs/source/user-guide/metrics.md +++ b/docs/source/user-guide/metrics.md @@ -27,12 +27,12 @@ DataFusion operators expose runtime metrics so you can understand where time is `BaselineMetrics` are available in most physical operators to capture common measurements. -| Metric | Description | -| --------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| elapsed_compute | CPU time the operator actively spends processing work. | -| output_rows | Total number of rows the operator produces. | -| output_bytes | Memory usage of all output batches. Note: This value may be overestimated. If multiple output `RecordBatch` instances share underlying memory buffers, their sizes will be counted multiple times. | -| output_batches | Total number of output batches the operator produces. | +| Metric | Description | +| --------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| elapsed_compute | CPU time the operator actively spends processing work. | +| output_rows | Total number of rows the operator produces. | +| output_bytes | Memory usage of all output batches, computed as the sum of each column's `Array::get_array_memory_size`. Note: This value may be overestimated: a buffer shared between two columns, or between an array and its children, is counted once per reference, and each array's own in-memory structure is included. | +| output_batches | Total number of output batches the operator produces. | ## Operator-specific Metrics From e2014b3f99231b7282ef8313b0f16ad7ebecdf38 Mon Sep 17 00:00:00 2001 From: Minh Nguyen Date: Wed, 7 Oct 2026 10:14:57 +0200 Subject: [PATCH 2/3] perf: compute output_bytes without building ArrayData Closes #26071 `BaselineMetrics::record_output` called `get_record_batch_memory_size` for every output batch, which materializes `ArrayData` and inserts every buffer address into a hash set. Compute `output_bytes` as the sum of each column's `Array::get_array_memory_size` instead, which never builds `ArrayData`. `get_record_batch_memory_size` is unchanged and remains what memory reservations use, since those need exact deduplicated accounting. --- 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 bbe8cf244947f..0184e1592514d 100644 --- a/datafusion/physical-expr-common/src/metrics/baseline.rs +++ b/datafusion/physical-expr-common/src/metrics/baseline.rs @@ -359,7 +359,7 @@ impl RecordOutput for RecordBatch { impl RecordOutput for &RecordBatch { fn record_output(self, bm: &BaselineMetrics) -> Self { bm.record_output(self.num_rows()); - bm.output_bytes.add(output_bytes(&self)); + bm.output_bytes.add(output_bytes(self)); bm.output_batches.add(1); self } From 7ec6dfad5cc0e5d758433955439e740c36431580 Mon Sep 17 00:00:00 2001 From: Minh Nguyen Date: Wed, 14 Oct 2026 10:44:20 +0200 Subject: [PATCH 3/3] test: update output_bytes expectations for get_array_memory_size accounting `BaselineMetrics::record_output` now sums each output column's `Array::get_array_memory_size()` instead of calling `get_record_batch_memory_size`, which deduplicates shared buffers and ignores per-array overhead. The new accounting adds `size_of::()` and `size_of::()` per column and counts a buffer once per reference, so `output_bytes` is systematically larger: a single Int64 column of 2 rows goes from 32 B to 128 B, and a 2-column 3-row aggregate batch goes from 1056 B to 1248 B. Update the pinned values in explain_analyze.slt and dynamic_filter_pushdown_config.slt to match. --- .../dynamic_filter_pushdown_config.slt | 2 +- .../test_files/explain_analyze.slt | 58 +++++++++---------- 2 files changed, 30 insertions(+), 30 deletions(-) diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 6a6bad99f0840..962c7f941c1f4 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -104,7 +104,7 @@ Plan with Metrics 03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, ] 04)------FilterExec: value@1 > 3, metrics=[output_rows=10, , selectivity=100% (10/10)] 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, ] -06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=1147.0 B, bytes_scanned=210.0 B, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=272.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=1147.0 B, bytes_scanned=210.0 B, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] statement ok set datafusion.explain.analyze_level = dev; diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt index 89567a920de68..2dd9df2a39555 100644 --- a/datafusion/sqllogictest/test_files/explain_analyze.slt +++ b/datafusion/sqllogictest/test_files/explain_analyze.slt @@ -148,7 +148,7 @@ SELECT a FROM generate_series(1, 100) as t1(a); ---- Plan with Metrics -01)ProjectionExec: expr=[value@0 as a], metrics=[output_rows=100, elapsed_compute=, output_bytes=64.0 KB, output_batches=1, expr_0_eval_time=] +01)ProjectionExec: expr=[value@0 as a], metrics=[output_rows=100, elapsed_compute=, output_bytes=64.1 KB, output_batches=1, expr_0_eval_time=] # 2 exprs @@ -159,7 +159,7 @@ SELECT a+1, pow(a,2) FROM generate_series(1, 100) as t1(a); ---- Plan with Metrics -01)ProjectionExec: expr=[a@0 + 1 as t1.a + Int64(1), power(CAST(a@0 AS Float64), 2) as pow(t1.a,Int64(2))], metrics=[output_rows=100, elapsed_compute=, output_bytes=1600.0 B, output_batches=1, expr_0_eval_time=, expr_1_eval_time=] +01)ProjectionExec: expr=[a@0 + 1 as t1.a + Int64(1), power(CAST(a@0 AS Float64), 2) as pow(t1.a,Int64(2))], metrics=[output_rows=100, elapsed_compute=, output_bytes=1792.0 B, output_batches=1, expr_0_eval_time=, expr_1_eval_time=] # common expressions @@ -170,8 +170,8 @@ SELECT a+1, a+1 as another_a_plus_one FROM generate_series(1, 100) as t1(a); ---- Plan with Metrics -01)ProjectionExec: expr=[__common_expr_1@0 as t1.a + Int64(1), __common_expr_1@0 as another_a_plus_one], metrics=[output_rows=100, elapsed_compute=, output_bytes=800.0 B, output_batches=1, expr_0_eval_time=, expr_1_eval_time=] -02)--ProjectionExec: expr=[a@0 + 1 as __common_expr_1], metrics=[output_rows=100, elapsed_compute=, output_bytes=800.0 B, output_batches=1, expr_0_eval_time=] +01)ProjectionExec: expr=[__common_expr_1@0 as t1.a + Int64(1), __common_expr_1@0 as another_a_plus_one], metrics=[output_rows=100, elapsed_compute=, output_bytes=1792.0 B, output_batches=1, expr_0_eval_time=, expr_1_eval_time=] +02)--ProjectionExec: expr=[a@0 + 1 as __common_expr_1], metrics=[output_rows=100, elapsed_compute=, output_bytes=896.0 B, output_batches=1, expr_0_eval_time=] statement ok @@ -340,10 +340,10 @@ FROM t1 JOIN t2 ON t1.k = t2.k; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)], metrics=[output_bytes=128.0 KB, build_mem_used=44.0 B] -02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)], metrics=[output_bytes=128.2 KB, build_mem_used=44.0 B] +02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 03)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] -04)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +04)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 05)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] statement ok @@ -361,11 +361,11 @@ FROM t1 JOIN t2 ON t1.k = t2.k; ---- Plan with Metrics -01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)], metrics=[output_bytes=128.0 KB, build_mem_used=44.0 B] -02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +01)HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(k@0, k@0)], metrics=[output_bytes=128.2 KB, build_mem_used=44.0 B] +02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 03)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] 04)--BufferExec: capacity=1024, metrics=[max_mem_used=128.0 B] -05)----ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +05)----ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 06)------DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] statement ok @@ -383,10 +383,10 @@ FROM t1 CROSS JOIN t2; ---- Plan with Metrics -01)CrossJoinExec, metrics=[output_bytes=96.0 B, build_mem_used=128.0 B] -02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +01)CrossJoinExec, metrics=[output_bytes=480.0 B, build_mem_used=128.0 B] +02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 03)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] -04)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +04)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 05)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] query TT @@ -401,10 +401,10 @@ FROM t1 JOIN t2 ON t1.k < t2.k; ---- Plan with Metrics -01)NestedLoopJoinExec: join_type=Inner, filter=k@0 < k@1, metrics=[output_bytes=128.0 KB, spilled_bytes=0.0 B, build_mem_used=128.0 B] -02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +01)NestedLoopJoinExec: join_type=Inner, filter=k@0 < k@1, metrics=[output_bytes=128.2 KB, spilled_bytes=0.0 B, build_mem_used=128.0 B] +02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 03)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] -04)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +04)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 05)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] statement ok @@ -422,11 +422,11 @@ 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=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] +01)PiecewiseMergeJoin: operator=Gt, join_type=Inner, on=(k > k), metrics=[output_bytes=128.2 KB, build_mem_used=144.0 B] +02)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=112.0 B] +03)----SortExec: expr=[column1@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=112.0 B, spilled_bytes=0.0 B] 04)------DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] -05)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +05)--ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 06)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] statement ok @@ -455,10 +455,10 @@ JOIN ea_smj_t2 ON ea_smj_t1.a = ea_smj_t2.a AND ea_smj_t2.b * 50 <= ea_smj_t1.b; ---- Plan with Metrics -01)SortMergeJoinExec: join_type=Inner, on=[(a@0, a@0)], filter=CAST(b@1 AS Int64) * 50 <= CAST(b@0 AS Int64), metrics=[output_bytes=320.0 KB, spilled_bytes=0.0 B, peak_mem_used=432.0 B] -02)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=60.0 B, spilled_bytes=0.0 B] +01)SortMergeJoinExec: join_type=Inner, on=[(a@0, a@0)], filter=CAST(b@1 AS Int64) * 50 <= CAST(b@0 AS Int64), metrics=[output_bytes=320.4 KB, spilled_bytes=0.0 B, peak_mem_used=432.0 B] +02)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=268.0 B, spilled_bytes=0.0 B] 03)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] -04)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=40.0 B, spilled_bytes=0.0 B] +04)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=248.0 B, spilled_bytes=0.0 B] 05)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] query TT @@ -468,10 +468,10 @@ FROM ea_smj_t1 LEFT SEMI JOIN ea_smj_t2 ON ea_smj_t1.a = ea_smj_t2.a; ---- Plan with Metrics -01)SortMergeJoinExec: join_type=LeftSemi, on=[(a@0, a@0)], metrics=[output_bytes=160.0 KB, spilled_bytes=0.0 B, peak_mem_used=0.0 B] -02)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=60.0 B, spilled_bytes=0.0 B] +01)SortMergeJoinExec: join_type=LeftSemi, on=[(a@0, a@0)], metrics=[output_bytes=160.2 KB, spilled_bytes=0.0 B, peak_mem_used=0.0 B] +02)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=268.0 B, spilled_bytes=0.0 B] 03)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] -04)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=32.0 B, spilled_bytes=0.0 B] +04)--SortExec: expr=[a@0 ASC], preserve_partitioning=[false], metrics=[output_bytes=144.0 B, spilled_bytes=0.0 B] 05)----DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] statement ok @@ -499,9 +499,9 @@ FROM t GROUP BY k; ---- Plan with Metrics -01)ProjectionExec: expr=[k@0 as k, count(Int64(1))@1 as count(*)], metrics=[output_bytes=1056.0 B] -02)--AggregateExec: mode=Single, gby=[k@0 as k], aggr=[count(Int64(1))], metrics=[output_bytes=1056.0 B, spilled_bytes=0.0 B, peak_mem_used=9.4 KB] -03)----ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=32.0 B] +01)ProjectionExec: expr=[k@0 as k, count(Int64(1))@1 as count(*)], metrics=[output_bytes=1248.0 B] +02)--AggregateExec: mode=Single, gby=[k@0 as k], aggr=[count(Int64(1))], metrics=[output_bytes=1248.0 B, spilled_bytes=0.0 B, peak_mem_used=9.4 KB] +03)----ProjectionExec: expr=[column1@0 as k], metrics=[output_bytes=128.0 B] 04)------DataSourceExec: partitions=1, partition_sizes=[1], metrics=[] statement ok