gortiz opened a new pull request, #19806: URL: https://github.com/apache/pinot/pull/19806
Fixes #19805 ## Summary This PR removes two full-size copies of each large intermediate result on its way between stages. It also writes and reads primitive arrays in bulk instead of one value at a time. The first user is exact `PERCENTILE`, whose `DoubleArrayList` state is often a G1 humongous object. The wire format does not change. - **Send side.** `SerializedIntermediateResult` gets `getSize()` and `writeTo(DataOutput)`. `DataBlockBuilder` and `BaseDataTableBuilder` write through them instead of `getBytes()`. `PercentileAggregationFunction` returns a subclass that writes the count and the doubles straight into the destination stream. The full-size `byte[]` is not created. - **Receive side.** `AggregationFunction.mergeSerializedIntermediateResult` merges a serialized value into the current result. The default deserializes and calls `merge()`. `PercentileAggregationFunction` overrides it and reads the doubles straight into the merged list. `MultistageAggregationExecutor` and `MultistageGroupByExecutor` use it for `OBJECT` columns of serialized blocks, through `AggregationFunctionUtils.mergeSerialized`. - **Bulk primitives.** `PinotOutputStream` gets `writeInts`, `writeLongs` and `writeDoubles`. The default encodes chunks of at most 8 KB. `PagedPinotOutputStream` puts the values straight into the current page through a typed buffer view. The new `PrimitiveArrayOutput` helper picks the bulk path for a `PinotOutputStream` and the chunked path for any other `DataOutput`. `INT/LONG/FLOAT/DOUBLE_ARRAY_LIST_SER_DE.deserialize` read with one typed-view `get`. ## Background A G1 humongous object is any object of half a region or more: 512 KB with 1 MB regions, or 64k doubles. Exact `PERCENTILE` with large groups produces many of them. Before this PR, every large group created two of them that existed only to move bytes: ``` send: DoubleArrayList ──serialize──> byte[] (humongous) ──copy──> paged output receive: data block ──deserialize──> DoubleArrayList (humongous) ──merge──> merged list after this PR: send: DoubleArrayList ──writeTo (bulk)──> paged output receive: data block ──mergeSerialized (bulk)──> merged list ``` ## Results Allocation for one list of 2^20 doubles (8 MB payload), as thread-allocated bytes: | Operation | Before | After | |---|---|---| | Build a data block | 16.9 MB | 8.5 MB (the paged output only) | | Merge into a list with room | 8.4 MB | about 0 | JMH on JDK 25, Apple M4 Pro, 3 forks x 5 iterations, 2^20 values (`BenchmarkPrimitiveArrayOutput`, `BenchmarkPrimitiveListDeserialize`): | Operation | One value at a time | Bulk | Speedup | |---|---|---|---| | Write doubles to `PagedPinotOutputStream` | 699 ops/s | 2,293 ops/s | 3.3x | | Write ints to `PagedPinotOutputStream` | 385 ops/s | 4,468 ops/s | 11.6x | | Write doubles to `DataOutputStream` (chunked) | 214 ops/s | 2,140 ops/s | 10x | | Deserialize doubles from a heap buffer | 1,935 ops/s | 6,416 ops/s | 3.3x | | Deserialize ints from a heap buffer | 1,924 ops/s | 12,821 ops/s | 6.7x | ## Notes for reviewers - **NaN.** `writeDoubles` writes raw bits, the same bytes as `ByteBuffer.putDouble` and `DOUBLE_ARRAY_LIST_SER_DE`. `DataOutput.writeDouble` canonicalizes NaN, so a per-value `writeDouble` loop would not match the serializer. - **Lazy serialization.** The `PERCENTILE` subclass reads the list when the builder writes it. The list must not change between `serializeIntermediateResult` and the write. The builders write the value right after they get it. `getBytes()` still works for other callers. - **Unchanged paths.** Row heap blocks and the `leafReturnFinalResult` path still use `merge()`. Other aggregation functions use the default implementations, which behave as before. - **Input checks.** The bulk deserializers reject a count larger than the remaining bytes before they allocate the array. They throw the same exception types as the old loop. ## Tests - `PrimitiveArrayOutputTest`: paged and chunked bulk writes produce the same bytes as per-value puts (with NaN payloads and page boundaries), the written size after a seek back, and invalid ranges. - `ObjectSerDeUtilsPrimitiveListTest`: bulk reads match the old per-value loop, round trips, and truncated or negative counts. - `PercentileIntermediateResultSerDeTest`: `writeTo` bytes match `getBytes()` in data blocks and data tables, `mergeSerialized` matches `merge()` and skips `null`, and allocation checks for the block build and the merge. - `SerializedIntermediateResultMergeTest`: MSE aggregation and group-by merges of serialized blocks match the results from row heap blocks. 🤖 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]
