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 739e6a90d78 Support multi-value columns in DISTINCTCOUNTULL and
SEGMENTPARTITIONEDDISTINCTCOUNT (#19302)
739e6a90d78 is described below
commit 739e6a90d780cf10d58c5a2ef5a39ec4acdae917
Author: Xiaotian (Jackie) Jiang <[email protected]>
AuthorDate: Wed Aug 19 10:04:31 2026 -0700
Support multi-value columns in DISTINCTCOUNTULL and
SEGMENTPARTITIONEDDISTINCTCOUNT (#19302)
---
.../DistinctCountBitmapAggregationFunction.java | 6 +-
.../DistinctCountULLAggregationFunction.java | 269 ++++++++++++++++++-
...artitionedDistinctCountAggregationFunction.java | 286 ++++++++++++++++++++-
.../function/SumPrecisionAggregationFunction.java | 12 +-
.../pinot/core/common/SyntheticBlockValSets.java | 216 +++++++++++-----
.../DistinctCountULLAggregationFunctionTest.java | 110 +++++++-
...tionedDistinctCountAggregationFunctionTest.java | 120 +++++++++
.../tests/custom/UuidAggregationTest.java | 11 +-
8 files changed, 937 insertions(+), 93 deletions(-)
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java
index e71632673b2..dea8eb8695a 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapAggregationFunction.java
@@ -39,9 +39,9 @@ import org.roaringbitmap.PeekableIntIterator;
import org.roaringbitmap.RoaringBitmap;
-/// The `DistinctCountBitmapAggregationFunction` calculates the number of
distinct values for a given single-value
-/// expression using RoaringBitmap. The bitmap stores the actual values for
`INT` expression, or hash code of the
-/// values for other data types (values with the same hash code will only be
counted once).
+/// The `DistinctCountBitmapAggregationFunction` calculates the number of
distinct values for a given single-value or
+/// multi-value expression using RoaringBitmap. The bitmap stores the actual
values for `INT` expression, or hash code
+/// of the values for other data types (values with the same hash code will
only be counted once).
public class DistinctCountBitmapAggregationFunction extends
BaseSingleInputAggregationFunction<RoaringBitmap, Integer> {
public DistinctCountBitmapAggregationFunction(List<ExpressionContext>
arguments) {
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunction.java
index 7ef3142ddb5..6289ac0bead 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunction.java
@@ -105,6 +105,15 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
DataType storedType = dataType.getStoredType();
+ if (blockValSet.isSingleValue()) {
+ aggregateSV(length, aggregationResultHolder, blockValSet, storedType);
+ } else {
+ aggregateMV(length, aggregationResultHolder, blockValSet, storedType);
+ }
+ }
+
+ protected void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet,
+ DataType storedType) {
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -158,6 +167,76 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
}
}
+ protected void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet,
+ DataType storedType) {
+ // For dictionary-encoded expression, store dictionary ids into the bitmap
+ Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
+ if (dictionary != null) {
+ RoaringBitmap dictIdBitmap = getDictIdBitmap(aggregationResultHolder,
dictionary);
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ for (int i = 0; i < length; i++) {
+ dictIdBitmap.add(dictIds[i]);
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store values into the UltraLogLog
+ UltraLogLog ull = getULL(aggregationResultHolder);
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int value : intValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (long value : longValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (float value : floatValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (double value : doubleValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (String value : stringValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_ULL aggregation function: "
+ storedType);
+ }
+ }
+
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -186,6 +265,15 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
DataType storedType = dataType.getStoredType();
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet, storedType);
+ } else {
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder,
blockValSet, storedType);
+ }
+ }
+
+ protected void aggregateSVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet, DataType storedType) {
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -196,7 +284,7 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
return;
}
- // For non-dictionary-encoded expression, store values into the
HyperLogLogPlus
+ // For non-dictionary-encoded expression, store values into the UltraLogLog
switch (storedType) {
case INT:
int[] intValues = blockValSet.getIntValuesSV();
@@ -246,6 +334,80 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
}
}
+ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet, DataType storedType) {
+ // For dictionary-encoded expression, store dictionary ids into the bitmap
+ Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
+ if (dictionary != null) {
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ for (int i = 0; i < length; i++) {
+ getDictIdBitmap(groupByResultHolder, groupKeyArray[i],
dictionary).add(dictIds[i]);
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store values into the UltraLogLog
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKeyArray[i]);
+ for (int value : intValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKeyArray[i]);
+ for (long value : longValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKeyArray[i]);
+ for (float value : floatValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKeyArray[i]);
+ for (double value : doubleValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKeyArray[i]);
+ for (String value : stringValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKeyArray[i]);
+ for (byte[] value : bytesValues[i]) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_ULL aggregation function: "
+ storedType);
+ }
+ }
+
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -276,6 +438,15 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
DataType storedType = dataType.getStoredType();
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet, storedType);
+ } else {
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet, storedType);
+ }
+ }
+
+ protected void aggregateSVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet, DataType storedType) {
// For dictionary-encoded expression, store dictionary ids into the bitmap
Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
if (dictionary != null) {
@@ -326,7 +497,101 @@ public class DistinctCountULLAggregationFunction extends
BaseSingleInputAggregat
break;
default:
throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation
function: " + storedType);
+ "Illegal data type for DISTINCT_COUNT_ULL aggregation function: "
+ storedType);
+ }
+ }
+
+ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet, DataType storedType) {
+ // For dictionary-encoded expression, store dictionary ids into the bitmap
+ Dictionary dictionary = blockValSet.isDictionaryEncoded() ?
blockValSet.getDictionary() : null;
+ if (dictionary != null) {
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ getDictIdBitmap(groupByResultHolder, groupKey,
dictionary).add(dictIds[i]);
+ }
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store values into the UltraLogLog
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ int[] intRow = intValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKey);
+ for (int value : intRow) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ long[] longRow = longValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKey);
+ for (long value : longRow) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ float[] floatRow = floatValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKey);
+ for (float value : floatRow) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ double[] doubleRow = doubleValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKey);
+ for (double value : doubleRow) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ String[] stringRow = stringValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKey);
+ for (String value : stringRow) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ }
+ break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ byte[][] bytesRow = bytesValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ UltraLogLog ull = getULL(groupByResultHolder, groupKey);
+ for (byte[] value : bytesRow) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_ULL aggregation function: "
+ storedType);
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SegmentPartitionedDistinctCountAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SegmentPartitionedDistinctCountAggregationFunction.java
index 1acae7fc332..2a2f5706e69 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SegmentPartitionedDistinctCountAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SegmentPartitionedDistinctCountAggregationFunction.java
@@ -40,7 +40,7 @@ import org.roaringbitmap.RoaringBitmap;
/// The `SegmentPartitionedDistinctCountAggregationFunction` calculates the
number of distinct values for a given
-/// single-value expression.
+/// single-value or multi-value expression.
///
/// IMPORTANT: This function relies on the expression values being partitioned
for each segment, where there is no
/// common values within different segments.
@@ -72,7 +72,14 @@ public class
SegmentPartitionedDistinctCountAggregationFunction extends BaseSing
public void aggregate(int length, AggregationResultHolder
aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
+ if (blockValSet.isSingleValue()) {
+ aggregateSV(length, aggregationResultHolder, blockValSet);
+ } else {
+ aggregateMV(length, aggregationResultHolder, blockValSet);
+ }
+ }
+ protected void aggregateSV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
// For dictionary-encoded expression, store dictionary ids into a
RoaringBitmap
if (blockValSet.isDictionaryEncoded()) {
int[] dictIds = blockValSet.getDictionaryIdsSV();
@@ -159,11 +166,119 @@ public class
SegmentPartitionedDistinctCountAggregationFunction extends BaseSing
}
}
+ protected void aggregateMV(int length, AggregationResultHolder
aggregationResultHolder, BlockValSet blockValSet) {
+ // For dictionary-encoded expression, store dictionary ids into a
RoaringBitmap
+ if (blockValSet.isDictionaryEncoded()) {
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ RoaringBitmap bitmap = aggregationResultHolder.getResult();
+ if (bitmap == null) {
+ bitmap = new RoaringBitmap();
+ aggregationResultHolder.setValue(bitmap);
+ }
+ for (int i = 0; i < length; i++) {
+ bitmap.add(dictIds[i]);
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store INT values into a
RoaringBitmap, other types into an OpenHashSet
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ RoaringBitmap bitmap = aggregationResultHolder.getResult();
+ if (bitmap == null) {
+ bitmap = new RoaringBitmap();
+ aggregationResultHolder.setValue(bitmap);
+ }
+ for (int i = 0; i < length; i++) {
+ bitmap.add(intValues[i]);
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ LongOpenHashSet longSet = aggregationResultHolder.getResult();
+ if (longSet == null) {
+ longSet = new LongOpenHashSet();
+ aggregationResultHolder.setValue(longSet);
+ }
+ for (int i = 0; i < length; i++) {
+ for (long value : longValues[i]) {
+ longSet.add(value);
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ FloatOpenHashSet floatSet = aggregationResultHolder.getResult();
+ if (floatSet == null) {
+ floatSet = new FloatOpenHashSet();
+ aggregationResultHolder.setValue(floatSet);
+ }
+ for (int i = 0; i < length; i++) {
+ for (float value : floatValues[i]) {
+ floatSet.add(value);
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ DoubleOpenHashSet doubleSet = aggregationResultHolder.getResult();
+ if (doubleSet == null) {
+ doubleSet = new DoubleOpenHashSet();
+ aggregationResultHolder.setValue(doubleSet);
+ }
+ for (int i = 0; i < length; i++) {
+ for (double value : doubleValues[i]) {
+ doubleSet.add(value);
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ ObjectOpenHashSet<String> stringSet =
aggregationResultHolder.getResult();
+ if (stringSet == null) {
+ stringSet = new ObjectOpenHashSet<>();
+ aggregationResultHolder.setValue(stringSet);
+ }
+ for (int i = 0; i < length; i++) {
+ for (String value : stringValues[i]) {
+ stringSet.add(value);
+ }
+ }
+ break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ ObjectOpenHashSet<ByteArray> bytesSet =
aggregationResultHolder.getResult();
+ if (bytesSet == null) {
+ bytesSet = new ObjectOpenHashSet<>();
+ aggregationResultHolder.setValue(bytesSet);
+ }
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValues[i]) {
+ bytesSet.add(new ByteArray(value));
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for PARTITIONED_DISTINCT_COUNT aggregation
function: " + storedType);
+ }
+ }
+
@Override
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);
+ }
+ }
+ protected void aggregateSVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
// For dictionary-encoded expression, store dictionary ids into a
RoaringBitmap
if (blockValSet.isDictionaryEncoded()) {
int[] dictIds = blockValSet.getDictionaryIdsSV();
@@ -218,11 +333,89 @@ public class
SegmentPartitionedDistinctCountAggregationFunction extends BaseSing
}
}
+ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ // For dictionary-encoded expression, store dictionary ids into a
RoaringBitmap
+ if (blockValSet.isDictionaryEncoded()) {
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ for (int i = 0; i < length; i++) {
+ for (int dictId : dictIds[i]) {
+ setIntValueForGroup(groupByResultHolder, groupKeyArray[i], dictId);
+ }
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store INT values into a
RoaringBitmap, other types into an OpenHashSet
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int value : intValues[i]) {
+ setIntValueForGroup(groupByResultHolder, groupKeyArray[i], value);
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (long value : longValues[i]) {
+ setLongValueForGroup(groupByResultHolder, groupKeyArray[i], value);
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (float value : floatValues[i]) {
+ setFloatValueForGroup(groupByResultHolder, groupKeyArray[i],
value);
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (double value : doubleValues[i]) {
+ setDoubleValueForGroup(groupByResultHolder, groupKeyArray[i],
value);
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (String value : stringValues[i]) {
+ setStringValueForGroup(groupByResultHolder, groupKeyArray[i],
value);
+ }
+ }
+ break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValues[i]) {
+ setBytesValueForGroup(groupByResultHolder, groupKeyArray[i], new
ByteArray(value));
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for PARTITIONED_DISTINCT_COUNT aggregation
function: " + storedType);
+ }
+ }
+
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
+ if (blockValSet.isSingleValue()) {
+ aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ } else {
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder,
blockValSet);
+ }
+ }
+ protected void aggregateSVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
// For dictionary-encoded expression, store dictionary ids into a
RoaringBitmap
if (blockValSet.isDictionaryEncoded()) {
int[] dictIds = blockValSet.getDictionaryIdsSV();
@@ -298,6 +491,97 @@ public class
SegmentPartitionedDistinctCountAggregationFunction extends BaseSing
}
}
+ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ // For dictionary-encoded expression, store dictionary ids into a
RoaringBitmap
+ if (blockValSet.isDictionaryEncoded()) {
+ int[][] dictIds = blockValSet.getDictionaryIdsMV();
+ for (int i = 0; i < length; i++) {
+ int[] rowDictIds = dictIds[i];
+ for (int groupKey : groupKeysArray[i]) {
+ for (int dictId : rowDictIds) {
+ setIntValueForGroup(groupByResultHolder, groupKey, dictId);
+ }
+ }
+ }
+ return;
+ }
+
+ // For non-dictionary-encoded expression, store INT values into a
RoaringBitmap, other types into an OpenHashSet
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ int[] intRow = intValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ for (int value : intRow) {
+ setIntValueForGroup(groupByResultHolder, groupKey, value);
+ }
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ long[] longRow = longValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ for (long value : longRow) {
+ setLongValueForGroup(groupByResultHolder, groupKey, value);
+ }
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ float[] floatRow = floatValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ for (float value : floatRow) {
+ setFloatValueForGroup(groupByResultHolder, groupKey, value);
+ }
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ double[] doubleRow = doubleValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ for (double value : doubleRow) {
+ setDoubleValueForGroup(groupByResultHolder, groupKey, value);
+ }
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ String[] stringRow = stringValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ for (String value : stringRow) {
+ setStringValueForGroup(groupByResultHolder, groupKey, value);
+ }
+ }
+ }
+ break;
+ case BYTES:
+ byte[][][] bytesValues = blockValSet.getBytesValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (byte[] value : bytesValues[i]) {
+ ByteArray byteArray = new ByteArray(value);
+ for (int groupKey : groupKeysArray[i]) {
+ setBytesValueForGroup(groupByResultHolder, groupKey, byteArray);
+ }
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for PARTITIONED_DISTINCT_COUNT aggregation
function: " + storedType);
+ }
+ }
+
@Override
public Long extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
return extractIntermediateResult(aggregationResultHolder.getResult());
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunction.java
index 7ff32f1043d..92281afc763 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/SumPrecisionAggregationFunction.java
@@ -253,8 +253,9 @@ public class SumPrecisionAggregationFunction extends
NullableSingleInputAggregat
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
+ BigDecimal value = BigDecimal.valueOf(intValues[i]);
for (int groupKey : groupKeysArray[i]) {
- updateGroupByResult(groupKey, groupByResultHolder,
BigDecimal.valueOf(intValues[i]));
+ updateGroupByResult(groupKey, groupByResultHolder, value);
}
}
});
@@ -265,8 +266,9 @@ public class SumPrecisionAggregationFunction extends
NullableSingleInputAggregat
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
+ BigDecimal value = BigDecimal.valueOf(longValues[i]);
for (int groupKey : groupKeysArray[i]) {
- updateGroupByResult(groupKey, groupByResultHolder,
BigDecimal.valueOf(longValues[i]));
+ updateGroupByResult(groupKey, groupByResultHolder, value);
}
}
});
@@ -279,8 +281,9 @@ public class SumPrecisionAggregationFunction extends
NullableSingleInputAggregat
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
+ BigDecimal value = new BigDecimal(stringValues[i]);
for (int groupKey : groupKeysArray[i]) {
- updateGroupByResult(groupKey, groupByResultHolder, new
BigDecimal(stringValues[i]));
+ updateGroupByResult(groupKey, groupByResultHolder, value);
}
}
});
@@ -303,8 +306,9 @@ public class SumPrecisionAggregationFunction extends
NullableSingleInputAggregat
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
+ BigDecimal value = BigDecimalUtils.deserialize(bytesValues[i]);
for (int groupKey : groupKeysArray[i]) {
- updateGroupByResult(groupKey, groupByResultHolder,
BigDecimalUtils.deserialize(bytesValues[i]));
+ updateGroupByResult(groupKey, groupByResultHolder, value);
}
}
});
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 e5251f3337b..eeed7b9e1ba 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
@@ -27,9 +27,10 @@ import java.util.function.Supplier;
import javax.annotation.Nullable;
import org.apache.pinot.core.plan.DocIdSetPlanNode;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
-import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.roaringbitmap.RoaringBitmap;
+
/// Synthetic [BlockValSet] for testing and benchmarking.
public class SyntheticBlockValSets {
private SyntheticBlockValSets() {
@@ -138,6 +139,105 @@ public class SyntheticBlockValSets {
}
}
+ /// A [BlockValSet] for a dictionary-encoded multi-value column, which
exposes dictionary ids rather than values.
+ ///
+ /// Functions that collect dictionary ids take a different path from the one
that reads values, and resolve the ids
+ /// against the dictionary only when the result is extracted.
+ public static class DictIdsMV extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final int[][] _dictIds;
+ final Dictionary _dictionary;
+ final DataType _valueType;
+
+ private DictIdsMV(@Nullable RoaringBitmap nullBitmap, int[][] dictIds,
Dictionary dictionary, DataType valueType) {
+ _nullBitmap = nullBitmap;
+ _dictIds = dictIds;
+ _dictionary = dictionary;
+ _valueType = valueType;
+ }
+
+ public static DictIdsMV create(@Nullable RoaringBitmap nullBitmap, int[][]
dictIds, Dictionary dictionary,
+ DataType valueType) {
+ return new DictIdsMV(nullBitmap, dictIds, dictionary, valueType);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return _valueType;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return false;
+ }
+
+ @Nullable
+ @Override
+ public Dictionary getDictionary() {
+ return _dictionary;
+ }
+
+ @Override
+ public int[][] getDictionaryIdsMV() {
+ return _dictIds;
+ }
+ }
+
+ /// A simple [BlockValSet] for nullable, not dictionary-encoded int values.
+ public static class Int extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final int[] _values;
+
+ private Int(@Nullable RoaringBitmap nullBitmap, int[] values) {
+ _nullBitmap = nullBitmap;
+ _values = values;
+ }
+
+ public static Int create(int numDocs, @Nullable RoaringBitmap nullBitmap,
IntSupplier supplier) {
+ Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs, "null bitmap larger than numDocs");
+ int[] values = new int[numDocs];
+ for (int i = 0; i < numDocs; i++) {
+ values[i] = supplier.getAsInt();
+ }
+ return new Int(nullBitmap, values);
+ }
+
+ public static Int create(@Nullable RoaringBitmap nullBitmap, int[] values)
{
+ return new Int(nullBitmap, values);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return DataType.INT;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return true;
+ }
+
+ @Override
+ public int[] getIntValuesSV() {
+ return _values;
+ }
+ }
+
/// A simple [BlockValSet] for nullable, not dictionary-encoded long values.
public static class Long extends Base {
@@ -159,8 +259,7 @@ public class SyntheticBlockValSets {
}
public static Long create(int numDocs, @Nullable RoaringBitmap nullBitmap,
LongSupplier supplier) {
- Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs,
- "null bitmap larger than numDocs");
+ Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs, "null bitmap larger than numDocs");
long[] values = new long[numDocs];
for (int i = 0; i < numDocs; i++) {
values[i] = supplier.getAsLong();
@@ -180,8 +279,8 @@ public class SyntheticBlockValSets {
}
@Override
- public FieldSpec.DataType getValueType() {
- return FieldSpec.DataType.LONG;
+ public DataType getValueType() {
+ return DataType.LONG;
}
@Override
@@ -195,6 +294,44 @@ public class SyntheticBlockValSets {
}
}
+ /// A simple [BlockValSet] for nullable, not dictionary-encoded multi-value
long values.
+ public static class LongMV extends Base {
+
+ @Nullable
+ final RoaringBitmap _nullBitmap;
+ final long[][] _values;
+
+ private LongMV(@Nullable RoaringBitmap nullBitmap, long[][] values) {
+ _nullBitmap = nullBitmap;
+ _values = values;
+ }
+
+ public static LongMV create(@Nullable RoaringBitmap nullBitmap, long[][]
values) {
+ return new LongMV(nullBitmap, values);
+ }
+
+ @Nullable
+ @Override
+ public RoaringBitmap getNullBitmap() {
+ return _nullBitmap;
+ }
+
+ @Override
+ public DataType getValueType() {
+ return DataType.LONG;
+ }
+
+ @Override
+ public boolean isSingleValue() {
+ return false;
+ }
+
+ @Override
+ public long[][] getLongValuesMV() {
+ return _values;
+ }
+ }
+
/// A simple [BlockValSet] for nullable, not dictionary-encoded double
values.
public static class Double extends Base {
@@ -216,8 +353,7 @@ public class SyntheticBlockValSets {
}
public static Double create(int numDocs, @Nullable RoaringBitmap
nullBitmap, DoubleSupplier supplier) {
- Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs,
- "null bitmap larger than numDocs");
+ Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs, "null bitmap larger than numDocs");
double[] values = new double[numDocs];
for (int i = 0; i < numDocs; i++) {
values[i] = supplier.getAsDouble();
@@ -237,8 +373,8 @@ public class SyntheticBlockValSets {
}
@Override
- public FieldSpec.DataType getValueType() {
- return FieldSpec.DataType.DOUBLE;
+ public DataType getValueType() {
+ return DataType.DOUBLE;
}
@Override
@@ -269,8 +405,7 @@ public class SyntheticBlockValSets {
}
public static Str create(int numDocs, @Nullable RoaringBitmap nullBitmap,
Supplier<String> supplier) {
- Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs,
- "null bitmap larger than numDocs");
+ Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs, "null bitmap larger than numDocs");
String[] values = new String[numDocs];
for (int i = 0; i < numDocs; i++) {
values[i] = supplier.get();
@@ -289,8 +424,8 @@ public class SyntheticBlockValSets {
}
@Override
- public FieldSpec.DataType getValueType() {
- return FieldSpec.DataType.STRING;
+ public DataType getValueType() {
+ return DataType.STRING;
}
@Override
@@ -317,8 +452,7 @@ public class SyntheticBlockValSets {
}
public static Bytes create(int numDocs, @Nullable RoaringBitmap
nullBitmap, Supplier<byte[]> supplier) {
- Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs,
- "null bitmap larger than numDocs");
+ Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs, "null bitmap larger than numDocs");
byte[][] values = new byte[numDocs][];
for (int i = 0; i < numDocs; i++) {
values[i] = supplier.get();
@@ -337,8 +471,8 @@ public class SyntheticBlockValSets {
}
@Override
- public FieldSpec.DataType getValueType() {
- return FieldSpec.DataType.BYTES;
+ public DataType getValueType() {
+ return DataType.BYTES;
}
@Override
@@ -351,52 +485,4 @@ public class SyntheticBlockValSets {
return _values;
}
}
-
- /// A simple [BlockValSet] for nullable, not dictionary-encoded int values.
- public static class Int extends Base {
-
- @Nullable
- final RoaringBitmap _nullBitmap;
- final int[] _values;
-
- private Int(@Nullable RoaringBitmap nullBitmap, int[] values) {
- _nullBitmap = nullBitmap;
- _values = values;
- }
-
- public static Int create(int numDocs, @Nullable RoaringBitmap nullBitmap,
IntSupplier supplier) {
- Preconditions.checkArgument(nullBitmap == null || nullBitmap.last() <
numDocs,
- "null bitmap larger than numDocs");
- int[] values = new int[numDocs];
- for (int i = 0; i < numDocs; i++) {
- values[i] = supplier.getAsInt();
- }
- return new Int(nullBitmap, values);
- }
-
- public static Int create(@Nullable RoaringBitmap nullBitmap, int[] values)
{
- return new Int(nullBitmap, values);
- }
-
- @Nullable
- @Override
- public RoaringBitmap getNullBitmap() {
- return _nullBitmap;
- }
-
- @Override
- public FieldSpec.DataType getValueType() {
- return FieldSpec.DataType.INT;
- }
-
- @Override
- public boolean isSingleValue() {
- return true;
- }
-
- @Override
- public int[] getIntValuesSV() {
- return _values;
- }
- }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunctionTest.java
index 06c05b366bd..b0fcb6cd068 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountULLAggregationFunctionTest.java
@@ -22,10 +22,23 @@ 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.groupby.GroupByResultHolder;
+import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
import org.apache.pinot.segment.spi.Constants;
-import org.testng.Assert;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.testng.annotations.Test;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
public class DistinctCountULLAggregationFunctionTest {
@@ -34,18 +47,18 @@ public class DistinctCountULLAggregationFunctionTest {
DistinctCountULLAggregationFunction function = new
DistinctCountULLAggregationFunction(
List.of(ExpressionContext.forIdentifier("col")));
- Assert.assertTrue(function.canUseStarTree(Map.of()));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"12")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
12)));
-
Assert.assertFalse(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
16)));
+ assertTrue(function.canUseStarTree(Map.of()));
+ assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"12")));
+ assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
12)));
+ assertFalse(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
16)));
function = new
DistinctCountULLAggregationFunction(List.of(ExpressionContext.forIdentifier("col"),
ExpressionContext.forLiteral(Literal.intValue(12))));
- Assert.assertTrue(function.canUseStarTree(Map.of()));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"12")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
12)));
-
Assert.assertFalse(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"16")));
+ assertTrue(function.canUseStarTree(Map.of()));
+ assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"12")));
+ assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
12)));
+ assertFalse(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"16")));
}
@Test
@@ -53,9 +66,80 @@ public class DistinctCountULLAggregationFunctionTest {
DistinctCountULLAggregationFunction function = new
DistinctCountULLAggregationFunction(
List.of(ExpressionContext.forIdentifier("col"),
ExpressionContext.forLiteral(Literal.stringValue("16"))));
- Assert.assertFalse(function.canUseStarTree(Map.of()));
-
Assert.assertFalse(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"12")));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
16)));
-
Assert.assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"16")));
+ assertFalse(function.canUseStarTree(Map.of()));
+ assertFalse(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"12")));
+ assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
16)));
+ assertTrue(function.canUseStarTree(Map.of(Constants.HLLPLUS_ULL_P_KEY,
"16")));
+ }
+
+ private static final ExpressionContext COLUMN =
ExpressionContext.forIdentifier("column");
+ private static final long[][] MV_ROWS = {{1L, 2L}, {3L, 4L}, {5L, 6L}, {1L,
3L}};
+ private static final long[] FLATTENED = {1L, 2L, 3L, 4L, 5L, 6L, 1L, 3L};
+
+ private static DistinctCountULLAggregationFunction create() {
+ return new DistinctCountULLAggregationFunction(List.of(COLUMN));
+ }
+
+ private static Map<ExpressionContext, BlockValSet> mvBlock() {
+ return Map.of(COLUMN, SyntheticBlockValSets.LongMV.create(null, MV_ROWS));
+ }
+
+ /// Aggregating a multi-value column gives exactly what aggregating its
flattened values as a single-value column
+ /// gives. Comparing the two rather than asserting an estimate keeps this
exact for a sketch.
+ @Test
+ public void testMVColumnMatchesFlattenedSVColumn() {
+ DistinctCountULLAggregationFunction function = create();
+
+ AggregationResultHolder mvHolder =
function.createAggregationResultHolder();
+ function.aggregate(MV_ROWS.length, mvHolder, mvBlock());
+
+ AggregationResultHolder svHolder =
function.createAggregationResultHolder();
+ function.aggregate(FLATTENED.length, svHolder,
+ Map.of(COLUMN, SyntheticBlockValSets.Long.create(null, FLATTENED)));
+
+
assertEquals(function.extractFinalResult(function.extractAggregationResult(mvHolder)),
+
function.extractFinalResult(function.extractAggregationResult(svHolder)));
+ }
+
+ /// Every value of a row lands in that row's group.
+ @Test
+ public void testMVColumnGroupBySV() {
+ DistinctCountULLAggregationFunction function = create();
+ GroupByResultHolder resultHolder = new ObjectGroupByResultHolder(2, 2);
+ // Rows 0 and 3 to group 0, giving 1, 2, 1, 3; rows 1 and 2 to group 1,
giving 3, 4, 5, 6
+ function.aggregateGroupBySV(MV_ROWS.length, new int[]{0, 1, 1, 0},
resultHolder, mvBlock());
+
+
assertEquals(function.extractFinalResult(function.extractGroupByResult(resultHolder,
0)), 3L);
+
assertEquals(function.extractFinalResult(function.extractGroupByResult(resultHolder,
1)), 4L);
+ }
+
+ /// A row's values land in every group key that row carries.
+ @Test
+ public void testMVColumnGroupByMV() {
+ DistinctCountULLAggregationFunction function = create();
+ GroupByResultHolder resultHolder = new ObjectGroupByResultHolder(2, 2);
+ int[][] groupKeys = {{0, 1}, {0}, {1}, {0, 1}};
+ function.aggregateGroupByMV(MV_ROWS.length, groupKeys, resultHolder,
mvBlock());
+
+ // Group 0 sees rows 0, 1 and 3, giving 1, 2, 3, 4; group 1 sees rows 0, 2
and 3, giving 1, 2, 5, 6, 3
+
assertEquals(function.extractFinalResult(function.extractGroupByResult(resultHolder,
0)), 4L);
+
assertEquals(function.extractFinalResult(function.extractGroupByResult(resultHolder,
1)), 5L);
+ }
+
+ /// A dictionary-encoded multi-value column collects dictionary ids and
resolves them against the dictionary only
+ /// when the result is extracted, so it is a separate path from the raw
multi-value column above.
+ @Test
+ public void testDictionaryEncodedMVColumn() {
+ Dictionary dictionary = mock(Dictionary.class);
+ when(dictionary.get(anyInt())).thenAnswer(invocation -> (long) (int)
invocation.getArgument(0) + 1);
+ // Ids 0..5 stand for values 1..6, so these rows carry the same values as
MV_ROWS
+ int[][] dictIds = {{0, 1}, {2, 3}, {4, 5}, {0, 2}};
+
+ DistinctCountULLAggregationFunction function = create();
+ AggregationResultHolder resultHolder =
function.createAggregationResultHolder();
+ function.aggregate(dictIds.length, resultHolder,
+ Map.of(COLUMN, SyntheticBlockValSets.DictIdsMV.create(null, dictIds,
dictionary, DataType.LONG)));
+
+ assertEquals((long)
function.extractFinalResult(function.extractAggregationResult(resultHolder)),
6L);
}
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SegmentPartitionedDistinctCountAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SegmentPartitionedDistinctCountAggregationFunctionTest.java
new file mode 100644
index 00000000000..f2ba6e5e43a
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/SegmentPartitionedDistinctCountAggregationFunctionTest.java
@@ -0,0 +1,120 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.query.aggregation.function;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.SyntheticBlockValSets;
+import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
+import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
+import org.apache.pinot.segment.spi.index.reader.Dictionary;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.testng.annotations.Test;
+
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.testng.Assert.assertEquals;
+
+
+/// Multi-value column support for `SEGMENTPARTITIONEDDISTINCTCOUNT`.
+///
+/// This function used to read the single-value accessors unconditionally, so
a multi-value column was not something
+/// it could aggregate at all. It now branches on
`BlockValSet.isSingleValue()` the way the HyperLogLog and bitmap
+/// families already did. Its counts are exact, so the answers are asserted
directly rather than as estimates.
+public class SegmentPartitionedDistinctCountAggregationFunctionTest {
+ private static final ExpressionContext COLUMN =
ExpressionContext.forIdentifier("column");
+ private static final long[][] MV_ROWS = {{1L, 2L}, {3L, 4L}, {5L, 6L}, {1L,
3L}};
+ private static final long[] FLATTENED = {1L, 2L, 3L, 4L, 5L, 6L, 1L, 3L};
+
+ private static SegmentPartitionedDistinctCountAggregationFunction create() {
+ return new
SegmentPartitionedDistinctCountAggregationFunction(List.of(COLUMN));
+ }
+
+ private static Map<ExpressionContext, BlockValSet> mvBlock() {
+ return Map.of(COLUMN, SyntheticBlockValSets.LongMV.create(null, MV_ROWS));
+ }
+
+ /// Every value of every row is counted, with repeats collapsing.
+ @Test
+ public void testMVColumnCountsEveryValue() {
+ SegmentPartitionedDistinctCountAggregationFunction function = create();
+ AggregationResultHolder resultHolder =
function.createAggregationResultHolder();
+ function.aggregate(MV_ROWS.length, resultHolder, mvBlock());
+
+ // 1, 2, 3, 4, 5 and 6, with 1 and 3 seen twice
+
assertEquals(function.extractFinalResult(function.extractAggregationResult(resultHolder)).longValue(),
6L);
+ }
+
+ /// Every value of a row lands in that row's group.
+ @Test
+ public void testMVColumnGroupBySV() {
+ SegmentPartitionedDistinctCountAggregationFunction function = create();
+ GroupByResultHolder resultHolder = new ObjectGroupByResultHolder(2, 2);
+ // Rows 0 and 3 to group 0, giving 1, 2, 1, 3; rows 1 and 2 to group 1,
giving 3, 4, 5, 6
+ function.aggregateGroupBySV(MV_ROWS.length, new int[]{0, 1, 1, 0},
resultHolder, mvBlock());
+
+
assertEquals(function.extractFinalResult(function.extractGroupByResult(resultHolder,
0)).longValue(), 3L);
+
assertEquals(function.extractFinalResult(function.extractGroupByResult(resultHolder,
1)).longValue(), 4L);
+ }
+
+ /// A row's values land in every group key that row carries.
+ @Test
+ public void testMVColumnGroupByMV() {
+ SegmentPartitionedDistinctCountAggregationFunction function = create();
+ GroupByResultHolder resultHolder = new ObjectGroupByResultHolder(2, 2);
+ int[][] groupKeys = {{0, 1}, {0}, {1}, {0, 1}};
+ function.aggregateGroupByMV(MV_ROWS.length, groupKeys, resultHolder,
mvBlock());
+
+ // Group 0 sees rows 0, 1 and 3, giving 1, 2, 3, 4; group 1 sees rows 0, 2
and 3, giving 1, 2, 5, 6, 3
+
assertEquals(function.extractFinalResult(function.extractGroupByResult(resultHolder,
0)).longValue(), 4L);
+
assertEquals(function.extractFinalResult(function.extractGroupByResult(resultHolder,
1)).longValue(), 5L);
+ }
+
+ /// A single-value column still takes the single-value path, so the split
did not disturb it.
+ @Test
+ public void testSVColumnStillAggregates() {
+ SegmentPartitionedDistinctCountAggregationFunction function = create();
+ AggregationResultHolder resultHolder =
function.createAggregationResultHolder();
+ function.aggregate(FLATTENED.length, resultHolder,
+ Map.of(COLUMN, SyntheticBlockValSets.Long.create(null, FLATTENED)));
+
+
assertEquals(function.extractFinalResult(function.extractAggregationResult(resultHolder)).longValue(),
6L);
+ }
+
+ /// A dictionary-encoded multi-value column collects dictionary ids and
resolves them against the dictionary only
+ /// when the result is extracted, so it is a separate path from the raw
multi-value column above.
+ @Test
+ public void testDictionaryEncodedMVColumn() {
+ Dictionary dictionary = mock(Dictionary.class);
+ when(dictionary.get(anyInt())).thenAnswer(invocation -> (long) (int)
invocation.getArgument(0) + 1);
+ // Ids 0..5 stand for values 1..6, so these rows carry the same values as
MV_ROWS
+ int[][] dictIds = {{0, 1}, {2, 3}, {4, 5}, {0, 2}};
+
+ SegmentPartitionedDistinctCountAggregationFunction function = create();
+ AggregationResultHolder resultHolder =
function.createAggregationResultHolder();
+ function.aggregate(dictIds.length, resultHolder,
+ Map.of(COLUMN, SyntheticBlockValSets.DictIdsMV.create(null, dictIds,
dictionary, DataType.LONG)));
+
+
assertEquals(function.extractFinalResult(function.extractAggregationResult(resultHolder)).longValue(),
6L);
+ }
+}
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/UuidAggregationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/UuidAggregationTest.java
index eff7f109d90..c7c95487b78 100644
---
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/UuidAggregationTest.java
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/custom/UuidAggregationTest.java
@@ -162,16 +162,17 @@ public class UuidAggregationTest extends
CustomDataQueryClusterIntegrationTest {
throws Exception {
setUseMultiStageQueryEngine(false);
for (String function : List.of("DISTINCTCOUNT", "DISTINCTCOUNTHLL",
"DISTINCTCOUNTHLLPLUS",
- "DISTINCTCOUNTBITMAP", "DISTINCTCOUNTTHETASKETCH",
"DISTINCTCOUNTCPCSKETCH")) {
+ "DISTINCTCOUNTBITMAP", "DISTINCTCOUNTTHETASKETCH",
"DISTINCTCOUNTCPCSKETCH", "DISTINCTCOUNTULL")) {
JsonNode rows = query(String.format("SELECT %1$s(%2$s), %1$s(%3$s),
%1$s(%4$s) FROM %5$s", function,
UUID_DICT_SV_COLUMN, UUID_RAW_SV_COLUMN, UUID_RAW_MV_COLUMN,
getTableName()));
assertCounts(rows.get(0), 3L, 3L, 4L);
}
- // DISTINCTCOUNTULL currently supports only single-value inputs.
- JsonNode rows = query(String.format("SELECT DISTINCTCOUNTULL(%s),
DISTINCTCOUNTULL(%s) FROM %s",
- UUID_DICT_SV_COLUMN, UUID_RAW_SV_COLUMN, getTableName()));
- assertCounts(rows.get(0), 3L, 3L);
+ // A dictionary-encoded multi-value column collects dictionary ids rather
than reading values, which is a
+ // separate path from the raw multi-value column above
+ JsonNode rows =
+ query(String.format("SELECT DISTINCTCOUNTULL(%s) FROM %s",
UUID_DICT_MV_COLUMN, getTableName()));
+ assertCounts(rows.get(0), 4L);
rows = query(String.format(
"SELECT DISTINCTCOUNTTHETASKETCH(%1$s, '', '%1$s = ''%3$s''', '$1'), "
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
