xiangfu0 commented on code in PR #19806:
URL: https://github.com/apache/pinot/pull/19806#discussion_r4283252258
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java:
##########
@@ -206,10 +212,88 @@ 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 newSize = Math.addExact(size, numValues);
+ int capacity = intermediateResult.elements().length;
+ if (newSize > capacity) {
+ // Grow by half like addAll(), capped below the maximum array size, but
never below the size set next.
+ int grownCapacity = (int) Math.min((long) capacity + (capacity >> 1),
Integer.MAX_VALUE - 8);
+ intermediateResult.ensureCapacity(Math.max(newSize, grownCapacity));
+ }
+ // size() zero-fills the new range, so it must run before the values are
copied in.
+ intermediateResult.size(newSize);
+ double[] elements = intermediateResult.elements();
+ if (numValues < PrimitiveArrayOutput.MIN_BULK_VALUES) {
+ for (int i = size; i < newSize; 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. On any other
+ /// [DataOutput] than a [PinotOutputStream], a list that fits in one
[PrimitiveArrayOutput#CHUNK_SIZE] chunk is
+ /// 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 (!(output instanceof PinotOutputStream)
+ && Integer.BYTES + (long) size * Double.BYTES <=
PrimitiveArrayOutput.CHUNK_SIZE) {
+ // The chunked path saves no memory while the list fits in one chunk,
and one write of a small array is
+ // cheaper than chunk encoding or a write per value (a synchronized
call on a DataOutputStream). A
+ // PinotOutputStream writes the values straight into its pages, at any
size.
+ output.write(getBytes());
Review Comment:
Verified this branching stays byte-identical to `getBytes()`: the fast path
writes the serialized array directly, and the general path's `writeInt(size)` +
`writeDoubles` (raw bits, big-endian) matches it for both PinotOutputStream and
chunked DataOutput. Widening the fast-path cutoff from 64 values to one 8KB
chunk is a sensible perf/allocation trade on the data-table path. Thanks for
addressing the earlier notes.
--
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]
