Conversation
github-actions
Bot
added
documentation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25051 +/- ##
==========================================
+ Coverage 81.90% 81.92% +0.01%
==========================================
Files 1134 1134
Lines 425261 426487 +1226
Branches 425261 426487 +1226
==========================================
+ Hits 348325 349399 +1074
- Misses 56295 56370 +75
- Partials 20641 20718 +77 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
jayzhan211
left a comment
jayzhan211
left a comment
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.
- 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.
87e724f to
8550e2a
Compare
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 |
|
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 , 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.
|
🤖 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 |
|
Regression coverage added:
Why the suggested shape was not adopted wholesale
|
|
@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.
… 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.
e6e4a85 moved
e7d2d7e added |
|
The chunked rewrite in Measured on this branch vs.
let result: Result<()> = (|| {
+ if grouped_update_metric.is_none() {
+ // Untimed fast path: identical to the pre-metrics loop.
+ for (&group_idx, offsets) in
+ groups_with_rows.iter().zip(offsets.windows(2))
+ {
+ let state = &mut self.states[group_idx];
+ sizes_pre += state.size();
+ let values_to_accumulate = slice_and_maybe_filter(
+ &values,
+ opt_filter.as_ref().map(|f| f.as_boolean()),
+ offsets,
+ )?;
+ f(state.accumulator.as_mut(), &values_to_accumulate)?;
+ let state = &mut self.states[group_idx];
+ state.indices.clear();
+ sizes_post += state.size();
+ }
+ return Ok(());
+ }
// Keep preparation bounded to avoid retaining one filtered array per
// group. Time only accumulator invocation: slicing and filtering
// are adapter work, not aggregate-owned subphase work.That's the exact patch I benchmarked. Note it also makes |
- GroupsAccumulatorAdapter: uninstrumented aggregates skip chunk/Vec path. - Removed dead `time_grouped_update`. - Timed submetric path unchanged. - `convert_to_state` unchanged.
Adds a new benchmark `groups_accumulator_adapter` to measure the performance of the uninstrumented update path of the `GroupsAccumulatorAdapter`. This benchmark is integrated into the existing benchmarks for `nth_value` and `approx_percentile_cont` to verify performance characteristics with grouped data (8192 groups).
80fd18b to
6a5c245
Compare
|
run benchmark groups_accumulator_adapter nth_value percentile_cont baseline:
ref: "93f2dae"
changed:
ref: "6a5c245" |
Implemented fast path |
4 participants
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 metrics measure complete
update,merge,state, andevaluatecalls, but they cannot show how much time an aggregate spends in meaningful internal work such as distinct-value handling.This PR adds an optional aggregate-owned submetrics contract so aggregate implementations can expose those internal phases while preserving stable ownership by aggregate expression and partition. Aggregates that do not request internal submetrics continue to use the existing accumulator paths without registering additional metrics.
array_agg(DISTINCT ...)is used as the representative implementation and reports time spent in distinct-value handling.What changes are included in this PR?
AggregateMetricandAggregateMetricsinterfaces and optional metric injection hooks forAccumulatorandGroupsAccumulator.agg_expr_{index}_internal_{subphase}_timeand the existingaggregatelabel.array_agg(DISTINCT ...)with aninternal_distincttimer for its distinct-value handling.GroupsAccumulatorAdapterso metric-enabled legacy accumulators can time grouped aggregate work once per bounded input chunk rather than once per group, while excluding adapter-owned filtering, slicing, size accounting, and state materialization from the internal timer.Time::add_duration_exactso repeated internal measurements do not round each zero-duration recording up to one nanosecond.array_agg(DISTINCT)SQL benchmark and focusedGroupsAccumulatorAdapter,nth_value, andapprox_percentile_contbenchmarks.Are these changes tested?
Yes. The patch adds tests covering:
array_agg(DISTINCT)metric recording for small and large update batches and merge batches.array_agg(DISTINCT ...)expressions.array_agg(DISTINCT)internal submetric.Timemerge behavior for a zero-duration measurement.The patch also adds benchmarks for the uninstrumented
GroupsAccumulatorAdapterpath, orderednth_value,approx_percentile_cont, and a 1M-group / two-row-per-grouparray_agg(DISTINCT)SQL workload.Are there any user-facing changes?
Yes.
EXPLAIN ANALYZEDev metrics can now include optional aggregate-owned internal timing metrics when an aggregate exposes them.These metrics use names of the form
agg_expr_{index}_internal_{subphase}_timeand carry the owning aggregate'saggregatelabel. Forarray_agg(DISTINCT ...), the new metric isagg_expr_{index}_internal_distinct_time.Internal submetrics supplement the existing
update,merge,state, andevaluatetimers; they are not a non-overlapping breakdown and should not be added to those phase timings.The metrics documentation is updated accordingly.
LLM-generated code disclosure
This PR includes LLM-generated code and comments. All LLM-generated content has been manually reviewed.