yashmayya commented on code in PR #19806:
URL: https://github.com/apache/pinot/pull/19806#discussion_r4263750102
##########
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:
Optional: below 64 values, this calls `writeLong` once per value, and each
call checks the page and updates `_written`. For 16–48 values, the MSE paged
write is 1.6–2x slower than master:
| Values | 8 | 16 | 32 | 48 | 63 |
|---|---|---|---|---|---|
| master | 16 | 20 | 31 | 44 | 56 |
| this PR | 19 | 31 | 59 | 88 | 72 |
| tight loop | 11 | 14 | 18 | 24 | 29 |
The `getBytes()` change from the previous comment brings this back to
master. A tight loop is faster still: put the values that fit in the page with
`_currentPage.putLong(pos, ...)`, then update `_offsetInPage` and `_written`
once. I verified that this loop writes the same bytes in 216 cases (page sizes
from 3 bytes to 16 KB, NaN payloads included).
##########
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:
SSE data-table writes (`BaseDataTableBuilder`) are now 2–4.5x slower than
master for lists of up to 64 values. Below 16 values, `writeDoublesChunked`
calls `DataOutputStream.writeLong` once per value, and each call is a
synchronized `ByteArrayOutputStream.write`. From 16 to 64 values, it pays for
the scratch buffer and the JNI swap. GROUP BY with many groups uses this path
on every server.
ns per list (JDK 25, M3 Max), many lists into one `DataOutputStream`:
| Values | 1 | 4 | 8 | 15 | 32 | 64 |
|---|---|---|---|---|---|---|
| master | 8.4 | 11 | 13 | 17 | 28 | 51 |
| this PR | 16 | 29 | 47 | 76 | 87 | 101 |
| suggestion | 8.8 | 13 | 16 | 20 | 34 | – |
Suggestion: below `MIN_BULK_VALUES`, write `getBytes()` here with one call,
like master does. This also fixes short lists on the paged path (next comment).
##########
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:
After the rebase, this lazy `getBytes()` conflicts with #19264.
`GroupByUtils.serializeChunk`
([L361](https://github.com/apache/pinot/blob/f71329f825363f5727ad7ae353450824d70b8143/pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java#L361))
keeps the `SerializedIntermediateResult` from a pre-pass. Then
`cloneIntermediate`
([L503](https://github.com/apache/pinot/blob/f71329f825363f5727ad7ae353450824d70b8143/pinot-core/src/main/java/org/apache/pinot/core/util/GroupByUtils.java#L503))
calls `getBytes()` later, from the derive tasks. That code expects immutable
bytes from the pre-pass. With this class, each derive set serializes the live
base list again.
Probe: 64 base groups of 512 KB lists, 4 grouping sets. The derive
serializes 192 times instead of 64 and allocates 64 MB more (594 vs 530 MB).
The results are equal. Only `groupingSetsBaseAggregation=true` (opt-in) is
affected. It is correct today only because no derive task writes a base list.
Suggestion: take the snapshot in `serializeChunk`:
```java
SerializedIntermediateResult serialized =
aggregationFunctions[i].serializeIntermediateResult(intermediate);
row[i] = new SerializedIntermediateResult(serialized.getType(),
serialized.getBytes());
```
Also add to the `SerializedIntermediateResult` Javadoc that a caller that
keeps the object must take a snapshot.
--
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]
