You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2026/10/08 19:52:08 UTC
[PR] Fix MSE group trimming for intermediate aggregation results [pinot]
xiangfu0 opened a new pull request, #19800:
URL: https://github.com/apache/pinot/pull/19800
MSE group trimming crashes when an ordered aggregate emits an intermediate sketch or set, such as `DISTINCTCOUNTHLL` or `DISTINCTCOUNTSMARTHLL`. The trim heap currently compares these objects directly as `Comparable`.
Compute final stored sort values once per group while retaining the original intermediate payloads for downstream merging. Copy funnel event queues before finalization because their finalizers consume the queue, and normalize raw sketch sort values to their final string representation.
### How to reproduce
Enable group trimming on an MSE aggregation following a join, with multiple groups and an aggregate in `ORDER BY`:
```sql
SET useMultiStageEngine = true;
SELECT /*+ joinOptions(join_strategy='lookup'),
aggOptions(is_enable_group_trim='true', mse_min_group_trim_size='1') */
f.groupKey,
DISTINCTCOUNTHLL(f.value) AS distinct_count
FROM facts f
JOIN dimension d ON f.joinKey = d.joinKey
GROUP BY f.groupKey
ORDER BY distinct_count DESC
LIMIT 1;
```
The ordered LEAF/INTERMEDIATE aggregation attempts to cast `HyperLogLog` to `Comparable`; the SmartHLL set path fails similarly. The focused regression reproduces this at both stages with six groups and two disjoint partials.
### Validation
- Before the fix, all eight distinct-count regression cases fail with set/sketch-to-`Comparable` errors.
- After the fix, `AggregateOperatorTest` and `MultistageGroupByExecutorTest`: 36 tests pass, including downstream merge counts, funnel event preservation, and raw HLL string ordering.
- Affected-module formatting, checkstyle, license formatting, and license checks pass.
```sh
./mvnw -pl pinot-query-runtime -am \
'-Dtest=AggregateOperatorTest,MultistageGroupByExecutorTest' \
-Dsurefire.failIfNoSpecifiedTests=false \
-Dcheckstyle.skip -Dspotless.check.skip -Dlicense.skip \
-Dshade.phase.prop=none \
-Dmaven.compiler.showDeprecation=true -Dmaven.compiler.showWarnings=true test
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4251199781
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/SetMergeStrategy.java:
##########
@@ -41,11 +42,13 @@ public List<Set> merge(List<Set> intermediateResult1, List<Set> intermediateResu
@Override
public LongArrayList extractFinalResult(List<Set> stepsSets) {
long[] result = new long[_numSteps];
- result[0] = stepsSets.get(0).size();
+ Set<?> firstStep = stepsSets.get(0);
+ Set<?> intersection = new HashSet<>(firstStep);
Review Comment:
Non-blocking: could we avoid copying the entire first-step population before an intersection is needed? A supported single-step set funnel previously finalized with one `size()` call; `new HashSet<>(firstStep)` now adds O(N) work and N hash entries, plus boxing when the source is a primitive fastutil set. Similarly, a two-step funnel with a million entrants and an empty second step now copies and removes all million entries, whereas the previous extraction did no member traversal. This also affects SSE and queries without group trimming.
We can preserve the intermediate state while reading the first count directly, skipping copying for single-step funnels, and constructing subsequent intersections from the smaller operand. Keeping primitive containers where practical would also avoid boxing.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4253915032
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -203,15 +258,15 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
int idx = 0;
while (idx < numGroups && groupKeyIterator.hasNext()) {
- Object[] row = getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes);
+ T row = decorate.apply(getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes));
sortedRows.add(row);
idx++;
}
while (groupKeyIterator.hasNext()) {
idx++;
// TODO: allocate new array row only if row enters set
- Object[] row = getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes);
+ T row = decorate.apply(getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes));
Review Comment:
Measured the object path before restructuring it, then fixed wrapper allocation in [1ca3cc9edc](https://github.com/apache/pinot/commit/1ca3cc9edc04a49b45236d02f63b2cf700109fec). Rejected candidates and entries polled from the heap are recycled, with their cached keys reset before reuse. This bounds `SortedRow` and key-array allocation to K+1 per trim; heap-owned entries are never modified and the original intermediate payloads remain mergeable.
| Aggregate | K | Before ms/op | After ms/op | Before B/op | After B/op |
|---|---:|---:|---:|---:|---:|
| AVG | 500 | 14.262 ± 6.078 | 7.525 ± 2.529 | 11,202,562 | 6,426,586 |
| AVG | 5,000 | 33.579 ± 29.211 | 16.044 ± 1.817 | 11,238,629 | 6,678,616 |
| Raw HLL | 500 | 1366.643 ± 146.105 | 1160.987 ± 24.530 | 10,406,050 | 5,630,094 |
| Raw HLL | 5,000 | 4282.882 ± 267.221 | 4182.569 ± 51.501 | 10,445,520 | 5,885,589 |
| SUM control | 500 | 4.227 ± 0.484 | 3.807 ± 0.099 | 4,002,223 | 4,002,221 |
| SUM control | 5,000 | 10.065 ± 0.464 | 8.819 ± 0.572 | 4,038,243 | 4,038,239 |
Total trim allocation falls 40.6–45.9% for the measured OBJECT cases. Native `SerializedHLL.compareTo()` still recomputes both operands' cardinalities: candidate raw-HLL trims take about 1.16s at K=500 and 4.18s at K=5,000 here. This change deliberately retains that native comparison contract; it does not add a public sketch cache or aggregation-function allowlist.
JMH 1.37, Temurin 25.0.4.1 on an Apple M4 Pro shared host; one thread, two forks, 3 × 1s warmup and 5 measured iterations per fork, GC profiler, 512 MiB initial/1 GiB maximum heap. Object measurements use 2s iterations; funnel/copy measurements use 1s. Errors shown are JMH 99.9% confidence errors.
The object fixture invokes production `getResult()` on 100,000 preaggregated groups, aggregate DESC then INT group ASC, with varied AVG counts and independent raw-HLL log2m=12 sketches containing 1–32 distinct inputs/group. Fixture creation, merging and correctness checks run in trial setup before warmup; the measured operation includes comparator construction, trimming and result draining. There are no allocating teardown checks. All 22 requested parameter cases completed and passed the independent result/preservation checks. The harness and raw results remain outside the PR.
These are trim/finalizer measurements, excluding upstream aggregation, planning, networking and whole-query QPS/p95/p99. The baseline AVG timings are noisy, so I am claiming the measured allocation reduction rather than a general latency speedup. SUM allocation is effectively unchanged; the observed control timings show no regression in these fixtures.
Validation: 67 focused MSE/funnel tests passed, as did all four affected-module `pinot-query-runtime` formatting/checkstyle/license checks. The new regression checks recycled-entry resets, ties, nulls, repeated trimming and payload preservation; a missing-reset mutant fails it. The fork branch and live PR head are verified at the pushed commit. Exact-head hosted CI is running. The recorded human approval is on the earlier `4292f86188` head; this follow-up has not received a fresh human review.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMaxStepAggregationFunction.java:
##########
@@ -51,6 +51,7 @@ public Integer extractFinalResult(@Nullable PriorityQueue<FunnelStepEvent> stepE
if (stepEvents == null || stepEvents.isEmpty()) {
return finalMaxStep;
}
+ stepEvents = new PriorityQueue<>(stepEvents);
Review Comment:
Completed the requested large-queue measurements for production MaxStep and MatchStep finalization at the pushed [1ca3cc9edc](https://github.com/apache/pinot/commit/1ca3cc9edc04a49b45236d02f63b2cf700109fec) head. The window-finalizer sources are unchanged from the measured `01a311b119` overlay. I kept the private queue copy because repeated extraction and subsequent merging must preserve the accumulator.
| Function | Events | First-window completion µs/op | Full traversal µs/op | Early B/op | Full B/op |
|---|---:|---:|---:|---:|---:|
| MAX_STEP | 10,000 | 1.330 ± 0.285 | 470.149 ± 18.146 | 40,160 | 40,163 |
| MAX_STEP | 100,000 | 8.430 ± 0.220 | 6241.274 ± 356.042 | 400,132 | 400,203 |
| MATCH_STEP | 10,000 | 0.914 ± 0.013 | 422.043 ± 12.989 | 40,160 | 40,219 |
| MATCH_STEP | 100,000 | 8.581 ± 0.400 | 5981.614 ± 117.375 | 400,160 | 400,258 |
The early case completes with events at timestamps 0, 1 and 2 in a window of 100, followed by a distant tail. It therefore terminates in the first three-event window while still copying all N references. The full case has no step-0 event, so it traverses the entire queue searching for a start; it is not a successful multi-window funnel.
Separately, the exact JDK constructor used by the finalizer (`new PriorityQueue<>(queue)`) measured:
| Queue events | Shallow copy µs/op | B/op |
|---:|---:|---:|
| 10,000 | 0.912 ± 0.093 | 40,048 |
| 100,000 | 8.454 ± 0.883 | 400,048 |
This confirms the O(N) shallow reference-copy allocation, including early completion; the events themselves are shared. The copy measurements are separate fixtures and should not be subtracted from finalization means to claim exact evaluator time.
JMH 1.37, Temurin 25.0.4.1 on an Apple M4 Pro shared host; one thread, two forks, 3 × 1s warmup and 5 measured iterations per fork, GC profiler, 512 MiB initial/1 GiB maximum heap. Object measurements use 2s iterations; funnel/copy measurements use 1s. Errors shown are JMH 99.9% confidence errors.
Fixtures are constructed before warmup. Setup checks two extractions and unchanged queue contents/event references after each; no teardown allocations contaminate the GC measurements. All eight finalizer cases and both isolated-copy cases completed. These are shared-host absolute costs, excluding real query resource-accounting/cancellation workloads and whole-query latency. The harness and raw results remain outside the PR.
Validation at the verified pushed head: 67 focused MSE/funnel tests passed, including repeated extraction followed by merging for all five window variants, plus all four affected-module checks. Exact-head hosted CI is running; the existing human approval remains on the earlier `4292f86188` head.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4253117012
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/BitmapMergeStrategy.java:
##########
@@ -41,11 +41,12 @@ public List<RoaringBitmap> merge(List<RoaringBitmap> intermediateResult1, List<R
@Override
public LongArrayList extractFinalResult(List<RoaringBitmap> stepsBitmaps) {
long[] result = new long[_numSteps];
- result[0] = stepsBitmaps.get(0).getCardinality();
+ RoaringBitmap intersection = stepsBitmaps.get(0).clone();
Review Comment:
Non-blocking cleanup: the bitmap path still clones the entire first population for single-step funnels or an empty second step. We can read the first cardinality directly, create a private intersection with `RoaringBitmap.and(first, second)` only when needed, then mutate that private result and stop once it is empty. This preserves the input state while avoiding unnecessary copies, including for the default and partitioned funnel callers.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMaxStepAggregationFunction.java:
##########
@@ -51,6 +51,7 @@ public Integer extractFinalResult(@Nullable PriorityQueue<FunnelStepEvent> stepE
if (stepEvents == null || stepEvents.isEmpty()) {
return finalMaxStep;
}
+ stepEvents = new PriorityQueue<>(stepEvents);
Review Comment:
Non-blocking performance follow-up for the window funnel finalizers: preserving the queue is necessary, but each extraction now copies all event references, even when evaluation can terminate early after finding a complete funnel. This is a shallow copy, not a copy of the events themselves. Please include large queues and early-completion cases in future object/funnel benchmarks; the current scalar benchmark does not measure this cost.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -203,15 +258,15 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
int idx = 0;
while (idx < numGroups && groupKeyIterator.hasNext()) {
- Object[] row = getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes);
+ T row = decorate.apply(getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes));
sortedRows.add(row);
idx++;
}
while (groupKeyIterator.hasNext()) {
idx++;
// TODO: allocate new array row only if row enters set
- Object[] row = getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes);
+ T row = decorate.apply(getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes));
Review Comment:
Non-blocking performance follow-up: the object path creates a `SortedRow` and sort-key array for every candidate, including candidates immediately discarded. Only top-K wrappers remain in the heap, but high group counts may produce substantial short-lived allocation. The scalar SUM benchmark does not cover this path. It would be useful to measure allocation and latency for object aggregates at high group counts before restructuring the shared heap loop. Also, caching a native raw-HLL wrapper avoids repeated extraction but does not cache its comparison cost: `SerializedHLL.compareTo()` still calls `cardinality()` on both operands.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4224821117
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
+ intermediateResult = new PriorityQueue<>(queue);
+ }
+ Object finalResult = aggFunction.extractFinalResult(intermediateResult);
+ sortKeys[index] = finalResult != null
Review Comment:
Addressed in `913f8c225d`. SUMPRECISION now keeps its native BigDecimal intermediate ordering, and finalized numeric sort keys are no longer converted through the SSE result metadata. The 5..9,100 regression fails on the prior PR head and now retains the 100 group. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
+ intermediateResult = new PriorityQueue<>(queue);
+ }
+ Object finalResult = aggFunction.extractFinalResult(intermediateResult);
+ sortKeys[index] = finalResult != null
+ ? TypeUtils.convert(finalResult, aggFunction.getFinalResultColumnType().getStoredType()) : null;
Review Comment:
Addressed in `913f8c225d`. Intermediate ordered trim now preserves all groups when an OBJECT aggregate has a non-numeric final sort type, including HISTOGRAM/ARRAY_AGG arrays. The HISTOGRAM regression reproduces the prior double[]-to-Comparable failure and now passes without finalizing the array key. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
+ intermediateResult = new PriorityQueue<>(queue);
+ }
+ Object finalResult = aggFunction.extractFinalResult(intermediateResult);
Review Comment:
Addressed in `913f8c225d`. Intermediate raw-sketch STRING orders now bypass trimming and retain every sketch for downstream merging and sorting. This avoids selecting groups by a partial serialization or changing the requested final STRING ordering to estimate ordering. The raw-HLL regression now verifies that all six partials survive a trim capacity of five. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
Review Comment:
Addressed in `913f8c225d`. Removed the PriorityQueue copy guard. Funnel sort keys now bypass intermediate trimming, so trim never consumes their live event state. The expanded six-group funnel regression verifies all events survive to FINAL. A broader non-destructive AggregationFunction contract and the pre-existing SSE path are separate from this minimal MSE bug fix. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,14 +188,73 @@ public void processBlock(MseBlock.Data block) {
}
}
+ /// Get aggregation result limited to first `maxRows` rows, ordered with `collations`.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numGroups = Math.min(_groupIdGenerator.getNumGroups(), maxRows);
Review Comment:
Addressed in `913f8c225d`. Added the early return when genuinely intermediate output has no more groups than the trim capacity. It uses the ordinary result path without sorting/finalization; DIRECT, FINAL and leafReturnFinalResult output keep their existing heap behavior. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4224478079
##########
pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/AggregateOperatorTest.java:
##########
@@ -351,6 +353,133 @@ public void testGroupTrimHintOverridesContextValue() {
assertEquals(getAggregateOperator(context, nodeHint, 5, List.of(new RelFieldCollation(1))).getGroupTrimSize(), 30);
}
+ @DataProvider
+ public Object[][] orderedGroupTrimCases() {
+ return new Object[][]{
+ {"COUNT", null, AggType.LEAF},
+ {"COUNT", null, AggType.INTERMEDIATE},
+ {"DISTINCTCOUNT", null, AggType.LEAF},
+ {"DISTINCTCOUNT", null, AggType.INTERMEDIATE},
+ {"DISTINCTCOUNTHLL", null, AggType.LEAF},
+ {"DISTINCTCOUNTHLL", null, AggType.INTERMEDIATE},
+ {"DISTINCTCOUNTSMARTHLL", null, AggType.LEAF},
+ {"DISTINCTCOUNTSMARTHLL", null, AggType.INTERMEDIATE},
+ {"DISTINCTCOUNTSMARTHLL", "threshold=1", AggType.LEAF},
+ {"DISTINCTCOUNTSMARTHLL", "threshold=1", AggType.INTERMEDIATE}
+ };
+ }
+
+ @Test(dataProvider = "orderedGroupTrimCases")
+ public void testOrderedGroupTrimPreservesIntermediateResults(String functionName, @Nullable String parameters,
+ AggType trimmedStage) {
+ OpChainExecutionContext context =
+ OperatorTestUtil.getContext(Map.of(QueryOptionKey.MSE_MIN_GROUP_TRIM_SIZE, "1"));
+ ColumnDataType finalType = functionName.equals("DISTINCTCOUNTHLL") || functionName.equals("COUNT")
+ ? ColumnDataType.LONG : INT;
Review Comment:
**Minor convention:** per the repo coding conventions (AGENTS.md, linked from CLAUDE.md): "Keep a ternary expression on one line when it fits. Otherwise, put `? <true-expression>` and `: <false-expression>` on separate lines." This ternary (and the one at `MultistageGroupByExecutor.java:308-309`, `? TypeUtils.convert(...) : null`) does not fit on one line, so the `:` branch belongs on its own line.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4225334254
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,15 +192,74 @@ public void processBlock(MseBlock.Data block) {
}
}
- /// Get aggregation result limited to first `maxRows` rows, ordered with `comparator`.
- public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
+ /// Select up to `maxRows` groups using `collations`, preserving all intermediate groups when trim is unsafe.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numKeys = _groupKeyIds.length;
+ int numCollations = collations.size();
+ AggregationFunction[] sortFunctions = new AggregationFunction[numCollations];
+ boolean finalizeSortValues = false;
+ if (_aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult) {
+ if (_groupIdGenerator.getNumGroups() <= maxRows) {
Review Comment:
Minor (test gap): this early-return branch — the common production case, since the default trim capacity is max(5*limit, 5000) and servers routinely hold fewer groups than that — is not exercised by any of the new tests: every ordered-trim test uses 6 groups against a trim capacity of 5, so `numGroups <= maxRows` is always false. A future edit that breaks this branch (e.g. moving it below the collation scan so funnel/sketch intermediates get finalized when groups are few, or returning sorted-but-finalized rows) would pass the whole suite. One parameterized case with trim capacity >= group count (e.g. `MSE_MIN_GROUP_TRIM_SIZE` large enough that `getTableCapacity` exceeds 6) asserting all groups survive with intact intermediates would pin it.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,15 +192,74 @@ public void processBlock(MseBlock.Data block) {
}
}
- /// Get aggregation result limited to first `maxRows` rows, ordered with `comparator`.
- public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
+ /// Select up to `maxRows` groups using `collations`, preserving all intermediate groups when trim is unsafe.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numKeys = _groupKeyIds.length;
+ int numCollations = collations.size();
+ AggregationFunction[] sortFunctions = new AggregationFunction[numCollations];
+ boolean finalizeSortValues = false;
+ if (_aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult) {
+ if (_groupIdGenerator.getNumGroups() <= maxRows) {
+ return getResult(maxRows);
+ }
+ for (int i = 0; i < numCollations; i++) {
+ int index = collations.get(i).getFieldIndex();
+ if (index < numKeys) {
+ continue;
+ }
+ AggregationFunction aggFunction = _aggFunctions[index - numKeys];
+ if (aggFunction.getIntermediateResultColumnType().getStoredType() != ColumnDataType.OBJECT) {
+ continue;
+ }
+ // These intermediate objects already compare by the aggregate's numeric value.
+ switch (aggFunction.getType()) {
+ case AVG, AVGMV, MINMAXRANGE, MINMAXRANGEMV, SUMPRECISION:
+ continue;
+ default:
+ break;
+ }
+ // Serialized sketches and array results lack a safe scalar trim order. Funnel finalization consumes events.
+ // Preserve every group for the downstream aggregate and sort instead of finalizing these intermediates.
+ if (!aggFunction.getFinalResultColumnType().getStoredType().isNumber()
+ || aggFunction.getType().name().startsWith("FUNNEL")) {
+ return getResult(Integer.MAX_VALUE);
+ }
+ sortFunctions[i] = aggFunction;
+ finalizeSortValues = true;
+ }
+ }
+ if (!finalizeSortValues) {
+ return getResult(new SortUtils.SortComparator(collations, true), maxRows, Function.identity(),
+ Function.identity());
+ }
+ SortUtils.SortComparator[] comparators = new SortUtils.SortComparator[numCollations];
+ for (int i = 0; i < numCollations; i++) {
+ comparators[i] = new SortUtils.SortComparator(List.of(collations.get(i).withFieldIndex(i)), true);
+ }
+ Comparator<SortedRow> comparator = (r1, r2) -> {
+ for (int i = 0; i < numCollations; i++) {
+ int index = collations.get(i).getFieldIndex();
Review Comment:
Nit (hot path): the lambda re-evaluates `collations.get(i).getFieldIndex()` on every comparator invocation, for every collation level — O(numGroups log trimSize) compares against up to `numGroupsLimit` (100K) groups. The indexes are fixed when the comparator is built; precomputing them next to `sortFunctions` removes two virtual calls per level per compare:
```suggestion
int[] sortColumnIndexes = new int[numCollations];
for (int i = 0; i < numCollations; i++) {
sortColumnIndexes[i] = collations.get(i).getFieldIndex();
}
Comparator<SortedRow> comparator = (r1, r2) -> {
for (int i = 0; i < numCollations; i++) {
int index = sortColumnIndexes[i];
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4225381031
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,15 +192,74 @@ public void processBlock(MseBlock.Data block) {
}
}
- /// Get aggregation result limited to first `maxRows` rows, ordered with `comparator`.
- public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
+ /// Select up to `maxRows` groups using `collations`, preserving all intermediate groups when trim is unsafe.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numKeys = _groupKeyIds.length;
+ int numCollations = collations.size();
+ AggregationFunction[] sortFunctions = new AggregationFunction[numCollations];
+ boolean finalizeSortValues = false;
+ if (_aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult) {
+ if (_groupIdGenerator.getNumGroups() <= maxRows) {
Review Comment:
Addressed in `725e9e2084`. Added one parameterized DISTINCTCOUNTHLL LEAF case with six groups and MSE_MIN_GROUP_TRIM_SIZE=10. It exercises the early return and verifies all six intermediate groups survive and merge across disjoint partials to cardinalities 2 through 12. The existing ten cases still exercise actual trimming. Validation: all 41 focused runtime tests and all four affected-module formatting/style/license checks pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,15 +192,74 @@ public void processBlock(MseBlock.Data block) {
}
}
- /// Get aggregation result limited to first `maxRows` rows, ordered with `comparator`.
- public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
+ /// Select up to `maxRows` groups using `collations`, preserving all intermediate groups when trim is unsafe.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numKeys = _groupKeyIds.length;
+ int numCollations = collations.size();
+ AggregationFunction[] sortFunctions = new AggregationFunction[numCollations];
+ boolean finalizeSortValues = false;
+ if (_aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult) {
+ if (_groupIdGenerator.getNumGroups() <= maxRows) {
+ return getResult(maxRows);
+ }
+ for (int i = 0; i < numCollations; i++) {
+ int index = collations.get(i).getFieldIndex();
+ if (index < numKeys) {
+ continue;
+ }
+ AggregationFunction aggFunction = _aggFunctions[index - numKeys];
+ if (aggFunction.getIntermediateResultColumnType().getStoredType() != ColumnDataType.OBJECT) {
+ continue;
+ }
+ // These intermediate objects already compare by the aggregate's numeric value.
+ switch (aggFunction.getType()) {
+ case AVG, AVGMV, MINMAXRANGE, MINMAXRANGEMV, SUMPRECISION:
+ continue;
+ default:
+ break;
+ }
+ // Serialized sketches and array results lack a safe scalar trim order. Funnel finalization consumes events.
+ // Preserve every group for the downstream aggregate and sort instead of finalizing these intermediates.
+ if (!aggFunction.getFinalResultColumnType().getStoredType().isNumber()
+ || aggFunction.getType().name().startsWith("FUNNEL")) {
+ return getResult(Integer.MAX_VALUE);
+ }
+ sortFunctions[i] = aggFunction;
+ finalizeSortValues = true;
+ }
+ }
+ if (!finalizeSortValues) {
+ return getResult(new SortUtils.SortComparator(collations, true), maxRows, Function.identity(),
+ Function.identity());
+ }
+ SortUtils.SortComparator[] comparators = new SortUtils.SortComparator[numCollations];
+ for (int i = 0; i < numCollations; i++) {
+ comparators[i] = new SortUtils.SortComparator(List.of(collations.get(i).withFieldIndex(i)), true);
+ }
+ Comparator<SortedRow> comparator = (r1, r2) -> {
+ for (int i = 0; i < numCollations; i++) {
+ int index = collations.get(i).getFieldIndex();
Review Comment:
Addressed in `725e9e2084`. Precomputed sortColumnIndexes alongside the per-field comparators and now read the cached int indexes during heap comparisons. This removes repeated collation list/field lookups from the object comparator; the scalar heap path is unchanged. Validation: all 41 focused runtime tests and all four affected-module formatting/style/license checks pass.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4224822309
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,14 +188,73 @@ public void processBlock(MseBlock.Data block) {
}
}
+ /// Get aggregation result limited to first `maxRows` rows, ordered with `collations`.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numGroups = Math.min(_groupIdGenerator.getNumGroups(), maxRows);
+ if (numGroups == 0) {
+ return List.of();
+ }
+
+ int numKeys = _groupKeyIds.length;
+ int numFunctions = _aggFunctions.length;
+ int[] columnsToFinalize = _aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult
+ ? collations.stream().mapToInt(RelFieldCollation::getFieldIndex).filter(index -> index >= numKeys)
+ .filter(index -> {
+ AggregationFunction aggFunction = _aggFunctions[index - numKeys];
+ ColumnDataType intermediateStoredType = aggFunction.getIntermediateResultColumnType().getStoredType();
+ return intermediateStoredType == ColumnDataType.OBJECT
Review Comment:
Addressed in `913f8c225d`. AVG/AVGMV and MINMAXRANGE/MINMAXRANGEMV now retain the original raw-row heap using their natural Comparable intermediates, along with SUMPRECISION. They allocate no per-group sort wrappers or boxed final keys. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
Review Comment:
Addressed in `913f8c225d`. Sort-key arrays now contain only ORDER BY columns, and values are finalized lazily when their collation is actually compared. Earlier keys short-circuit later finalization, and each cache also distinguishes an uninitialized key from a finalized null. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,14 +188,73 @@ public void processBlock(MseBlock.Data block) {
}
}
+ /// Get aggregation result limited to first `maxRows` rows, ordered with `collations`.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numGroups = Math.min(_groupIdGenerator.getNumGroups(), maxRows);
+ if (numGroups == 0) {
+ return List.of();
+ }
+
+ int numKeys = _groupKeyIds.length;
+ int numFunctions = _aggFunctions.length;
+ int[] columnsToFinalize = _aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult
+ ? collations.stream().mapToInt(RelFieldCollation::getFieldIndex).filter(index -> index >= numKeys)
+ .filter(index -> {
+ AggregationFunction aggFunction = _aggFunctions[index - numKeys];
+ ColumnDataType intermediateStoredType = aggFunction.getIntermediateResultColumnType().getStoredType();
+ return intermediateStoredType == ColumnDataType.OBJECT
+ || intermediateStoredType != aggFunction.getFinalResultColumnType().getStoredType();
+ }).distinct().toArray()
+ : new int[0];
+ SortUtils.SortComparator sortComparator = new SortUtils.SortComparator(collations, true);
+ // Scalar intermediate values already have their final stored representation; keep the original heap path.
+ if (columnsToFinalize.length == 0) {
+ return getResult(sortComparator, maxRows);
+ }
+ Comparator<SortedRow> comparator = (r1, r2) -> sortComparator.compare(r1.sortKeys(), r2.sortKeys());
Review Comment:
Addressed in `913f8c225d`. Both paths now share one generic top-K heap loop. The raw-row path uses Function.identity(), while the object path decorates and unwraps compact cached sort rows. Heap selection, draining and rowsProcessed accounting are implemented once. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
Review Comment:
Addressed in `913f8c225d`. Removed the unreachable empty-finalization branch while replacing getSortedRow with the compact lazy cache. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
##########
pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/AggregateOperatorTest.java:
##########
@@ -351,6 +353,133 @@ public void testGroupTrimHintOverridesContextValue() {
assertEquals(getAggregateOperator(context, nodeHint, 5, List.of(new RelFieldCollation(1))).getGroupTrimSize(), 30);
}
+ @DataProvider
Review Comment:
Addressed in `913f8c225d`. Replaced the five INTERMEDIATE parameter rows with reachable LEAF leafReturnFinalResult=true cases. The ten rows now cover both LEAF output modes, followed by FINAL merging, without increasing the parameterized test count. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4225333128
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,14 +188,73 @@ public void processBlock(MseBlock.Data block) {
}
}
+ /// Get aggregation result limited to first `maxRows` rows, ordered with `collations`.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numGroups = Math.min(_groupIdGenerator.getNumGroups(), maxRows);
+ if (numGroups == 0) {
+ return List.of();
+ }
+
+ int numKeys = _groupKeyIds.length;
+ int numFunctions = _aggFunctions.length;
+ int[] columnsToFinalize = _aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult
+ ? collations.stream().mapToInt(RelFieldCollation::getFieldIndex).filter(index -> index >= numKeys)
+ .filter(index -> {
+ AggregationFunction aggFunction = _aggFunctions[index - numKeys];
+ ColumnDataType intermediateStoredType = aggFunction.getIntermediateResultColumnType().getStoredType();
+ return intermediateStoredType == ColumnDataType.OBJECT
Review Comment:
Re-review of 913f8c2: the allowlist restores the raw-heap path as requested, and `AvgPair`/`MinMaxRangePair`/`BigDecimal` natural order does match the final numeric order for all normal values — resolved for the reported cost issue. One caveat the switch comment ("already compare by the aggregate's numeric value") glosses over: `AvgPair.compareTo` and `MinMaxRangePair.compareTo` compare with raw `>`/`<` doubles, so a NaN average/range compares equal (0) to every other group. That makes the trim heap's comparator non-transitive when any group's sum is NaN (e.g. a DOUBLE column containing NaN), so which groups survive the leaf trim becomes insertion-order dependent, while the downstream FINAL sort uses `Double.compareTo`, which ranks NaN greatest — a NaN group that belongs in the top-K under `ORDER BY AVG(x) DESC` can be silently trimmed. The previous head's finalized `Double` sort keys had the total order; this head trades that back for zero allocation (and matches pre-PR maste
r behavior, so this is an edge-case note rather than a blocker). If you want the allowlist watertight, fixing `AvgPair`/`MinMaxRangePair.compareTo` to use `Double.compare` would align all three orderings without touching this executor.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #19800:
URL: https://github.com/apache/pinot/pull/19800#issuecomment-6089902563
Addressed [this review](https://github.com/apache/pinot/pull/19800#pullrequestreview-5475272499) in [c4513289bd](https://github.com/apache/pinot/commit/c4513289bd160d84b1518a9a6c7e575a48284821).
Removed the aggregate-name special cases and numeric-only restriction. Intermediate OBJECT aggregate keys now use lazy, cached native `extractFinalResult()` values, following SSE, while retaining the original mergeable rows. Group keys, scalar intermediates and already-final values keep their existing heap path; the no-trim early return remains.
Moved extraction safety into the funnel implementations: all five window finalizers consume private queue copies, and bitmap/set counts use private cumulative intersections. The group-by extraction contract now documents preservation of intermediate information for later merging.
The comparison boundary is explicit in the code and PR description: native trimming compares SUMPRECISION numerically as BigDecimal, raw HLL by cardinality, and histogram/funnel lists lexicographically. Final MSE output conversion still yields hexadecimal strings for raw HLL (downstream ORDER BY is lexical) and primitive arrays (generic downstream sorting remains unsupported). This follows SSE's approximate intermediate candidate ranking; it does not establish exact final MSE ORDER BY equivalence for converted outputs. Aligning those downstream semantics requires carrying aggregate ordering information through planning/exchanges, beyond this fix.
Validation: 58 focused tests pass (41 MSE, 17 funnel); all seven new extraction/merge cases fail on the prior funnel implementations. Both affected modules pass all four formatting/checkstyle/license checks, with no compiler warnings on changed code. The pushed PR head is verified; hosted CI for this commit is queued/running, and human approval remains pending. The previous scalar SUM measurements do not cover the object/funnel path.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4253416921
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/BitmapMergeStrategy.java:
##########
@@ -41,11 +41,12 @@ public List<RoaringBitmap> merge(List<RoaringBitmap> intermediateResult1, List<R
@Override
public LongArrayList extractFinalResult(List<RoaringBitmap> stepsBitmaps) {
long[] result = new long[_numSteps];
- result[0] = stepsBitmaps.get(0).getCardinality();
+ RoaringBitmap intersection = stepsBitmaps.get(0).clone();
Review Comment:
Fixed in [01a311b119](https://github.com/apache/pinot/commit/01a311b11948fc4b23909e51af1daaef7e12f227). The first cardinality is read directly; single-step funnels and empty operands do not copy or intersect populations. The first nonempty intersection uses `RoaringBitmap.and()`, later steps reuse that private result, and empty intersections stop evaluation. Input bitmaps remain intact for both default and partitioned callers.
Local validation: 27 focused finalizer/null-handling/bitmap/partitioned/partitioned-sorted tests passed, including repeated extraction followed by merging. All four `pinot-core` formatting/checkstyle/license checks passed, with no warnings on added lines. Two regressions cover no-clone single/empty cases, later private intersections, unchanged populations and stopping before later steps; both fail against the preceding `4292f86188` finalizer and pass with this change.
The push is verified. Hosted CI is running for the new head. The approval at `4292f86188` remains recorded; this follow-up commit has not had a new human review.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -203,15 +258,15 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
int idx = 0;
while (idx < numGroups && groupKeyIterator.hasNext()) {
- Object[] row = getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes);
+ T row = decorate.apply(getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes));
sortedRows.add(row);
idx++;
}
while (groupKeyIterator.hasNext()) {
idx++;
// TODO: allocate new array row only if row enters set
- Object[] row = getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes);
+ T row = decorate.apply(getRow(groupKeyIterator, numKeys, numFunctions, resultStoredTypes));
Review Comment:
Agreed. The object-sort path allocates a `SortedRow` and key array for each candidate, including discarded candidates. Lazy caching avoids repeated extraction for reached collations; `SerializedHLL.compareTo()` still calls `cardinality()` on both operands. The scalar SUM benchmark supplies no allocation or latency evidence for these object costs.
I have kept the shared heap loop unchanged in this bug fix. Following your non-blocking recommendation, object measurements remain a follow-up: high group counts, varying top-K, allocation and latency, including raw HLL. Leaving this thread open for that measurement rather than marking it fixed.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelMaxStepAggregationFunction.java:
##########
@@ -51,6 +51,7 @@ public Integer extractFinalResult(@Nullable PriorityQueue<FunnelStepEvent> stepE
if (stepEvents == null || stepEvents.isEmpty()) {
return finalMaxStep;
}
+ stepEvents = new PriorityQueue<>(stepEvents);
Review Comment:
Agreed. Nonempty window funnels make a shallow O(N) copy of event references before MaxStep/MatchStep can terminate on completion, preserving the accumulator for later extraction and merging. The events themselves are reused. The scalar benchmark does not measure this cost.
Following your non-blocking recommendation, future funnel measurements should include large queues completing in the first window and queues requiring full evaluation, with first-window size considered separately from total queue size. Leaving this thread open for those measurements.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4225331983
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
Review Comment:
Re-review of 913f8c2: the reachable funnel bug is fixed (FUNNEL* now bypasses trim entirely), and the heap no longer finalizes live funnel state. One residual on this thread: the check is a fail-unsafe denylist. `cacheSortValue` finalizes the live intermediate that stays in the output row, and the only guard against a destructive finalizer is `getType().name().startsWith("FUNNEL")`. There is already a concrete non-FUNNEL counterexample in the codebase: `DistinctCountOffHeapAggregationFunction.extractFinalResult` calls `set.close()` (frees native memory) on an OBJECT intermediate with INT final type, so it passes every gate here. It is unreachable today only because its `aggregateGroupBySV`/`extractGroupByResult` throw `UnsupportedOperationException` — if group-by support lands (or any plugin adds a destructive numeric-final finalizer), ordered trim silently corrupts (here, use-after-free) with no compile error or test failure. Since the untrimmed fallback is always correct, cons
ider failing safe at zero cost to current behavior: route unknown OBJECT intermediates to `getResult(Integer.MAX_VALUE)` unless the function is known safe to finalize (invert to an allowlist), or at minimum add DISTINCTCOUNTOFFHEAP next to the FUNNEL check with a comment naming the invariant.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19800:
URL: https://github.com/apache/pinot/pull/19800#issuecomment-6069279202
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19800?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 100.00%. Comparing base ([`6f87b71`](https://app.codecov.io/gh/apache/pinot/commit/6f87b7150f40464f455e1a52b2c2a6184f49ccb7?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`cc68cab`](https://app.codecov.io/gh/apache/pinot/commit/cc68cabe99baf1c48e3128556a85e16cfb022a70?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19800 +/- ##
==============================================
+ Coverage 68.54% 100.00% +31.45%
+ Complexity 1486 6 -1480
==============================================
Files 3525 3 -3522
Lines 229852 6 -229846
Branches 36507 0 -36507
==============================================
- Hits 157558 6 -157552
+ Misses 59965 0 -59965
+ Partials 12329 0 -12329
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (ø)` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (ø)` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (+31.45%)` | :arrow_up: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (ø)` | |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ø> (+31.45%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19800/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19800?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4224823585
##########
pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/AggregateOperatorTest.java:
##########
@@ -351,6 +353,133 @@ public void testGroupTrimHintOverridesContextValue() {
assertEquals(getAggregateOperator(context, nodeHint, 5, List.of(new RelFieldCollation(1))).getGroupTrimSize(), 30);
}
+ @DataProvider
+ public Object[][] orderedGroupTrimCases() {
+ return new Object[][]{
+ {"COUNT", null, AggType.LEAF},
+ {"COUNT", null, AggType.INTERMEDIATE},
+ {"DISTINCTCOUNT", null, AggType.LEAF},
+ {"DISTINCTCOUNT", null, AggType.INTERMEDIATE},
+ {"DISTINCTCOUNTHLL", null, AggType.LEAF},
+ {"DISTINCTCOUNTHLL", null, AggType.INTERMEDIATE},
+ {"DISTINCTCOUNTSMARTHLL", null, AggType.LEAF},
+ {"DISTINCTCOUNTSMARTHLL", null, AggType.INTERMEDIATE},
+ {"DISTINCTCOUNTSMARTHLL", "threshold=1", AggType.LEAF},
+ {"DISTINCTCOUNTSMARTHLL", "threshold=1", AggType.INTERMEDIATE}
+ };
+ }
+
+ @Test(dataProvider = "orderedGroupTrimCases")
+ public void testOrderedGroupTrimPreservesIntermediateResults(String functionName, @Nullable String parameters,
+ AggType trimmedStage) {
+ OpChainExecutionContext context =
+ OperatorTestUtil.getContext(Map.of(QueryOptionKey.MSE_MIN_GROUP_TRIM_SIZE, "1"));
+ ColumnDataType finalType = functionName.equals("DISTINCTCOUNTHLL") || functionName.equals("COUNT")
+ ? ColumnDataType.LONG : INT;
Review Comment:
Addressed in `913f8c225d`. Reformatted the multiline test ternaries with separate ? and : branches. The executor conversion ternary was removed with the SSE sort-key normalization; the remaining short ternary fits on one line. Validation: all 40 focused runtime tests and the four affected-module hygiene checks pass.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4224476613
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
+ intermediateResult = new PriorityQueue<>(queue);
+ }
+ Object finalResult = aggFunction.extractFinalResult(intermediateResult);
+ sortKeys[index] = finalResult != null
+ ? TypeUtils.convert(finalResult, aggFunction.getFinalResultColumnType().getStoredType()) : null;
Review Comment:
**ClassCastException for array-typed final results (narrow regression).** For aggregates whose final stored type is a primitive array (`HISTOGRAM`, non-distinct `ARRAY_AGG` → `DOUBLE_ARRAY`/`*_ARRAY`), the finalized sort key becomes `double[]`/`long[]` via `TypeUtils.convert`, and `SortUtils.SortComparator` then does `((Comparable) v1).compareTo(v2)` → `ClassCastException`. Pre-PR the heap compared the fastutil `DoubleArrayList`/`IntArrayList` intermediates, which implement `Comparable`, so e.g. `ORDER BY COUNT(*) DESC, HISTOGRAM(x,0,10,5) LIMIT 10` with group trim completed the leaf trim and could succeed end-to-end when merged counts never tie at FINAL; now the leaf throws whenever the comparator reaches the array column (tie on the earlier key, or array column first). Consider excluding non-Comparable final stored types from `columnsToFinalize` (keep the Comparable intermediate) or skipping ordered trim for them.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
+ intermediateResult = new PriorityQueue<>(queue);
+ }
+ Object finalResult = aggFunction.extractFinalResult(intermediateResult);
Review Comment:
**Raw-sketch trim keys are near-arbitrary for Base64-serialized sketches.** For `DISTINCTCOUNTRAWTHETASKETCH`, CPC/ULL/KLL raw variants, the sort key is the Base64 string of the *partial* (per-worker) sketch. Base64 character order does not track byte values and merging rewrites headers/counts, so which groups survive the leaf trim is close to unrelated to the final merged-string order the downstream sort uses — silent, nondeterministic top-K, where pre-PR this path failed fast with ClassCastException. (Raw HLL is more benign: hex of max-merged registers is monotone-ish.) Note SSE `TableResizer` ranks these by `Serialized*.compareTo` (cardinality/quantile), so the two leaf trims disagree. `testOrderedRawHllGroupTrimMatchesDirectResults` uses a single input partial, where partial == merged by construction, so it cannot catch this. Ranking raw sketches by their estimate instead of their serialization string would make the trim meaningful.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
Review Comment:
**The `instanceof PriorityQueue` copy-guard hard-codes one function family into the generic executor.** The new invariant is really "`extractFinalResult` must not mutate its argument, because the same object is serialized downstream for merging", but nothing states or enforces it; this type-sniff covers only the funnel-window family. Known destructive finalizers it misses are currently unreachable in MSE (`FUNNELCOUNT`'s `BitmapMergeStrategy`/`SetMergeStrategy` intersect step bitmaps/sets in place but require dictionary-encoded input; `DistinctCountOffHeap.close()` frees memory but throws earlier), so this is latent — until the next destructive finalizer or an INTERMEDIATE-emitting planner arrives, at which point the corruption is silent wrong results. Note the identical unguarded pattern is already live (pre-existing) in SSE: `TableResizer.AggregationFunctionExtractor.extract` finalizes the live funnel queue when trimming `ORDER BY funnelMaxStep(...)`, draining the queues that
are then shipped to FINAL. A deeper fix — non-destructive finalization as an `AggregationFunction` contract (or copying inside the funnel functions) — would fix both engines and let this special case be deleted.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,14 +188,73 @@ public void processBlock(MseBlock.Data block) {
}
}
+ /// Get aggregation result limited to first `maxRows` rows, ordered with `collations`.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numGroups = Math.min(_groupIdGenerator.getNumGroups(), maxRows);
Review Comment:
**Missing early-out when nothing can be trimmed.** When `_groupIdGenerator.getNumGroups() <= maxRows` no row is dropped, and LEAF/INTERMEDIATE output order is discarded by the re-aggregating FINAL stage — yet this path still clones, finalizes (HLL cardinality, funnel queue copy + full window drain, theta union flush) and heap-sorts every group. With the default trim size of max(5×limit, 5000), servers routinely have fewer groups than the trim capacity, so this is the common case. `if (_groupIdGenerator.getNumGroups() <= maxRows) { return getResult(maxRows); }` at the top skips all of it (and the GROUPS_TRIMMED stat is unaffected: rowsProcessed never exceeds trim size in that case).
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,14 +188,73 @@ public void processBlock(MseBlock.Data block) {
}
}
+ /// Get aggregation result limited to first `maxRows` rows, ordered with `collations`.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numGroups = Math.min(_groupIdGenerator.getNumGroups(), maxRows);
+ if (numGroups == 0) {
+ return List.of();
+ }
+
+ int numKeys = _groupKeyIds.length;
+ int numFunctions = _aggFunctions.length;
+ int[] columnsToFinalize = _aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult
+ ? collations.stream().mapToInt(RelFieldCollation::getFieldIndex).filter(index -> index >= numKeys)
+ .filter(index -> {
+ AggregationFunction aggFunction = _aggFunctions[index - numKeys];
+ ColumnDataType intermediateStoredType = aggFunction.getIntermediateResultColumnType().getStoredType();
+ return intermediateStoredType == ColumnDataType.OBJECT
Review Comment:
**AVG/MINMAXRANGE are pushed onto the decorated path unnecessarily.** `AvgPair` and `MinMaxRangePair` implement `Comparable` with ordering that matches their final double value, and pre-PR they trimmed correctly on the zero-allocation raw heap. Because their intermediate stored type is OBJECT they now pay `row.clone()` + `SortedRow` + boxed finalization per group (several extra MB of garbage at 100K groups) for identical ordering. A per-function capability ("intermediate natural order equals final order") or a small allowlist would keep them on the fast path.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
Review Comment:
**Per-group allocation and eager finalization are heavier than needed.** Every iterated group allocates a full-width `row.clone()` plus a `SortedRow`, including the candidates the heap rejects immediately, even though the comparator only reads the collation columns; and every column in `columnsToFinalize` is finalized for every group even when an earlier scalar collation key (e.g. `ORDER BY COUNT(*) DESC, PERCENTILE(...)`) decides nearly all comparisons — `SortComparator` short-circuits, so tie-break-only sketch/percentile columns are finalized N times for nothing. A compact sort-key array sized to `collations.size()` (like SSE's `IntermediateRecord`), or lazy finalization on first tie, would cut both costs.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -186,14 +188,73 @@ public void processBlock(MseBlock.Data block) {
}
}
+ /// Get aggregation result limited to first `maxRows` rows, ordered with `collations`.
+ public List<Object[]> getResult(List<RelFieldCollation> collations, int maxRows) {
+ int numGroups = Math.min(_groupIdGenerator.getNumGroups(), maxRows);
+ if (numGroups == 0) {
+ return List.of();
+ }
+
+ int numKeys = _groupKeyIds.length;
+ int numFunctions = _aggFunctions.length;
+ int[] columnsToFinalize = _aggType.isOutputIntermediateFormat() && !_leafReturnFinalResult
+ ? collations.stream().mapToInt(RelFieldCollation::getFieldIndex).filter(index -> index >= numKeys)
+ .filter(index -> {
+ AggregationFunction aggFunction = _aggFunctions[index - numKeys];
+ ColumnDataType intermediateStoredType = aggFunction.getIntermediateResultColumnType().getStoredType();
+ return intermediateStoredType == ColumnDataType.OBJECT
+ || intermediateStoredType != aggFunction.getFinalResultColumnType().getStoredType();
+ }).distinct().toArray()
+ : new int[0];
+ SortUtils.SortComparator sortComparator = new SortUtils.SortComparator(collations, true);
+ // Scalar intermediate values already have their final stored representation; keep the original heap path.
+ if (columnsToFinalize.length == 0) {
+ return getResult(sortComparator, maxRows);
+ }
+ Comparator<SortedRow> comparator = (r1, r2) -> sortComparator.compare(r1.sortKeys(), r2.sortKeys());
Review Comment:
**Duplicated top-K heap loop.** Lines 214-248 are a line-for-line copy of the private `getResult(Comparator, int)` below (fill, peek/poll/offer replace — including the copied `TODO: allocate new array row only if row enters set` — `_rowsProcessed` accounting, drain + `Collections.reverse`), differing only in the element type and the `.row()` unwrap. A later fix to one copy (the TODO, trim-stat accounting, eviction logic) will silently miss the other, and which copy runs depends on the sort column's type. One generic helper serves both — e.g. `<T> List<Object[]> topK(Comparator<T> cmp, int maxRows, Function<Object[],T> decorate, Function<T,Object[]> unwrap)` with `Function.identity()` for the scalar path (keeping its no-allocation property) — and the fill/replace step already exists as `SelectionOperatorUtils.addToPriorityQueue`, which `TopNSortOperator` in this module uses.
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
Review Comment:
**Dead branch.** `getSortedRow` is only called from the two loops that run after the `columnsToFinalize.length == 0` early return at line 211-213, so this `return new SortedRow(row, row)` can never execute. It also contradicts the record's doc ("keeping final sort values separate from the mergeable output row"). Suggest removing it.
##########
pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/AggregateOperatorTest.java:
##########
@@ -351,6 +353,133 @@ public void testGroupTrimHintOverridesContextValue() {
assertEquals(getAggregateOperator(context, nodeHint, 5, List.of(new RelFieldCollation(1))).getGroupTrimSize(), 30);
}
+ @DataProvider
Review Comment:
**Test coverage is aimed at an unreachable stage while a reachable variant is untested.** Five of the ten `orderedGroupTrimCases` trim at `AggType.INTERMEDIATE`, which no planner emits (only plan serde handles it; `PinotAggregateExchangeNodeInsertRule` has an open TODO). Meanwhile no new test covers `leafReturnFinalResult=true`, which guards the reachable gate at `MultistageGroupByExecutor` line 200: if that `!_leafReturnFinalResult` condition is ever dropped, a LEAF with `is_leaf_return_final_result` would call `extractFinalResult` on an already-final value (ClassCastException for DISTINCTCOUNT/AVG) and all 13 new test invocations would still pass. Consider swapping some INTERMEDIATE rows for `leafReturnFinalResult=true` LEAF rows.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4224469629
##########
pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MultistageGroupByExecutor.java:
##########
@@ -230,6 +291,30 @@ public List<Object[]> getResult(Comparator<Object[]> comparator, int maxRows) {
return result;
}
+ private SortedRow getSortedRow(Object[] row, int[] columnsToFinalize) {
+ if (columnsToFinalize.length == 0) {
+ return new SortedRow(row, row);
+ }
+ // Extract each sort value once per group, retaining intermediate values for downstream aggregation.
+ Object[] sortKeys = row.clone();
+ for (int index : columnsToFinalize) {
+ AggregationFunction aggFunction = _aggFunctions[index - _groupKeyIds.length];
+ Object intermediateResult = row[index];
+ // Funnel finalization consumes the event queue, so extract from a copy.
+ if (intermediateResult instanceof PriorityQueue<?> queue) {
+ intermediateResult = new PriorityQueue<>(queue);
+ }
+ Object finalResult = aggFunction.extractFinalResult(intermediateResult);
+ sortKeys[index] = finalResult != null
Review Comment:
**Wrong trim ordering for SUMPRECISION (regression).** The sort key is converted with `aggFunction.getFinalResultColumnType().getStoredType()` — the SSE wire type — not the type the MSE FINAL stage emits. `SumPrecisionAggregationFunction.getFinalResultColumnType()` returns `STRING` (with a TODO), so `TypeUtils.convert` turns each `BigDecimal` sum into `toString()` and the trim heap orders groups lexicographically: `"9" > "2500" > "100" > "10"`, and scientific notation like `1E+3` misorders further. Before this PR the heap compared the `BigDecimal` intermediates numerically (they are `Comparable`), which was correct; MSE FINAL/DIRECT and SSE `TableResizer` also compare numerically. With group trim on a LEAF above a join and `ORDER BY SUMPRECISION(...) DESC LIMIT 1` over sums {5..9, 100}, the 100-group is trimmed and the query silently returns 9. Same root cause gives subtler divergence for any function whose declared final type disagrees with the planner type (e.g. `PERCENTILES
MARTTDIGEST` pinned to ARG0 truncates the double before comparing). The sort key should be built in the type the downstream sort actually compares (the planned output type), or such columns should keep their natural `Comparable` intermediate ordering.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 merged PR #19800:
URL: https://github.com/apache/pinot/pull/19800
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Fix MSE group trimming for intermediate aggregation results [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19800:
URL: https://github.com/apache/pinot/pull/19800#discussion_r4251403186
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/SetMergeStrategy.java:
##########
@@ -41,11 +42,13 @@ public List<Set> merge(List<Set> intermediateResult1, List<Set> intermediateResu
@Override
public LongArrayList extractFinalResult(List<Set> stepsSets) {
long[] result = new long[_numSteps];
- result[0] = stepsSets.get(0).size();
+ Set<?> firstStep = stepsSets.get(0);
+ Set<?> intersection = new HashSet<>(firstStep);
Review Comment:
Fixed in [4292f86188](https://github.com/apache/pinot/commit/4292f86188e9cd01f7b48b0d7541e62ec4b4b235). The first count comes directly from `size()`, single-step funnels do not copy the population, and an empty operand stops intersection work immediately. Nonempty intersections copy the smaller operand; later steps reuse the private intersection when it is already smaller. Matching INT/LONG/FLOAT/DOUBLE sets retain primitive containers and typed intersection operations, while all input populations remain intact for later merging.
Local validation: 64 focused funnel/MSE tests passed, plus all four `pinot-core` formatting/checkstyle/license checks, with no compiler warnings on added lines. The regression rejects iteration over a mocked million-entry first step in single-step, empty-second-step and skewed-step cases. Primitive/string cases cover repeated extraction and preserved populations, including distinct NaN payloads and signed zeros. Running these tests against the previous `c4513289bd` finalizer produces three expected failures (large population traversal and float/double NaN counts); the fixed code passes.
The push is verified at this head. Hosted CI for the new head is pending; human approval remains pending.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org