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 3c92923b82d Give the value-accumulating aggregations the query's null
handling option, and add their missing multi-value paths (#19326)
3c92923b82d is described below
commit 3c92923b82d0ee75ecaa863d07bf7f9356ec2e35
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Fri Aug 21 12:17:05 2026 -0700
Give the value-accumulating aggregations the query's null handling option,
and add their missing multi-value paths (#19326)
---
.../function/AggregationFunctionFactory.java | 19 +-
.../function/FourthMomentAggregationFunction.java | 87 ++--
.../function/HistogramAggregationFunction.java | 452 ++++++++++++++---
.../function/IdSetAggregationFunction.java | 537 ++++++++++++++-------
.../function/StUnionAggregationFunction.java | 148 +++++-
.../array/SumArrayDoubleAggregationFunction.java | 63 ++-
.../array/SumArrayLongAggregationFunction.java | 64 ++-
.../pinot/core/common/SyntheticBlockValSets.java | 232 +++++++++
.../AggregationFunctionNullContractTest.java | 13 +-
.../function/StUnionAggregationFunctionTest.java | 4 +-
.../function/ValueAggregationNullHandlingTest.java | 453 +++++++++++++++++
11 files changed, 1704 insertions(+), 368 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
index 4354ae48282..519333ab345 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionFactory.java
@@ -299,9 +299,9 @@ public class AggregationFunctionFactory {
return new ListAggFunction(arguments.get(0), separator,
nullHandlingEnabled);
}
case SUMARRAYLONG:
- return new SumArrayLongAggregationFunction(arguments);
+ return new SumArrayLongAggregationFunction(arguments,
nullHandlingEnabled);
case SUMARRAYDOUBLE:
- return new SumArrayDoubleAggregationFunction(arguments);
+ return new SumArrayDoubleAggregationFunction(arguments,
nullHandlingEnabled);
case ARRAYAGG: {
Preconditions.checkArgument(numArguments >= 2,
"ARRAY_AGG expects 2 or 3 arguments, got: %s. The function can
be used as "
@@ -420,7 +420,7 @@ public class AggregationFunctionFactory {
case DISTINCTAVG:
return new DistinctAvgAggregationFunction(arguments,
nullHandlingEnabled);
case IDSET:
- return new IdSetAggregationFunction(arguments);
+ return new IdSetAggregationFunction(arguments,
nullHandlingEnabled);
case COUNTMV:
return new CountMVAggregationFunction(arguments,
nullHandlingEnabled);
case MINMV:
@@ -454,9 +454,9 @@ public class AggregationFunctionFactory {
case DISTINCTAVGMV:
return new DistinctAvgMVAggregationFunction(arguments,
nullHandlingEnabled);
case STUNION:
- return new StUnionAggregationFunction(arguments);
+ return new StUnionAggregationFunction(arguments,
nullHandlingEnabled);
case HISTOGRAM:
- return new HistogramAggregationFunction(arguments);
+ return new HistogramAggregationFunction(arguments,
nullHandlingEnabled);
case COVARPOP:
return new CovarianceAggregationFunction(arguments, false,
nullHandlingEnabled);
case COVARSAMP:
@@ -474,11 +474,14 @@ public class AggregationFunctionFactory {
case STDDEVSAMP:
return new VarianceAggregationFunction(arguments, true, true,
nullHandlingEnabled);
case SKEWNESS:
- return new FourthMomentAggregationFunction(arguments,
FourthMomentAggregationFunction.Type.SKEWNESS);
+ return new FourthMomentAggregationFunction(arguments,
FourthMomentAggregationFunction.Type.SKEWNESS,
+ nullHandlingEnabled);
case KURTOSIS:
- return new FourthMomentAggregationFunction(arguments,
FourthMomentAggregationFunction.Type.KURTOSIS);
+ return new FourthMomentAggregationFunction(arguments,
FourthMomentAggregationFunction.Type.KURTOSIS,
+ nullHandlingEnabled);
case FOURTHMOMENT:
- return new FourthMomentAggregationFunction(arguments,
FourthMomentAggregationFunction.Type.MOMENT);
+ return new FourthMomentAggregationFunction(arguments,
FourthMomentAggregationFunction.Type.MOMENT,
+ nullHandlingEnabled);
case DISTINCTCOUNTTUPLESKETCH:
// mode actually doesn't matter here because we only care about
keys, not values
return new
DistinctCountIntegerTupleSketchAggregationFunction(arguments,
IntegerSummary.Mode.Sum,
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FourthMomentAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FourthMomentAggregationFunction.java
index d4a7dd871dc..64f4162fccd 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FourthMomentAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FourthMomentAggregationFunction.java
@@ -35,7 +35,8 @@ import
org.apache.pinot.segment.local.customobject.PinotFourthMoment;
import org.apache.pinot.segment.spi.AggregationFunctionType;
-public class FourthMomentAggregationFunction extends
BaseSingleInputAggregationFunction<PinotFourthMoment, Double> {
+public class FourthMomentAggregationFunction
+ extends NullableSingleInputAggregationFunction<PinotFourthMoment, Double> {
private final Type _type;
@@ -43,8 +44,9 @@ public class FourthMomentAggregationFunction extends
BaseSingleInputAggregationF
KURTOSIS, SKEWNESS, MOMENT
}
- public FourthMomentAggregationFunction(List<ExpressionContext> arguments,
Type type) {
- super(verifySingleArgument(arguments, type.name()));
+ public FourthMomentAggregationFunction(List<ExpressionContext> arguments,
Type type,
+ boolean nullHandlingEnabled) {
+ super(verifySingleArgument(arguments, type.name()), nullHandlingEnabled);
_type = type;
}
@@ -77,65 +79,75 @@ public class FourthMomentAggregationFunction extends
BaseSingleInputAggregationF
Map<ExpressionContext, BlockValSet> blockValSetMap) {
double[] values =
StatisticalAggregationFunctionUtils.getValSet(blockValSetMap, _expression);
- PinotFourthMoment m4 = aggregationResultHolder.getResult();
- if (m4 == null) {
- m4 = new PinotFourthMoment();
- aggregationResultHolder.setValue(m4);
- }
-
- for (int i = 0; i < length; i++) {
- m4.increment(values[i]);
- }
+ // The moment is created inside the range, so a block with no non-null row
leaves the holder untouched and
+ // extractFinalResult sees the null that means nothing was aggregated
+ forEachNotNull(length, blockValSetMap.get(_expression), (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ PinotFourthMoment m4 = aggregationResultHolder.getResult();
+ if (m4 == null) {
+ m4 = new PinotFourthMoment();
+ aggregationResultHolder.setValue(m4);
+ }
+ for (int i = from; i < to; i++) {
+ m4.increment(values[i]);
+ }
+ });
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
double[] values =
StatisticalAggregationFunctionUtils.getValSet(blockValSetMap, _expression);
- for (int i = 0; i < length; i++) {
- PinotFourthMoment m4 = groupByResultHolder.getResult(groupKeyArray[i]);
- if (m4 == null) {
- m4 = new PinotFourthMoment();
- groupByResultHolder.setValueForKey(groupKeyArray[i], m4);
+ forEachNotNull(length, blockValSetMap.get(_expression), (from, to) -> {
+ for (int i = from; i < to; i++) {
+ PinotFourthMoment m4 = groupByResultHolder.getResult(groupKeyArray[i]);
+ if (m4 == null) {
+ m4 = new PinotFourthMoment();
+ groupByResultHolder.setValueForKey(groupKeyArray[i], m4);
+ }
+ m4.increment(values[i]);
}
- m4.increment(values[i]);
- }
+ });
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
double[] values =
StatisticalAggregationFunctionUtils.getValSet(blockValSetMap, _expression);
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- PinotFourthMoment m4 = groupByResultHolder.getResult(groupKey);
- if (m4 == null) {
- m4 = new PinotFourthMoment();
- groupByResultHolder.setValueForKey(groupKey, m4);
+ forEachNotNull(length, blockValSetMap.get(_expression), (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ PinotFourthMoment m4 = groupByResultHolder.getResult(groupKey);
+ if (m4 == null) {
+ m4 = new PinotFourthMoment();
+ groupByResultHolder.setValueForKey(groupKey, m4);
+ }
+ m4.increment(values[i]);
}
- m4.increment(values[i]);
}
- }
+ });
}
+ @Nullable
@Override
public PinotFourthMoment extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
PinotFourthMoment m4 = aggregationResultHolder.getResult();
- if (m4 == null) {
- return new PinotFourthMoment();
- } else {
+ if (m4 != null) {
return m4;
}
+ // With the option disabled an untouched holder still renders the empty
accumulator, which is the
+ // intermediate this mode has always emitted; with it enabled the null is
the signal that nothing was
+ // aggregated.
+ return _nullHandlingEnabled ? null : new PinotFourthMoment();
}
+ @Nullable
@Override
public PinotFourthMoment extractGroupByResult(GroupByResultHolder
groupByResultHolder, int groupKey) {
PinotFourthMoment m4 = groupByResultHolder.getResult(groupKey);
- if (m4 == null) {
- return new PinotFourthMoment();
- } else {
- return m4;
- }
+ return m4 != null ? m4 : (_nullHandlingEnabled ? null : new
PinotFourthMoment());
}
@Override
@@ -169,7 +181,10 @@ public class FourthMomentAggregationFunction extends
BaseSingleInputAggregationF
@Override
public Double extractFinalResult(@Nullable PinotFourthMoment m4) {
if (m4 == null) {
- return null;
+ // A null intermediate result means nothing was aggregated. With null
handling enabled the skewness of nothing
+ // is NULL; with it disabled it is what an untouched moment renders to,
which Commons Math reports as NaN below
+ // the three samples skewness needs and the four kurtosis needs.
+ return _nullHandlingEnabled ? null : Double.NaN;
}
switch (_type) {
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 b4ca9bacc91..80830edcd92 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
@@ -20,6 +20,7 @@ package org.apache.pinot.core.query.aggregation.function;
import com.google.common.base.Preconditions;
import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import java.math.BigDecimal;
import java.util.List;
import java.util.Map;
import javax.annotation.Nullable;
@@ -41,7 +42,8 @@ import org.apache.pinot.spi.utils.ArrayCopyUtils;
/// usage example:
/// `Histogram(columnName, ARRAY\[0,1,10,100\])` to specify bins \[0,1),
\[1,10), \[10,1000\] or
/// `Histogram(columnName, 0, 1000, 10)` to specify 10 equal-length bins
\[0,100), \[100,200), ..., \[900,1000\]
-public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunction<DoubleArrayList, DoubleArrayList> {
+public class HistogramAggregationFunction
+ extends NullableSingleInputAggregationFunction<DoubleArrayList,
DoubleArrayList> {
private static final String ARRAY_CONSTRUCTOR = "arrayvalueconstructor";
private static final int INVALID_BIN = -1;
@@ -51,8 +53,8 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
double _upper;
double _binLength;
- public HistogramAggregationFunction(List<ExpressionContext> arguments) {
- super(arguments.get(0));
+ public HistogramAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
+ super(arguments.get(0), nullHandlingEnabled);
int numArguments = arguments.size();
Preconditions.checkArgument(numArguments == 4 || numArguments == 2,
"Histogram expects 2 or 4 arguments, got: %s;"
+ " usage example: `Histogram(columnName, ARRAY[0,1,10,100])` to
specify bins [0,1), [1,10), [10,1000] or "
@@ -150,6 +152,14 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
///
/// @param val input value
/// @return bin id
+ /// Counts one value into the supplied histogram, ignoring values that fall
outside every bin.
+ private void increment(double[] histogram, double value) {
+ int binId = getBinId(value);
+ if (binId != INVALID_BIN) {
+ histogram[binId] += 1;
+ }
+ }
+
private int getBinId(double val) {
if (val > _upper || val < _lower) {
return INVALID_BIN;
@@ -191,14 +201,17 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
return new ObjectGroupByResultHolder(initialCapacity, maxCapacity);
}
+ @Nullable
@Override
public DoubleArrayList extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
- DoubleArrayList aggregationResultHolderResult =
aggregationResultHolder.getResult();
- if (aggregationResultHolderResult == null) {
- return DoubleVectorOpUtils.createAndInitialize(getNumBins());
- } else {
- return aggregationResultHolderResult;
+ DoubleArrayList histogram = aggregationResultHolder.getResult();
+ if (histogram != null) {
+ return histogram;
}
+ // With the option disabled an untouched holder still renders the empty
accumulator, which is the
+ // intermediate this mode has always emitted; with it enabled the null is
the signal that nothing was
+ // aggregated.
+ return _nullHandlingEnabled ? null :
DoubleVectorOpUtils.createAndInitialize(getNumBins());
}
@Nullable
@@ -237,6 +250,11 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
@Nullable
@Override
public DoubleArrayList extractFinalResult(@Nullable DoubleArrayList
doubleArrayList) {
+ if (doubleArrayList == null) {
+ // A null intermediate result means nothing was aggregated. With null
handling enabled the histogram of nothing
+ // is NULL; with it disabled it is the all-zero histogram, which is the
answer this mode has always given.
+ return _nullHandlingEnabled ? null :
DoubleVectorOpUtils.createAndInitialize(getNumBins());
+ }
return doubleArrayList;
}
@@ -244,50 +262,153 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- Preconditions.checkState(blockValSet.isSingleValue(), "Histogram currently
only supports single-valued column");
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ } else {
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ }
+ }
+
+ private void aggregateSVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
int[] values = blockValSet.getIntValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- double value = values[i];
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ int value = values[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- }
+ });
break;
}
case LONG: {
long[] values = blockValSet.getLongValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- double value = values[i];
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ long value = values[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- }
+ });
break;
}
case FLOAT: {
float[] values = blockValSet.getFloatValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- double value = values[i];
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ float value = values[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- }
+ });
break;
}
case DOUBLE: {
double[] values = blockValSet.getDoubleValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- double value = values[i];
- for (int groupKey : groupKeysArray[i]) {
- setGroupByResult(groupKey, groupByResultHolder, value);
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ double value = values[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
}
- }
+ });
+ break;
+ }
+ case BIG_DECIMAL: {
+ BigDecimal[] values = blockValSet.getBigDecimalValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ double value = values[i].doubleValue();
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
+ }
+ });
break;
}
default:
- throw new IllegalStateException("Cannot compute histogram for
non-numeric type: " + blockValSet.getValueType());
+ throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ + blockValSet.getValueType());
+ }
+ }
+
+ private void aggregateMVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ switch (blockValSet.getValueType().getStoredType()) {
+ case INT: {
+ int[][] values = blockValSet.getIntValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (int value : values[i]) {
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
+ }
+ }
+ });
+ break;
+ }
+ case LONG: {
+ long[][] values = blockValSet.getLongValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (long value : values[i]) {
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
+ }
+ }
+ });
+ break;
+ }
+ case FLOAT: {
+ float[][] values = blockValSet.getFloatValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (float value : values[i]) {
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
+ }
+ }
+ });
+ break;
+ }
+ case DOUBLE: {
+ double[][] values = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (double value : values[i]) {
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder, value);
+ }
+ }
+ }
+ });
+ break;
+ }
+ case BIG_DECIMAL: {
+ BigDecimal[][] values = blockValSet.getBigDecimalValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (BigDecimal value : values[i]) {
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupKey, groupByResultHolder,
value.doubleValue());
+ }
+ }
+ }
+ });
+ break;
+ }
+ default:
+ throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ + blockValSet.getValueType());
}
}
@@ -295,37 +416,133 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet);
+ } else {
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet);
+ }
+ }
+
+ private void aggregateSVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
int[] values = blockValSet.getIntValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- setGroupByResult(groupKeyArray[i], groupByResultHolder, values[i]);
- }
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ int value = values[i];
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ });
break;
}
case LONG: {
long[] values = blockValSet.getLongValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- setGroupByResult(groupKeyArray[i], groupByResultHolder, values[i]);
- }
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ long value = values[i];
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ });
break;
}
case FLOAT: {
float[] values = blockValSet.getFloatValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- setGroupByResult(groupKeyArray[i], groupByResultHolder, values[i]);
- }
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ float value = values[i];
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ });
break;
}
case DOUBLE: {
double[] values = blockValSet.getDoubleValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- setGroupByResult(groupKeyArray[i], groupByResultHolder, values[i]);
- }
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ double value = values[i];
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ });
+ break;
+ }
+ case BIG_DECIMAL: {
+ BigDecimal[] values = blockValSet.getBigDecimalValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ double value = values[i].doubleValue();
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ });
break;
}
default:
- throw new IllegalStateException("Cannot compute histogram for
non-numeric type: " + blockValSet.getValueType());
+ throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ + blockValSet.getValueType());
+ }
+ }
+
+ private void aggregateMVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ switch (blockValSet.getValueType().getStoredType()) {
+ case INT: {
+ int[][] values = blockValSet.getIntValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (int value : values[i]) {
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ }
+ });
+ break;
+ }
+ case LONG: {
+ long[][] values = blockValSet.getLongValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (long value : values[i]) {
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ }
+ });
+ break;
+ }
+ case FLOAT: {
+ float[][] values = blockValSet.getFloatValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (float value : values[i]) {
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ }
+ });
+ break;
+ }
+ case DOUBLE: {
+ double[][] values = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (double value : values[i]) {
+ setGroupByResult(groupKeyArray[i], groupByResultHolder, value);
+ }
+ }
+ });
+ break;
+ }
+ case BIG_DECIMAL: {
+ BigDecimal[][] values = blockValSet.getBigDecimalValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (BigDecimal value : values[i]) {
+ setGroupByResult(groupKeyArray[i], groupByResultHolder,
value.doubleValue());
+ }
+ }
+ });
+ break;
+ }
+ default:
+ throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ + blockValSet.getValueType());
}
}
@@ -345,56 +562,149 @@ public class HistogramAggregationFunction extends
BaseSingleInputAggregationFunc
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- //TODO: Add MV support for histogram
- Preconditions.checkState(blockValSet.isSingleValue(), "Histogram currently
only supports single-valued column");
- double[] histogram = new double[this.getNumBins()];
+ if (blockValSet.isSingleValue()) {
+ aggregateSV(length, aggregationResultHolder, blockValSet);
+ } else {
+ aggregateMV(length, aggregationResultHolder, blockValSet);
+ }
+ }
+
+ private void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ double[] histogram = new double[getNumBins()];
+ int numRows;
switch (blockValSet.getValueType().getStoredType()) {
case INT: {
int[] values = blockValSet.getIntValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- int binId = this.getBinId(values[i]);
- if (binId != INVALID_BIN) {
- histogram[binId] += 1;
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ increment(histogram, values[i]);
}
- }
- setAggregationResult(aggregationResultHolder, histogram);
+ return acum + to - from;
+ });
break;
}
case LONG: {
long[] values = blockValSet.getLongValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- int binId = this.getBinId(values[i]);
- if (binId != INVALID_BIN) {
- histogram[binId] += 1;
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ increment(histogram, values[i]);
}
- }
- setAggregationResult(aggregationResultHolder, histogram);
+ return acum + to - from;
+ });
break;
}
case FLOAT: {
float[] values = blockValSet.getFloatValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- int binId = this.getBinId(values[i]);
- if (binId != INVALID_BIN) {
- histogram[binId] += 1;
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ increment(histogram, values[i]);
}
- }
- setAggregationResult(aggregationResultHolder, histogram);
+ return acum + to - from;
+ });
break;
}
case DOUBLE: {
double[] values = blockValSet.getDoubleValuesSV();
- for (int i = 0; i < length && i < values.length; i++) {
- int binId = this.getBinId(values[i]);
- if (binId != INVALID_BIN) {
- histogram[binId] += 1;
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ increment(histogram, values[i]);
}
- }
- setAggregationResult(aggregationResultHolder, histogram);
+ return acum + to - from;
+ });
+ break;
+ }
+ case BIG_DECIMAL: {
+ BigDecimal[] values = blockValSet.getBigDecimalValuesSV();
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ increment(histogram, values[i].doubleValue());
+ }
+ return acum + to - from;
+ });
break;
}
default:
- throw new IllegalStateException("Cannot compute histogram for
non-numeric type: " + blockValSet.getValueType());
+ throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ + blockValSet.getValueType());
+ }
+ // The histogram is published only when a row reached it, so a block with
no non-null row leaves the holder
+ // untouched and extractFinalResult sees the null that means nothing was
aggregated. It is published once rather
+ // than per range, because the buffer accumulates across ranges and adding
it again would recount earlier rows.
+ if (numRows > 0) {
+ setAggregationResult(aggregationResultHolder, histogram);
+ }
+ }
+
+ private void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ double[] histogram = new double[getNumBins()];
+ int numRows;
+ switch (blockValSet.getValueType().getStoredType()) {
+ case INT: {
+ int[][] values = blockValSet.getIntValuesMV();
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (int value : values[i]) {
+ increment(histogram, value);
+ }
+ }
+ return acum + to - from;
+ });
+ break;
+ }
+ case LONG: {
+ long[][] values = blockValSet.getLongValuesMV();
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (long value : values[i]) {
+ increment(histogram, value);
+ }
+ }
+ return acum + to - from;
+ });
+ break;
+ }
+ case FLOAT: {
+ float[][] values = blockValSet.getFloatValuesMV();
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (float value : values[i]) {
+ increment(histogram, value);
+ }
+ }
+ return acum + to - from;
+ });
+ break;
+ }
+ case DOUBLE: {
+ double[][] values = blockValSet.getDoubleValuesMV();
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (double value : values[i]) {
+ increment(histogram, value);
+ }
+ }
+ return acum + to - from;
+ });
+ break;
+ }
+ case BIG_DECIMAL: {
+ BigDecimal[][] values = blockValSet.getBigDecimalValuesMV();
+ numRows = foldNotNull(length, blockValSet, 0, (acum, from, to) -> {
+ for (int i = from; i < to && i < values.length; i++) {
+ for (BigDecimal value : values[i]) {
+ increment(histogram, value.doubleValue());
+ }
+ }
+ return acum + to - from;
+ });
+ break;
+ }
+ default:
+ throw new IllegalStateException("Cannot compute histogram for
non-numeric type: "
+ + blockValSet.getValueType());
+ }
+ if (numRows > 0) {
+ setAggregationResult(aggregationResultHolder, histogram);
}
}
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 30cacf08e95..2e93c26cc97 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
@@ -55,7 +55,7 @@ import org.apache.pinot.spi.data.FieldSpec.DataType;
/// - fpp: Desired false positive probability for the BloomFilter, must be
positive and less than 1.0. (Default 0.03)
///
/// Example: IDSET(col,
'sizeThresholdInBytes=1000;expectedInsertions=10000;fpp=0.03')
-public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction<IdSet, String> {
+public class IdSetAggregationFunction extends
NullableSingleInputAggregationFunction<IdSet, String> {
private static final char PARAMETER_DELIMITER = ';';
private static final char PARAMETER_KEY_VALUE_SEPARATOR = '=';
private static final String UPPER_CASE_SIZE_THRESHOLD_IN_BYTES =
"SIZETHRESHOLDINBYTES";
@@ -66,8 +66,8 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
private final int _expectedInsertions;
private final double _fpp;
- public IdSetAggregationFunction(List<ExpressionContext> arguments) {
- super(arguments.get(0));
+ public IdSetAggregationFunction(List<ExpressionContext> arguments, boolean
nullHandlingEnabled) {
+ super(arguments.get(0), nullHandlingEnabled);
if (arguments.size() == 1) {
_sizeThresholdInBytes = IdSets.DEFAULT_SIZE_THRESHOLD_IN_BYTES;
_expectedInsertions = IdSets.DEFAULT_EXPECTED_INSERTIONS;
@@ -127,94 +127,182 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- DataType storedType = blockValSet.getValueType().getStoredType();
- IdSet idSet = getIdSet(aggregationResultHolder, storedType);
if (blockValSet.isSingleValue()) {
- switch (storedType) {
- case INT:
- int[] intValuesSV = blockValSet.getIntValuesSV();
- for (int i = 0; i < length; i++) {
+ aggregateSV(length, aggregationResultHolder, blockValSet);
+ } else {
+ aggregateMV(length, aggregationResultHolder, blockValSet);
+ }
+ }
+
+ private void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[] intValuesSV = blockValSet.getIntValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
idSet.add(intValuesSV[i]);
}
- break;
- case LONG:
- long[] longValuesSV = blockValSet.getLongValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case LONG:
+ long[] longValuesSV = blockValSet.getLongValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
idSet.add(longValuesSV[i]);
}
- break;
- case FLOAT:
- float[] floatValuesSV = blockValSet.getFloatValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case FLOAT:
+ float[] floatValuesSV = blockValSet.getFloatValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
idSet.add(floatValuesSV[i]);
}
- break;
- case DOUBLE:
- double[] doubleValuesSV = blockValSet.getDoubleValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case DOUBLE:
+ double[] doubleValuesSV = blockValSet.getDoubleValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
idSet.add(doubleValuesSV[i]);
}
- break;
- case STRING:
- String[] stringValuesSV = blockValSet.getStringValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case STRING:
+ String[] stringValuesSV = blockValSet.getStringValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
idSet.add(stringValuesSV[i]);
}
- break;
- case BYTES:
- byte[][] bytesValuesSV = blockValSet.getBytesValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case BYTES:
+ byte[][] bytesValuesSV = blockValSet.getBytesValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
idSet.add(bytesValuesSV[i]);
}
- break;
- default:
- throw new IllegalStateException("Illegal SV data type for ID_SET
aggregation function: " + storedType);
- }
- } else {
- switch (storedType) {
- case INT:
- int[][] intValuesMV = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ default:
+ throw new IllegalStateException("Illegal SV data type for ID_SET
aggregation function: " + storedType);
+ }
+ }
+
+ private void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[][] intValuesMV = blockValSet.getIntValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
for (int intValue : intValuesMV[i]) {
idSet.add(intValue);
}
}
- break;
- case LONG:
- long[][] longValuesMV = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case LONG:
+ long[][] longValuesMV = blockValSet.getLongValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
for (long longValue : longValuesMV[i]) {
idSet.add(longValue);
}
}
- break;
- case FLOAT:
- float[][] floatValuesMV = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case FLOAT:
+ float[][] floatValuesMV = blockValSet.getFloatValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
for (float floatValue : floatValuesMV[i]) {
idSet.add(floatValue);
}
}
- break;
- case DOUBLE:
- double[][] doubleValuesMV = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case DOUBLE:
+ double[][] doubleValuesMV = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
for (double doubleValue : doubleValuesMV[i]) {
idSet.add(doubleValue);
}
}
- break;
- case STRING:
- String[][] stringValuesMV = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case STRING:
+ String[][] stringValuesMV = blockValSet.getStringValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
for (String stringValue : stringValuesMV[i]) {
idSet.add(stringValue);
}
}
- break;
- default:
- throw new IllegalStateException("Illegal MV data type for ID_SET
aggregation function: " + storedType);
- }
+ });
+ break;
+ case BYTES:
+ byte[][][] bytesValuesMV = blockValSet.getBytesValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ IdSet idSet = getIdSet(aggregationResultHolder, storedType);
+ for (int i = from; i < to; i++) {
+ for (byte[] bytesValue : bytesValuesMV[i]) {
+ idSet.add(bytesValue);
+ }
+ }
+ });
+ break;
+ default:
+ throw new IllegalStateException("Illegal MV data type for ID_SET
aggregation function: " + storedType);
}
}
@@ -222,98 +310,142 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- DataType storedType = blockValSet.getValueType().getStoredType();
if (blockValSet.isSingleValue()) {
- switch (storedType) {
- case INT:
- int[] intValuesSV = blockValSet.getIntValuesSV();
- for (int i = 0; i < length; i++) {
+ aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet);
+ } else {
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet);
+ }
+ }
+
+ private void aggregateSVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[] intValuesSV = blockValSet.getIntValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.INT).add(intValuesSV[i]);
}
- break;
- case LONG:
- long[] longValuesSV = blockValSet.getLongValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case LONG:
+ long[] longValuesSV = blockValSet.getLongValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.LONG).add(longValuesSV[i]);
}
- break;
- case FLOAT:
- float[] floatValuesSV = blockValSet.getFloatValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case FLOAT:
+ float[] floatValuesSV = blockValSet.getFloatValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.FLOAT).add(floatValuesSV[i]);
}
- break;
- case DOUBLE:
- double[] doubleValuesSV = blockValSet.getDoubleValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case DOUBLE:
+ double[] doubleValuesSV = blockValSet.getDoubleValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.DOUBLE).add(doubleValuesSV[i]);
}
- break;
- case STRING:
- String[] stringValuesSV = blockValSet.getStringValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case STRING:
+ String[] stringValuesSV = blockValSet.getStringValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.STRING).add(stringValuesSV[i]);
}
- break;
- case BYTES:
- byte[][] bytesValuesSV = blockValSet.getBytesValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case BYTES:
+ byte[][] bytesValuesSV = blockValSet.getBytesValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.BYTES).add(bytesValuesSV[i]);
}
- break;
- default:
- throw new IllegalStateException("Illegal SV data type for ID_SET
aggregation function: " + storedType);
- }
- } else {
- switch (storedType) {
- case INT:
- int[][] intValuesMV = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ default:
+ throw new IllegalStateException("Illegal SV data type for ID_SET
aggregation function: " + storedType);
+ }
+ }
+
+ private void aggregateMVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[][] intValuesMV = blockValSet.getIntValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
IdSet idSet = getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.INT);
for (int intValue : intValuesMV[i]) {
idSet.add(intValue);
}
}
- break;
- case LONG:
- long[][] longValuesMV = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case LONG:
+ long[][] longValuesMV = blockValSet.getLongValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
IdSet idSet = getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.LONG);
for (long longValue : longValuesMV[i]) {
idSet.add(longValue);
}
}
- break;
- case FLOAT:
- float[][] floatValuesMV = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case FLOAT:
+ float[][] floatValuesMV = blockValSet.getFloatValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
IdSet idSet = getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.FLOAT);
for (float floatValue : floatValuesMV[i]) {
idSet.add(floatValue);
}
}
- break;
- case DOUBLE:
- double[][] doubleValuesMV = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case DOUBLE:
+ double[][] doubleValuesMV = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
IdSet idSet = getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.DOUBLE);
for (double doubleValue : doubleValuesMV[i]) {
idSet.add(doubleValue);
}
}
- break;
- case STRING:
- String[][] stringValuesMV = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case STRING:
+ String[][] stringValuesMV = blockValSet.getStringValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
IdSet idSet = getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.STRING);
for (String stringValue : stringValuesMV[i]) {
idSet.add(stringValue);
}
}
- break;
- default:
- throw new IllegalStateException("Illegal MV data type for ID_SET
aggregation function: " + storedType);
- }
+ });
+ break;
+ case BYTES:
+ byte[][][] bytesValuesMV = blockValSet.getBytesValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ IdSet idSet = getIdSet(groupByResultHolder, groupKeyArray[i],
DataType.BYTES);
+ for (byte[] bytesValue : bytesValuesMV[i]) {
+ idSet.add(bytesValue);
+ }
+ }
+ });
+ break;
+ default:
+ throw new IllegalStateException("Illegal MV data type for ID_SET
aggregation function: " + storedType);
}
}
@@ -321,71 +453,96 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- DataType storedType = blockValSet.getValueType().getStoredType();
if (blockValSet.isSingleValue()) {
- switch (storedType) {
- case INT:
- int[] intValuesSV = blockValSet.getIntValuesSV();
- for (int i = 0; i < length; i++) {
+ aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ } else {
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ }
+ }
+
+ private void aggregateSVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[] intValuesSV = blockValSet.getIntValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
int intValue = intValuesSV[i];
for (int groupKey : groupKeysArray[i]) {
getIdSet(groupByResultHolder, groupKey,
DataType.INT).add(intValue);
}
}
- break;
- case LONG:
- long[] longValuesSV = blockValSet.getLongValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case LONG:
+ long[] longValuesSV = blockValSet.getLongValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
long longValue = longValuesSV[i];
for (int groupKey : groupKeysArray[i]) {
getIdSet(groupByResultHolder, groupKey,
DataType.LONG).add(longValue);
}
}
- break;
- case FLOAT:
- float[] floatValuesSV = blockValSet.getFloatValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case FLOAT:
+ float[] floatValuesSV = blockValSet.getFloatValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
float floatValue = floatValuesSV[i];
for (int groupKey : groupKeysArray[i]) {
getIdSet(groupByResultHolder, groupKey,
DataType.FLOAT).add(floatValue);
}
}
- break;
- case DOUBLE:
- double[] doubleValuesSV = blockValSet.getDoubleValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case DOUBLE:
+ double[] doubleValuesSV = blockValSet.getDoubleValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
double doubleValue = doubleValuesSV[i];
for (int groupKey : groupKeysArray[i]) {
getIdSet(groupByResultHolder, groupKey,
DataType.DOUBLE).add(doubleValue);
}
}
- break;
- case STRING:
- String[] stringValuesSV = blockValSet.getStringValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case STRING:
+ String[] stringValuesSV = blockValSet.getStringValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
String stringValue = stringValuesSV[i];
for (int groupKey : groupKeysArray[i]) {
getIdSet(groupByResultHolder, groupKey,
DataType.STRING).add(stringValue);
}
}
- break;
- case BYTES:
- byte[][] bytesValuesSV = blockValSet.getBytesValuesSV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case BYTES:
+ byte[][] bytesValuesSV = blockValSet.getBytesValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
byte[] bytesValue = bytesValuesSV[i];
for (int groupKey : groupKeysArray[i]) {
getIdSet(groupByResultHolder, groupKey,
DataType.BYTES).add(bytesValue);
}
}
- break;
- default:
- throw new IllegalStateException("Illegal SV data type for ID_SET
aggregation function: " + storedType);
- }
- } else {
- switch (storedType) {
- case INT:
- int[][] intValuesMV = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ default:
+ throw new IllegalStateException("Illegal SV data type for ID_SET
aggregation function: " + storedType);
+ }
+ }
+
+ private void aggregateMVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[][] intValuesMV = blockValSet.getIntValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
int[] intValues = intValuesMV[i];
for (int groupKey : groupKeysArray[i]) {
IdSet idSet = getIdSet(groupByResultHolder, groupKey,
DataType.INT);
@@ -394,10 +551,12 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
}
}
}
- break;
- case LONG:
- long[][] longValuesMV = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case LONG:
+ long[][] longValuesMV = blockValSet.getLongValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
long[] longValues = longValuesMV[i];
for (int groupKey : groupKeysArray[i]) {
IdSet idSet = getIdSet(groupByResultHolder, groupKey,
DataType.LONG);
@@ -406,10 +565,12 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
}
}
}
- break;
- case FLOAT:
- float[][] floatValuesMV = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case FLOAT:
+ float[][] floatValuesMV = blockValSet.getFloatValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
float[] floatValues = floatValuesMV[i];
for (int groupKey : groupKeysArray[i]) {
IdSet idSet = getIdSet(groupByResultHolder, groupKey,
DataType.FLOAT);
@@ -418,10 +579,12 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
}
}
}
- break;
- case DOUBLE:
- double[][] doubleValuesMV = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case DOUBLE:
+ double[][] doubleValuesMV = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
double[] doubleValues = doubleValuesMV[i];
for (int groupKey : groupKeysArray[i]) {
IdSet idSet = getIdSet(groupByResultHolder, groupKey,
DataType.DOUBLE);
@@ -430,10 +593,12 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
}
}
}
- break;
- case STRING:
- String[][] stringValuesMV = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
+ });
+ break;
+ case STRING:
+ String[][] stringValuesMV = blockValSet.getStringValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
String[] stringValues = stringValuesMV[i];
for (int groupKey : groupKeysArray[i]) {
IdSet idSet = getIdSet(groupByResultHolder, groupKey,
DataType.STRING);
@@ -442,23 +607,45 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
}
}
}
- break;
- default:
- throw new IllegalStateException("Illegal MV data type for ID_SET
aggregation function: " + storedType);
- }
+ });
+ break;
+ case BYTES:
+ byte[][][] bytesValuesMV = blockValSet.getBytesValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ byte[][] bytesValues = bytesValuesMV[i];
+ for (int groupKey : groupKeysArray[i]) {
+ IdSet idSet = getIdSet(groupByResultHolder, groupKey,
DataType.BYTES);
+ for (byte[] bytesValue : bytesValues) {
+ idSet.add(bytesValue);
+ }
+ }
+ }
+ });
+ break;
+ default:
+ throw new IllegalStateException("Illegal MV data type for ID_SET
aggregation function: " + storedType);
}
}
@Override
+ @Nullable
public IdSet extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
IdSet idSet = aggregationResultHolder.getResult();
- return idSet != null ? idSet : IdSets.emptyIdSet();
+ if (idSet != null) {
+ return idSet;
+ }
+ // With the option disabled an untouched holder still renders the empty
accumulator, which is the
+ // intermediate this mode has always emitted; with it enabled the null is
the signal that nothing was
+ // aggregated.
+ return _nullHandlingEnabled ? null : IdSets.emptyIdSet();
}
@Override
+ @Nullable
public IdSet extractGroupByResult(GroupByResultHolder groupByResultHolder,
int groupKey) {
IdSet idSet = groupByResultHolder.getResult(groupKey);
- return idSet != null ? idSet : IdSets.emptyIdSet();
+ return idSet != null ? idSet : (_nullHandlingEnabled ? null :
IdSets.emptyIdSet());
}
@Override
@@ -490,12 +677,18 @@ public class IdSetAggregationFunction extends
BaseSingleInputAggregationFunction
@Nullable
@Override
public String extractFinalResult(@Nullable IdSet intermediateResult) {
- // A null intermediate result means nothing was aggregated, and there is
no id set to serialize
- if (intermediateResult == null) {
- return null;
+ // A null intermediate result means nothing was aggregated. With null
handling enabled there is no id set to
+ // serialize and the answer is NULL; with it disabled it is the serialized
empty id set, which is the answer this
+ // mode has always given.
+ IdSet idSet = intermediateResult;
+ if (idSet == null) {
+ if (_nullHandlingEnabled) {
+ return null;
+ }
+ idSet = IdSets.emptyIdSet();
}
try {
- return intermediateResult.toBase64String();
+ return idSet.toBase64String();
} catch (IOException e) {
throw new RuntimeException("Caught exception while serializing IdSet",
e);
}
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 039fa8912b3..5ad959a33fe 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
@@ -39,10 +39,10 @@ import org.locationtech.jts.geom.util.GeometryCombiner;
import org.locationtech.jts.operation.union.UnaryUnionOp;
-public class StUnionAggregationFunction extends
BaseSingleInputAggregationFunction<Geometry, ByteArray> {
+public class StUnionAggregationFunction extends
NullableSingleInputAggregationFunction<Geometry, ByteArray> {
- public StUnionAggregationFunction(List<ExpressionContext> arguments) {
- super(verifySingleArgument(arguments, "ST_UNION"));
+ public StUnionAggregationFunction(List<ExpressionContext> arguments, boolean
nullHandlingEnabled) {
+ super(verifySingleArgument(arguments, "ST_UNION"), nullHandlingEnabled);
}
@Override
@@ -63,49 +63,142 @@ public class StUnionAggregationFunction extends
BaseSingleInputAggregationFuncti
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- byte[][] bytesArray = blockValSetMap.get(_expression).getBytesValuesSV();
- Geometry geometry = aggregationResultHolder.getResult();
- for (int i = 0; i < length; i++) {
- geometry = union(geometry,
GeometrySerializer.deserialize(bytesArray[i]));
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ if (blockValSet.isSingleValue()) {
+ aggregateSV(length, aggregationResultHolder, blockValSet);
+ } else {
+ aggregateMV(length, aggregationResultHolder, blockValSet);
}
- aggregationResultHolder.setValue(geometry);
+ }
+
+ private void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ byte[][] bytesArray = blockValSet.getBytesValuesSV();
+ // The holder is written only from inside the range, so a block with no
non-null row leaves it untouched and
+ // extractFinalResult sees the null that means nothing was aggregated
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ Geometry geometry = aggregationResultHolder.getResult();
+ for (int i = from; i < to; i++) {
+ geometry = union(geometry,
GeometrySerializer.deserialize(bytesArray[i]));
+ }
+ aggregationResultHolder.setValue(geometry);
+ });
+ }
+
+ /// Every geometry of a multi-value row is folded into the same union, so a
row contributes once per value.
+ private void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ byte[][][] bytesArrays = blockValSet.getBytesValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ Geometry geometry = aggregationResultHolder.getResult();
+ for (int i = from; i < to; i++) {
+ for (byte[] bytes : bytesArrays[i]) {
+ geometry = union(geometry, GeometrySerializer.deserialize(bytes));
+ }
+ }
+ aggregationResultHolder.setValue(geometry);
+ });
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- byte[][] bytesArray = blockValSetMap.get(_expression).getBytesValuesSV();
- for (int i = 0; i < length; i++) {
- int groupKey = groupKeyArray[i];
- Geometry value = GeometrySerializer.deserialize(bytesArray[i]);
- Geometry geometry = groupByResultHolder.getResult(groupKey);
- groupByResultHolder.setValueForKey(groupKey, union(geometry, value));
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet);
+ } else {
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet);
}
}
+ private void aggregateSVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ byte[][] bytesArray = blockValSet.getBytesValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int groupKey = groupKeyArray[i];
+ Geometry value = GeometrySerializer.deserialize(bytesArray[i]);
+ groupByResultHolder.setValueForKey(groupKey,
union(groupByResultHolder.getResult(groupKey), value));
+ }
+ });
+ }
+
+ private void aggregateMVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ byte[][][] bytesArrays = blockValSet.getBytesValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ int groupKey = groupKeyArray[i];
+ for (byte[] bytes : bytesArrays[i]) {
+ Geometry value = GeometrySerializer.deserialize(bytes);
+ groupByResultHolder.setValueForKey(groupKey,
union(groupByResultHolder.getResult(groupKey), value));
+ }
+ }
+ });
+ }
+
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- byte[][] bytesArray = blockValSetMap.get(_expression).getBytesValuesSV();
- for (int i = 0; i < length; i++) {
- Geometry value = GeometrySerializer.deserialize(bytesArray[i]);
- for (int groupKey : groupKeysArray[i]) {
- Geometry geometry = groupByResultHolder.getResult(groupKey);
- groupByResultHolder.setValueForKey(groupKey, union(geometry, value));
- }
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ } else {
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
}
}
+ private void aggregateSVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ byte[][] bytesArray = blockValSet.getBytesValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ // Deserialized once per row, not once per group key the row belongs to
+ Geometry value = GeometrySerializer.deserialize(bytesArray[i]);
+ for (int groupKey : groupKeysArray[i]) {
+ groupByResultHolder.setValueForKey(groupKey,
union(groupByResultHolder.getResult(groupKey), value));
+ }
+ }
+ });
+ }
+
+ private void aggregateMVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ byte[][][] bytesArrays = blockValSet.getBytesValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (byte[] bytes : bytesArrays[i]) {
+ Geometry value = GeometrySerializer.deserialize(bytes);
+ for (int groupKey : groupKeysArray[i]) {
+ groupByResultHolder.setValueForKey(groupKey,
union(groupByResultHolder.getResult(groupKey), value));
+ }
+ }
+ }
+ });
+ }
+
+ @Nullable
@Override
public Geometry extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
Geometry geometry = aggregationResultHolder.getResult();
- return geometry == null ? GeometryUtils.EMPTY_POINT : geometry;
+ if (geometry != null) {
+ return geometry;
+ }
+ // With the option disabled an untouched holder still renders the empty
accumulator, which is the
+ // intermediate this mode has always emitted; with it enabled the null is
the signal that nothing was
+ // aggregated.
+ return _nullHandlingEnabled ? null : GeometryUtils.EMPTY_POINT;
}
+ @Nullable
@Override
public Geometry extractGroupByResult(GroupByResultHolder
groupByResultHolder, int groupKey) {
Geometry geometry = groupByResultHolder.getResult(groupKey);
- return geometry == null ? GeometryUtils.EMPTY_POINT : geometry;
+ return geometry != null ? geometry : (_nullHandlingEnabled ? null :
GeometryUtils.EMPTY_POINT);
}
@Override
@@ -137,8 +230,13 @@ public class StUnionAggregationFunction extends
BaseSingleInputAggregationFuncti
@Nullable
@Override
public ByteArray extractFinalResult(@Nullable Geometry geometry) {
- // A null intermediate result means nothing was aggregated; the union of
no geometries is NULL, matching ST_Union
- return geometry != null ? new
ByteArray(GeometrySerializer.serialize(geometry)) : null;
+ if (geometry == null) {
+ // A null intermediate result means nothing was aggregated. With null
handling enabled the union of no
+ // geometries is NULL, matching ST_Union; with it disabled it is the
empty point, which is the answer this mode
+ // has always given.
+ return _nullHandlingEnabled ? null : new
ByteArray(GeometrySerializer.serialize(GeometryUtils.EMPTY_POINT));
+ }
+ return new ByteArray(GeometrySerializer.serialize(geometry));
}
/// Returns the union of the supplied geometries.
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 a10997e448b..f19468d3a49 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
@@ -29,17 +29,17 @@ 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.ObjectAggregationResultHolder;
-import
org.apache.pinot.core.query.aggregation.function.BaseSingleInputAggregationFunction;
+import
org.apache.pinot.core.query.aggregation.function.NullableSingleInputAggregationFunction;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
import org.apache.pinot.segment.spi.AggregationFunctionType;
public class SumArrayDoubleAggregationFunction
- extends BaseSingleInputAggregationFunction<DoubleArrayList,
DoubleArrayList> {
+ extends NullableSingleInputAggregationFunction<DoubleArrayList,
DoubleArrayList> {
- public SumArrayDoubleAggregationFunction(List<ExpressionContext> arguments) {
- super(verifySingleArgument(arguments, "SUM_ARRAY"));
+ public SumArrayDoubleAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
+ super(verifySingleArgument(arguments, "SUM_ARRAY"), nullHandlingEnabled);
}
@Override
@@ -60,39 +60,50 @@ public class SumArrayDoubleAggregationFunction
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- double[][] values = blockValSetMap.get(_expression).getDoubleValuesMV();
- if (aggregationResultHolder.getResult() == null) {
- aggregationResultHolder.setValue(new DoubleArrayList());
- }
- DoubleArrayList result = aggregationResultHolder.getResult();
- for (int i = 0; i < length; i++) {
- double[] value = values[i];
- aggregateMerge(value, result);
- }
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ double[][] values = blockValSet.getDoubleValuesMV();
+ // The accumulator is created inside the range, so a block with no
non-null row leaves the holder untouched and
+ // extractFinalResult sees the null that means nothing was aggregated
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ DoubleArrayList result = aggregationResultHolder.getResult();
+ if (result == null) {
+ result = new DoubleArrayList();
+ aggregationResultHolder.setValue(result);
+ }
+ for (int i = from; i < to; i++) {
+ aggregateMerge(values[i], result);
+ }
+ });
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- double[][] valuesArray =
blockValSetMap.get(_expression).getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- double[] values = valuesArray[i];
- int groupKey = groupKeyArray[i];
- setGroupByResult(groupByResultHolder, values, groupKey);
- }
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ double[][] valuesArray = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ setGroupByResult(groupByResultHolder, valuesArray[i],
groupKeyArray[i]);
+ }
+ });
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- double[][] valuesArray =
blockValSetMap.get(_expression).getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- double[] values = valuesArray[i];
- int[] groupKeys = groupKeysArray[i];
- for (int groupKey : groupKeys) {
- setGroupByResult(groupByResultHolder, values, groupKey);
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ double[][] valuesArray = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ double[] values = valuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupByResultHolder, values, groupKey);
+ }
}
- }
+ });
}
private void setGroupByResult(GroupByResultHolder groupByResultHolder,
double[] values, int groupKey) {
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 78035b9fa7f..22c2d84a4c8 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
@@ -29,16 +29,17 @@ 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.ObjectAggregationResultHolder;
-import
org.apache.pinot.core.query.aggregation.function.BaseSingleInputAggregationFunction;
+import
org.apache.pinot.core.query.aggregation.function.NullableSingleInputAggregationFunction;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
import org.apache.pinot.segment.spi.AggregationFunctionType;
-public class SumArrayLongAggregationFunction extends
BaseSingleInputAggregationFunction<LongArrayList, LongArrayList> {
+public class SumArrayLongAggregationFunction
+ extends NullableSingleInputAggregationFunction<LongArrayList,
LongArrayList> {
- public SumArrayLongAggregationFunction(List<ExpressionContext> arguments) {
- super(verifySingleArgument(arguments, "SUM_ARRAY"));
+ public SumArrayLongAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
+ super(verifySingleArgument(arguments, "SUM_ARRAY"), nullHandlingEnabled);
}
@Override
@@ -59,39 +60,50 @@ public class SumArrayLongAggregationFunction extends
BaseSingleInputAggregationF
@Override
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[][] values = blockValSetMap.get(_expression).getLongValuesMV();
- if (aggregationResultHolder.getResult() == null) {
- aggregationResultHolder.setValue(new LongArrayList());
- }
- LongArrayList result = aggregationResultHolder.getResult();
- for (int i = 0; i < length; i++) {
- long[] value = values[i];
- aggregateMerge(value, result);
- }
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ long[][] values = blockValSet.getLongValuesMV();
+ // The accumulator is created inside the range, so a block with no
non-null row leaves the holder untouched and
+ // extractFinalResult sees the null that means nothing was aggregated
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ if (to == from) {
+ return;
+ }
+ LongArrayList result = aggregationResultHolder.getResult();
+ if (result == null) {
+ result = new LongArrayList();
+ aggregationResultHolder.setValue(result);
+ }
+ for (int i = from; i < to; i++) {
+ aggregateMerge(values[i], result);
+ }
+ });
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[][] valuesArray = blockValSetMap.get(_expression).getLongValuesMV();
- for (int i = 0; i < length; i++) {
- long[] values = valuesArray[i];
- int groupKey = groupKeyArray[i];
- setGroupByResult(groupByResultHolder, values, groupKey);
- }
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ long[][] valuesArray = blockValSet.getLongValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ setGroupByResult(groupByResultHolder, valuesArray[i],
groupKeyArray[i]);
+ }
+ });
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[][] valuesArray = blockValSetMap.get(_expression).getLongValuesMV();
- for (int i = 0; i < length; i++) {
- long[] values = valuesArray[i];
- int[] groupKeys = groupKeysArray[i];
- for (int groupKey : groupKeys) {
- setGroupByResult(groupByResultHolder, values, groupKey);
+ BlockValSet blockValSet = blockValSetMap.get(_expression);
+ long[][] valuesArray = blockValSet.getLongValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ long[] values = valuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ setGroupByResult(groupByResultHolder, values, groupKey);
+ }
}
- }
+ });
}
private void setGroupByResult(GroupByResultHolder groupByResultHolder,
long[] values, int groupKey) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/common/SyntheticBlockValSets.java
b/pinot-core/src/test/java/org/apache/pinot/core/common/SyntheticBlockValSets.java
index 76164c81a2f..ddb6dc6c314 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/common/SyntheticBlockValSets.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/common/SyntheticBlockValSets.java
@@ -238,6 +238,44 @@ public class SyntheticBlockValSets {
}
}
+ /// A simple [BlockValSet] for nullable, not dictionary-encoded multi-value
int values.
+ public static class IntMV extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final int[][] _values;
+
+ private IntMV(@Nullable RoaringBitmap nullBitmap, int[][] values) {
+ _nullBitmap = nullBitmap;
+ _values = values;
+ }
+
+ public static IntMV create(@Nullable RoaringBitmap nullBitmap, int[][]
values) {
+ return new IntMV(nullBitmap, values);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return DataType.INT;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return false;
+ }
+
+ @Override
+ public int[][] getIntValuesMV() {
+ return _values;
+ }
+ }
+
/// A simple [BlockValSet] for nullable, not dictionary-encoded long values.
public static class Long extends Base {
@@ -332,6 +370,82 @@ public class SyntheticBlockValSets {
}
}
+ /// A simple [BlockValSet] for nullable, not dictionary-encoded float values.
+ public static class Float extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final float[] _values;
+
+ private Float(@Nullable RoaringBitmap nullBitmap, float[] values) {
+ _nullBitmap = nullBitmap;
+ _values = values;
+ }
+
+ public static Float create(@Nullable RoaringBitmap nullBitmap, float[]
values) {
+ return new Float(nullBitmap, values);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return DataType.FLOAT;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return true;
+ }
+
+ @Override
+ public float[] getFloatValuesSV() {
+ return _values;
+ }
+ }
+
+ /// A simple [BlockValSet] for nullable, not dictionary-encoded multi-value
float values.
+ public static class FloatMV extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final float[][] _values;
+
+ private FloatMV(@Nullable RoaringBitmap nullBitmap, float[][] values) {
+ _nullBitmap = nullBitmap;
+ _values = values;
+ }
+
+ public static FloatMV create(@Nullable RoaringBitmap nullBitmap, float[][]
values) {
+ return new FloatMV(nullBitmap, values);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return DataType.FLOAT;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return false;
+ }
+
+ @Override
+ public float[][] getFloatValuesMV() {
+ return _values;
+ }
+ }
+
/// A simple [BlockValSet] for nullable, not dictionary-encoded double
values.
public static class Double extends Base {
@@ -388,6 +502,124 @@ public class SyntheticBlockValSets {
}
}
+ /// A simple [BlockValSet] for nullable, not dictionary-encoded multi-value
double values.
+ public static class DoubleMV extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final double[][] _values;
+
+ private DoubleMV(@Nullable RoaringBitmap nullBitmap, double[][] values) {
+ _nullBitmap = nullBitmap;
+ _values = values;
+ }
+
+ public static DoubleMV create(@Nullable RoaringBitmap nullBitmap,
double[][] values) {
+ return new DoubleMV(nullBitmap, values);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return DataType.DOUBLE;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return false;
+ }
+
+ @Override
+ public double[][] getDoubleValuesMV() {
+ return _values;
+ }
+ }
+
+ /// A simple [BlockValSet] for nullable, not dictionary-encoded BigDecimal
values.
+ ///
+ /// Named `BigDec` rather than `BigDecimal` for the same reason [Str] is not
named `String`: a nested class of that
+ /// name shadows [java.math.BigDecimal] across the whole enclosing class,
which silently changes the signature of
+ /// every `getBigDecimal*` method declared here so that it no longer
implements [BlockValSet].
+ public static class BigDec extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final BigDecimal[] _values;
+
+ private BigDec(@Nullable RoaringBitmap nullBitmap, BigDecimal[] values) {
+ _nullBitmap = nullBitmap;
+ _values = values;
+ }
+
+ public static BigDec create(@Nullable RoaringBitmap nullBitmap,
BigDecimal[] values) {
+ return new BigDec(nullBitmap, values);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return DataType.BIG_DECIMAL;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return true;
+ }
+
+ @Override
+ public BigDecimal[] getBigDecimalValuesSV() {
+ return _values;
+ }
+ }
+
+ /// A simple [BlockValSet] for nullable, not dictionary-encoded multi-value
BigDecimal values.
+ public static class BigDecMV extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final BigDecimal[][] _values;
+
+ private BigDecMV(@Nullable RoaringBitmap nullBitmap, BigDecimal[][]
values) {
+ _nullBitmap = nullBitmap;
+ _values = values;
+ }
+
+ public static BigDecMV create(@Nullable RoaringBitmap nullBitmap,
BigDecimal[][] values) {
+ return new BigDecMV(nullBitmap, values);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return DataType.BIG_DECIMAL;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return false;
+ }
+
+ @Override
+ public BigDecimal[][] getBigDecimalValuesMV() {
+ return _values;
+ }
+ }
+
/// A simple [BlockValSet] for nullable, not dictionary-encoded string
values.
///
/// Named `Str` rather than `String`: a nested class called `String` shadows
`java.lang.String` across
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
index 2a1d77138ef..4e282e463a6 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionNullContractTest.java
@@ -203,7 +203,13 @@ public class AggregationFunctionNullContractTest {
AggregationFunctionType.AVG, AggregationFunctionType.MINMAXRANGE,
AggregationFunctionType.VARPOP,
AggregationFunctionType.STDDEVPOP, AggregationFunctionType.PERCENTILE,
AggregationFunctionType.PERCENTILEEST,
AggregationFunctionType.PERCENTILETDIGEST,
- AggregationFunctionType.PERCENTILEKLL,
AggregationFunctionType.PERCENTILESMARTTDIGEST}) {
+ AggregationFunctionType.PERCENTILEKLL,
AggregationFunctionType.PERCENTILESMARTTDIGEST,
+ // These carry a legacy sentinel with the option disabled - NaN, the
empty point, an all-zero histogram, an
+ // empty id set - so only the enabled answer is NULL, and only the
enabled answer is checked here
+ AggregationFunctionType.SKEWNESS, AggregationFunctionType.KURTOSIS,
AggregationFunctionType.STUNION,
+ AggregationFunctionType.HISTOGRAM, AggregationFunctionType.IDSET,
+ // These two answer NULL in both modes, which is why neither needs a
mode-aware branch
+ AggregationFunctionType.SUMARRAYLONG,
AggregationFunctionType.SUMARRAYDOUBLE}) {
// Built through the shared argument shapes: the percentile families
disagree on whether the percentile is a
// name suffix or an argument, and only some accept both
AggregationFunction function = tryCreate(type, true);
@@ -241,7 +247,9 @@ public class AggregationFunctionNullContractTest {
AggregationFunctionType.PERCENTILETDIGEST,
AggregationFunctionType.PERCENTILERAWTDIGEST,
AggregationFunctionType.PERCENTILESMARTTDIGEST,
AggregationFunctionType.PERCENTILEKLL,
AggregationFunctionType.PERCENTILERAWKLL,
AggregationFunctionType.VARPOP, AggregationFunctionType.VARSAMP,
- AggregationFunctionType.STDDEVPOP, AggregationFunctionType.STDDEVSAMP,
AggregationFunctionType.MINMV,
+ AggregationFunctionType.STDDEVPOP, AggregationFunctionType.STDDEVSAMP,
+ // Both are the same PinotFourthMoment accumulator, so threading the
option into it moves the pair
+ AggregationFunctionType.SKEWNESS, AggregationFunctionType.KURTOSIS,
AggregationFunctionType.MINMV,
AggregationFunctionType.MAXMV, AggregationFunctionType.SUMMV,
AggregationFunctionType.AVGMV,
AggregationFunctionType.MINMAXRANGEMV,
AggregationFunctionType.DISTINCTCOUNTMV,
AggregationFunctionType.DISTINCTSUMMV,
AggregationFunctionType.DISTINCTAVGMV,
@@ -252,6 +260,7 @@ public class AggregationFunctionNullContractTest {
AggregationFunctionType.MINLONG, AggregationFunctionType.MAXLONG,
AggregationFunctionType.SUMINT,
AggregationFunctionType.SUMLONG, AggregationFunctionType.SUMPRECISION,
AggregationFunctionType.FIRSTWITHTIME,
AggregationFunctionType.LASTWITHTIME, AggregationFunctionType.ARRAYAGG,
AggregationFunctionType.LISTAGG,
+ AggregationFunctionType.IDSET, AggregationFunctionType.HISTOGRAM,
// Given the option so they can skip null rows; a row counts only when
both input columns are non-null
AggregationFunctionType.COVARPOP, AggregationFunctionType.COVARSAMP,
// Given the option so they can skip null rows. These two were in this
set once before, on the strength of an
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 1e6371906d1..a114799ec2e 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
@@ -40,7 +40,7 @@ public class StUnionAggregationFunctionTest {
public void testMergeHandlesGeometryCollectionInputs()
throws Exception {
ExpressionContext expression =
ExpressionContext.forIdentifier("geometryColumn");
- StUnionAggregationFunction aggregationFunction = new
StUnionAggregationFunction(List.of(expression));
+ 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)");
Geometry geometryCollection =
@@ -57,7 +57,7 @@ public class StUnionAggregationFunctionTest {
public void testAggregateHandlesMixedDimensionSequence()
throws Exception {
ExpressionContext expression =
ExpressionContext.forIdentifier("geometryColumn");
- StUnionAggregationFunction aggregationFunction = new
StUnionAggregationFunction(List.of(expression));
+ 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(10 10,
15 15)");
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ValueAggregationNullHandlingTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ValueAggregationNullHandlingTest.java
new file mode 100644
index 00000000000..54d2d187ad5
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ValueAggregationNullHandlingTest.java
@@ -0,0 +1,453 @@
+/**
+ * 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 it.unimi.dsi.fastutil.doubles.DoubleArrayList;
+import it.unimi.dsi.fastutil.longs.LongArrayList;
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.SyntheticBlockValSets;
+import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import
org.apache.pinot.core.query.aggregation.function.array.SumArrayDoubleAggregationFunction;
+import
org.apache.pinot.core.query.aggregation.function.array.SumArrayLongAggregationFunction;
+import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
+import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
+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.PinotFourthMoment;
+import org.apache.pinot.segment.local.utils.GeometrySerializer;
+import org.apache.pinot.segment.local.utils.GeometryUtils;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.locationtech.jts.geom.Geometry;
+import org.roaringbitmap.RoaringBitmap;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Null handling, multi-value columns and the empty-input answer for the
value-accumulating aggregations:
+/// `SKEWNESS`, `KURTOSIS`, `STUNION`, `SUMARRAYLONG`, `SUMARRAYDOUBLE`,
`HISTOGRAM` and `IDSET`.
+///
+/// [AggregationFunctionNullContractTest] drives one synthetic single-value
block through `aggregate` only, and it
+/// cannot construct most of these from its shared argument shapes at all, so
the group-by paths, the multi-value
+/// paths and the per-mode empty answers are checked nowhere else.
+///
+/// Each of these functions renders "nothing aggregated" differently with the
option disabled — `NaN`, the empty
+/// point, an all-zero histogram, an empty id set, `NULL` — and that sentinel
is a backward-compatibility
+/// constraint, so both modes are asserted rather than only the SQL one.
+public class ValueAggregationNullHandlingTest {
+ private static final ExpressionContext COLUMN =
ExpressionContext.forIdentifier("column");
+ private static final RoaringBitmap ROW1_NULL = RoaringBitmap.bitmapOf(1);
+
+ private static RoaringBitmap allNull(int length) {
+ RoaringBitmap bitmap = new RoaringBitmap();
+ bitmap.add(0L, length);
+ return bitmap;
+ }
+
+ private static Map<ExpressionContext, BlockValSet> block(BlockValSet
blockValSet) {
+ return Map.of(COLUMN, blockValSet);
+ }
+
+ // ---------- SKEWNESS / KURTOSIS ----------
+
+ private static FourthMomentAggregationFunction skewness(boolean
nullHandlingEnabled) {
+ return new FourthMomentAggregationFunction(List.of(COLUMN),
FourthMomentAggregationFunction.Type.SKEWNESS,
+ nullHandlingEnabled);
+ }
+
+ private static PinotFourthMoment
aggregateDoubles(FourthMomentAggregationFunction function, RoaringBitmap
nullBitmap,
+ double[] values) {
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(values.length, holder,
block(SyntheticBlockValSets.Double.create(nullBitmap, values)));
+ return function.extractAggregationResult(holder);
+ }
+
+ /// A null row contributes nothing, so the skewness equals that of the same
block with the null row removed.
+ @Test
+ public void testSkewnessSkipsNullRows() {
+ double[] withNull = {1.0d, 999.0d, 2.0d, 4.0d, 8.0d};
+ double[] without = {1.0d, 2.0d, 4.0d, 8.0d};
+
+ Double skipped = skewness(true).extractFinalResult(
+ aggregateDoubles(skewness(true), RoaringBitmap.bitmapOf(1), withNull));
+ Double expected =
skewness(true).extractFinalResult(aggregateDoubles(skewness(true), null,
without));
+
+ assertNotNull(skipped);
+ assertEquals(skipped, expected);
+ }
+
+ /// With the option enabled the skewness of nothing is NULL; with it
disabled it is the NaN an untouched moment
+ /// renders to, which is the answer that mode has always given.
+ ///
+ /// The two modes need different inputs to reach "nothing aggregated". With
the option enabled an all-null block
+ /// does it, because every row is skipped. With it disabled the null bitmap
is ignored and those same rows are
+ /// aggregated as values, so only a holder that was never touched gets there.
+ @Test
+ public void testSkewnessEmptyAnswerDiffersByMode() {
+ double[] values = {1.0d, 2.0d, 4.0d};
+
+
assertNull(skewness(true).extractFinalResult(aggregateDoubles(skewness(true),
allNull(3), values)));
+
+ FourthMomentAggregationFunction disabled = skewness(false);
+ Double result = disabled.extractFinalResult(
+
disabled.extractAggregationResult(disabled.createAggregationResultHolder()));
+ assertNotNull(result);
+ assertTrue(result.isNaN(), "expected NaN, got " + result);
+ }
+
+ /// With the option disabled a null row is read as the column default and
still aggregated, which is the answer
+ /// that mode has always given.
+ @Test
+ public void testSkewnessCountsNullRowsWhenOptionDisabled() {
+ double[] values = {1.0d, 2.0d, 4.0d};
+
+ Double withBitmap =
skewness(false).extractFinalResult(aggregateDoubles(skewness(false),
allNull(3), values));
+ Double withoutBitmap =
skewness(false).extractFinalResult(aggregateDoubles(skewness(false), null,
values));
+
+ assertNotNull(withBitmap);
+ assertEquals(withBitmap, withoutBitmap);
+ }
+
+ // ---------- STUNION ----------
+
+ private static StUnionAggregationFunction stUnion(boolean
nullHandlingEnabled) {
+ return new StUnionAggregationFunction(List.of(COLUMN),
nullHandlingEnabled);
+ }
+
+ private static Geometry point(double x, double y) {
+ return GeometryUtils.GEOMETRY_FACTORY.createPoint(new
org.locationtech.jts.geom.Coordinate(x, y));
+ }
+
+ private static byte[] serialized(double x, double y) {
+ return GeometrySerializer.serialize(point(x, y));
+ }
+
+ /// Only the geometries of the rows that carry one are unioned.
+ @Test
+ public void testStUnionSkipsNullRows() {
+ byte[][] values = {serialized(0, 0), serialized(5, 5)};
+
+ StUnionAggregationFunction function = stUnion(true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(2, holder,
block(SyntheticBlockValSets.Bytes.create(ROW1_NULL, values)));
+
+ assertEquals(function.extractAggregationResult(holder), point(0, 0));
+ }
+
+ /// With the option enabled the union of nothing is NULL; with it disabled
it is the empty point.
+ @Test
+ public void testStUnionEmptyAnswerDiffersByMode() {
+ byte[][] values = {serialized(0, 0), serialized(5, 5)};
+
+ StUnionAggregationFunction enabled = stUnion(true);
+ AggregationResultHolder enabledHolder =
enabled.createAggregationResultHolder();
+ enabled.aggregate(2, enabledHolder,
block(SyntheticBlockValSets.Bytes.create(allNull(2), values)));
+
assertNull(enabled.extractFinalResult(enabled.extractAggregationResult(enabledHolder)));
+
+ StUnionAggregationFunction disabled = stUnion(false);
+ assertEquals(disabled.extractFinalResult(
+
disabled.extractAggregationResult(disabled.createAggregationResultHolder())),
+ new
ByteArray(GeometrySerializer.serialize(GeometryUtils.EMPTY_POINT)));
+ }
+
+ /// Every geometry of a multi-value row is folded into the union.
+ @Test
+ public void testStUnionMVColumnUnionsEveryGeometry() {
+ byte[][][] rows = {
+ {serialized(0, 0), serialized(1, 1)},
+ {serialized(5, 5)}
+ };
+
+ StUnionAggregationFunction function = stUnion(true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(2, holder,
block(SyntheticBlockValSets.BytesMV.create(ROW1_NULL, rows)));
+
+ // Row 1 is null, so only the two geometries of row 0 are unioned
+ Geometry result = function.extractAggregationResult(holder);
+ assertNotNull(result);
+ assertEquals(result.getNumGeometries(), 2);
+ assertTrue(result.covers(point(0, 0)));
+ assertTrue(result.covers(point(1, 1)));
+ }
+
+ /// A row's geometries land in that row's group, and a group whose only row
is null is never created.
+ @Test
+ public void testStUnionMVColumnGroupBySV() {
+ byte[][][] rows = {
+ {serialized(0, 0), serialized(1, 1)},
+ {serialized(5, 5)}
+ };
+
+ StUnionAggregationFunction function = stUnion(true);
+ GroupByResultHolder holder = new ObjectGroupByResultHolder(2, 2);
+ function.aggregateGroupBySV(2, new int[]{0, 1}, holder,
+ block(SyntheticBlockValSets.BytesMV.create(ROW1_NULL, rows)));
+
+ Geometry group0 = function.extractGroupByResult(holder, 0);
+ assertNotNull(group0);
+ assertEquals(group0.getNumGeometries(), 2);
+ assertNull(function.extractGroupByResult(holder, 1));
+ }
+
+ // ---------- SUMARRAYLONG / SUMARRAYDOUBLE ----------
+
+ /// A null row contributes none of its array elements.
+ @Test
+ public void testSumArrayLongSkipsNullRows() {
+ long[][] rows = {{1L, 2L}, {100L, 100L}, {10L, 20L}};
+
+ SumArrayLongAggregationFunction function = new
SumArrayLongAggregationFunction(List.of(COLUMN), true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(3, holder,
block(SyntheticBlockValSets.LongMV.create(ROW1_NULL, rows)));
+
+
assertEquals(function.extractFinalResult(function.extractAggregationResult(holder)),
+ new LongArrayList(new long[]{11L, 22L}));
+ }
+
+ @Test
+ public void testSumArrayDoubleSkipsNullRows() {
+ double[][] rows = {{1.5d, 2.5d}, {100d, 100d}, {10d, 20d}};
+
+ SumArrayDoubleAggregationFunction function = new
SumArrayDoubleAggregationFunction(List.of(COLUMN), true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(3, holder,
block(SyntheticBlockValSets.DoubleMV.create(ROW1_NULL, rows)));
+
+
assertEquals(function.extractFinalResult(function.extractAggregationResult(holder)),
+ new DoubleArrayList(new double[]{11.5d, 22.5d}));
+ }
+
+ /// Unlike the others here, the sum of no arrays is NULL in both modes,
which is why neither needs a mode-aware
+ /// branch in extractFinalResult.
+ @Test
+ public void testSumArrayEmptyAnswerIsNullInBothModes() {
+ long[][] rows = {{1L}, {2L}};
+ for (boolean nullHandlingEnabled : new boolean[]{true, false}) {
+ SumArrayLongAggregationFunction function =
+ new SumArrayLongAggregationFunction(List.of(COLUMN),
nullHandlingEnabled);
+ assertNull(function.extractFinalResult(
+
function.extractAggregationResult(function.createAggregationResultHolder())),
+ "SUM_ARRAY over nothing must be NULL with nullHandlingEnabled=" +
nullHandlingEnabled);
+ if (nullHandlingEnabled) {
+ AggregationResultHolder holder =
function.createAggregationResultHolder();
+ function.aggregate(2, holder,
block(SyntheticBlockValSets.LongMV.create(allNull(2), rows)));
+
assertNull(function.extractFinalResult(function.extractAggregationResult(holder)));
+ }
+ }
+ }
+
+ /// A zero-length block still reaches the range callback, and must not
create the accumulator: doing so would
+ /// turn the empty answer into an empty array instead of NULL.
+ @Test
+ public void testSumArrayZeroLengthBlockLeavesTheHolderUntouched() {
+ long[][] rows = {{1L}, {2L}};
+
+ for (boolean nullHandlingEnabled : new boolean[]{true, false}) {
+ SumArrayLongAggregationFunction function =
+ new SumArrayLongAggregationFunction(List.of(COLUMN),
nullHandlingEnabled);
+ AggregationResultHolder holder =
function.createAggregationResultHolder();
+ function.aggregate(0, holder,
block(SyntheticBlockValSets.LongMV.create(null, rows)));
+ assertNull(function.extractAggregationResult(holder),
+ "a zero-length block must leave the holder untouched with
nullHandlingEnabled=" + nullHandlingEnabled);
+ }
+ }
+
+ // ---------- HISTOGRAM ----------
+
+ /// Two equal-length bins over [0, 10): [0, 5) and [5, 10].
+ private static HistogramAggregationFunction histogram(boolean
nullHandlingEnabled) {
+ return new HistogramAggregationFunction(List.of(COLUMN,
+ ExpressionContext.forLiteral(Literal.doubleValue(0)),
+ ExpressionContext.forLiteral(Literal.doubleValue(10)),
+ ExpressionContext.forLiteral(Literal.intValue(2))),
nullHandlingEnabled);
+ }
+
+ private static DoubleArrayList histogramOf(HistogramAggregationFunction
function, BlockValSet blockValSet,
+ int length) {
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(length, holder, block(blockValSet));
+ return
function.extractFinalResult(function.extractAggregationResult(holder));
+ }
+
+ /// A null row is counted into no bin.
+ @Test
+ public void testHistogramSkipsNullRows() {
+ double[] values = {1.0d, 7.0d, 6.0d};
+
+ assertEquals(histogramOf(histogram(true),
SyntheticBlockValSets.Double.create(ROW1_NULL, values), 3),
+ new DoubleArrayList(new double[]{1.0d, 1.0d}));
+ }
+
+ /// With the option enabled the histogram of nothing is NULL; with it
disabled it is the all-zero histogram.
+ @Test
+ public void testHistogramEmptyAnswerDiffersByMode() {
+ double[] values = {1.0d, 7.0d, 6.0d};
+
+ assertNull(histogramOf(histogram(true),
SyntheticBlockValSets.Double.create(allNull(3), values), 3));
+ assertEquals(histogramOf(histogram(false),
SyntheticBlockValSets.Double.create(allNull(3), values), 0),
+ new DoubleArrayList(new double[]{0.0d, 0.0d}));
+ }
+
+ /// Every value of a multi-value row is counted, which is a path this
function did not support before.
+ @Test
+ public void testHistogramMVColumnCountsEveryValue() {
+ double[][] rows = {{1.0d, 2.0d}, {7.0d}, {6.0d, 9.0d}};
+
+ assertEquals(histogramOf(histogram(true),
SyntheticBlockValSets.DoubleMV.create(ROW1_NULL, rows), 3),
+ new DoubleArrayList(new double[]{2.0d, 2.0d}));
+ }
+
+ /// BIG_DECIMAL is numeric and now lands in the same bins as the other
numeric types, single- and multi-value.
+ @Test
+ public void testHistogramBigDecimalColumn() {
+ BigDecimal[] sv = {new BigDecimal("1.5"), new BigDecimal("99"), new
BigDecimal("6.5")};
+ assertEquals(histogramOf(histogram(true),
SyntheticBlockValSets.BigDec.create(ROW1_NULL, sv), 3),
+ new DoubleArrayList(new double[]{1.0d, 1.0d}));
+
+ BigDecimal[][] mv = {{new BigDecimal("1.5"), new BigDecimal("2.5")}, {new
BigDecimal("99")}};
+ assertEquals(histogramOf(histogram(true),
SyntheticBlockValSets.BigDecMV.create(ROW1_NULL, mv), 2),
+ new DoubleArrayList(new double[]{2.0d, 0.0d}));
+ }
+
+ /// A row's values land in that row's group, and a group whose only row is
null is never created.
+ @Test
+ public void testHistogramMVColumnGroupBySV() {
+ double[][] rows = {{1.0d, 2.0d}, {7.0d}, {6.0d}};
+
+ HistogramAggregationFunction function = histogram(true);
+ GroupByResultHolder holder = new ObjectGroupByResultHolder(2, 2);
+ function.aggregateGroupBySV(3, new int[]{0, 1, 0}, holder,
+ block(SyntheticBlockValSets.DoubleMV.create(ROW1_NULL, rows)));
+
+ assertEquals(function.extractGroupByResult(holder, 0), new
DoubleArrayList(new double[]{2.0d, 1.0d}));
+ assertNull(function.extractGroupByResult(holder, 1));
+ }
+
+ /// A null row is skipped for every group key it would have fed.
+ @Test
+ public void testHistogramMVColumnGroupByMV() {
+ double[][] rows = {{1.0d, 2.0d}, {7.0d}};
+
+ HistogramAggregationFunction function = histogram(true);
+ GroupByResultHolder holder = new ObjectGroupByResultHolder(2, 2);
+ function.aggregateGroupByMV(2, new int[][]{{0, 1}, {0, 1}}, holder,
+ block(SyntheticBlockValSets.DoubleMV.create(ROW1_NULL, rows)));
+
+ assertEquals(function.extractGroupByResult(holder, 0), new
DoubleArrayList(new double[]{2.0d, 0.0d}));
+ assertEquals(function.extractGroupByResult(holder, 1), new
DoubleArrayList(new double[]{2.0d, 0.0d}));
+ }
+
+ // ---------- IDSET ----------
+
+ private static IdSetAggregationFunction idSet(boolean nullHandlingEnabled) {
+ return new IdSetAggregationFunction(List.of(COLUMN), nullHandlingEnabled);
+ }
+
+ /// Only the ids of the rows that carry one are collected.
+ @Test
+ public void testIdSetSkipsNullRows() {
+ int[] values = {10, 20, 30};
+
+ IdSetAggregationFunction function = idSet(true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(3, holder,
block(SyntheticBlockValSets.Int.create(ROW1_NULL, values)));
+
+ IdSet result = function.extractAggregationResult(holder);
+ assertNotNull(result);
+ assertTrue(result.contains(10));
+ assertTrue(result.contains(30));
+ assertFalse(result.contains(20));
+ }
+
+ /// With the option enabled the id set of nothing is NULL; with it disabled
it is the empty id set.
+ @Test
+ public void testIdSetEmptyAnswerDiffersByMode() {
+ int[] values = {10, 20};
+
+ IdSetAggregationFunction enabled = idSet(true);
+ AggregationResultHolder holder = enabled.createAggregationResultHolder();
+ enabled.aggregate(2, holder,
block(SyntheticBlockValSets.Int.create(allNull(2), values)));
+ assertNull(enabled.extractAggregationResult(holder));
+ assertNull(enabled.extractFinalResult(null));
+
+ IdSetAggregationFunction disabled = idSet(false);
+ IdSet empty =
disabled.extractAggregationResult(disabled.createAggregationResultHolder());
+ assertNotNull(empty);
+ assertEquals(empty.getType(), IdSets.emptyIdSet().getType());
+ }
+
+ /// Every id of a multi-value row is collected.
+ @Test
+ public void testIdSetMVColumnCollectsEveryValue() {
+ int[][] rows = {{10, 11}, {99}, {30}};
+
+ IdSetAggregationFunction function = idSet(true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(3, holder,
block(SyntheticBlockValSets.IntMV.create(ROW1_NULL, rows)));
+
+ IdSet result = function.extractAggregationResult(holder);
+ assertNotNull(result);
+ assertTrue(result.contains(10));
+ assertTrue(result.contains(11));
+ assertTrue(result.contains(30));
+ assertFalse(result.contains(99));
+ }
+
+ /// A multi-value BYTES column was rejected before, although IdSets supports
BYTES and the single-value case
+ /// already worked.
+ @Test
+ public void testIdSetMVBytesColumn() {
+ byte[][][] rows = {{{1, 2}, {3, 4}}, {{9, 9}}};
+
+ IdSetAggregationFunction function = idSet(true);
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(2, holder,
block(SyntheticBlockValSets.BytesMV.create(ROW1_NULL, rows)));
+
+ IdSet result = function.extractAggregationResult(holder);
+ assertNotNull(result);
+ assertTrue(result.contains(new byte[]{1, 2}));
+ assertTrue(result.contains(new byte[]{3, 4}));
+ assertFalse(result.contains(new byte[]{9, 9}));
+ }
+
+ /// A row's ids land in that row's group, and a group whose only row is null
is never created.
+ @Test
+ public void testIdSetMVColumnGroupBySV() {
+ int[][] rows = {{10, 11}, {99}, {30}};
+
+ IdSetAggregationFunction function = idSet(true);
+ GroupByResultHolder holder = new ObjectGroupByResultHolder(2, 2);
+ function.aggregateGroupBySV(3, new int[]{0, 1, 0}, holder,
+ block(SyntheticBlockValSets.IntMV.create(ROW1_NULL, rows)));
+
+ IdSet group0 = function.extractGroupByResult(holder, 0);
+ assertNotNull(group0);
+ assertTrue(group0.contains(10));
+ assertTrue(group0.contains(30));
+ assertNull(function.extractGroupByResult(holder, 1));
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
