kosiew commented on PR #25051:
URL: https://github.com/apache/datafusion/pull/25051#issuecomment-5634850128
@jayzhan211
`c79745312f^..68945a9a20` removes the per-row metric work on both paths.
- `DistinctArrayAggAccumulator::merge_batch_impl` now calls the untimed
`update_batch_impl(..., false)` for each list-state row and records
`internal_distinct` around the complete direct merge once. The existing `merge`
timer remains the call-boundary timer; `internal_distinct` remains the
aggregate-owned deduplication submetric.
- Added `Accumulator::merge_batch_grouped`, matching `update_batch_grouped`.
The grouped adapter invokes this untimed entry point and owns the grouped
submetric timing, so grouped merge no longer starts a timer or atomically
records a duration per group/state row.
- `GroupsAccumulatorAdapter::convert_to_state` now calls
`update_batch_grouped`, not `update_batch`. It prepares factory/slice/filter
work outside the submetric, times only aggregate-owned deduplication, and emits
one accumulated metric update for the logical conversion batch. Preparation is
bounded in 64-row chunks, so timer reads scale with chunks rather than rows
while avoiding retention of all prepared arrays.
- The legacy grouped factory shares an `OnceLock` metric-handle cache. Only
its first accumulator receives `set_metrics` and resolves `metric("distinct")`;
conversion, update, and merge do not resolve it per row/group.
Regression coverage added:
- `distinct_accumulator_records_merge_metric_once`: a three-row
`List<Int32>` merge records one internal duration.
- `legacy_grouped_distinct_merge_records_metric_once`: grouped merge
resolves the metric once and records one duration for three state rows.
- `adapter_convert_to_state_records_metric_once` plus the legacy grouped
conversion/update test: conversion records once and metric lookup is not
repeated per conversion row.
- `adapter_convert_to_state_excludes_state_materialization_from_metric`:
verifies factory/input preparation/state materialization remain outside the
aggregate-owned submetric.
### Why the suggested shape was not adopted wholesale
- We intentionally do not use one uninterrupted `Instant` around all of
`convert_to_state`. That would either retain prepared slice/filter arrays for
the entire input batch, making memory proportional to row count, or include
factory, slice/filter preparation, `state`, and result materialization in an
aggregate-owned deduplication metric. The 64-row preparation chunks retain
bounded memory and time only accumulator invocation. The remaining timestamp
cost is once per chunk, rather than once per row, and the accumulated duration
is committed with one metric update per logical batch.
- We retain `agg_expr_N_internal_distinct_time` during merge. `merge` is the
aggregate call-boundary timer, while `internal_distinct` is the aggregate-owned
deduplication diagnostic. They intentionally overlap; the internal metric
complements the phase timer and is not an additive subdivision of it. Removing
merge-side `internal_distinct` would make that diagnostic depend on execution
phase rather than report all distinct-deduplication work. `metrics.md` now
explicitly says internal submetrics may overlap phase timers and must not be
added to phase timings as a breakdown.
[Benchmark results for 1M distinct
groups](https://github.com/apache/datafusion/pull/25051#issuecomment-5634652310)
--
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]
