This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 967bbe2aa34 Fix sketch accumulator initialization, tuple sizing, and
KLL null handling (#20228)
967bbe2aa34 is described below
commit 967bbe2aa349f45533eff2345be89ffd2ec3efeb
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Thu Nov 12 16:54:18 2026 -0800
Fix sketch accumulator initialization, tuple sizing, and KLL null handling
(#20228)
---
...ValueIntegerTupleSketchAggregationFunction.java | 3 -
.../DistinctCountCPCSketchAggregationFunction.java | 10 +-
...CountIntegerTupleSketchAggregationFunction.java | 3 -
...stinctCountRawCPCSketchAggregationFunction.java | 2 -
...inctCountRawThetaSketchAggregationFunction.java | 2 -
...istinctCountThetaSketchAggregationFunction.java | 9 +-
.../IntegerTupleSketchAggregationFunction.java | 13 +--
.../function/PercentileKLLAggregationFunction.java | 23 ++--
...aluesIntegerTupleSketchAggregationFunction.java | 3 -
.../IntermediateResultByteBufferSerDeTest.java | 15 ++-
.../PercentileKLLBytesAggregationTest.java | 113 ++++++++++++++++++++
.../SketchAccumulatorSerializationTest.java | 117 +++++++++++++++++++++
.../IntegerTupleSketchValueAggregator.java | 20 ++--
.../local/customobject/CpcSketchAccumulator.java | 33 ++----
.../customobject/CustomObjectAccumulator.java | 40 +++----
.../local/customobject/ThetaSketchAccumulator.java | 48 +++------
.../customobject/TupleIntSketchAccumulator.java | 55 +++-------
.../IntegerTupleSketchValueAggregatorTest.java | 37 ++++++-
.../customobject/CpcSketchAccumulatorTest.java | 67 +++++++++---
.../customobject/ThetaSketchAccumulatorTest.java | 72 ++++++++++---
.../TupleIntSketchAccumulatorTest.java | 102 +++++++++++++++---
21 files changed, 547 insertions(+), 240 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AvgValueIntegerTupleSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AvgValueIntegerTupleSketchAggregationFunction.java
index 1d34b244d27..5532a709031 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AvgValueIntegerTupleSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AvgValueIntegerTupleSketchAggregationFunction.java
@@ -57,9 +57,6 @@ public class AvgValueIntegerTupleSketchAggregationFunction
if (accumulator == null) {
return null;
}
- accumulator.setNominalEntries(_nominalEntries);
- accumulator.setSetOperations(_setOps);
- accumulator.setThreshold(_accumulatorThreshold);
TupleSketch<IntegerSummary> result = accumulator.getResult();
if (result.isEmpty() || result.getRetainedEntries() == 0) {
// there is nothing to average, return null
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java
index 4a16b767ad6..d9856b6a2e0 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunction.java
@@ -598,8 +598,6 @@ public class DistinctCountCPCSketchAggregationFunction
if (intermediateResult2.isEmpty()) {
return intermediateResult1;
}
- intermediateResult1.setLgNominalEntries(_lgNominalEntries);
- intermediateResult1.setThreshold(_accumulatorThreshold);
intermediateResult1.merge(intermediateResult2);
return intermediateResult1;
}
@@ -616,10 +614,8 @@ public class DistinctCountCPCSketchAggregationFunction
@Override
public CpcSketchAccumulator deserializeIntermediateResult(CustomObject
customObject) {
- // The accumulator serializes as its sketch, so a new accumulator takes
the sketch read back
- CpcSketchAccumulator cpcSketchAccumulator = new CpcSketchAccumulator();
-
cpcSketchAccumulator.apply(CpcSketch.heapify(MemorySegment.ofBuffer(customObject.getBuffer())));
- return cpcSketchAccumulator;
+ return new CpcSketchAccumulator(_lgNominalEntries, _accumulatorThreshold,
+ CpcSketch.heapify(MemorySegment.ofBuffer(customObject.getBuffer())));
}
@Override
@@ -632,8 +628,6 @@ public class DistinctCountCPCSketchAggregationFunction
if (intermediateResult == null) {
return 0L;
}
- intermediateResult.setLgNominalEntries(_lgNominalEntries);
- intermediateResult.setThreshold(_accumulatorThreshold);
return Math.round(intermediateResult.getResult().getEstimate());
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountIntegerTupleSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountIntegerTupleSketchAggregationFunction.java
index 4e071706eb4..0dac654caab 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountIntegerTupleSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountIntegerTupleSketchAggregationFunction.java
@@ -51,9 +51,6 @@ public class
DistinctCountIntegerTupleSketchAggregationFunction extends IntegerT
if (accumulator == null) {
return 0L;
}
- accumulator.setNominalEntries(_nominalEntries);
- accumulator.setSetOperations(_setOps);
- accumulator.setThreshold(_accumulatorThreshold);
return Double.valueOf(accumulator.getResult().getEstimate()).longValue();
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawCPCSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawCPCSketchAggregationFunction.java
index 819bbc49987..0e305b03bfa 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawCPCSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawCPCSketchAggregationFunction.java
@@ -55,8 +55,6 @@ public class DistinctCountRawCPCSketchAggregationFunction
extends DistinctCountC
if (intermediateResult == null) {
return null;
}
- intermediateResult.setLgNominalEntries(_lgNominalEntries);
- intermediateResult.setThreshold(_accumulatorThreshold);
return new SerializedCPCSketch(intermediateResult.getResult());
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawThetaSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawThetaSketchAggregationFunction.java
index f12861f0317..85544b87821 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawThetaSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawThetaSketchAggregationFunction.java
@@ -61,8 +61,6 @@ public class DistinctCountRawThetaSketchAggregationFunction
extends DistinctCoun
for (Object object : accumulators) {
ThetaSketchAccumulator accumulator = convertSketchAccumulator(object);
- accumulator.setThreshold(_accumulatorThreshold);
- accumulator.setSetOperationBuilder(_setOperationBuilder);
mergedSketches.add(accumulator.getResult());
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java
index ac30ec148d4..e1ea546eec6 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountThetaSketchAggregationFunction.java
@@ -1321,8 +1321,6 @@ public class DistinctCountThetaSketchAggregationFunction
mergedAccumulators.add(thetaSketchAccumulator1);
continue;
}
- thetaSketchAccumulator1.setSetOperationBuilder(_setOperationBuilder);
- thetaSketchAccumulator1.setThreshold(_accumulatorThreshold);
thetaSketchAccumulator1.merge(thetaSketchAccumulator2);
mergedAccumulators.add(thetaSketchAccumulator1);
}
@@ -1381,9 +1379,8 @@ public class DistinctCountThetaSketchAggregationFunction
int numBytes = buffer.getInt();
ByteBuffer slice = buffer.slice();
slice.limit(numBytes);
- ThetaSketchAccumulator thetaSketchAccumulator = new
ThetaSketchAccumulator();
-
thetaSketchAccumulator.apply(ObjectSerDeUtils.DATA_SKETCH_THETA_SER_DE.deserialize(slice));
- thetaSketchAccumulators.add(thetaSketchAccumulator);
+ thetaSketchAccumulators.add(new
ThetaSketchAccumulator(_setOperationBuilder, _accumulatorThreshold,
+ ObjectSerDeUtils.DATA_SKETCH_THETA_SER_DE.deserialize(slice)));
buffer.position(buffer.position() + numBytes);
}
return thetaSketchAccumulators;
@@ -1404,8 +1401,6 @@ public class DistinctCountThetaSketchAggregationFunction
for (Object accumulatorObject : accumulators) {
ThetaSketchAccumulator accumulator =
convertSketchAccumulator(accumulatorObject);
- accumulator.setThreshold(_accumulatorThreshold);
- accumulator.setSetOperationBuilder(_setOperationBuilder);
mergedSketches.add(accumulator.getResult());
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/IntegerTupleSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/IntegerTupleSketchAggregationFunction.java
index cf5396b8586..e9f80a1c6af 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/IntegerTupleSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/IntegerTupleSketchAggregationFunction.java
@@ -236,9 +236,6 @@ public class IntegerTupleSketchAggregationFunction
if (intermediateResult2.isEmpty()) {
return intermediateResult1;
}
- intermediateResult1.setThreshold(_accumulatorThreshold);
- intermediateResult1.setNominalEntries(_nominalEntries);
- intermediateResult1.setSetOperations(_setOps);
intermediateResult1.merge(intermediateResult2);
return intermediateResult1;
}
@@ -255,11 +252,8 @@ public class IntegerTupleSketchAggregationFunction
@Override
public TupleIntSketchAccumulator deserializeIntermediateResult(CustomObject
customObject) {
- // The accumulator serializes as its sketch, so a new accumulator takes
the sketch read back
- TupleIntSketchAccumulator tupleIntSketchAccumulator = new
TupleIntSketchAccumulator();
-
tupleIntSketchAccumulator.apply(TupleSketch.heapifySketch(MemorySegment.ofBuffer(customObject.getBuffer()),
- new IntegerSummaryDeserializer()));
- return tupleIntSketchAccumulator;
+ return new TupleIntSketchAccumulator(_setOps, _nominalEntries,
_accumulatorThreshold,
+
TupleSketch.heapifySketch(MemorySegment.ofBuffer(customObject.getBuffer()), new
IntegerSummaryDeserializer()));
}
@Override
@@ -278,9 +272,6 @@ public class IntegerTupleSketchAggregationFunction
}
accumulator = emptyAccumulator();
}
- accumulator.setNominalEntries(_nominalEntries);
- accumulator.setSetOperations(_setOps);
- accumulator.setThreshold(_accumulatorThreshold);
return
Base64.getEncoder().encodeToString(accumulator.getResult().toByteArray());
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLAggregationFunction.java
index d6ce82f0c4d..da4952c5225 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLAggregationFunction.java
@@ -104,10 +104,10 @@ public class PercentileKLLAggregationFunction
DataType dataType = valueSet.getValueType();
boolean singleValue = valueSet.isSingleValue();
if (dataType == DataType.BYTES && singleValue) {
- KllDoublesSketch[] deserializedSketches =
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
+ byte[][] serializedSketches = valueSet.getBytesValuesSV();
forEachNotNull(length, valueSet, (from, to) -> {
for (int i = from; i < to; i++) {
- sketch.merge(deserializedSketches[i]);
+ sketch.merge(deserializeSketch(serializedSketches[i]));
}
});
return;
@@ -148,11 +148,11 @@ public class PercentileKLLAggregationFunction
DataType dataType = valueSet.getValueType();
boolean singleValue = valueSet.isSingleValue();
if (dataType == DataType.BYTES && singleValue) {
- KllDoublesSketch[] deserializedSketches =
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
+ byte[][] serializedSketches = valueSet.getBytesValuesSV();
forEachNotNull(length, valueSet, (from, to) -> {
for (int i = from; i < to; i++) {
KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder,
groupKeyArray[i]);
- sketch.merge(deserializedSketches[i]);
+ sketch.merge(deserializeSketch(serializedSketches[i]));
}
});
return;
@@ -197,12 +197,13 @@ public class PercentileKLLAggregationFunction
DataType dataType = valueSet.getValueType();
boolean singleValue = valueSet.isSingleValue();
if (dataType == DataType.BYTES && singleValue) {
- KllDoublesSketch[] deserializedSketches =
deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
+ byte[][] serializedSketches = valueSet.getBytesValuesSV();
forEachNotNull(length, valueSet, (from, to) -> {
for (int i = from; i < to; i++) {
+ KllDoublesSketch rowSketch =
deserializeSketch(serializedSketches[i]);
for (int groupKey : groupKeysArray[i]) {
KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder,
groupKey);
- sketch.merge(deserializedSketches[i]);
+ sketch.merge(rowSketch);
}
}
});
@@ -265,13 +266,9 @@ public class PercentileKLLAggregationFunction
return sketch;
}
- /// Deserializes the sketches from the bytes.
- protected KllDoublesSketch[] deserializeSketches(byte[][]
serializedSketches) {
- KllDoublesSketch[] sketches = new
KllDoublesSketch[serializedSketches.length];
- for (int i = 0; i < serializedSketches.length; i++) {
- sketches[i] =
KllDoublesSketch.wrap(MemorySegment.ofArray(serializedSketches[i]).asReadOnly());
- }
- return sketches;
+ /// Wraps the serialized sketch for a non-null input row without copying its
bytes.
+ private static KllDoublesSketch deserializeSketch(byte[] serializedSketch) {
+ return
KllDoublesSketch.wrap(MemorySegment.ofArray(serializedSketch).asReadOnly());
}
@Nullable
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumValuesIntegerTupleSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumValuesIntegerTupleSketchAggregationFunction.java
index d3b8c1bed8c..48cc151dd88 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumValuesIntegerTupleSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumValuesIntegerTupleSketchAggregationFunction.java
@@ -56,9 +56,6 @@ public class SumValuesIntegerTupleSketchAggregationFunction
extends IntegerTuple
return _nullHandlingEnabled ? null : 0L;
}
double retainedTotal = 0L;
- accumulator.setNominalEntries(_nominalEntries);
- accumulator.setSetOperations(_setOps);
- accumulator.setThreshold(_accumulatorThreshold);
TupleSketch<IntegerSummary> result = accumulator.getResult();
TupleSketchIterator<IntegerSummary> summaries = result.iterator();
while (summaries.next()) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntermediateResultByteBufferSerDeTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntermediateResultByteBufferSerDeTest.java
index a0c5463060d..5208272911f 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntermediateResultByteBufferSerDeTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntermediateResultByteBufferSerDeTest.java
@@ -50,9 +50,11 @@ import org.apache.datasketches.cpc.CpcSketch;
import org.apache.datasketches.frequencies.FrequentItemsSketch;
import org.apache.datasketches.frequencies.FrequentLongsSketch;
import org.apache.datasketches.kll.KllDoublesSketch;
+import org.apache.datasketches.theta.ThetaSetOperationBuilder;
import org.apache.datasketches.theta.ThetaSketch;
import org.apache.datasketches.theta.UpdatableThetaSketch;
import org.apache.datasketches.tuple.aninteger.IntegerSummary;
+import org.apache.datasketches.tuple.aninteger.IntegerSummarySetOperations;
import org.apache.datasketches.tuple.aninteger.IntegerTupleSketch;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.Literal;
@@ -228,24 +230,19 @@ public class IntermediateResultByteBufferSerDeTest {
CpcSketch sketch = new CpcSketch(12);
sketch.update(1);
sketch.update(2);
- CpcSketchAccumulator accumulator = new CpcSketchAccumulator();
- accumulator.apply(sketch);
- return accumulator;
+ return new CpcSketchAccumulator(12, 2, sketch);
}
private static TupleIntSketchAccumulator tupleIntSketchAccumulator() {
IntegerTupleSketch sketch = new IntegerTupleSketch(12,
IntegerSummary.Mode.Sum);
sketch.update(1L, 1);
sketch.update(2L, 1);
- TupleIntSketchAccumulator accumulator = new TupleIntSketchAccumulator();
- accumulator.apply(sketch.compact());
- return accumulator;
+ return new TupleIntSketchAccumulator(
+ new IntegerSummarySetOperations(IntegerSummary.Mode.Sum,
IntegerSummary.Mode.Sum), 16384, 2, sketch.compact());
}
private static ThetaSketchAccumulator thetaSketchAccumulator() {
- ThetaSketchAccumulator accumulator = new ThetaSketchAccumulator();
- accumulator.apply(thetaSketch(1, 2));
- return accumulator;
+ return new ThetaSketchAccumulator(new ThetaSetOperationBuilder(), 2,
thetaSketch(1, 2));
}
private static ThetaSketch thetaSketch(long... values) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLBytesAggregationTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLBytesAggregationTest.java
new file mode 100644
index 00000000000..2bf3ca26f1c
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLBytesAggregationTest.java
@@ -0,0 +1,113 @@
+/**
+ * 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.core.query.aggregation.function;
+
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.apache.datasketches.kll.KllDoublesSketch;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.SyntheticBlockValSets;
+import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
+import org.roaringbitmap.RoaringBitmap;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Checks that KLL aggregation only deserializes included, non-null BYTES
rows.
+/// Test state is local to each invocation, so these tests can run
concurrently.
+public class PercentileKLLBytesAggregationTest {
+ private static final ExpressionContext COLUMN =
ExpressionContext.forIdentifier("column");
+
+ @DataProvider
+ public Object[][] bytesInputs() {
+ KllDoublesSketch sketch = KllDoublesSketch.newHeapInstance();
+ sketch.update(42.0);
+ byte[] valid = sketch.toByteArray();
+ byte[] invalid = new byte[0];
+ return new Object[][]{
+ // Null BYTES values use an empty default, which is not a serialized
KLL sketch.
+ {true, 2, RoaringBitmap.bitmapOf(1), new byte[][]{valid, invalid}, 1},
+ // Entries after length must not be read, regardless of null handling.
+ {true, 1, null, new byte[][]{valid, invalid}, 1},
+ {false, 1, null, new byte[][]{valid, invalid}, 1},
+ {true, 0, null, new byte[][]{invalid}, 0},
+ {false, 0, null, new byte[][]{invalid}, 0},
+ {true, 2, RoaringBitmap.bitmapOf(0, 1), new byte[][]{invalid,
invalid}, 0}
+ };
+ }
+
+ @Test(dataProvider = "bytesInputs")
+ public void testAggregate(boolean nullHandlingEnabled, int length, @Nullable
RoaringBitmap nullBitmap,
+ byte[][] values, int expectedCount) {
+ PercentileKLLAggregationFunction function = function(nullHandlingEnabled);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(length, holder, block(nullBitmap, values));
+ assertResult(function.extractAggregationResult(holder), expectedCount);
+ }
+
+ @Test(dataProvider = "bytesInputs")
+ public void testAggregateGroupBySV(boolean nullHandlingEnabled, int length,
@Nullable RoaringBitmap nullBitmap,
+ byte[][] values, int expectedCount) {
+ PercentileKLLAggregationFunction function = function(nullHandlingEnabled);
+ GroupByResultHolder holder = function.createGroupByResultHolder(1, 1);
+ function.aggregateGroupBySV(length, new int[length], holder,
block(nullBitmap, values));
+ assertResult(function.extractGroupByResult(holder, 0), expectedCount);
+ }
+
+ @Test(dataProvider = "bytesInputs")
+ public void testAggregateGroupByMV(boolean nullHandlingEnabled, int length,
@Nullable RoaringBitmap nullBitmap,
+ byte[][] values, int expectedCount) {
+ PercentileKLLAggregationFunction function = function(nullHandlingEnabled);
+ GroupByResultHolder holder = function.createGroupByResultHolder(2, 2);
+ int[][] groupKeys = new int[length][];
+ for (int i = 0; i < length; i++) {
+ groupKeys[i] = new int[]{0, 1};
+ }
+ function.aggregateGroupByMV(length, groupKeys, holder, block(nullBitmap,
values));
+ assertResult(function.extractGroupByResult(holder, 0), expectedCount);
+ assertResult(function.extractGroupByResult(holder, 1), expectedCount);
+ }
+
+ private static PercentileKLLAggregationFunction function(boolean
nullHandlingEnabled) {
+ return new PercentileKLLAggregationFunction(
+ List.of(COLUMN,
ExpressionContext.forLiteral(Literal.doubleValue(50.0))), nullHandlingEnabled);
+ }
+
+ private static Map<ExpressionContext, BlockValSet> block(@Nullable
RoaringBitmap nullBitmap, byte[][] values) {
+ return Map.of(COLUMN, SyntheticBlockValSets.Bytes.create(nullBitmap,
values));
+ }
+
+ private static void assertResult(@Nullable KllDoublesSketch sketch, int
expectedCount) {
+ if (expectedCount == 0) {
+ assertTrue(sketch == null || sketch.isEmpty());
+ } else {
+ assertNotNull(sketch);
+ assertEquals(sketch.getN(), (long) expectedCount);
+ assertEquals(sketch.getQuantile(0.5), 42.0);
+ }
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SketchAccumulatorSerializationTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SketchAccumulatorSerializationTest.java
new file mode 100644
index 00000000000..32352218bea
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SketchAccumulatorSerializationTest.java
@@ -0,0 +1,117 @@
+/**
+ * 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.core.query.aggregation.function;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+import org.apache.datasketches.cpc.CpcSketch;
+import org.apache.datasketches.cpc.CpcUnion;
+import org.apache.datasketches.theta.ThetaSetOperationBuilder;
+import org.apache.datasketches.theta.ThetaUnion;
+import org.apache.datasketches.theta.UpdatableThetaSketch;
+import org.apache.datasketches.tuple.TupleUnion;
+import org.apache.datasketches.tuple.aninteger.IntegerSummary;
+import org.apache.datasketches.tuple.aninteger.IntegerSummarySetOperations;
+import org.apache.datasketches.tuple.aninteger.IntegerTupleSketch;
+import org.apache.pinot.common.CustomObject;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/// Checks that deserialized sketches retain their bytes and use the receiving
query's settings when merged.
+/// All mutable state belongs to individual test invocations, so the tests can
run concurrently.
+public class SketchAccumulatorSerializationTest {
+ @DataProvider
+ public Object[][] thresholds() {
+ return new Object[][]{{1}, {3}};
+ }
+
+ @Test(dataProvider = "thresholds")
+ public void testTuple(int threshold) {
+ IntegerTupleSketch left = new IntegerTupleSketch(12,
IntegerSummary.Mode.Sum);
+ IntegerTupleSketch right = new IntegerTupleSketch(12,
IntegerSummary.Mode.Sum);
+ for (int i = 0; i < 1000; i++) {
+ left.update(i, 1);
+ right.update(i + 500, 2);
+ }
+ TupleUnion<IntegerSummary> expected =
+ new TupleUnion<>(128, new
IntegerSummarySetOperations(IntegerSummary.Mode.Max, IntegerSummary.Mode.Max));
+ expected.union(left.compact());
+ expected.union(right.compact());
+ assertLifecycle(new
IntegerTupleSketchAggregationFunction(arguments(threshold),
IntegerSummary.Mode.Max, false),
+ 48, left.compact().toByteArray(), right.compact().toByteArray(),
expected.getResult().toByteArray());
+ }
+
+ @Test(dataProvider = "thresholds")
+ public void testCpc(int threshold) {
+ CpcSketch left = new CpcSketch(12);
+ CpcSketch right = new CpcSketch(12);
+ for (int i = 0; i < 1000; i++) {
+ left.update(i);
+ right.update(i + 500);
+ }
+ CpcUnion expected = new CpcUnion(7);
+ expected.update(left);
+ expected.update(right);
+ assertLifecycle(new
DistinctCountCPCSketchAggregationFunction(arguments(threshold), false), 49,
+ left.toByteArray(), right.toByteArray(),
expected.getResult().toByteArray());
+ }
+
+ @Test(dataProvider = "thresholds")
+ public void testTheta(int threshold) {
+ UpdatableThetaSketch left = UpdatableThetaSketch.builder().build();
+ UpdatableThetaSketch right = UpdatableThetaSketch.builder().build();
+ for (int i = 0; i < 1000; i++) {
+ left.update(i);
+ right.update(i + 500);
+ }
+ ThetaUnion expected = new
ThetaSetOperationBuilder().setNominalEntries(128).buildUnion();
+ expected.union(left.compact());
+ expected.union(right.compact());
+ assertLifecycle(new
DistinctCountThetaSketchAggregationFunction(arguments(threshold), false), 21,
+ thetaList(left.compact().toByteArray()),
thetaList(right.compact().toByteArray()),
+ thetaList(expected.getResult(false, null).toByteArray()));
+ }
+
+ private static List<ExpressionContext> arguments(int threshold) {
+ return List.of(ExpressionContext.forIdentifier("column"),
ExpressionContext.forLiteral(
+ Literal.stringValue("nominalEntries=128;accumulatorThreshold=" +
threshold)));
+ }
+
+ private static byte[] thetaList(byte[] sketch) {
+ return ByteBuffer.allocate(3 * Integer.BYTES +
sketch.length).putInt(1).putInt(47).putInt(sketch.length)
+ .put(sketch).array();
+ }
+
+ private static <I> void assertLifecycle(AggregationFunction<I, ?> function,
int type, byte[] leftBytes,
+ byte[] rightBytes, byte[] expectedBytes) {
+ I left = function.deserializeIntermediateResult(new CustomObject(type,
ByteBuffer.wrap(leftBytes)));
+ // Receiving one sketch must not downsample it, including when the
accumulator threshold is one.
+ AggregationFunction.SerializedResult serialized =
function.serializeIntermediateResult(left);
+ assertEquals(serialized.getType(), type);
+ assertEquals(serialized.getBytes(), leftBytes);
+ I right = function.deserializeIntermediateResult(new CustomObject(type,
ByteBuffer.wrap(rightBytes)));
+ I merged = function.merge(left, right);
+ assertEquals(function.serializeIntermediateResult(merged).getBytes(),
expectedBytes);
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/IntegerTupleSketchValueAggregator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/IntegerTupleSketchValueAggregator.java
index ec37d68e6d0..4a75aa66dd2 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/IntegerTupleSketchValueAggregator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/IntegerTupleSketchValueAggregator.java
@@ -22,6 +22,7 @@ import com.google.common.annotations.VisibleForTesting;
import java.lang.foreign.MemorySegment;
import java.util.List;
import javax.annotation.Nullable;
+import org.apache.datasketches.common.Util;
import org.apache.datasketches.tuple.TupleSketch;
import org.apache.datasketches.tuple.TupleUnion;
import org.apache.datasketches.tuple.aninteger.IntegerSummary;
@@ -126,22 +127,13 @@ public class IntegerTupleSketchValueAggregator implements
ValueAggregator<byte[]
return true;
}
- /// Returns the maximum number of storage bytes required for a Compact
Integer Tuple Sketch with the given
- /// number of actual entries. Note that this assumes the worst case of the
sketch in
- /// estimation mode, which requires storing theta and count.
- /// @return the maximum number of storage bytes required for a Compact
Integer Tuple Sketch with the given number
- /// of entries.
+ /// Returns the maximum serialized size after unioning at the configured
nominal entries. The union rounds the
+ /// nominal entries up to a power of two. In estimation mode the compact
sketch has a 24-byte header, followed by
+ /// an 8-byte hash and a 5-byte integer summary (value and mode) for each
retained entry.
@Override
public int getMaxAggregatedValueByteSize() {
- if (_nominalEntries == 0) {
- return 8;
- }
- if (_nominalEntries == 1) {
- return 16;
- }
- int longSizeInBytes = Long.BYTES;
- int intSizeInBytes = Integer.BYTES;
- return (_nominalEntries * (longSizeInBytes + intSizeInBytes)) + 24;
+ long nominalEntries = Util.ceilingPowerOf2(_nominalEntries);
+ return Math.toIntExact(nominalEntries * (Long.BYTES + Integer.BYTES +
Byte.BYTES) + 24);
}
@Override
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulator.java
index abcd6e9380b..82c4c23fffd 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulator.java
@@ -22,26 +22,20 @@ import org.apache.datasketches.cpc.CpcSketch;
import org.apache.datasketches.cpc.CpcUnion;
-/// Intermediate state used by `DistinctCountCPCSketchAggregationFunction`
which gives
-/// the end user more control over how sketches are merged for performance.
-/// The end user can set parameters that trade-off more memory usage for more
pre-aggregation.
+/// Buffers CPC sketches for deferred merges with the configured precision.
+/// Instances are mutable and not thread-safe.
public class CpcSketchAccumulator extends CustomObjectAccumulator<CpcSketch> {
- private int _lgNominalEntries = 4;
+ private final int _lgNominalEntries;
private CpcUnion _union;
- public CpcSketchAccumulator() {
- }
-
- // Note: The accumulator is serialized as a sketch. This means that the
majority of the processing
- // happens on serialization. Therefore, when deserialized, the values may be
null and will
- // require re-initialisation. Since the primary use case is at query time
for the Broker
- // and Server, these properties are already in memory and are re-set.
public CpcSketchAccumulator(int lgNominalEntries, int threshold) {
super(threshold);
_lgNominalEntries = lgNominalEntries;
}
- public void setLgNominalEntries(int lgNominalEntries) {
+ /// Retains a deserialized sketch without unioning it, using the receiving
query's configuration for future merges.
+ public CpcSketchAccumulator(int lgNominalEntries, int threshold, CpcSketch
initialSketch) {
+ super(threshold, initialSketch);
_lgNominalEntries = lgNominalEntries;
}
@@ -61,20 +55,13 @@ public class CpcSketchAccumulator extends
CustomObjectAccumulator<CpcSketch> {
@Override
public CpcSketch getResult() {
+ // Return a single pending sketch unchanged. Its precision is not adjusted
until a union is needed.
+ if (getNumInputs() == 1 && _accumulator.size() == 1) {
+ return _accumulator.get(0);
+ }
if (_union == null) {
_union = new CpcUnion(_lgNominalEntries);
}
- // Return the default update "gadget" sketch as a compact sketch
- if (isEmpty()) {
- return _union.getResult();
- }
- // Corner-case: the parameters are not strictly respected when there is a
single sketch.
- // This single sketch might have been the result of a previously
accumulated union and
- // would already have the parameters set. The sketch is returned as-is
without adjusting
- // nominal entries which requires an additional union operation.
- if (getNumInputs() == 1 && _accumulator != null && _accumulator.size() ==
1) {
- return _accumulator.get(0);
- }
flush();
return _union.getResult();
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CustomObjectAccumulator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CustomObjectAccumulator.java
index 55cba5452a7..c02d37f030d 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CustomObjectAccumulator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/CustomObjectAccumulator.java
@@ -22,34 +22,27 @@ import com.google.common.base.Preconditions;
import java.util.ArrayList;
-/// Intermediate state used by some aggregation functions which gives the end
user more control over how custom objects
-/// are merged for performance reasons. Some custom objects such as
DataSketches have better merge performance when
-/// more than two items are merged at once through the elimination of
intermediate bookkeeping overheads.
-///
-/// The end user can set a value for the "threshold" parameter that defers the
merge operation until at least as many
-/// items are ready to be merged, or the callee forces the merge directly via
"getResult" - e.g. on serialization.
-/// This data structure trades-off more memory usage for a greater degree of
pre-aggregation in the accumulator state.
+/// Buffers intermediate values until the configured threshold is reached or a
result is requested.
+/// Subclasses merge pending values in [#flush()]. Instances are mutable and
not thread-safe.
public abstract class CustomObjectAccumulator<T> {
protected ArrayList<T> _accumulator;
- private int _threshold;
+ private final int _threshold;
private int _numInputs = 0;
- public CustomObjectAccumulator() {
- this(2);
- }
-
- public CustomObjectAccumulator(int threshold) {
- setThreshold(threshold);
- }
-
- /// Sets the threshold that determines how much memory to use for the
internal accumulator state before
- /// the intermediate state is merged.
- /// @param threshold the threshold \[> 0\].
- public void setThreshold(int threshold) {
+ protected CustomObjectAccumulator(int threshold) {
Preconditions.checkArgument(threshold > 0, "Invalid threshold: %s, must be
positive", threshold);
_threshold = threshold;
}
+ /// Retains an initial result without flushing, even when the threshold is
one. A deserialized sketch can then
+ /// be returned unchanged until another input is added, rather than being
unioned again during deserialization.
+ protected CustomObjectAccumulator(int threshold, T initialResult) {
+ this(threshold);
+ _accumulator = new ArrayList<>(1);
+ _accumulator.add(Preconditions.checkNotNull(initialResult));
+ _numInputs = 1;
+ }
+
/// Returns the configured threshold for this accumulator.
public int getThreshold() {
return _threshold;
@@ -65,9 +58,7 @@ public abstract class CustomObjectAccumulator<T> {
return _numInputs == 0;
}
- /// Forces the item T in internal state to be merged with all pending items
in the accumulator state
- /// and returns the result.
- /// @return T result of the merge.
+ /// Returns the accumulated result, merging pending inputs as needed.
public abstract T getResult();
/// Merges the pending items into the internal state and clears them,
without building a result.
@@ -83,8 +74,7 @@ public abstract class CustomObjectAccumulator<T> {
applyInternal(result);
}
- /// Adds a new item to the accumulator state. If the accumulator state is
equal to the threshold value,
- /// the internal state is updated and the accumulator state is cleared.
+ /// Adds an input and flushes pending inputs when their count reaches the
threshold.
/// @param item the item to add to the accumulator state, cannot be null.
public void apply(T item) {
Preconditions.checkNotNull(item);
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/ThetaSketchAccumulator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/ThetaSketchAccumulator.java
index 94dba63d18d..178f05ebdf9 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/ThetaSketchAccumulator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/ThetaSketchAccumulator.java
@@ -24,30 +24,21 @@ import org.apache.datasketches.theta.ThetaSketch;
import org.apache.datasketches.theta.ThetaUnion;
-/// Intermediate state used by `DistinctCountThetaSketchAggregationFunction`
which gives
-/// the end user more control over how sketches are merged for performance.
-/// In particular, the Theta Sketch Union "early-stop" optimisation can be
used - ordered sketches require no further
-/// processing beyond the minimum Theta value.
-/// The union operation initialises an empty "gadget" bookkeeping sketch that
is updated with hashed entries
-/// that fall below the minimum Theta value for all input sketches ("Broder
Rule"). When the initial Theta value is
-/// set to the minimum immediately, further gains can be realised.
+/// Buffers theta sketches and unions pending inputs in ascending theta order
to reduce intermediate retained entries.
+/// Instances are mutable and not thread-safe. The supplied builder must not
be modified after construction.
public class ThetaSketchAccumulator extends
CustomObjectAccumulator<ThetaSketch> {
- private ThetaSetOperationBuilder _setOperationBuilder = new
ThetaSetOperationBuilder();
+ private final ThetaSetOperationBuilder _setOperationBuilder;
private ThetaUnion _union;
- public ThetaSketchAccumulator() {
- }
-
- // Note: The accumulator is serialized as a sketch. This means that the
majority of the processing
- // happens on serialization. Therefore, when deserialized, the values may be
null and will
- // require re-initialisation. Since the primary use case is at query time
for the Broker
- // and Server, these properties are already in memory and are re-set.
public ThetaSketchAccumulator(ThetaSetOperationBuilder setOperationBuilder,
int threshold) {
super(threshold);
_setOperationBuilder = setOperationBuilder;
}
- public void setSetOperationBuilder(ThetaSetOperationBuilder
setOperationBuilder) {
+ /// Retains a deserialized sketch without unioning it, using the receiving
query's configuration for future merges.
+ public ThetaSketchAccumulator(ThetaSetOperationBuilder setOperationBuilder,
int threshold,
+ ThetaSketch initialSketch) {
+ super(threshold, initialSketch);
_setOperationBuilder = setOperationBuilder;
}
@@ -60,15 +51,7 @@ public class ThetaSketchAccumulator extends
CustomObjectAccumulator<ThetaSketch>
_union = _setOperationBuilder.buildUnion();
}
- // Performance optimization: ensure that the minimum Theta is used for
"early stop".
- // The "early stop" optimization is implemented in the Apache Datasketches
ThetaUnion operation for
- // ordered and compact Theta sketches. Internally, a compact and ordered
Theta sketch can be
- // compared to a sorted array of K items. When performing a union, only
those items from
- // the input sketch less than Theta need to be processed. The loop
terminates as soon as a hash
- // is seen that is > Theta.
- // The following "sort" improves on this further by selecting the minimal
Theta value up-front,
- // which results in fewer redundant entries being retained and
subsequently discarded during the
- // union operation.
+ // Start with the lowest theta so later inputs do not retain hashes that
would be discarded by that input.
_accumulator.sort(Comparator.comparingDouble(ThetaSketch::getTheta));
for (ThetaSketch accumulatedSketch : _accumulator) {
_union.union(accumulatedSketch);
@@ -78,20 +61,13 @@ public class ThetaSketchAccumulator extends
CustomObjectAccumulator<ThetaSketch>
@Override
public ThetaSketch getResult() {
+ // Return a single pending sketch unchanged. Its nominal entries are not
adjusted until a union is needed.
+ if (getNumInputs() == 1 && _accumulator.size() == 1) {
+ return _accumulator.get(0);
+ }
if (_union == null) {
_union = _setOperationBuilder.buildUnion();
}
- // Return the default update "gadget" sketch as a compact sketch
- if (isEmpty()) {
- return _union.getResult(false, null);
- }
- // Corner-case: the parameters are not strictly respected when there is a
single sketch.
- // This single sketch might have been the result of a previously
accumulated union and
- // would already have the parameters set. The sketch is returned as-is
without adjusting
- // nominal entries which requires an additional union operation.
- if (getNumInputs() == 1 && _accumulator != null && _accumulator.size() ==
1) {
- return _accumulator.get(0);
- }
flush();
return _union.getResult(false, null);
}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/TupleIntSketchAccumulator.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/TupleIntSketchAccumulator.java
index 3f5a16392e6..ccc11c40aaf 100644
---
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/TupleIntSketchAccumulator.java
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/TupleIntSketchAccumulator.java
@@ -25,38 +25,25 @@ import
org.apache.datasketches.tuple.aninteger.IntegerSummary;
import org.apache.datasketches.tuple.aninteger.IntegerSummarySetOperations;
-/// Intermediate state used by `IntegerTupleSketchAggregationFunction` which
gives
-/// the end user more control over how sketches are merged for performance.
-/// In particular, the Theta Sketch Union "early-stop" optimisation can be
used - ordered sketches require no further
-/// processing beyond the minimum Theta value. This applies to Tuple sketches
because they are an extension of the
-/// Theta sketch.
-/// The union operation initialises an empty "gadget" bookkeeping sketch that
is updated with hashed entries
-/// that fall below the minimum Theta value for all input sketches ("Broder
Rule"). When the initial Theta value is
-/// set to the minimum immediately, further gains can be realised.
+/// Buffers integer tuple sketches and reuses a union with the configured
summary operations and nominal entries.
+/// Instances are mutable and not thread-safe.
public class TupleIntSketchAccumulator extends
CustomObjectAccumulator<TupleSketch<IntegerSummary>> {
- private IntegerSummarySetOperations _setOperations;
- private int _nominalEntries;
+ private final IntegerSummarySetOperations _setOperations;
+ private final int _nominalEntries;
private TupleUnion<IntegerSummary> _union;
- public TupleIntSketchAccumulator() {
- }
-
- // Note: The accumulator is serialized as a sketch. This means that the
majority of the processing
- // happens on serialization. Therefore, when deserialized, the values may be
null and will
- // require re-initialisation. Since the primary use case is at query time
for the Broker
- // and Server, these properties are already in memory and are re-set.
public TupleIntSketchAccumulator(IntegerSummarySetOperations setOperations,
int nominalEntries, int threshold) {
super(threshold);
_nominalEntries = nominalEntries;
_setOperations = setOperations;
}
- public void setSetOperations(IntegerSummarySetOperations setOperations) {
- _setOperations = setOperations;
- }
-
- public void setNominalEntries(int nominalEntries) {
+ /// Retains a deserialized sketch without unioning it, using the receiving
query's configuration for future merges.
+ public TupleIntSketchAccumulator(IntegerSummarySetOperations setOperations,
int nominalEntries, int threshold,
+ TupleSketch<IntegerSummary> initialSketch) {
+ super(threshold, initialSketch);
_nominalEntries = nominalEntries;
+ _setOperations = setOperations;
}
@Override
@@ -68,15 +55,6 @@ public class TupleIntSketchAccumulator extends
CustomObjectAccumulator<TupleSket
_union = new TupleUnion<>(_nominalEntries, _setOperations);
}
- // Performance optimization: ensure that the minimum Theta is used for
"early stop".
- // The "early stop" optimization is implemented in the Apache Datasketches
TupleUnion operation for
- // ordered and compact Theta sketches. Internally, a compact and ordered
Theta sketch can be
- // compared to a sorted array of K items. When performing a union, only
those items from
- // the input sketch less than Theta need to be processed. The loop
terminates as soon as a hash
- // is seen that is > Theta.
- // The following "sort" improves on this further by selecting the minimal
Theta value up-front,
- // which results in fewer redundant entries being retained and
subsequently discarded during the
- // union operation.
_accumulator.sort(Comparator.comparingDouble(TupleSketch::getTheta));
for (TupleSketch<IntegerSummary> accumulatedSketch : _accumulator) {
_union.union(accumulatedSketch);
@@ -86,20 +64,13 @@ public class TupleIntSketchAccumulator extends
CustomObjectAccumulator<TupleSket
@Override
public TupleSketch<IntegerSummary> getResult() {
+ // Return a single pending sketch unchanged. Its nominal entries are not
adjusted until a union is needed.
+ if (getNumInputs() == 1 && _accumulator.size() == 1) {
+ return _accumulator.get(0);
+ }
if (_union == null) {
_union = new TupleUnion<>(_nominalEntries, _setOperations);
}
- // Return the default update "gadget" sketch as a compact sketch
- if (isEmpty()) {
- return _union.getResult();
- }
- // Corner-case: the parameters are not strictly respected when there is a
single sketch.
- // This single sketch might have been the result of a previously
accumulated union and
- // would already have the parameters set. The sketch is returned as-is
without adjusting
- // nominal entries which requires an additional union operation.
- if (getNumInputs() == 1 && _accumulator != null && _accumulator.size() ==
1) {
- return _accumulator.get(0);
- }
flush();
return _union.getResult();
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/IntegerTupleSketchValueAggregatorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/IntegerTupleSketchValueAggregatorTest.java
index 4ea2d49fab5..6e2825511b3 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/IntegerTupleSketchValueAggregatorTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/aggregator/IntegerTupleSketchValueAggregatorTest.java
@@ -23,9 +23,13 @@ import org.apache.datasketches.tuple.TupleSketch;
import org.apache.datasketches.tuple.TupleUnion;
import org.apache.datasketches.tuple.aninteger.IntegerSummary;
import org.apache.datasketches.tuple.aninteger.IntegerTupleSketch;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
public class IntegerTupleSketchValueAggregatorTest {
@@ -60,7 +64,7 @@ public class IntegerTupleSketchValueAggregatorTest {
new IntegerTupleSketchValueAggregator(List.of(),
IntegerSummary.Mode.Sum);
TupleSketch<IntegerSummary> merged = toSketch(agg.applyAggregatedValue(s1,
s2));
assertEquals(merged.getEstimate(), 2.0);
- assertEquals(agg.getMaxAggregatedValueByteSize(), 196632);
+ assertEquals(agg.getMaxAggregatedValueByteSize(), 213016);
}
@Test
@@ -73,7 +77,36 @@ public class IntegerTupleSketchValueAggregatorTest {
new IntegerTupleSketchValueAggregator(List.of(),
IntegerSummary.Mode.Sum);
TupleSketch<IntegerSummary> merged = toSketch(agg.applyRawValue(s1,
agg.serializeAggregatedValue(s2)));
assertEquals(merged.getEstimate(), 2.0);
- assertEquals(agg.getMaxAggregatedValueByteSize(), 196632);
+ assertEquals(agg.getMaxAggregatedValueByteSize(), 213016);
+ }
+
+ @DataProvider
+ public Object[][] nominalEntries() {
+ return new Object[][]{{null, 16384}, {1, 1}, {3, 4}, {1000, 1024}};
+ }
+
+ @Test(dataProvider = "nominalEntries")
+ public void testMaxSerializedSize(Integer nominalEntries, int
retainedEntries) {
+ List<ExpressionContext> arguments = nominalEntries == null
+ ? List.of()
+ :
List.of(ExpressionContext.forLiteral(Literal.intValue(nominalEntries)));
+ IntegerTupleSketchValueAggregator agg = new
IntegerTupleSketchValueAggregator(arguments, IntegerSummary.Mode.Sum);
+ int maxSize = agg.getMaxAggregatedValueByteSize();
+ for (int inputSize : new int[]{0, 1, retainedEntries * 2}) {
+ IntegerTupleSketch input = new IntegerTupleSketch(16,
IntegerSummary.Mode.Sum);
+ for (int i = 0; i < inputSize; i++) {
+ input.update(i, 1);
+ }
+ Object result =
agg.getInitialAggregatedValue(input.compact().toByteArray());
+ byte[] bytes = agg.serializeAggregatedValue(result);
+ assertTrue(bytes.length <= maxSize, "Serialized sketch exceeds the size
reserved by the index writer");
+ if (inputSize > retainedEntries) {
+ TupleSketch<IntegerSummary> sketch = toSketch(result);
+ assertTrue(sketch.isEstimationMode());
+ assertEquals(sketch.getRetainedEntries(), retainedEntries);
+ assertEquals(bytes.length, maxSize);
+ }
+ }
}
@SuppressWarnings("unchecked")
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulatorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulatorTest.java
index 86326de573e..9d3e88f0436 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulatorTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulatorTest.java
@@ -22,9 +22,13 @@ package org.apache.pinot.segment.local.customobject;
import java.util.stream.IntStream;
import org.apache.datasketches.cpc.CpcSketch;
import org.apache.datasketches.cpc.CpcUnion;
-import org.testng.Assert;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
public class CpcSketchAccumulatorTest {
private final int _lgNominalEntries = 20;
@@ -33,8 +37,8 @@ public class CpcSketchAccumulatorTest {
@Test
public void testEmptyAccumulator() {
CpcSketchAccumulator accumulator = new
CpcSketchAccumulator(_lgNominalEntries, 2);
- Assert.assertTrue(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(), 0.0);
+ assertTrue(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), 0.0);
}
@Test
@@ -45,8 +49,47 @@ public class CpcSketchAccumulatorTest {
CpcSketchAccumulator accumulator = new
CpcSketchAccumulator(_lgNominalEntries, 2);
accumulator.apply(sketch);
- Assert.assertFalse(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch.getEstimate());
+ assertFalse(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), sketch.getEstimate());
+ }
+
+ @Test
+ public void testSeededSketchIsReturnedUnchanged() {
+ for (int numEntries : new int[]{0, 1000}) {
+ CpcSketch sketch = new CpcSketch(12);
+ IntStream.range(0, numEntries).forEach(sketch::update);
+ for (int threshold = 1; threshold <= 3; threshold++) {
+ CpcSketchAccumulator accumulator = new CpcSketchAccumulator(10,
threshold, sketch);
+
+ assertFalse(accumulator.isEmpty());
+ assertEquals(accumulator.getNumInputs(), 1);
+ assertSame(accumulator.getResult(), sketch);
+ assertSame(accumulator.getResult(), sketch);
+ }
+ }
+ }
+
+ @Test
+ public void testMergeAfterReadingSeededSketch() {
+ CpcSketch input1 = new CpcSketch(12);
+ IntStream.range(0, 1000).forEach(input1::update);
+ byte[] bytes = input1.toByteArray();
+ CpcSketch sketch2 = new CpcSketch(12);
+ IntStream.range(500, 1500).forEach(sketch2::update);
+ for (int threshold = 1; threshold <= 3; threshold++) {
+ CpcSketch sketch1 = CpcSketch.heapify(bytes);
+ CpcSketchAccumulator accumulator = new CpcSketchAccumulator(10,
threshold, sketch1);
+ assertEquals(accumulator.getResult().toByteArray(), bytes);
+ accumulator.merge(new CpcSketchAccumulator(10, threshold, sketch2));
+
+ CpcUnion expected = new CpcUnion(10);
+ expected.update(sketch1);
+ expected.update(sketch2);
+ CpcSketch result = accumulator.getResult();
+ assertEquals(accumulator.getNumInputs(), 2);
+ assertEquals(result.getLgK(), 10);
+ assertEquals(result.getEstimate(), expected.getResult().getEstimate(),
0.0);
+ }
}
@Test
@@ -62,7 +105,7 @@ public class CpcSketchAccumulatorTest {
accumulator2.apply(sketch2);
accumulator1.merge(accumulator2);
- Assert.assertEquals(accumulator1.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate(),
+ assertEquals(accumulator1.getResult().getEstimate(), sketch1.getEstimate()
+ sketch2.getEstimate(),
_epsilon);
}
@@ -77,7 +120,7 @@ public class CpcSketchAccumulatorTest {
accumulator.apply(sketch1);
accumulator.apply(sketch2);
- Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate(), _epsilon);
+ assertEquals(accumulator.getResult().getEstimate(), sketch1.getEstimate()
+ sketch2.getEstimate(), _epsilon);
}
@Test
@@ -92,7 +135,7 @@ public class CpcSketchAccumulatorTest {
accumulator.apply(sketch);
expected.update(sketch);
}
- Assert.assertEquals(accumulator.getResult().getEstimate(),
expected.getResult().getEstimate(), _epsilon,
+ assertEquals(accumulator.getResult().getEstimate(),
expected.getResult().getEstimate(), _epsilon,
"threshold " + threshold);
}
}
@@ -107,8 +150,8 @@ public class CpcSketchAccumulatorTest {
accumulator.apply(sketch);
}
double first = accumulator.getResult().getEstimate();
- Assert.assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
- Assert.assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
+ assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
+ assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
}
@Test
@@ -118,7 +161,7 @@ public class CpcSketchAccumulatorTest {
accumulator.merge(emptyAccumulator);
- Assert.assertTrue(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(), 0.0);
+ assertTrue(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), 0.0);
}
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/ThetaSketchAccumulatorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/ThetaSketchAccumulatorTest.java
index 088d4f9ddfc..00e4296a510 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/ThetaSketchAccumulatorTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/ThetaSketchAccumulatorTest.java
@@ -19,15 +19,20 @@
package org.apache.pinot.segment.local.customobject;
+import java.lang.foreign.MemorySegment;
import java.util.stream.IntStream;
import org.apache.datasketches.theta.ThetaSetOperationBuilder;
import org.apache.datasketches.theta.ThetaSketch;
import org.apache.datasketches.theta.ThetaUnion;
import org.apache.datasketches.theta.UpdatableThetaSketch;
-import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
public class ThetaSketchAccumulatorTest {
private ThetaSetOperationBuilder _setOperationBuilder;
@@ -40,8 +45,8 @@ public class ThetaSketchAccumulatorTest {
@Test
public void testEmptyAccumulator() {
ThetaSketchAccumulator accumulator = new
ThetaSketchAccumulator(_setOperationBuilder, 2);
- Assert.assertTrue(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(), 0.0);
+ assertTrue(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), 0.0);
}
@Test
@@ -53,8 +58,51 @@ public class ThetaSketchAccumulatorTest {
ThetaSketchAccumulator accumulator = new
ThetaSketchAccumulator(_setOperationBuilder, 2);
accumulator.apply(sketch);
- Assert.assertFalse(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch.getEstimate());
+ assertFalse(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), sketch.getEstimate());
+ }
+
+ @Test
+ public void testSeededSketchIsReturnedUnchanged() {
+ ThetaSetOperationBuilder builder = new
ThetaSetOperationBuilder().setNominalEntries(128);
+ for (int numEntries : new int[]{0, 1000}) {
+ UpdatableThetaSketch input = UpdatableThetaSketch.builder().build();
+ IntStream.range(0, numEntries).forEach(input::update);
+ ThetaSketch sketch = input.compact();
+ for (int threshold = 1; threshold <= 3; threshold++) {
+ ThetaSketchAccumulator accumulator = new
ThetaSketchAccumulator(builder, threshold, sketch);
+
+ assertFalse(accumulator.isEmpty());
+ assertEquals(accumulator.getNumInputs(), 1);
+ assertSame(accumulator.getResult(), sketch);
+ assertSame(accumulator.getResult(), sketch);
+ }
+ }
+ }
+
+ @Test
+ public void testMergeAfterReadingSeededSketch() {
+ UpdatableThetaSketch input1 = UpdatableThetaSketch.builder().build();
+ IntStream.range(0, 1000).forEach(input1::update);
+ byte[] bytes = input1.compact().toByteArray();
+ UpdatableThetaSketch input2 = UpdatableThetaSketch.builder().build();
+ IntStream.range(500, 1500).forEach(input2::update);
+ ThetaSketch sketch2 = input2.compact();
+ ThetaSetOperationBuilder builder = new
ThetaSetOperationBuilder().setNominalEntries(128);
+ for (int threshold = 1; threshold <= 3; threshold++) {
+ ThetaSketch sketch1 = ThetaSketch.heapify(MemorySegment.ofArray(bytes));
+ ThetaSketchAccumulator accumulator = new ThetaSketchAccumulator(builder,
threshold, sketch1);
+ assertEquals(accumulator.getResult().toByteArray(), bytes);
+ accumulator.merge(new ThetaSketchAccumulator(builder, threshold,
sketch2));
+
+ ThetaUnion expected = builder.buildUnion();
+ expected.union(sketch1);
+ expected.union(sketch2);
+ ThetaSketch result = accumulator.getResult();
+ assertEquals(accumulator.getNumInputs(), 2);
+ assertEquals(result.getRetainedEntries(true), 128);
+ assertEquals(result.getEstimate(), expected.getResult().getEstimate(),
0.0);
+ }
}
@Test
@@ -72,7 +120,7 @@ public class ThetaSketchAccumulatorTest {
accumulator2.apply(sketch2);
accumulator1.merge(accumulator2);
- Assert.assertEquals(accumulator1.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate());
+ assertEquals(accumulator1.getResult().getEstimate(), sketch1.getEstimate()
+ sketch2.getEstimate());
}
@Test
@@ -88,7 +136,7 @@ public class ThetaSketchAccumulatorTest {
accumulator.apply(sketch1);
accumulator.apply(sketch2);
- Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate());
+ assertEquals(accumulator.getResult().getEstimate(), sketch1.getEstimate()
+ sketch2.getEstimate());
}
@Test
@@ -104,7 +152,7 @@ public class ThetaSketchAccumulatorTest {
accumulator.apply(sketch);
expected.union(sketch);
}
- Assert.assertEquals(accumulator.getResult().getEstimate(),
expected.getResult().getEstimate(), 0.0,
+ assertEquals(accumulator.getResult().getEstimate(),
expected.getResult().getEstimate(), 0.0,
"threshold " + threshold);
}
}
@@ -119,8 +167,8 @@ public class ThetaSketchAccumulatorTest {
accumulator.apply(input.compact());
}
double first = accumulator.getResult().getEstimate();
- Assert.assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
- Assert.assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
+ assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
+ assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
}
@Test
@@ -130,7 +178,7 @@ public class ThetaSketchAccumulatorTest {
accumulator.merge(emptyAccumulator);
- Assert.assertTrue(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(), 0.0);
+ assertTrue(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), 0.0);
}
}
diff --git
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/TupleIntSketchAccumulatorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/TupleIntSketchAccumulatorTest.java
index ba90de81b65..91567c30207 100644
---
a/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/TupleIntSketchAccumulatorTest.java
+++
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/TupleIntSketchAccumulatorTest.java
@@ -19,18 +19,24 @@
package org.apache.pinot.segment.local.customobject;
+import java.lang.foreign.MemorySegment;
import java.util.stream.IntStream;
import org.apache.datasketches.tuple.CompactTupleSketch;
import org.apache.datasketches.tuple.TupleSketch;
import org.apache.datasketches.tuple.TupleSketchIterator;
import org.apache.datasketches.tuple.TupleUnion;
import org.apache.datasketches.tuple.aninteger.IntegerSummary;
+import org.apache.datasketches.tuple.aninteger.IntegerSummaryDeserializer;
import org.apache.datasketches.tuple.aninteger.IntegerSummarySetOperations;
import org.apache.datasketches.tuple.aninteger.IntegerTupleSketch;
-import org.testng.Assert;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertSame;
+import static org.testng.Assert.assertTrue;
+
public class TupleIntSketchAccumulatorTest {
private IntegerSummarySetOperations _setOps;
@@ -45,8 +51,8 @@ public class TupleIntSketchAccumulatorTest {
@Test
public void testEmptyAccumulator() {
TupleIntSketchAccumulator accumulator = new
TupleIntSketchAccumulator(_setOps, _nominalEntries, 2);
- Assert.assertTrue(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(), 0.0);
+ assertTrue(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), 0.0);
}
@Test
@@ -58,8 +64,78 @@ public class TupleIntSketchAccumulatorTest {
TupleIntSketchAccumulator accumulator = new
TupleIntSketchAccumulator(_setOps, _nominalEntries, 2);
accumulator.apply(sketch);
- Assert.assertFalse(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch.getEstimate());
+ assertFalse(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), sketch.getEstimate());
+ }
+
+ @Test
+ public void testSeededSketchIsReturnedUnchanged() {
+ for (int numEntries : new int[]{0, 1000}) {
+ IntegerTupleSketch input = new IntegerTupleSketch(_lgK,
IntegerSummary.Mode.Sum);
+ IntStream.range(0, numEntries).forEach(i -> input.update(i, 1));
+ CompactTupleSketch<IntegerSummary> sketch = input.compact();
+ for (int threshold = 1; threshold <= 3; threshold++) {
+ TupleIntSketchAccumulator accumulator = new
TupleIntSketchAccumulator(_setOps, 16, threshold, sketch);
+
+ // A received sketch stays intact until another input needs to be
merged, even at threshold 1.
+ assertFalse(accumulator.isEmpty());
+ assertEquals(accumulator.getNumInputs(), 1);
+ assertSame(accumulator.getResult(), sketch);
+ assertSame(accumulator.getResult(), sketch);
+ }
+ }
+ }
+
+ @Test
+ public void testMergeSeededSketches() {
+ IntegerTupleSketch input1 = new IntegerTupleSketch(_lgK,
IntegerSummary.Mode.Sum);
+ input1.update(1L, 2);
+ IntegerTupleSketch input2 = new IntegerTupleSketch(_lgK,
IntegerSummary.Mode.Sum);
+ input2.update(1L, 3);
+ input2.update(2L, 4);
+ for (int threshold = 1; threshold <= 3; threshold++) {
+ TupleIntSketchAccumulator accumulator =
+ new TupleIntSketchAccumulator(_setOps, _nominalEntries, threshold,
input1.compact());
+ TupleIntSketchAccumulator other =
+ new TupleIntSketchAccumulator(_setOps, _nominalEntries, threshold,
input2.compact());
+ accumulator.merge(other);
+
+ TupleSketch<IntegerSummary> result = accumulator.getResult();
+ assertEquals(accumulator.getNumInputs(), 2);
+ assertEquals(result.getEstimate(), 2.0);
+ assertEquals(summarySum(result), 9L);
+ assertEquals(summarySum(accumulator.getResult()), 9L);
+ }
+ }
+
+ @Test
+ public void testMergeAfterReadingSeededSketch() {
+ IntegerTupleSketch input1 = new IntegerTupleSketch(_lgK,
IntegerSummary.Mode.Sum);
+ IntegerTupleSketch input2 = new IntegerTupleSketch(_lgK,
IntegerSummary.Mode.Sum);
+ IntStream.range(0, 1000).forEach(i -> {
+ input1.update(i, 1);
+ input2.update(i, 2);
+ });
+ byte[] bytes = input1.compact().toByteArray();
+ CompactTupleSketch<IntegerSummary> sketch2 = input2.compact();
+ IntegerSummarySetOperations setOps =
+ new IntegerSummarySetOperations(IntegerSummary.Mode.Max,
IntegerSummary.Mode.Max);
+ for (int threshold = 1; threshold <= 3; threshold++) {
+ TupleSketch<IntegerSummary> sketch1 =
+ TupleSketch.heapifySketch(MemorySegment.ofArray(bytes), new
IntegerSummaryDeserializer());
+ TupleIntSketchAccumulator accumulator = new
TupleIntSketchAccumulator(setOps, 16, threshold, sketch1);
+ assertEquals(accumulator.getResult().toByteArray(), bytes);
+ accumulator.merge(new TupleIntSketchAccumulator(setOps, 16, threshold,
sketch2));
+
+ TupleUnion<IntegerSummary> expected = new TupleUnion<>(16, setOps);
+ expected.union(sketch1);
+ expected.union(sketch2);
+ TupleSketch<IntegerSummary> result = accumulator.getResult();
+ TupleSketch<IntegerSummary> expectedResult = expected.getResult();
+ assertEquals(result.getRetainedEntries(),
expectedResult.getRetainedEntries());
+ assertEquals(result.getTheta(), expectedResult.getTheta());
+ assertEquals(summarySum(result), summarySum(expectedResult));
+ }
}
@Test
@@ -77,7 +153,7 @@ public class TupleIntSketchAccumulatorTest {
accumulator2.apply(sketch2);
accumulator1.merge(accumulator2);
- Assert.assertEquals(accumulator1.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate());
+ assertEquals(accumulator1.getResult().getEstimate(), sketch1.getEstimate()
+ sketch2.getEstimate());
}
@Test
@@ -93,7 +169,7 @@ public class TupleIntSketchAccumulatorTest {
accumulator.apply(sketch1);
accumulator.apply(sketch2);
- Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate());
+ assertEquals(accumulator.getResult().getEstimate(), sketch1.getEstimate()
+ sketch2.getEstimate());
}
// Summaries are summed, so a sketch merged twice doubles a summary while
leaving the estimate right.
@@ -112,8 +188,8 @@ public class TupleIntSketchAccumulatorTest {
}
TupleSketch<IntegerSummary> actual = accumulator.getResult();
TupleSketch<IntegerSummary> want = expected.getResult();
- Assert.assertEquals(actual.getEstimate(), want.getEstimate(), 0.0,
"threshold " + threshold);
- Assert.assertEquals(summarySum(actual), summarySum(want), "threshold " +
threshold);
+ assertEquals(actual.getEstimate(), want.getEstimate(), 0.0, "threshold "
+ threshold);
+ assertEquals(summarySum(actual), summarySum(want), "threshold " +
threshold);
}
}
@@ -127,8 +203,8 @@ public class TupleIntSketchAccumulatorTest {
accumulator.apply(input.compact());
}
long first = summarySum(accumulator.getResult());
- Assert.assertEquals(summarySum(accumulator.getResult()), first);
- Assert.assertEquals(summarySum(accumulator.getResult()), first);
+ assertEquals(summarySum(accumulator.getResult()), first);
+ assertEquals(summarySum(accumulator.getResult()), first);
}
private static long summarySum(TupleSketch<IntegerSummary> sketch) {
@@ -147,7 +223,7 @@ public class TupleIntSketchAccumulatorTest {
accumulator.merge(emptyAccumulator);
- Assert.assertTrue(accumulator.isEmpty());
- Assert.assertEquals(accumulator.getResult().getEstimate(), 0.0);
+ assertTrue(accumulator.isEmpty());
+ assertEquals(accumulator.getResult().getEstimate(), 0.0);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
