jayzhan211 commented on PR #25051:
URL: https://github.com/apache/datafusion/pull/25051#issuecomment-5618493056
**`update_batch`'s timer is paid per row on the merge and convert-to-state
paths**
`merge_batch` fans out to `self.update_batch(&[val])` once per state row
(array_agg.rs:1044), and `update_batch` → `update_batch_impl(values, true)`
does `Instant::now()` + `elapsed()` + an atomic add on each call. I confirmed
it with a probe: a 5-row `List<Int32>` state batch produces 5 metric recordings
inside a single `merge_batch`.
Two problems:
1. Metrics are always *collected* (`analyze_level` only gates display), so
`distinct_metric` is always `Some` and every Final/FinalPartitioned
`array_agg(DISTINCT)` merge now pays two clock reads plus an atomic per row.
That's the same per-row overhead the `grouped_update_batch_metric` /
`update_batch_grouped` split was added to remove on the update path — the merge
path just didn't get the treatment.
2. Merge time is recorded into `agg_expr_N_internal_distinct_time` while
also being counted by the existing `merge` timer. That contradicts
`metrics.md`: "Grouped accumulation records this once per input batch, rather
than once per group", and "complement the `update`, `merge`, `state`, and
`evaluate` timers rather than subdividing".
Suggest timing once at the batch boundary and using the untimed impl inside:
```diff
fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
if states.is_empty() {
return Ok(());
}
assert_eq_or_internal_err!(states.len(), 1, "expects single state");
+ // Time the whole merge once: the per-element calls below must not
each
+ // take a timestamp.
+ let distinct_metric = self.distinct_metric.clone();
+ let distinct_start = distinct_metric.as_ref().map(|_|
Instant::now());
+
// The DISTINCT state is `List<value>`.
- states[0]
+ let result = states[0]
.as_list::<i32>()
.iter()
.flatten()
- .try_for_each(|val| self.update_batch(&[val]))
+ .try_for_each(|val| self.update_batch_impl(&[val], false));
+
+ if let (Some(metric), Some(start)) = (distinct_metric,
distinct_start) {
+ metric.add_duration(start.elapsed());
+ }
+ result
}
```
The same shape exists in `GroupsAccumulatorAdapter::convert_to_state`
(functions-aggregate-common/src/aggregate/groups_accumulator.rs:478): it
builds a
fresh accumulator per row through the factory — so `set_metrics` →
`metric()` also
runs per row — and then calls `update_batch`, timing each row separately on
the
skip-partial-aggregation path.
```diff
- converted_accumulator.update_batch(&values_to_accumulate)?;
+ // Row-at-a-time conversion: use the untimed variant so this
path
+ // does not take a timestamp per row.
+
converted_accumulator.update_batch_grouped(&values_to_accumulate)?;
```
If you want convert-to-state time attributed to the subphase, hoist a single
`Instant` around the `for row_idx in 0..num_rows` loop using the metric from
the
first converted accumulator, the way `invoke_per_accumulator` does.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
