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/08 04:00:18 UTC
(datafusion) 01/01: Refine and document AggregateExec metrics (#24757)
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-24757-84ccbb15be15e843b8cf65e6df5ef664f0fd0984
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 8a9228164fddeaa03c9ee510b641990e403f2eb5
Author: kosiew <ko...@gmail.com>
AuthorDate: Tue Sep 8 03:59:53 2026 +0000
Refine and document AggregateExec metrics (#24757)
## Which issue does this PR close?
* Part of #23570
## Rationale for this change
Aggregate-specific metrics are a user-visible observability surface in
`EXPLAIN ANALYZE`, but their phase boundaries and relationship to
individual aggregate expressions need to be clear and consistent.
In particular, users should be able to understand what each aggregate
timer measures, map indexed metrics such as `agg_expr_0_arguments_time`
back to the corresponding expression in the operator's `aggr=[...]`
list, and understand which metrics are available on different grouped
aggregation paths without reading the implementation.
This PR also tightens the timing boundaries of the grouped aggregate
metrics so that the documented metrics correspond more directly to the
work they describe.
## What changes are included in this PR?
* Documents `AggregateExec` operator-level and per-aggregate metrics,
including their phase boundaries, indexed naming convention, aggregate
expression labels, visibility, and path-specific coverage.
* Groups aggregate timing into explicit phases for:
* group-key preparation,
* aggregate argument and filter evaluation,
* accumulator `update` / `merge` operations, and
* output emission.
* Ensures accumulator `aggregation_time` measures a single interval
containing the relevant accumulator operations rather than accumulating
overlapping/prefix timing.
* Includes group interning and ordering setup in
`time_calculating_group_ids`.
* Includes collectively evaluated aggregate filters in
`aggregate_arguments_time`.
* Adds `topk_maintenance_time` for Grouped TopK priority-map maintenance
and avoids registering `aggregation_time` for that accumulator-free
path.
* Times grouped output materialization consistently with
`emitting_time`.
* Adds `EXPLAIN ANALYZE` coverage demonstrating how indexed
per-aggregate metrics map to multiple aggregate expressions and how
expression labels appear in verbose output.
## Are these changes tested?
Yes.
The patch adds and updates tests that verify:
* `EXPLAIN ANALYZE` reports `agg_expr_0_arguments_time`,
`agg_expr_1_arguments_time`, and `agg_expr_2_arguments_time` for a query
containing `SUM(c5)`, `SUM(c6)`, and `COUNT(c7)`.
* `EXPLAIN ANALYZE VERBOSE` associates those indexed metrics with the
corresponding rendered aggregate expressions.
* Grouped TopK reports `topk_maintenance_time` and
`time_calculating_group_ids` while not registering `aggregation_time`.
* Regular aggregate paths do not report `topk_maintenance_time`.
* Aggregate execution without grouped emission does not report
`emitting_time`.
* `GroupByMetrics::new_topk` does not register `aggregation_time`.
## Are there any user-facing changes?
Yes. Aggregate metrics shown through `EXPLAIN ANALYZE` are clarified and
their timing boundaries are made more consistent.
The metrics documentation now describes the `AggregateExec`
operator-level metrics and per-expression timer phases, explains how
indexed metrics map to aggregate expressions, and documents
path-specific differences such as Grouped TopK and non-grouped
aggregation.
These are observability and documentation changes; this PR does not
introduce a public API change.
## LLM-generated code disclosure
This PR includes LLM-generated code and comments. All LLM-generated
content has been manually reviewed.
---
datafusion/core/tests/sql/explain_analyze.rs | 41 +++
.../src/aggregates/aggregate_hash_table/common.rs | 203 ++++++-----
.../aggregate_hash_table/common_ordered.rs | 147 ++++----
.../src/aggregates/aggregate_stream.rs | 1 +
.../src/aggregates/group_values/metrics.rs | 94 ++++-
.../src/aggregates/grouped_hash_stream.rs | 386 +++++++++++----------
.../src/aggregates/grouped_topk_stream.rs | 109 +++---
datafusion/physical-plan/src/aggregates/mod.rs | 14 +
docs/source/user-guide/metrics.md | 121 +++++++
9 files changed, 706 insertions(+), 410 deletions(-)
diff --git a/datafusion/core/tests/sql/explain_analyze.rs b/datafusion/core/tests/sql/explain_analyze.rs
index f8498cbedf..2ca04ad6b3 100644
--- a/datafusion/core/tests/sql/explain_analyze.rs
+++ b/datafusion/core/tests/sql/explain_analyze.rs
@@ -1064,6 +1064,47 @@ async fn csv_explain_analyze_verbose() {
assert_contains!(formatted, verbose_needle);
}
+#[tokio::test]
+#[cfg_attr(coverage, ignore)]
+async fn explain_analyze_aggregate_metrics_map_indices_to_expressions() {
+ let ctx =
+ SessionContext::new_with_config(SessionConfig::new().with_target_partitions(1));
+ register_aggregate_csv_by_sql(&ctx).await;
+
+ let query =
+ "SELECT c1, SUM(c5), SUM(c6), COUNT(c7) FROM aggregate_test_100 GROUP BY c1";
+ let normal = execute_to_batches(&ctx, &format!("EXPLAIN ANALYZE {query}")).await;
+ let normal = arrow::util::pretty::pretty_format_batches(&normal)
+ .unwrap()
+ .to_string();
+ assert_contains!(
+ normal.as_str(),
+ "aggr=[sum(aggregate_test_100.c5), sum(aggregate_test_100.c6), count(aggregate_test_100.c7)]"
+ );
+ assert_contains!(normal.as_str(), "agg_expr_0_arguments_time");
+ assert_contains!(normal.as_str(), "agg_expr_1_arguments_time");
+ assert_contains!(normal.as_str(), "agg_expr_2_arguments_time");
+ assert!(!normal.contains("aggregate="));
+
+ let verbose =
+ execute_to_batches(&ctx, &format!("EXPLAIN ANALYZE VERBOSE {query}")).await;
+ let verbose = arrow::util::pretty::pretty_format_batches(&verbose)
+ .unwrap()
+ .to_string();
+ assert_contains!(
+ verbose.as_str(),
+ "agg_expr_0_arguments_time{partition=0, aggregate=sum(aggregate_test_100.c5)}"
+ );
+ assert_contains!(
+ verbose.as_str(),
+ "agg_expr_1_arguments_time{partition=0, aggregate=sum(aggregate_test_100.c6)}"
+ );
+ assert_contains!(
+ verbose.as_str(),
+ "agg_expr_2_arguments_time{partition=0, aggregate=count(aggregate_test_100.c7)}"
+ );
+}
+
#[tokio::test]
async fn explain_logical_plan_only() {
let mut config = ConfigOptions::new();
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 0c481a6b3b..75468198f5 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs
@@ -173,24 +173,23 @@ impl<AggrMode> AggregateHashTable<AggrMode> {
batch: &RecordBatch,
) -> Result<EvaluatedAggregateBatch> {
let state = self.state.building();
- let timer = self.group_by_metrics.time_calculating_group_ids.timer();
- // outer vec: one per each grouping set
- // inner vec: all group by exprs for the current grouping set
- let grouping_set_args = evaluate_group_by(&state.group_by, batch)?;
- drop(timer);
-
- let timer = self.group_by_metrics.aggregate_arguments_time.timer();
- // The evaluated args for each accumulator
- let accumulator_args = state
- .accumulators
- .iter()
- .enumerate()
- .map(|(idx, acc)| {
- self.aggregate_argument_metrics
- .time(idx, || acc.evaluate_acc_args(batch))
- })
- .collect::<Result<Vec<_>>>()?;
- drop(timer);
+ // Outer vec: one per grouping set; inner vec: group-by expressions.
+ let grouping_set_args = self
+ .group_by_metrics
+ .time_group_key_preparation(|| evaluate_group_by(&state.group_by, batch))?;
+
+ // The evaluated args for each accumulator.
+ let accumulator_args = self.group_by_metrics.time_aggregate_arguments(|| {
+ state
+ .accumulators
+ .iter()
+ .enumerate()
+ .map(|(idx, acc)| {
+ self.aggregate_argument_metrics
+ .time(idx, || acc.evaluate_acc_args(batch))
+ })
+ .collect::<Result<Vec<_>>>()
+ })?;
Ok(EvaluatedAggregateBatch {
grouping_set_args,
@@ -212,26 +211,31 @@ impl<AggrMode> AggregateHashTable<AggrMode> {
) -> Result<()> {
let evaluated_batch = self.evaluate_batch(batch)?;
let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics);
+ let group_by_metrics = self.group_by_metrics.clone();
let state = self.state.building_mut();
- let _timer = self.group_by_metrics.aggregation_time.timer();
for group_values in &evaluated_batch.grouping_set_args {
- state
- .group_values
- .intern(group_values, &mut state.batch_group_indices)?;
+ group_by_metrics.time_group_key_preparation(|| {
+ state
+ .group_values
+ .intern(group_values, &mut state.batch_group_indices)
+ })?;
+
let group_indices = &state.batch_group_indices;
let total_num_groups = state.group_values.len();
-
- for (idx, (acc, values)) in state
- .accumulators
- .iter_mut()
- .zip(evaluated_batch.accumulator_args.iter())
- .enumerate()
- {
- accumulator_metrics.time(idx, accumulator_phase, || {
- aggregate_fn(acc, values, group_indices, total_num_groups)
- })?;
- }
+ group_by_metrics.time_aggregation(|| {
+ for (idx, (acc, values)) in state
+ .accumulators
+ .iter_mut()
+ .zip(evaluated_batch.accumulator_args.iter())
+ .enumerate()
+ {
+ accumulator_metrics.time(idx, accumulator_phase, || {
+ aggregate_fn(acc, values, group_indices, total_num_groups)
+ })?;
+ }
+ Ok::<(), datafusion_common::DataFusionError>(())
+ })?;
}
Ok(())
@@ -265,16 +269,17 @@ impl<AggrMode> AggregateHashTable<AggrMode> {
// Accumulator output consumes internal state. Materialize all
// groups once, then slice the materialized batch on later polls.
let emit_to = EmitTo::All;
- let timer = self.group_by_metrics.emitting_time.timer();
- let mut columns = state.group_values.emit(emit_to)?;
- for (idx, acc) in state.accumulators.iter_mut().enumerate() {
- columns.extend(accumulator_metrics.time(
- idx,
- accumulator_phase,
- || materialize_accumulator_fn(acc, emit_to),
- )?);
- }
- drop(timer);
+ let columns = self.group_by_metrics.time_emitting(|| {
+ let mut columns = state.group_values.emit(emit_to)?;
+ for (idx, acc) in state.accumulators.iter_mut().enumerate() {
+ columns.extend(accumulator_metrics.time(
+ idx,
+ accumulator_phase,
+ || materialize_accumulator_fn(acc, emit_to),
+ )?);
+ }
+ Ok::<_, datafusion_common::DataFusionError>(columns)
+ })?;
let batch = RecordBatch::try_new(output_schema, columns)?;
debug_assert!(batch.num_rows() > 0);
@@ -334,19 +339,23 @@ impl<AggrMode> AggregateHashTable<AggrMode> {
) -> Result<Option<RecordBatch>> {
let state_schema = Arc::clone(&self.state_schema);
let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics);
+ let group_by_metrics = self.group_by_metrics.clone();
let state = self.state.building_mut();
if state.group_values.is_empty() {
return Ok(None);
}
- let mut output = state.group_values.emit(EmitTo::All)?;
- for (idx, acc) in state.accumulators.iter_mut().enumerate() {
- output.extend(accumulator_metrics.time(
- idx,
- AccumulatorPhase::State,
- || acc.state(EmitTo::All),
- )?);
- }
+ let output = group_by_metrics.time_emitting(|| {
+ let mut output = state.group_values.emit(EmitTo::All)?;
+ for (idx, acc) in state.accumulators.iter_mut().enumerate() {
+ output.extend(accumulator_metrics.time(
+ idx,
+ AccumulatorPhase::State,
+ || acc.state(EmitTo::All),
+ )?);
+ }
+ Ok::<_, datafusion_common::DataFusionError>(output)
+ })?;
let batch = RecordBatch::try_new(state_schema, output)?;
debug_assert!(batch.num_rows() > 0);
@@ -401,57 +410,71 @@ impl<AggrMode> AggregateHashTable<AggrMode> {
/// already expanded keys as plain group columns (see
/// [`PhysicalGroupBy::as_final`]) and never own a grouping set.
pub(super) fn init_empty_grouping_sets(&mut self) -> Result<()> {
+ let group_by_metrics = self.group_by_metrics.clone();
let state = self.state.building_mut();
if !state.group_by.has_grouping_set() || !state.group_values.is_empty() {
return Ok(());
}
let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics);
- let max_ordinal = max_duplicate_ordinal(state.group_by.groups());
- let mut ordinals: HashMap<&[bool], usize> = HashMap::new();
- let group_schema = state.group_by.group_schema(&self.input_schema)?;
- let n_expr = state.group_by.expr().len();
- let mut any_interned = false;
-
- for group in state.group_by.groups() {
- let ordinal = {
- let entry = ordinals.entry(group.as_slice()).or_insert(0);
- let ordinal = *entry;
- *entry += 1;
- ordinal
- };
-
- if !group.iter().all(|&is_null| is_null) {
- continue;
- }
+ let any_interned = group_by_metrics.time_group_key_preparation(|| {
+ let max_ordinal = max_duplicate_ordinal(state.group_by.groups());
+ let mut ordinals: HashMap<&[bool], usize> = HashMap::new();
+ let group_schema = state.group_by.group_schema(&self.input_schema)?;
+ let n_expr = state.group_by.expr().len();
+ let mut any_interned = false;
+
+ for group in state.group_by.groups() {
+ let ordinal = {
+ let entry = ordinals.entry(group.as_slice()).or_insert(0);
+ let ordinal = *entry;
+ *entry += 1;
+ ordinal
+ };
- let mut cols: Vec<ArrayRef> = group_schema
- .fields()
- .iter()
- .take(n_expr)
- .map(|field| new_null_array(field.data_type(), 1))
- .collect();
- cols.push(group_id_array(group, ordinal, max_ordinal, 1)?);
+ if !group.iter().all(|&is_null| is_null) {
+ continue;
+ }
- state
- .group_values
- .intern(&cols, &mut state.batch_group_indices)?;
- any_interned = true;
- }
+ let mut cols: Vec<ArrayRef> = group_schema
+ .fields()
+ .iter()
+ .take(n_expr)
+ .map(|field| new_null_array(field.data_type(), 1))
+ .collect();
+ cols.push(group_id_array(group, ordinal, max_ordinal, 1)?);
+
+ state
+ .group_values
+ .intern(&cols, &mut state.batch_group_indices)?;
+ any_interned = true;
+ }
+ Ok::<_, datafusion_common::DataFusionError>(any_interned)
+ })?;
if any_interned {
let total_groups = state.group_values.len();
let false_filter = BooleanArray::from(vec![false]);
- for (idx, acc) in state.accumulators.iter_mut().enumerate() {
- let null_args = acc.null_arguments(&self.input_schema)?;
- let values = EvaluatedAccumulatorArgs {
- arguments: null_args,
- filter: Some(Arc::new(false_filter.clone())),
- };
- accumulator_metrics.time(idx, AccumulatorPhase::Update, || {
- acc.update_batch(&values, &[0], total_groups)
- })?;
- }
+ let values = state
+ .accumulators
+ .iter()
+ .map(|acc| {
+ Ok(EvaluatedAccumulatorArgs {
+ arguments: acc.null_arguments(&self.input_schema)?,
+ filter: Some(Arc::new(false_filter.clone())),
+ })
+ })
+ .collect::<Result<Vec<_>>>()?;
+ group_by_metrics.time_aggregation(|| {
+ for (idx, (acc, values)) in
+ state.accumulators.iter_mut().zip(values.iter()).enumerate()
+ {
+ accumulator_metrics.time(idx, AccumulatorPhase::Update, || {
+ acc.update_batch(values, &[0], total_groups)
+ })?;
+ }
+ Ok::<(), datafusion_common::DataFusionError>(())
+ })?;
}
Ok(())
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs
index 4ced967a09..418f3f376b 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs
@@ -244,22 +244,22 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
&self,
batch: &RecordBatch,
) -> Result<EvaluatedAggregateBatch> {
- let timer = self.group_by_metrics.time_calculating_group_ids.timer();
- let grouping_set_args = evaluate_group_by(&self.buffer.group_by, batch)?;
- drop(timer);
+ let grouping_set_args =
+ self.group_by_metrics.time_group_key_preparation(|| {
+ evaluate_group_by(&self.buffer.group_by, batch)
+ })?;
- let timer = self.group_by_metrics.aggregate_arguments_time.timer();
- let accumulator_args = self
- .buffer
- .accumulators
- .iter()
- .enumerate()
- .map(|(idx, acc)| {
- self.aggregate_argument_metrics
- .time(idx, || acc.evaluate_acc_args(batch))
- })
- .collect::<Result<Vec<_>>>()?;
- drop(timer);
+ let accumulator_args = self.group_by_metrics.time_aggregate_arguments(|| {
+ self.buffer
+ .accumulators
+ .iter()
+ .enumerate()
+ .map(|(idx, acc)| {
+ self.aggregate_argument_metrics
+ .time(idx, || acc.evaluate_acc_args(batch))
+ })
+ .collect::<Result<Vec<_>>>()
+ })?;
Ok(EvaluatedAggregateBatch {
grouping_set_args,
@@ -326,14 +326,17 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
}
let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics);
- let mut output = self.buffer.group_values.emit(EmitTo::All)?;
- for (idx, acc) in self.buffer.accumulators.iter_mut().enumerate() {
- output.extend(accumulator_metrics.time(
- idx,
- AccumulatorPhase::State,
- || acc.state(EmitTo::All),
- )?);
- }
+ let output = self.group_by_metrics.time_emitting(|| {
+ let mut output = self.buffer.group_values.emit(EmitTo::All)?;
+ for (idx, acc) in self.buffer.accumulators.iter_mut().enumerate() {
+ output.extend(accumulator_metrics.time(
+ idx,
+ AccumulatorPhase::State,
+ || acc.state(EmitTo::All),
+ )?);
+ }
+ Ok::<_, datafusion_common::DataFusionError>(output)
+ })?;
let batch = RecordBatch::try_new(Arc::clone(&self.state_schema), output)?;
debug_assert!(batch.num_rows() > 0);
@@ -381,38 +384,43 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
accumulator_phase: AccumulatorPhase,
) -> Result<()> {
let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics);
+ let group_by_metrics = self.group_by_metrics.clone();
for group_values in &evaluated_batch.grouping_set_args {
- let starting_num_groups = self.buffer.group_values.len();
- self.buffer
- .group_values
- .intern(group_values, &mut self.buffer.group_indices)?;
- let total_num_groups = self.buffer.group_values.len();
- if total_num_groups > starting_num_groups {
- self.buffer.group_ordering.new_groups(
- group_values,
- &self.buffer.group_indices,
- total_num_groups,
- )?;
- }
-
- let timer = self.group_by_metrics.aggregation_time.timer();
- for (idx, (acc, values)) in self
- .buffer
- .accumulators
- .iter_mut()
- .zip(evaluated_batch.accumulator_args.iter())
- .enumerate()
- {
- accumulator_metrics.time(idx, accumulator_phase, || {
- aggregate_fn(
- acc,
- values,
+ let total_num_groups = group_by_metrics.time_group_key_preparation(|| {
+ let starting_num_groups = self.buffer.group_values.len();
+ self.buffer
+ .group_values
+ .intern(group_values, &mut self.buffer.group_indices)?;
+ let total_num_groups = self.buffer.group_values.len();
+ if total_num_groups > starting_num_groups {
+ self.buffer.group_ordering.new_groups(
+ group_values,
&self.buffer.group_indices,
total_num_groups,
- )
- })?;
- }
- drop(timer);
+ )?;
+ }
+ Ok::<_, datafusion_common::DataFusionError>(total_num_groups)
+ })?;
+
+ group_by_metrics.time_aggregation(|| {
+ for (idx, (acc, values)) in self
+ .buffer
+ .accumulators
+ .iter_mut()
+ .zip(evaluated_batch.accumulator_args.iter())
+ .enumerate()
+ {
+ accumulator_metrics.time(idx, accumulator_phase, || {
+ aggregate_fn(
+ acc,
+ values,
+ &self.buffer.group_indices,
+ total_num_groups,
+ )
+ })?;
+ }
+ Ok::<(), datafusion_common::DataFusionError>(())
+ })?;
}
Ok(())
@@ -441,24 +449,27 @@ impl<AggrMode> OrderedAggregateTable<AggrMode> {
self.clamp_emit_to(self.buffer.group_values.len(), emit_to);
let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics);
- let timer = self.group_by_metrics.emitting_time.timer();
- let mut output = self.buffer.group_values.emit(emit_to)?;
- if should_remove_groups {
- match emit_to {
- EmitTo::First(n) => self.buffer.group_ordering.remove_groups(n),
- // `EmitTo::All` is only used after `input_done`, when all
- // buffered groups are known complete and the ordering state is
- // no longer needed.
- EmitTo::All => {}
+ let output = self.group_by_metrics.time_emitting(|| {
+ let mut output = self.buffer.group_values.emit(emit_to)?;
+ if should_remove_groups {
+ match emit_to {
+ EmitTo::First(n) => self.buffer.group_ordering.remove_groups(n),
+ // `EmitTo::All` is only used after `input_done`, when all
+ // buffered groups are known complete and the ordering state is
+ // no longer needed.
+ EmitTo::All => {}
+ }
}
- }
- for (idx, acc) in self.buffer.accumulators.iter_mut().enumerate() {
- output.extend(accumulator_metrics.time(idx, accumulator_phase, || {
- materialize_accumulator_fn(acc, emit_to)
- })?);
- }
- drop(timer);
+ for (idx, acc) in self.buffer.accumulators.iter_mut().enumerate() {
+ output.extend(accumulator_metrics.time(
+ idx,
+ accumulator_phase,
+ || materialize_accumulator_fn(acc, emit_to),
+ )?);
+ }
+ Ok::<_, datafusion_common::DataFusionError>(output)
+ })?;
let batch = RecordBatch::try_new(Arc::clone(&self.output_schema), output)?;
debug_assert!(batch.num_rows() > 0);
diff --git a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs
index a3fc731cb8..23f74e6352 100644
--- a/datafusion/physical-plan/src/aggregates/aggregate_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/aggregate_stream.rs
@@ -701,6 +701,7 @@ mod tests {
}
assert!(aggregate_metrics(&metrics, "merge").is_empty());
assert!(aggregate_metrics(&metrics, "state").is_empty());
+ assert!(metrics.sum_by_name("emitting_time").is_none());
Ok(())
}
diff --git a/datafusion/physical-plan/src/aggregates/group_values/metrics.rs b/datafusion/physical-plan/src/aggregates/group_values/metrics.rs
index 4b5bfd1725..d6405bb403 100644
--- a/datafusion/physical-plan/src/aggregates/group_values/metrics.rs
+++ b/datafusion/physical-plan/src/aggregates/group_values/metrics.rs
@@ -143,38 +143,87 @@ impl AggregateAccumulatorMetrics {
#[derive(Clone)]
pub(crate) struct GroupByMetrics {
- /// Time spent calculating the group IDs from the evaluated grouping columns.
- pub(crate) time_calculating_group_ids: Time,
- /// Time spent evaluating the inputs to the aggregate functions.
- pub(crate) aggregate_arguments_time: Time,
- /// Time spent evaluating the aggregate expressions themselves
- /// (e.g. summing all elements and counting number of elements for `avg` aggregate).
- pub(crate) aggregation_time: Time,
- /// Time spent emitting the final results and constructing the record batch
- /// which includes finalizing the grouping expressions
- /// (e.g. emit from the hash table in case of hash aggregation) and the accumulators
- pub(crate) emitting_time: Time,
+ /// Group-key preparation: grouping-expression evaluation, interning, and ordering updates.
+ time_calculating_group_ids: Time,
+ /// Aggregate-input evaluation, including collectively evaluated filters.
+ aggregate_arguments_time: Time,
+ /// Input-processing accumulator operations (`update` and `merge`).
+ aggregation_time: Option<Time>,
+ /// Output emission: group values and accumulator states or final values.
+ emitting_time: Time,
+ /// Grouped TopK priority-map batch setup, insertion, comparison, and NULL handling.
+ topk_maintenance_time: Option<Time>,
}
impl GroupByMetrics {
pub(crate) fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
+ Self::new_with_options(metrics, partition, true, false)
+ }
+
+ /// Creates metrics for Grouped TopK, which does not invoke accumulators.
+ pub(crate) fn new_topk(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
+ Self::new_with_options(metrics, partition, false, true)
+ }
+
+ fn new_with_options(
+ metrics: &ExecutionPlanMetricsSet,
+ partition: usize,
+ include_aggregation_time: bool,
+ include_topk_maintenance_time: bool,
+ ) -> Self {
Self {
time_calculating_group_ids: MetricBuilder::new(metrics)
.subset_time("time_calculating_group_ids", partition),
aggregate_arguments_time: MetricBuilder::new(metrics)
.subset_time("aggregate_arguments_time", partition),
- aggregation_time: MetricBuilder::new(metrics)
- .subset_time("aggregation_time", partition),
+ aggregation_time: include_aggregation_time.then(|| {
+ MetricBuilder::new(metrics).subset_time("aggregation_time", partition)
+ }),
emitting_time: MetricBuilder::new(metrics)
.subset_time("emitting_time", partition),
+ topk_maintenance_time: include_topk_maintenance_time.then(|| {
+ MetricBuilder::new(metrics)
+ .subset_time("topk_maintenance_time", partition)
+ }),
}
}
+
+ /// Times grouping-expression evaluation, group interning, and ordering updates.
+ pub(crate) fn time_group_key_preparation<R>(&self, f: impl FnOnce() -> R) -> R {
+ let _timer = self.time_calculating_group_ids.timer();
+ f()
+ }
+
+ /// Times aggregate argument and collectively evaluated filter expressions.
+ pub(crate) fn time_aggregate_arguments<R>(&self, f: impl FnOnce() -> R) -> R {
+ let _timer = self.aggregate_arguments_time.timer();
+ f()
+ }
+
+ /// Times one interval containing all input-processing accumulator operations.
+ pub(crate) fn time_aggregation<R>(&self, f: impl FnOnce() -> R) -> R {
+ let _timer = self.aggregation_time.as_ref().map(Time::timer);
+ f()
+ }
+
+ /// Times group-value and accumulator-state/final-value output emission.
+ pub(crate) fn time_emitting<R>(&self, f: impl FnOnce() -> R) -> R {
+ let _timer = self.emitting_time.timer();
+ f()
+ }
+
+ /// Times Grouped TopK priority-map maintenance after group keys are prepared.
+ pub(crate) fn time_topk_maintenance<R>(&self, f: impl FnOnce() -> R) -> R {
+ let _timer = self.topk_maintenance_time.as_ref().map(Time::timer);
+ f()
+ }
}
#[cfg(test)]
mod tests {
+ use super::GroupByMetrics;
use crate::aggregates::{AggregateExec, AggregateMode, PhysicalGroupBy};
- use crate::metrics::{MetricValue, MetricsSet};
+ use crate::metrics::{ExecutionPlanMetricsSet, MetricValue, MetricsSet};
use crate::test::TestMemoryExec;
use crate::{ExecutionPlan, collect};
use arrow::array::{Float64Array, UInt32Array};
@@ -207,6 +256,23 @@ mod tests {
assert!(emitting_time.unwrap().as_usize() > 0);
}
+ #[test]
+ fn groupby_metrics_without_accumulators_do_not_register_aggregation_time() {
+ let metrics = ExecutionPlanMetricsSet::new();
+ let groupby_metrics = GroupByMetrics::new_topk(&metrics, 0);
+ let mut called = false;
+
+ groupby_metrics.time_aggregation(|| called = true);
+
+ assert!(called);
+ assert!(
+ metrics
+ .clone_inner()
+ .sum_by_name("aggregation_time")
+ .is_none()
+ );
+ }
+
fn aggregate_metric_names_and_labels(
metrics: &MetricsSet,
suffix: &str,
diff --git a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
index 3f6f3f8ce8..24bb4c16d8 100644
--- a/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/grouped_hash_stream.rs
@@ -59,7 +59,6 @@ use datafusion_physical_expr_common::sort_expr::LexOrdering;
use datafusion_physical_expr_common::utils::evaluate_expressions_to_arrays;
use crate::sorts::IncrementalSortIterator;
-use datafusion_common::instant::Instant;
use datafusion_common::utils::memory::get_record_batch_memory_size;
use futures::ready;
use futures::stream::{Stream, StreamExt};
@@ -869,120 +868,109 @@ impl RecordBatchStream for GroupedHashAggregateStream {
impl GroupedHashAggregateStream {
/// Perform group-by aggregation for the given [`RecordBatch`].
fn group_aggregate_batch(&mut self, batch: &RecordBatch) -> Result<()> {
- // Evaluate the grouping expressions
- let group_by_values = if self.spill_state.is_stream_merging {
- evaluate_group_by(&self.spill_state.merging_group_by, batch)?
- } else {
- evaluate_group_by(&self.group_by, batch)?
- };
+ let is_stream_merging = self.spill_state.is_stream_merging;
- // Only create the timer if there are actual aggregate arguments to evaluate
- let timer = match (
- self.spill_state.is_stream_merging,
- self.spill_state.merging_aggregate_arguments.is_empty(),
- self.aggregate_arguments.is_empty(),
- ) {
- (true, false, _) | (false, _, false) => {
- Some(self.group_by_metrics.aggregate_arguments_time.timer())
+ // Evaluate the grouping expressions; interning happens below in the
+ // same group-key-preparation phase.
+ let group_by_values = self.group_by_metrics.time_group_key_preparation(|| {
+ if is_stream_merging {
+ evaluate_group_by(&self.spill_state.merging_group_by, batch)
+ } else {
+ evaluate_group_by(&self.group_by, batch)
}
- _ => None,
- };
+ })?;
- // Evaluate the aggregation expressions.
- let aggregate_arguments = if self.spill_state.is_stream_merging {
+ // Evaluate aggregate arguments and filters together. Per-aggregate
+ // argument timers remain limited to their argument expressions because
+ // filters are evaluated collectively here.
+ let aggregate_arguments = if is_stream_merging {
&self.spill_state.merging_aggregate_arguments
} else {
&self.aggregate_arguments
};
- let input_values = aggregate_arguments
- .iter()
- .enumerate()
- .map(|(idx, expr)| {
- self.aggregate_argument_metrics
- .time(idx, || evaluate_expressions_to_arrays(expr, batch))
- })
- .collect::<Result<Vec<_>>>()?;
- drop(timer);
-
- // Evaluate the filter expressions, if any, against the inputs
- let filter_values = if self.spill_state.is_stream_merging {
- let filter_expressions = vec![None; self.accumulators.len()];
- evaluate_optional(&filter_expressions, batch)?
- } else {
- evaluate_optional(&self.filter_expressions, batch)?
- };
+ let (input_values, filter_values) =
+ self.group_by_metrics.time_aggregate_arguments(|| {
+ let input_values = aggregate_arguments
+ .iter()
+ .enumerate()
+ .map(|(idx, expr)| {
+ self.aggregate_argument_metrics
+ .time(idx, || evaluate_expressions_to_arrays(expr, batch))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ let filter_values = if is_stream_merging {
+ let filter_expressions = vec![None; self.accumulators.len()];
+ evaluate_optional(&filter_expressions, batch)?
+ } else {
+ evaluate_optional(&self.filter_expressions, batch)?
+ };
+ Ok::<_, DataFusionError>((input_values, filter_values))
+ })?;
for group_values in &group_by_values {
- let groups_start_time = Instant::now();
-
- // calculate the group indices for each input row
- let starting_num_groups = self.group_values.len();
- self.group_values
- .intern(group_values, &mut self.current_group_indices)?;
- let group_indices = &self.current_group_indices;
-
- // Update ordering information if necessary
- let total_num_groups = self.group_values.len();
- if total_num_groups > starting_num_groups {
- self.group_ordering.new_groups(
- group_values,
- group_indices,
- total_num_groups,
- )?;
- }
-
- // Use this instant for both measurements to save a syscall
- let agg_start_time = Instant::now();
- self.group_by_metrics
- .time_calculating_group_ids
- .add_duration(agg_start_time - groups_start_time);
-
- // Gather the inputs to call the actual accumulator
- let t = self
- .accumulators
- .iter_mut()
- .zip(input_values.iter())
- .zip(filter_values.iter());
-
- for (idx, ((acc, values), opt_filter)) in t.enumerate() {
- let opt_filter = opt_filter.as_ref().map(|filter| filter.as_boolean());
-
- // Call the appropriate method on each aggregator with
- // the entire input row and the relevant group indexes
- if self.mode.input_mode() == AggregateInputMode::Raw
- && !self.spill_state.is_stream_merging
- {
- self.aggregate_accumulator_metrics.time(
- idx,
- AccumulatorPhase::Update,
- || {
- acc.update_batch(
- values,
- group_indices,
- opt_filter,
- total_num_groups,
- )
- },
- )?;
- } else {
- assert_or_internal_err!(
- opt_filter.is_none(),
- "aggregate filter should be applied in partial stage, there should be no filter in final stage"
- );
+ // Calculate group indices and update ordering information.
+ let total_num_groups =
+ self.group_by_metrics.time_group_key_preparation(|| {
+ let starting_num_groups = self.group_values.len();
+ self.group_values
+ .intern(group_values, &mut self.current_group_indices)?;
+ let group_indices = &self.current_group_indices;
+ let total_num_groups = self.group_values.len();
+ if total_num_groups > starting_num_groups {
+ self.group_ordering.new_groups(
+ group_values,
+ group_indices,
+ total_num_groups,
+ )?;
+ }
+ Ok::<_, DataFusionError>(total_num_groups)
+ })?;
- // if aggregation is over intermediate states,
- // use merge
- self.aggregate_accumulator_metrics.time(
- idx,
- AccumulatorPhase::Merge,
- || acc.merge_batch(values, group_indices, total_num_groups),
- )?;
+ // Convert filters before timing accumulator calls. The timer covers
+ // one interval across all accumulator operations, rather than a
+ // prefix sum charged once per aggregate expression.
+ let filters = filter_values
+ .iter()
+ .map(|filter| filter.as_ref().map(|filter| filter.as_boolean()))
+ .collect::<Vec<_>>();
+ self.group_by_metrics.time_aggregation(|| {
+ let group_indices = &self.current_group_indices;
+ let t = self
+ .accumulators
+ .iter_mut()
+ .zip(input_values.iter())
+ .zip(filters.iter());
+
+ for (idx, ((acc, values), opt_filter)) in t.enumerate() {
+ if self.mode.input_mode() == AggregateInputMode::Raw && !is_stream_merging
+ {
+ self.aggregate_accumulator_metrics.time(
+ idx,
+ AccumulatorPhase::Update,
+ || {
+ acc.update_batch(
+ values,
+ group_indices,
+ *opt_filter,
+ total_num_groups,
+ )
+ },
+ )?;
+ } else {
+ assert_or_internal_err!(
+ opt_filter.is_none(),
+ "aggregate filter should be applied in partial stage, there should be no filter in final stage"
+ );
+
+ self.aggregate_accumulator_metrics.time(
+ idx,
+ AccumulatorPhase::Merge,
+ || acc.merge_batch(values, group_indices, total_num_groups),
+ )?;
+ }
}
- }
- // Recorded once per grouping set, after all its accumulators ran
- self.group_by_metrics
- .aggregation_time
- .add_elapsed(agg_start_time);
+ Ok::<(), DataFusionError>(())
+ })?;
}
Ok(())
@@ -1077,31 +1065,31 @@ impl GroupedHashAggregateStream {
return Ok(None);
}
- let timer = self.group_by_metrics.emitting_time.timer();
- let mut output = self.group_values.emit(emit_to)?;
- if let EmitTo::First(n) = emit_to {
- self.group_ordering.remove_groups(n);
- }
+ let group_by_metrics = self.group_by_metrics.clone();
+ let output = group_by_metrics.time_emitting(|| {
+ let mut output = self.group_values.emit(emit_to)?;
+ if let EmitTo::First(n) = emit_to {
+ self.group_ordering.remove_groups(n);
+ }
- // Next output each aggregate value
- for (idx, acc) in self.accumulators.iter_mut().enumerate() {
- if self.mode.output_mode() == AggregateOutputMode::Final && !spilling {
- output.push(self.aggregate_accumulator_metrics.time(
- idx,
- AccumulatorPhase::Evaluate,
- || acc.evaluate(emit_to),
- )?)
- } else {
- // Output partial state: either because we're in a non-final mode,
- // or because we're spilling and will merge/re-evaluate later.
- output.extend(self.aggregate_accumulator_metrics.time(
- idx,
- AccumulatorPhase::State,
- || acc.state(emit_to),
- )?)
+ // Next output each aggregate value.
+ for (idx, acc) in self.accumulators.iter_mut().enumerate() {
+ if self.mode.output_mode() == AggregateOutputMode::Final && !spilling {
+ output.push(self.aggregate_accumulator_metrics.time(
+ idx,
+ AccumulatorPhase::Evaluate,
+ || acc.evaluate(emit_to),
+ )?)
+ } else {
+ output.extend(self.aggregate_accumulator_metrics.time(
+ idx,
+ AccumulatorPhase::State,
+ || acc.state(emit_to),
+ )?)
+ }
}
- }
- drop(timer);
+ Ok::<_, DataFusionError>(output)
+ })?;
// emit reduces the memory usage. Ignore Err from update_memory_reservation. Even if it is
// over the target memory size after emission, we can emit again rather than returning Err.
@@ -1127,47 +1115,50 @@ impl GroupedHashAggregateStream {
return Ok(());
}
- let max_ordinal = max_duplicate_ordinal(self.group_by.groups());
- let mut ordinals: std::collections::HashMap<&[bool], usize> =
- std::collections::HashMap::new();
- let group_schema = self.group_by.group_schema(&self.input_schema)?;
- let n_expr = self.group_by.expr().len();
- let mut any_interned = false;
-
- for group in self.group_by.groups() {
- let ordinal = {
- let entry = ordinals.entry(group.as_slice()).or_insert(0);
- let o = *entry;
- *entry += 1;
- o
- };
+ let group_by_metrics = self.group_by_metrics.clone();
+ let any_interned = group_by_metrics.time_group_key_preparation(|| {
+ let max_ordinal = max_duplicate_ordinal(self.group_by.groups());
+ let mut ordinals: std::collections::HashMap<&[bool], usize> =
+ std::collections::HashMap::new();
+ let group_schema = self.group_by.group_schema(&self.input_schema)?;
+ let n_expr = self.group_by.expr().len();
+ let mut any_interned = false;
+
+ for group in self.group_by.groups() {
+ let ordinal = {
+ let entry = ordinals.entry(group.as_slice()).or_insert(0);
+ let o = *entry;
+ *entry += 1;
+ o
+ };
- if !group.iter().all(|&is_null| is_null) {
- continue;
- }
+ if !group.iter().all(|&is_null| is_null) {
+ continue;
+ }
- // Build the group key: one NULL per group-by expression, then the grouping_id.
- let mut cols: Vec<ArrayRef> = group_schema
- .fields()
- .iter()
- .take(n_expr)
- .map(|f| new_null_array(f.data_type(), 1))
- .collect();
- cols.push(group_id_array(group, ordinal, max_ordinal, 1)?);
-
- let starting_groups = self.group_values.len();
- self.group_values
- .intern(&cols, &mut self.current_group_indices)?;
- let total_groups = self.group_values.len();
- if total_groups > starting_groups {
- self.group_ordering.new_groups(
- &cols,
- &self.current_group_indices,
- total_groups,
- )?;
+ let mut cols: Vec<ArrayRef> = group_schema
+ .fields()
+ .iter()
+ .take(n_expr)
+ .map(|f| new_null_array(f.data_type(), 1))
+ .collect();
+ cols.push(group_id_array(group, ordinal, max_ordinal, 1)?);
+
+ let starting_groups = self.group_values.len();
+ self.group_values
+ .intern(&cols, &mut self.current_group_indices)?;
+ let total_groups = self.group_values.len();
+ if total_groups > starting_groups {
+ self.group_ordering.new_groups(
+ &cols,
+ &self.current_group_indices,
+ total_groups,
+ )?;
+ }
+ any_interned = true;
}
- any_interned = true;
- }
+ Ok::<_, DataFusionError>(any_interned)
+ })?;
if any_interned {
// Prime each accumulator for the registered group count with no data.
@@ -1212,18 +1203,28 @@ impl GroupedHashAggregateStream {
})
.collect::<Result<Vec<_>>>()?;
let false_filter = BooleanArray::from(vec![false]);
- for (idx, (acc, args)) in self
- .accumulators
- .iter_mut()
- .zip(null_args.iter())
- .enumerate()
- {
- self.aggregate_accumulator_metrics.time(
- idx,
- AccumulatorPhase::Update,
- || acc.update_batch(args, &[0], Some(&false_filter), total_groups),
- )?;
- }
+ group_by_metrics.time_aggregation(|| {
+ for (idx, (acc, args)) in self
+ .accumulators
+ .iter_mut()
+ .zip(null_args.iter())
+ .enumerate()
+ {
+ self.aggregate_accumulator_metrics.time(
+ idx,
+ AccumulatorPhase::Update,
+ || {
+ acc.update_batch(
+ args,
+ &[0],
+ Some(&false_filter),
+ total_groups,
+ )
+ },
+ )?;
+ }
+ Ok::<(), DataFusionError>(())
+ })?;
}
Ok(())
@@ -1435,19 +1436,24 @@ impl GroupedHashAggregateStream {
/// Transforms input batch to intermediate aggregate state, without grouping it
fn transform_to_states(&self, batch: &RecordBatch) -> Result<RecordBatch> {
- let mut group_values = evaluate_group_by(&self.group_by, batch)?;
- let timer = self.group_by_metrics.aggregate_arguments_time.timer();
- let input_values = self
- .aggregate_arguments
- .iter()
- .enumerate()
- .map(|(idx, expr)| {
- self.aggregate_argument_metrics
- .time(idx, || evaluate_expressions_to_arrays(expr, batch))
- })
- .collect::<Result<Vec<_>>>()?;
- drop(timer);
- let filter_values = evaluate_optional(&self.filter_expressions, batch)?;
+ let mut group_values = self
+ .group_by_metrics
+ .time_group_key_preparation(|| evaluate_group_by(&self.group_by, batch))?;
+
+ let (input_values, filter_values) =
+ self.group_by_metrics.time_aggregate_arguments(|| {
+ let input_values = self
+ .aggregate_arguments
+ .iter()
+ .enumerate()
+ .map(|(idx, expr)| {
+ self.aggregate_argument_metrics
+ .time(idx, || evaluate_expressions_to_arrays(expr, batch))
+ })
+ .collect::<Result<Vec<_>>>()?;
+ let filter_values = evaluate_optional(&self.filter_expressions, batch)?;
+ Ok::<_, DataFusionError>((input_values, filter_values))
+ })?;
assert_eq_or_internal_err!(
group_values.len(),
diff --git a/datafusion/physical-plan/src/aggregates/grouped_topk_stream.rs b/datafusion/physical-plan/src/aggregates/grouped_topk_stream.rs
index a5b274cffe..13ead73930 100644
--- a/datafusion/physical-plan/src/aggregates/grouped_topk_stream.rs
+++ b/datafusion/physical-plan/src/aggregates/grouped_topk_stream.rs
@@ -74,7 +74,7 @@ impl GroupedTopKAggregateStream {
let group_by = Arc::clone(&aggr.group_by);
let input = aggr.input.execute(partition, Arc::clone(context))?;
let baseline_metrics = BaselineMetrics::new(&aggr.metrics, partition);
- let group_by_metrics = GroupByMetrics::new(&aggr.metrics, partition);
+ let group_by_metrics = GroupByMetrics::new_topk(&aggr.metrics, partition);
let aggregate_argument_metrics = AggregateArgumentMetrics::new(
&aggr.metrics,
partition,
@@ -151,38 +151,40 @@ impl GroupedTopKAggregateStream {
}
fn intern(&mut self, ids: &ArrayRef, vals: &ArrayRef) -> Result<()> {
- let _timer = self.group_by_metrics.time_calculating_group_ids.timer();
+ let group_by_metrics = self.group_by_metrics.clone();
+ group_by_metrics.time_topk_maintenance(|| {
+ let len = ids.len();
+ self.priority_map
+ .set_batch(Arc::clone(ids), Arc::clone(vals));
- let len = ids.len();
- self.priority_map
- .set_batch(Arc::clone(ids), Arc::clone(vals));
-
- let has_nulls = vals.null_count() > 0;
- if has_nulls && self.is_group_by_only() {
- self.null_group_seen = true;
- }
- // Keep the common no-NULL path free of NULL bookkeeping. Once a NULL
- // group exists, use the NULL-aware path until it has been resolved.
- let track_null_groups = !self.is_group_by_only()
- && (has_nulls || self.priority_map.has_null_groups());
- for row_idx in 0..len {
- if has_nulls && vals.is_null(row_idx) {
- // MIN/MAX ignore NULL inputs, but a group whose values are all
- // NULL must still be emitted with a NULL aggregate value, so
- // track it. (GROUP BY-only aggregations handle NULL group keys
- // via `null_group_seen` instead.)
- if !self.is_group_by_only() {
- self.priority_map.insert_null(row_idx);
- }
- continue;
+ let has_nulls = vals.null_count() > 0;
+ let is_group_by_only = self.is_group_by_only();
+ if has_nulls && is_group_by_only {
+ self.null_group_seen = true;
}
- if track_null_groups {
- self.priority_map.insert_with_null_groups(row_idx)?;
- } else {
- self.priority_map.insert(row_idx)?;
+ // Keep the common no-NULL path free of NULL bookkeeping. Once a NULL
+ // group exists, use the NULL-aware path until it has been resolved.
+ let track_null_groups =
+ !is_group_by_only && (has_nulls || self.priority_map.has_null_groups());
+ for row_idx in 0..len {
+ if has_nulls && vals.is_null(row_idx) {
+ // MIN/MAX ignore NULL inputs, but a group whose values are all
+ // NULL must still be emitted with a NULL aggregate value, so
+ // track it. (GROUP BY-only aggregations handle NULL group keys
+ // via `null_group_seen` instead.)
+ if !is_group_by_only {
+ self.priority_map.insert_null(row_idx);
+ }
+ continue;
+ }
+ if track_null_groups {
+ self.priority_map.insert_with_null_groups(row_idx)?;
+ } else {
+ self.priority_map.insert(row_idx)?;
+ }
}
- }
- Ok(())
+ Ok(())
+ })
}
fn emit_columns(&mut self) -> Result<Vec<ArrayRef>> {
@@ -230,7 +232,6 @@ impl Stream for GroupedTopKAggregateStream {
return Poll::Ready(None);
}
let elapsed_compute = self.baseline_metrics.elapsed_compute().clone();
- let emitting_time = self.group_by_metrics.emitting_time.clone();
while let Poll::Ready(res) = self.input.poll_next_unpin(cx) {
let _timer = elapsed_compute.timer();
match res {
@@ -247,7 +248,10 @@ impl Stream for GroupedTopKAggregateStream {
print_batches(std::slice::from_ref(&batch))?;
}
self.row_count += batch.num_rows();
- let group_by_values = evaluate_group_by(&self.group_by, &batch)?;
+ let group_by_values =
+ self.group_by_metrics.time_group_key_preparation(|| {
+ evaluate_group_by(&self.group_by, &batch)
+ })?;
assert_eq!(
group_by_values.len(),
1,
@@ -264,18 +268,18 @@ impl Stream for GroupedTopKAggregateStream {
Arc::clone(&group_by_values)
} else {
// MIN/MAX case: evaluate aggregate expressions
- let _timer =
- self.group_by_metrics.aggregate_arguments_time.timer();
- let input_values = self
- .aggregate_arguments
- .iter()
- .enumerate()
- .map(|(idx, expr)| {
- self.aggregate_argument_metrics.time(idx, || {
- evaluate_expressions_to_arrays(expr, &batch)
- })
- })
- .collect::<Result<Vec<_>>>()?;
+ let input_values =
+ self.group_by_metrics.time_aggregate_arguments(|| {
+ self.aggregate_arguments
+ .iter()
+ .enumerate()
+ .map(|(idx, expr)| {
+ self.aggregate_argument_metrics.time(idx, || {
+ evaluate_expressions_to_arrays(expr, &batch)
+ })
+ })
+ .collect::<Result<Vec<_>>>()
+ })?;
assert_eq!(input_values.len(), 1, "Exactly 1 input required");
assert_eq!(input_values[0].len(), 1, "Exactly 1 input required");
Arc::clone(&input_values[0][0])
@@ -294,11 +298,11 @@ impl Stream for GroupedTopKAggregateStream {
self.done = true;
return Poll::Ready(None);
}
- let batch = {
- let _timer = emitting_time.timer();
+ let group_by_metrics = self.group_by_metrics.clone();
+ let batch = group_by_metrics.time_emitting(|| {
let cols = self.emit_columns()?;
- RecordBatch::try_new(Arc::clone(&self.schema), cols)?
- };
+ RecordBatch::try_new(Arc::clone(&self.schema), cols)
+ })?;
let batch = batch.record_output(&self.baseline_metrics);
trace!(
"partition {} emit batch with {} rows",
@@ -398,6 +402,15 @@ mod tests {
assert!(argument_metric.is_some());
assert!(argument_metric.unwrap().value().as_usize() > 0);
+ let topk_maintenance_time = metrics.sum_by_name("topk_maintenance_time");
+ assert!(topk_maintenance_time.is_some());
+ assert!(topk_maintenance_time.unwrap().as_usize() > 0);
+
+ let group_id_time = metrics.sum_by_name("time_calculating_group_ids");
+ assert!(group_id_time.is_some());
+ assert!(group_id_time.unwrap().as_usize() > 0);
+ assert!(metrics.sum_by_name("aggregation_time").is_none());
+
Ok(())
}
}
diff --git a/datafusion/physical-plan/src/aggregates/mod.rs b/datafusion/physical-plan/src/aggregates/mod.rs
index ba08c4b003..774c08535c 100644
--- a/datafusion/physical-plan/src/aggregates/mod.rs
+++ b/datafusion/physical-plan/src/aggregates/mod.rs
@@ -4407,6 +4407,13 @@ mod tests {
| 2 |
+---+
");
+ assert!(
+ partial_aggregate
+ .metrics()
+ .unwrap()
+ .sum_by_name("topk_maintenance_time")
+ .is_none()
+ );
let final_input =
TestMemoryExec::try_new_exec(&[input_batches], Arc::clone(&schema), None)?;
@@ -4441,6 +4448,13 @@ mod tests {
| 2 |
+---+
");
+ assert!(
+ final_aggregate
+ .metrics()
+ .unwrap()
+ .sum_by_name("topk_maintenance_time")
+ .is_none()
+ );
Ok(())
}
diff --git a/docs/source/user-guide/metrics.md b/docs/source/user-guide/metrics.md
index 306094c438..5bb4895a3d 100644
--- a/docs/source/user-guide/metrics.md
+++ b/docs/source/user-guide/metrics.md
@@ -61,6 +61,127 @@ subsequent join processing time.
| avg_fanout | Average number of build-side join-key matches per matched probe-side row before applying any join filter. |
| array_map_created_count | Number of times `HashJoinExec` created an `ArrayMap` for perfect hash join lookup execution. |
+### AggregateExec
+
+`AggregateExec` exposes the common `BaselineMetrics`, the operator-level
+metrics below, and a timer per aggregate expression and execution phase.
+
+| Metric | Description |
+| -------------------------- | ------------------------------------------------------------------------------------------------------------------------------------- |
+| time_calculating_group_ids | Time spent preparing group keys; see the note below for path-specific coverage. |
+| aggregate_arguments_time | Total time spent evaluating the inputs to the aggregate functions. `agg_expr_{index}_arguments_time` breaks this down per expression. |
+| aggregation_time | Time spent invoking accumulator `update` and `merge` operations. |
+| emitting_time | Time spent materializing group values and accumulator `state` or `evaluate` results. |
+| topk_maintenance_time | Time spent maintaining Grouped TopK's priority map, including batch setup, insertion, comparison, and NULL handling. |
+| skipped_aggregation_rows | Number of input rows passed through without aggregating them, when partial aggregation is skipped. |
+| reduction_factor | Rows emitted per row consumed by a partial aggregation, displayed as `66.67% (2/3)`. |
+| spill_count | Number of spill files written when the aggregation exceeds its memory budget. |
+| spilled_bytes | Total number of bytes written to spill files. |
+| spilled_rows | Total number of rows written to spill files. |
+| peak_mem_used | Peak tracked memory held by the grouped aggregation, in bytes; recorded by the fallback grouped hash path only. |
+
+These operator-level metrics are recorded by the grouped aggregation paths
+only: an `AggregateExec` without a `GROUP BY` reports just `BaselineMetrics`
+and the per-aggregate timers. `reduction_factor` and `skipped_aggregation_rows`
+are recorded in partial mode only. `skipped_aggregation_rows` is recorded only
+when partial-aggregation skipping is enabled for a single, non-grouping-sets
+`GROUP BY` whose input is not ordered by its grouping expressions; a
+`datafusion.execution.skip_partial_aggregation_probe_ratio_threshold` of `>= 1.0`
+disables the feature.
+
+`time_calculating_group_ids` covers both grouping-expression evaluation and
+resolving the resulting rows to group IDs, including interning and ordering
+setup. `aggregation_time` covers only accumulator `update` and `merge` calls.
+On accumulator-backed grouped aggregation paths, `state` and `evaluate` are
+included in `emitting_time`, including when partial aggregation materializes
+state under memory pressure. For an `AggregateExec` without a `GROUP BY`, the
+per-aggregate `state` and `evaluate` timers are recorded during
+`elapsed_compute`; it does not report grouped operator metrics such as
+`emitting_time`. Partial aggregation that skips aggregation reports
+`convert_to_state` instead of `aggregation_time` for those rows.
+
+When the specialized Grouped TopK path is selected for a limited grouped
+aggregate, it has no accumulators. Its group-key expression evaluation is
+included in `time_calculating_group_ids`, and its priority-map work is reported
+by `topk_maintenance_time`; it does not report accumulator phases. Planner
+selection is query-shape dependent: a limited `DISTINCT` query without
+ordering requirements uses the regular aggregate path instead.
+
+The per-aggregate timers are named `agg_expr_{index}_{phase}_time`, where
+`index` is the zero-based position of an aggregate expression in the operator
+and `phase` is one of the following:
+
+| Phase | Description |
+| ------------------ | --------------------------------------------------------------------------------------------- |
+| `arguments` | Evaluating the aggregate's argument expressions into input arrays. |
+| `update` | Updating an accumulator from raw input values. |
+| `merge` | Merging partial accumulator states. |
+| `state` | Obtaining an accumulator's intermediate state for partial output or aggregate spilling. |
+| `convert_to_state` | Converting raw aggregate inputs directly to partial state without normal accumulator updates. |
+| `evaluate` | Evaluating an accumulator to its final result. |
+
+For example, when a partial stage is planned, the partial `AggregateExec` for
+`SELECT SUM(a), SUM(b) FROM t` reports `agg_expr_0_arguments_time` and
+`agg_expr_0_update_time` for `SUM(a)`,
+and `agg_expr_1_arguments_time` and `agg_expr_1_update_time` for `SUM(b)`. The
+index is positional and refers to the same position in the `aggr=[...]` list
+printed on the operator's plan line, which is how an indexed timer is mapped
+back to an aggregate expression. Because the index is part of the metric name,
+otherwise identical functions over different columns stay distinct when
+per-partition metrics are combined.
+
+Each per-aggregate timer additionally carries an `aggregate` label holding the
+rendered aggregate expression (for example, `sum(t.a)`). Combining metrics
+across partitions drops labels, so this label is only shown in the "Plan with
+Full Metrics" section of `EXPLAIN ANALYZE VERBOSE`, which reports metrics per
+partition.
+
+`arguments` is recorded in every mode. The accumulator phases that are present
+depend on the aggregate mode and implementation. For non-grouped aggregation,
+partial mode records `update` and `state`, partial reduce mode records `merge`
+and `state`, final mode records `merge` and `evaluate`, and single mode records
+`update` and `evaluate`. Hash aggregation uses `update`, `state`, and
+`convert_to_state` in partial mode; `merge` and `state` in partial-reduce mode;
+`merge`, `state`, and `evaluate` in final mode; and `update`, `state`, `merge`,
+and `evaluate` in single mode. Its `state` timers measure intermediate-state
+emission, including during spilling. The grouped TopK aggregate path records
+only the per-aggregate `arguments` timer, because it maintains values directly
+rather than using accumulators.
+
+Where an aggregate has a `FILTER` clause, its evaluation is included in
+`aggregate_arguments_time`. Per-aggregate `arguments` timers include the filter
+when it is evaluated per aggregate. The legacy grouped hash path evaluates
+filters collectively, so its per-aggregate `arguments` timers cover argument
+expressions only; their sum need not equal `aggregate_arguments_time`.
+
+Except for the `Summary` metric `reduction_factor`, these operator-level and
+per-aggregate metrics are `Dev` metrics. They appear in `EXPLAIN ANALYZE` when
+`datafusion.explain.analyze_level` includes `Dev` (the default), but are omitted
+at the `Summary` level. The normal display combines partitions; use `EXPLAIN ANALYZE VERBOSE` to additionally show the per-partition values together with
+each per-aggregate timer's `aggregate` label. For a query
+such as the following, the per-expression metrics stay readable, and the
+operator's `aggr=[...]` list names the aggregate behind each timer index:
+
+```sql
+EXPLAIN ANALYZE
+SELECT k, SUM(a), SUM(b), COUNT(c)
+FROM t
+GROUP BY k;
+```
+
+The abbreviated partial-aggregate plan line below shows that mapping (timings
+and unrelated metrics are omitted):
+
+```text
+AggregateExec: mode=Partial, gby=[k@0 as k], aggr=[sum(t.a), sum(t.b), count(t.c)],
+ metrics=[..., agg_expr_0_arguments_time=..., agg_expr_1_arguments_time=...,
+ agg_expr_2_arguments_time=...]
+```
+
+Thus `agg_expr_0_*` is `sum(t.a)`, `agg_expr_1_*` is `sum(t.b)`, and
+`agg_expr_2_*` is `count(t.c)`. In verbose output, the corresponding
+`aggregate` labels provide the same mapping per partition.
+
## TODO
Add metrics for the remaining operators
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org