gortiz opened a new pull request, #19066:
URL: https://github.com/apache/pinot/pull/19066
## Problem
`StreamingGroupByCombineOperator` can crash with `IndexOutOfBoundsException`
(or
silently return wrong aggregates) on high-cardinality group-by queries.
The operator's worker threads produce a per-segment `GroupByResultsBlock`
and hand
it to the consumer thread through the blocking queue. When a block carries a
raw
`AggregationGroupByResult`, that result is backed by the producing worker
thread's
`DictionaryBasedGroupKeyGenerator` group-key map — which is **thread-local
and
reused**. After enqueuing the block, the same worker advances to its next
segment
and `clearAndTrim()`/`expand()`s that thread-local map, while the consumer
thread is
still iterating the earlier block in `mergeBlock` →
`AggregationGroupByResult.getResultForGroupId(...)`. The torn read produces a
corrupted group id that indexes the segment's result holder out of bounds,
e.g.:
```
Caught exception while streaming group-by results:
Index 524288 out of bounds for length 524288
```
### When it triggers
- The segment's group cardinality exceeds `maxInitialResultHolderCapacity`
(default `10000`). Below that, `DictionaryBasedGroupKeyGenerator` uses a
per-instance `ArrayBasedHolder` instead of the reused thread-local
`IntGroupIdMap`,
so there is no aliasing.
- `streamingGroupByFlushThreshold` is set, so group-by is routed through
`StreamingGroupByCombineOperator`.
The non-streaming `GroupByCombineOperator` is **not** affected: it merges
each
segment inline on the producing worker thread, before that thread reuses its
thread-local map.
## Fix
Materialize each raw per-segment `AggregationGroupByResult` into
self-contained
`IntermediateRecord`s **on the producing worker thread**, before the block
is handed
to the consumer — mirroring how the non-streaming operator merges inline.
- New `BaseStreamingCombineOperator.detachFromWorkerThreadState(T)` hook
(identity by
default), invoked in `processSegments` before each block is enqueued.
- `StreamingGroupByCombineOperator` overrides it to iterate the group-key
iterator,
extract results, and rebuild a block backed only by `IntermediateRecord`s.
The
consumer-side `mergeBlock` then only ever consumes self-contained records.
- Adds `IntermediateRecord.withoutOrderByValues(Key, Record)` (the
constructor is
package-private and the streaming operator lives in a different package).
- The selection-only streaming combine operator keeps the no-op default.
## Testing
Adds
`StreamingGroupByCombineOperatorTest#testHighCardinalityConcurrentMergeIsCorrect`
(16 segments × 20000 groups, rows shuffled per segment so each segment's
group-id↔key
mapping differs, `maxExecutionThreads=1`). It reliably reproduces the crash
on the
current code and passes with the fix. Existing tests in the class continue
to pass.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
