You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by ya...@apache.org on 2026/02/05 01:00:11 UTC
(pinot) branch master updated: Consolidate MV aggregations into regular SV aggregations (#17519)
This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 d6ca0c78b86 Consolidate MV aggregations into regular SV aggregations (#17519)
d6ca0c78b86 is described below
commit d6ca0c78b866a316bdde86eeb4ff115a5b277c06
Author: Yash Mayya <ya...@gmail.com>
AuthorDate: Wed Feb 4 17:00:05 2026 -0800
Consolidate MV aggregations into regular SV aggregations (#17519)
---
.../DistinctCountBitmapAggregationFunction.java | 223 +++++++++++++++++++
.../DistinctCountBitmapMVAggregationFunction.java | 198 +----------------
.../DistinctCountHLLAggregationFunction.java | 233 ++++++++++++++++++++
.../DistinctCountHLLMVAggregationFunction.java | 212 +-----------------
.../DistinctCountHLLPlusAggregationFunction.java | 236 +++++++++++++++++++++
.../DistinctCountHLLPlusMVAggregationFunction.java | 212 +-----------------
.../function/PercentileAggregationFunction.java | 69 +++++-
.../function/PercentileEstAggregationFunction.java | 126 ++++++++---
.../PercentileEstMVAggregationFunction.java | 28 +--
.../function/PercentileKLLAggregationFunction.java | 112 ++++++++--
.../PercentileKLLMVAggregationFunction.java | 64 +-----
.../function/PercentileMVAggregationFunction.java | 28 +--
.../PercentileTDigestAggregationFunction.java | 128 ++++++++---
.../PercentileTDigestMVAggregationFunction.java | 28 +--
...stinctCountBitmapMVAggregationFunctionTest.java | 95 +++++++++
.../DistinctCountHLLMVAggregationFunctionTest.java | 95 +++++++++
...tinctCountHLLPlusMVAggregationFunctionTest.java | 95 +++++++++
.../PercentileEstMVAggregationFunctionTest.java | 95 +++++++++
.../PercentileKLLMVAggregationFunctionTest.java | 95 +++++++++
.../PercentileMVAggregationFunctionTest.java | 95 +++++++++
...PercentileTDigestMVAggregationFunctionTest.java | 95 +++++++++
.../DistinctCountBitmapWithMVStarTreeV2Test.java | 56 +++++
.../v2/DistinctCountHLLWithMVStarTreeV2Test.java | 57 +++++
.../v2/PercentileEstWithMVStarTreeV2Test.java | 61 ++++++
.../v2/PercentileTDigestWithMVStarTreeV2Test.java | 63 ++++++
.../local/aggregator/AvgValueAggregator.java | 4 +-
.../local/aggregator/CountMVValueAggregator.java | 11 +-
.../DistinctCountBitmapValueAggregator.java | 18 +-
.../DistinctCountHLLPlusValueAggregator.java | 18 +-
.../DistinctCountHLLValueAggregator.java | 18 +-
.../local/aggregator/MaxValueAggregator.java | 4 +-
.../aggregator/MinMaxRangeValueAggregator.java | 4 +-
.../local/aggregator/MinValueAggregator.java | 4 +-
.../aggregator/PercentileEstValueAggregator.java | 18 +-
.../PercentileTDigestValueAggregator.java | 18 +-
.../local/aggregator/SumValueAggregator.java | 4 +-
36 files changed, 2064 insertions(+), 856 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 a7cdb5e53e9..ea27bb872cd 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
@@ -91,6 +91,15 @@ public class DistinctCountBitmapAggregationFunction extends BaseSingleInputAggre
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -136,6 +145,66 @@ public class DistinctCountBitmapAggregationFunction extends BaseSingleInputAggre
}
}
+ protected void aggregateMV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet blockValSet,
+ DataType storedType) {
+ // For dictionary-encoded expression, store dictionary ids into the bitmap
+ Dictionary dictionary = blockValSet.getDictionary();
+ 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 hash code of the values into the bitmap
+ RoaringBitmap valueBitmap = getValueBitmap(aggregationResultHolder);
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ valueBitmap.add(intValues[i]);
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (long value : longValues[i]) {
+ valueBitmap.add(Long.hashCode(value));
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (float value : floatValues[i]) {
+ valueBitmap.add(Float.hashCode(value));
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (double value : doubleValues[i]) {
+ valueBitmap.add(Double.hashCode(value));
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (String value : stringValues[i]) {
+ valueBitmap.add(value.hashCode());
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
+ }
+ }
+
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -158,6 +227,15 @@ public class DistinctCountBitmapAggregationFunction extends BaseSingleInputAggre
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -206,6 +284,68 @@ public class DistinctCountBitmapAggregationFunction extends BaseSingleInputAggre
}
}
+ 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.getDictionary();
+ 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 hash code of the values into the bitmap
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ getValueBitmap(groupByResultHolder, groupKeyArray[i]).add(intValues[i]);
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
+ for (long value : longValues[i]) {
+ bitmap.add(Long.hashCode(value));
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
+ for (float value : floatValues[i]) {
+ bitmap.add(Float.hashCode(value));
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
+ for (double value : doubleValues[i]) {
+ bitmap.add(Double.hashCode(value));
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
+ for (String value : stringValues[i]) {
+ bitmap.add(value.hashCode());
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
+ }
+ }
+
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -230,6 +370,15 @@ public class DistinctCountBitmapAggregationFunction extends BaseSingleInputAggre
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -278,6 +427,80 @@ public class DistinctCountBitmapAggregationFunction extends BaseSingleInputAggre
}
}
+ 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.getDictionary();
+ 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 hash code of the values into the bitmap
+ switch (storedType) {
+ case INT:
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ getValueBitmap(groupByResultHolder, groupKey).add(intValues[i]);
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
+ for (long value : longValues[i]) {
+ bitmap.add(Long.hashCode(value));
+ }
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
+ for (float value : floatValues[i]) {
+ bitmap.add(Float.hashCode(value));
+ }
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
+ for (double value : doubleValues[i]) {
+ bitmap.add(Double.hashCode(value));
+ }
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
+ for (String value : stringValues[i]) {
+ bitmap.add(value.hashCode());
+ }
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_BITMAP aggregation function: " + storedType);
+ }
+ }
+
@Override
public RoaringBitmap extractAggregationResult(AggregationResultHolder aggregationResultHolder) {
Object result = aggregationResultHolder.getResult();
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapMVAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapMVAggregationFunction.java
index cb278ec5868..c56d988333d 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapMVAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapMVAggregationFunction.java
@@ -25,9 +25,6 @@ import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.segment.spi.AggregationFunctionType;
-import org.apache.pinot.segment.spi.index.reader.Dictionary;
-import org.apache.pinot.spi.data.FieldSpec.DataType;
-import org.roaringbitmap.RoaringBitmap;
/**
@@ -50,207 +47,22 @@ public class DistinctCountBitmapMVAggregationFunction extends DistinctCountBitma
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 hash code of the values into the bitmap
- RoaringBitmap valueBitmap = getValueBitmap(aggregationResultHolder);
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValues = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- valueBitmap.add(intValues[i]);
- }
- break;
- case LONG:
- long[][] longValues = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- for (long value : longValues[i]) {
- valueBitmap.add(Long.hashCode(value));
- }
- }
- break;
- case FLOAT:
- float[][] floatValues = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- for (float value : floatValues[i]) {
- valueBitmap.add(Float.hashCode(value));
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValues = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- for (double value : doubleValues[i]) {
- valueBitmap.add(Double.hashCode(value));
- }
- }
- break;
- case STRING:
- String[][] stringValues = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- for (String value : stringValues[i]) {
- valueBitmap.add(value.hashCode());
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_BITMAP_MV aggregation function: " + storedType);
- }
+ aggregateMV(length, aggregationResultHolder, blockValSet, blockValSet.getValueType().getStoredType());
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 hash code of the values into the bitmap
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValues = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- getValueBitmap(groupByResultHolder, groupKeyArray[i]).add(intValues[i]);
- }
- break;
- case LONG:
- long[][] longValues = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
- for (long value : longValues[i]) {
- bitmap.add(Long.hashCode(value));
- }
- }
- break;
- case FLOAT:
- float[][] floatValues = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
- for (float value : floatValues[i]) {
- bitmap.add(Float.hashCode(value));
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValues = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
- for (double value : doubleValues[i]) {
- bitmap.add(Double.hashCode(value));
- }
- }
- break;
- case STRING:
- String[][] stringValues = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKeyArray[i]);
- for (String value : stringValues[i]) {
- bitmap.add(value.hashCode());
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_BITMAP_MV aggregation function: " + storedType);
- }
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder, blockValSet,
+ blockValSet.getValueType().getStoredType());
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 hash code of the values into the bitmap
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValues = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- getValueBitmap(groupByResultHolder, groupKey).add(intValues[i]);
- }
- }
- break;
- case LONG:
- long[][] longValues = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
- for (long value : longValues[i]) {
- bitmap.add(Long.hashCode(value));
- }
- }
- }
- break;
- case FLOAT:
- float[][] floatValues = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
- for (float value : floatValues[i]) {
- bitmap.add(Float.hashCode(value));
- }
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValues = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
- for (double value : doubleValues[i]) {
- bitmap.add(Double.hashCode(value));
- }
- }
- }
- break;
- case STRING:
- String[][] stringValues = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- RoaringBitmap bitmap = getValueBitmap(groupByResultHolder, groupKey);
- for (String value : stringValues[i]) {
- bitmap.add(value.hashCode());
- }
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_BITMAP_MV aggregation function: " + storedType);
- }
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder, blockValSet,
+ blockValSet.getValueType().getStoredType());
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunction.java
index 9f7e1b34bcc..66748f390cc 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLAggregationFunction.java
@@ -103,6 +103,15 @@ public class DistinctCountHLLAggregationFunction extends BaseSingleInputAggregat
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -149,6 +158,67 @@ public class DistinctCountHLLAggregationFunction 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.getDictionary();
+ 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 HyperLogLog
+ HyperLogLog hyperLogLog = getHyperLogLog(aggregationResultHolder);
+ switch (storedType) {
+ case INT:
+ int[][] intValuesArray = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int value : intValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValuesArray = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (long value : longValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValuesArray = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (float value : floatValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (double value : doubleValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValuesArray = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (String value : stringValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType);
+ }
+ }
+
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -175,6 +245,15 @@ public class DistinctCountHLLAggregationFunction extends BaseSingleInputAggregat
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -222,6 +301,70 @@ public class DistinctCountHLLAggregationFunction 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.getDictionary();
+ 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 HyperLogLog
+ switch (storedType) {
+ case INT:
+ int[][] intValuesArray = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
+ for (int value : intValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValuesArray = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
+ for (long value : longValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValuesArray = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
+ for (float value : floatValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
+ for (double value : doubleValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValuesArray = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
+ for (String value : stringValuesArray[i]) {
+ hyperLogLog.offer(value);
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType);
+ }
+ }
+
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -251,6 +394,15 @@ public class DistinctCountHLLAggregationFunction extends BaseSingleInputAggregat
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -298,6 +450,87 @@ public class DistinctCountHLLAggregationFunction extends BaseSingleInputAggregat
}
}
+ 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.getDictionary();
+ 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 HyperLogLog
+ switch (storedType) {
+ case INT:
+ int[][] intValuesArray = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ int[] intValues = intValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
+ for (int value : intValues) {
+ hyperLogLog.offer(value);
+ }
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValuesArray = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ long[] longValues = longValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
+ for (long value : longValues) {
+ hyperLogLog.offer(value);
+ }
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValuesArray = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ float[] floatValues = floatValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
+ for (float value : floatValues) {
+ hyperLogLog.offer(value);
+ }
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ double[] doubleValues = doubleValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
+ for (double value : doubleValues) {
+ hyperLogLog.offer(value);
+ }
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValuesArray = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ String[] stringValues = stringValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
+ for (String value : stringValues) {
+ hyperLogLog.offer(value);
+ }
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException("Illegal data type for DISTINCT_COUNT_HLL aggregation function: " + storedType);
+ }
+ }
+
@Override
public HyperLogLog extractAggregationResult(AggregationResultHolder aggregationResultHolder) {
Object result = aggregationResultHolder.getResult();
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLMVAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLMVAggregationFunction.java
index deee3ba93d4..8a6a0f273f1 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLMVAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLMVAggregationFunction.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.query.aggregation.function;
-import com.clearspring.analytics.stream.cardinality.HyperLogLog;
import java.util.List;
import java.util.Map;
import org.apache.pinot.common.request.context.ExpressionContext;
@@ -26,9 +25,6 @@ import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.segment.spi.AggregationFunctionType;
-import org.apache.pinot.segment.spi.index.reader.Dictionary;
-import org.apache.pinot.spi.data.FieldSpec.DataType;
-import org.roaringbitmap.RoaringBitmap;
public class DistinctCountHLLMVAggregationFunction extends DistinctCountHLLAggregationFunction {
@@ -46,220 +42,22 @@ public class DistinctCountHLLMVAggregationFunction extends DistinctCountHLLAggre
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 HyperLogLog
- HyperLogLog hyperLogLog = getHyperLogLog(aggregationResultHolder);
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValuesArray = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- for (int value : intValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- case LONG:
- long[][] longValuesArray = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- for (long value : longValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- case FLOAT:
- float[][] floatValuesArray = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- for (float value : floatValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- for (double value : doubleValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- case STRING:
- String[][] stringValuesArray = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- for (String value : stringValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_HLL_MV aggregation function: " + storedType);
- }
+ aggregateMV(length, aggregationResultHolder, blockValSet, blockValSet.getValueType().getStoredType());
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 HyperLogLog
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValuesArray = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
- for (int value : intValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- case LONG:
- long[][] longValuesArray = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
- for (long value : longValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- case FLOAT:
- float[][] floatValuesArray = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
- for (float value : floatValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
- for (double value : doubleValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- case STRING:
- String[][] stringValuesArray = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKeyArray[i]);
- for (String value : stringValuesArray[i]) {
- hyperLogLog.offer(value);
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_HLL_MV aggregation function: " + storedType);
- }
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder, blockValSet,
+ blockValSet.getValueType().getStoredType());
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 HyperLogLog
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValuesArray = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- int[] intValues = intValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
- for (int value : intValues) {
- hyperLogLog.offer(value);
- }
- }
- }
- break;
- case LONG:
- long[][] longValuesArray = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- long[] longValues = longValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
- for (long value : longValues) {
- hyperLogLog.offer(value);
- }
- }
- }
- break;
- case FLOAT:
- float[][] floatValuesArray = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- float[] floatValues = floatValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
- for (float value : floatValues) {
- hyperLogLog.offer(value);
- }
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- double[] doubleValues = doubleValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
- for (double value : doubleValues) {
- hyperLogLog.offer(value);
- }
- }
- }
- break;
- case STRING:
- String[][] stringValuesArray = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- String[] stringValues = stringValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLog hyperLogLog = getHyperLogLog(groupByResultHolder, groupKey);
- for (String value : stringValues) {
- hyperLogLog.offer(value);
- }
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_HLL_MV aggregation function: " + storedType);
- }
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder, blockValSet,
+ blockValSet.getValueType().getStoredType());
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusAggregationFunction.java
index 9c7012bc4c2..25f555dc588 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusAggregationFunction.java
@@ -113,6 +113,15 @@ public class DistinctCountHLLPlusAggregationFunction extends BaseSingleInputAggr
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -160,6 +169,68 @@ public class DistinctCountHLLPlusAggregationFunction extends BaseSingleInputAggr
}
}
+ protected void aggregateMV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet blockValSet,
+ DataType storedType) {
+ // For dictionary-encoded expression, store dictionary ids into the bitmap
+ Dictionary dictionary = blockValSet.getDictionary();
+ 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 HyperLogLogPlus
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(aggregationResultHolder);
+ switch (storedType) {
+ case INT:
+ int[][] intValuesArray = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int value : intValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValuesArray = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (long value : longValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValuesArray = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (float value : floatValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (double value : doubleValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValuesArray = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (String value : stringValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation function: " + storedType);
+ }
+ }
+
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -186,6 +257,15 @@ public class DistinctCountHLLPlusAggregationFunction extends BaseSingleInputAggr
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -234,6 +314,71 @@ public class DistinctCountHLLPlusAggregationFunction extends BaseSingleInputAggr
}
}
+ 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.getDictionary();
+ 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 HyperLogLogPlus
+ switch (storedType) {
+ case INT:
+ int[][] intValuesArray = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
+ for (int value : intValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValuesArray = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
+ for (long value : longValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValuesArray = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
+ for (float value : floatValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
+ for (double value : doubleValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValuesArray = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
+ for (String value : stringValuesArray[i]) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation function: " + storedType);
+ }
+ }
+
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -263,6 +408,15 @@ public class DistinctCountHLLPlusAggregationFunction extends BaseSingleInputAggr
return;
}
+ 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.getDictionary();
if (dictionary != null) {
@@ -311,6 +465,88 @@ public class DistinctCountHLLPlusAggregationFunction extends BaseSingleInputAggr
}
}
+ 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.getDictionary();
+ 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 HyperLogLogPlus
+ switch (storedType) {
+ case INT:
+ int[][] intValuesArray = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ int[] intValues = intValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
+ for (int value : intValues) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ }
+ break;
+ case LONG:
+ long[][] longValuesArray = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ long[] longValues = longValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
+ for (long value : longValues) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ }
+ break;
+ case FLOAT:
+ float[][] floatValuesArray = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ float[] floatValues = floatValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
+ for (float value : floatValues) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ }
+ break;
+ case DOUBLE:
+ double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ double[] doubleValues = doubleValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
+ for (double value : doubleValues) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ }
+ break;
+ case STRING:
+ String[][] stringValuesArray = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ String[] stringValues = stringValuesArray[i];
+ for (int groupKey : groupKeysArray[i]) {
+ HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
+ for (String value : stringValues) {
+ hyperLogLogPlus.offer(value);
+ }
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException(
+ "Illegal data type for DISTINCT_COUNT_HLL_PLUS aggregation function: " + storedType);
+ }
+ }
+
@Override
public HyperLogLogPlus extractAggregationResult(AggregationResultHolder aggregationResultHolder) {
Object result = aggregationResultHolder.getResult();
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusMVAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusMVAggregationFunction.java
index 00abb1f5d28..890cf3cc3c4 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusMVAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusMVAggregationFunction.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.query.aggregation.function;
-import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
import java.util.List;
import java.util.Map;
import org.apache.pinot.common.request.context.ExpressionContext;
@@ -26,9 +25,6 @@ import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.segment.spi.AggregationFunctionType;
-import org.apache.pinot.segment.spi.index.reader.Dictionary;
-import org.apache.pinot.spi.data.FieldSpec.DataType;
-import org.roaringbitmap.RoaringBitmap;
public class DistinctCountHLLPlusMVAggregationFunction extends DistinctCountHLLPlusAggregationFunction {
@@ -46,220 +42,22 @@ public class DistinctCountHLLPlusMVAggregationFunction extends DistinctCountHLLP
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 HyperLogLog
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(aggregationResultHolder);
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValuesArray = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- for (int value : intValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- case LONG:
- long[][] longValuesArray = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- for (long value : longValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- case FLOAT:
- float[][] floatValuesArray = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- for (float value : floatValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- for (double value : doubleValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- case STRING:
- String[][] stringValuesArray = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- for (String value : stringValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_HLL_MV aggregation function: " + storedType);
- }
+ aggregateMV(length, aggregationResultHolder, blockValSet, blockValSet.getValueType().getStoredType());
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 HyperLogLog
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValuesArray = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
- for (int value : intValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- case LONG:
- long[][] longValuesArray = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
- for (long value : longValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- case FLOAT:
- float[][] floatValuesArray = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
- for (float value : floatValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
- for (double value : doubleValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- case STRING:
- String[][] stringValuesArray = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKeyArray[i]);
- for (String value : stringValuesArray[i]) {
- hyperLogLogPlus.offer(value);
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_HLL_MV aggregation function: " + storedType);
- }
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder, blockValSet,
+ blockValSet.getValueType().getStoredType());
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
-
- // For dictionary-encoded expression, store dictionary ids into the bitmap
- Dictionary dictionary = blockValSet.getDictionary();
- 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 HyperLogLog
- DataType storedType = blockValSet.getValueType().getStoredType();
- switch (storedType) {
- case INT:
- int[][] intValuesArray = blockValSet.getIntValuesMV();
- for (int i = 0; i < length; i++) {
- int[] intValues = intValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
- for (int value : intValues) {
- hyperLogLogPlus.offer(value);
- }
- }
- }
- break;
- case LONG:
- long[][] longValuesArray = blockValSet.getLongValuesMV();
- for (int i = 0; i < length; i++) {
- long[] longValues = longValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
- for (long value : longValues) {
- hyperLogLogPlus.offer(value);
- }
- }
- }
- break;
- case FLOAT:
- float[][] floatValuesArray = blockValSet.getFloatValuesMV();
- for (int i = 0; i < length; i++) {
- float[] floatValues = floatValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
- for (float value : floatValues) {
- hyperLogLogPlus.offer(value);
- }
- }
- }
- break;
- case DOUBLE:
- double[][] doubleValuesArray = blockValSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- double[] doubleValues = doubleValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
- for (double value : doubleValues) {
- hyperLogLogPlus.offer(value);
- }
- }
- }
- break;
- case STRING:
- String[][] stringValuesArray = blockValSet.getStringValuesMV();
- for (int i = 0; i < length; i++) {
- String[] stringValues = stringValuesArray[i];
- for (int groupKey : groupKeysArray[i]) {
- HyperLogLogPlus hyperLogLogPlus = getHyperLogLogPlus(groupByResultHolder, groupKey);
- for (String value : stringValues) {
- hyperLogLogPlus.offer(value);
- }
- }
- }
- break;
- default:
- throw new IllegalStateException(
- "Illegal data type for DISTINCT_COUNT_HLL_MV aggregation function: " + storedType);
- }
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder, blockValSet,
+ blockValSet.getValueType().getStoredType());
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java
index 3b1fcc09ef1..e87d860f93f 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileAggregationFunction.java
@@ -78,8 +78,16 @@ public class PercentileAggregationFunction extends NullableSingleInputAggregatio
@Override
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- DoubleArrayList valueList = getValueList(aggregationResultHolder);
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) {
+ DoubleArrayList valueList = getValueList(aggregationResultHolder);
double[] valueArray = blockValSet.getDoubleValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
@@ -88,10 +96,31 @@ public class PercentileAggregationFunction extends NullableSingleInputAggregatio
});
}
+ protected void aggregateMV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet blockValSet) {
+ DoubleArrayList valueList = getValueList(aggregationResultHolder);
+ double[][] valuesArray = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (double value : valuesArray[i]) {
+ valueList.add(value);
+ }
+ }
+ });
+ }
+
@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) {
double[] valueArray = blockValSet.getDoubleValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
@@ -101,10 +130,32 @@ public class PercentileAggregationFunction extends NullableSingleInputAggregatio
});
}
+ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ double[][] valuesArray = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ DoubleArrayList valueList = getValueList(groupByResultHolder, groupKeyArray[i]);
+ for (double value : valuesArray[i]) {
+ valueList.add(value);
+ }
+ }
+ });
+ }
+
@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) {
double[] valueArray = blockValSet.getDoubleValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
@@ -117,6 +168,22 @@ public class PercentileAggregationFunction extends NullableSingleInputAggregatio
});
}
+ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ 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]) {
+ DoubleArrayList valueList = getValueList(groupByResultHolder, groupKey);
+ for (double value : values) {
+ valueList.add(value);
+ }
+ }
+ }
+ });
+ }
+
@Override
public DoubleArrayList extractAggregationResult(AggregationResultHolder aggregationResultHolder) {
DoubleArrayList doubleArrayList = aggregationResultHolder.getResult();
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileEstAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileEstAggregationFunction.java
index 196c2afc87b..079cbebac85 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileEstAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileEstAggregationFunction.java
@@ -80,15 +80,7 @@ public class PercentileEstAggregationFunction extends NullableSingleInputAggrega
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- if (blockValSet.getValueType() != DataType.BYTES) {
- long[] longValues = blockValSet.getLongValuesSV();
- QuantileDigest quantileDigest = getDefaultQuantileDigest(aggregationResultHolder);
- forEachNotNull(length, blockValSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- quantileDigest.add(longValues[i]);
- }
- });
- } else {
+ if (blockValSet.getValueType() == DataType.BYTES) {
// Serialized QuantileDigest
byte[][] bytesValues = blockValSet.getBytesValuesSV();
foldNotNull(length, blockValSet, (QuantileDigest) aggregationResultHolder.getResult(), (quantile, from, toEx) -> {
@@ -107,21 +99,43 @@ public class PercentileEstAggregationFunction extends NullableSingleInputAggrega
}
return quantileDigest;
});
+ return;
}
+
+ if (blockValSet.isSingleValue()) {
+ aggregateSV(length, aggregationResultHolder, blockValSet);
+ } else {
+ aggregateMV(length, aggregationResultHolder, blockValSet);
+ }
+ }
+
+ protected void aggregateSV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet blockValSet) {
+ long[] longValues = blockValSet.getLongValuesSV();
+ QuantileDigest quantileDigest = getDefaultQuantileDigest(aggregationResultHolder);
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ quantileDigest.add(longValues[i]);
+ }
+ });
+ }
+
+ protected void aggregateMV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet blockValSet) {
+ long[][] valuesArray = blockValSet.getLongValuesMV();
+ QuantileDigest quantileDigest = getDefaultQuantileDigest(aggregationResultHolder);
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (long value : valuesArray[i]) {
+ quantileDigest.add(value);
+ }
+ }
+ });
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- if (blockValSet.getValueType() != DataType.BYTES) {
- long[] longValues = blockValSet.getLongValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- getDefaultQuantileDigest(groupByResultHolder, groupKeyArray[i]).add(longValues[i]);
- }
- });
- } else {
+ if (blockValSet.getValueType() == DataType.BYTES) {
// Serialized QuantileDigest
byte[][] bytesValues = blockValSet.getBytesValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
@@ -136,24 +150,44 @@ public class PercentileEstAggregationFunction extends NullableSingleInputAggrega
}
}
});
+ return;
+ }
+
+ 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) {
+ long[] longValues = blockValSet.getLongValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ getDefaultQuantileDigest(groupByResultHolder, groupKeyArray[i]).add(longValues[i]);
+ }
+ });
+ }
+
+ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ long[][] valuesArray = blockValSet.getLongValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ QuantileDigest quantileDigest = getDefaultQuantileDigest(groupByResultHolder, groupKeyArray[i]);
+ for (long value : valuesArray[i]) {
+ quantileDigest.add(value);
+ }
+ }
+ });
+ }
+
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- if (blockValSet.getValueType() != DataType.BYTES) {
- long[] longValues = blockValSet.getLongValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- long value = longValues[i];
- for (int groupKey : groupKeysArray[i]) {
- getDefaultQuantileDigest(groupByResultHolder, groupKey).add(value);
- }
- }
- });
- } else {
+ if (blockValSet.getValueType() == DataType.BYTES) {
// Serialized QuantileDigest
byte[][] bytesValues = blockValSet.getBytesValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
@@ -171,7 +205,43 @@ public class PercentileEstAggregationFunction extends NullableSingleInputAggrega
}
}
});
+ return;
}
+
+ 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) {
+ long[] longValues = blockValSet.getLongValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ long value = longValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ getDefaultQuantileDigest(groupByResultHolder, groupKey).add(value);
+ }
+ }
+ });
+ }
+
+ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ 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]) {
+ QuantileDigest quantileDigest = getDefaultQuantileDigest(groupByResultHolder, groupKey);
+ for (long value : values) {
+ quantileDigest.add(value);
+ }
+ }
+ }
+ });
}
@Override
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileEstMVAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileEstMVAggregationFunction.java
index 5a861714620..011cca6ee00 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileEstMVAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileEstMVAggregationFunction.java
@@ -23,7 +23,6 @@ import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
-import org.apache.pinot.segment.local.customobject.QuantileDigest;
import org.apache.pinot.segment.spi.AggregationFunctionType;
@@ -53,39 +52,18 @@ public class PercentileEstMVAggregationFunction extends PercentileEstAggregation
@Override
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- long[][] valuesArray = blockValSetMap.get(_expression).getLongValuesMV();
- QuantileDigest quantileDigest = getDefaultQuantileDigest(aggregationResultHolder);
- for (int i = 0; i < length; i++) {
- for (long value : valuesArray[i]) {
- quantileDigest.add(value);
- }
- }
+ aggregateMV(length, aggregationResultHolder, blockValSetMap.get(_expression));
}
@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++) {
- QuantileDigest quantileDigest = getDefaultQuantileDigest(groupByResultHolder, groupKeyArray[i]);
- for (long value : valuesArray[i]) {
- quantileDigest.add(value);
- }
- }
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder, blockValSetMap.get(_expression));
}
@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];
- for (int groupKey : groupKeysArray[i]) {
- QuantileDigest quantileDigest = getDefaultQuantileDigest(groupByResultHolder, groupKey);
- for (long value : values) {
- quantileDigest.add(value);
- }
- }
- }
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder, blockValSetMap.get(_expression));
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLAggregationFunction.java
index 5d8a7d52e91..21498595ad1 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLAggregationFunction.java
@@ -116,16 +116,38 @@ public class PercentileKLLAggregationFunction
sketch.merge(deserializedSketches[i]);
}
});
+ return;
+ }
+
+ if (valueSet.isSingleValue()) {
+ aggregateSV(length, aggregationResultHolder, valueSet, sketch);
} else {
- double[] values = valueSet.getDoubleValuesSV();
- forEachNotNull(length, valueSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- sketch.update(values[i]);
- }
- });
+ aggregateMV(length, aggregationResultHolder, valueSet, sketch);
}
}
+ protected void aggregateSV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet valueSet,
+ KllDoublesSketch sketch) {
+ double[] values = valueSet.getDoubleValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ sketch.update(values[i]);
+ }
+ });
+ }
+
+ protected void aggregateMV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet valueSet,
+ KllDoublesSketch sketch) {
+ double[][] values = valueSet.getDoubleValuesMV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (double val : values[i]) {
+ sketch.update(val);
+ }
+ }
+ });
+ }
+
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -141,17 +163,40 @@ public class PercentileKLLAggregationFunction
sketch.merge(deserializedSketches[i]);
}
});
+ return;
+ }
+
+ if (valueSet.isSingleValue()) {
+ aggregateSVGroupBySV(length, groupKeyArray, groupByResultHolder, valueSet);
} else {
- double[] values = valueSet.getDoubleValuesSV();
- forEachNotNull(length, valueSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKeyArray[i]);
- sketch.update(values[i]);
- }
- });
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder, valueSet);
}
}
+ protected void aggregateSVGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet valueSet) {
+ double[] values = valueSet.getDoubleValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKeyArray[i]);
+ sketch.update(values[i]);
+ }
+ });
+ }
+
+ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet valueSet) {
+ double[][] values = valueSet.getDoubleValuesMV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKeyArray[i]);
+ for (double val : values[i]) {
+ sketch.update(val);
+ }
+ }
+ });
+ }
+
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -169,17 +214,42 @@ public class PercentileKLLAggregationFunction
}
}
});
+ return;
+ }
+
+ if (valueSet.isSingleValue()) {
+ aggregateSVGroupByMV(length, groupKeysArray, groupByResultHolder, valueSet);
} else {
- double[] values = valueSet.getDoubleValuesSV();
- forEachNotNull(length, valueSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- for (int groupKey : groupKeysArray[i]) {
- KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKey);
- sketch.update(values[i]);
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder, valueSet);
+ }
+ }
+
+ protected void aggregateSVGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet valueSet) {
+ double[] values = valueSet.getDoubleValuesSV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKey);
+ sketch.update(values[i]);
+ }
+ }
+ });
+ }
+
+ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet valueSet) {
+ double[][] values = valueSet.getDoubleValuesMV();
+ forEachNotNull(length, valueSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (int groupKey : groupKeysArray[i]) {
+ KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKey);
+ for (double val : values[i]) {
+ sketch.update(val);
}
}
- });
- }
+ }
+ });
}
/**
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLMVAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLMVAggregationFunction.java
index 26af8dea447..af316e63e00 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLMVAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLMVAggregationFunction.java
@@ -20,13 +20,11 @@ package org.apache.pinot.core.query.aggregation.function;
import java.util.List;
import java.util.Map;
-import org.apache.datasketches.kll.KllDoublesSketch;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.segment.spi.AggregationFunctionType;
-import org.apache.pinot.spi.data.FieldSpec.DataType;
public class PercentileKLLMVAggregationFunction extends PercentileKLLAggregationFunction {
@@ -39,75 +37,19 @@ public class PercentileKLLMVAggregationFunction extends PercentileKLLAggregation
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet valueSet = blockValSetMap.get(_expression);
- DataType valueType = valueSet.getValueType();
- KllDoublesSketch sketch = getOrCreateSketch(aggregationResultHolder);
-
- if (valueType == DataType.BYTES) {
- // Assuming the column contains serialized data sketches
- KllDoublesSketch[] deserializedSketches = deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- for (int i = 0; i < length; i++) {
- sketch.merge(deserializedSketches[i]);
- }
- } else {
- double[][] values = valueSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- for (double val : values[i]) {
- sketch.update(val);
- }
- }
- }
+ aggregateMV(length, aggregationResultHolder, valueSet, getOrCreateSketch(aggregationResultHolder));
}
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- BlockValSet valueSet = blockValSetMap.get(_expression);
- DataType valueType = valueSet.getValueType();
-
- if (valueType == DataType.BYTES) {
- // serialized sketch
- KllDoublesSketch[] deserializedSketches = deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- for (int i = 0; i < length; i++) {
- KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKeyArray[i]);
- sketch.merge(deserializedSketches[i]);
- }
- } else {
- double[][] values = valueSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKeyArray[i]);
- for (double val : values[i]) {
- sketch.update(val);
- }
- }
- }
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder, blockValSetMap.get(_expression));
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- BlockValSet valueSet = blockValSetMap.get(_expression);
- DataType valueType = valueSet.getValueType();
-
- if (valueType == DataType.BYTES) {
- // serialized sketch
- KllDoublesSketch[] deserializedSketches = deserializeSketches(blockValSetMap.get(_expression).getBytesValuesSV());
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- KllDoublesSketch sketch = this.getOrCreateSketch(groupByResultHolder, groupKey);
- sketch.merge(deserializedSketches[i]);
- }
- }
- } else {
- double[][] values = valueSet.getDoubleValuesMV();
- for (int i = 0; i < length; i++) {
- for (int groupKey : groupKeysArray[i]) {
- KllDoublesSketch sketch = getOrCreateSketch(groupByResultHolder, groupKey);
- for (double val : values[i]) {
- sketch.update(val);
- }
- }
- }
- }
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder, blockValSetMap.get(_expression));
}
@Override
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileMVAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileMVAggregationFunction.java
index 620763ea759..8423a50e3a3 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileMVAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileMVAggregationFunction.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.query.aggregation.function;
-import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import java.util.Map;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.core.common.BlockValSet;
@@ -52,39 +51,18 @@ public class PercentileMVAggregationFunction extends PercentileAggregationFuncti
@Override
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- double[][] valuesArray = blockValSetMap.get(_expression).getDoubleValuesMV();
- DoubleArrayList valueList = getValueList(aggregationResultHolder);
- for (int i = 0; i < length; i++) {
- for (double value : valuesArray[i]) {
- valueList.add(value);
- }
- }
+ aggregateMV(length, aggregationResultHolder, blockValSetMap.get(_expression));
}
@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++) {
- DoubleArrayList valueList = getValueList(groupByResultHolder, groupKeyArray[i]);
- for (double value : valuesArray[i]) {
- valueList.add(value);
- }
- }
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder, blockValSetMap.get(_expression));
}
@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];
- for (int groupKey : groupKeysArray[i]) {
- DoubleArrayList valueList = getValueList(groupByResultHolder, groupKey);
- for (double value : values) {
- valueList.add(value);
- }
- }
- }
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder, blockValSetMap.get(_expression));
}
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java
index c840d434f52..799fdc75016 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestAggregationFunction.java
@@ -105,15 +105,7 @@ public class PercentileTDigestAggregationFunction extends NullableSingleInputAgg
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- if (blockValSet.getValueType() != DataType.BYTES) {
- double[] doubleValues = blockValSet.getDoubleValuesSV();
- TDigest tDigest = getDefaultTDigest(aggregationResultHolder, _compressionFactor);
- forEachNotNull(length, blockValSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- tDigest.add(doubleValues[i]);
- }
- });
- } else {
+ if (blockValSet.getValueType() == DataType.BYTES) {
// Serialized TDigest
byte[][] bytesValues = blockValSet.getBytesValuesSV();
foldNotNull(length, blockValSet, (TDigest) aggregationResultHolder.getResult(), (tDigest, from, toEx) -> {
@@ -130,21 +122,43 @@ public class PercentileTDigestAggregationFunction extends NullableSingleInputAgg
}
return tDigest;
});
+ return;
+ }
+
+ if (blockValSet.isSingleValue()) {
+ aggregateSV(length, aggregationResultHolder, blockValSet);
+ } else {
+ aggregateMV(length, aggregationResultHolder, blockValSet);
}
}
+ protected void aggregateSV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet blockValSet) {
+ double[] doubleValues = blockValSet.getDoubleValuesSV();
+ TDigest tDigest = getDefaultTDigest(aggregationResultHolder, _compressionFactor);
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ tDigest.add(doubleValues[i]);
+ }
+ });
+ }
+
+ protected void aggregateMV(int length, AggregationResultHolder aggregationResultHolder, BlockValSet blockValSet) {
+ double[][] valuesArray = blockValSet.getDoubleValuesMV();
+ TDigest tDigest = getDefaultTDigest(aggregationResultHolder, _compressionFactor);
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ for (double value : valuesArray[i]) {
+ tDigest.add(value);
+ }
+ }
+ });
+ }
+
@Override
public void aggregateGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- if (blockValSet.getValueType() != DataType.BYTES) {
- double[] doubleValues = blockValSet.getDoubleValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- getDefaultTDigest(groupByResultHolder, groupKeyArray[i], _compressionFactor).add(doubleValues[i]);
- }
- });
- } else {
+ if (blockValSet.getValueType() == DataType.BYTES) {
// Serialized TDigest
byte[][] bytesValues = blockValSet.getBytesValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
@@ -159,25 +173,45 @@ public class PercentileTDigestAggregationFunction extends NullableSingleInputAgg
}
}
});
+ return;
}
+
+ 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) {
+ double[] doubleValues = blockValSet.getDoubleValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ getDefaultTDigest(groupByResultHolder, groupKeyArray[i], _compressionFactor).add(doubleValues[i]);
+ }
+ });
+ }
+
+ protected void aggregateMVGroupBySV(int length, int[] groupKeyArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ double[][] valuesArray = blockValSet.getDoubleValuesMV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ TDigest tDigest = getDefaultTDigest(groupByResultHolder, groupKeyArray[i], _compressionFactor);
+ for (double value : valuesArray[i]) {
+ tDigest.add(value);
+ }
+ }
+ });
}
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
- if (blockValSet.getValueType() != DataType.BYTES) {
- double[] doubleValues = blockValSet.getDoubleValuesSV();
- forEachNotNull(length, blockValSet, (from, to) -> {
- for (int i = from; i < to; i++) {
- double value = doubleValues[i];
- for (int groupKey : groupKeysArray[i]) {
- getDefaultTDigest(groupByResultHolder, groupKey, _compressionFactor).add(value);
- }
- }
- });
- } else {
- // Serialized QuantileDigest
+ if (blockValSet.getValueType() == DataType.BYTES) {
+ // Serialized TDigest
byte[][] bytesValues = blockValSet.getBytesValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
@@ -193,7 +227,43 @@ public class PercentileTDigestAggregationFunction extends NullableSingleInputAgg
}
}
});
+ return;
}
+
+ 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) {
+ double[] doubleValues = blockValSet.getDoubleValuesSV();
+ forEachNotNull(length, blockValSet, (from, to) -> {
+ for (int i = from; i < to; i++) {
+ double value = doubleValues[i];
+ for (int groupKey : groupKeysArray[i]) {
+ getDefaultTDigest(groupByResultHolder, groupKey, _compressionFactor).add(value);
+ }
+ }
+ });
+ }
+
+ protected void aggregateMVGroupByMV(int length, int[][] groupKeysArray, GroupByResultHolder groupByResultHolder,
+ BlockValSet blockValSet) {
+ 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]) {
+ TDigest tDigest = getDefaultTDigest(groupByResultHolder, groupKey, _compressionFactor);
+ for (double value : values) {
+ tDigest.add(value);
+ }
+ }
+ }
+ });
}
@Override
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestMVAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestMVAggregationFunction.java
index a6b7884e6e8..6b4d20e00e1 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestMVAggregationFunction.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestMVAggregationFunction.java
@@ -18,7 +18,6 @@
*/
package org.apache.pinot.core.query.aggregation.function;
-import com.tdunning.math.stats.TDigest;
import java.util.Map;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.core.common.BlockValSet;
@@ -61,39 +60,18 @@ public class PercentileTDigestMVAggregationFunction extends PercentileTDigestAgg
@Override
public void aggregate(int length, AggregationResultHolder aggregationResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
- double[][] valuesArray = blockValSetMap.get(_expression).getDoubleValuesMV();
- TDigest tDigest = getDefaultTDigest(aggregationResultHolder, _compressionFactor);
- for (int i = 0; i < length; i++) {
- for (double value : valuesArray[i]) {
- tDigest.add(value);
- }
- }
+ aggregateMV(length, aggregationResultHolder, blockValSetMap.get(_expression));
}
@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++) {
- TDigest tDigest = getDefaultTDigest(groupByResultHolder, groupKeyArray[i], _compressionFactor);
- for (double value : valuesArray[i]) {
- tDigest.add(value);
- }
- }
+ aggregateMVGroupBySV(length, groupKeyArray, groupByResultHolder, blockValSetMap.get(_expression));
}
@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];
- for (int groupKey : groupKeysArray[i]) {
- TDigest tDigest = getDefaultTDigest(groupByResultHolder, groupKey, _compressionFactor);
- for (double value : values) {
- tDigest.add(value);
- }
- }
- }
+ aggregateMVGroupByMV(length, groupKeysArray, groupByResultHolder, blockValSetMap.get(_expression));
}
}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapMVAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapMVAggregationFunctionTest.java
new file mode 100644
index 00000000000..229d694e83e
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountBitmapMVAggregationFunctionTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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 org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.testng.annotations.Test;
+
+
+public class DistinctCountBitmapMVAggregationFunctionTest extends AbstractAggregationFunctionTest {
+
+ @Test
+ public void testAggregationMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.INT)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3;4"}
+ )
+ // Distinct values: 1, 2, 3, 4 = 4 distinct
+ .whenQuery("select distinctcountbitmap(mv) from testTable")
+ .thenResultIs("INTEGER", "4");
+ }
+
+ @Test
+ public void testAggregationMVGroupBySV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.INT)
+ .addSingleValueDimension("sv", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3", "k1"},
+ new Object[]{"4;5", "k2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3", "k1"},
+ new Object[]{"5;6", "k2"}
+ )
+ .whenQuery("select sv, distinctcountbitmap(mv) from testTable group by sv order by sv")
+ .thenResultIs("STRING | INTEGER",
+ "k1 | 3", // distinct: 1, 2, 3
+ "k2 | 3"); // distinct: 4, 5, 6
+ }
+
+ @Test
+ public void testAggregationMVGroupByMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("nums", FieldSpec.DataType.INT)
+ .addMultiValueDimension("tags", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ // Column order is alphabetical: nums, tags
+ new Object[]{"1;2", "tag1;tag2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3", "tag1;tag2"}
+ )
+ .whenQuery("select tags, distinctcountbitmap(nums) from testTable group by tags order by tags")
+ .thenResultIs("STRING | INTEGER",
+ "tag1 | 3", // distinct: 1, 2, 3
+ "tag2 | 3"); // distinct: 1, 2, 3
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLMVAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLMVAggregationFunctionTest.java
new file mode 100644
index 00000000000..fe24e95b946
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLMVAggregationFunctionTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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 org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.testng.annotations.Test;
+
+
+public class DistinctCountHLLMVAggregationFunctionTest extends AbstractAggregationFunctionTest {
+
+ @Test
+ public void testAggregationMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.INT)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3;4"}
+ )
+ // Distinct values: 1, 2, 3, 4 = 4 distinct
+ .whenQuery("select distinctcounthll(mv) from testTable")
+ .thenResultIs("LONG", "4");
+ }
+
+ @Test
+ public void testAggregationMVGroupBySV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.INT)
+ .addSingleValueDimension("sv", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3", "k1"},
+ new Object[]{"4;5", "k2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3", "k1"},
+ new Object[]{"5;6", "k2"}
+ )
+ .whenQuery("select sv, distinctcounthll(mv) from testTable group by sv order by sv")
+ .thenResultIs("STRING | LONG",
+ "k1 | 3", // distinct: 1, 2, 3
+ "k2 | 3"); // distinct: 4, 5, 6
+ }
+
+ @Test
+ public void testAggregationMVGroupByMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("nums", FieldSpec.DataType.INT)
+ .addMultiValueDimension("tags", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ // Column order is alphabetical: nums, tags
+ new Object[]{"1;2", "tag1;tag2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3", "tag1;tag2"}
+ )
+ .whenQuery("select tags, distinctcounthll(nums) from testTable group by tags order by tags")
+ .thenResultIs("STRING | LONG",
+ "tag1 | 3", // distinct: 1, 2, 3
+ "tag2 | 3"); // distinct: 1, 2, 3
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusMVAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusMVAggregationFunctionTest.java
new file mode 100644
index 00000000000..9a2ca5f86e7
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountHLLPlusMVAggregationFunctionTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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 org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.testng.annotations.Test;
+
+
+public class DistinctCountHLLPlusMVAggregationFunctionTest extends AbstractAggregationFunctionTest {
+
+ @Test
+ public void testAggregationMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.INT)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3;4"}
+ )
+ // Distinct values: 1, 2, 3, 4 = 4 distinct
+ .whenQuery("select distinctcounthllplus(mv) from testTable")
+ .thenResultIs("LONG", "4");
+ }
+
+ @Test
+ public void testAggregationMVGroupBySV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.INT)
+ .addSingleValueDimension("sv", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3", "k1"},
+ new Object[]{"4;5", "k2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3", "k1"},
+ new Object[]{"5;6", "k2"}
+ )
+ .whenQuery("select sv, distinctcounthllplus(mv) from testTable group by sv order by sv")
+ .thenResultIs("STRING | LONG",
+ "k1 | 3", // distinct: 1, 2, 3
+ "k2 | 3"); // distinct: 4, 5, 6
+ }
+
+ @Test
+ public void testAggregationMVGroupByMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("nums", FieldSpec.DataType.INT)
+ .addMultiValueDimension("tags", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ // Column order is alphabetical: nums, tags
+ new Object[]{"1;2", "tag1;tag2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"2;3", "tag1;tag2"}
+ )
+ .whenQuery("select tags, distinctcounthllplus(nums) from testTable group by tags order by tags")
+ .thenResultIs("STRING | LONG",
+ "tag1 | 3", // distinct: 1, 2, 3
+ "tag2 | 3"); // distinct: 1, 2, 3
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileEstMVAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileEstMVAggregationFunctionTest.java
new file mode 100644
index 00000000000..c77334f2286
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileEstMVAggregationFunctionTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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 org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.testng.annotations.Test;
+
+
+public class PercentileEstMVAggregationFunctionTest extends AbstractAggregationFunctionTest {
+
+ @Test
+ public void testAggregationMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.LONG)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3;4;5"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"6;7;8;9;10"}
+ )
+ // All values: 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 - p50 should be around 6
+ .whenQuery("select percentileest(mv, 50) from testTable")
+ .thenResultIs("LONG", "6");
+ }
+
+ @Test
+ public void testAggregationMVGroupBySV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.LONG)
+ .addSingleValueDimension("sv", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3;4;5", "k1"},
+ new Object[]{"10;20;30", "k2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"6;7;8;9;10", "k1"},
+ new Object[]{"40;50", "k2"}
+ )
+ .whenQuery("select sv, percentileest(mv, 50) from testTable group by sv order by sv")
+ .thenResultIs("STRING | LONG",
+ "k1 | 6", // values: 1-10, p50 ~= 6
+ "k2 | 30"); // values: 10, 20, 30, 40, 50, p50 ~= 30
+ }
+
+ @Test
+ public void testAggregationMVGroupByMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("nums", FieldSpec.DataType.LONG)
+ .addMultiValueDimension("tags", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ // Column order is alphabetical: nums, tags
+ new Object[]{"1;2;3", "tag1;tag2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"4;5;6", "tag1;tag2"}
+ )
+ .whenQuery("select tags, percentileest(nums, 50) from testTable group by tags order by tags")
+ .thenResultIs("STRING | LONG",
+ "tag1 | 4", // nums: 1, 2, 3, 4, 5, 6, p50 ~= 4
+ "tag2 | 4"); // nums: 1, 2, 3, 4, 5, 6, p50 ~= 4
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLMVAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLMVAggregationFunctionTest.java
new file mode 100644
index 00000000000..3303b97ac92
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileKLLMVAggregationFunctionTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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 org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.testng.annotations.Test;
+
+
+public class PercentileKLLMVAggregationFunctionTest extends AbstractAggregationFunctionTest {
+
+ @Test
+ public void testAggregationMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.DOUBLE)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1.0;2.0;3.0;4.0;5.0"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"6.0;7.0;8.0;9.0;10.0"}
+ )
+ // All values: 1-10, p50 should be around 5
+ .whenQuery("select percentilekll(mv, 50) from testTable")
+ .thenResultIs("DOUBLE", "5.0");
+ }
+
+ @Test
+ public void testAggregationMVGroupBySV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.DOUBLE)
+ .addSingleValueDimension("sv", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1.0;2.0;3.0;4.0;5.0", "k1"},
+ new Object[]{"10.0;20.0;30.0", "k2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"6.0;7.0;8.0;9.0;10.0", "k1"},
+ new Object[]{"40.0;50.0", "k2"}
+ )
+ .whenQuery("select sv, percentilekll(mv, 50) from testTable group by sv order by sv")
+ .thenResultIs("STRING | DOUBLE",
+ "k1 | 5.0", // values: 1-10, p50 ~= 5
+ "k2 | 30.0"); // values: 10, 20, 30, 40, 50, p50 ~= 30
+ }
+
+ @Test
+ public void testAggregationMVGroupByMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("nums", FieldSpec.DataType.DOUBLE)
+ .addMultiValueDimension("tags", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ // Column order is alphabetical: nums, tags
+ new Object[]{"1.0;2.0;3.0", "tag1;tag2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"4.0;5.0;6.0", "tag1;tag2"}
+ )
+ .whenQuery("select tags, percentilekll(nums, 50) from testTable group by tags order by tags")
+ .thenResultIs("STRING | DOUBLE",
+ "tag1 | 3.0", // nums: 1, 2, 3, 4, 5, 6, p50 ~= 3
+ "tag2 | 3.0"); // nums: 1, 2, 3, 4, 5, 6, p50 ~= 3
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileMVAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileMVAggregationFunctionTest.java
new file mode 100644
index 00000000000..a246f37b1bf
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileMVAggregationFunctionTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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 org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.testng.annotations.Test;
+
+
+public class PercentileMVAggregationFunctionTest extends AbstractAggregationFunctionTest {
+
+ @Test
+ public void testAggregationMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.INT)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3;4;5"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"6;7;8;9;10"}
+ )
+ // All values: 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 - p50 should be 6
+ .whenQuery("select percentile(mv, 50) from testTable")
+ .thenResultIs("DOUBLE", "6.0");
+ }
+
+ @Test
+ public void testAggregationMVGroupBySV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.INT)
+ .addSingleValueDimension("sv", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1;2;3;4;5", "k1"},
+ new Object[]{"10;20;30", "k2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"6;7;8;9;10", "k1"},
+ new Object[]{"40;50", "k2"}
+ )
+ .whenQuery("select sv, percentile(mv, 50) from testTable group by sv order by sv")
+ .thenResultIs("STRING | DOUBLE",
+ "k1 | 6.0", // values: 1-10, p50 = 6
+ "k2 | 30.0"); // values: 10, 20, 30, 40, 50, p50 = 30
+ }
+
+ @Test
+ public void testAggregationMVGroupByMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("nums", FieldSpec.DataType.INT)
+ .addMultiValueDimension("tags", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ // Column order is alphabetical: nums, tags
+ new Object[]{"1;2;3", "tag1;tag2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"4;5;6", "tag1;tag2"}
+ )
+ .whenQuery("select tags, percentile(nums, 50) from testTable group by tags order by tags")
+ .thenResultIs("STRING | DOUBLE",
+ "tag1 | 4.0", // nums: 1, 2, 3, 4, 5, 6, p50 = 4
+ "tag2 | 4.0"); // nums: 1, 2, 3, 4, 5, 6, p50 = 4
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestMVAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestMVAggregationFunctionTest.java
new file mode 100644
index 00000000000..59decfbaadd
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileTDigestMVAggregationFunctionTest.java
@@ -0,0 +1,95 @@
+/**
+ * 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 org.apache.pinot.queries.FluentQueryTest;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.testng.annotations.Test;
+
+
+public class PercentileTDigestMVAggregationFunctionTest extends AbstractAggregationFunctionTest {
+
+ @Test
+ public void testAggregationMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.DOUBLE)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1.0;2.0;3.0;4.0;5.0"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"6.0;7.0;8.0;9.0;10.0"}
+ )
+ // All values: 1-10, p50 should be around 5
+ .whenQuery("select percentiletdigest(mv, 50) from testTable")
+ .thenResultIs("DOUBLE", "5.5");
+ }
+
+ @Test
+ public void testAggregationMVGroupBySV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("mv", FieldSpec.DataType.DOUBLE)
+ .addSingleValueDimension("sv", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ new Object[]{"1.0;2.0;3.0;4.0;5.0", "k1"},
+ new Object[]{"10.0;20.0;30.0", "k2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"6.0;7.0;8.0;9.0;10.0", "k1"},
+ new Object[]{"40.0;50.0", "k2"}
+ )
+ .whenQuery("select sv, percentiletdigest(mv, 50) from testTable group by sv order by sv")
+ .thenResultIs("STRING | DOUBLE",
+ "k1 | 5.5", // values: 1-10, p50 ~= 5.5
+ "k2 | 30.0"); // values: 10, 20, 30, 40, 50, p50 ~= 30
+ }
+
+ @Test
+ public void testAggregationMVGroupByMV() {
+ FluentQueryTest.withBaseDir(_baseDir)
+ .givenTable(
+ new Schema.SchemaBuilder()
+ .setSchemaName("testTable")
+ .setEnableColumnBasedNullHandling(true)
+ .addMultiValueDimension("nums", FieldSpec.DataType.DOUBLE)
+ .addMultiValueDimension("tags", FieldSpec.DataType.STRING)
+ .build(), SINGLE_FIELD_TABLE_CONFIG)
+ .onFirstInstance(
+ // Column order is alphabetical: nums, tags
+ new Object[]{"1.0;2.0;3.0", "tag1;tag2"}
+ )
+ .andOnSecondInstance(
+ new Object[]{"4.0;5.0;6.0", "tag1;tag2"}
+ )
+ .whenQuery("select tags, percentiletdigest(nums, 50) from testTable group by tags order by tags")
+ .thenResultIs("STRING | DOUBLE",
+ "tag1 | 3.5", // nums: 1, 2, 3, 4, 5, 6, p50 ~= 3.5
+ "tag2 | 3.5"); // nums: 1, 2, 3, 4, 5, 6, p50 ~= 3.5
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/DistinctCountBitmapWithMVStarTreeV2Test.java b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/DistinctCountBitmapWithMVStarTreeV2Test.java
new file mode 100644
index 00000000000..b58a9d31cc6
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/DistinctCountBitmapWithMVStarTreeV2Test.java
@@ -0,0 +1,56 @@
+/**
+ * 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.startree.v2;
+
+import java.util.Random;
+import org.apache.pinot.segment.local.aggregator.DistinctCountBitmapValueAggregator;
+import org.apache.pinot.segment.local.aggregator.ValueAggregator;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.roaringbitmap.RoaringBitmap;
+
+import static org.testng.Assert.assertEquals;
+
+
+public class DistinctCountBitmapWithMVStarTreeV2Test extends BaseStarTreeV2Test<Object, RoaringBitmap> {
+
+ @Override
+ ValueAggregator<Object, RoaringBitmap> getValueAggregator() {
+ return new DistinctCountBitmapValueAggregator();
+ }
+
+ @Override
+ DataType getRawValueType() {
+ return DataType.INT;
+ }
+
+ @Override
+ Object getRandomRawValue(Random random) {
+ return random.nextInt(100);
+ }
+
+ @Override
+ void assertAggregatedValue(RoaringBitmap starTreeResult, RoaringBitmap nonStarTreeResult) {
+ assertEquals(starTreeResult, nonStarTreeResult);
+ }
+
+ @Override
+ protected boolean isAggColSingleValueField() {
+ return false;
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/DistinctCountHLLWithMVStarTreeV2Test.java b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/DistinctCountHLLWithMVStarTreeV2Test.java
new file mode 100644
index 00000000000..3925c93f89c
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/DistinctCountHLLWithMVStarTreeV2Test.java
@@ -0,0 +1,57 @@
+/**
+ * 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.startree.v2;
+
+import com.clearspring.analytics.stream.cardinality.HyperLogLog;
+import java.util.Collections;
+import java.util.Random;
+import org.apache.pinot.segment.local.aggregator.DistinctCountHLLValueAggregator;
+import org.apache.pinot.segment.local.aggregator.ValueAggregator;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+
+import static org.testng.Assert.assertEquals;
+
+
+public class DistinctCountHLLWithMVStarTreeV2Test extends BaseStarTreeV2Test<Object, HyperLogLog> {
+
+ @Override
+ ValueAggregator<Object, HyperLogLog> getValueAggregator() {
+ return new DistinctCountHLLValueAggregator(Collections.emptyList());
+ }
+
+ @Override
+ DataType getRawValueType() {
+ return DataType.INT;
+ }
+
+ @Override
+ Object getRandomRawValue(Random random) {
+ return random.nextInt(100);
+ }
+
+ @Override
+ void assertAggregatedValue(HyperLogLog starTreeResult, HyperLogLog nonStarTreeResult) {
+ assertEquals(starTreeResult.cardinality(), nonStarTreeResult.cardinality());
+ }
+
+ @Override
+ protected boolean isAggColSingleValueField() {
+ return false;
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/PercentileEstWithMVStarTreeV2Test.java b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/PercentileEstWithMVStarTreeV2Test.java
new file mode 100644
index 00000000000..61b7531442e
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/PercentileEstWithMVStarTreeV2Test.java
@@ -0,0 +1,61 @@
+/**
+ * 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.startree.v2;
+
+import java.util.Random;
+import org.apache.pinot.core.query.aggregation.function.PercentileEstAggregationFunction;
+import org.apache.pinot.segment.local.aggregator.PercentileEstValueAggregator;
+import org.apache.pinot.segment.local.aggregator.ValueAggregator;
+import org.apache.pinot.segment.local.customobject.QuantileDigest;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+
+import static org.testng.Assert.assertEquals;
+
+
+public class PercentileEstWithMVStarTreeV2Test extends BaseStarTreeV2Test<Object, QuantileDigest> {
+ private static final int MAX_VALUE = 10000;
+
+ @Override
+ ValueAggregator<Object, QuantileDigest> getValueAggregator() {
+ return new PercentileEstValueAggregator();
+ }
+
+ @Override
+ DataType getRawValueType() {
+ return DataType.LONG;
+ }
+
+ @Override
+ Object getRandomRawValue(Random random) {
+ return random.nextInt(MAX_VALUE);
+ }
+
+ @Override
+ void assertAggregatedValue(QuantileDigest starTreeResult, QuantileDigest nonStarTreeResult) {
+ double delta = MAX_VALUE * PercentileEstAggregationFunction.DEFAULT_MAX_ERROR;
+ for (int i = 0; i <= 100; i++) {
+ assertEquals(starTreeResult.getQuantile(i / 100.0), nonStarTreeResult.getQuantile(i / 100.0), delta);
+ }
+ }
+
+ @Override
+ protected boolean isAggColSingleValueField() {
+ return false;
+ }
+}
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/PercentileTDigestWithMVStarTreeV2Test.java b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/PercentileTDigestWithMVStarTreeV2Test.java
new file mode 100644
index 00000000000..3c8dd749540
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/startree/v2/PercentileTDigestWithMVStarTreeV2Test.java
@@ -0,0 +1,63 @@
+/**
+ * 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.startree.v2;
+
+import com.tdunning.math.stats.TDigest;
+import java.util.Collections;
+import java.util.Random;
+import org.apache.pinot.segment.local.aggregator.PercentileTDigestValueAggregator;
+import org.apache.pinot.segment.local.aggregator.ValueAggregator;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+
+import static org.testng.Assert.assertEquals;
+
+
+public class PercentileTDigestWithMVStarTreeV2Test extends BaseStarTreeV2Test<Object, TDigest> {
+ private static final int MAX_VALUE = 10000;
+ private static final double DELTA = 0.05;
+
+ @Override
+ ValueAggregator<Object, TDigest> getValueAggregator() {
+ return new PercentileTDigestValueAggregator(Collections.emptyList());
+ }
+
+ @Override
+ DataType getRawValueType() {
+ return DataType.DOUBLE;
+ }
+
+ @Override
+ Object getRandomRawValue(Random random) {
+ return random.nextDouble() * MAX_VALUE;
+ }
+
+ @Override
+ void assertAggregatedValue(TDigest starTreeResult, TDigest nonStarTreeResult) {
+ for (int i = 0; i <= 100; i++) {
+ double percentile = i / 100.0;
+ assertEquals(starTreeResult.quantile(percentile), nonStarTreeResult.quantile(percentile),
+ MAX_VALUE * DELTA, "Percentile " + i + " did not match");
+ }
+ }
+
+ @Override
+ protected boolean isAggColSingleValueField() {
+ return false;
+ }
+}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/AvgValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/AvgValueAggregator.java
index 3bd007af5d0..6810ec80324 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/AvgValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/AvgValueAggregator.java
@@ -96,9 +96,7 @@ public class AvgValueAggregator implements ValueAggregator<Object, AvgPair> {
Object[] values = (Object[]) rawValue;
AvgPair avgPair = new AvgPair();
for (Object value : values) {
- if (value != null) {
- avgPair.apply(ValueAggregatorUtils.toDouble(value));
- }
+ avgPair.apply(ValueAggregatorUtils.toDouble(value));
}
return avgPair;
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/CountMVValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/CountMVValueAggregator.java
index 7b26515fb18..ea7584502d5 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/CountMVValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/CountMVValueAggregator.java
@@ -49,19 +49,12 @@ public class CountMVValueAggregator extends CountValueAggregator {
}
/**
- * Processes a multi-value array and returns the count of non-null values.
+ * Processes a multi-value array and returns the count of values.
* The rawValue can be an Object[] array containing values.
*/
private Long processMultiValueArray(Object rawValue) {
if (rawValue instanceof Object[]) {
- Object[] values = (Object[]) rawValue;
- long count = 0;
- for (Object value : values) {
- if (value != null) {
- count++;
- }
- }
- return count;
+ return (long) ((Object[]) rawValue).length;
} else {
return 1L;
}
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountBitmapValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountBitmapValueAggregator.java
index 64cceab6bb5..9aa7fd87aaf 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountBitmapValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountBitmapValueAggregator.java
@@ -50,7 +50,7 @@ public class DistinctCountBitmapValueAggregator implements ValueAggregator<Objec
_maxByteSize = Math.max(_maxByteSize, bytes.length);
} else {
initialValue = new RoaringBitmap();
- initialValue.add(rawValue.hashCode());
+ addToValue(initialValue, rawValue);
_maxByteSize = Math.max(_maxByteSize, initialValue.serializedSizeInBytes());
}
return initialValue;
@@ -61,12 +61,26 @@ public class DistinctCountBitmapValueAggregator implements ValueAggregator<Objec
if (rawValue instanceof byte[]) {
value.or(deserializeAggregatedValue((byte[]) rawValue));
} else {
- value.add(rawValue.hashCode());
+ addToValue(value, rawValue);
}
_maxByteSize = Math.max(_maxByteSize, value.serializedSizeInBytes());
return value;
}
+ /**
+ * Adds a raw value (single value or multi-value array) to the RoaringBitmap.
+ */
+ protected void addToValue(RoaringBitmap bitmap, Object rawValue) {
+ if (rawValue instanceof Object[]) {
+ Object[] values = (Object[]) rawValue;
+ for (Object value : values) {
+ bitmap.add(value.hashCode());
+ }
+ } else {
+ bitmap.add(rawValue.hashCode());
+ }
+ }
+
@Override
public RoaringBitmap applyAggregatedValue(RoaringBitmap value, RoaringBitmap aggregatedValue) {
value.or(aggregatedValue);
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregator.java
index 6c1a50e5c57..93ff8feb95b 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLPlusValueAggregator.java
@@ -76,7 +76,7 @@ public class DistinctCountHLLPlusValueAggregator implements ValueAggregator<Obje
_maxByteSize = bytes.length;
} else {
initialValue = new HyperLogLogPlus(_p, _sp);
- initialValue.offer(rawValue);
+ offerValue(initialValue, rawValue);
_maxByteSize = HyperLogLogPlusUtils.byteSize(_p, _sp);
}
return initialValue;
@@ -91,11 +91,25 @@ public class DistinctCountHLLPlusValueAggregator implements ValueAggregator<Obje
throw new RuntimeException(e);
}
} else {
- value.offer(rawValue);
+ offerValue(value, rawValue);
}
return value;
}
+ /**
+ * Offers a raw value (single value or multi-value array) to the HyperLogLogPlus.
+ */
+ protected void offerValue(HyperLogLogPlus hll, Object rawValue) {
+ if (rawValue instanceof Object[]) {
+ Object[] values = (Object[]) rawValue;
+ for (Object value : values) {
+ hll.offer(value);
+ }
+ } else {
+ hll.offer(rawValue);
+ }
+ }
+
@Override
public HyperLogLogPlus applyAggregatedValue(HyperLogLogPlus value, HyperLogLogPlus aggregatedValue) {
try {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLValueAggregator.java
index 6414231f913..eccb03c83fa 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/DistinctCountHLLValueAggregator.java
@@ -70,7 +70,7 @@ public class DistinctCountHLLValueAggregator implements ValueAggregator<Object,
_maxByteSize = bytes.length;
} else {
initialValue = new HyperLogLog(_log2m);
- initialValue.offer(rawValue);
+ offerValue(initialValue, rawValue);
_maxByteSize = HyperLogLogUtils.byteSize(initialValue);
}
return initialValue;
@@ -85,11 +85,25 @@ public class DistinctCountHLLValueAggregator implements ValueAggregator<Object,
throw new RuntimeException(e);
}
} else {
- value.offer(rawValue);
+ offerValue(value, rawValue);
}
return value;
}
+ /**
+ * Offers a raw value (single value or multi-value array) to the HyperLogLog.
+ */
+ protected void offerValue(HyperLogLog hll, Object rawValue) {
+ if (rawValue instanceof Object[]) {
+ Object[] values = (Object[]) rawValue;
+ for (Object value : values) {
+ hll.offer(value);
+ }
+ } else {
+ hll.offer(rawValue);
+ }
+ }
+
@Override
public HyperLogLog applyAggregatedValue(HyperLogLog value, HyperLogLog aggregatedValue) {
try {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MaxValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MaxValueAggregator.java
index d93fd7a9f72..8305c8bb2db 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MaxValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MaxValueAggregator.java
@@ -84,9 +84,7 @@ public class MaxValueAggregator implements ValueAggregator<Object, Double> {
Object[] values = (Object[]) rawValue;
double max = Double.NEGATIVE_INFINITY;
for (Object value : values) {
- if (value != null) {
- max = Math.max(max, ValueAggregatorUtils.toDouble(value));
- }
+ max = Math.max(max, ValueAggregatorUtils.toDouble(value));
}
return max;
} else {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MinMaxRangeValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MinMaxRangeValueAggregator.java
index 2307e3aab79..535c8f62490 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MinMaxRangeValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MinMaxRangeValueAggregator.java
@@ -90,9 +90,7 @@ public class MinMaxRangeValueAggregator implements ValueAggregator<Object, MinMa
Object[] values = (Object[]) rawValue;
MinMaxRangePair minMaxRangePair = new MinMaxRangePair();
for (Object value : values) {
- if (value != null) {
- minMaxRangePair.apply(ValueAggregatorUtils.toDouble(value));
- }
+ minMaxRangePair.apply(ValueAggregatorUtils.toDouble(value));
}
return minMaxRangePair;
} else {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MinValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MinValueAggregator.java
index 64b0e98b94d..b7ee8f6cbf0 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MinValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/MinValueAggregator.java
@@ -84,9 +84,7 @@ public class MinValueAggregator implements ValueAggregator<Object, Double> {
Object[] values = (Object[]) rawValue;
double min = Double.POSITIVE_INFINITY;
for (Object value : values) {
- if (value != null) {
- min = Math.min(min, ValueAggregatorUtils.toDouble(value));
- }
+ min = Math.min(min, ValueAggregatorUtils.toDouble(value));
}
return min;
} else {
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileEstValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileEstValueAggregator.java
index 739b0a82b76..4b4cbc8af31 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileEstValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileEstValueAggregator.java
@@ -53,7 +53,7 @@ public class PercentileEstValueAggregator implements ValueAggregator<Object, Qua
_maxByteSize = Math.max(_maxByteSize, bytes.length);
} else {
initialValue = new QuantileDigest(DEFAULT_MAX_ERROR);
- initialValue.add(toLong(rawValue));
+ addToDigest(initialValue, rawValue);
_maxByteSize = Math.max(_maxByteSize, initialValue.getByteSize());
}
return initialValue;
@@ -64,12 +64,26 @@ public class PercentileEstValueAggregator implements ValueAggregator<Object, Qua
if (rawValue instanceof byte[]) {
value.merge(deserializeAggregatedValue((byte[]) rawValue));
} else {
- value.add(toLong(rawValue));
+ addToDigest(value, rawValue);
}
_maxByteSize = Math.max(_maxByteSize, value.getByteSize());
return value;
}
+ /**
+ * Adds a raw value (single value or multi-value array) to the QuantileDigest.
+ */
+ protected void addToDigest(QuantileDigest digest, Object rawValue) {
+ if (rawValue instanceof Object[]) {
+ Object[] values = (Object[]) rawValue;
+ for (Object value : values) {
+ digest.add(toLong(value));
+ }
+ } else {
+ digest.add(toLong(rawValue));
+ }
+ }
+
private static long toLong(Object rawValue) {
if (rawValue instanceof Number) {
return ((Number) rawValue).longValue();
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java
index 0e1a2caf342..772b6a52891 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/PercentileTDigestValueAggregator.java
@@ -63,7 +63,7 @@ public class PercentileTDigestValueAggregator implements ValueAggregator<Object,
_maxByteSize = Math.max(_maxByteSize, bytes.length);
} else {
initialValue = TDigest.createMergingDigest(_compressionFactor);
- initialValue.add(ValueAggregatorUtils.toDouble(rawValue));
+ addToDigest(initialValue, rawValue);
_maxByteSize = Math.max(_maxByteSize, initialValue.byteSize());
}
return initialValue;
@@ -74,12 +74,26 @@ public class PercentileTDigestValueAggregator implements ValueAggregator<Object,
if (rawValue instanceof byte[]) {
value.add(deserializeAggregatedValue((byte[]) rawValue));
} else {
- value.add(ValueAggregatorUtils.toDouble(rawValue));
+ addToDigest(value, rawValue);
}
_maxByteSize = Math.max(_maxByteSize, value.byteSize());
return value;
}
+ /**
+ * Adds a raw value (single value or multi-value array) to the TDigest.
+ */
+ protected void addToDigest(TDigest digest, Object rawValue) {
+ if (rawValue instanceof Object[]) {
+ Object[] values = (Object[]) rawValue;
+ for (Object value : values) {
+ digest.add(ValueAggregatorUtils.toDouble(value));
+ }
+ } else {
+ digest.add(ValueAggregatorUtils.toDouble(rawValue));
+ }
+ }
+
@Override
public TDigest applyAggregatedValue(TDigest value, TDigest aggregatedValue) {
value.add(aggregatedValue);
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/SumValueAggregator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/SumValueAggregator.java
index 882e5beed2c..2c788586616 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/SumValueAggregator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/aggregator/SumValueAggregator.java
@@ -87,9 +87,7 @@ public class SumValueAggregator implements ValueAggregator<Object, Double> {
Object[] values = (Object[]) rawValue;
double sum = 0.0;
for (Object value : values) {
- if (value != null) {
- sum += ValueAggregatorUtils.toDouble(value);
- }
+ sum += ValueAggregatorUtils.toDouble(value);
}
return sum;
} else {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org