This is an automated email from the ASF dual-hosted git repository.
xiangfu0 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 7835c7dc378 Support MODE on STRING, BOOLEAN and TIMESTAMP inputs
(#19991)
7835c7dc378 is described below
commit 7835c7dc378c21738bb94ecedf89161add936417
Author: Xiang Fu <[email protected]>
AuthorDate: Mon Oct 26 19:18:49 2026 -0700
Support MODE on STRING, BOOLEAN and TIMESTAMP inputs (#19991)
* 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
explicit form requires the expression's stored type to match the data type.
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.
* Use the null-safe ORDER BY comparator for aggregation values
Aggregation results can be null with null handling disabled too, e.g.
MINSTRING or a typed MODE over an empty group, so ORDER BY on them failed
with an NPE in TableResizer. Also use the is_skip_leaf_stage_group_by hint
in
the MODE runtime case.
* Note where MODE's query result type comes from
* Preserve legacy BOOLEAN and TIMESTAMP MODE calls by default
---
.../apache/pinot/core/common/ObjectSerDeUtils.java | 53 +++++-
.../apache/pinot/core/data/table/TableResizer.java | 16 +-
.../function/ModeAggregationFunction.java | 201 +++++++++++++++++---
.../AggregateFunctionRewriteOptimizer.java | 28 +++
.../query/reduce/AggregationDataTableReducer.java | 5 +-
.../pinot/core/common/ObjectSerDeUtilsTest.java | 17 ++
.../function/ModeAggregationFunctionTest.java | 181 ++++++++++++++++++
.../AggregateFunctionRewriteOptimizerTest.java | 25 +++
.../org/apache/pinot/queries/ModeQueriesTest.java | 210 ++++++++++++++++++++-
.../PinotAggregateTypeArgumentRewriteRule.java | 55 +++---
.../org/apache/pinot/query/QueryEnvironment.java | 3 +-
.../src/test/resources/queries/AggregatePlans.json | 26 +++
.../src/test/resources/queries/UDFAggregates.json | 41 ++++
.../pinot/segment/spi/AggregationFunctionType.java | 12 +-
14 files changed, 818 insertions(+), 55 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 4259bd3873b..b72af2c0717 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;
@@ -177,7 +179,8 @@ public class ObjectSerDeUtils {
BigDecimalSet(53),
BytesArrayList(54),
BigDecimalLongPair(55),
- BytesLongPair(56);
+ BytesLongPair(56),
+ String2LongMap(57);
private final int _value;
@@ -1680,6 +1683,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
@@ -2134,6 +2184,7 @@ public class ObjectSerDeUtils {
BYTES_ARRAY_LIST_SER_DE,
BIG_DECIMAL_LONG_PAIR_SER_DE,
BYTES_LONG_PAIR_SER_DE,
+ STRING_2_LONG_MAP_SER_DE,
};
//@formatter:on
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/data/table/TableResizer.java
b/pinot-core/src/main/java/org/apache/pinot/core/data/table/TableResizer.java
index fdfd1b1ecd4..15d4660ef5f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/data/table/TableResizer.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/data/table/TableResizer.java
@@ -104,16 +104,22 @@ public class TableResizer {
_orderByValueExtractors = new
OrderByValueExtractor[_numOrderByExpressions];
Comparator[] comparators = new Comparator[_numOrderByExpressions];
int[] nullComparisonResults = new int[_numOrderByExpressions];
+ /// Grouping-set queries produce genuine null group-key values for
rolled-up columns regardless of the
+ /// user's null-handling option, so ORDER BY over a grouping column must
use the null-safe comparator.
+ boolean nullAwareComparison =
queryContext.requiresNullAwareKeyEvaluation();
for (int i = 0; i < _numOrderByExpressions; i++) {
OrderByExpressionContext orderByExpression = orderByExpressions.get(i);
- _orderByValueExtractors[i] =
getOrderByValueExtractor(orderByExpression.getExpression());
+ ExpressionContext expression = orderByExpression.getExpression();
+ _orderByValueExtractors[i] = getOrderByValueExtractor(expression);
comparators[i] = orderByExpression.isAsc() ? Comparator.naturalOrder() :
Comparator.reverseOrder();
nullComparisonResults[i] = orderByExpression.isNullsLast() ? -1 : 1;
+ // Aggregation results can be null with null handling disabled too, e.g.
MINSTRING over an empty group.
+ if (expression.getType() != ExpressionContext.Type.LITERAL &&
!_groupByExpressionIndexMap.containsKey(
+ expression)) {
+ nullAwareComparison = true;
+ }
}
- /// Grouping-set queries produce genuine null group-key values for
rolled-up columns regardless of the
- /// user's null-handling option, so ORDER BY over a grouping column must
use the null-safe comparator.
- boolean nullHandlingEnabled =
queryContext.requiresNullAwareKeyEvaluation();
- if (nullHandlingEnabled) {
+ if (nullAwareComparison) {
_intermediateRecordComparator = (o1, o2) -> {
for (int i = 0; i < _numOrderByExpressions; i++) {
Object v1 = o1._values[i];
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..56b1ec7acaa 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,71 @@ 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. The expression must have the
+/// same stored 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 of the values, which must match the result type
when it is given. BOOLEAN and TIMESTAMP
+ /// values are stored as INT and LONG.
+ private DataType getStoredType(DataType valueType) {
+ DataType storedType = valueType.getStoredType();
+ if (_resultType != ColumnDataType.DOUBLE) {
+ Preconditions.checkState(storedType ==
_resultType.getStoredType().toDataType(),
+ "Illegal data type for MODE aggregation function with %s result:
%s", _resultType, valueType);
+ } else {
+ 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 +128,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 +185,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,12 +218,12 @@ 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) {
+ 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);
@@ -201,13 +253,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.put(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 +316,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 +355,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 +388,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 +422,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 +455,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 +497,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 +545,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 +563,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 +576,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 +587,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 +816,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 e7ed844ff79..518fefc9881 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
@@ -36,6 +36,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(stringColumn[, 'MIN'|'MAX']) -> MODE(stringColumn, 'MIN'|'MAX',
'STRING')
///
/// Applied only when enabled through the `autoRewriteAggregationType` query
option:
/// - MIN(stringType) -> MINSTRING
@@ -98,6 +99,10 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
addDataTypeArgument(function, normalizedName, fieldSpec);
return;
}
+ if (normalizedName.equals(AggregationFunctionType.MODE.name()) &&
function.getOperandsSize() <= 2) {
+ addModeDataTypeArgument(function, fieldSpec);
+ return;
+ }
if (rewriteToTypedVariants) {
// The typed variants follow the stored type (e.g. TIMESTAMP as LONG),
while the data type argument above keeps
// the logical type.
@@ -134,6 +139,29 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
functionName, fieldSpec.isSingleValueField() ? "" : "multi-value ",
fieldSpec.getDataType()));
}
+ /// Appends the default tie reducer when missing, then STRING for a STRING
or JSON column.
+ private static void addModeDataTypeArgument(Function function, FieldSpec
fieldSpec) {
+ if (!fieldSpec.isSingleValueField()) {
+ return;
+ }
+ String dataType;
+ switch (fieldSpec.getDataType()) {
+ case STRING:
+ 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 d02250c778c..27b71ab841a 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.math.BigDecimal;
@@ -684,6 +685,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 1051598be29..ffef0397a46 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
@@ -46,6 +46,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; ";
@@ -101,6 +102,30 @@ 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'), "
+ + "MODE(timestampCol, 'AVG') 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) = true ORDER BY MODE(timestampCol,
'MAX')");
+ }
+
+ @Test
+ public void testModeKeepsOtherInputs() {
+ // Numeric, BOOLEAN and TIMESTAMP MODE keep their DOUBLE results
+ assertUnchanged("SELECT MODE(intCol), MODE(longCol, 'AVG'), MODE(floatCol,
'MAX'), MODE(doubleCol) FROM testTable");
+ for (String column : new String[]{
+ "boolCol", "timestampCol", "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..d6948dafe7d 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;
@@ -57,8 +63,10 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.expectThrows;
/// Queries test for MODE queries.
@@ -80,23 +88,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 +165,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 +180,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 +416,167 @@ public class ModeQueriesTest extends BaseQueriesTest {
return entries.toArray(new Object[0][]);
}
+ @Test(dataProvider = "testNonNumericAggregationOnlyDataProvider")
+ public void testNonNumericAggregationOnly(String query, Object[]
expectedRow) {
+ // STRING is inferred; BOOLEAN and TIMESTAMP use the explicit typed form
+ 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, 'MIN', 'BOOLEAN'), "
+ + "MODE(booleanNoDictColumn, 'MIN', 'BOOLEAN'),
MODE(timestampColumn, 'MIN', 'TIMESTAMP'), "
+ + "MODE(timestampNoDictColumn, 'MIN', 'TIMESTAMP'),
MODE(jsonColumn) FROM testTable", expectedMinRow
+ }, {
+ "SELECT MODE(stringColumn, 'MIN'), MODE(stringNoDictColumn,
'MIN'), MODE(booleanColumn, 'MIN', 'BOOLEAN'), "
+ + "MODE(booleanNoDictColumn, 'MIN', 'BOOLEAN'),
MODE(timestampColumn, 'MIN', 'TIMESTAMP'), "
+ + "MODE(timestampNoDictColumn, 'MIN', 'TIMESTAMP'), "
+ + "MODE(jsonColumn, 'MIN') FROM testTable", expectedMinRow
+ }, {
+ "SELECT MODE(stringColumn, 'MAX'), MODE(stringNoDictColumn,
'MAX'), MODE(booleanColumn, 'MAX', 'BOOLEAN'), "
+ + "MODE(booleanNoDictColumn, 'MAX', 'BOOLEAN'),
MODE(timestampColumn, 'MAX', 'TIMESTAMP'), "
+ + "MODE(timestampNoDictColumn, 'MAX', 'TIMESTAMP'), "
+ + "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, 'MIN', 'BOOLEAN'), "
+ + "MODE(booleanNoDictColumn, 'MAX', 'BOOLEAN'),
MODE(timestampColumn, 'MIN', 'TIMESTAMP'), "
+ + "MODE(timestampNoDictColumn, 'MAX', 'TIMESTAMP') "
+ + "FROM testTable GROUP BY intColumn ORDER BY
MODE(stringColumn) LIMIT 1000"
+ }, {
+ "SELECT intMVColumn, MODE(stringColumn), MODE(stringNoDictColumn),
MODE(booleanColumn, 'MIN', 'BOOLEAN'), "
+ + "MODE(booleanNoDictColumn, 'MAX', 'BOOLEAN'),
MODE(timestampColumn, 'MIN', 'TIMESTAMP'), "
+ + "MODE(timestampNoDictColumn, 'MAX', 'TIMESTAMP') "
+ + "FROM testTable GROUP BY intMVColumn ORDER BY
MODE(stringNoDictColumn) LIMIT 1000"
+ }
+ };
+ }
+
+ @Test
+ public void testBooleanTimestampKeepLegacySemantics() {
+ String query = "SELECT intColumn, MODE(booleanColumn),
MODE(timestampColumn, 'AVG'), "
+ + "MODE(timestampColumn) + 1, CAST(MODE(booleanColumn) AS STRING) FROM
testTable "
+ + "GROUP BY intColumn HAVING MODE(booleanColumn) = 1.0 ORDER BY
intColumn LIMIT 1000";
+ ResultTable table = getBrokerResponseForOptimizedQuery(query,
SCHEMA).getResultTable();
+ assertEquals(table.getDataSchema().getColumnDataTypes(), new
ColumnDataType[]{
+ ColumnDataType.INT, ColumnDataType.DOUBLE, ColumnDataType.DOUBLE,
ColumnDataType.DOUBLE, ColumnDataType.OBJECT
+ });
+ List<Integer> expectedKeys = _values.keySet().stream().filter(key -> key %
3 == 0).sorted().toList();
+ assertEquals(table.getRows().size(), expectedKeys.size());
+ for (int i = 0; i < expectedKeys.size(); i++) {
+ int key = expectedKeys.get(i);
+ assertEquals(table.getRows().get(i), new Object[]{
+ key, 1.0, (double) (TIMESTAMP_BASE + key), (double) (TIMESTAMP_BASE
+ key + 1), "1.0"
+ });
+ }
+ }
+
+ @Test
+ public void testNonNumericEmptyInput() {
+ String query = "SELECT MODE(stringColumn), MODE(stringNoDictColumn),
MODE(booleanColumn, 'MIN', 'BOOLEAN'), "
+ + "MODE(timestampNoDictColumn, 'MIN', 'TIMESTAMP') 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 dataType = SCHEMA.getFieldSpecFor(column).getDataType().name();
+ String query = "SELECT MODE(" + column + ", 'AVG', '" + dataType + "')
FROM testTable";
+ BadQueryRequestException exception =
+ expectThrows(BadQueryRequestException.class, () ->
getBrokerResponseForOptimizedQuery(query, SCHEMA));
+ assertTrue(exception.getMessage().contains("Unsupported MODE reducer
for"), exception.getMessage());
+ }
+ }
+
+ @Test
+ public void testNonNumericRejectsMismatchedDataType() {
+ for (String query : new String[]{
+ "SELECT MODE(intColumn, 'MIN', 'STRING') FROM testTable",
+ "SELECT MODE(longNoDictColumn, 'MIN', 'BOOLEAN') FROM testTable",
+ "SELECT MODE(stringNoDictColumn, 'MIN', 'TIMESTAMP') FROM testTable"
+ }) {
+ BrokerResponseNative brokerResponse = getBrokerResponse(query);
+ assertFalse(brokerResponse.getExceptions().isEmpty(), query);
+ String message = brokerResponse.getExceptions().get(0).getMessage();
+ assertTrue(message.contains("Illegal data type for MODE aggregation
function"), message);
+ }
+ }
+
+ @Test
+ public void testNonNumericOrderByEmptyGroup() {
+ // With null handling disabled, MODE of an empty group is null, which
sorts last for ASC and first for DESC
+ for (boolean asc : new boolean[]{true, false}) {
+ String query = "SELECT intColumn, MODE(booleanColumn, 'MIN', 'BOOLEAN')
FILTER (WHERE intColumn < 5) "
+ + "FROM testTable GROUP BY intColumn "
+ + "ORDER BY MODE(booleanColumn, 'MIN', 'BOOLEAN') FILTER (WHERE
intColumn < 5) " + (asc ? "ASC" : "DESC")
+ + ", intColumn LIMIT 3";
+ Comparator<Boolean> modeOrder = asc ?
Comparator.nullsLast(Comparator.<Boolean>naturalOrder())
+ : Comparator.nullsFirst(Comparator.<Boolean>reverseOrder());
+ List<Integer> expectedKeys = _values.keySet().stream()
+ .sorted(Comparator.comparing((Integer key) -> key < 5 ? key % 3 == 0
: null, modeOrder)
+ .thenComparing(Comparator.naturalOrder()))
+ .limit(3)
+ .toList();
+ List<Object[]> rows = getBrokerResponseForOptimizedQuery(query,
SCHEMA).getResultTable().getRows();
+ assertEquals(rows.size(), 3, query);
+ for (int i = 0; i < 3; i++) {
+ int key = expectedKeys.get(i);
+ assertEquals(rows.get(i), new Object[]{key, key < 5 ? (Object) (key %
3 == 0) : null}, query);
+ }
+ }
+ }
+
@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 aa739af7230..a2b2c2cccd4 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.
///
/// The servers require the data type argument, so the rule is required rather
than an optimization: `QueryEnvironment`
@@ -65,7 +66,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,18 +97,20 @@ 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();
List<AggregateCall> aggCalls = new ArrayList<>(orgAggCalls.size());
- Map<String, Integer> literalIndexByDataType = new HashMap<>();
+ Map<String, Integer> literalIndexes = new HashMap<>();
for (AggregateCall aggCall : orgAggCalls) {
- String dataType = getDataTypeArgument(aggCall, input);
- if (dataType == null) {
+ List<String> literalArguments = getLiteralArguments(aggCall, input);
+ if (literalArguments == null) {
aggCalls.add(aggCall);
continue;
}
List<Integer> argList = new ArrayList<>(aggCall.getArgList());
- argList.add(literalIndexByDataType.computeIfAbsent(dataType, type -> {
- projects.add(rexBuilder.makeLiteral(type));
- return projects.size() - 1;
- }));
+ 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));
}
@@ -116,20 +119,30 @@ public class PinotAggregateTypeArgumentRewriteRule
extends RelOptRule {
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. Throws
- /// for a data argument without an explicit form, such as an array from a
multi-value column.
+ /// 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()))) {
+ return List.of(getDataTypeArgument(name, getSqlTypeName(input,
argList.get(0))));
+ }
+ 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.
Throws for a data argument without an
+ /// explicit form, such as an array from a multi-value column.
+ private static String getDataTypeArgument(String name, SqlTypeName dataType)
{
switch (dataType) {
case TINYINT:
case SMALLINT:
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
index 31531c92139..da842624029 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/QueryEnvironment.java
@@ -605,7 +605,8 @@ public class QueryEnvironment {
for (RelOptRule relOptRule : basicRules) {
hepProgramBuilder.addRuleInstance(relOptRule);
}
- // Required to plan FIRST_WITH_TIME / LAST_WITH_TIME without a data type,
so the rule options do not filter it.
+ // Required to plan FIRST_WITH_TIME / LAST_WITH_TIME without a data type
and MODE on strings, so the rule options
+ // do not filter it.
hepProgramBuilder.addRuleInstance(PinotAggregateTypeArgumentRewriteRule.INSTANCE);
// ----
diff --git a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
index f1dbfd4813e..3be6c9e0d64 100644
--- a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
+++ b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
@@ -197,6 +197,32 @@
"\n"
]
},
+ {
+ "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 ca555e963d8..816e9e2b637 100644
--- a/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
+++ b/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
@@ -23,6 +23,25 @@
[150, 150, 1.5, 1.5, "c", false, "1E+317", "2020-02-03 03:32:12",
"DEADBE01", "00000000-0000-0000-0000-000000000006"],
[175, 175, 1.75, 1.75, "c", true, "1E+328", "2020-01-02 03:32:12",
"DEADBE02", "00000000-0000-0000-0000-000000000007"]
]
+ },
+ "partitioned_tbl": {
+ "schema": [
+ {"name": "num", "type": "INT"},
+ {"name": "name", "type": "STRING"},
+ {"name": "flag", "type": "BOOLEAN"},
+ {"name": "event_ts", "type": "TIMESTAMP"}
+ ],
+ "inputs": [
+ [1, "a", true, "2020-01-01 00:00:00"],
+ [3, "c", false, "2020-01-02 00:00:00"],
+ [3, "yyy", true, "2020-01-03 00:00:00"],
+ [4, "e", true, "2020-01-04 00:00:00"],
+ [4, "e", false, "2020-01-04 00:00:00"],
+ [4, "f", false, "2020-01-05 00:00:00"],
+ [7, "f", true, "2020-01-07 00:00:00"],
+ [7, "d", true, "2020-01-06 00:00:00"]
+ ],
+ "partitionColumns": ["num"]
}
},
"queries": [
@@ -108,6 +127,28 @@
"sql": "SELECT /*+ aggOptions(is_skip_leaf_stage_aggregate='true') */
string_col, FIRST_WITH_TIME(decimal_col, timestamp_col),
LAST_WITH_TIME(decimal_col, timestamp_col), FIRST_WITH_TIME(bytes_col,
timestamp_col), LAST_WITH_TIME(bytes_col, timestamp_col) FROM {tbl} GROUP BY
string_col",
"outputs": [["a", "1E-307", "1E+307", "deadbeef", "fdeadbee"], ["b",
"1E-308", "1E+308", "efdeadbe", "beefdead"], ["c", "1E+328", "1E-317",
"deadbe02", "adbeefde"]]
},
+ {
+ "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_group_by='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 input
type",
+ "sql": "SELECT /*+ aggOptions(is_leaf_return_final_result='true') */
{partitioned_tbl}.num, MODE({partitioned_tbl}.name),
MODE({partitioned_tbl}.flag), MODE({partitioned_tbl}.event_ts) FROM
{partitioned_tbl} /*+ tableOptions(partition_function='hashcode',
partition_key='num', partition_size='4') */ GROUP BY {partitioned_tbl}.num",
+ "outputs": [[1, "a", true, "2020-01-01 00:00:00.0"], [3, "c", false,
"2020-01-02 00:00:00.0"], [4, "e", false, "2020-01-04 00:00:00.0"], [7, "d",
true, "2020-01-06 00:00:00.0"]]
+ },
{
"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 7534ec713f6..06907a7b730 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", ReturnTypes.ARG0, OperandTypes.or(
OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY),
@@ -428,6 +428,16 @@ public enum AggregationFunctionType {
}
}
+ /// Returns DOUBLE, or the type of the data argument for the explicit data
type form MODE(expr, reducer, dataType).
+ /// Only a leaf stage that returns final results uses it; the query's result
type comes from Calcite's MODE (ARG0).
+ 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;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
