Add aggregate-internal submetrics - #25051
Draft
kosiew wants to merge 32 commits into
Draft
Conversation
github-actions
Bot
added
documentation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25051 +/- ##
==========================================
+ Coverage 81.92% 81.94% +0.01%
==========================================
Files 1133 1133
Lines 423288 424363 +1075
Branches 423288 424363 +1075
==========================================
+ Hits 346797 347761 +964
- Misses 55902 55977 +75
- Partials 20589 20625 +36 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
kosiew
marked this pull request as ready for review
September 8, 2026 07:33
jayzhan211
reviewed
Sep 8, 2026
jayzhan211
left a comment
jayzhan211
left a comment
Contributor
There was a problem hiding this comment.
Thanks @kosiew , here is a suggestion:
DistinctArrayAggAccumulator has no GroupsAccumulator, so it always runs through GroupsAccumulatorAdapter. That adapter calls the factory once per new group and update_batch once per group per batch, which means this PR adds, per group: a parking_lot::Mutex lock + HashMap lookup + Arc clone on construction (via set_metrics -> metric("distinct")), and two Instant::now() calls plus an atomic fetch_add on every update_batch. The existing update timer only fires once per batch around the whole adapter call, so this is a new cost class for high-cardinality GROUP BY k, array_agg(DISTINCT v) with small groups.
There is also a reporting artifact: Time::add_duration clamps each addition to at least 1 ns, so with millions of one-row groups the internal_distinct time is inflated by ~1 ns per group per batch.
Could you run a benchmark against main for something like
SELECT k, array_agg(DISTINCT v) FROM t GROUP BY k
with ~1M distinct k and 1-2 rows per group, and post the numbers? If it regresses, one mitigation is to resolve the metric once in the factory closure rather than per accumulator, so set_metrics does not take the lock per group:
let agg_expr_captured = Arc::clone(agg_expr);
let factory = move || {
agg_expr_captured.create_accumulator_with_metrics(Arc::clone(&metrics))
};
becomes something where AggregateSubMetrics::metric is called once up front and the accumulator receives the resolved Arc<dyn AggregateMetric> (or a small pre-resolved struct). The per-batch Instant::now() pair is harder to avoid without restructuring; if the numbers show it matters, skipping the timer for batches below a small row threshold would keep the metric meaningful for the cases where it is actually informative.
kosiew
marked this pull request as draft
September 9, 2026 04:06
Closed
- Added harness implementation: - `benchmarks/sql_benchmarks/array_agg_distinct/array_agg_distinct.suite` – defines the benchmark suite, test parameters, and execution configuration for the data‑free SQL harness. - `benchmarks/sql_benchmarks/array_agg_distinct/benchmarks/q01.benchmark` – contains the specific query benchmark (`q01`) that exercises the `array_agg(DISTINCT …)` workload without requiring any input data. - Workload characteristics: - Simulates **2 M range rows** → **1 M groups**. - Each group contains **2 rows** with **2 distinct values**, providing a realistic yet data‑free test scenario for aggregation performance. - add bench.sh wrapper for array_agg_distinct
Reduce the data-free grouped array_agg(DISTINCT) workload while preserving its two-rows-per-group and two-distinct-values-per-group shape.
kosiew
force-pushed
the
aggmetrics-06-23570
branch
from
September 10, 2026 03:16
87e724f to
8550e2a
Compare
Contributor
Author
I amended the GroupsAccumulatorAdapter to time the entire grouped dispatch once per input batch, while each DistinctArrayAggAccumulator executes the same update logic without individually recording another timer.
I added a benchmark and rebased the benchmark to before adding internal submetrics so I can compare benchmark before vs after. empty commit is the mark right before the commits adding internal submetrics. I duplicated this branch to another PR and ran benchmark there. 1M distinct-> did not finish before repo killed it 100kIn this PR's benchmark run |
Contributor
Author
|
run benchmark sql env:
CARGO_BUILD_JOBS: 1
BENCH_NAME: array_agg_distinct
BENCH_QUERY: 1
baseline:
ref: "a997d8313e4c85c5a6ff14ada6876809592df42b"
changed:
ref: "7c3ebca0bebd388356d6c63103f1163b479f9766" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing 7c3ebca (7c3ebca) to a997d83 diff Run configurationrun benchmark sql
env:
BENCH_NAME: "array_agg_distinct"
BENCH_QUERY: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: "a997d8313e4c85c5a6ff14ada6876809592df42b"
changed:
ref: "7c3ebca0bebd388356d6c63103f1163b479f9766"Results will be posted here when complete File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing 7c3ebca (7c3ebca) to a997d83 diff Run configurationrun benchmark sql
env:
BENCH_NAME: "array_agg_distinct"
BENCH_QUERY: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: "a997d8313e4c85c5a6ff14ada6876809592df42b"
changed:
ref: "7c3ebca0bebd388356d6c63103f1163b479f9766"CPU Details (lscpu)Details
Resource Usagesql — base (merge-base)
sql — branch
File an issue against this benchmark runner |
kosiew
marked this pull request as ready for review
September 10, 2026 08:03
Contributor
|
@kosiew , here is another suggestion:
Two problems:
Suggest timing once at the batch boundary and using the untimed impl inside: 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 - 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 |
…comprehensive tests
- Introduce optional `AggregateMetric(s)` API with default no‑op setters for backward compatibility.
- Add lazy‑stable internal metrics: `agg_expr_{i}_internal_{subphase}_time` for fine‑grained phase tracking.
- Wire the metrics across all execution paths: stream, grouped, hash, ordered, and replay.
- Implement `array_agg(DISTINCT)` distinct‑timer to measure distinct‑aggregation latency.
- Extend test coverage:
- Partition merge scenarios.
- Repeated DISTINCT expression handling.
- Update documentation to reflect new API, metric naming, and wiring details.
…ove performance Cache one adapter per subphase; no repeated wrapper allocation - Introduce a single reusable adapter instance per subphase, eliminating the need to allocate multiple wrapper objects. This reduces memory churn and improves performance during metric collection. Skip `Arc` clone/clock read when no metric - Detect when there is no active metric to record and skip the unnecessary `Arc` clone and system clock reads. This lowers CPU overhead for subphases that don't emit metrics. Make submetric implementation details private - Move internal helpers and type-specific logic for submetrics behind `pub(super)` or module‑level privacy boundaries. This hides implementation details from external users, enhancing encapsulation and reducing the risk of misuse.
- Updated the test to verify that both `internal_distinct` timers are positive (`>0`) - This resolves the blocker where timers could be zero, causing test failures - Ensures correct initialization and behavior of the timer logic - Improves the reliability and confidence of timer‑related functionality
… partition display, and empty input behavior - Clarify identity/cardinality format as **(expr index, subphase, partition)** - Note that replacement accumulators share a single timer - Explain how partitions are combined in the normal display - Document that construction‑time requests can cause metrics to appear on empty input
- Counts the retained metric‑handle field in the accumulator’s size. - Adds a regression test to verify the size calculation under various inputs. - Updates the exact distinct‑size expectation to match the corrected behavior.
- Restored shared Time::add min‑1ns behavior. - Removed exact‑duration API/use. - Added regression test merge‑zero → recorded 1ns. - Updated submetric zero test.
…per‑batch 1 ns inflation - Time::add: legacy min‑1 ns merge unchanged. - Time::add_duration_exact: restored, scoped API. - Aggregate submetrics use exact adds → no per‑batch 1 ns inflation. - Tests cover both contracts.
…ile‑time test for DistinctArrayAggAccumulator - Updated `AggregateMetric` to implement the `RefUnwindSafe` trait, ensuring safe reference semantics during unwind operations. - Added a compile‑time test (`distinct_array_agg_accumulator_unwindsafe`) that verifies `DistinctArrayAggAccumulator` maintains both `UnwindSafe` and `RefUnwindSafe` guarantees, preventing panics in error‑recovery scenarios.
- Direct `GroupedHashAggregateStream` usage. - Multiple groups → adapter per‑group accumulators. - Asserts positive `agg_expr_0_internal_distinct_time`.
…mprove per-group metrics - Skip internal DISTINCT timing for batches <16 to reduce overhead. - Avoid per-group `Instant::now()`, metric `Arc` clone, and atomic updates for small batches. - Metrics are still recorded for batches >=16 to retain visibility where needed. - Added threshold tests to verify the new behavior.
…for accumulators This change introduces two new methods to the Accumulator trait: 1. `grouped_update_batch_metric`: Returns an optional metric that can be used to time grouped updates once per batch instead of per group. 2. `update_batch_grouped`: Updates state when called by a grouped accumulator adapter, with support for using the grouped update metric. The GroupsAccumulatorAdapter now uses these new methods to avoid timing every per-group call, recording one interval for the full batch instead. This reduces the overhead of metric collection when there are many groups. DistinctArrayAggAccumulator is updated to take advantage of the new grouped update functionality, skipping per-group timing for deduplication operations. This refactor improves performance for grouped aggregations by making metric collection proportional to batch count rather than group cardinality.
…tion clone - Replace direct `Option` clone with proper handling to avoid unnecessary allocations. - Switch to `datafusion_common::instant::Instant` for more accurate timing measurements. - Update related logic to ensure correctness and performance improvements.
…etric-count regressions - array_agg(DISTINCT) merge: one metric timer/state batch. - convert_to_state: untimed row updates + one conversion timer. - Added metric-count regressions.
…ng, bound prep chunks, time only accumulator calls, and emit one metric update per logical batch - Preps slice/filter outside submetric timing. - Times only accumulator calls. - Bounded 64‑group prep chunks; avoids unbounded retained arrays. - One metric update per logical batch.
…pped batch updates and adds regression test **Details:** - Bounded 64‑row prep chunks for `convert_to_state`. - Timer only wraps `update_batch_grouped`. - Excludes factory, slice/filter, `state/evaluate`, and result materialization steps. - Single metric update / logical conversion batch per operation. - Added RED→GREEN filtered multi‑row state‑materialization regression test.
…s, import AggregateMetrics - **Clear conditional metric lookup** – Refactored the conditional logic that retrieves metrics, removing redundant checks and making the code path easier to follow. - **Merged adjacent metrics impl blocks** – Consolidated neighboring implementation blocks that handled similar metric types, reducing duplication and improving maintainability. - **Imported AggregateMetrics; removed qualified repeats** – Added the `AggregateMetrics` import and eliminated unnecessary fully‑qualified references throughout the codebase.
kosiew
force-pushed
the
aggmetrics-06-23570
branch
from
September 11, 2026 08:24
7c3ebca to
68945a9
Compare
… propagate to accumulators - Updated `common.rs` to introduce a helper that constructs an `AggregateMetrics` owned by the execution context. - The helper returns an `Arc<dyn AggregateMetrics>` which is now passed to both the initial and replacement accumulator instances. - Ensures consistent metric aggregation ownership across accumulator updates, resolving the blocker.
kosiew
marked this pull request as draft
September 11, 2026 09:50
Contributor
Author
|
run benchmark sql env:
CARGO_BUILD_JOBS: 1
BENCH_NAME: array_agg_distinct
BENCH_QUERY: 1
baseline:
ref: "93f2dae"
changed:
ref: "eab374e" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing eab374e (eab374e) to 93f2dae diff Run configurationrun benchmark sql
env:
BENCH_NAME: "array_agg_distinct"
BENCH_QUERY: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: "93f2dae"
changed:
ref: "eab374e"Results will be posted here when complete File an issue against this benchmark runner |
- Internal submetrics may overlap enclosing phase timers. - They are supplementary diagnostics. - Do not add them as phase‑time breakdowns.
|
🤖 Benchmark completed (GKE) | trigger Instance: Comparing eab374e (eab374e) to 93f2dae diff Run configurationrun benchmark sql
env:
BENCH_NAME: "array_agg_distinct"
BENCH_QUERY: "1"
CARGO_BUILD_JOBS: "1"
baseline:
ref: "93f2dae"
changed:
ref: "eab374e"CPU Details (lscpu)Details
Resource Usagesql — base (merge-base)
sql — branch
File an issue against this benchmark runner |
Contributor
Author
|
Regression coverage added:
Why the suggested shape was not adopted wholesale
|
kosiew
marked this pull request as ready for review
September 11, 2026 13:18
Contributor
|
@kosiew , here is a suggestion:
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(())
})();
This also contradicts the comment three lines above ("slicing and filtering are adapter work, not aggregate-owned subphase work") and is inconsistent with Splitting into three passes keeps the one- - 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 |
…handling in GroupsAccumulatorAdapter The `sizes_pre` accumulation is now performed before the timer starts, preventing adapter work from being measured as part of the aggregate duration. The inner accumulation loop now breaks early on error and uses borrowed references, improving error propagation and reducing unnecessary copies.
- Added `groups_accumulator.rs` regression test file. - Introduced a sleeping `SlowSizeAccumulator` that delays `size()` by 50 ms. - Asserts that the grouped metric excludes the 50 ms `size()` delay, confirming correct grouping behavior.
kosiew
marked this pull request as draft
September 11, 2026 16:24
… add retry regression - Clear successful groups before propagating later-group error to prevent contaminating subsequent error handling. - Added retry regression test to catch duplicate update scenarios.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
This is the last of a series of PRs to close the issue
Rationale for this change
Existing per-aggregate
update,merge,state, andevaluatetimers attribute complete accumulator operations to an aggregate expression, but they cannot show how much time an aggregate spends in meaningful internal subphases.This makes it difficult to diagnose cases such as
array_agg(DISTINCT ...), where distinct-value handling can be a significant part of aggregate execution.This PR adds an optional aggregate-owned submetrics contract so aggregate implementations can expose internal timing while preserving stable ownership by aggregate expression and keeping the existing call-boundary metrics unchanged.
What changes are included in this PR?
This PR:
AggregateMetricandAggregateMetricsinterfaces and optionalset_metricshooks forAccumulatorandGroupsAccumulator.agg_expr_{index}_internal_{subphase}_timewith the owning aggregate expression attached through theaggregatelabel.array_agg(DISTINCT ...)with aninternal_distinctsubmetric covering its distinct-value processing.GroupsAccumulatorAdapterso grouped legacy accumulators can record aggregate-owned work once per bounded input chunk rather than once per group, while excluding adapter-owned filtering, slicing, size accounting, and state materialization from the submetric.Time::add_duration_exactso aggregate submetrics can accumulate measured durations without rounding each recording up to one nanosecond.array_agg_distinctSQL benchmark covering a high-cardinality groupedarray_agg(DISTINCT)workload.Are these changes tested?
Yes. The patch adds tests covering:
convert_to_statemetric recording;DistinctArrayAggAccumulatormetric recording for small and large batches;array_agg(DISTINCT ...)expressions retaining separate names and aggregate labels;array_agg(DISTINCT ...)internal metric;Timemetrics.The patch also adds a data-free SQL benchmark for 1M groups with two rows and two distinct values per group:
Are there any user-facing changes?
Yes.
EXPLAIN ANALYZEcan now expose optional aggregate-owned internal timing metrics when an aggregate requests them.For
array_agg(DISTINCT ...), the distinct-value processing time is reported as:agg_expr_{index}_internal_distinct_timewith an
aggregatelabel identifying the owning aggregate expression.These metrics complement the existing
update,merge,state, andevaluatetimers. They may overlap those enclosing timers and should not be added to them as a timing breakdown.Aggregates that do not request internal submetrics continue without additional internal metrics.
The accumulator traits also gain default metric hooks, so existing implementations are not required to implement them.
LLM-generated code disclosure
This PR includes LLM-generated code and comments. All LLM-generated content has been manually reviewed.