This is an automated email from the ASF dual-hosted git repository. xiangfu0 pushed a commit to branch xiangfu0/agg-rewrite/03-mode in repository https://gitbox.apache.org/repos/asf/pinot.git
commit 2423329bf60705cb18c6808a12563942246688e7 Author: Xiang Fu <[email protected]> AuthorDate: Tue Oct 20 18:28:33 2026 -0700 Support MODE on STRING, BOOLEAN and TIMESTAMP inputs MODE on a STRING, BOOLEAN or TIMESTAMP input returns the input's type, with MIN (default) and MAX tie reducers. Both engines rewrite the call to MODE(expr, 'MIN'|'MAX', 'dataType'): the single-stage optimizer for STRING/JSON, BOOLEAN and TIMESTAMP columns, and the multi-stage rule for VARCHAR arguments, since BOOLEAN and TIMESTAMP already return their own type there. String counts are serialized with a new String2LongMap object type, and the explicit form also sets the type of a leaf-returned final result. The aggregation-only reducer now skips formatting NULL values, like the group-by reducer. Single-stage MODE on BOOLEAN and TIMESTAMP columns used to return DOUBLE and accept the AVG reducer; it now returns BOOLEAN/TIMESTAMP and rejects AVG. Servers older than the broker reject the rewritten call. --- .../apache/pinot/core/common/ObjectSerDeUtils.java | 53 +++++- .../function/ModeAggregationFunction.java | 203 ++++++++++++++++++--- .../AggregateFunctionRewriteOptimizer.java | 30 +++ .../query/reduce/AggregationDataTableReducer.java | 5 +- .../pinot/core/common/ObjectSerDeUtilsTest.java | 17 ++ .../function/ModeAggregationFunctionTest.java | 181 ++++++++++++++++++ .../AggregateFunctionRewriteOptimizerTest.java | 23 +++ .../org/apache/pinot/queries/ModeQueriesTest.java | 146 ++++++++++++++- .../PinotAggregateTypeArgumentRewriteRule.java | 68 ++++--- .../src/test/resources/queries/AggregatePlans.json | 26 +++ .../src/test/resources/queries/UDFAggregates.json | 39 ++++ .../pinot/segment/spi/AggregationFunctionType.java | 11 +- .../apache/pinot/spi/utils/CommonConstants.java | 2 +- 13 files changed, 743 insertions(+), 61 deletions(-) diff --git a/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java b/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java index b3c17dc809c..3557298404c 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/common/ObjectSerDeUtils.java @@ -49,6 +49,8 @@ import it.unimi.dsi.fastutil.longs.LongArrayList; import it.unimi.dsi.fastutil.longs.LongIterator; import it.unimi.dsi.fastutil.longs.LongOpenHashSet; import it.unimi.dsi.fastutil.longs.LongSet; +import it.unimi.dsi.fastutil.objects.Object2LongMap; +import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap; import it.unimi.dsi.fastutil.objects.ObjectArrayList; import it.unimi.dsi.fastutil.objects.ObjectLinkedOpenHashSet; import it.unimi.dsi.fastutil.objects.ObjectOpenHashSet; @@ -173,7 +175,8 @@ public class ObjectSerDeUtils { FunnelStepEventAccumulator(51), BigDecimalArrayList(52), BigDecimalSet(53), - BytesArrayList(54); + BytesArrayList(54), + String2LongMap(55); private final int _value; @@ -1637,6 +1640,53 @@ public class ObjectSerDeUtils { } }; + public static final ObjectSerDe<Object2LongMap<String>> STRING_2_LONG_MAP_SER_DE = + new ObjectSerDe<Object2LongMap<String>>() { + + @Override + public byte[] serialize(Object2LongMap<String> map) { + int size = map.size(); + // Besides the key bytes, we store: size, length for each key, value for each key + long bufferSize = Integer.BYTES + (long) size * (Integer.BYTES + Long.BYTES); + byte[][] keyBytesArray = new byte[size][]; + long[] values = new long[size]; + int index = 0; + for (Object2LongMap.Entry<String> entry : map.object2LongEntrySet()) { + byte[] keyBytes = entry.getKey().getBytes(UTF_8); + bufferSize += keyBytes.length; + keyBytesArray[index] = keyBytes; + values[index++] = entry.getLongValue(); + } + Preconditions.checkState(bufferSize <= Integer.MAX_VALUE, "Buffer size exceeds 2GB"); + byte[] bytes = new byte[(int) bufferSize]; + ByteBuffer byteBuffer = ByteBuffer.wrap(bytes); + byteBuffer.putInt(size); + for (int i = 0; i < size; i++) { + byteBuffer.putInt(keyBytesArray[i].length); + byteBuffer.put(keyBytesArray[i]); + byteBuffer.putLong(values[i]); + } + return bytes; + } + + @Override + public Object2LongOpenHashMap<String> deserialize(byte[] bytes) { + return deserialize(ByteBuffer.wrap(bytes)); + } + + @Override + public Object2LongOpenHashMap<String> deserialize(ByteBuffer byteBuffer) { + int size = byteBuffer.getInt(); + Object2LongOpenHashMap<String> map = new Object2LongOpenHashMap<>(size); + for (int i = 0; i < size; i++) { + byte[] keyBytes = new byte[byteBuffer.getInt()]; + byteBuffer.get(keyBytes); + map.put(new String(keyBytes, UTF_8), byteBuffer.getLong()); + } + return map; + } + }; + public static final ObjectSerDe<Int2LongMap> INT_2_LONG_MAP_SER_DE = new ObjectSerDe<Int2LongMap>() { @Override @@ -2089,6 +2139,7 @@ public class ObjectSerDeUtils { BIG_DECIMAL_ARRAY_LIST_SER_DE, BIG_DECIMAL_SET_SER_DE, BYTES_ARRAY_LIST_SER_DE, + STRING_2_LONG_MAP_SER_DE, }; //@formatter:on diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunction.java b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunction.java index 95ccf22bbe4..ae0bd6a4841 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunction.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunction.java @@ -30,8 +30,11 @@ import it.unimi.dsi.fastutil.ints.Int2LongMap; import it.unimi.dsi.fastutil.ints.Int2LongOpenHashMap; import it.unimi.dsi.fastutil.longs.Long2LongMap; import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap; +import it.unimi.dsi.fastutil.objects.Object2LongMap; +import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap; import it.unimi.dsi.fastutil.objects.ObjectIterator; import java.util.List; +import java.util.Locale; import java.util.Map; import javax.annotation.Nullable; import org.apache.pinot.common.CustomObject; @@ -51,33 +54,68 @@ import org.apache.pinot.spi.data.FieldSpec.DataType; /// This function is used for Mode calculations. /// -/// The function can be used as MODE(expression, multiModeReducerType) +/// The function can be used as MODE(expression, multiModeReducerType, dataType) /// /// Following arguments are supported: /// -/// - Expression: expression that contains the column to be calculated mode on, can be any Numeric column +/// - Expression: expression that contains the column to be calculated mode on /// - MultiModeReducerType (optional): the reducer to use in case of multiple modes present in data +/// - DataType (optional): STRING, BOOLEAN or TIMESTAMP, which is also the result type. Values are read as this type, +/// and only the MIN and MAX reducers are supported. Without it, the expression must be numeric and the result is +/// DOUBLE. @SuppressWarnings({"rawtypes", "unchecked"}) -public class ModeAggregationFunction extends BaseSingleInputAggregationFunction<Map<? extends Number, Long>, Double> { +public class ModeAggregationFunction extends BaseSingleInputAggregationFunction<Map<?, Long>, Comparable<?>> { private static final double DEFAULT_FINAL_RESULT = Double.NEGATIVE_INFINITY; private final MultiModeReducerType _multiModeReducerType; + private final ColumnDataType _resultType; public ModeAggregationFunction(List<ExpressionContext> arguments, boolean nullHandlingEnabled) { super(arguments.get(0), nullHandlingEnabled); int numArguments = arguments.size(); - Preconditions.checkArgument(numArguments <= 2, "Mode expects at most 2 arguments, got: %s", numArguments); + Preconditions.checkArgument(numArguments <= 3, "Mode expects at most 3 arguments, got: %s", numArguments); if (numArguments > 1) { _multiModeReducerType = MultiModeReducerType.valueOf(arguments.get(1).getLiteral().getStringValue()); } else { _multiModeReducerType = MultiModeReducerType.MIN; } + if (numArguments > 2) { + _resultType = getResultType(arguments.get(2).getLiteral().getStringValue()); + Preconditions.checkArgument(_multiModeReducerType != MultiModeReducerType.AVG, + "Unsupported MODE reducer for %s: AVG", _resultType); + } else { + _resultType = ColumnDataType.DOUBLE; + } + } + + private static ColumnDataType getResultType(String dataType) { + switch (DataType.valueOf(dataType.toUpperCase(Locale.ROOT))) { + case STRING: + return ColumnDataType.STRING; + case BOOLEAN: + return ColumnDataType.BOOLEAN; + case TIMESTAMP: + return ColumnDataType.TIMESTAMP; + default: + throw new IllegalArgumentException("Unsupported data type for MODE: " + dataType); + } + } + + /// Returns the stored type to read the values as. BOOLEAN and TIMESTAMP values are read as INT and LONG. + private DataType getStoredType(DataType valueType) { + if (_resultType != ColumnDataType.DOUBLE) { + return _resultType.getStoredType().toDataType(); + } + DataType storedType = valueType.getStoredType(); + Preconditions.checkState(storedType.isNumeric(), "Illegal data type for MODE aggregation function: %s", + storedType); + return storedType; } /// Helper method to create a value map for the given value type. - private static Map<? extends Number, Long> getValueMap(DataType valueType) { + private static Map<?, Long> getValueMap(DataType valueType) { switch (valueType) { case INT: return new Int2LongOpenHashMap(); @@ -87,15 +125,16 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< return new Float2LongOpenHashMap(); case DOUBLE: return new Double2LongOpenHashMap(); + case STRING: + return new Object2LongOpenHashMap<String>(); default: throw new IllegalStateException("Illegal data type for MODE aggregation function: " + valueType); } } /// Returns the value map from the result holder or creates a new one if it does not exist. - private static Map<? extends Number, Long> getValueMap(AggregationResultHolder aggregationResultHolder, - DataType valueType) { - Map<? extends Number, Long> valueMap = aggregationResultHolder.getResult(); + private static Map<?, Long> getValueMap(AggregationResultHolder aggregationResultHolder, DataType valueType) { + Map<?, Long> valueMap = aggregationResultHolder.getResult(); if (valueMap == null) { valueMap = getValueMap(valueType); aggregationResultHolder.setValue(valueMap); @@ -143,6 +182,16 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< valueMap.merge(value, 1, Long::sum); } + /// Helper method to set STRING value for the given group keys into the result holder. + private static void setValueForGroupKeys(GroupByResultHolder groupByResultHolder, int groupKey, String value) { + Object2LongOpenHashMap<String> valueMap = groupByResultHolder.getResult(groupKey); + if (valueMap == null) { + valueMap = new Object2LongOpenHashMap<>(); + groupByResultHolder.setValueForKey(groupKey, valueMap); + } + valueMap.addTo(value, 1); + } + /// Returns the dictionary id count map from the result holder or creates a new one if it does not exist. protected static Int2IntOpenHashMap getDictIdCountMap(AggregationResultHolder aggregationResultHolder, Dictionary dictionary) { @@ -166,25 +215,26 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } /// Helper method to read dictionary and convert dictionary ids to values for dictionary-encoded expression. - private static Map<? extends Number, Long> convertToValueMap(DictIdsWrapper dictIdsWrapper) { + /// Counts are added up because values read as another type can collide. + private Map<?, Long> convertToValueMap(DictIdsWrapper dictIdsWrapper) { Dictionary dictionary = dictIdsWrapper._dictionary; Int2IntOpenHashMap dictIdCountMap = dictIdsWrapper._dictIdCountMap; int numValues = dictIdCountMap.size(); ObjectIterator<Int2IntMap.Entry> iterator = Int2IntMaps.fastIterator(dictIdCountMap); - DataType storedType = dictionary.getValueType(); + DataType storedType = getStoredType(dictionary.getValueType()); switch (storedType) { case INT: Int2LongOpenHashMap intValueMap = new Int2LongOpenHashMap(numValues); while (iterator.hasNext()) { Int2IntMap.Entry next = iterator.next(); - intValueMap.put(dictionary.getIntValue(next.getIntKey()), next.getIntValue()); + intValueMap.addTo(dictionary.getIntValue(next.getIntKey()), next.getIntValue()); } return intValueMap; case LONG: Long2LongOpenHashMap longValueMap = new Long2LongOpenHashMap(numValues); while (iterator.hasNext()) { Int2IntMap.Entry next = iterator.next(); - longValueMap.put(dictionary.getLongValue(next.getIntKey()), next.getIntValue()); + longValueMap.addTo(dictionary.getLongValue(next.getIntKey()), next.getIntValue()); } return longValueMap; case FLOAT: @@ -201,13 +251,20 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< doubleValueMap.put(dictionary.getDoubleValue(next.getIntKey()), next.getIntValue()); } return doubleValueMap; + case STRING: + Object2LongOpenHashMap<String> stringValueMap = new Object2LongOpenHashMap<>(numValues); + while (iterator.hasNext()) { + Int2IntMap.Entry next = iterator.next(); + stringValueMap.addTo(dictionary.getStringValue(next.getIntKey()), next.getIntValue()); + } + return stringValueMap; default: throw new IllegalStateException("Illegal data type for MODE aggregation function: " + storedType); } } /// Helper method to extract segment level intermediate result from the inner segment result. - private static Map<? extends Number, Long> extractIntermediateResult(@Nullable Object result) { + private Map<?, Long> extractIntermediateResult(@Nullable Object result) { if (result == null) { // NOTE: Return an empty Int2LongOpenHashMap for empty result. return new Int2LongOpenHashMap(); @@ -257,8 +314,8 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } // For non-dictionary-encoded expression, store values into the value map - DataType storedType = blockValSet.getValueType().getStoredType(); - Map<? extends Number, Long> valueMap = getValueMap(aggregationResultHolder, storedType); + DataType storedType = getStoredType(blockValSet.getValueType()); + Map<?, Long> valueMap = getValueMap(aggregationResultHolder, storedType); switch (storedType) { case INT: Int2LongOpenHashMap intMap = (Int2LongOpenHashMap) valueMap; @@ -296,6 +353,15 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } }); break; + case STRING: + Object2LongOpenHashMap<String> stringMap = (Object2LongOpenHashMap<String>) valueMap; + String[] stringValues = blockValSet.getStringValuesSV(); + forEachNotNull(length, blockValSet, (from, to) -> { + for (int i = from; i < to; i++) { + stringMap.addTo(stringValues[i], 1); + } + }); + break; default: throw new IllegalStateException("Illegal data type for MODE aggregation function: " + storedType); } @@ -320,7 +386,7 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } // For non-dictionary-encoded expression, store values into the value map - DataType storedType = blockValSet.getValueType().getStoredType(); + DataType storedType = getStoredType(blockValSet.getValueType()); switch (storedType) { case INT: int[] intValues = blockValSet.getIntValuesSV(); @@ -354,6 +420,14 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } }); break; + case STRING: + String[] stringValues = blockValSet.getStringValuesSV(); + forEachNotNull(length, blockValSet, (from, to) -> { + for (int i = from; i < to; i++) { + setValueForGroupKeys(groupByResultHolder, groupKeyArray[i], stringValues[i]); + } + }); + break; default: throw new IllegalStateException("Illegal data type for MODE aggregation function: " + storedType); } @@ -379,7 +453,7 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } // For non-dictionary-encoded expression, store values into the value map - DataType storedType = blockValSet.getValueType().getStoredType(); + DataType storedType = getStoredType(blockValSet.getValueType()); switch (storedType) { case INT: int[] intValues = blockValSet.getIntValuesSV(); @@ -421,24 +495,33 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } }); break; + case STRING: + String[] stringValues = blockValSet.getStringValuesSV(); + forEachNotNull(length, blockValSet, (from, to) -> { + for (int i = from; i < to; i++) { + for (int groupKey : groupKeysArray[i]) { + setValueForGroupKeys(groupByResultHolder, groupKey, stringValues[i]); + } + } + }); + break; default: throw new IllegalStateException("Illegal data type for MODE aggregation function: " + storedType); } } @Override - public Map<? extends Number, Long> extractAggregationResult(AggregationResultHolder aggregationResultHolder) { + public Map<?, Long> extractAggregationResult(AggregationResultHolder aggregationResultHolder) { return extractIntermediateResult(aggregationResultHolder.getResult()); } @Override - public Map<? extends Number, Long> extractGroupByResult(GroupByResultHolder groupByResultHolder, int groupKey) { + public Map<?, Long> extractGroupByResult(GroupByResultHolder groupByResultHolder, int groupKey) { return extractIntermediateResult(groupByResultHolder.getResult(groupKey)); } @Override - public Map<? extends Number, Long> merge(Map<? extends Number, Long> intermediateResult1, - Map<? extends Number, Long> intermediateResult2) { + public Map<?, Long> merge(Map<?, Long> intermediateResult1, Map<?, Long> intermediateResult2) { if (intermediateResult1.isEmpty()) { return intermediateResult2; } @@ -460,6 +543,10 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< && intermediateResult2 instanceof Double2LongOpenHashMap) { ((Double2LongOpenHashMap) intermediateResult2).double2LongEntrySet().fastForEach( e -> ((Double2LongOpenHashMap) intermediateResult1).merge(e.getDoubleKey(), e.getLongValue(), Long::sum)); + } else if (intermediateResult1 instanceof Object2LongOpenHashMap + && intermediateResult2 instanceof Object2LongOpenHashMap) { + ((Object2LongOpenHashMap<String>) intermediateResult2).object2LongEntrySet().fastForEach( + e -> ((Object2LongOpenHashMap<String>) intermediateResult1).addTo(e.getKey(), e.getLongValue())); } else { throw new IllegalStateException( "Illegal data type for Intermediate Result of MODE aggregation function: " + intermediateResult1.getClass() @@ -474,7 +561,7 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } @Override - public SerializedIntermediateResult serializeIntermediateResult(Map<? extends Number, Long> longMap) { + public SerializedIntermediateResult serializeIntermediateResult(Map<?, Long> longMap) { if (longMap instanceof Int2LongMap) { return new SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.Int2LongMap.getValue(), ObjectSerDeUtils.INT_2_LONG_MAP_SER_DE.serialize((Int2LongMap) longMap)); @@ -487,6 +574,9 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } else if (longMap instanceof Double2LongMap) { return new SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.Double2LongMap.getValue(), ObjectSerDeUtils.DOUBLE_2_LONG_MAP_SER_DE.serialize((Double2LongMap) longMap)); + } else if (longMap instanceof Object2LongMap) { + return new SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.String2LongMap.getValue(), + ObjectSerDeUtils.STRING_2_LONG_MAP_SER_DE.serialize((Object2LongMap<String>) longMap)); } else { throw new IllegalStateException( "Illegal data type for Intermediate Result of MODE aggregation function: " + longMap.getClass() @@ -495,18 +585,22 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } @Override - public Map<? extends Number, Long> deserializeIntermediateResult(CustomObject customObject) { + public Map<?, Long> deserializeIntermediateResult(CustomObject customObject) { return ObjectSerDeUtils.deserialize(customObject); } @Override public ColumnDataType getFinalResultColumnType() { - return ColumnDataType.DOUBLE; + return _resultType; } @Nullable @Override - public Double extractFinalResult(@Nullable Map<? extends Number, Long> intermediateResult) { + public Comparable<?> extractFinalResult(@Nullable Map<?, Long> intermediateResult) { + if (_resultType != ColumnDataType.DOUBLE) { + return intermediateResult == null || intermediateResult.isEmpty() ? null + : extractTypedFinalResult(intermediateResult); + } // A null intermediate result means nothing was aggregated, and the mode of nothing is NULL. An empty map is a // different thing: it is what an untouched single-stage result holder produces, and it keeps its historical // sentinel below so that path is not silently changed. @@ -720,6 +814,65 @@ public class ModeAggregationFunction extends BaseSingleInputAggregationFunction< } } + /// Returns the most frequent value read as the result type, breaking ties with the MIN or MAX reducer. + private Comparable<?> extractTypedFinalResult(Map<?, Long> intermediateResult) { + boolean max = _multiModeReducerType == MultiModeReducerType.MAX; + if (intermediateResult instanceof Int2LongOpenHashMap) { + ObjectIterator<Int2LongMap.Entry> iterator = + ((Int2LongOpenHashMap) intermediateResult).int2LongEntrySet().fastIterator(); + Int2LongMap.Entry first = iterator.next(); + long maxFrequency = first.getLongValue(); + int mode = first.getIntKey(); + while (iterator.hasNext()) { + Int2LongMap.Entry next = iterator.next(); + long frequency = next.getLongValue(); + int value = next.getIntKey(); + if (frequency > maxFrequency || (frequency == maxFrequency && (max ? value > mode : value < mode))) { + maxFrequency = frequency; + mode = value; + } + } + return mode; + } else if (intermediateResult instanceof Long2LongOpenHashMap) { + ObjectIterator<Long2LongMap.Entry> iterator = + ((Long2LongOpenHashMap) intermediateResult).long2LongEntrySet().fastIterator(); + Long2LongMap.Entry first = iterator.next(); + long maxFrequency = first.getLongValue(); + long mode = first.getLongKey(); + while (iterator.hasNext()) { + Long2LongMap.Entry next = iterator.next(); + long frequency = next.getLongValue(); + long value = next.getLongKey(); + if (frequency > maxFrequency || (frequency == maxFrequency && (max ? value > mode : value < mode))) { + maxFrequency = frequency; + mode = value; + } + } + return mode; + } else if (intermediateResult instanceof Object2LongOpenHashMap) { + ObjectIterator<Object2LongMap.Entry<String>> iterator = + ((Object2LongOpenHashMap<String>) intermediateResult).object2LongEntrySet().fastIterator(); + Object2LongMap.Entry<String> first = iterator.next(); + long maxFrequency = first.getLongValue(); + String mode = first.getKey(); + while (iterator.hasNext()) { + Object2LongMap.Entry<String> next = iterator.next(); + long frequency = next.getLongValue(); + String value = next.getKey(); + if (frequency > maxFrequency || (frequency == maxFrequency && (max ? value.compareTo(mode) > 0 + : value.compareTo(mode) < 0))) { + maxFrequency = frequency; + mode = value; + } + } + return mode; + } else { + throw new IllegalStateException( + "Illegal data type for Intermediate Result of MODE aggregation function: " + intermediateResult.getClass() + .getSimpleName()); + } + } + private enum MultiModeReducerType { MIN, MAX, AVG } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizer.java index 5bbc3274e05..ca344be85e5 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizer.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizer.java @@ -35,6 +35,7 @@ import org.apache.pinot.spi.utils.CommonConstants; /// Always applied: /// - FIRST_WITH_TIME(dataColumn, timeColumn) -> FIRST_WITH_TIME(dataColumn, timeColumn, 'dataType') /// - LAST_WITH_TIME(dataColumn, timeColumn) -> LAST_WITH_TIME(dataColumn, timeColumn, 'dataType') +/// - MODE(nonNumericColumn[, 'MIN'|'MAX']) -> MODE(nonNumericColumn, 'MIN'|'MAX', 'dataType') /// /// Applied only when enabled through the `autoRewriteAggregationType` query option: /// - MIN(stringType) -> MINSTRING @@ -97,6 +98,10 @@ public class AggregateFunctionRewriteOptimizer implements StatementOptimizer { addDataTypeArgument(function, fieldSpec); return; } + if (normalizedName.equals(AggregationFunctionType.MODE.name()) && function.getOperandsSize() <= 2) { + addModeDataTypeArgument(function, fieldSpec); + return; + } if (rewriteToTypedVariants) { rewriteToTypedVariant(function, functionName, fieldSpec.getDataType().getStoredType()); } @@ -126,6 +131,31 @@ public class AggregateFunctionRewriteOptimizer implements StatementOptimizer { } } + /// Appends the default tie reducer when missing, then the data column's logical type when it is non-numeric. + private static void addModeDataTypeArgument(Function function, FieldSpec fieldSpec) { + if (!fieldSpec.isSingleValueField()) { + return; + } + String dataType; + switch (fieldSpec.getDataType()) { + case STRING: + case BOOLEAN: + case TIMESTAMP: + dataType = fieldSpec.getDataType().name(); + break; + case JSON: + // Same as the multi-stage engine, which types a JSON column as VARCHAR. + dataType = FieldSpec.DataType.STRING.name(); + break; + default: + return; + } + if (function.getOperandsSize() == 1) { + function.addToOperands(RequestUtils.getLiteralExpression("MIN")); + } + function.addToOperands(RequestUtils.getLiteralExpression(dataType)); + } + private static void rewriteToTypedVariant(Function function, String functionName, FieldSpec.DataType operandType) { // Rewrite MIN(stringCol) and MAX(stringCol) to MINSTRING / MAXSTRING // Rewrite MIN(longCol) and MAX(longCol) to MINLONG / MAXLONG diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java index 5c8143975f9..546d3ca7a8e 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/reduce/AggregationDataTableReducer.java @@ -242,7 +242,10 @@ public class AggregationDataTableReducer implements DataTableReducer { int numColumns = columnDataTypes.length; for (Object[] rewrittenRow : rows) { for (int j = 0; j < numColumns; j++) { - rewrittenRow[j] = columnDataTypes[j].format(rewrittenRow[j]); + Object value = rewrittenRow[j]; + if (value != null) { + rewrittenRow[j] = columnDataTypes[j].format(value); + } } } diff --git a/pinot-core/src/test/java/org/apache/pinot/core/common/ObjectSerDeUtilsTest.java b/pinot-core/src/test/java/org/apache/pinot/core/common/ObjectSerDeUtilsTest.java index 41c969962a5..6f77923c060 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/common/ObjectSerDeUtilsTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/common/ObjectSerDeUtilsTest.java @@ -30,6 +30,7 @@ import it.unimi.dsi.fastutil.ints.IntOpenHashSet; import it.unimi.dsi.fastutil.ints.IntSet; import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap; import it.unimi.dsi.fastutil.longs.LongArrayList; +import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap; import it.unimi.dsi.fastutil.objects.ObjectArrayList; import it.unimi.dsi.fastutil.objects.ObjectLinkedOpenHashSet; import java.nio.BufferOverflowException; @@ -650,6 +651,22 @@ public class ObjectSerDeUtilsTest { } } + @Test + public void testString2LongMap() { + for (int i = 0; i < NUM_ITERATIONS; i++) { + int size = RANDOM.nextInt(100); + Object2LongOpenHashMap<String> expected = new Object2LongOpenHashMap<>(size); + for (int j = 0; j < size; j++) { + expected.put(RandomStringUtils.secure().next(RANDOM.nextInt(5)), RANDOM.nextLong()); + } + + byte[] bytes = ObjectSerDeUtils.STRING_2_LONG_MAP_SER_DE.serialize(expected); + Object2LongOpenHashMap<String> actual = ObjectSerDeUtils.deserialize(bytes, ObjectType.String2LongMap); + + assertEquals(actual, expected, ERROR_MESSAGE); + } + } + @Test public void testCpcSketch() { for (int i = 0; i < NUM_ITERATIONS; i++) { diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunctionTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunctionTest.java index 78a7cdcada9..9aa4d02d572 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunctionTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/ModeAggregationFunctionTest.java @@ -19,6 +19,17 @@ package org.apache.pinot.core.query.aggregation.function; +import it.unimi.dsi.fastutil.ints.Int2LongOpenHashMap; +import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap; +import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap; +import java.nio.ByteBuffer; +import java.sql.Timestamp; +import java.util.List; +import java.util.Map; +import org.apache.pinot.common.CustomObject; +import org.apache.pinot.common.request.context.ExpressionContext; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; +import org.apache.pinot.core.query.aggregation.function.AggregationFunction.SerializedIntermediateResult; import org.apache.pinot.queries.FluentQueryTest; import org.apache.pinot.spi.config.table.FieldConfig; import org.apache.pinot.spi.data.FieldSpec.DataType; @@ -26,6 +37,10 @@ import org.apache.pinot.spi.utils.PinotDataType; import org.testng.annotations.DataProvider; import org.testng.annotations.Test; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertNull; +import static org.testng.Assert.assertThrows; + public class ModeAggregationFunctionTest extends AbstractAggregationFunctionTest { @@ -232,4 +247,170 @@ public class ModeAggregationFunctionTest extends AbstractAggregationFunctionTest .whenQuery("select 'cte1' as cte1, 'cte2' as cte2, mode(myField) as mode from testTable group by cte1, cte2") .thenResultIs("STRING | STRING | DOUBLE", "cte1 | cte2 | 1"); } + + @DataProvider(name = "nonNumericScenarios") + Object[] nonNumericScenarios() { + long earlier = 1_600_000_000_000L; + long later = 1_700_000_000_000L; + return new Object[]{ + new NonNumericScenario(DataType.STRING, true, "a", "b", "a", "b", "null"), + new NonNumericScenario(DataType.STRING, false, "a", "b", "a", "b", "null"), + new NonNumericScenario(DataType.BOOLEAN, true, "false", "true", false, true, false), + new NonNumericScenario(DataType.BOOLEAN, false, "false", "true", false, true, false), + new NonNumericScenario(DataType.TIMESTAMP, true, String.valueOf(earlier), String.valueOf(later), + new Timestamp(earlier).toString(), new Timestamp(later).toString(), new Timestamp(0).toString()), + new NonNumericScenario(DataType.TIMESTAMP, false, String.valueOf(earlier), String.valueOf(later), + new Timestamp(earlier).toString(), new Timestamp(later).toString(), new Timestamp(0).toString()) + }; + } + + /// Inputs are the lower and the higher value; results are as the broker returns them. + public class NonNumericScenario { + private final DataType _dataType; + private final boolean _dictionary; + private final String _low; + private final String _high; + private final Object _lowResult; + private final Object _highResult; + private final Object _defaultNullValueResult; + + public NonNumericScenario(DataType dataType, boolean dictionary, String low, String high, Object lowResult, + Object highResult, Object defaultNullValueResult) { + _dataType = dataType; + _dictionary = dictionary; + _low = low; + _high = high; + _lowResult = lowResult; + _highResult = highResult; + _defaultNullValueResult = defaultNullValueResult; + } + + public FluentQueryTest.DeclaringTable getDeclaringTable(boolean nullHandlingEnabled) { + FieldConfig.EncodingType encodingType = + _dictionary ? FieldConfig.EncodingType.DICTIONARY : FieldConfig.EncodingType.RAW; + return givenSingleNullableFieldTable(_dataType, nullHandlingEnabled, builder -> { + builder.withEncodingType(encodingType); + builder.withCompressionCodec(FieldConfig.CompressionCodec.PASS_THROUGH); + }); + } + + public String mode(String reducer) { + return "mode(myField, '" + reducer + "', '" + _dataType + "')"; + } + + @Override + public String toString() { + return "NonNumericScenario{" + "dt=" + _dataType + ", dict=" + _dictionary + '}'; + } + } + + @Test(dataProvider = "nonNumericScenarios") + void nonNumericAggrWithNull(NonNumericScenario scenario) { + scenario.getDeclaringTable(true) + .onFirstInstance("myField", "null", scenario._low, scenario._high) + .andOnSecondInstance("myField", "null", scenario._high, "null") + .whenQuery("select " + scenario.mode("MIN") + " as mode from testTable") + .thenResultIs(new Object[]{scenario._highResult}) + .whenQuery("select 'cte', " + scenario.mode("MIN") + " as mode from testTable group by 'cte'") + .thenResultIs(new Object[]{"cte", scenario._highResult}); + } + + @Test(dataProvider = "nonNumericScenarios") + void nonNumericAggrWithoutNull(NonNumericScenario scenario) { + // Nulls are read as the default null value, which becomes the most frequent value + scenario.getDeclaringTable(false) + .onFirstInstance("myField", "null", scenario._low, scenario._high) + .andOnSecondInstance("myField", "null", scenario._high, "null") + .whenQuery("select " + scenario.mode("MIN") + " as mode from testTable") + .thenResultIs(new Object[]{scenario._defaultNullValueResult}) + .whenQuery("select 'cte', " + scenario.mode("MIN") + " as mode from testTable group by 'cte'") + .thenResultIs(new Object[]{"cte", scenario._defaultNullValueResult}); + } + + @Test(dataProvider = "nonNumericScenarios") + void nonNumericAggrOnlyNulls(NonNumericScenario scenario) { + scenario.getDeclaringTable(true) + .onFirstInstance("myField", "null", "null") + .andOnSecondInstance("myField", "null", "null") + .whenQuery("select " + scenario.mode("MIN") + " as mode from testTable") + .thenResultIs(new Object[]{null}) + .whenQuery("select 'cte', " + scenario.mode("MAX") + " as mode from testTable group by 'cte'") + .thenResultIs(new Object[]{"cte", null}); + } + + @Test(dataProvider = "nonNumericScenarios") + void nonNumericAggrTieReducers(NonNumericScenario scenario) { + scenario.getDeclaringTable(false) + .onFirstInstance("myField", scenario._high, scenario._low) + .andOnSecondInstance("myField", scenario._low, scenario._high) + .whenQuery("select " + scenario.mode("MIN") + ", " + scenario.mode("MAX") + " from testTable") + .thenResultIs(new Object[]{scenario._lowResult, scenario._highResult}) + .whenQuery("select 'cte', " + scenario.mode("MIN") + ", " + scenario.mode("MAX") + + " from testTable group by 'cte'") + .thenResultIs(new Object[]{"cte", scenario._lowResult, scenario._highResult}); + } + + @Test + void nonNumericSerializedMerge() { + ModeAggregationFunction stringMode = createFunction("MAX", "STRING"); + assertEquals(stringMode.getFinalResultColumnType(), ColumnDataType.STRING); + Object2LongOpenHashMap<String> strings1 = new Object2LongOpenHashMap<>(Map.of("a", 2L, "b", 1L)); + Object2LongOpenHashMap<String> strings2 = new Object2LongOpenHashMap<>(Map.of("b", 1L, "c", 2L)); + Map<?, Long> mergedStrings = stringMode.merge(roundTrip(stringMode, strings1), roundTrip(stringMode, strings2)); + assertEquals(mergedStrings, Map.of("a", 2L, "b", 2L, "c", 2L)); + assertEquals(stringMode.extractFinalResult(mergedStrings), "c"); + assertEquals(createFunction("MIN", "STRING").extractFinalResult(mergedStrings), "a"); + + ModeAggregationFunction booleanMode = createFunction("MIN", "BOOLEAN"); + assertEquals(booleanMode.getFinalResultColumnType(), ColumnDataType.BOOLEAN); + Int2LongOpenHashMap booleans = new Int2LongOpenHashMap(Map.of(0, 3L, 1, 3L)); + assertEquals(booleanMode.extractFinalResult(roundTrip(booleanMode, booleans)), 0); + assertEquals(createFunction("MAX", "BOOLEAN").extractFinalResult(booleans), 1); + + ModeAggregationFunction timestampMode = createFunction("MAX", "TIMESTAMP"); + assertEquals(timestampMode.getFinalResultColumnType(), ColumnDataType.TIMESTAMP); + Long2LongOpenHashMap timestamps1 = new Long2LongOpenHashMap(Map.of(Long.MAX_VALUE - 1, 2L, 5L, 1L)); + Long2LongOpenHashMap timestamps2 = new Long2LongOpenHashMap(Map.of(Long.MAX_VALUE, 2L)); + Map<?, Long> mergedTimestamps = + timestampMode.merge(roundTrip(timestampMode, timestamps1), roundTrip(timestampMode, timestamps2)); + // The two modes are equal as doubles, so they must be compared as longs + assertEquals(timestampMode.extractFinalResult(mergedTimestamps), Long.MAX_VALUE); + assertEquals(createFunction("MIN", "TIMESTAMP").extractFinalResult(mergedTimestamps), Long.MAX_VALUE - 1); + } + + @Test + void nonNumericEmptyIntermediateResult() { + for (String dataType : new String[]{"STRING", "BOOLEAN", "TIMESTAMP"}) { + ModeAggregationFunction function = createFunction("MIN", dataType); + // A server without matching rows sends an empty map + Map<?, Long> empty = + roundTrip(function, function.extractAggregationResult(function.createAggregationResultHolder())); + assertEquals(empty, Map.of()); + assertNull(function.extractFinalResult(empty)); + assertNull(function.extractFinalResult((Map<?, Long>) null)); + } + ModeAggregationFunction stringMode = createFunction("MIN", "STRING"); + Object2LongOpenHashMap<String> strings = new Object2LongOpenHashMap<>(Map.of("a", 1L)); + assertEquals(stringMode.merge(roundTrip(stringMode, new Int2LongOpenHashMap()), strings), strings); + assertEquals(stringMode.merge(strings, new Int2LongOpenHashMap()), strings); + } + + @Test + void nonNumericRejectsUnsupportedArguments() { + assertThrows(IllegalArgumentException.class, () -> createFunction("AVG", "STRING")); + assertThrows(IllegalArgumentException.class, () -> createFunction("MIN", "INT")); + assertThrows(IllegalArgumentException.class, () -> createFunction("MIN", "BYTES")); + } + + private static ModeAggregationFunction createFunction(String reducer, String dataType) { + return new ModeAggregationFunction(List.of(ExpressionContext.forIdentifier("myField"), + ExpressionContext.forLiteral(DataType.STRING, reducer), + ExpressionContext.forLiteral(DataType.STRING, dataType)), false); + } + + private static Map<?, Long> roundTrip(ModeAggregationFunction function, Map<?, Long> intermediateResult) { + SerializedIntermediateResult serialized = function.serializeIntermediateResult(intermediateResult); + return function.deserializeIntermediateResult( + new CustomObject(serialized.getType(), ByteBuffer.wrap(serialized.getBytes()))); + } } diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java index d44bc13a13a..d93623464bb 100644 --- a/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java @@ -43,6 +43,7 @@ public class AggregateFunctionRewriteOptimizerTest { .addSingleValueDimension("bytesCol", DataType.BYTES) .addSingleValueDimension("uuidCol", DataType.UUID) .addMultiValueDimension("mvCol", DataType.INT) + .addMultiValueDimension("mvStringCol", DataType.STRING) .addSingleValueDimension("timeCol", DataType.LONG) .build(); private static final String AUTO_REWRITE = "SET autoRewriteAggregationType=true; "; @@ -88,6 +89,28 @@ public class AggregateFunctionRewriteOptimizerTest { + "ORDER BY LAST_WITH_TIME(timestampCol, timeCol, 'TIMESTAMP')"); } + @Test + public void testModeInfersNonNumericDataType() { + assertRewrite("SELECT MODE(stringCol), MODE(jsonCol), MODE(boolCol, 'MAX'), MODE(timestampCol, 'AVG') " + + "FROM testTable", + "SELECT MODE(stringCol, 'MIN', 'STRING'), MODE(jsonCol, 'MIN', 'STRING'), MODE(boolCol, 'MAX', 'BOOLEAN'), " + + "MODE(timestampCol, 'AVG', 'TIMESTAMP') FROM testTable"); + assertRewrite("SELECT intCol, MODE(stringCol) FILTER (WHERE boolCol) AS m FROM testTable GROUP BY intCol " + + "HAVING MODE(boolCol) = true ORDER BY MODE(timestampCol, 'MAX')", + "SELECT intCol, MODE(stringCol, 'MIN', 'STRING') FILTER (WHERE boolCol) AS m FROM testTable GROUP BY intCol " + + "HAVING MODE(boolCol, 'MIN', 'BOOLEAN') = true ORDER BY MODE(timestampCol, 'MAX', 'TIMESTAMP')"); + } + + @Test + public void testModeKeepsOtherInputs() { + // Numeric MODE keeps its DOUBLE result + assertUnchanged("SELECT MODE(intCol), MODE(longCol, 'AVG'), MODE(floatCol, 'MAX'), MODE(doubleCol) FROM testTable"); + for (String column : new String[]{"bytesCol", "uuidCol", "mvStringCol", "missingCol", "trim(stringCol)"}) { + assertUnchanged("SELECT MODE(" + column + ") FROM testTable"); + } + assertUnchanged("SELECT MODE(stringCol, 'MAX', 'STRING') FROM testTable"); + } + @Test public void testTypedVariantsRequireQueryOption() { String query = "SELECT MIN(stringCol) AS m, MAX(longCol), SUM(intCol), SUM(longCol), MIN(doubleCol) " diff --git a/pinot-core/src/test/java/org/apache/pinot/queries/ModeQueriesTest.java b/pinot-core/src/test/java/org/apache/pinot/queries/ModeQueriesTest.java index b79c9c1fbce..758418731fb 100644 --- a/pinot-core/src/test/java/org/apache/pinot/queries/ModeQueriesTest.java +++ b/pinot-core/src/test/java/org/apache/pinot/queries/ModeQueriesTest.java @@ -21,8 +21,10 @@ package org.apache.pinot.queries; import com.google.common.collect.Lists; import java.io.File; import java.io.IOException; +import java.sql.Timestamp; import java.util.ArrayList; import java.util.Arrays; +import java.util.Comparator; import java.util.HashMap; import java.util.Iterator; import java.util.List; @@ -31,6 +33,9 @@ import java.util.Random; import java.util.stream.Collectors; import org.apache.commons.io.FileUtils; import org.apache.pinot.common.response.broker.BrokerResponseNative; +import org.apache.pinot.common.response.broker.ResultTable; +import org.apache.pinot.common.utils.DataSchema; +import org.apache.pinot.common.utils.DataSchema.ColumnDataType; import org.apache.pinot.common.utils.HashUtil; import org.apache.pinot.core.operator.blocks.results.AggregationResultsBlock; import org.apache.pinot.core.operator.blocks.results.GroupByResultsBlock; @@ -49,6 +54,7 @@ import org.apache.pinot.spi.config.table.TableType; import org.apache.pinot.spi.data.FieldSpec.DataType; import org.apache.pinot.spi.data.Schema; import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.exception.BadQueryRequestException; import org.apache.pinot.spi.utils.ReadMode; import org.apache.pinot.spi.utils.builder.TableConfigBuilder; import org.testng.annotations.AfterClass; @@ -59,6 +65,7 @@ import org.testng.annotations.Test; import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotNull; import static org.testng.Assert.assertTrue; +import static org.testng.Assert.expectThrows; /// Queries test for MODE queries. @@ -80,23 +87,44 @@ public class ModeQueriesTest extends BaseQueriesTest { private static final String LONG_NO_DICT_COLUMN = "longNoDictColumn"; private static final String FLOAT_NO_DICT_COLUMN = "floatNoDictColumn"; private static final String DOUBLE_NO_DICT_COLUMN = "doubleNoDictColumn"; + private static final String STRING_COLUMN = "stringColumn"; + private static final String STRING_NO_DICT_COLUMN = "stringNoDictColumn"; + private static final String JSON_COLUMN = "jsonColumn"; + private static final String BOOLEAN_COLUMN = "booleanColumn"; + private static final String BOOLEAN_NO_DICT_COLUMN = "booleanNoDictColumn"; + private static final String TIMESTAMP_COLUMN = "timestampColumn"; + private static final String TIMESTAMP_NO_DICT_COLUMN = "timestampNoDictColumn"; private static final Schema SCHEMA = new Schema.SchemaBuilder().addSingleValueDimension(INT_COLUMN, DataType.INT) .addMultiValueDimension(INT_MV_COLUMN, DataType.INT).addSingleValueDimension(INT_NO_DICT_COLUMN, DataType.INT) .addSingleValueDimension(LONG_COLUMN, DataType.LONG).addSingleValueDimension(LONG_NO_DICT_COLUMN, DataType.LONG) .addSingleValueDimension(FLOAT_COLUMN, DataType.FLOAT) .addSingleValueDimension(FLOAT_NO_DICT_COLUMN, DataType.FLOAT) .addSingleValueDimension(DOUBLE_COLUMN, DataType.DOUBLE) - .addSingleValueDimension(DOUBLE_NO_DICT_COLUMN, DataType.DOUBLE).build(); + .addSingleValueDimension(DOUBLE_NO_DICT_COLUMN, DataType.DOUBLE) + .addSingleValueDimension(STRING_COLUMN, DataType.STRING) + .addSingleValueDimension(STRING_NO_DICT_COLUMN, DataType.STRING) + .addSingleValueDimension(JSON_COLUMN, DataType.JSON) + .addSingleValueDimension(BOOLEAN_COLUMN, DataType.BOOLEAN) + .addSingleValueDimension(BOOLEAN_NO_DICT_COLUMN, DataType.BOOLEAN) + .addSingleValueDimension(TIMESTAMP_COLUMN, DataType.TIMESTAMP) + .addSingleValueDimension(TIMESTAMP_NO_DICT_COLUMN, DataType.TIMESTAMP).build(); private static final TableConfig TABLE_CONFIG = new TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME) .setNoDictionaryColumns( - Lists.newArrayList(INT_NO_DICT_COLUMN, LONG_NO_DICT_COLUMN, FLOAT_NO_DICT_COLUMN, DOUBLE_NO_DICT_COLUMN)) + Lists.newArrayList(INT_NO_DICT_COLUMN, LONG_NO_DICT_COLUMN, FLOAT_NO_DICT_COLUMN, DOUBLE_NO_DICT_COLUMN, + STRING_NO_DICT_COLUMN, BOOLEAN_NO_DICT_COLUMN, TIMESTAMP_NO_DICT_COLUMN)) .build(); private static final double DELTA = 0.00001; + private static final long TIMESTAMP_BASE = 1_600_000_000_000L; private HashMap<Integer, Long> _values; private Double _expectedResultMin; private Double _expectedResultMax; private Double _expectedResultAvg; + private String _expectedStringMin; + private String _expectedStringMax; + private boolean _expectedBoolean; + private String _expectedTimestampMin; + private String _expectedTimestampMax; private IndexSegment _indexSegment; private List<IndexSegment> _indexSegments; @@ -136,6 +164,13 @@ public class ModeQueriesTest extends BaseQueriesTest { record.putValue(FLOAT_NO_DICT_COLUMN, (float) value); record.putValue(DOUBLE_COLUMN, (double) value); record.putValue(DOUBLE_NO_DICT_COLUMN, (double) value); + record.putValue(STRING_COLUMN, String.valueOf(value)); + record.putValue(STRING_NO_DICT_COLUMN, String.valueOf(value)); + record.putValue(JSON_COLUMN, String.valueOf(value)); + record.putValue(BOOLEAN_COLUMN, value % 3 == 0); + record.putValue(BOOLEAN_NO_DICT_COLUMN, value % 3 == 0); + record.putValue(TIMESTAMP_COLUMN, TIMESTAMP_BASE + value); + record.putValue(TIMESTAMP_NO_DICT_COLUMN, TIMESTAMP_BASE + value); records.add(record); } long maxOccurrences = _values.values().stream().max(Long::compareTo).get(); @@ -144,6 +179,15 @@ public class ModeQueriesTest extends BaseQueriesTest { _expectedResultMin = Arrays.stream(modes).min().getAsDouble(); _expectedResultMax = Arrays.stream(modes).max().getAsDouble(); _expectedResultAvg = Arrays.stream(modes).average().getAsDouble(); + // Ties between string modes are broken in string order, e.g. "10" < "9" + List<String> stringModes = Arrays.stream(modes).mapToObj(mode -> String.valueOf((int) mode)).toList(); + _expectedStringMin = stringModes.stream().min(Comparator.naturalOrder()).get(); + _expectedStringMax = stringModes.stream().max(Comparator.naturalOrder()).get(); + long numTrueValues = _values.entrySet().stream().filter(e -> e.getKey() % 3 == 0).mapToLong(Map.Entry::getValue) + .sum(); + _expectedBoolean = 2 * numTrueValues > NUM_RECORDS; + _expectedTimestampMin = new Timestamp(TIMESTAMP_BASE + _expectedResultMin.longValue()).toString(); + _expectedTimestampMax = new Timestamp(TIMESTAMP_BASE + _expectedResultMax.longValue()).toString(); SegmentGeneratorConfig segmentGeneratorConfig = new SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA); segmentGeneratorConfig.setTableName(RAW_TABLE_NAME); @@ -371,6 +415,104 @@ public class ModeQueriesTest extends BaseQueriesTest { return entries.toArray(new Object[0][]); } + @Test(dataProvider = "testNonNumericAggregationOnlyDataProvider") + public void testNonNumericAggregationOnly(String query, Object[] expectedRow) { + // The optimizer supplies the data type, so the result keeps the column's type + ResultTable resultTable = getBrokerResponseForOptimizedQuery(query, SCHEMA).getResultTable(); + assertEquals(resultTable.getDataSchema(), new DataSchema(new String[]{ + "mode(stringColumn)", "mode(stringNoDictColumn)", "mode(booleanColumn)", "mode(booleanNoDictColumn)", + "mode(timestampColumn)", "mode(timestampNoDictColumn)", "mode(jsonColumn)" + }, new ColumnDataType[]{ + ColumnDataType.STRING, ColumnDataType.STRING, ColumnDataType.BOOLEAN, ColumnDataType.BOOLEAN, + ColumnDataType.TIMESTAMP, ColumnDataType.TIMESTAMP, ColumnDataType.STRING + })); + assertEquals(resultTable.getRows().size(), 1); + assertEquals(resultTable.getRows().get(0), expectedRow); + } + + @DataProvider + public Object[][] testNonNumericAggregationOnlyDataProvider() { + Object[] expectedMinRow = new Object[]{ + _expectedStringMin, _expectedStringMin, _expectedBoolean, _expectedBoolean, _expectedTimestampMin, + _expectedTimestampMin, _expectedStringMin + }; + Object[] expectedMaxRow = new Object[]{ + _expectedStringMax, _expectedStringMax, _expectedBoolean, _expectedBoolean, _expectedTimestampMax, + _expectedTimestampMax, _expectedStringMax + }; + return new Object[][]{ + { + "SELECT MODE(stringColumn), MODE(stringNoDictColumn), MODE(booleanColumn), MODE(booleanNoDictColumn), " + + "MODE(timestampColumn), MODE(timestampNoDictColumn), MODE(jsonColumn) FROM testTable", expectedMinRow + }, { + "SELECT MODE(stringColumn, 'MIN'), MODE(stringNoDictColumn, 'MIN'), MODE(booleanColumn, 'MIN'), " + + "MODE(booleanNoDictColumn, 'MIN'), MODE(timestampColumn, 'MIN'), MODE(timestampNoDictColumn, 'MIN'), " + + "MODE(jsonColumn, 'MIN') FROM testTable", expectedMinRow + }, { + "SELECT MODE(stringColumn, 'MAX'), MODE(stringNoDictColumn, 'MAX'), MODE(booleanColumn, 'MAX'), " + + "MODE(booleanNoDictColumn, 'MAX'), MODE(timestampColumn, 'MAX'), MODE(timestampNoDictColumn, 'MAX'), " + + "MODE(jsonColumn, 'MAX') FROM testTable", expectedMaxRow + } + }; + } + + @Test(dataProvider = "testNonNumericAggregationGroupByDataProvider") + public void testNonNumericAggregationGroupBy(String query) { + ResultTable resultTable = getBrokerResponseForOptimizedQuery(query, SCHEMA).getResultTable(); + assertEquals(resultTable.getDataSchema().getColumnDataTypes(), new ColumnDataType[]{ + ColumnDataType.INT, ColumnDataType.STRING, ColumnDataType.STRING, ColumnDataType.BOOLEAN, + ColumnDataType.BOOLEAN, ColumnDataType.TIMESTAMP, ColumnDataType.TIMESTAMP + }); + List<Object[]> rows = resultTable.getRows(); + assertEquals(rows.size(), _values.size()); + String previous = ""; + for (Object[] row : rows) { + // Each group has a single value per column + int key = (Integer) row[0]; + String timestamp = new Timestamp(TIMESTAMP_BASE + key).toString(); + assertEquals(row, new Object[]{ + key, String.valueOf(key), String.valueOf(key), key % 3 == 0, key % 3 == 0, timestamp, timestamp + }); + // ORDER BY repeats the call, so it must be rewritten the same way as the SELECT list + assertTrue(((String) row[1]).compareTo(previous) > 0); + previous = (String) row[1]; + } + } + + @DataProvider + public Object[][] testNonNumericAggregationGroupByDataProvider() { + return new Object[][]{ + { + "SELECT intColumn, MODE(stringColumn), MODE(stringNoDictColumn), MODE(booleanColumn), " + + "MODE(booleanNoDictColumn, 'MAX'), MODE(timestampColumn), MODE(timestampNoDictColumn, 'MAX') " + + "FROM testTable GROUP BY intColumn ORDER BY MODE(stringColumn) LIMIT 1000" + }, { + "SELECT intMVColumn, MODE(stringColumn), MODE(stringNoDictColumn), MODE(booleanColumn), " + + "MODE(booleanNoDictColumn, 'MAX'), MODE(timestampColumn), MODE(timestampNoDictColumn, 'MAX') " + + "FROM testTable GROUP BY intMVColumn ORDER BY MODE(stringNoDictColumn) LIMIT 1000" + } + }; + } + + @Test + public void testNonNumericEmptyInput() { + String query = "SELECT MODE(stringColumn), MODE(stringNoDictColumn), MODE(booleanColumn), " + + "MODE(timestampNoDictColumn) FROM testTable WHERE intColumn < 0"; + List<Object[]> rows = getBrokerResponseForOptimizedQuery(query, SCHEMA).getResultTable().getRows(); + assertEquals(rows.size(), 1); + assertEquals(rows.get(0), new Object[]{null, null, null, null}); + } + + @Test + public void testNonNumericRejectsAvg() { + for (String column : new String[]{STRING_COLUMN, BOOLEAN_COLUMN, TIMESTAMP_COLUMN}) { + String query = "SELECT MODE(" + column + ", 'AVG') FROM testTable"; + BadQueryRequestException exception = + expectThrows(BadQueryRequestException.class, () -> getBrokerResponseForOptimizedQuery(query, SCHEMA)); + assertTrue(exception.getMessage().contains("Unsupported MODE reducer for"), exception.getMessage()); + } + } + @AfterClass public void tearDown() throws IOException { diff --git a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java index dc4f8455363..64b712a3e1f 100644 --- a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java +++ b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java @@ -43,9 +43,10 @@ import org.apache.pinot.segment.spi.AggregationFunctionType; /// Supplies the explicit data type argument from the type of the data argument: /// - FIRST_WITH_TIME(dataColumn, timeColumn) -> FIRST_WITH_TIME(dataColumn, timeColumn, 'dataType') /// - LAST_WITH_TIME(dataColumn, timeColumn) -> LAST_WITH_TIME(dataColumn, timeColumn, 'dataType') +/// - MODE(stringColumn) -> MODE(stringColumn, 'MIN', 'STRING') /// -/// The result type of the call is unchanged. The literal is projected directly beneath the aggregate, appended to the -/// existing project when there is one, because [PinotAggregateExchangeNodeInsertRule] and `AggregatePushdownRule` +/// The result type of the call is unchanged. The literals are projected directly beneath the aggregate, appended to +/// the existing project when there is one, because [PinotAggregateExchangeNodeInsertRule] and `AggregatePushdownRule` /// inline literal arguments only from the immediate project. public class PinotAggregateTypeArgumentRewriteRule extends RelOptRule { public static final PinotAggregateTypeArgumentRewriteRule INSTANCE = @@ -66,7 +67,7 @@ public class PinotAggregateTypeArgumentRewriteRule extends RelOptRule { // Requiring a call without its data type is also what makes the rule terminate. Aggregate aggRel = call.rel(0); RelNode input = aggRel.getInput(); - return aggRel.getAggCallList().stream().anyMatch(aggCall -> getDataTypeArgument(aggCall, input) != null); + return aggRel.getAggCallList().stream().anyMatch(aggCall -> getLiteralArguments(aggCall, input) != null); } @Override @@ -96,47 +97,54 @@ public class PinotAggregateTypeArgumentRewriteRule extends RelOptRule { // Appending keeps the existing field indexes, so the group set and the other calls are unaffected. List<AggregateCall> orgAggCalls = aggRel.getAggCallList(); - int numAggCalls = orgAggCalls.size(); - int[] literalIndexes = new int[numAggCalls]; - Map<String, Integer> literalIndexByDataType = new HashMap<>(); - for (int i = 0; i < numAggCalls; i++) { - String dataType = getDataTypeArgument(orgAggCalls.get(i), input); - literalIndexes[i] = dataType == null ? -1 : literalIndexByDataType.computeIfAbsent(dataType, type -> { - projects.add(rexBuilder.makeLiteral(type)); - return projects.size() - 1; - }); - } - RelNode newInput = LogicalProject.create(projectInput, hints, projects, (List<String>) null, Set.of()); - - List<AggregateCall> aggCalls = new ArrayList<>(numAggCalls); - for (int i = 0; i < numAggCalls; i++) { - AggregateCall aggCall = orgAggCalls.get(i); - if (literalIndexes[i] < 0) { + List<AggregateCall> aggCalls = new ArrayList<>(orgAggCalls.size()); + Map<String, Integer> literalIndexes = new HashMap<>(); + for (AggregateCall aggCall : orgAggCalls) { + List<String> literalArguments = getLiteralArguments(aggCall, input); + if (literalArguments == null) { aggCalls.add(aggCall); continue; } List<Integer> argList = new ArrayList<>(aggCall.getArgList()); - argList.add(literalIndexes[i]); + for (String literalArgument : literalArguments) { + argList.add(literalIndexes.computeIfAbsent(literalArgument, literal -> { + projects.add(rexBuilder.makeLiteral(literal)); + return projects.size() - 1; + })); + } // Keeps the original result type, so the rewrite does not change the result schema. aggCalls.add(aggCall.withArgList(argList)); } + RelNode newInput = LogicalProject.create(projectInput, hints, projects, (List<String>) null, Set.of()); call.transformTo( aggRel.copy(aggRel.getTraitSet(), newInput, aggRel.getGroupSet(), aggRel.getGroupSets(), aggCalls)); } - /// Returns the data type argument to append, or `null` when the call already has one or is not such a call. + /// Returns the literal arguments to append, or `null` when the call already has them or is not such a call. @Nullable - private static String getDataTypeArgument(AggregateCall aggCall, RelNode input) { - if (aggCall.getArgList().size() != 2) { - return null; - } + private static List<String> getLiteralArguments(AggregateCall aggCall, RelNode input) { + List<Integer> argList = aggCall.getArgList(); String name = AggregationFunctionType.getNormalizedAggregationFunctionName(aggCall.getAggregation().getName()); - if (!name.equals(AggregationFunctionType.FIRSTWITHTIME.name()) - && !name.equals(AggregationFunctionType.LASTWITHTIME.name())) { - return null; + if (name.equals(AggregationFunctionType.MODE.name())) { + // BOOLEAN and TIMESTAMP already return their own type, cast from the DOUBLE result. + return argList.size() == 1 && SqlTypeName.CHAR_TYPES.contains(getSqlTypeName(input, argList.get(0))) + ? List.of("MIN", "STRING") : null; } - SqlTypeName dataType = - input.getRowType().getFieldList().get(aggCall.getArgList().get(0)).getType().getSqlTypeName(); + if (argList.size() == 2 && (name.equals(AggregationFunctionType.FIRSTWITHTIME.name()) + || name.equals(AggregationFunctionType.LASTWITHTIME.name()))) { + String dataType = getDataTypeArgument(getSqlTypeName(input, argList.get(0))); + return dataType != null ? List.of(dataType) : null; + } + return null; + } + + private static SqlTypeName getSqlTypeName(RelNode input, int fieldIndex) { + return input.getRowType().getFieldList().get(fieldIndex).getType().getSqlTypeName(); + } + + /// Returns the data type argument of FIRST_WITH_TIME / LAST_WITH_TIME, or `null` when the type is not supported. + @Nullable + private static String getDataTypeArgument(SqlTypeName dataType) { switch (dataType) { case TINYINT: case SMALLINT: diff --git a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json index c10f7185b79..ed55bef43f2 100644 --- a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json +++ b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json @@ -189,6 +189,32 @@ "sql": "EXPLAIN PLAN FOR SELECT FIRST_WITH_TIME(a.col4, a.ts) FROM a", "expectedException": ".*Unsupported data type for .*\\(dataColumn, timeColumn\\): DECIMAL.*" }, + { + "description": "MODE on a STRING argument gets the tie reducer and data type; numeric, BOOLEAN and TIMESTAMP MODE are unchanged", + "sql": "EXPLAIN PLAN FOR SELECT MODE(a.col1), MODE(a.col3), MODE(a.col5), MODE(a.ts_timestamp), MODE(UPPER(a.col2)) FROM a", + "output": [ + "Execution Plan", + "\nPinotLogicalAggregate(group=[{}], agg#0=[MODE($0, _UTF-8'MIN', _UTF-8'STRING')], agg#1=[MODE($1)], agg#2=[MODE($2)], agg#3=[MODE($3)], agg#4=[MODE($4, _UTF-8'MIN', _UTF-8'STRING')], aggType=[FINAL])", + "\n PinotLogicalExchange(distribution=[hash])", + "\n PinotLogicalAggregate(group=[{}], agg#0=[MODE($0, _UTF-8'MIN', _UTF-8'STRING')], agg#1=[MODE($1)], agg#2=[MODE($2)], agg#3=[MODE($3)], agg#4=[MODE($4, _UTF-8'MIN', _UTF-8'STRING')], aggType=[LEAF])", + "\n LogicalProject($f0=[$0], $f1=[$2], $f2=[$4], $f3=[$8], $f4=[UPPER($1)], $f5=[_UTF-8'MIN'], $f6=[_UTF-8'STRING'])", + "\n PinotLogicalTableScan(table=[[default, a]])", + "\n" + ] + }, + { + "description": "MODE and FIRST_WITH_TIME share the projected data type literal in a grouped aggregate", + "sql": "EXPLAIN PLAN FOR SELECT a.col3, MODE(a.col2), FIRST_WITH_TIME(a.col1, a.ts) FROM a GROUP BY a.col3", + "output": [ + "Execution Plan", + "\nPinotLogicalAggregate(group=[{0}], agg#0=[MODE($1, _UTF-8'MIN', _UTF-8'STRING')], agg#1=[FIRSTWITHTIME($2, $2, _UTF-8'STRING')], aggType=[FINAL])", + "\n PinotLogicalExchange(distribution=[hash[0]])", + "\n PinotLogicalAggregate(group=[{0}], agg#0=[MODE($1, _UTF-8'MIN', _UTF-8'STRING')], agg#1=[FIRSTWITHTIME($2, $3, _UTF-8'STRING')], aggType=[LEAF])", + "\n LogicalProject($f0=[$2], $f1=[$1], $f2=[$0], $f3=[$7], $f4=[_UTF-8'MIN'], $f5=[_UTF-8'STRING'])", + "\n PinotLogicalTableScan(table=[[default, a]])", + "\n" + ] + }, { "description": "MIN on STRING column rewritten to MINSTRING", "sql": "SET autoRewriteAggregationType=true; EXPLAIN PLAN FOR SELECT MIN(a.col1) FROM a", diff --git a/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json b/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json index 01a19d7837c..ffc9a51c59e 100644 --- a/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json +++ b/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json @@ -21,6 +21,23 @@ [150, 150, 1.5, 1.5, "c", false, "1E+317", "2020-02-03 03:32:12"], [175, 175, 1.75, 1.75, "c", true, "1E+328", "2020-01-02 03:32:12"] ] + }, + "partitioned_tbl": { + "schema": [ + {"name": "num", "type": "INT"}, + {"name": "name", "type": "STRING"} + ], + "inputs": [ + [1, "a"], + [3, "c"], + [3, "yyy"], + [4, "e"], + [4, "e"], + [4, "f"], + [7, "f"], + [7, "d"] + ], + "partitionColumns": ["num"] } }, "queries": [ @@ -77,6 +94,28 @@ "sql": "SELECT FIRST_WITH_TIME(timestamp_col, timestamp_col), LAST_WITH_TIME(timestamp_col, timestamp_col), FIRST_WITH_TIME(timestamp_col, timestamp_col, 'TIMESTAMP') FROM {tbl}", "outputs": [["2020-01-01 03:10:12.0", "2020-05-01 03:32:12.0", "2020-01-01 03:10:12.0"]] }, + { + "comment": "MODE returns the type of a non-numeric argument, with ties broken by the smallest value", + "sql": "SELECT MODE(string_col), MODE(UPPER(string_col)), MODE(bool_col), MODE(timestamp_col), MODE(int_col) FROM {tbl}", + "outputs": [["c", "C", false, "2020-01-01 03:10:12.0", 2]] + }, + { + "sql": "SELECT bool_col, MODE(string_col), MODE(timestamp_col) FROM {tbl} GROUP BY bool_col", + "outputs": [[false, "b", "2020-01-01 08:32:12.0"], [true, "a", "2020-01-01 03:10:12.0"]] + }, + { + "sql": "SELECT /*+ aggOptions(is_skip_leaf_stage_aggregate='true') */ int_col, MODE(string_col), MODE(bool_col), MODE(timestamp_col) FROM {tbl} GROUP BY int_col", + "outputs": [[2, "a", true, "2020-01-01 03:10:12.0"], [3, "b", false, "2020-01-01 08:32:12.0"], [100, "b", false, "2020-02-06 03:32:12.0"], [101, "c", false, "2020-05-01 03:32:12.0"], [150, "c", false, "2020-02-03 03:32:12.0"], [175, "c", true, "2020-01-02 03:32:12.0"]] + }, + { + "sql": "SELECT MODE(string_col) FROM {tbl} WHERE int_col < 0", + "outputs": [[null]] + }, + { + "comment": "The leaf returns the final MODE, which keeps the STRING type", + "sql": "SELECT /*+ aggOptions(is_leaf_return_final_result='true') */ {partitioned_tbl}.num, MODE({partitioned_tbl}.name) FROM {partitioned_tbl} /*+ tableOptions(partition_function='hashcode', partition_key='num', partition_size='4') */ GROUP BY {partitioned_tbl}.num", + "outputs": [[1, "a"], [3, "c"], [4, "e"], [7, "d"]] + }, { "sql": "SELECT SUMPRECISION(decimal_col) FROM {tbl}", "outputs": [["10000000000100000000110000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000.00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000 [...] diff --git a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java index bbd582c6cb0..acd44aedd52 100644 --- a/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java +++ b/pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java @@ -63,7 +63,7 @@ public enum AggregationFunctionType { SUMLONG("sumLong", ReturnTypes.AGG_SUM, OperandTypes.or(OperandTypes.INTEGER, OperandTypes.ARRAY_OF_INTEGER)), SUMPRECISION("sumPrecision", ReturnTypes.explicit(SqlTypeName.DECIMAL), OperandTypes.ANY, SqlTypeName.OTHER), AVG("avg", SqlTypeName.OTHER, SqlTypeName.DOUBLE), - MODE("mode", SqlTypeName.OTHER, SqlTypeName.DOUBLE), + MODE("mode", null, null, ReturnTypes.explicit(SqlTypeName.OTHER), new ModeFinalReturnTypeInference()), ANYVALUE("anyValue", ReturnTypes.ARG0, OperandTypes.ANY, SqlTypeName.OTHER), FIRSTWITHTIME("firstWithTime", new ValueWithTimeReturnTypeInference(), OperandTypes.or( OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY), @@ -447,6 +447,15 @@ public enum AggregationFunctionType { } } + /// Returns DOUBLE, or the type of the data argument for the explicit data type form MODE(expr, reducer, dataType). + private static class ModeFinalReturnTypeInference implements SqlReturnTypeInference { + @Override + public RelDataType inferReturnType(SqlOperatorBinding opBinding) { + return opBinding.getOperandCount() == 3 ? ReturnTypes.ARG0.inferReturnType(opBinding) + : opBinding.getTypeFactory().createSqlType(SqlTypeName.DOUBLE); + } + } + private static class ArrayReturnTypeInference implements SqlReturnTypeInference { final SqlTypeName _sqlTypeName; diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java index 6fc50e1c84d..375b17dc1b0 100644 --- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java +++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java @@ -1291,7 +1291,7 @@ public class CommonConstants { public static final String AGGREGATE_UNION_TRANSPOSE = "AggregateUnionTranspose"; public static final String AGGREGATE_REDUCE_FUNCTIONS = "AggregateReduceFunctions"; public static final String AGGREGATE_FUNCTION_REWRITE = "AggregateFunctionRewrite"; - /// Supplies the data type argument of FIRST_WITH_TIME / LAST_WITH_TIME(dataColumn, timeColumn). + /// Supplies the data type argument of FIRST_WITH_TIME / LAST_WITH_TIME(dataColumn, timeColumn) and MODE(string). public static final String AGGREGATE_TYPE_ARGUMENT_REWRITE = "AggregateTypeArgumentRewrite"; /// Inert unless [CommonConstants.Broker#USE_APPROXIMATE_FUNCTION] is on; name it here to switch it off entirely. public static final String APPROXIMATE_AGGREGATE_REWRITE = "ApproximateAggregateRewrite"; --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
