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 6d522d9a77d Do not build a discarded result when an accumulator
reaches its threshold (#19449)
6d522d9a77d is described below
commit 6d522d9a77da83bb88e49d2349f44c5e4b542640
Author: David Cromberge <[email protected]>
AuthorDate: Thu Sep 17 23:25:31 2026 +0200
Do not build a discarded result when an accumulator reaches its threshold
(#19449)
---
.../local/customobject/CpcSketchAccumulator.java | 24 ++++++-----
.../customobject/CustomObjectAccumulator.java | 8 ++--
.../local/customobject/ThetaSketchAccumulator.java | 38 +++++++++--------
.../customobject/TupleIntSketchAccumulator.java | 38 +++++++++--------
.../customobject/CpcSketchAccumulatorTest.java | 32 +++++++++++++++
.../customobject/ThetaSketchAccumulatorTest.java | 33 +++++++++++++++
.../TupleIntSketchAccumulatorTest.java | 47 ++++++++++++++++++++++
7 files changed, 176 insertions(+), 44 deletions(-)
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 8cbe53dc0b8..abcd6e9380b 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
@@ -46,11 +46,21 @@ public class CpcSketchAccumulator extends
CustomObjectAccumulator<CpcSketch> {
}
@Override
- public CpcSketch getResult() {
- return unionAll();
+ protected void flush() {
+ if (_accumulator == null || _accumulator.isEmpty()) {
+ return;
+ }
+ if (_union == null) {
+ _union = new CpcUnion(_lgNominalEntries);
+ }
+ for (CpcSketch accumulatedSketch : _accumulator) {
+ _union.update(accumulatedSketch);
+ }
+ _accumulator.clear();
}
- private CpcSketch unionAll() {
+ @Override
+ public CpcSketch getResult() {
if (_union == null) {
_union = new CpcUnion(_lgNominalEntries);
}
@@ -62,14 +72,10 @@ public class CpcSketchAccumulator extends
CustomObjectAccumulator<CpcSketch> {
// 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) {
+ if (getNumInputs() == 1 && _accumulator != null && _accumulator.size() ==
1) {
return _accumulator.get(0);
}
-
- for (CpcSketch accumulatedSketch : _accumulator) {
- _union.update(accumulatedSketch);
- }
-
+ 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 01397d4ff4b..55cba5452a7 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
@@ -66,10 +66,13 @@ public abstract class CustomObjectAccumulator<T> {
}
/// Forces the item T in internal state to be merged with all pending items
in the accumulator state
- /// and returns the result. This should not result in the accumulator state
being updated or cleared.
+ /// and returns the result.
/// @return T result of the merge.
public abstract T getResult();
+ /// Merges the pending items into the internal state and clears them,
without building a result.
+ protected abstract void flush();
+
/// Merges another accumulator with this one, by extracting the result from
"other".
/// @param other the custom object accumulator to merge.
public void merge(CustomObjectAccumulator<T> other) {
@@ -96,8 +99,7 @@ public abstract class CustomObjectAccumulator<T> {
_numInputs += 1;
if (_accumulator.size() >= _threshold) {
- getResult();
- _accumulator.clear();
+ flush();
}
}
}
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 fd3cb2159f1..94dba63d18d 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
@@ -52,25 +52,13 @@ public class ThetaSketchAccumulator extends
CustomObjectAccumulator<ThetaSketch>
}
@Override
- public ThetaSketch getResult() {
- return unionAll();
- }
-
- private ThetaSketch unionAll() {
+ protected void flush() {
+ if (_accumulator == null || _accumulator.isEmpty()) {
+ return;
+ }
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) {
- return _accumulator.get(0);
- }
// 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
@@ -86,7 +74,25 @@ public class ThetaSketchAccumulator extends
CustomObjectAccumulator<ThetaSketch>
_union.union(accumulatedSketch);
}
_accumulator.clear();
+ }
+ @Override
+ public ThetaSketch getResult() {
+ 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 e4a118c5c1f..3f5a16392e6 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
@@ -60,25 +60,13 @@ public class TupleIntSketchAccumulator extends
CustomObjectAccumulator<TupleSket
}
@Override
- public TupleSketch<IntegerSummary> getResult() {
- return unionAll();
- }
-
- private TupleSketch<IntegerSummary> unionAll() {
+ protected void flush() {
+ if (_accumulator == null || _accumulator.isEmpty()) {
+ return;
+ }
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) {
- return _accumulator.get(0);
- }
// 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
@@ -94,7 +82,25 @@ public class TupleIntSketchAccumulator extends
CustomObjectAccumulator<TupleSket
_union.union(accumulatedSketch);
}
_accumulator.clear();
+ }
+ @Override
+ public TupleSketch<IntegerSummary> getResult() {
+ 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/customobject/CpcSketchAccumulatorTest.java
b/pinot-segment-local/src/test/java/org/apache/pinot/segment/local/customobject/CpcSketchAccumulatorTest.java
index a86144ed036..86326de573e 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
@@ -21,6 +21,7 @@ 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;
@@ -79,6 +80,37 @@ public class CpcSketchAccumulatorTest {
Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate(), _epsilon);
}
+ @Test
+ public void testInputsSurviveRepeatedFlushes() {
+ for (int threshold = 1; threshold <= 4; threshold++) {
+ CpcSketchAccumulator accumulator = new
CpcSketchAccumulator(_lgNominalEntries, threshold);
+ CpcUnion expected = new CpcUnion(_lgNominalEntries);
+ for (int i = 0; i < 7; i++) {
+ CpcSketch sketch = new CpcSketch(_lgNominalEntries);
+ int base = i * 1000;
+ IntStream.range(base, base + 1000).forEach(sketch::update);
+ accumulator.apply(sketch);
+ expected.update(sketch);
+ }
+ Assert.assertEquals(accumulator.getResult().getEstimate(),
expected.getResult().getEstimate(), _epsilon,
+ "threshold " + threshold);
+ }
+ }
+
+ @Test
+ public void testRepeatedGetResultIsStable() {
+ CpcSketchAccumulator accumulator = new
CpcSketchAccumulator(_lgNominalEntries, 2);
+ for (int i = 0; i < 5; i++) {
+ CpcSketch sketch = new CpcSketch(_lgNominalEntries);
+ int base = i * 1000;
+ IntStream.range(base, base + 1000).forEach(sketch::update);
+ accumulator.apply(sketch);
+ }
+ double first = accumulator.getResult().getEstimate();
+ Assert.assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
+ Assert.assertEquals(accumulator.getResult().getEstimate(), first, 0.0);
+ }
+
@Test
public void testUnionWithEmptyInput() {
CpcSketchAccumulator accumulator = new
CpcSketchAccumulator(_lgNominalEntries, 3);
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 c9e05c51d70..088d4f9ddfc 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
@@ -22,6 +22,7 @@ package org.apache.pinot.segment.local.customobject;
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;
@@ -90,6 +91,38 @@ public class ThetaSketchAccumulatorTest {
Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate());
}
+ @Test
+ public void testInputsSurviveRepeatedFlushes() {
+ for (int threshold = 1; threshold <= 4; threshold++) {
+ ThetaSketchAccumulator accumulator = new
ThetaSketchAccumulator(_setOperationBuilder, threshold);
+ ThetaUnion expected = _setOperationBuilder.buildUnion();
+ for (int i = 0; i < 7; i++) {
+ UpdatableThetaSketch input = UpdatableThetaSketch.builder().build();
+ int base = i * 1000;
+ IntStream.range(base, base + 1000).forEach(input::update);
+ ThetaSketch sketch = input.compact();
+ accumulator.apply(sketch);
+ expected.union(sketch);
+ }
+ Assert.assertEquals(accumulator.getResult().getEstimate(),
expected.getResult().getEstimate(), 0.0,
+ "threshold " + threshold);
+ }
+ }
+
+ @Test
+ public void testRepeatedGetResultIsStable() {
+ ThetaSketchAccumulator accumulator = new
ThetaSketchAccumulator(_setOperationBuilder, 2);
+ for (int i = 0; i < 5; i++) {
+ UpdatableThetaSketch input = UpdatableThetaSketch.builder().build();
+ int base = i * 1000;
+ IntStream.range(base, base + 1000).forEach(input::update);
+ 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);
+ }
+
@Test
public void testUnionWithEmptyInput() {
ThetaSketchAccumulator accumulator = new
ThetaSketchAccumulator(_setOperationBuilder, 3);
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 fd7761b6c86..ba90de81b65 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
@@ -21,6 +21,9 @@ package org.apache.pinot.segment.local.customobject;
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.IntegerSummarySetOperations;
import org.apache.datasketches.tuple.aninteger.IntegerTupleSketch;
@@ -93,6 +96,50 @@ public class TupleIntSketchAccumulatorTest {
Assert.assertEquals(accumulator.getResult().getEstimate(),
sketch1.getEstimate() + sketch2.getEstimate());
}
+ // Summaries are summed, so a sketch merged twice doubles a summary while
leaving the estimate right.
+ @Test
+ public void testInputsSurviveRepeatedFlushes() {
+ for (int threshold = 1; threshold <= 4; threshold++) {
+ TupleIntSketchAccumulator accumulator = new
TupleIntSketchAccumulator(_setOps, _nominalEntries, threshold);
+ TupleUnion<IntegerSummary> expected = new TupleUnion<>(_nominalEntries,
_setOps);
+ for (int i = 0; i < 7; i++) {
+ IntegerTupleSketch input = new IntegerTupleSketch(_lgK,
IntegerSummary.Mode.Sum);
+ int base = i * 1000;
+ IntStream.range(base, base + 1000).forEach(k -> input.update(k, 1));
+ CompactTupleSketch<IntegerSummary> sketch = input.compact();
+ accumulator.apply(sketch);
+ expected.union(sketch);
+ }
+ 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);
+ }
+ }
+
+ @Test
+ public void testRepeatedGetResultIsStable() {
+ TupleIntSketchAccumulator accumulator = new
TupleIntSketchAccumulator(_setOps, _nominalEntries, 2);
+ for (int i = 0; i < 5; i++) {
+ IntegerTupleSketch input = new IntegerTupleSketch(_lgK,
IntegerSummary.Mode.Sum);
+ int base = i * 1000;
+ IntStream.range(base, base + 1000).forEach(k -> input.update(k, 1));
+ accumulator.apply(input.compact());
+ }
+ long first = summarySum(accumulator.getResult());
+ Assert.assertEquals(summarySum(accumulator.getResult()), first);
+ Assert.assertEquals(summarySum(accumulator.getResult()), first);
+ }
+
+ private static long summarySum(TupleSketch<IntegerSummary> sketch) {
+ long sum = 0;
+ TupleSketchIterator<IntegerSummary> it = sketch.iterator();
+ while (it.next()) {
+ sum += it.getSummary().getValue();
+ }
+ return sum;
+ }
+
@Test
public void testUnionWithEmptyInput() {
TupleIntSketchAccumulator accumulator = new
TupleIntSketchAccumulator(_setOps, _nominalEntries, 3);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
