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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
