yashmayya commented on code in PR #19806:
URL: https://github.com/apache/pinot/pull/19806#discussion_r4275410383
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java:
##########
@@ -206,10 +211,84 @@ 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);
+ double[] elements = intermediateResult.elements();
+ if (numValues < PrimitiveArrayOutput.MIN_BULK_VALUES) {
+ for (int i = size; i < size + numValues; i++) {
+ elements[i] = buffer.getDouble();
+ }
+ } else {
+ buffer.asDoubleBuffer().get(elements, size, numValues);
+ buffer.position(buffer.position() + numValues * Double.BYTES);
+ }
+ return intermediateResult;
+ }
+
@Override
public SerializedIntermediateResult
serializeIntermediateResult(DoubleArrayList doubleArrayList) {
- return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.DoubleArrayList.getValue(),
- ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE.serialize(doubleArrayList));
+ return new SerializedDoubleArrayList(doubleArrayList);
+ }
+
+ /// Serializes a [DoubleArrayList] on demand, with the bytes of
ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE: a
+ /// big-endian int count, then the big-endian doubles. [#writeTo] writes the
values in bulk straight into the
+ /// destination (see [PrimitiveArrayOutput]), so the builders never hold the
full-size byte array; lists shorter
+ /// than [PrimitiveArrayOutput#MIN_BULK_VALUES] are written as one small
array. Not thread-safe; the list must not
+ /// change until it is written.
+ private static final class SerializedDoubleArrayList extends
SerializedIntermediateResult {
+ private final DoubleArrayList _values;
+
+ SerializedDoubleArrayList(DoubleArrayList values) {
+ super(ObjectSerDeUtils.ObjectType.DoubleArrayList.getValue());
+ _values = values;
+ }
+
+ @Override
+ public byte[] getBytes() {
+ return ObjectSerDeUtils.DOUBLE_ARRAY_LIST_SER_DE.serialize(_values);
+ }
+
+ @Override
+ public int getSize() {
+ return Math.toIntExact(Integer.BYTES + (long) _values.size() *
Double.BYTES);
+ }
+
+ @Override
+ public void writeTo(DataOutput output)
+ throws IOException {
+ int size = _values.size();
+ if (size < PrimitiveArrayOutput.MIN_BULK_VALUES) {
Review Comment:
Optional: below 64 values, this writes `getBytes()` for every output. So
PERCENTILE never uses the new tight loop in `PagedPinotOutputStream`, and short
lists on the MSE path only match master. On the SSE path, lists of 64 values or
more use the chunked path. That path saves no memory while the list fits in one
8 KB chunk, and on M3 it is slower than master there.
Suggestion:
```java
int size = _values.size();
if (!(output instanceof PinotOutputStream)
&& Integer.BYTES + (long) size * Double.BYTES <=
PrimitiveArrayOutput.CHUNK_SIZE) {
output.write(getBytes());
return;
}
output.writeInt(size);
PrimitiveArrayOutput.writeDoubles(output, _values.elements(), 0, size);
```
ns per list (JDK 25, M3 Max), master / now / suggestion:
| Path | 16 values | 32 values | 64 values | 128 values | 256 values |
|---|---|---|---|---|---|
| MSE paged | 20 / 20 / 15 | 31 / 33 / 18 | – | – | – |
| SSE data table | – | – | 51 / 107 / 59 | 98 / 150 / 108 | 191 / 250 / 205 |
The suggestion writes the same bytes on both outputs. I verified 24 cases
from 0 to 5,000 values, with NaN payloads.
--
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]
