xiangfu0 commented on code in PR #20186:
URL: https://github.com/apache/pinot/pull/20186#discussion_r4499831391
##########
pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/GroupingSetsQueriesTest.java:
##########
@@ -630,6 +634,7 @@ public Object[][] baseAggregationEquivalenceCases() {
{"sum-cube", "SUM(" + LNG + "), COUNT(*)", "CUBE(" + D1 + ", " + D2 +
")", false},
{"distinctcount-groupingsets", "DISTINCTCOUNT(" + LNG + "), COUNT(*)",
"GROUPING SETS ((" + D1 + "), (" + D2 + "), ())", true},
+ {"exprmax-rollup", "EXPR_MAX(" + LNG + ", " + DBL + "), COUNT(*)",
"ROLLUP(" + D1 + ", " + D2 + ")", false},
Review Comment:
This new case fails every run in [Integration Test Set
1](https://github.com/apache/pinot/actions/runs/42324823352/job/128257564603).
The broker rejects the query with `EXPR_MAX is not enabled on this broker`,
because `ExprMinMaxRewriter` isn't set up for this cluster. With no
`resultTable` in the response, `rowsByKey` throws an NPE at :645. Could you
enable the rewriter for this cluster, or move the case to one that has it?
Until then, the new EXPRMIN/EXPRMAX grouping-sets derive path has no end-to-end
coverage.
##########
pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/ObjectSerDeUtils.java:
##########
@@ -0,0 +1,1120 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.utils;
+
+import com.clearspring.analytics.stream.cardinality.HyperLogLog;
+import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
+import com.clearspring.analytics.stream.cardinality.RegisterSet;
+import com.dynatrace.hash4j.distinctcount.UltraLogLog;
+import it.unimi.dsi.fastutil.doubles.Double2LongMap;
+import it.unimi.dsi.fastutil.doubles.Double2LongOpenHashMap;
+import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.doubles.DoubleIterator;
+import it.unimi.dsi.fastutil.doubles.DoubleOpenHashSet;
+import it.unimi.dsi.fastutil.doubles.DoubleSet;
+import it.unimi.dsi.fastutil.floats.Float2LongMap;
+import it.unimi.dsi.fastutil.floats.Float2LongOpenHashMap;
+import it.unimi.dsi.fastutil.floats.FloatArrayList;
+import it.unimi.dsi.fastutil.floats.FloatIterator;
+import it.unimi.dsi.fastutil.floats.FloatOpenHashSet;
+import it.unimi.dsi.fastutil.floats.FloatSet;
+import it.unimi.dsi.fastutil.ints.Int2LongMap;
+import it.unimi.dsi.fastutil.ints.Int2LongOpenHashMap;
+import it.unimi.dsi.fastutil.ints.IntArrayList;
+import it.unimi.dsi.fastutil.ints.IntIterator;
+import it.unimi.dsi.fastutil.ints.IntOpenHashSet;
+import it.unimi.dsi.fastutil.ints.IntSet;
+import it.unimi.dsi.fastutil.longs.Long2LongMap;
+import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap;
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import it.unimi.dsi.fastutil.longs.LongIterator;
+import it.unimi.dsi.fastutil.longs.LongOpenHashSet;
+import it.unimi.dsi.fastutil.longs.LongSet;
+import it.unimi.dsi.fastutil.objects.Object2LongMap;
+import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap;
+import it.unimi.dsi.fastutil.objects.ObjectArrayList;
+import it.unimi.dsi.fastutil.objects.ObjectLinkedOpenHashSet;
+import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet;
+import java.io.IOException;
+import java.lang.foreign.MemorySegment;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.IntBuffer;
+import java.util.Set;
+import org.apache.datasketches.theta.ThetaSketch;
+import org.apache.datasketches.theta.UpdatableThetaSketch;
+import
org.apache.pinot.segment.local.customobject.tdigest.SerializedTDigestInput;
+import org.apache.pinot.segment.local.customobject.tdigest.TDigest;
+import org.apache.pinot.segment.local.customobject.tdigest.TDigestAccumulator;
+import org.apache.pinot.segment.spi.memory.PinotOutputStream;
+import org.apache.pinot.segment.spi.memory.PrimitiveArrays;
+import org.apache.pinot.spi.utils.BigDecimalUtils;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.Utf8Utils;
+
+
+/// The codecs of the intermediate results that aggregation functions carry in
`OBJECT` columns, and of the
+/// pre-aggregated values that value aggregators store in `BYTES` columns, one
per type defined in this module or below
+/// it. An aggregation function owns the serialization of its intermediate
result: it serializes and deserializes the
+/// result, with the codec of its type when there is one, and declares as its
own constant the wire type value that
+/// precedes the payload. A value identifies one intermediate result type for
good: a new type takes a value above
+/// every one declared so far, and a value never changes or gets reused once
its type is retired.
+public class ObjectSerDeUtils {
+ private ObjectSerDeUtils() {
+ }
+
+ /// Serializer/De-serializer for a specific type of object.
+ ///
+ /// @param <T> Type of the object
+ public interface ObjectSerDe<T> {
+
+ /// Serializes a value into a byte array.
+ byte[] serialize(T value);
+
+ /// Writes the serialized form of a value to an output stream, which takes
values straight into its pages. The
+ /// default writes the byte array; a codec that writes its value in parts
overrides this to skip the full-size
+ /// array.
+ default void serialize(T value, PinotOutputStream output)
+ throws IOException {
+ output.write(serialize(value));
+ }
+
+ /// De-serializes a value from the remaining bytes of a byte buffer, which
end where the payload ends. Where the
+ /// position is left afterwards depends on the codec.
+ T deserialize(ByteBuffer buffer);
+
+ /// De-serializes a value from a byte array. The default wraps the array,
which costs nothing. A codec whose native
+ /// input is an array overrides this one too, so that an array caller
skips the copy its buffer overload makes.
+ default T deserialize(byte[] bytes) {
+ return deserialize(ByteBuffer.wrap(bytes));
+ }
+ }
+
+ /// Serializer/De-serializer that knows the payload size of a value without
serializing it, so the value can be
+ /// written straight into a buffer reserved for it.
+ ///
+ /// @param <T> Type of the object
+ public interface SizedObjectSerDe<T> extends ObjectSerDe<T> {
+
+ /// Returns the exact payload size of a value. Must not serialize the
value or allocate the payload. The value must
+ /// stay unchanged between sizing and serialization.
+ int getSerializedSize(T value);
+
+ /// Writes the payload at the current position of the big-endian `buffer`,
which has at least
+ /// [#getSerializedSize(Object)] bytes remaining, and advances the
position by the payload size. Only the position
+ /// changes: the buffer is not retained and the value is not mutated.
Supports heap and direct buffers.
+ void serialize(T value, ByteBuffer buffer);
+
+ @Override
+ default byte[] serialize(T value) {
+ byte[] bytes = new byte[getSerializedSize(value)];
+ serialize(value, ByteBuffer.wrap(bytes));
+ return bytes;
+ }
+ }
+
+ public static final SizedObjectSerDe<IntArrayList> INT_ARRAY_LIST_SER_DE =
new SizedObjectSerDe<>() {
+
+ @Override
+ public int getSerializedSize(IntArrayList intArrayList) {
+ return Math.addExact(Integer.BYTES,
Math.multiplyExact(intArrayList.size(), Integer.BYTES));
+ }
+
+ @Override
+ public void serialize(IntArrayList intArrayList, ByteBuffer buffer) {
+ int size = intArrayList.size();
+ buffer.putInt(size);
+ int[] values = intArrayList.elements();
+ for (int i = 0; i < size; i++) {
+ buffer.putInt(values[i]);
+ }
+ }
+
+ @Override
+ public void serialize(IntArrayList intArrayList, PinotOutputStream output)
+ throws IOException {
+ int size = intArrayList.size();
+ output.writeInt(size);
+ output.writeInts(intArrayList.elements(), 0, size);
+ }
+
+ @Override
+ public IntArrayList deserialize(ByteBuffer buffer) {
+ int numValues = buffer.getInt();
+ int[] values = new int[numValues];
Review Comment:
The INT/LONG/FLOAT/DOUBLE list codecs (here and at 194, 228, 262) no longer
have #19806's `checkRemaining` guard, which compared the count to the remaining
bytes before allocating. `testTruncatedAndNegativeCounts` was deleted along
with it. With a corrupt or truncated payload, a large count now tries to
allocate up to ~16 GB and fails with an OOM instead of a
`BufferUnderflowException`. A negative count now throws
`NegativeArraySizeException`.
`PercentileAggregationFunction.mergeSerializedIntermediateResult` still does
this check. Could you restore the guard and its test?
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/memory/PinotOutputStream.java:
##########
@@ -18,112 +18,92 @@
*/
package org.apache.pinot.segment.spi.memory;
-import com.google.common.primitives.Longs;
import java.io.DataOutput;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.nio.charset.StandardCharsets;
+/// A big-endian [DataOutput] with a seekable position and bulk writes of
primitive arrays.
public abstract class PinotOutputStream extends OutputStream implements
DataOutput {
- /// Return the current position in the OutputStream.
- ///
- /// @return current position in bytes from the start of the stream
+ /// Returns the current position, in bytes from the start of the stream.
public abstract long getCurrentOffset();
- /// Seek to a new position in the OutputStream.
+ /// Seeks to a new position.
///
- /// @param newPos the new position to seek to
- /// @throws IllegalArgumentException If the new position is negative
- public abstract void seek(long newPos);
-
- /// Moves the current offset, applying the given change.
- /// @param change the change to apply to the current offset
/// @throws IllegalArgumentException if the new position is negative
- public void moveCurrentOffset(long change) {
- long newOffset = getCurrentOffset() + change;
- seek(newOffset);
- }
+ public abstract void seek(long newPos);
@Override
- public void writeBoolean(boolean v) throws IOException {
+ public void writeBoolean(boolean v)
+ throws IOException {
write(v ? 1 : 0);
}
@Override
- public void writeByte(int v) throws IOException {
+ public void writeByte(int v)
+ throws IOException {
write(v);
}
@Override
- public void writeChar(int v) throws IOException {
+ public abstract void writeShort(int v)
+ throws IOException;
+
+ @Override
+ public void writeChar(int v)
+ throws IOException {
writeShort(v);
}
@Override
- public void writeChars(String s) throws IOException {
- for (int i = 0; i < s.length(); i++) {
- writeChar(s.charAt(i));
- }
- }
+ public abstract void writeInt(int v)
Review Comment:
This is why [Binary Compatibility
Check](https://github.com/apache/pinot/actions/runs/42324823341/job/128257565104)
fails. In release-1.6.0, `moveCurrentOffset`, `writeInt`, `writeLong`,
`writeShort` and `write(DataBuffer, long, long)` are concrete public methods.
This PR removes `moveCurrentOffset` and makes the other four abstract. In-repo
code is fine because `PagedPinotOutputStream` overrides all of them, but
external subclasses and `DataBlockSerde` plugins would break. Keeping the old
implementations costs nothing (e.g. `moveCurrentOffset` →
`seek(getCurrentOffset() + change)`). If you'd rather not keep them, please
list the change under Backward incompatibility.
--
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]
