gortiz commented on code in PR #19806:
URL: https://github.com/apache/pinot/pull/19806#discussion_r4273551127
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java:
##########
@@ -206,10 +211,77 @@ 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. 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() {
Review Comment:
Good catch, thanks. Fixed in 6ba32b82df as you suggested: `serializeChunk`
now keeps a byte-array snapshot, `new
SerializedIntermediateResult(serialized.getType(), serialized.getBytes())`, so
the derive tasks read immutable bytes and each base list is serialized once. I
also added to the `SerializedIntermediateResult` javadoc that a caller that
keeps the object must take a snapshot.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java:
##########
@@ -206,10 +211,77 @@ 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. 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();
+ output.writeInt(size);
+ PrimitiveArrayOutput.writeDoubles(output, _values.elements(), 0, size);
Review Comment:
Agreed, and thanks for the numbers. Fixed in 6ba32b82df: below
`MIN_BULK_VALUES`, `SerializedDoubleArrayList.writeTo` now writes `getBytes()`
in one call, like master. I extended
`testSerializedIntermediateResultMatchesByteArray` with 63 and 64 values so
both branches are compared with the byte-array serializer.
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/memory/PagedPinotOutputStream.java:
##########
@@ -212,6 +213,78 @@ public void write(byte[] b, int off, int len)
_written = Math.max(_written, _offsetInPage + _currentPageStartOffset);
}
+ /// Puts as many values as fit in the current page straight into it, through
a big-endian typed view. A value that
+ /// does not fit in the rest of the page (the page offset depends on
everything written before, so it is not aligned
+ /// to the value width) is written with [#writeInt(int)], which splits it
across the two pages. When fewer than
+ /// [PrimitiveArrayOutput#MIN_BULK_VALUES] values go to the page, they are
also written one by one, because the
+ /// typed-view copy has a fixed cost that a few values do not pay back.
+ @Override
+ public void writeInts(int[] values, int offset, int length)
+ throws IOException {
+ Objects.checkFromIndexSize(offset, length, values.length);
+ int end = offset + length;
+ int i = offset;
+ while (i < end) {
+ int numValues = Math.min(end - i, remainingInPage() / Integer.BYTES);
+ if (numValues < PrimitiveArrayOutput.MIN_BULK_VALUES) {
+ writeInt(values[i++]);
+ continue;
+ }
+ currentPageView().asIntBuffer().put(values, i, numValues);
+ i += numValues;
+ _offsetInPage += numValues * Integer.BYTES;
+ _written = Math.max(_written, _offsetInPage + _currentPageStartOffset);
+ }
+ }
+
+ /// Same as [#writeInts(int[], int, int)], for longs.
+ @Override
+ public void writeLongs(long[] values, int offset, int length)
+ throws IOException {
+ Objects.checkFromIndexSize(offset, length, values.length);
+ int end = offset + length;
+ int i = offset;
+ while (i < end) {
+ int numValues = Math.min(end - i, remainingInPage() / Long.BYTES);
+ if (numValues < PrimitiveArrayOutput.MIN_BULK_VALUES) {
+ writeLong(values[i++]);
+ continue;
+ }
+ currentPageView().asLongBuffer().put(values, i, numValues);
+ i += numValues;
+ _offsetInPage += numValues * Long.BYTES;
+ _written = Math.max(_written, _offsetInPage + _currentPageStartOffset);
+ }
+ }
+
+ /// Same as [#writeInts(int[], int, int)], for doubles.
+ @Override
+ public void writeDoubles(double[] values, int offset, int length)
+ throws IOException {
+ Objects.checkFromIndexSize(offset, length, values.length);
+ int end = offset + length;
+ int i = offset;
+ while (i < end) {
+ int numValues = Math.min(end - i, remainingInPage() / Double.BYTES);
+ if (numValues < PrimitiveArrayOutput.MIN_BULK_VALUES) {
+ // Raw bits, like the typed view (and ByteBuffer#putDouble):
writeDouble() would canonicalize NaN.
+ writeLong(Double.doubleToRawLongBits(values[i++]));
Review Comment:
Took it, in 6ba32b82df. `writeInts`/`writeLongs`/`writeDoubles` now put the
values that fit in the current page with `_currentPage.putX(position, ...)` and
update `_offsetInPage` and `_written` once. The typed-view copy is still used
from 64 values on, and a value that straddles two pages still goes through
`writeInt`/`writeLong`. `PrimitiveArrayOutputTest` compares it with per-value
puts for page sizes of 3, 7, 13, 100 and the default, at several start offsets
and lengths on both sides of 16 and 64.
--
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]
