gortiz commented on code in PR #19806:
URL: https://github.com/apache/pinot/pull/19806#discussion_r4283034129
##########
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:
Thanks, taken as you suggested in ac3922df63. `writeTo` now writes
`getBytes()` only when the output is not a `PinotOutputStream` and the list
fits in one `CHUNK_SIZE` chunk, so the MSE path always gets the paged puts and
the SSE path stays on one small write up to 1023 values.
`PercentileIntermediateResultSerDeTest` now also covers 1023, 1024 and 5000
values, and writes every case to a `PagedPinotOutputStream` with small pages
too.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java:
##########
@@ -220,13 +250,31 @@ public SerializedIntermediateResult(int type, byte[]
buffer) {
_bytes = buffer;
}
+ /// Constructor for subclasses that serialize on demand. They must
override [#getBytes()], [#getSize()] and
+ /// [#writeTo(DataOutput)].
+ protected SerializedIntermediateResult(int type) {
+ _type = type;
+ _bytes = null;
+ }
+
public int getType() {
return _type;
}
public byte[] getBytes() {
return _bytes;
}
+
+ /// Returns the number of bytes [#writeTo(DataOutput)] writes, which is
the length of [#getBytes()].
+ public int getSize() {
Review Comment:
Good point, done in ac3922df63. `getSize()` now throws an
`IllegalStateException` that names the class and says it must override
`getSize()`. I went with an explicit check instead of an `assert` so it also
fires with assertions off.
##########
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));
Review Comment:
Agreed, the cap could end up below the size set right after. In ac3922df63
the cap applies only to the grow-by-half target, and the new size (now computed
with `Math.addExact`) is the lower bound, so `ensureCapacity` always gets at
least what `size()` needs.
--
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]
