jayzhan211 commented on PR #25051:
URL: https://github.com/apache/datafusion/pull/25051#issuecomment-5636719616
**`invoke_per_accumulator`: `state.size()` is inside the timed region**
`groups_accumulator.rs:322-335` starts the timer, then calls `state.size()`
twice per group inside it:
```rust
let start = grouped_update_metric.as_ref().map(|_| Instant::now());
let chunk_result: Result<()> = (|| {
for (group_idx, values) in values_to_accumulate {
let state = &mut self.states[group_idx];
sizes_pre += state.size(); // <-- adapter accounting, timed
f(state.accumulator.as_mut(), &values)?;
state.indices.clear();
sizes_post += state.size(); // <-- adapter accounting, timed
}
Ok(())
})();
```
`AccumulatorState::size()` calls `Accumulator::size()`, and
`DistinctArrayAggAccumulator::size()` (array_agg.rs:1227) walks
`state.group_rows` summing `r.row().data().len()`, plus `converter.size()` and
`rows_buffer.size()`. That's two O(D) passes per group inside the timer, where
D is the group's distinct-set size, against O(rows-in-group) of actual distinct
work. As D grows across batches — precisely the `array_agg(DISTINCT)` case this
metric exists to diagnose — the accounting dominates the number being reported,
and it grows with result cardinality rather than with distinct work.
This also contradicts the comment three lines above ("slicing and filtering
are adapter work, not aggregate-owned subphase work") and is inconsistent with
`convert_to_state`, which deliberately keeps `state()` outside the timer and
has `adapter_convert_to_state_excludes_state_materialization_from_metric` to
prove it.
Splitting into three passes keeps the one-`Instant`-pair-per-chunk property
you were after:
```diff
- let start = grouped_update_metric.as_ref().map(|_|
Instant::now());
- let chunk_result: Result<()> = (|| {
- for (group_idx, values) in values_to_accumulate {
- let state = &mut self.states[group_idx];
- sizes_pre += state.size();
- f(state.accumulator.as_mut(), &values)?;
-
- // clear out the state so they are empty for next
- // iteration
- state.indices.clear();
- sizes_post += state.size();
- }
- Ok(())
- })();
+ // Size accounting is adapter work: keep it out of the
timer.
+ for (group_idx, _) in &values_to_accumulate {
+ sizes_pre += self.states[*group_idx].size();
+ }
+
+ let start = grouped_update_metric.as_ref().map(|_|
Instant::now());
+ let mut chunk_result = Ok(());
+ for (group_idx, values) in &values_to_accumulate {
+ chunk_result =
+ f(self.states[*group_idx].accumulator.as_mut(),
values);
+ if chunk_result.is_err() {
+ break;
+ }
+ }
if let Some(start) = start {
aggregate_duration += start.elapsed();
}
chunk_result?;
+
+ for (group_idx, _) in &values_to_accumulate {
+ let state = &mut self.states[*group_idx];
+ // clear out the state so they are empty for next
iteration
+ state.indices.clear();
+ sizes_post += state.size();
+ }
```
A regression test in the shape of
`adapter_convert_to_state_excludes_state_materialization_from_metric` — an
accumulator whose `size()` sleeps, asserting the recorded duration stays small
— would lock this in.
--
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]
