Jackie-Jiang commented on code in PR #20186:
URL: https://github.com/apache/pinot/pull/20186#discussion_r4503961572
##########
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:
The shared custom cluster doesn't enable the EXPR_MIN/EXPR_MAX rewriters,
and they are JVM-wide, so I moved the case instead of enabling them there.
`ExprMinMaxTest.testGroupingSetsBaseAggregationMatchesExpansion` now runs
`ROLLUP(groupByIntColumn, groupByIntColumn2)` with `expr_max` and `expr_min`,
with `groupingSetsBaseAggregation` on and off, and expects the same rows. It
goes through the single-stage servers and broker, so it merges deserialized
`ExprMinMaxObject`s across base groups. I checked that the base-aggregation run
does take the derive path. `GroupingSetsQueriesTest` is back to its master
version.
##########
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:
Restored `checkRemaining` in the INT/LONG/FLOAT/DOUBLE list codecs, along
with `testTruncatedAndNegativeCounts`. It now runs once per payload before the
array is allocated, and checks the bound with a `long` multiply. A truncated
payload throws `BufferUnderflowException` again and a negative count
`IllegalArgumentException`, as with #19806.
##########
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:
The break is intentional. `PagedPinotOutputStream` is the only
implementation, and making the primitive and bulk writes abstract means an
implementation can't silently fall back to a slow per-byte default. I listed
the change under Backward incompatibility in the description:
`moveCurrentOffset` is gone, use `seek(getCurrentOffset() + change)`, and
several writes are now abstract, including the new `writeFloats`.
--
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]
