yashmayya commented on code in PR #19806:
URL: https://github.com/apache/pinot/pull/19806#discussion_r4251365991
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java:
##########
@@ -206,10 +211,70 @@ public ColumnDataType getIntermediateResultColumnType() {
return ColumnDataType.OBJECT;
}
+ @Override
+ public DoubleArrayList mergeSerializedIntermediateResult(@Nullable
DoubleArrayList intermediateResult,
+ CustomObject serialized) {
+ if (intermediateResult == null) {
+ return deserializeIntermediateResult(serialized);
+ }
+ // Reads the values in bulk straight into the merged list, in the format of
+ // ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE: an int count, then the
doubles, in the buffer's byte order. Same
+ // growth as addAll(), without the incoming list.
+ ByteBuffer buffer = serialized.getBuffer();
Review Comment:
Over gRPC, the largest values still get a full-size copy before this merge
runs.
1. The sender splits a block into 8 MB messages
(`pinot.query.runner.max.msg.size.bytes` / 2).
2. The receiver builds a `CompoundDataBuffer` from these messages.
3. `getCustomObject` → `copyOrView` copies any value that crosses a message
boundary into a new `byte[]`.
A value of 8 MB or more always crosses. `BlockSplitter` does not split these
blocks, because it assumes 8 bytes per column.
Probe: `GrpcSendingMailbox.toByteStrings` →
`DataBlockUtils.deserialize(List)` → `mergeSerialized`:
| Block | Values copied | Allocated |
|---|---|---|
| 1 × 8 MB list | 1 of 1 | 8 MB (single buffer: 0) |
| 3 × 8 MB | 3 of 3 | 24 MB |
| 64 × 512 KB | 4 of 64 | 2 MB |
So for one 8 MB list, the receiver goes from two humongous copies to one,
not zero. `testMergeSerializedDoesNotAllocateIncomingList` uses a single
`ByteBuffer`, so it misses this. Can you update the description? A follow-up
can read across buffers through `DataBuffer`.
##########
pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java:
##########
@@ -510,11 +522,11 @@ public DoubleArrayList deserialize(byte[] bytes) {
@Override
public DoubleArrayList deserialize(ByteBuffer byteBuffer) {
int numValues = byteBuffer.getInt();
- DoubleArrayList doubleArrayList = new DoubleArrayList(numValues);
- for (int i = 0; i < numValues; i++) {
- doubleArrayList.add(byteBuffer.getDouble());
- }
- return doubleArrayList;
+ checkRemaining(byteBuffer, numValues, Double.BYTES);
+ double[] values = new double[numValues];
+ byteBuffer.asDoubleBuffer().get(values);
Review Comment:
Lists of up to ~16 values get 2–6x slower with the bulk path. On a
little-endian CPU, a big-endian typed view copies through
`Unsafe.copySwapMemory0`. This is a JNI call, not an intrinsic. It costs about
35 ns per call, and the JDK uses it for any copy over 6 bytes.
GROUP BY with many groups makes many small lists. These list SerDes also
serve HISTOGRAM, ARRAY_AGG, SUM_ARRAY and PERCENTILESMARTTDIGEST.
JMH, ns per list (JDK 25, M3 Max), before → PR:
| Values per list | 1 | 4 | 16 | 64 | 1024 |
|---|---|---|---|---|---|
| Deserialize (your benchmark, DOUBLE) | 6.5 → 43 | 8.9 → 42 | 16 → 41 | 45
→ 53 | 557 → 434 |
| Paged write | 8 → 41 | 16 → 43 | 20 → 46 | 59 → 62 | 1,092 → 725 |
| PERCENTILE merge | 29 → 56 | 22 → 56 | 67 → 65 | 241 → 122 | 3,296 → 1,295
|
Suggestion: use a scalar loop below about 32 values. This applies here, in
`PagedPinotOutputStream.write*`, in `PrimitiveArrayOutput.write*Chunked`, and
in the PERCENTILE merge. With a 32-value cutoff, I measured 6/15/28 ns (write)
and 12/21/50 ns (merge) at 1/4/16 values. Large lists keep their gains.
Can you also add small sizes to the two benchmarks? They only test 64K and
1M.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java:
##########
@@ -206,10 +211,70 @@ public ColumnDataType getIntermediateResultColumnType() {
return ColumnDataType.OBJECT;
}
+ @Override
+ public DoubleArrayList mergeSerializedIntermediateResult(@Nullable
DoubleArrayList intermediateResult,
+ CustomObject serialized) {
+ if (intermediateResult == null) {
+ return deserializeIntermediateResult(serialized);
+ }
+ // Reads the values in bulk straight into the merged list, in the format of
+ // ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE: an int count, then the
doubles, in the buffer's byte order. Same
+ // growth as addAll(), without the incoming list.
+ ByteBuffer buffer = serialized.getBuffer();
+ int numValues = buffer.getInt();
+ if (numValues < 0) {
+ throw new IllegalArgumentException("Negative number of values: " +
numValues);
+ }
+ if (numValues > buffer.remaining() / Double.BYTES) {
+ throw new BufferUnderflowException();
+ }
+ int size = intermediateResult.size();
+ int capacity = intermediateResult.elements().length;
+ if (size + numValues > capacity) {
+ intermediateResult.ensureCapacity(
+ (int) Math.min(Math.max((long) size + numValues, (long) capacity +
(capacity >> 1)), Integer.MAX_VALUE - 8));
+ }
+ // size() zero-fills the new range, so it must run before the values are
copied in.
+ intermediateResult.size(size + numValues);
Review Comment:
Optional: `size()` fills the new range with zeros, then the bulk read
overwrites it. For 8 MB, this merge takes 262 µs. Without the fill, it takes
144 µs (777 µs before the PR).
To skip the fill, read into `elements()` past `size`, then return
`DoubleArrayList.wrap(elements, size + numValues)`. Both callers use the return
value. The trade-off: the old list object then shares its array with the new
one.
--
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]
