Skip to content

perf: avoid ArrayData conversion in BaselineMetrics for every column #26105

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
minhh-nguyen wants to merge 4 commits into
base: main
Choose a base branch
from
Open

perf: avoid ArrayData conversion in BaselineMetrics for every column #26105

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
81ffb0c
perf: compute output_bytes without building ArrayData
minhh-nguyen Oct 7, 2026
e2014b3
perf: compute output_bytes without building ArrayData
minhh-nguyen Oct 7, 2026
7ec6dfa
test: update output_bytes expectations for get_array_memory_size acco…
minhh-nguyen Oct 14, 2026
e10ca04
Merge branch 'main' into perf/26071-avoid-arraydata-conversion
minhh-nguyen Oct 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 25 additions & 8 deletions datafusion/physical-expr-common/src/metrics/baseline.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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: <https://github.com/apache/datafusion/issues/16841>
output_bytes: Count,

Expand Down Expand Up @@ -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
Expand All @@ -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
}
Expand All @@ -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
}
Expand Down
3 changes: 2 additions & 1 deletion datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,8 @@ Plan with Metrics
03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, <slt:ignore>]
04)------FilterExec: value@1 > 3, metrics=[output_rows=10, <slt:ignore>, selectivity=100% (10/10)]
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, <slt:ignore>]
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=<slt:ignore>, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully 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=<slt:ignore>, 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=<slt:ignore>, 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=<slt:ignore>, scan_efficiency_ratio=18.31% (210/1.15 K)]


statement ok
set datafusion.explain.analyze_level = dev;
Expand Down
58 changes: 29 additions & 29 deletions datafusion/sqllogictest/test_files/explain_analyze.slt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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=<slt:ignore>, output_bytes=64.0 KB, output_batches=1, expr_0_eval_time=<slt:ignore>]
01)ProjectionExec: expr=[value@0 as a], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=64.1 KB, output_batches=1, expr_0_eval_time=<slt:ignore>]
<slt:ignore>

# 2 exprs
Expand All @@ -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=<slt:ignore>, output_bytes=1600.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>, expr_1_eval_time=<slt:ignore>]
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=<slt:ignore>, output_bytes=1792.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>, expr_1_eval_time=<slt:ignore>]
<slt:ignore>

# common expressions
Expand All @@ -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=<slt:ignore>, output_bytes=800.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>, expr_1_eval_time=<slt:ignore>]
02)--ProjectionExec: expr=[a@0 + 1 as __common_expr_1], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=800.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>]
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=<slt:ignore>, output_bytes=1792.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>, expr_1_eval_time=<slt:ignore>]
02)--ProjectionExec: expr=[a@0 + 1 as __common_expr_1], metrics=[output_rows=100, elapsed_compute=<slt:ignore>, output_bytes=896.0 B, output_batches=1, expr_0_eval_time=<slt:ignore>]
<slt:ignore>

statement ok
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
12 changes: 6 additions & 6 deletions docs/source/user-guide/metrics.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down