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 96fe7141147 Merge final results of array, raw sketch and other
aggregation functions (#20098)
96fe7141147 is described below
commit 96fe714114751e5655b674c9e2fb8411d35dd4c2
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Fri Oct 30 16:15:53 2026 -0700
Merge final results of array, raw sketch and other aggregation functions
(#20098)
---
.../aggregation/function/AggregationFunction.java | 11 +-
.../function/AnyValueAggregationFunction.java | 5 +
...ValueIntegerTupleSketchAggregationFunction.java | 6 +
...stinctCountRawCPCSketchAggregationFunction.java | 17 ++
.../DistinctCountRawHLLAggregationFunction.java | 15 +-
...DistinctCountRawHLLPlusAggregationFunction.java | 15 +-
...inctCountRawThetaSketchAggregationFunction.java | 17 ++
.../DistinctCountRawULLAggregationFunction.java | 15 +-
.../FrequentLongsSketchAggregationFunction.java | 10 ++
.../FrequentStringsSketchAggregationFunction.java | 10 ++
.../function/HistogramAggregationFunction.java | 9 ++
.../function/IdSetAggregationFunction.java | 9 ++
.../IntegerTupleSketchAggregationFunction.java | 11 ++
.../PercentileRawEstAggregationFunction.java | 16 +-
.../PercentileRawKLLAggregationFunction.java | 14 ++
.../PercentileRawKLLMVAggregationFunction.java | 14 ++
.../PercentileRawTDigestAggregationFunction.java | 15 +-
.../function/StUnionAggregationFunction.java | 7 +
.../function/SumPrecisionAggregationFunction.java | 24 +--
.../function/array/BaseArrayAggFunction.java | 20 +++
.../function/array/ListAggFunction.java | 15 +-
.../array/SumArrayDoubleAggregationFunction.java | 10 ++
.../array/SumArrayLongAggregationFunction.java | 10 ++
.../funnel/FunnelCountAggregationFunction.java | 5 +-
...FunnelStepDurationStatsAggregationFunction.java | 5 -
.../function/AggregationFunctionFactoryTest.java | 6 +
.../function/AnyValueAggregationFunctionTest.java | 6 +
...tinctCountCPCSketchAggregationFunctionTest.java | 65 +++++---
.../DistinctCountULLAggregationFunctionTest.java | 15 ++
.../aggregation/function/FinalResultMergeTest.java | 176 +++++++++++++++++++++
.../IntegerTupleSketchAggregationFunctionTest.java | 65 ++++++--
.../aggregation/function/ListAggFunctionTest.java | 34 ++++
.../function/StUnionAggregationFunctionTest.java | 14 ++
.../SumPrecisionAggregationFunctionTest.java | 19 +++
.../function/funnel/FunnelFinalResultTest.java | 16 ++
.../apache/pinot/queries/ArrayAggQueriesTest.java | 27 ++--
.../DistinctCountThetaSketchQueriesTest.java | 7 +
...erSegmentAggregationSingleValueQueriesTest.java | 16 ++
38 files changed, 699 insertions(+), 72 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
index 86955a830ed..32d516d2e41 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunction.java
@@ -322,9 +322,16 @@ public interface AggregationFunction<IntermediateResult,
FinalResult> {
@Nullable
FinalResult extractFinalResult(@Nullable IntermediateResult
intermediateResult);
- /// Merges two final results, neither of which is `null`. This can be used
to optimized certain functions (e.g.
+ /// Merges two final results, neither of which is `null`. This can be used
to optimize certain functions (e.g.
/// DISTINCT_COUNT) when data is partitioned on each server, where we may
directly request servers to return final
- /// result and merge them on broker. The result may be `finalResult1`
itself, modified in place.
+ /// result and merge them on broker. Final results are only merged across
partitions of the aggregated values, so no
+ /// value is aggregated into both final results.
+ ///
+ /// Final results are merged in the form they are transported in, the value
their column stores: a `STRING` final
+ /// result is a string even where [#extractFinalResult] returns an object
stored as its string, e.g. the serialized
+ /// sketch of a raw sketch function. The merged result takes the same form.
Like other column values, a final result
+ /// can be shared, e.g. when the multi-stage engine broadcasts a block to
several stages, so the merge must not modify
+ /// either final result. It may return one of them unchanged.
///
/// Call [AggregationFunctionUtils#mergeFinalResult] instead of calling this
directly, on the same terms as [#merge].
default FinalResult mergeFinalResult(FinalResult finalResult1, FinalResult
finalResult2) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
index 73f9097b291..256dc6274e8 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
@@ -173,6 +173,11 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
return left;
}
+ @Override
+ public Object mergeFinalResult(Object finalResult1, Object finalResult2) {
+ return finalResult1;
+ }
+
@Override
public void aggregate(int length, AggregationResultHolder holder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
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 f279a157aae..1d34b244d27 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
@@ -73,4 +73,10 @@ public class AvgValueIntegerTupleSketchAggregationFunction
double estimate = retainedTotal / result.getRetainedEntries();
return Math.round(estimate);
}
+
+ /// Averages cannot be merged.
+ @Override
+ public Comparable mergeFinalResult(Comparable finalResult1, Comparable
finalResult2) {
+ throw new UnsupportedOperationException("Cannot merge final results for
function: " + getType());
+ }
}
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 bf3e8c94725..819bbc49987 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
@@ -18,8 +18,11 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.util.Base64;
import java.util.List;
import javax.annotation.Nullable;
+import org.apache.datasketches.cpc.CpcSketch;
+import org.apache.datasketches.cpc.CpcUnion;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.segment.local.customobject.CpcSketchAccumulator;
@@ -56,4 +59,18 @@ public class DistinctCountRawCPCSketchAggregationFunction
extends DistinctCountC
intermediateResult.setThreshold(_accumulatorThreshold);
return new SerializedCPCSketch(intermediateResult.getResult());
}
+
+ /// Merges two final results, which are serialized sketches in string form,
into the string form of the merged
+ /// sketch.
+ @Override
+ public String mergeFinalResult(Comparable finalResult1, Comparable
finalResult2) {
+ CpcUnion union = new CpcUnion(_lgNominalEntries);
+ union.update(deserializeFinalResult(finalResult1));
+ union.update(deserializeFinalResult(finalResult2));
+ return new SerializedCPCSketch(union.getResult()).toString();
+ }
+
+ private static CpcSketch deserializeFinalResult(Comparable finalResult) {
+ return CpcSketch.heapify(Base64.getDecoder().decode((String) finalResult));
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawHLLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawHLLAggregationFunction.java
index 330dbb966c6..20ed32efcef 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawHLLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawHLLAggregationFunction.java
@@ -26,14 +26,16 @@ import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.segment.local.customobject.SerializedHLL;
import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.utils.BytesUtils;
public class DistinctCountRawHLLAggregationFunction
- extends BaseSingleInputAggregationFunction<HyperLogLog, SerializedHLL> {
+ extends BaseSingleInputAggregationFunction<HyperLogLog, Comparable<?>> {
private final DistinctCountHLLAggregationFunction
_distinctCountHLLAggregationFunction;
public DistinctCountRawHLLAggregationFunction(List<ExpressionContext>
arguments, boolean nullHandlingEnabled) {
@@ -128,4 +130,15 @@ public class DistinctCountRawHLLAggregationFunction
}
return new SerializedHLL(intermediateResult);
}
+
+ /// Merges two final results, which are serialized HLLs in string form, into
the string form of the merged HLL.
+ @Override
+ public String mergeFinalResult(Comparable<?> finalResult1, Comparable<?>
finalResult2) {
+ HyperLogLog merged = merge(deserializeFinalResult(finalResult1),
deserializeFinalResult(finalResult2));
+ return new SerializedHLL(merged).toString();
+ }
+
+ private static HyperLogLog deserializeFinalResult(Comparable<?> finalResult)
{
+ return
ObjectSerDeUtils.HYPER_LOG_LOG_SER_DE.deserialize(BytesUtils.toBytes((String)
finalResult));
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawHLLPlusAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawHLLPlusAggregationFunction.java
index 75ebb62c5dc..17d9b012cd6 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawHLLPlusAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawHLLPlusAggregationFunction.java
@@ -26,14 +26,16 @@ import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.segment.local.customobject.SerializedHLLPlus;
import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.utils.BytesUtils;
public class DistinctCountRawHLLPlusAggregationFunction
- extends BaseSingleInputAggregationFunction<HyperLogLogPlus,
SerializedHLLPlus> {
+ extends BaseSingleInputAggregationFunction<HyperLogLogPlus, Comparable<?>>
{
private final DistinctCountHLLPlusAggregationFunction
_distinctCountHLLPlusAggregationFunction;
public DistinctCountRawHLLPlusAggregationFunction(List<ExpressionContext>
arguments, boolean nullHandlingEnabled) {
@@ -129,4 +131,15 @@ public class DistinctCountRawHLLPlusAggregationFunction
}
return new SerializedHLLPlus(intermediateResult);
}
+
+ /// Merges two final results, which are serialized HLL+s in string form,
into the string form of the merged HLL+.
+ @Override
+ public String mergeFinalResult(Comparable<?> finalResult1, Comparable<?>
finalResult2) {
+ HyperLogLogPlus merged = merge(deserializeFinalResult(finalResult1),
deserializeFinalResult(finalResult2));
+ return new SerializedHLLPlus(merged).toString();
+ }
+
+ private static HyperLogLogPlus deserializeFinalResult(Comparable<?>
finalResult) {
+ return
ObjectSerDeUtils.HYPER_LOG_LOG_PLUS_SER_DE.deserialize(BytesUtils.toBytes((String)
finalResult));
+ }
}
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 20c93225836..f12861f0317 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
@@ -18,11 +18,13 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.lang.foreign.MemorySegment;
import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import javax.annotation.Nullable;
import org.apache.datasketches.theta.ThetaSketch;
+import org.apache.datasketches.theta.ThetaUnion;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.segment.local.customobject.ThetaSketchAccumulator;
@@ -67,4 +69,19 @@ public class DistinctCountRawThetaSketchAggregationFunction
extends DistinctCoun
ThetaSketch sketch = evaluatePostAggregationExpression(mergedSketches);
return Base64.getEncoder().encodeToString(sketch.toByteArray());
}
+
+ /// Merges two final results, which are serialized sketches in string form,
into the string form of their union.
+ /// Final results are only merged across partitions of the aggregated
values, so the union of the sketches that the
+ /// post-aggregation expression returns for each partition is the sketch it
returns for all of them.
+ @Override
+ public String mergeFinalResult(Comparable finalResult1, Comparable
finalResult2) {
+ ThetaUnion union = _setOperationBuilder.buildUnion();
+ union.union(deserializeFinalResult(finalResult1));
+ union.union(deserializeFinalResult(finalResult2));
+ return Base64.getEncoder().encodeToString(union.getResult().toByteArray());
+ }
+
+ private static ThetaSketch deserializeFinalResult(Comparable finalResult) {
+ return
ThetaSketch.wrap(MemorySegment.ofArray(Base64.getDecoder().decode((String)
finalResult)));
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawULLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawULLAggregationFunction.java
index e2104d6fdae..e2f5b14ccee 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawULLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountRawULLAggregationFunction.java
@@ -19,6 +19,7 @@
package org.apache.pinot.core.query.aggregation.function;
import com.dynatrace.hash4j.distinctcount.UltraLogLog;
+import java.util.Base64;
import java.util.List;
import javax.annotation.Nullable;
import org.apache.pinot.common.request.context.ExpressionContext;
@@ -41,12 +42,24 @@ public class DistinctCountRawULLAggregationFunction extends
DistinctCountULLAggr
return new SerializedULL(intermediateResult);
}
+ @Override
public AggregationFunctionType getType() {
- return AggregationFunctionType.DISTINCTCOUNTULL;
+ return AggregationFunctionType.DISTINCTCOUNTRAWULL;
}
@Override
public ColumnDataType getFinalResultColumnType() {
return ColumnDataType.STRING;
}
+
+ /// Merges two final results, which are serialized ULLs in string form, into
the string form of the merged ULL.
+ @Override
+ public String mergeFinalResult(Comparable finalResult1, Comparable
finalResult2) {
+ UltraLogLog merged = merge(deserializeFinalResult(finalResult1),
deserializeFinalResult(finalResult2));
+ return new SerializedULL(merged).toString();
+ }
+
+ private static UltraLogLog deserializeFinalResult(Comparable finalResult) {
+ return UltraLogLog.wrap(Base64.getDecoder().decode((String) finalResult));
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentLongsSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentLongsSketchAggregationFunction.java
index 15cee829ebf..48253a23b74 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentLongsSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentLongsSketchAggregationFunction.java
@@ -21,6 +21,7 @@ package org.apache.pinot.core.query.aggregation.function;
import com.google.common.base.Preconditions;
import java.lang.foreign.MemorySegment;
+import java.util.Base64;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
@@ -303,4 +304,13 @@ public class FrequentLongsSketchAggregationFunction
// no disabled-mode identity to preserve here.
return sketch != null ? new SerializedFrequentLongsSketch(sketch) : null;
}
+
+ /// Merges two final results, which are serialized sketches in string form,
into the string form of the merged
+ /// sketch.
+ @Override
+ public String mergeFinalResult(Comparable<?> finalResult1, Comparable<?>
finalResult2) {
+ FrequentLongsSketch sketch1 =
deserializeSketch(Base64.getDecoder().decode((String) finalResult1));
+ FrequentLongsSketch sketch2 =
deserializeSketch(Base64.getDecoder().decode((String) finalResult2));
+ return new SerializedFrequentLongsSketch(merge(sketch1,
sketch2)).toString();
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentStringsSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentStringsSketchAggregationFunction.java
index 27fa032fe98..af0ad716a79 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentStringsSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FrequentStringsSketchAggregationFunction.java
@@ -21,6 +21,7 @@ package org.apache.pinot.core.query.aggregation.function;
import com.google.common.base.Preconditions;
import java.lang.foreign.MemorySegment;
+import java.util.Base64;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
@@ -296,4 +297,13 @@ public class FrequentStringsSketchAggregationFunction
// no disabled-mode identity to preserve here.
return sketch != null ? new SerializedFrequentStringsSketch(sketch) : null;
}
+
+ /// Merges two final results, which are serialized sketches in string form,
into the string form of the merged
+ /// sketch.
+ @Override
+ public String mergeFinalResult(Comparable<?> finalResult1, Comparable<?>
finalResult2) {
+ FrequentItemsSketch<String> sketch1 =
deserializeSketch(Base64.getDecoder().decode((String) finalResult1));
+ FrequentItemsSketch<String> sketch2 =
deserializeSketch(Base64.getDecoder().decode((String) finalResult2));
+ return new SerializedFrequentStringsSketch(merge(sketch1,
sketch2)).toString();
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
index 64c48f522bd..2e39377a92a 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/HistogramAggregationFunction.java
@@ -258,6 +258,15 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
return ArrayListUtils.toDoubleArray(doubleArrayList);
}
+ @Override
+ public double[] mergeFinalResult(double[] finalResult1, double[]
finalResult2) {
+ double[] mergedResult = finalResult1.clone();
+ for (int i = 0; i < mergedResult.length; i++) {
+ mergedResult[i] += finalResult2[i];
+ }
+ return mergedResult;
+ }
+
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/IdSetAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/IdSetAggregationFunction.java
index 7c1ef748058..0c616d2a2d2 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/IdSetAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/IdSetAggregationFunction.java
@@ -658,6 +658,15 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
}
}
+ @Override
+ public String mergeFinalResult(String finalResult1, String finalResult2) {
+ try {
+ return merge(IdSets.fromBase64String(finalResult1),
IdSets.fromBase64String(finalResult2)).toBase64String();
+ } catch (IOException e) {
+ throw new RuntimeException("Caught exception while merging IdSets", e);
+ }
+ }
+
/// Returns the IdSet from the result holder or creates a new one if it does
not exist.
private IdSet getIdSet(AggregationResultHolder aggregationResultHolder,
DataType valueType) {
IdSet idSet = aggregationResultHolder.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 c7366e8227b..c53f19076f0 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
@@ -26,6 +26,7 @@ import java.util.Map;
import javax.annotation.Nullable;
import org.apache.commons.lang3.StringUtils;
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.IntegerSummaryDeserializer;
import org.apache.datasketches.tuple.aninteger.IntegerSummarySetOperations;
@@ -279,6 +280,16 @@ public class IntegerTupleSketchAggregationFunction
return
Base64.getEncoder().encodeToString(accumulator.getResult().toByteArray());
}
+ /// Merges two final results, which are serialized sketches in string form,
into the string form of the merged
+ /// sketch.
+ @Override
+ public Comparable mergeFinalResult(Comparable finalResult1, Comparable
finalResult2) {
+ TupleUnion<IntegerSummary> union = new TupleUnion<>(_nominalEntries,
_setOps);
+ union.union(deserializeSketch(Base64.getDecoder().decode((String)
finalResult1)));
+ union.union(deserializeSketch(Base64.getDecoder().decode((String)
finalResult2)));
+ return Base64.getEncoder().encodeToString(union.getResult().toByteArray());
+ }
+
@Override
public boolean canUseStarTree(Map<String, Object> functionParameters) {
Object nominalEntriesParam =
functionParameters.get(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawEstAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawEstAggregationFunction.java
index 263b8d90ff4..b4842c8d61b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawEstAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawEstAggregationFunction.java
@@ -24,17 +24,19 @@ import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.segment.local.customobject.QuantileDigest;
import org.apache.pinot.segment.local.customobject.SerializedQuantileDigest;
import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.utils.BytesUtils;
/// The `PercentileRawEstAggregationFunction` returns the serialized
`QuantileDigest` data structure of the
/// `PercentileEstAggregationFunction`.
public class PercentileRawEstAggregationFunction
- extends BaseSingleInputAggregationFunction<QuantileDigest,
SerializedQuantileDigest> {
+ extends BaseSingleInputAggregationFunction<QuantileDigest, Comparable<?>> {
private final PercentileEstAggregationFunction
_percentileEstAggregationFunction;
public PercentileRawEstAggregationFunction(ExpressionContext
expressionContext, double percentile,
@@ -146,4 +148,16 @@ public class PercentileRawEstAggregationFunction
}
return new SerializedQuantileDigest(intermediateResult,
_percentileEstAggregationFunction._percentile);
}
+
+ /// Merges two final results, which are serialized digests in string form,
into the string form of the merged
+ /// digest.
+ @Override
+ public String mergeFinalResult(Comparable<?> finalResult1, Comparable<?>
finalResult2) {
+ QuantileDigest merged = merge(deserializeFinalResult(finalResult1),
deserializeFinalResult(finalResult2));
+ return new SerializedQuantileDigest(merged,
_percentileEstAggregationFunction._percentile).toString();
+ }
+
+ private static QuantileDigest deserializeFinalResult(Comparable<?>
finalResult) {
+ return
ObjectSerDeUtils.QUANTILE_DIGEST_SER_DE.deserialize(BytesUtils.toBytes((String)
finalResult));
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawKLLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawKLLAggregationFunction.java
index e4e2df10bfb..ab3af559594 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawKLLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawKLLAggregationFunction.java
@@ -18,11 +18,13 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.util.Base64;
import java.util.List;
import javax.annotation.Nullable;
import org.apache.datasketches.kll.KllDoublesSketch;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.segment.local.customobject.SerializedKLL;
import org.apache.pinot.segment.spi.AggregationFunctionType;
@@ -59,4 +61,16 @@ public class PercentileRawKLLAggregationFunction extends
PercentileKLLAggregatio
}
return new SerializedKLL(sketch, _percentile);
}
+
+ /// Merges two final results, which are serialized sketches in string form,
into the string form of the merged
+ /// sketch.
+ @Override
+ public String mergeFinalResult(Comparable<?> finalResult1, Comparable<?>
finalResult2) {
+ KllDoublesSketch merged = merge(deserializeFinalResult(finalResult1),
deserializeFinalResult(finalResult2));
+ return new SerializedKLL(merged, _percentile).toString();
+ }
+
+ private static KllDoublesSketch deserializeFinalResult(Comparable<?>
finalResult) {
+ return
ObjectSerDeUtils.KLL_SKETCH_SER_DE.deserialize(Base64.getDecoder().decode((String)
finalResult));
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawKLLMVAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawKLLMVAggregationFunction.java
index 7979374c533..e40857dc193 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawKLLMVAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawKLLMVAggregationFunction.java
@@ -18,11 +18,13 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.util.Base64;
import java.util.List;
import javax.annotation.Nullable;
import org.apache.datasketches.kll.KllDoublesSketch;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.segment.local.customobject.SerializedKLL;
import org.apache.pinot.segment.spi.AggregationFunctionType;
@@ -59,4 +61,16 @@ public class PercentileRawKLLMVAggregationFunction extends
PercentileKLLMVAggreg
}
return new SerializedKLL(sketch, _percentile);
}
+
+ /// Merges two final results, which are serialized sketches in string form,
into the string form of the merged
+ /// sketch.
+ @Override
+ public String mergeFinalResult(Comparable<?> finalResult1, Comparable<?>
finalResult2) {
+ KllDoublesSketch merged = merge(deserializeFinalResult(finalResult1),
deserializeFinalResult(finalResult2));
+ return new SerializedKLL(merged, _percentile).toString();
+ }
+
+ private static KllDoublesSketch deserializeFinalResult(Comparable<?>
finalResult) {
+ return
ObjectSerDeUtils.KLL_SKETCH_SER_DE.deserialize(Base64.getDecoder().decode((String)
finalResult));
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawTDigestAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawTDigestAggregationFunction.java
index 5e4240f33a3..54fdff1339f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawTDigestAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileRawTDigestAggregationFunction.java
@@ -31,12 +31,13 @@ import
org.apache.pinot.segment.local.customobject.tdigest.SerializedTDigest;
import org.apache.pinot.segment.local.customobject.tdigest.TDigest;
import org.apache.pinot.segment.local.customobject.tdigest.TDigestAccumulator;
import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.utils.BytesUtils;
/// The `PercentileRawTDigestAggregationFunction` returns the serialized
`TDigest` data structure of the
/// `PercentileEstAggregationFunction`.
public class PercentileRawTDigestAggregationFunction
- extends BaseSingleInputAggregationFunction<TDigest, SerializedTDigest> {
+ extends BaseSingleInputAggregationFunction<TDigest, Comparable<?>> {
private final PercentileTDigestAggregationFunction
_percentileTDigestAggregationFunction;
public PercentileRawTDigestAggregationFunction(ExpressionContext
expressionContext, int percentile,
@@ -162,4 +163,16 @@ public class PercentileRawTDigestAggregationFunction
}
return new SerializedTDigest(intermediateResult,
_percentileTDigestAggregationFunction._percentile);
}
+
+ /// Merges two final results, which are serialized digests in string form,
into the string form of the merged
+ /// digest.
+ @Override
+ public String mergeFinalResult(Comparable<?> finalResult1, Comparable<?>
finalResult2) {
+ TDigest merged = merge(deserializeFinalResult(finalResult1),
deserializeFinalResult(finalResult2));
+ return new SerializedTDigest(merged,
_percentileTDigestAggregationFunction._percentile).toString();
+ }
+
+ private static TDigest deserializeFinalResult(Comparable<?> finalResult) {
+ return
ObjectSerDeUtils.TDIGEST_SER_DE.deserialize(BytesUtils.toBytes((String)
finalResult));
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/StUnionAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/StUnionAggregationFunction.java
index d46de5cffbe..3113d178c9b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/StUnionAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/StUnionAggregationFunction.java
@@ -233,6 +233,13 @@ public class StUnionAggregationFunction extends
BaseSingleInputAggregationFuncti
return new ByteArray(GeometrySerializer.serialize(geometry));
}
+ @Override
+ public ByteArray mergeFinalResult(ByteArray finalResult1, ByteArray
finalResult2) {
+ Geometry geometry1 =
GeometrySerializer.deserialize(finalResult1.getBytes());
+ Geometry geometry2 =
GeometrySerializer.deserialize(finalResult2.getBytes());
+ return new ByteArray(GeometrySerializer.serialize(merge(geometry1,
geometry2)));
+ }
+
/// Returns the union of the supplied geometries.
///
/// When either operand is a `GeometryCollection`,
[Geometry#union(Geometry)] can produce invalid
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunction.java
index 64f2064b1eb..4306f032107 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunction.java
@@ -370,19 +370,25 @@ public class SumPrecisionAggregationFunction extends
BaseSingleInputAggregationF
@Nullable
@Override
public BigDecimal extractFinalResult(@Nullable BigDecimal
intermediateResult) {
- if (intermediateResult == null) {
- return null;
- }
- if (_precision == null) {
- return intermediateResult;
- }
- BigDecimal result = intermediateResult.round(new MathContext(_precision,
RoundingMode.HALF_EVEN));
- return _scale == null ? result : result.setScale(_scale,
RoundingMode.HALF_EVEN);
+ return intermediateResult != null ? round(intermediateResult) : null;
}
+ /// Merges two final results and rounds the sum to the precision and scale
of the function. The final results are
+ /// already rounded, so the merged result can differ from the rounded exact
sum: with precision 1 and scale 0,
+ /// partition sums `0.5` and `0.5` have final results `0` and `0`, which
merge to `0`, while their exact sum `1.0`
+ /// rounds to `1`.
@Override
public BigDecimal mergeFinalResult(BigDecimal finalResult1, BigDecimal
finalResult2) {
- return merge(finalResult1, finalResult2);
+ return round(merge(finalResult1, finalResult2));
+ }
+
+ /// Rounds the sum to the precision and scale of the function, if specified.
+ private BigDecimal round(BigDecimal sum) {
+ if (_precision == null) {
+ return sum;
+ }
+ BigDecimal result = sum.round(new MathContext(_precision,
RoundingMode.HALF_EVEN));
+ return _scale == null ? result : result.setScale(_scale,
RoundingMode.HALF_EVEN);
}
public BigDecimal getDefaultResult(GroupByResultHolder groupByResultHolder,
int groupKey) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/BaseArrayAggFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/BaseArrayAggFunction.java
index 7abc7c9f0da..ad4cde3b825 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/BaseArrayAggFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/BaseArrayAggFunction.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.query.aggregation.function.array;
+import java.lang.reflect.Array;
import java.nio.ByteBuffer;
import java.util.Map;
import javax.annotation.Nullable;
@@ -100,4 +101,23 @@ public abstract class BaseArrayAggFunction<I, F> extends
BaseSingleInputAggregat
public I extractGroupByResult(GroupByResultHolder groupByResultHolder, int
groupKey) {
return groupByResultHolder.getResult(groupKey);
}
+
+ /// Merges two final results by concatenating them. Final results are only
merged across partitions of the
+ /// aggregated values, so the values of the distinct variants do not overlap.
+ @Override
+ public F mergeFinalResult(F finalResult1, F finalResult2) {
+ int length1 = Array.getLength(finalResult1);
+ if (length1 == 0) {
+ return finalResult2;
+ }
+ int length2 = Array.getLength(finalResult2);
+ if (length2 == 0) {
+ return finalResult1;
+ }
+ //noinspection unchecked
+ F mergedResult = (F)
Array.newInstance(finalResult1.getClass().getComponentType(), length1 +
length2);
+ System.arraycopy(finalResult1, 0, mergedResult, 0, length1);
+ System.arraycopy(finalResult2, 0, mergedResult, length1, length2);
+ return mergedResult;
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/ListAggFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/ListAggFunction.java
index 78c02b54d22..71c371ba8af 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/ListAggFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/ListAggFunction.java
@@ -23,6 +23,7 @@ import it.unimi.dsi.fastutil.objects.ObjectCollection;
import java.util.Arrays;
import java.util.Map;
import javax.annotation.Nullable;
+import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
@@ -198,12 +199,18 @@ public class ListAggFunction extends
BaseSingleInputAggregationFunction<ObjectCo
return ColumnDataType.STRING;
}
+ /// Joins the values with the separator. Returns `null` when there are no
values, which keeps them apart from a
+ /// single empty-string value.
@Nullable
@Override
public String extractFinalResult(@Nullable ObjectCollection<String> strings)
{
- if (strings == null) {
- return null;
- }
- return StringUtils.join(strings, _separator);
+ return CollectionUtils.isNotEmpty(strings) ? StringUtils.join(strings,
_separator) : null;
+ }
+
+ /// Merges two final results by joining them with the separator. Final
results are only merged across partitions of
+ /// the aggregated values, so the values of the distinct variant do not
overlap.
+ @Override
+ public String mergeFinalResult(String finalResult1, String finalResult2) {
+ return finalResult1 + _separator + finalResult2;
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayDoubleAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayDoubleAggregationFunction.java
index dfa9693adaf..3f19d975dbe 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayDoubleAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayDoubleAggregationFunction.java
@@ -19,6 +19,7 @@
package org.apache.pinot.core.query.aggregation.function.array;
import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
@@ -177,4 +178,13 @@ public class SumArrayDoubleAggregationFunction
public double[] extractFinalResult(@Nullable DoubleArrayList result) {
return result != null ? ArrayListUtils.toDoubleArray(result) : null;
}
+
+ @Override
+ public double[] mergeFinalResult(double[] finalResult1, double[]
finalResult2) {
+ double[] mergedResult = Arrays.copyOf(finalResult1,
Math.max(finalResult1.length, finalResult2.length));
+ for (int i = 0; i < finalResult2.length; i++) {
+ mergedResult[i] += finalResult2[i];
+ }
+ return mergedResult;
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayLongAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayLongAggregationFunction.java
index 11f52798e1e..99c0ef57fea 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayLongAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/array/SumArrayLongAggregationFunction.java
@@ -19,6 +19,7 @@
package org.apache.pinot.core.query.aggregation.function.array;
import it.unimi.dsi.fastutil.longs.LongArrayList;
+import java.util.Arrays;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
@@ -177,4 +178,13 @@ public class SumArrayLongAggregationFunction
public long[] extractFinalResult(@Nullable LongArrayList result) {
return result != null ? ArrayListUtils.toLongArray(result) : null;
}
+
+ @Override
+ public long[] mergeFinalResult(long[] finalResult1, long[] finalResult2) {
+ long[] mergedResult = Arrays.copyOf(finalResult1,
Math.max(finalResult1.length, finalResult2.length));
+ for (int i = 0; i < finalResult2.length; i++) {
+ mergedResult[i] += finalResult2[i];
+ }
+ return mergedResult;
+ }
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountAggregationFunction.java
index f405a75114f..bebe3eb2bd9 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelCountAggregationFunction.java
@@ -182,10 +182,11 @@ public class FunnelCountAggregationFunction<A, I>
implements AggregationFunction
@Override
public long[] mergeFinalResult(long[] finalResult1, long[] finalResult2) {
+ long[] mergedResult = finalResult1.clone();
for (int i = 0; i < _numSteps; i++) {
- finalResult1[i] += finalResult2[i];
+ mergedResult[i] += finalResult2[i];
}
- return finalResult1;
+ return mergedResult;
}
@Override
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelStepDurationStatsAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelStepDurationStatsAggregationFunction.java
index b3518a92905..2cdd28c403b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelStepDurationStatsAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/funnel/window/FunnelStepDurationStatsAggregationFunction.java
@@ -257,9 +257,4 @@ public class FunnelStepDurationStatsAggregationFunction
extends FunnelBaseAggreg
}
return maxStep;
}
-
- @Override
- public double[] mergeFinalResult(double[] finalResult1, double[]
finalResult2) {
- return finalResult1;
- }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactoryTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactoryTest.java
index 0d44e9bcf00..d10465d426c 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactoryTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactoryTest.java
@@ -202,6 +202,12 @@ public class AggregationFunctionFactoryTest {
assertEquals(aggregationFunction.getType(),
AggregationFunctionType.DISTINCTCOUNTRAWHLLPLUS);
assertEquals(aggregationFunction.getResultColumnName(),
function.toString());
+ function = getFunction("DiStInCtCoUnTrAwUlL");
+ aggregationFunction =
AggregationFunctionFactory.getAggregationFunction(function, false);
+ assertTrue(aggregationFunction instanceof
DistinctCountRawULLAggregationFunction);
+ assertEquals(aggregationFunction.getType(),
AggregationFunctionType.DISTINCTCOUNTRAWULL);
+ assertEquals(aggregationFunction.getResultColumnName(),
function.toString());
+
function = getFunction("FaStHlL");
aggregationFunction =
AggregationFunctionFactory.getAggregationFunction(function, false);
assertTrue(aggregationFunction instanceof FastHLLAggregationFunction);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
index 77d27f06195..516c0c74358 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
@@ -511,6 +511,12 @@ public class AnyValueAggregationFunctionTest extends
AbstractAggregationFunction
serialize(bigDecimalFunction, value)), new BigDecimal("1")), value);
}
+ @Test
+ void testMergeFinalResult() {
+ AnyValueAggregationFunction function = createFunction("LONG");
+ assertEquals(function.mergeFinalResult(1L, 2L), 1L);
+ }
+
private static AnyValueAggregationFunction createFunction(String dataType) {
return new AnyValueAggregationFunction(List.of(EXPRESSION,
ExpressionContext.forLiteral(DataType.STRING, dataType)),
true);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunctionTest.java
index 051a7f86f0c..fb5441805dc 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountCPCSketchAggregationFunctionTest.java
@@ -30,9 +30,10 @@ import
org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.segment.local.customobject.CpcSketchAccumulator;
import org.apache.pinot.segment.local.customobject.SerializedCPCSketch;
import org.apache.pinot.segment.spi.Constants;
-import org.testng.Assert;
import org.testng.annotations.Test;
+import static org.testng.Assert.*;
+
public class DistinctCountCPCSketchAggregationFunctionTest {
@@ -41,18 +42,18 @@ public class DistinctCountCPCSketchAggregationFunctionTest {
DistinctCountCPCSketchAggregationFunction function =
new
DistinctCountCPCSketchAggregationFunction(List.of(ExpressionContext.forIdentifier("col")),
false);
- Assert.assertTrue(function.canUseStarTree(Map.of()));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
"12")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
12)));
-
Assert.assertFalse(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
11)));
+ assertTrue(function.canUseStarTree(Map.of()));
+ assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
"12")));
+ assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
12)));
+ assertFalse(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
11)));
function = new DistinctCountCPCSketchAggregationFunction(
List.of(ExpressionContext.forIdentifier("col"),
ExpressionContext.forLiteral(Literal.intValue(12))), false);
- Assert.assertTrue(function.canUseStarTree(Map.of()));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
"12")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
12)));
-
Assert.assertFalse(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
11)));
+ assertTrue(function.canUseStarTree(Map.of()));
+ assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
"12")));
+ assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
12)));
+ assertFalse(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
11)));
}
@Test
@@ -62,10 +63,10 @@ public class DistinctCountCPCSketchAggregationFunctionTest {
ExpressionContext.forLiteral(Literal.stringValue("nominalEntries=8192"))),
false);
// Default lgK = 12 / K=4096
- Assert.assertFalse(function.canUseStarTree(Map.of()));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
"14")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
13)));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
"13")));
+ assertFalse(function.canUseStarTree(Map.of()));
+ assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
"14")));
+ assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
13)));
+ assertTrue(function.canUseStarTree(Map.of(Constants.CPCSKETCH_LGK_KEY,
"13")));
}
/// Tests the empty result path used by BROKER_EVALUATE when all segments
are pruned.
@@ -78,15 +79,15 @@ public class DistinctCountCPCSketchAggregationFunctionTest {
AggregationResultHolder holder = function.createAggregationResultHolder();
CpcSketchAccumulator accumulator =
function.extractAggregationResult(holder);
- Assert.assertTrue(accumulator.isEmpty());
+ assertTrue(accumulator.isEmpty());
// extractFinalResult should return 0 for an empty accumulator
Comparable result = function.extractFinalResult(accumulator);
- Assert.assertEquals(result, 0L);
+ assertEquals(result, 0L);
// The result must be convertible via the declared column type
Object converted = function.getFinalResultColumnType().convert(result);
- Assert.assertEquals(converted, 0L);
+ assertEquals(converted, 0L);
}
@Test
@@ -96,22 +97,22 @@ public class DistinctCountCPCSketchAggregationFunctionTest {
AggregationResultHolder holder = function.createAggregationResultHolder();
CpcSketchAccumulator accumulator =
function.extractAggregationResult(holder);
- Assert.assertTrue(accumulator.isEmpty());
+ assertTrue(accumulator.isEmpty());
// extractFinalResult should return a SerializedCPCSketch
SerializedCPCSketch rawResult = function.extractFinalResult(accumulator);
- Assert.assertNotNull(rawResult);
+ assertNotNull(rawResult);
// The declared column type is STRING; convert() must produce a String
- Assert.assertEquals(function.getFinalResultColumnType(),
ColumnDataType.STRING);
+ assertEquals(function.getFinalResultColumnType(), ColumnDataType.STRING);
Object converted = function.getFinalResultColumnType().convert(rawResult);
- Assert.assertTrue(converted instanceof String);
+ assertTrue(converted instanceof String);
// The string should be a valid Base64-encoded CPC sketch that round-trips
String base64 = (String) converted;
byte[] bytes = Base64.getDecoder().decode(base64);
CpcSketch deserialized = CpcSketch.heapify(MemorySegment.ofArray(bytes));
- Assert.assertEquals(deserialized.getEstimate(), 0.0);
+ assertEquals(deserialized.getEstimate(), 0.0);
}
@Test
@@ -128,14 +129,30 @@ public class
DistinctCountCPCSketchAggregationFunctionTest {
// merge(empty, non-empty) should return non-empty
CpcSketchAccumulator result = function.merge(empty1, nonEmpty);
- Assert.assertSame(result, nonEmpty);
+ assertSame(result, nonEmpty);
// merge(non-empty, empty) should return non-empty
result = function.merge(nonEmpty, empty2);
- Assert.assertSame(result, nonEmpty);
+ assertSame(result, nonEmpty);
// merge(empty, empty) should return empty
result = function.merge(new CpcSketchAccumulator(12, 2), new
CpcSketchAccumulator(12, 2));
- Assert.assertTrue(result.isEmpty());
+ assertTrue(result.isEmpty());
+ }
+
+ @Test
+ public void testMergeRawFinalResults() {
+ DistinctCountRawCPCSketchAggregationFunction function =
+ new
DistinctCountRawCPCSketchAggregationFunction(List.of(ExpressionContext.forIdentifier("col")),
false);
+ CpcSketch sketch1 = new CpcSketch(12);
+ sketch1.update("a");
+ sketch1.update("b");
+ CpcSketch sketch2 = new CpcSketch(12);
+ sketch2.update("c");
+
+ String merged = function.mergeFinalResult(new
SerializedCPCSketch(sketch1).toString(),
+ new SerializedCPCSketch(sketch2).toString());
+ CpcSketch mergedSketch =
CpcSketch.heapify(MemorySegment.ofArray(Base64.getDecoder().decode(merged)));
+ assertEquals(Math.round(mergedSketch.getEstimate()), 3L);
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunctionTest.java
index 0abed9d5690..2a31b0a3aa7 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunctionTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import com.dynatrace.hash4j.distinctcount.UltraLogLog;
import java.util.List;
import java.util.Map;
import org.apache.pinot.common.request.Literal;
@@ -27,6 +28,7 @@ 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.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
+import org.apache.pinot.segment.local.customobject.SerializedULL;
import org.apache.pinot.segment.spi.Constants;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.data.FieldSpec.DataType;
@@ -90,6 +92,19 @@ public class DistinctCountULLAggregationFunctionTest {
return Map.of(COLUMN, SyntheticBlockValSets.LongMV.create(null, MV_ROWS));
}
+ @Test
+ public void testMergeRawFinalResults() {
+ DistinctCountRawULLAggregationFunction function =
+ new
DistinctCountRawULLAggregationFunction(List.of(ExpressionContext.forIdentifier("col")),
false);
+ UltraLogLog ull1 =
UltraLogLog.create(12).add(0x0123456789ABCDEFL).add(0x1111111111111111L);
+ UltraLogLog ull2 = UltraLogLog.create(12).add(0x7777777777777777L);
+ UltraLogLog expected =
+
UltraLogLog.create(12).add(0x0123456789ABCDEFL).add(0x1111111111111111L).add(0x7777777777777777L);
+
+ String merged = function.mergeFinalResult(new
SerializedULL(ull1).toString(), new SerializedULL(ull2).toString());
+ assertEquals(merged, new SerializedULL(expected).toString());
+ }
+
/// Aggregating a multi-value column gives exactly what aggregating its
flattened values as a single-value column
/// gives. Comparing the two rather than asserting an estimate keeps this
exact for a sketch.
@Test
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/FinalResultMergeTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/FinalResultMergeTest.java
new file mode 100644
index 00000000000..7e226a85d46
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/FinalResultMergeTest.java
@@ -0,0 +1,176 @@
+/**
+ * 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.io.IOException;
+import java.lang.foreign.MemorySegment;
+import java.math.BigDecimal;
+import java.util.Base64;
+import java.util.List;
+import org.apache.datasketches.common.ArrayOfStringsSerDe;
+import org.apache.datasketches.frequencies.FrequentItemsSketch;
+import org.apache.datasketches.frequencies.FrequentLongsSketch;
+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.query.aggregation.function.array.ArrayAggBigDecimalFunction;
+import
org.apache.pinot.core.query.aggregation.function.array.ArrayAggDistinctIntFunction;
+import
org.apache.pinot.core.query.aggregation.function.array.ArrayAggIntFunction;
+import
org.apache.pinot.core.query.aggregation.function.array.SumArrayDoubleAggregationFunction;
+import
org.apache.pinot.core.query.aggregation.function.array.SumArrayLongAggregationFunction;
+import org.apache.pinot.core.query.utils.idset.IdSet;
+import org.apache.pinot.core.query.utils.idset.IdSets;
+import
org.apache.pinot.segment.local.customobject.SerializedFrequentLongsSketch;
+import
org.apache.pinot.segment.local.customobject.SerializedFrequentStringsSketch;
+import org.apache.pinot.segment.local.customobject.SerializedKLL;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Merges the final results that servers or leaf stage workers return for
partitions of the aggregated values, in the
+/// form they are transported in.
+public class FinalResultMergeTest {
+ private static final ExpressionContext COLUMN =
ExpressionContext.forIdentifier("column");
+
+ @Test
+ public void testHistogram() {
+ HistogramAggregationFunction function = new
HistogramAggregationFunction(List.of(COLUMN,
+ ExpressionContext.forLiteral(Literal.doubleValue(0)),
ExpressionContext.forLiteral(Literal.doubleValue(10)),
+ ExpressionContext.forLiteral(Literal.intValue(2))), false);
+ double[] finalResult1 = {1, 2};
+ double[] finalResult2 = {3, 4};
+ assertEquals(function.mergeFinalResult(finalResult1, finalResult2), new
double[]{4, 6});
+ assertEquals(finalResult1, new double[]{1, 2});
+ assertEquals(finalResult2, new double[]{3, 4});
+ }
+
+ @Test
+ public void testSumArrayLong() {
+ SumArrayLongAggregationFunction function = new
SumArrayLongAggregationFunction(List.of(COLUMN), false);
+ long[] longer = {1, 2, 3};
+ long[] shorter = {4};
+ assertEquals(function.mergeFinalResult(longer, shorter), new long[]{5, 2,
3});
+ assertEquals(function.mergeFinalResult(shorter, longer), new long[]{5, 2,
3});
+ assertEquals(longer, new long[]{1, 2, 3});
+ assertEquals(shorter, new long[]{4});
+ }
+
+ @Test
+ public void testSumArrayDouble() {
+ SumArrayDoubleAggregationFunction function = new
SumArrayDoubleAggregationFunction(List.of(COLUMN), false);
+ double[] longer = {1, 2};
+ double[] shorter = {0.5};
+ assertEquals(function.mergeFinalResult(longer, shorter), new double[]{1.5,
2});
+ assertEquals(function.mergeFinalResult(shorter, longer), new double[]{1.5,
2});
+ assertEquals(longer, new double[]{1, 2});
+ assertEquals(shorter, new double[]{0.5});
+ }
+
+ @Test
+ public void testArrayAgg() {
+ ArrayAggIntFunction function = new ArrayAggIntFunction(COLUMN,
DataType.INT, false);
+ assertEquals(function.mergeFinalResult(new int[]{1, 2}, new int[]{2}), new
int[]{1, 2, 2});
+ assertEquals(function.mergeFinalResult(new int[0], new int[]{3}), new
int[]{3});
+ assertEquals(function.mergeFinalResult(new int[]{3}, new int[0]), new
int[]{3});
+
+ ArrayAggDistinctIntFunction distinctFunction = new
ArrayAggDistinctIntFunction(COLUMN, DataType.INT, false);
+ assertEquals(distinctFunction.mergeFinalResult(new int[]{1, 2}, new
int[]{3}), new int[]{1, 2, 3});
+
+ ArrayAggBigDecimalFunction bigDecimalFunction = new
ArrayAggBigDecimalFunction(COLUMN, false);
+ BigDecimal[] merged =
+ bigDecimalFunction.mergeFinalResult(new BigDecimal[]{BigDecimal.ONE},
new BigDecimal[]{BigDecimal.TEN});
+ assertEquals(merged, new BigDecimal[]{BigDecimal.ONE, BigDecimal.TEN});
+ }
+
+ @Test
+ public void testPercentileRawKLL() {
+ List<ExpressionContext> arguments = List.of(COLUMN,
ExpressionContext.forLiteral(Literal.doubleValue(50)));
+ for (PercentileKLLAggregationFunction function : List.of(new
PercentileRawKLLAggregationFunction(arguments, false),
+ new PercentileRawKLLMVAggregationFunction(arguments, false))) {
+ KllDoublesSketch sketch1 = KllDoublesSketch.newHeapInstance();
+ sketch1.update(1);
+ sketch1.update(2);
+ KllDoublesSketch sketch2 = KllDoublesSketch.newHeapInstance();
+ sketch2.update(3);
+ String merged = (String) function.mergeFinalResult(new
SerializedKLL(sketch1, 50).toString(),
+ new SerializedKLL(sketch2, 50).toString());
+ KllDoublesSketch mergedSketch =
+
KllDoublesSketch.heapify(MemorySegment.ofArray(Base64.getDecoder().decode(merged)));
+ assertEquals(mergedSketch.getN(), 3);
+ assertEquals(mergedSketch.getMinItem(), 1.0);
+ assertEquals(mergedSketch.getMaxItem(), 3.0);
+ }
+ }
+
+ @Test
+ public void testFrequentLongsSketch() {
+ FrequentLongsSketchAggregationFunction function =
+ new FrequentLongsSketchAggregationFunction(List.of(COLUMN), false);
+ FrequentLongsSketch sketch1 = new FrequentLongsSketch(8);
+ sketch1.update(1, 3);
+ sketch1.update(2);
+ FrequentLongsSketch sketch2 = new FrequentLongsSketch(8);
+ sketch2.update(3, 2);
+ String merged = function.mergeFinalResult(new
SerializedFrequentLongsSketch(sketch1).toString(),
+ new SerializedFrequentLongsSketch(sketch2).toString());
+ FrequentLongsSketch mergedSketch =
+
FrequentLongsSketch.getInstance(MemorySegment.ofArray(Base64.getDecoder().decode(merged)));
+ assertEquals(mergedSketch.getStreamLength(), 6);
+ assertEquals(mergedSketch.getEstimate(1), 3);
+ assertEquals(mergedSketch.getEstimate(2), 1);
+ assertEquals(mergedSketch.getEstimate(3), 2);
+ }
+
+ @Test
+ public void testFrequentStringsSketch() {
+ FrequentStringsSketchAggregationFunction function =
+ new FrequentStringsSketchAggregationFunction(List.of(COLUMN), false);
+ FrequentItemsSketch<String> sketch1 = new FrequentItemsSketch<>(8);
+ sketch1.update("a", 3);
+ sketch1.update("b");
+ FrequentItemsSketch<String> sketch2 = new FrequentItemsSketch<>(8);
+ sketch2.update("c", 2);
+ String merged = function.mergeFinalResult(new
SerializedFrequentStringsSketch(sketch1).toString(),
+ new SerializedFrequentStringsSketch(sketch2).toString());
+ FrequentItemsSketch<String> mergedSketch = FrequentItemsSketch.getInstance(
+ MemorySegment.ofArray(Base64.getDecoder().decode(merged)), new
ArrayOfStringsSerDe());
+ assertEquals(mergedSketch.getStreamLength(), 6);
+ assertEquals(mergedSketch.getEstimate("a"), 3);
+ assertEquals(mergedSketch.getEstimate("b"), 1);
+ assertEquals(mergedSketch.getEstimate("c"), 2);
+ }
+
+ @Test
+ public void testIdSet()
+ throws IOException {
+ IdSetAggregationFunction function = new
IdSetAggregationFunction(List.of(COLUMN), false);
+ IdSet idSet1 = IdSets.create(DataType.INT);
+ idSet1.add(1);
+ IdSet idSet2 = IdSets.create(DataType.INT);
+ idSet2.add(2);
+ IdSet merged =
IdSets.fromBase64String(function.mergeFinalResult(idSet1.toBase64String(),
idSet2.toBase64String()));
+ assertTrue(merged.contains(1));
+ assertTrue(merged.contains(2));
+ assertFalse(merged.contains(3));
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntegerTupleSketchAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntegerTupleSketchAggregationFunctionTest.java
index 20c1eb7d836..08c26999b10 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntegerTupleSketchAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/IntegerTupleSketchAggregationFunctionTest.java
@@ -18,15 +18,25 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.lang.foreign.MemorySegment;
+import java.util.Base64;
import java.util.List;
import java.util.Map;
+import org.apache.datasketches.tuple.TupleSketch;
+import org.apache.datasketches.tuple.TupleSketchIterator;
import org.apache.datasketches.tuple.aninteger.IntegerSummary;
+import org.apache.datasketches.tuple.aninteger.IntegerSummaryDeserializer;
+import org.apache.datasketches.tuple.aninteger.IntegerTupleSketch;
import org.apache.pinot.common.request.Literal;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.segment.spi.Constants;
-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.assertThrows;
+import static org.testng.Assert.assertTrue;
+
public class IntegerTupleSketchAggregationFunctionTest {
@@ -36,10 +46,10 @@ public class IntegerTupleSketchAggregationFunctionTest {
new
IntegerTupleSketchAggregationFunction(List.of(ExpressionContext.forIdentifier("col")),
IntegerSummary.Mode.Sum, false);
- Assert.assertTrue(function.canUseStarTree(Map.of()));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
"16384")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
16384)));
-
Assert.assertFalse(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
8192)));
+ assertTrue(function.canUseStarTree(Map.of()));
+
assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
"16384")));
+
assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
16384)));
+
assertFalse(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
8192)));
}
@Test
@@ -49,10 +59,45 @@ public class IntegerTupleSketchAggregationFunctionTest {
IntegerSummary.Mode.Sum, false);
// Default StarTree lgK = 14 / K=16384
- Assert.assertFalse(function.canUseStarTree(Map.of()));
-
Assert.assertFalse(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
"16384")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
"65536")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
32768)));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
"32768")));
+ assertFalse(function.canUseStarTree(Map.of()));
+
assertFalse(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
"16384")));
+
assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
"65536")));
+
assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
32768)));
+
assertTrue(function.canUseStarTree(Map.of(Constants.THETA_TUPLE_SKETCH_NOMINAL_ENTRIES,
"32768")));
+ }
+
+ @Test
+ public void testMergeRawFinalResults() {
+ IntegerTupleSketchAggregationFunction function =
+ new
IntegerTupleSketchAggregationFunction(List.of(ExpressionContext.forIdentifier("col")),
+ IntegerSummary.Mode.Sum, false);
+ IntegerTupleSketch sketch1 = new IntegerTupleSketch(4,
IntegerSummary.Mode.Sum);
+ sketch1.update(1L, 1);
+ sketch1.update(2L, 2);
+ IntegerTupleSketch sketch2 = new IntegerTupleSketch(4,
IntegerSummary.Mode.Sum);
+ sketch2.update(3L, 3);
+
+ String merged = (String) function.mergeFinalResult(serialize(sketch1),
serialize(sketch2));
+ TupleSketch<IntegerSummary> mergedSketch = TupleSketch.heapifySketch(
+ MemorySegment.ofArray(Base64.getDecoder().decode(merged)), new
IntegerSummaryDeserializer());
+ assertEquals(mergedSketch.getEstimate(), 3.0);
+ int sum = 0;
+ TupleSketchIterator<IntegerSummary> iterator = mergedSketch.iterator();
+ while (iterator.next()) {
+ sum += iterator.getSummary().getValue();
+ }
+ assertEquals(sum, 6);
+ }
+
+ @Test
+ public void testMergeAvgValueFinalResults() {
+ AvgValueIntegerTupleSketchAggregationFunction function =
+ new
AvgValueIntegerTupleSketchAggregationFunction(List.of(ExpressionContext.forIdentifier("col")),
+ IntegerSummary.Mode.Sum, false);
+ assertThrows(UnsupportedOperationException.class, () ->
function.mergeFinalResult(1L, 2L));
+ }
+
+ private static String serialize(IntegerTupleSketch sketch) {
+ return Base64.getEncoder().encodeToString(sketch.compact().toByteArray());
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ListAggFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ListAggFunctionTest.java
index 92fce918ad5..c78059c16b9 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ListAggFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ListAggFunctionTest.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.util.List;
import java.util.Map;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.core.common.SyntheticBlockValSets;
@@ -27,9 +28,11 @@ import
org.apache.pinot.core.query.aggregation.function.array.ListAggFunction;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.roaringbitmap.RoaringBitmap;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
public class ListAggFunctionTest extends AbstractAggregationFunctionTest {
@@ -148,4 +151,35 @@ public class ListAggFunctionTest extends
AbstractAggregationFunctionTest {
assertEquals(fn.extractFinalResult(gb.getResult(0)), "A");
assertEquals(fn.extractFinalResult(gb.getResult(1)), "A:B:C");
}
+
+ @Test
+ public void testMergeFinalResult() {
+ ListAggFunction fn = new
ListAggFunction(ExpressionContext.forIdentifier("myField"), ",", false);
+ assertEquals(fn.mergeFinalResult("A,B", "C"), "A,B,C");
+ assertEquals(fn.mergeFinalResult("", "A"), ",A");
+ assertEquals(fn.mergeFinalResult("A", ""), "A,");
+
+ ListAggDistinctFunction distinctFn =
+ new
ListAggDistinctFunction(ExpressionContext.forIdentifier("myField"), ",", false);
+ assertEquals(distinctFn.mergeFinalResult("A,B", "C"), "A,B,C");
+ assertEquals(distinctFn.mergeFinalResult("", "A"), ",A");
+ assertEquals(distinctFn.mergeFinalResult("A", ""), "A,");
+ }
+
+ @Test
+ public void testFinalResultOfNullValues() {
+ RoaringBitmap nullBitmap = RoaringBitmap.bitmapOf(0, 1);
+ for (ListAggFunction fn : List.of(
+ new ListAggFunction(ExpressionContext.forIdentifier("myField"), ",",
true),
+ new
ListAggDistinctFunction(ExpressionContext.forIdentifier("myField"), ",", true)
+ )) {
+ AggregationResultHolder holder = fn.createAggregationResultHolder();
+ fn.aggregate(2, holder,
Map.of(ExpressionContext.forIdentifier("myField"),
+ SyntheticBlockValSets.Str.create(nullBitmap, new String[]{"", ""})));
+ String finalResult = fn.extractFinalResult(holder.getResult());
+ assertNull(finalResult);
+ assertEquals(AggregationFunctionUtils.mergeFinalResult(fn, finalResult,
"A"), "A");
+ assertEquals(AggregationFunctionUtils.mergeFinalResult(fn, "A",
finalResult), "A");
+ }
+ }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/StUnionAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/StUnionAggregationFunctionTest.java
index cad60ae7b68..2efaa6c6df6 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/StUnionAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/StUnionAggregationFunctionTest.java
@@ -27,6 +27,7 @@ import
org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.segment.local.utils.GeometrySerializer;
import org.apache.pinot.segment.local.utils.GeometryUtils;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.utils.ByteArray;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.geom.util.GeometryCombiner;
import org.locationtech.jts.operation.union.UnaryUnionOp;
@@ -79,6 +80,19 @@ public class StUnionAggregationFunctionTest {
assertEquals(aggregationResultHolder.getResult(), expected);
}
+ @Test
+ public void testMergeFinalResults()
+ throws Exception {
+ ExpressionContext expression =
ExpressionContext.forIdentifier("geometryColumn");
+ StUnionAggregationFunction aggregationFunction = new
StUnionAggregationFunction(List.of(expression), false);
+ Geometry polygon = GeometryUtils.GEOMETRY_WKT_READER.read("POLYGON((0 0, 0
5, 5 5, 5 0, 0 0))");
+ Geometry line = GeometryUtils.GEOMETRY_WKT_READER.read("LINESTRING(20 20,
25 25)");
+
+ ByteArray merged = aggregationFunction.mergeFinalResult(new
ByteArray(GeometrySerializer.serialize(polygon)),
+ new ByteArray(GeometrySerializer.serialize(line)));
+ assertEquals(GeometrySerializer.deserialize(merged.getBytes()),
aggregationFunction.merge(polygon, line));
+ }
+
private static class BytesBlockValSet extends SyntheticBlockValSets.Base {
private final byte[][] _values;
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunctionTest.java
index 7b5345700e0..4b43e089170 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunctionTest.java
@@ -18,6 +18,10 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.math.BigDecimal;
+import java.util.List;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.queries.FluentQueryTest;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.FieldSpec.DataType;
@@ -26,6 +30,8 @@ import org.apache.pinot.spi.data.Schema;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
+import static org.testng.Assert.assertEquals;
+
public class SumPrecisionAggregationFunctionTest extends
AbstractAggregationFunctionTest {
@@ -145,6 +151,19 @@ public class SumPrecisionAggregationFunctionTest extends
AbstractAggregationFunc
);
}
+ @Test
+ void mergeFinalResultRoundsToPrecisionAndScale() {
+ ExpressionContext column = ExpressionContext.forIdentifier("myField");
+ ExpressionContext precision =
ExpressionContext.forLiteral(Literal.intValue(3));
+ SumPrecisionAggregationFunction function = new
SumPrecisionAggregationFunction(List.of(column, precision), false);
+ assertEquals(function.mergeFinalResult(new BigDecimal("999"), new
BigDecimal("999")), new BigDecimal("2.00E+3"));
+
+ ExpressionContext scale =
ExpressionContext.forLiteral(Literal.intValue(1));
+ function = new SumPrecisionAggregationFunction(List.of(column, precision,
scale), false);
+ assertEquals(function.mergeFinalResult(new BigDecimal("999.0"), new
BigDecimal("999.0")),
+ new BigDecimal("2000.0"));
+ }
+
private String getStringValueOfSum(int sum, DataType dataType) {
if (dataType == DataType.FLOAT || dataType == DataType.DOUBLE) {
return String.valueOf((double) sum);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelFinalResultTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelFinalResultTest.java
index 4de127fbc34..9cc85adf9e0 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelFinalResultTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/funnel/FunnelFinalResultTest.java
@@ -112,6 +112,22 @@ public class FunnelFinalResultTest {
assertEquals(function.mergeFinalResult(new String[0], oneFunnel),
oneFunnel);
}
+ @Test
+ public void testFunnelCountFinalResultMerge() {
+ List<ExpressionContext> arguments = List.of(
+ ExpressionContext.forFunction(new
FunctionContext(FunctionContext.Type.TRANSFORM, "steps",
+ List.of(ExpressionContext.forIdentifier("step0"),
ExpressionContext.forIdentifier("step1")))),
+ ExpressionContext.forFunction(new
FunctionContext(FunctionContext.Type.TRANSFORM, "correlateby",
+ List.of(ExpressionContext.forIdentifier("userId")))));
+ FunnelCountAggregationFunction<?, ?> function =
+ (FunnelCountAggregationFunction<?, ?>) new
FunnelCountAggregationFunctionFactory(arguments, true).get();
+ long[] finalResult1 = {2, 1};
+ long[] finalResult2 = {1, 0};
+ assertEquals(function.mergeFinalResult(finalResult1, finalResult2), new
long[]{3, 1});
+ assertEquals(finalResult1, new long[]{2, 1});
+ assertEquals(finalResult2, new long[]{1, 0});
+ }
+
@Test
public void testBitmapFinalExtractionAvoidsUnnecessaryCopies() {
RoaringBitmap firstStep = mock(RoaringBitmap.class);
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/ArrayAggQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/ArrayAggQueriesTest.java
index 674203533e9..dcb34efd42d 100644
--- a/pinot-core/src/test/java/org/apache/pinot/queries/ArrayAggQueriesTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/ArrayAggQueriesTest.java
@@ -25,6 +25,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
@@ -45,6 +46,7 @@ import org.apache.pinot.spi.config.table.TableType;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
+import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.AfterClass;
@@ -189,17 +191,20 @@ public class ArrayAggQueriesTest extends BaseQueriesTest {
assertEquals(((List) aggregationResult.get(i)).size(), NUM_RECORDS);
}
- // Inter segments
- ResultTable resultTable = getBrokerResponse(query).getResultTable();
- assertEquals(resultTable.getRows().get(0).length, 8);
- assertEquals(((int[]) resultTable.getRows().get(0)[0]).length, 4 *
NUM_RECORDS);
- assertEquals(((long[]) resultTable.getRows().get(0)[1]).length, 4 *
NUM_RECORDS);
- assertEquals(((float[]) resultTable.getRows().get(0)[2]).length, 4 *
NUM_RECORDS);
- assertEquals(((double[]) resultTable.getRows().get(0)[3]).length, 4 *
NUM_RECORDS);
- assertEquals(((String[]) resultTable.getRows().get(0)[4]).length, 4 *
NUM_RECORDS);
- assertEquals(((String[]) resultTable.getRows().get(0)[5]).length, 4 *
NUM_RECORDS);
- assertEquals(((String[]) resultTable.getRows().get(0)[6]).length, 4 *
NUM_RECORDS);
- assertEquals(((String[]) resultTable.getRows().get(0)[7]).length, 4 *
NUM_RECORDS);
+ // Inter segments, with and without servers returning final results
+ for (Map<String, String> queryOptions : List.of(Map.<String, String>of(),
+ Map.of(QueryOptionKey.SERVER_RETURN_FINAL_RESULT, "true"))) {
+ ResultTable resultTable = getBrokerResponse(query,
queryOptions).getResultTable();
+ assertEquals(resultTable.getRows().get(0).length, 8);
+ assertEquals(((int[]) resultTable.getRows().get(0)[0]).length, 4 *
NUM_RECORDS);
+ assertEquals(((long[]) resultTable.getRows().get(0)[1]).length, 4 *
NUM_RECORDS);
+ assertEquals(((float[]) resultTable.getRows().get(0)[2]).length, 4 *
NUM_RECORDS);
+ assertEquals(((double[]) resultTable.getRows().get(0)[3]).length, 4 *
NUM_RECORDS);
+ assertEquals(((String[]) resultTable.getRows().get(0)[4]).length, 4 *
NUM_RECORDS);
+ assertEquals(((String[]) resultTable.getRows().get(0)[5]).length, 4 *
NUM_RECORDS);
+ assertEquals(((String[]) resultTable.getRows().get(0)[6]).length, 4 *
NUM_RECORDS);
+ assertEquals(((String[]) resultTable.getRows().get(0)[7]).length, 4 *
NUM_RECORDS);
+ }
}
@Test
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountThetaSketchQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountThetaSketchQueriesTest.java
index 7d498377724..c7080d07216 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountThetaSketchQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/DistinctCountThetaSketchQueriesTest.java
@@ -25,6 +25,7 @@ import java.util.Arrays;
import java.util.Base64;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import org.apache.commons.io.FileUtils;
import org.apache.datasketches.theta.ThetaSketch;
import org.apache.datasketches.theta.UpdatableThetaSketch;
@@ -50,6 +51,7 @@ import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.spi.data.readers.GenericRow;
import org.apache.pinot.spi.exception.BadQueryRequestException;
+import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.AfterClass;
@@ -302,6 +304,11 @@ public class DistinctCountThetaSketchQueriesTest extends
BaseQueriesTest {
ThetaSketch sketch =
ObjectSerDeUtils.DATA_SKETCH_THETA_SER_DE.deserialize(Base64.getDecoder().decode(serializedSketch));
assertEquals(Math.round(sketch.getEstimate()), NUM_RECORDS);
+
+ brokerResponse = getBrokerResponse(query,
Map.of(QueryOptionKey.SERVER_RETURN_FINAL_RESULT, "true"));
+ serializedSketch = (String)
brokerResponse.getResultTable().getRows().get(0)[0];
+ sketch =
ObjectSerDeUtils.DATA_SKETCH_THETA_SER_DE.deserialize(Base64.getDecoder().decode(serializedSketch));
+ assertEquals(Math.round(sketch.getEstimate()), NUM_RECORDS);
}
@Test
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentAggregationSingleValueQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentAggregationSingleValueQueriesTest.java
index d85a8d9e500..5912d03d3fb 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentAggregationSingleValueQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/InterSegmentAggregationSingleValueQueriesTest.java
@@ -20,6 +20,7 @@ package org.apache.pinot.queries;
import java.util.Arrays;
import java.util.List;
+import java.util.Map;
import java.util.function.Function;
import org.apache.pinot.common.response.broker.BrokerResponseNative;
import org.apache.pinot.common.response.broker.ResultTable;
@@ -28,6 +29,7 @@ import
org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.plan.maker.InstancePlanMakerImplV2;
import org.apache.pinot.spi.utils.BytesUtils;
+import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.testng.annotations.Test;
import static org.testng.Assert.assertFalse;
@@ -36,6 +38,8 @@ import static org.testng.Assert.assertTrue;
public class InterSegmentAggregationSingleValueQueriesTest extends
BaseSingleValueQueriesTest {
private static final String GROUP_BY = " GROUP BY column9 ORDER BY v1 DESC,
v2 DESC LIMIT 1";
+ private static final Map<String, String> SERVER_RETURN_FINAL_RESULT =
+ Map.of(QueryOptionKey.SERVER_RETURN_FINAL_RESULT, "true");
// Allow 5% quantile error due to the randomness of TDigest merge
private static final double PERCENTILE_TDIGEST_DELTA = 0.05 *
Integer.MAX_VALUE;
@@ -324,6 +328,10 @@ public class InterSegmentAggregationSingleValueQueriesTest
extends BaseSingleVal
QueriesTestUtils.testInterSegmentsResult(brokerResponse, 120000L, 0L, 0L,
120000L, expectedResultTable,
cardinalityExtractor);
+ brokerResponse = getBrokerResponse(query, SERVER_RETURN_FINAL_RESULT);
+ QueriesTestUtils.testInterSegmentsResult(brokerResponse, 120000L, 0L, 0L,
120000L, expectedResultTable,
+ cardinalityExtractor);
+
brokerResponse = getBrokerResponse(query + FILTER);
expectedResultTable = new ResultTable(expectedDataSchema,
List.<Object[]>of(new Object[]{1886L, 4492L}));
QueriesTestUtils.testInterSegmentsResult(brokerResponse, 24516L, 252256L,
49032L, 120000L, expectedResultTable,
@@ -358,6 +366,10 @@ public class InterSegmentAggregationSingleValueQueriesTest
extends BaseSingleVal
QueriesTestUtils.testInterSegmentsResult(brokerResponse, 120000L, 0L, 0L,
120000L, expectedResultTable,
cardinalityExtractor);
+ brokerResponse = getBrokerResponse(query, SERVER_RETURN_FINAL_RESULT);
+ QueriesTestUtils.testInterSegmentsResult(brokerResponse, 120000L, 0L, 0L,
120000L, expectedResultTable,
+ cardinalityExtractor);
+
brokerResponse = getBrokerResponse(query + FILTER);
expectedResultTable = new ResultTable(expectedDataSchema,
List.<Object[]>of(new Object[]{1885L, 4545L}));
QueriesTestUtils.testInterSegmentsResult(brokerResponse, 24516L, 252256L,
49032L, 120000L, expectedResultTable,
@@ -585,6 +597,8 @@ public class InterSegmentAggregationSingleValueQueriesTest
extends BaseSingleVal
percentile, percentile);
QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(rawQuery),
getBrokerResponse(regularQuery),
quantileExtractor);
+ QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(rawQuery,
SERVER_RETURN_FINAL_RESULT),
+ getBrokerResponse(regularQuery), quantileExtractor);
QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(rawQuery +
FILTER),
getBrokerResponse(regularQuery + FILTER), quantileExtractor);
QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(rawQuery +
GROUP_BY),
@@ -619,6 +633,8 @@ public class InterSegmentAggregationSingleValueQueriesTest
extends BaseSingleVal
percentile, percentile);
QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(rawQuery),
getBrokerResponse(regularQuery),
quantileExtractor, PERCENTILE_TDIGEST_DELTA);
+ QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(rawQuery,
SERVER_RETURN_FINAL_RESULT),
+ getBrokerResponse(regularQuery), quantileExtractor,
PERCENTILE_TDIGEST_DELTA);
QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(rawQuery +
FILTER),
getBrokerResponse(regularQuery + FILTER), quantileExtractor,
PERCENTILE_TDIGEST_DELTA);
QueriesTestUtils.testInterSegmentsResult(getBrokerResponse(rawQuery +
GROUP_BY),
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
