siddharthteotia commented on code in PR #10128:
URL: https://github.com/apache/pinot/pull/10128#discussion_r1073857754
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctAggregateAggregationFunction.java:
##########
@@ -136,8 +202,84 @@ public void aggregate(int length, AggregationResultHolder
aggregationResultHolde
}
}
- @Override
- public void aggregateGroupBySV(int length, int[] groupKeyArray,
GroupByResultHolder groupByResultHolder,
+ /**
+ * Performs aggregation for a MV column
+ */
+ protected void mvAggregate(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 value set
+ DataType storedType = blockValSet.getValueType().getStoredType();
+ Set valueSet = getValueSet(aggregationResultHolder, storedType);
+ switch (storedType) {
+ case INT:
+ IntOpenHashSet intSet = (IntOpenHashSet) valueSet;
+ int[][] intValues = blockValSet.getIntValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (int value : intValues[i]) {
+ intSet.add(value);
+ }
+ }
+ break;
+ case LONG:
+ LongOpenHashSet longSet = (LongOpenHashSet) valueSet;
+ long[][] longValues = blockValSet.getLongValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (long value : longValues[i]) {
+ longSet.add(value);
+ }
+ }
+ break;
+ case FLOAT:
+ FloatOpenHashSet floatSet = (FloatOpenHashSet) valueSet;
+ float[][] floatValues = blockValSet.getFloatValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (float value : floatValues[i]) {
+ floatSet.add(value);
+ }
+ }
+ break;
+ case DOUBLE:
+ DoubleOpenHashSet doubleSet = (DoubleOpenHashSet) valueSet;
+ double[][] doubleValues = blockValSet.getDoubleValuesMV();
+ for (int i = 0; i < length; i++) {
+ for (double value : doubleValues[i]) {
+ doubleSet.add(value);
+ }
+ }
+ break;
+ case STRING:
+ ObjectOpenHashSet<String> stringSet = (ObjectOpenHashSet<String>)
valueSet;
+ String[][] stringValues = blockValSet.getStringValuesMV();
+ for (int i = 0; i < length; i++) {
+ //noinspection ManualArrayToCollectionCopy
+ for (String value : stringValues[i]) {
+ //noinspection UseBulkOperation
+ stringSet.add(value);
+ }
+ }
+ break;
+ default:
+ throw new IllegalStateException("Illegal data type for
DISTINCT_COUNT_MV aggregation function: " + storedType);
Review Comment:
Doesn't look like exception message is correct probably because copied over
from existing code ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
