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 dc759c51e66 Support FIRST_WITH_TIME and LAST_WITH_TIME without a data
type (#19986)
dc759c51e66 is described below
commit dc759c51e667286ae9f9fcb0f6e4dc9ddbe34b1f
Author: Xiang Fu <[email protected]>
AuthorDate: Fri Oct 23 16:28:27 2026 -0700
Support FIRST_WITH_TIME and LAST_WITH_TIME without a data type (#19986)
Both query engines rewrite FIRST_WITH_TIME(dataColumn, timeColumn) and
LAST_WITH_TIME(dataColumn, timeColumn) to the explicit form with the data
column's type, and the explicit form now supports every single-value data type
(adding TIMESTAMP, BIG_DECIMAL, BYTES, UUID and JSON).
---
.../apache/pinot/core/common/ObjectSerDeUtils.java | 47 +++++-
.../function/AggregationFunctionUtils.java | 3 +
...igDecimalValueWithTimeAggregationFunction.java} | 50 ++++---
...irstBytesValueWithTimeAggregationFunction.java} | 67 +++++----
...rstLastWithTimeAggregationFunctionProvider.java | 25 ++++
.../FirstLongValueWithTimeAggregationFunction.java | 16 +-
.../function/FirstWithTimeAggregationFunction.java | 12 +-
...igDecimalValueWithTimeAggregationFunction.java} | 50 ++++---
...LastBytesValueWithTimeAggregationFunction.java} | 67 +++++----
.../LastLongValueWithTimeAggregationFunction.java | 16 +-
.../function/LastWithTimeAggregationFunction.java | 12 +-
.../AggregateFunctionRewriteOptimizer.java | 109 +++++++++-----
.../pinot/core/common/ObjectSerDeUtilsTest.java | 34 +++++
.../function/AggregationFunctionUtilsTest.java | 25 +++-
.../AggregateFunctionRewriteOptimizerTest.java | 147 ++++++++++++++++++
.../pinot/queries/FirstWithTimeQueriesTest.java | 148 +++++++++++++++++++
.../pinot/queries/LastWithTimeQueriesTest.java | 148 +++++++++++++++++++
.../PinotAggregateTypeArgumentRewriteRule.java | 164 +++++++++++++++++++++
.../org/apache/pinot/query/QueryEnvironment.java | 3 +
.../apache/pinot/query/QueryCompilationTest.java | 11 ++
.../pinot/query/QueryPlannerRuleOptionsTest.java | 10 ++
.../src/test/resources/queries/AggregatePlans.json | 39 +++++
.../query/runtime/queries/QueryRunnerTestBase.java | 9 ++
.../src/test/resources/queries/UDFAggregates.json | 76 +++++++++-
.../local/customobject/BigDecimalLongPair.java | 52 +++++++
.../segment/local/customobject/BytesLongPair.java | 51 +++++++
.../pinot/segment/spi/AggregationFunctionType.java | 10 +-
27 files changed, 1239 insertions(+), 162 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..4259bd3873b 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
@@ -85,6 +85,8 @@ import
org.apache.pinot.core.query.aggregation.utils.exprminmax.ExprMinMaxObject
import org.apache.pinot.core.query.utils.idset.IdSet;
import org.apache.pinot.core.query.utils.idset.IdSets;
import org.apache.pinot.segment.local.customobject.AvgPair;
+import org.apache.pinot.segment.local.customobject.BigDecimalLongPair;
+import org.apache.pinot.segment.local.customobject.BytesLongPair;
import org.apache.pinot.segment.local.customobject.CovarianceTuple;
import org.apache.pinot.segment.local.customobject.CpcSketchAccumulator;
import org.apache.pinot.segment.local.customobject.DoubleLongPair;
@@ -173,7 +175,9 @@ public class ObjectSerDeUtils {
FunnelStepEventAccumulator(51),
BigDecimalArrayList(52),
BigDecimalSet(53),
- BytesArrayList(54);
+ BytesArrayList(54),
+ BigDecimalLongPair(55),
+ BytesLongPair(56);
private final int _value;
@@ -287,6 +291,10 @@ public class ObjectSerDeUtils {
return ObjectType.DoubleLongPair;
} else if (value instanceof StringLongPair) {
return ObjectType.StringLongPair;
+ } else if (value instanceof BigDecimalLongPair) {
+ return ObjectType.BigDecimalLongPair;
+ } else if (value instanceof BytesLongPair) {
+ return ObjectType.BytesLongPair;
} else if (value instanceof CovarianceTuple) {
return ObjectType.CovarianceTuple;
} else if (value instanceof VarianceTuple) {
@@ -810,6 +818,41 @@ public class ObjectSerDeUtils {
return StringLongPair.fromByteBuffer(byteBuffer);
}
};
+ public static final ObjectSerDe<BigDecimalLongPair>
BIG_DECIMAL_LONG_PAIR_SER_DE =
+ new ObjectSerDe<BigDecimalLongPair>() {
+
+ @Override
+ public byte[] serialize(BigDecimalLongPair bigDecimalLongPair) {
+ return bigDecimalLongPair.toBytes();
+ }
+
+ @Override
+ public BigDecimalLongPair deserialize(byte[] bytes) {
+ return BigDecimalLongPair.fromBytes(bytes);
+ }
+
+ @Override
+ public BigDecimalLongPair deserialize(ByteBuffer byteBuffer) {
+ return BigDecimalLongPair.fromByteBuffer(byteBuffer);
+ }
+ };
+ public static final ObjectSerDe<BytesLongPair> BYTES_LONG_PAIR_SER_DE = new
ObjectSerDe<BytesLongPair>() {
+
+ @Override
+ public byte[] serialize(BytesLongPair bytesLongPair) {
+ return bytesLongPair.toBytes();
+ }
+
+ @Override
+ public BytesLongPair deserialize(byte[] bytes) {
+ return BytesLongPair.fromBytes(bytes);
+ }
+
+ @Override
+ public BytesLongPair deserialize(ByteBuffer byteBuffer) {
+ return BytesLongPair.fromByteBuffer(byteBuffer);
+ }
+ };
public static final ObjectSerDe<CovarianceTuple>
COVARIANCE_TUPLE_OBJECT_SER_DE = new ObjectSerDe<CovarianceTuple>() {
@Override
@@ -2089,6 +2132,8 @@ public class ObjectSerDeUtils {
BIG_DECIMAL_ARRAY_LIST_SER_DE,
BIG_DECIMAL_SET_SER_DE,
BYTES_ARRAY_LIST_SER_DE,
+ BIG_DECIMAL_LONG_PAIR_SER_DE,
+ BYTES_LONG_PAIR_SER_DE,
};
//@formatter:on
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
index 649ab7ad9c2..13c93f4023d 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtils.java
@@ -77,6 +77,7 @@ import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.apache.pinot.spi.exception.BadQueryRequestException;
import org.apache.pinot.spi.query.QueryThreadContext;
import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.UuidUtils;
/// The `AggregationFunctionUtils` class provides utility methods for
aggregation function.
@@ -355,6 +356,8 @@ public class AggregationFunctionUtils {
return dataTable.getString(rowId, colId);
case BYTES:
return dataTable.getBytes(rowId, colId).getBytes();
+ case UUID:
+ return UuidUtils.toUUID(dataTable.getBytes(rowId, colId));
case INT_ARRAY:
return dataTable.getIntArray(rowId, colId);
case LONG_ARRAY:
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstBigDecimalValueWithTimeAggregationFunction.java
similarity index 64%
copy from
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
copy to
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstBigDecimalValueWithTimeAggregationFunction.java
index 382abe1be19..155779960ad 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstBigDecimalValueWithTimeAggregationFunction.java
@@ -18,57 +18,59 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.math.BigDecimal;
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.common.BlockValSet;
import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
-import org.apache.pinot.segment.local.customobject.LongLongPair;
+import org.apache.pinot.segment.local.customobject.BigDecimalLongPair;
import org.apache.pinot.segment.local.customobject.ValueLongPair;
-/// This function is used for FirstWithTime calculations for data column with
long type.
+/// This function is used for FirstWithTime calculations for data column with
big decimal type.
///
-/// The function can be used as FirstWithTime(dataExpression, timeExpression,
'long')
+/// The function can be used as FirstWithTime(dataExpression, timeExpression,
'big_decimal')
///
/// Following arguments are supported:
///
-/// - dataExpression: expression that contains the long data column to be
calculated first on
+/// - dataExpression: expression that contains the big decimal data column to
be calculated first on
/// - timeExpression: expression that contains the column to be used to decide
which data is first, can be any
/// Numeric column
-public class FirstLongValueWithTimeAggregationFunction extends
FirstWithTimeAggregationFunction<Long> {
- private final static ValueLongPair<Long> DEFAULT_VALUE_TIME_PAIR = new
LongLongPair(Long.MIN_VALUE, Long.MAX_VALUE);
+public class FirstBigDecimalValueWithTimeAggregationFunction extends
FirstWithTimeAggregationFunction<BigDecimal> {
+ private final static ValueLongPair<BigDecimal> DEFAULT_VALUE_TIME_PAIR =
+ new BigDecimalLongPair(BigDecimal.ZERO, Long.MAX_VALUE);
- public FirstLongValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
+ public FirstBigDecimalValueWithTimeAggregationFunction(ExpressionContext
dataCol, ExpressionContext timeCol,
boolean nullHandlingEnabled) {
- super(dataCol, timeCol, ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE,
nullHandlingEnabled);
+ super(dataCol, timeCol, ObjectSerDeUtils.BIG_DECIMAL_LONG_PAIR_SER_DE,
nullHandlingEnabled);
}
@Override
- public ValueLongPair<Long> constructValueLongPair(Long value, long time) {
- return new LongLongPair(value, time);
+ public ValueLongPair<BigDecimal> constructValueLongPair(BigDecimal value,
long time) {
+ return new BigDecimalLongPair(value, time);
}
@Override
- public ValueLongPair<Long> getDefaultValueTimePair() {
+ public ValueLongPair<BigDecimal> getDefaultValueTimePair() {
return DEFAULT_VALUE_TIME_PAIR;
}
@Override
- public Long readCell(BlockValSet block, int docId) {
- return block.getLongValuesSV()[docId];
+ public BigDecimal readCell(BlockValSet block, int docId) {
+ return block.getBigDecimalValuesSV()[docId];
}
@Override
public void aggregateGroupResultWithRawDataSv(int length, int[]
groupKeyArray,
GroupByResultHolder groupByResultHolder, BlockValSet blockValSet,
BlockValSet timeValSet) {
- long[] longValues = blockValSet.getLongValuesSV();
+ BigDecimal[] bigDecimalValues = blockValSet.getBigDecimalValuesSV();
long[] timeValues = timeValSet.getLongValuesSV();
forEachNotNull(length, blockValSet, timeValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- long data = longValues[i];
+ BigDecimal data = bigDecimalValues[i];
long time = timeValues[i];
setGroupByResult(groupKeyArray[i], groupByResultHolder, data, time);
}
@@ -78,12 +80,12 @@ public class FirstLongValueWithTimeAggregationFunction
extends FirstWithTimeAggr
@Override
public void aggregateGroupResultWithRawDataMv(int length, int[][]
groupKeysArray,
GroupByResultHolder groupByResultHolder, BlockValSet blockValSet,
BlockValSet timeValSet) {
- long[] longValues = blockValSet.getLongValuesSV();
+ BigDecimal[] bigDecimalValues = blockValSet.getBigDecimalValuesSV();
long[] timeValues = timeValSet.getLongValuesSV();
forEachNotNull(length, blockValSet, timeValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- long value = longValues[i];
+ BigDecimal value = bigDecimalValues[i];
long time = timeValues[i];
for (int groupKey : groupKeysArray[i]) {
setGroupByResult(groupKey, groupByResultHolder, value, time);
@@ -94,22 +96,22 @@ public class FirstLongValueWithTimeAggregationFunction
extends FirstWithTimeAggr
@Override
public String getResultColumnName() {
- return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'LONG')";
+ return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'BIG_DECIMAL')";
}
@Override
- public SerializedIntermediateResult
serializeIntermediateResult(ValueLongPair<Long> longLongPair) {
- return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.LongLongPair.getValue(),
- ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE.serialize((LongLongPair)
longLongPair));
+ public SerializedIntermediateResult
serializeIntermediateResult(ValueLongPair<BigDecimal> bigDecimalLongPair) {
+ return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.BigDecimalLongPair.getValue(),
+
ObjectSerDeUtils.BIG_DECIMAL_LONG_PAIR_SER_DE.serialize((BigDecimalLongPair)
bigDecimalLongPair));
}
@Override
- public ValueLongPair<Long> deserializeIntermediateResult(CustomObject
customObject) {
- return
ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE.deserialize(customObject.getBuffer());
+ public ValueLongPair<BigDecimal> deserializeIntermediateResult(CustomObject
customObject) {
+ return
ObjectSerDeUtils.BIG_DECIMAL_LONG_PAIR_SER_DE.deserialize(customObject.getBuffer());
}
@Override
public ColumnDataType getFinalResultColumnType() {
- return ColumnDataType.LONG;
+ return ColumnDataType.BIG_DECIMAL;
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstBytesValueWithTimeAggregationFunction.java
similarity index 55%
copy from
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
copy to
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstBytesValueWithTimeAggregationFunction.java
index 382abe1be19..d3aec379777 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstBytesValueWithTimeAggregationFunction.java
@@ -18,57 +18,72 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import com.google.common.base.Preconditions;
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.common.BlockValSet;
import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
-import org.apache.pinot.segment.local.customobject.LongLongPair;
+import org.apache.pinot.segment.local.customobject.BytesLongPair;
import org.apache.pinot.segment.local.customobject.ValueLongPair;
+import org.apache.pinot.spi.utils.ByteArray;
-/// This function is used for FirstWithTime calculations for data column with
long type.
+/// This function is used for FirstWithTime calculations for data column with
bytes or UUID type.
///
-/// The function can be used as FirstWithTime(dataExpression, timeExpression,
'long')
+/// The function can be used as FirstWithTime(dataExpression, timeExpression,
'bytes')
///
/// Following arguments are supported:
///
-/// - dataExpression: expression that contains the long data column to be
calculated first on
+/// - dataExpression: expression that contains the bytes or UUID data column
to be calculated first on
/// - timeExpression: expression that contains the column to be used to decide
which data is first, can be any
/// Numeric column
-public class FirstLongValueWithTimeAggregationFunction extends
FirstWithTimeAggregationFunction<Long> {
- private final static ValueLongPair<Long> DEFAULT_VALUE_TIME_PAIR = new
LongLongPair(Long.MIN_VALUE, Long.MAX_VALUE);
+public class FirstBytesValueWithTimeAggregationFunction extends
FirstWithTimeAggregationFunction<ByteArray> {
+ private final ColumnDataType _resultType;
+ private final ValueLongPair<ByteArray> _defaultValueTimePair;
- public FirstLongValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
- boolean nullHandlingEnabled) {
- super(dataCol, timeCol, ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE,
nullHandlingEnabled);
+ /// The result type is BYTES or UUID; both are stored as bytes.
+ public FirstBytesValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
+ boolean nullHandlingEnabled, ColumnDataType resultType) {
+ super(dataCol, timeCol, ObjectSerDeUtils.BYTES_LONG_PAIR_SER_DE,
nullHandlingEnabled);
+ Preconditions.checkArgument(resultType == ColumnDataType.BYTES ||
resultType == ColumnDataType.UUID,
+ "Unsupported result type: %s", resultType);
+ _resultType = resultType;
+ // An empty value is not a valid UUID, so the default value is the result
type's null placeholder (the nil UUID).
+ _defaultValueTimePair = new BytesLongPair((ByteArray)
resultType.getNullPlaceholder(), Long.MAX_VALUE);
}
@Override
- public ValueLongPair<Long> constructValueLongPair(Long value, long time) {
- return new LongLongPair(value, time);
+ public ValueLongPair<ByteArray> constructValueLongPair(ByteArray value, long
time) {
+ return new BytesLongPair(value, time);
}
@Override
- public ValueLongPair<Long> getDefaultValueTimePair() {
- return DEFAULT_VALUE_TIME_PAIR;
+ public ValueLongPair<ByteArray> getDefaultValueTimePair() {
+ return _defaultValueTimePair;
}
+ /// A BYTES data column holds values, not serialized pairs.
@Override
- public Long readCell(BlockValSet block, int docId) {
- return block.getLongValuesSV()[docId];
+ protected boolean isSerializedPairInput(BlockValSet blockValSet) {
+ return false;
+ }
+
+ @Override
+ public ByteArray readCell(BlockValSet block, int docId) {
+ return new ByteArray(block.getBytesValuesSV()[docId]);
}
@Override
public void aggregateGroupResultWithRawDataSv(int length, int[]
groupKeyArray,
GroupByResultHolder groupByResultHolder, BlockValSet blockValSet,
BlockValSet timeValSet) {
- long[] longValues = blockValSet.getLongValuesSV();
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
long[] timeValues = timeValSet.getLongValuesSV();
forEachNotNull(length, blockValSet, timeValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- long data = longValues[i];
+ ByteArray data = new ByteArray(bytesValues[i]);
long time = timeValues[i];
setGroupByResult(groupKeyArray[i], groupByResultHolder, data, time);
}
@@ -78,12 +93,12 @@ public class FirstLongValueWithTimeAggregationFunction
extends FirstWithTimeAggr
@Override
public void aggregateGroupResultWithRawDataMv(int length, int[][]
groupKeysArray,
GroupByResultHolder groupByResultHolder, BlockValSet blockValSet,
BlockValSet timeValSet) {
- long[] longValues = blockValSet.getLongValuesSV();
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
long[] timeValues = timeValSet.getLongValuesSV();
forEachNotNull(length, blockValSet, timeValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- long value = longValues[i];
+ ByteArray value = new ByteArray(bytesValues[i]);
long time = timeValues[i];
for (int groupKey : groupKeysArray[i]) {
setGroupByResult(groupKey, groupByResultHolder, value, time);
@@ -94,22 +109,22 @@ public class FirstLongValueWithTimeAggregationFunction
extends FirstWithTimeAggr
@Override
public String getResultColumnName() {
- return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'LONG')";
+ return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'" + _resultType + "')";
}
@Override
- public SerializedIntermediateResult
serializeIntermediateResult(ValueLongPair<Long> longLongPair) {
- return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.LongLongPair.getValue(),
- ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE.serialize((LongLongPair)
longLongPair));
+ public SerializedIntermediateResult
serializeIntermediateResult(ValueLongPair<ByteArray> bytesLongPair) {
+ return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.BytesLongPair.getValue(),
+ ObjectSerDeUtils.BYTES_LONG_PAIR_SER_DE.serialize((BytesLongPair)
bytesLongPair));
}
@Override
- public ValueLongPair<Long> deserializeIntermediateResult(CustomObject
customObject) {
- return
ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE.deserialize(customObject.getBuffer());
+ public ValueLongPair<ByteArray> deserializeIntermediateResult(CustomObject
customObject) {
+ return
ObjectSerDeUtils.BYTES_LONG_PAIR_SER_DE.deserialize(customObject.getBuffer());
}
@Override
public ColumnDataType getFinalResultColumnType() {
- return ColumnDataType.LONG;
+ return _resultType;
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLastWithTimeAggregationFunctionProvider.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLastWithTimeAggregationFunctionProvider.java
index 35e7ca647e5..c6ecb8c91d6 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLastWithTimeAggregationFunctionProvider.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLastWithTimeAggregationFunctionProvider.java
@@ -22,6 +22,7 @@ import com.google.common.base.Preconditions;
import java.util.List;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.request.context.FunctionContext;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.spi.data.FieldSpec.DataType;
@@ -60,12 +61,24 @@ public final class
FirstLastWithTimeAggregationFunctionProvider {
return new FirstIntValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled, false);
case LONG:
return new FirstLongValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
+ case TIMESTAMP:
+ return new FirstLongValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled,
+ ColumnDataType.TIMESTAMP);
case FLOAT:
return new FirstFloatValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
case DOUBLE:
return new FirstDoubleValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
+ case BIG_DECIMAL:
+ return new FirstBigDecimalValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
case STRING:
+ case JSON:
return new FirstStringValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
+ case BYTES:
+ return new FirstBytesValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled,
+ ColumnDataType.BYTES);
+ case UUID:
+ return new FirstBytesValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled,
+ ColumnDataType.UUID);
default:
throw new IllegalArgumentException("Unsupported data type for
FIRST_WITH_TIME: " + dataType);
}
@@ -100,12 +113,24 @@ public final class
FirstLastWithTimeAggregationFunctionProvider {
return new LastIntValueWithTimeAggregationFunction(dataCol, timeCol,
nullHandlingEnabled, false);
case LONG:
return new LastLongValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
+ case TIMESTAMP:
+ return new LastLongValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled,
+ ColumnDataType.TIMESTAMP);
case FLOAT:
return new LastFloatValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
case DOUBLE:
return new LastDoubleValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
+ case BIG_DECIMAL:
+ return new LastBigDecimalValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
case STRING:
+ case JSON:
return new LastStringValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled);
+ case BYTES:
+ return new LastBytesValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled,
+ ColumnDataType.BYTES);
+ case UUID:
+ return new LastBytesValueWithTimeAggregationFunction(dataCol,
timeCol, nullHandlingEnabled,
+ ColumnDataType.UUID);
default:
throw new IllegalArgumentException("Unsupported data type for
LAST_WITH_TIME: " + dataType);
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
index 382abe1be19..e56f309a0d6 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstLongValueWithTimeAggregationFunction.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import com.google.common.base.Preconditions;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
@@ -40,9 +41,20 @@ import
org.apache.pinot.segment.local.customobject.ValueLongPair;
public class FirstLongValueWithTimeAggregationFunction extends
FirstWithTimeAggregationFunction<Long> {
private final static ValueLongPair<Long> DEFAULT_VALUE_TIME_PAIR = new
LongLongPair(Long.MIN_VALUE, Long.MAX_VALUE);
+ private final ColumnDataType _resultType;
+
public FirstLongValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
boolean nullHandlingEnabled) {
+ this(dataCol, timeCol, nullHandlingEnabled, ColumnDataType.LONG);
+ }
+
+ /// The result type is LONG or TIMESTAMP; both are stored as long values.
+ public FirstLongValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
+ boolean nullHandlingEnabled, ColumnDataType resultType) {
super(dataCol, timeCol, ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE,
nullHandlingEnabled);
+ Preconditions.checkArgument(resultType == ColumnDataType.LONG ||
resultType == ColumnDataType.TIMESTAMP,
+ "Unsupported result type: %s", resultType);
+ _resultType = resultType;
}
@Override
@@ -94,7 +106,7 @@ public class FirstLongValueWithTimeAggregationFunction
extends FirstWithTimeAggr
@Override
public String getResultColumnName() {
- return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'LONG')";
+ return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'" + _resultType + "')";
}
@Override
@@ -110,6 +122,6 @@ public class FirstLongValueWithTimeAggregationFunction
extends FirstWithTimeAggr
@Override
public ColumnDataType getFinalResultColumnType() {
- return ColumnDataType.LONG;
+ return _resultType;
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstWithTimeAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstWithTimeAggregationFunction.java
index f41e4f63423..f10a28cf746 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstWithTimeAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/FirstWithTimeAggregationFunction.java
@@ -64,6 +64,12 @@ public abstract class FirstWithTimeAggregationFunction<V
extends Comparable<V>>
public abstract ValueLongPair<V> getDefaultValueTimePair();
+ /// Whether the data input holds serialized value-time pairs, as a
pre-aggregated BYTES column does, rather than data
+ /// values.
+ protected boolean isSerializedPairInput(BlockValSet blockValSet) {
+ return blockValSet.getValueType() == DataType.BYTES;
+ }
+
public abstract V readCell(BlockValSet block, int docId);
public abstract void aggregateGroupResultWithRawDataSv(int length,
@@ -99,7 +105,7 @@ public abstract class FirstWithTimeAggregationFunction<V
extends Comparable<V>>
BlockValSet blockValSet = blockValSetMap.get(_expression);
BlockValSet blockTimeSet = blockValSetMap.get(_timeCol);
- if (blockValSet.getValueType() != DataType.BYTES) {
+ if (!isSerializedPairInput(blockValSet)) {
IntLongPair defaultPair = new IntLongPair(Integer.MIN_VALUE,
Long.MAX_VALUE);
long[] timeValues = blockTimeSet.getLongValuesSV();
@@ -154,7 +160,7 @@ public abstract class FirstWithTimeAggregationFunction<V
extends Comparable<V>>
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
BlockValSet timeValSet = blockValSetMap.get(_timeCol);
- if (blockValSet.getValueType() != DataType.BYTES) {
+ if (!isSerializedPairInput(blockValSet)) {
aggregateGroupResultWithRawDataSv(length, groupKeyArray,
groupByResultHolder,
blockValSet, timeValSet);
} else {
@@ -177,7 +183,7 @@ public abstract class FirstWithTimeAggregationFunction<V
extends Comparable<V>>
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
BlockValSet timeValSet = blockValSetMap.get(_timeCol);
- if (blockValSet.getValueType() != DataType.BYTES) {
+ if (!isSerializedPairInput(blockValSet)) {
aggregateGroupResultWithRawDataMv(length, groupKeysArray,
groupByResultHolder, blockValSet, timeValSet);
} else {
// Serialized ValueTimePair
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastBigDecimalValueWithTimeAggregationFunction.java
similarity index 64%
copy from
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
copy to
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastBigDecimalValueWithTimeAggregationFunction.java
index 0e543204709..3c4be3a02fe 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastBigDecimalValueWithTimeAggregationFunction.java
@@ -18,57 +18,59 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.math.BigDecimal;
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.common.BlockValSet;
import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
-import org.apache.pinot.segment.local.customobject.LongLongPair;
+import org.apache.pinot.segment.local.customobject.BigDecimalLongPair;
import org.apache.pinot.segment.local.customobject.ValueLongPair;
-/// This function is used for LastWithTime calculations for data column with
long type.
+/// This function is used for LastWithTime calculations for data column with
big decimal type.
///
-/// The function can be used as LastWithTime(dataExpression, timeExpression,
'long')
+/// The function can be used as LastWithTime(dataExpression, timeExpression,
'big_decimal')
///
/// Following arguments are supported:
///
-/// - dataExpression: expression that contains the long data column to be
calculated last on
+/// - dataExpression: expression that contains the big decimal data column to
be calculated last on
/// - timeExpression: expression that contains the column to be used to decide
which data is last, can be any
/// Numeric column
-public class LastLongValueWithTimeAggregationFunction extends
LastWithTimeAggregationFunction<Long> {
- private final static ValueLongPair<Long> DEFAULT_VALUE_TIME_PAIR = new
LongLongPair(Long.MIN_VALUE, Long.MIN_VALUE);
+public class LastBigDecimalValueWithTimeAggregationFunction extends
LastWithTimeAggregationFunction<BigDecimal> {
+ private final static ValueLongPair<BigDecimal> DEFAULT_VALUE_TIME_PAIR =
+ new BigDecimalLongPair(BigDecimal.ZERO, Long.MIN_VALUE);
- public LastLongValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
+ public LastBigDecimalValueWithTimeAggregationFunction(ExpressionContext
dataCol, ExpressionContext timeCol,
boolean nullHandlingEnabled) {
- super(dataCol, timeCol, ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE,
nullHandlingEnabled);
+ super(dataCol, timeCol, ObjectSerDeUtils.BIG_DECIMAL_LONG_PAIR_SER_DE,
nullHandlingEnabled);
}
@Override
- public ValueLongPair<Long> constructValueLongPair(Long value, long time) {
- return new LongLongPair(value, time);
+ public ValueLongPair<BigDecimal> constructValueLongPair(BigDecimal value,
long time) {
+ return new BigDecimalLongPair(value, time);
}
@Override
- public ValueLongPair<Long> getDefaultValueTimePair() {
+ public ValueLongPair<BigDecimal> getDefaultValueTimePair() {
return DEFAULT_VALUE_TIME_PAIR;
}
@Override
- public Long readCell(BlockValSet block, int docId) {
- return block.getLongValuesSV()[docId];
+ public BigDecimal readCell(BlockValSet block, int docId) {
+ return block.getBigDecimalValuesSV()[docId];
}
@Override
public void aggregateGroupResultWithRawDataSv(int length, int[]
groupKeyArray,
GroupByResultHolder groupByResultHolder, BlockValSet blockValSet,
BlockValSet timeValSet) {
- long[] longValues = blockValSet.getLongValuesSV();
+ BigDecimal[] bigDecimalValues = blockValSet.getBigDecimalValuesSV();
long[] timeValues = timeValSet.getLongValuesSV();
forEachNotNull(length, blockValSet, timeValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- long data = longValues[i];
+ BigDecimal data = bigDecimalValues[i];
long time = timeValues[i];
setGroupByResult(groupKeyArray[i], groupByResultHolder, data, time);
}
@@ -78,12 +80,12 @@ public class LastLongValueWithTimeAggregationFunction
extends LastWithTimeAggreg
@Override
public void aggregateGroupResultWithRawDataMv(int length, int[][]
groupKeysArray,
GroupByResultHolder groupByResultHolder, BlockValSet blockValSet,
BlockValSet timeValSet) {
- long[] longValues = blockValSet.getLongValuesSV();
+ BigDecimal[] bigDecimalValues = blockValSet.getBigDecimalValuesSV();
long[] timeValues = timeValSet.getLongValuesSV();
forEachNotNull(length, blockValSet, timeValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- long value = longValues[i];
+ BigDecimal value = bigDecimalValues[i];
long time = timeValues[i];
for (int groupKey : groupKeysArray[i]) {
setGroupByResult(groupKey, groupByResultHolder, value, time);
@@ -94,22 +96,22 @@ public class LastLongValueWithTimeAggregationFunction
extends LastWithTimeAggreg
@Override
public String getResultColumnName() {
- return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'LONG')";
+ return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'BIG_DECIMAL')";
}
@Override
- public SerializedIntermediateResult
serializeIntermediateResult(ValueLongPair<Long> longLongPair) {
- return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.LongLongPair.getValue(),
- ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE.serialize((LongLongPair)
longLongPair));
+ public SerializedIntermediateResult
serializeIntermediateResult(ValueLongPair<BigDecimal> bigDecimalLongPair) {
+ return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.BigDecimalLongPair.getValue(),
+
ObjectSerDeUtils.BIG_DECIMAL_LONG_PAIR_SER_DE.serialize((BigDecimalLongPair)
bigDecimalLongPair));
}
@Override
- public ValueLongPair<Long> deserializeIntermediateResult(CustomObject
customObject) {
- return
ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE.deserialize(customObject.getBuffer());
+ public ValueLongPair<BigDecimal> deserializeIntermediateResult(CustomObject
customObject) {
+ return
ObjectSerDeUtils.BIG_DECIMAL_LONG_PAIR_SER_DE.deserialize(customObject.getBuffer());
}
@Override
public ColumnDataType getFinalResultColumnType() {
- return ColumnDataType.LONG;
+ return ColumnDataType.BIG_DECIMAL;
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastBytesValueWithTimeAggregationFunction.java
similarity index 55%
copy from
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
copy to
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastBytesValueWithTimeAggregationFunction.java
index 0e543204709..338092948e2 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastBytesValueWithTimeAggregationFunction.java
@@ -18,57 +18,72 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import com.google.common.base.Preconditions;
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.common.BlockValSet;
import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
-import org.apache.pinot.segment.local.customobject.LongLongPair;
+import org.apache.pinot.segment.local.customobject.BytesLongPair;
import org.apache.pinot.segment.local.customobject.ValueLongPair;
+import org.apache.pinot.spi.utils.ByteArray;
-/// This function is used for LastWithTime calculations for data column with
long type.
+/// This function is used for LastWithTime calculations for data column with
bytes or UUID type.
///
-/// The function can be used as LastWithTime(dataExpression, timeExpression,
'long')
+/// The function can be used as LastWithTime(dataExpression, timeExpression,
'bytes')
///
/// Following arguments are supported:
///
-/// - dataExpression: expression that contains the long data column to be
calculated last on
+/// - dataExpression: expression that contains the bytes or UUID data column
to be calculated last on
/// - timeExpression: expression that contains the column to be used to decide
which data is last, can be any
/// Numeric column
-public class LastLongValueWithTimeAggregationFunction extends
LastWithTimeAggregationFunction<Long> {
- private final static ValueLongPair<Long> DEFAULT_VALUE_TIME_PAIR = new
LongLongPair(Long.MIN_VALUE, Long.MIN_VALUE);
+public class LastBytesValueWithTimeAggregationFunction extends
LastWithTimeAggregationFunction<ByteArray> {
+ private final ColumnDataType _resultType;
+ private final ValueLongPair<ByteArray> _defaultValueTimePair;
- public LastLongValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
- boolean nullHandlingEnabled) {
- super(dataCol, timeCol, ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE,
nullHandlingEnabled);
+ /// The result type is BYTES or UUID; both are stored as bytes.
+ public LastBytesValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
+ boolean nullHandlingEnabled, ColumnDataType resultType) {
+ super(dataCol, timeCol, ObjectSerDeUtils.BYTES_LONG_PAIR_SER_DE,
nullHandlingEnabled);
+ Preconditions.checkArgument(resultType == ColumnDataType.BYTES ||
resultType == ColumnDataType.UUID,
+ "Unsupported result type: %s", resultType);
+ _resultType = resultType;
+ // An empty value is not a valid UUID, so the default value is the result
type's null placeholder (the nil UUID).
+ _defaultValueTimePair = new BytesLongPair((ByteArray)
resultType.getNullPlaceholder(), Long.MIN_VALUE);
}
@Override
- public ValueLongPair<Long> constructValueLongPair(Long value, long time) {
- return new LongLongPair(value, time);
+ public ValueLongPair<ByteArray> constructValueLongPair(ByteArray value, long
time) {
+ return new BytesLongPair(value, time);
}
@Override
- public ValueLongPair<Long> getDefaultValueTimePair() {
- return DEFAULT_VALUE_TIME_PAIR;
+ public ValueLongPair<ByteArray> getDefaultValueTimePair() {
+ return _defaultValueTimePair;
}
+ /// A BYTES data column holds values, not serialized pairs.
@Override
- public Long readCell(BlockValSet block, int docId) {
- return block.getLongValuesSV()[docId];
+ protected boolean isSerializedPairInput(BlockValSet blockValSet) {
+ return false;
+ }
+
+ @Override
+ public ByteArray readCell(BlockValSet block, int docId) {
+ return new ByteArray(block.getBytesValuesSV()[docId]);
}
@Override
public void aggregateGroupResultWithRawDataSv(int length, int[]
groupKeyArray,
GroupByResultHolder groupByResultHolder, BlockValSet blockValSet,
BlockValSet timeValSet) {
- long[] longValues = blockValSet.getLongValuesSV();
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
long[] timeValues = timeValSet.getLongValuesSV();
forEachNotNull(length, blockValSet, timeValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- long data = longValues[i];
+ ByteArray data = new ByteArray(bytesValues[i]);
long time = timeValues[i];
setGroupByResult(groupKeyArray[i], groupByResultHolder, data, time);
}
@@ -78,12 +93,12 @@ public class LastLongValueWithTimeAggregationFunction
extends LastWithTimeAggreg
@Override
public void aggregateGroupResultWithRawDataMv(int length, int[][]
groupKeysArray,
GroupByResultHolder groupByResultHolder, BlockValSet blockValSet,
BlockValSet timeValSet) {
- long[] longValues = blockValSet.getLongValuesSV();
+ byte[][] bytesValues = blockValSet.getBytesValuesSV();
long[] timeValues = timeValSet.getLongValuesSV();
forEachNotNull(length, blockValSet, timeValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- long value = longValues[i];
+ ByteArray value = new ByteArray(bytesValues[i]);
long time = timeValues[i];
for (int groupKey : groupKeysArray[i]) {
setGroupByResult(groupKey, groupByResultHolder, value, time);
@@ -94,22 +109,22 @@ public class LastLongValueWithTimeAggregationFunction
extends LastWithTimeAggreg
@Override
public String getResultColumnName() {
- return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'LONG')";
+ return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'" + _resultType + "')";
}
@Override
- public SerializedIntermediateResult
serializeIntermediateResult(ValueLongPair<Long> longLongPair) {
- return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.LongLongPair.getValue(),
- ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE.serialize((LongLongPair)
longLongPair));
+ public SerializedIntermediateResult
serializeIntermediateResult(ValueLongPair<ByteArray> bytesLongPair) {
+ return new
SerializedIntermediateResult(ObjectSerDeUtils.ObjectType.BytesLongPair.getValue(),
+ ObjectSerDeUtils.BYTES_LONG_PAIR_SER_DE.serialize((BytesLongPair)
bytesLongPair));
}
@Override
- public ValueLongPair<Long> deserializeIntermediateResult(CustomObject
customObject) {
- return
ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE.deserialize(customObject.getBuffer());
+ public ValueLongPair<ByteArray> deserializeIntermediateResult(CustomObject
customObject) {
+ return
ObjectSerDeUtils.BYTES_LONG_PAIR_SER_DE.deserialize(customObject.getBuffer());
}
@Override
public ColumnDataType getFinalResultColumnType() {
- return ColumnDataType.LONG;
+ return _resultType;
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
index 0e543204709..3d5250282d4 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastLongValueWithTimeAggregationFunction.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import com.google.common.base.Preconditions;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
@@ -40,9 +41,20 @@ import
org.apache.pinot.segment.local.customobject.ValueLongPair;
public class LastLongValueWithTimeAggregationFunction extends
LastWithTimeAggregationFunction<Long> {
private final static ValueLongPair<Long> DEFAULT_VALUE_TIME_PAIR = new
LongLongPair(Long.MIN_VALUE, Long.MIN_VALUE);
+ private final ColumnDataType _resultType;
+
public LastLongValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
boolean nullHandlingEnabled) {
+ this(dataCol, timeCol, nullHandlingEnabled, ColumnDataType.LONG);
+ }
+
+ /// The result type is LONG or TIMESTAMP; both are stored as long values.
+ public LastLongValueWithTimeAggregationFunction(ExpressionContext dataCol,
ExpressionContext timeCol,
+ boolean nullHandlingEnabled, ColumnDataType resultType) {
super(dataCol, timeCol, ObjectSerDeUtils.LONG_LONG_PAIR_SER_DE,
nullHandlingEnabled);
+ Preconditions.checkArgument(resultType == ColumnDataType.LONG ||
resultType == ColumnDataType.TIMESTAMP,
+ "Unsupported result type: %s", resultType);
+ _resultType = resultType;
}
@Override
@@ -94,7 +106,7 @@ public class LastLongValueWithTimeAggregationFunction
extends LastWithTimeAggreg
@Override
public String getResultColumnName() {
- return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'LONG')";
+ return getType().getName().toLowerCase() + "(" + _expression + "," +
_timeCol + ",'" + _resultType + "')";
}
@Override
@@ -110,6 +122,6 @@ public class LastLongValueWithTimeAggregationFunction
extends LastWithTimeAggreg
@Override
public ColumnDataType getFinalResultColumnType() {
- return ColumnDataType.LONG;
+ return _resultType;
}
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastWithTimeAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastWithTimeAggregationFunction.java
index 4896a43fd71..a1bd5d2e529 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastWithTimeAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/LastWithTimeAggregationFunction.java
@@ -65,6 +65,12 @@ public abstract class LastWithTimeAggregationFunction<V
extends Comparable<V>>
public abstract ValueLongPair<V> getDefaultValueTimePair();
+ /// Whether the data input holds serialized value-time pairs, as a
pre-aggregated BYTES column does, rather than data
+ /// values.
+ protected boolean isSerializedPairInput(BlockValSet blockValSet) {
+ return blockValSet.getValueType() == DataType.BYTES;
+ }
+
public abstract V readCell(BlockValSet block, int docId);
public abstract void aggregateGroupResultWithRawDataSv(int length,
@@ -100,7 +106,7 @@ public abstract class LastWithTimeAggregationFunction<V
extends Comparable<V>>
BlockValSet blockValSet = blockValSetMap.get(_expression);
BlockValSet blockTimeSet = blockValSetMap.get(_timeCol);
- if (blockValSet.getValueType() != DataType.BYTES) {
+ if (!isSerializedPairInput(blockValSet)) {
IntLongPair defaultPair = new IntLongPair(Integer.MIN_VALUE,
Long.MIN_VALUE);
long[] timeValues = blockTimeSet.getLongValuesSV();
@@ -154,7 +160,7 @@ public abstract class LastWithTimeAggregationFunction<V
extends Comparable<V>>
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
BlockValSet timeValSet = blockValSetMap.get(_timeCol);
- if (blockValSet.getValueType() != DataType.BYTES) {
+ if (!isSerializedPairInput(blockValSet)) {
aggregateGroupResultWithRawDataSv(length, groupKeyArray,
groupByResultHolder,
blockValSet, timeValSet);
} else {
@@ -175,7 +181,7 @@ public abstract class LastWithTimeAggregationFunction<V
extends Comparable<V>>
Map<ExpressionContext, BlockValSet> blockValSetMap) {
BlockValSet blockValSet = blockValSetMap.get(_expression);
BlockValSet timeValSet = blockValSetMap.get(_timeCol);
- if (blockValSet.getValueType() != DataType.BYTES) {
+ if (!isSerializedPairInput(blockValSet)) {
aggregateGroupResultWithRawDataMv(length, groupKeysArray,
groupByResultHolder, blockValSet, timeValSet);
} else {
// Serialized ValueTimePair
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 f88c1cf372d..e7ed844ff79 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
@@ -23,21 +23,29 @@ import javax.annotation.Nullable;
import org.apache.pinot.common.request.Expression;
import org.apache.pinot.common.request.Function;
import org.apache.pinot.common.request.PinotQuery;
+import org.apache.pinot.common.utils.request.RequestUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.spi.data.FieldSpec;
import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.exception.BadQueryRequestException;
import org.apache.pinot.spi.utils.CommonConstants;
/// Rewrites certain aggregation functions based on operand types to support
polymorphic aggregations.
///
-/// Currently supported rewrites:
+/// Always applied:
+/// - FIRST_WITH_TIME(dataColumn, timeColumn) -> FIRST_WITH_TIME(dataColumn,
timeColumn, 'dataType')
+/// - LAST_WITH_TIME(dataColumn, timeColumn) -> LAST_WITH_TIME(dataColumn,
timeColumn, 'dataType')
+///
+/// Applied only when enabled through the `autoRewriteAggregationType` query
option:
/// - MIN(stringType) -> MINSTRING
/// - MAX(stringType) -> MAXSTRING
/// - MIN(longType) -> MINLONG
/// - MAX(longType) -> MAXLONG
/// - SUM(longType) -> SUMLONG
/// - SUM(intType) -> SUMINT
+///
+/// Aggregations nested in other expressions, such as an alias, a FILTER
clause or a transform, are rewritten as well.
public class AggregateFunctionRewriteOptimizer implements StatementOptimizer {
@Override
@@ -46,38 +54,24 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
return;
}
- // Only perform auto rewrite when enabled through query option.
- if (pinotQuery.getQueryOptions() == null ||
!Boolean.parseBoolean(pinotQuery.getQueryOptions().get(
-
CommonConstants.Broker.Request.QueryOptionKey.AUTO_REWRITE_AGGREGATION_TYPE))) {
- return;
- }
-
- List<Expression> selectList = pinotQuery.getSelectList();
- if (selectList != null) {
- for (Expression expression : selectList) {
- maybeRewriteAggregateFunction(expression, schema);
- }
- }
-
- List<Expression> groupByList = pinotQuery.getGroupByList();
- if (groupByList != null) {
- for (Expression expression : groupByList) {
- maybeRewriteAggregateFunction(expression, schema);
- }
- }
+ boolean rewriteToTypedVariants = pinotQuery.getQueryOptions() != null &&
Boolean.parseBoolean(
+
pinotQuery.getQueryOptions().get(CommonConstants.Broker.Request.QueryOptionKey.AUTO_REWRITE_AGGREGATION_TYPE));
+ rewrite(pinotQuery.getSelectList(), schema, rewriteToTypedVariants);
+ rewrite(pinotQuery.getGroupByList(), schema, rewriteToTypedVariants);
+ rewrite(pinotQuery.getOrderByList(), schema, rewriteToTypedVariants);
+ rewrite(pinotQuery.getFilterExpression(), schema, rewriteToTypedVariants);
+ rewrite(pinotQuery.getHavingExpression(), schema, rewriteToTypedVariants);
+ }
- List<Expression> orderByList = pinotQuery.getOrderByList();
- if (orderByList != null) {
- for (Expression expression : orderByList) {
- maybeRewriteAggregateFunction(expression, schema);
+ private static void rewrite(@Nullable List<Expression> expressions, Schema
schema, boolean rewriteToTypedVariants) {
+ if (expressions != null) {
+ for (Expression expression : expressions) {
+ rewrite(expression, schema, rewriteToTypedVariants);
}
}
-
- maybeRewriteAggregateFunction(pinotQuery.getFilterExpression(), schema);
- maybeRewriteAggregateFunction(pinotQuery.getHavingExpression(), schema);
}
- private void maybeRewriteAggregateFunction(@Nullable Expression expression,
Schema schema) {
+ private static void rewrite(@Nullable Expression expression, Schema schema,
boolean rewriteToTypedVariants) {
if (expression == null || !expression.isSetFunctionCall()) {
return;
}
@@ -85,23 +79,62 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
Function function = expression.getFunctionCall();
String functionName = function.getOperator();
if (!AggregationFunctionType.isAggregationFunction(functionName)) {
+ rewrite(function.getOperands(), schema, rewriteToTypedVariants);
return;
}
- Expression operand = function.getOperands().get(0);
- FieldSpec.DataType operandType;
// TODO: Handle more complex expressions (e.g. MIN(trim(stringCol)) )
- if (operand.isSetIdentifier()) {
- String columnName = operand.getIdentifier().getName();
- FieldSpec fieldSpec = schema.getFieldSpecFor(columnName);
- if (fieldSpec == null) {
- return;
- }
- operandType = fieldSpec.getDataType().getStoredType();
- } else {
+ if (function.getOperandsSize() == 0 ||
!function.getOperands().get(0).isSetIdentifier()) {
+ return;
+ }
+ FieldSpec fieldSpec =
schema.getFieldSpecFor(function.getOperands().get(0).getIdentifier().getName());
+ if (fieldSpec == null) {
+ return;
+ }
+
+ String normalizedName =
AggregationFunctionType.getNormalizedAggregationFunctionName(functionName);
+ if ((normalizedName.equals(AggregationFunctionType.FIRSTWITHTIME.name())
+ || normalizedName.equals(AggregationFunctionType.LASTWITHTIME.name()))
&& function.getOperandsSize() == 2) {
+ addDataTypeArgument(function, normalizedName, 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.
+ rewriteToTypedVariant(function, functionName,
fieldSpec.getDataType().getStoredType());
+ }
+ }
+
+ /// Appends the data column's logical type. The supported types match
`PinotAggregateTypeArgumentRewriteRule` in the
+ /// multi-stage engine.
+ private static void addDataTypeArgument(Function function, String
functionName, FieldSpec fieldSpec) {
+ if (fieldSpec.isSingleValueField()) {
+ switch (fieldSpec.getDataType()) {
+ case INT:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ case BIG_DECIMAL:
+ case STRING:
+ case BOOLEAN:
+ case TIMESTAMP:
+ case BYTES:
+ case UUID:
+
function.addToOperands(RequestUtils.getLiteralExpression(fieldSpec.getDataType().name()));
+ return;
+ case JSON:
+ // Same as the multi-stage engine, which types a JSON column as
VARCHAR.
+
function.addToOperands(RequestUtils.getLiteralExpression(FieldSpec.DataType.STRING.name()));
+ return;
+ default:
+ break;
+ }
+ }
+ throw new BadQueryRequestException(String.format("Unsupported data type
for %s(dataColumn, timeColumn): %s%s",
+ functionName, fieldSpec.isSingleValueField() ? "" : "multi-value ",
fieldSpec.getDataType()));
+ }
+ 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
if ((functionName.equals(AggregationFunctionType.MIN.getName())
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..d02250c778c 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
@@ -32,6 +32,7 @@ import it.unimi.dsi.fastutil.longs.Long2LongOpenHashMap;
import it.unimi.dsi.fastutil.longs.LongArrayList;
import it.unimi.dsi.fastutil.objects.ObjectArrayList;
import it.unimi.dsi.fastutil.objects.ObjectLinkedOpenHashSet;
+import java.math.BigDecimal;
import java.nio.BufferOverflowException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
@@ -60,6 +61,8 @@ import
org.apache.pinot.core.query.aggregation.function.PercentileEstAggregation
import
org.apache.pinot.core.query.aggregation.function.PercentileTDigestAggregationFunction;
import org.apache.pinot.core.query.aggregation.function.funnel.FunnelStepEvent;
import org.apache.pinot.segment.local.customobject.AvgPair;
+import org.apache.pinot.segment.local.customobject.BigDecimalLongPair;
+import org.apache.pinot.segment.local.customobject.BytesLongPair;
import org.apache.pinot.segment.local.customobject.CpcSketchAccumulator;
import org.apache.pinot.segment.local.customobject.DoubleLongPair;
import org.apache.pinot.segment.local.customobject.FloatLongPair;
@@ -76,6 +79,7 @@ import
org.apache.pinot.segment.local.customobject.tdigest.TDigest;
import org.apache.pinot.segment.local.customobject.tdigest.TDigest.Centroid;
import org.apache.pinot.segment.local.customobject.tdigest.TDigestAccumulator;
import org.apache.pinot.segment.local.utils.UltraLogLogUtils;
+import org.apache.pinot.spi.utils.ByteArray;
import org.roaringbitmap.RoaringBitmap;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@@ -407,6 +411,36 @@ public class ObjectSerDeUtilsTest {
}
}
+ @Test
+ public void testBigDecimalValueTimePair() {
+ for (int i = 0; i < NUM_ITERATIONS; i++) {
+ ValueLongPair<BigDecimal> expected =
+ new BigDecimalLongPair(BigDecimal.valueOf(RANDOM.nextLong(),
RANDOM.nextInt(20)), RANDOM.nextLong());
+
+ byte[] bytes = ObjectSerDeUtils.serialize(expected);
+ ValueLongPair<BigDecimal> actual =
+ ObjectSerDeUtils.deserialize(bytes,
ObjectSerDeUtils.ObjectType.BigDecimalLongPair);
+
+ assertEquals(actual.getValue(), expected.getValue(), ERROR_MESSAGE);
+ assertEquals(actual.getTime(), expected.getTime(), ERROR_MESSAGE);
+ }
+ }
+
+ @Test
+ public void testBytesValueTimePair() {
+ for (int i = 0; i < NUM_ITERATIONS; i++) {
+ byte[] value = new byte[RANDOM.nextInt(32)];
+ RANDOM.nextBytes(value);
+ ValueLongPair<ByteArray> expected = new BytesLongPair(new
ByteArray(value), RANDOM.nextLong());
+
+ byte[] bytes = ObjectSerDeUtils.serialize(expected);
+ ValueLongPair<ByteArray> actual = ObjectSerDeUtils.deserialize(bytes,
ObjectSerDeUtils.ObjectType.BytesLongPair);
+
+ assertEquals(actual.getValue(), expected.getValue(), ERROR_MESSAGE);
+ assertEquals(actual.getTime(), expected.getTime(), ERROR_MESSAGE);
+ }
+ }
+
@Test
public void testHyperLogLog() {
for (int i = 0; i < NUM_ITERATIONS; i++) {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtilsTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtilsTest.java
index 53cd082fc0c..1687ab8196a 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtilsTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AggregationFunctionUtilsTest.java
@@ -18,10 +18,19 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.io.IOException;
+import java.util.UUID;
+import org.apache.pinot.common.datatable.DataTable;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.core.common.datatable.DataTableBuilder;
+import org.apache.pinot.core.common.datatable.DataTableBuilderFactory;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.datasource.DataSource;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.exception.BadQueryRequestException;
+import org.apache.pinot.spi.utils.ByteArray;
+import org.apache.pinot.spi.utils.UuidUtils;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
@@ -33,7 +42,8 @@ import static org.testng.Assert.fail;
/// Unit test for {@link AggregationFunctionUtils#getAggregationResult}, the
metadata/dictionary based aggregation
-/// result resolver used by the non-scan based and partial metadata based
aggregation paths.
+/// result resolver used by the non-scan based and partial metadata based
aggregation paths, and for
+/// {@link AggregationFunctionUtils#getConvertedFinalResult}.
@SuppressWarnings("rawtypes")
public class AggregationFunctionUtilsTest {
@@ -114,4 +124,17 @@ public class AggregationFunctionUtilsTest {
() ->
AggregationFunctionUtils.getAggregationResult(mockFunction(AggregationFunctionType.MIN),
null, 100,
"TEST"));
}
+
+ @Test
+ public void testConvertedFinalResultForUuid()
+ throws IOException {
+ UUID uuid = UUID.randomUUID();
+ DataTableBuilder dataTableBuilder =
DataTableBuilderFactory.getDataTableBuilder(
+ new DataSchema(new String[]{"uuid"}, new
ColumnDataType[]{ColumnDataType.UUID}));
+ dataTableBuilder.startRow();
+ dataTableBuilder.setColumn(0, new ByteArray(UuidUtils.toBytes(uuid)));
+ dataTableBuilder.finishRow();
+ DataTable dataTable = dataTableBuilder.build();
+ assertEquals(AggregationFunctionUtils.getConvertedFinalResult(dataTable,
ColumnDataType.UUID, 0, 0), uuid);
+ }
}
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
new file mode 100644
index 00000000000..1051598be29
--- /dev/null
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java
@@ -0,0 +1,147 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.core.query.optimizer.statement;
+
+import org.apache.pinot.common.request.PinotQuery;
+import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.exception.BadQueryRequestException;
+import org.apache.pinot.sql.parsers.CalciteSqlParser;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.expectThrows;
+
+
+/// Verifies which aggregation calls are rewritten from the schema, with and
without the `autoRewriteAggregationType`
+/// query option. Each test compiles its own queries, so the class has no
shared mutable state.
+public class AggregateFunctionRewriteOptimizerTest {
+ private static final Schema SCHEMA = new
Schema.SchemaBuilder().setSchemaName("testTable")
+ .addSingleValueDimension("intCol", DataType.INT)
+ .addSingleValueDimension("longCol", DataType.LONG)
+ .addSingleValueDimension("floatCol", DataType.FLOAT)
+ .addSingleValueDimension("doubleCol", DataType.DOUBLE)
+ .addSingleValueDimension("bigDecimalCol", DataType.BIG_DECIMAL)
+ .addSingleValueDimension("stringCol", DataType.STRING)
+ .addSingleValueDimension("boolCol", DataType.BOOLEAN)
+ .addSingleValueDimension("timestampCol", DataType.TIMESTAMP)
+ .addSingleValueDimension("jsonCol", DataType.JSON)
+ .addSingleValueDimension("bytesCol", DataType.BYTES)
+ .addSingleValueDimension("uuidCol", DataType.UUID)
+ .addMultiValueDimension("mvCol", DataType.INT)
+ .addSingleValueDimension("timeCol", DataType.LONG)
+ .build();
+ private static final String AUTO_REWRITE = "SET
autoRewriteAggregationType=true; ";
+
+ @DataProvider
+ public Object[][] dataTypes() {
+ return new Object[][]{
+ {"intCol", "INT"}, {"longCol", "LONG"}, {"floatCol", "FLOAT"},
{"doubleCol", "DOUBLE"},
+ {"bigDecimalCol", "BIG_DECIMAL"}, {"stringCol", "STRING"}, {"jsonCol",
"STRING"}, {"boolCol", "BOOLEAN"},
+ {"timestampCol", "TIMESTAMP"}, {"bytesCol", "BYTES"}, {"uuidCol",
"UUID"}
+ };
+ }
+
+ @Test(dataProvider = "dataTypes")
+ public void testValueWithTimeInfersDataType(String column, String dataType) {
+ for (String function : new String[]{"FIRST_WITH_TIME", "LASTWITHTIME"}) {
+ assertRewrite("SELECT " + function + "(" + column + ", timeCol) FROM
testTable",
+ "SELECT " + function + "(" + column + ", timeCol, '" + dataType +
"') FROM testTable");
+ }
+ }
+
+ @Test
+ public void testValueWithTimeRejectsMultiValueDataColumn() {
+ PinotQuery query = CalciteSqlParser.compileToPinotQuery("SELECT
FIRST_WITH_TIME(mvCol, timeCol) FROM testTable");
+ BadQueryRequestException exception =
expectThrows(BadQueryRequestException.class,
+ () -> new AggregateFunctionRewriteOptimizer().optimize(query, SCHEMA));
+ assertEquals(exception.getMessage(),
+ "Unsupported data type for FIRSTWITHTIME(dataColumn, timeColumn):
multi-value INT");
+ }
+
+ @Test
+ public void testValueWithTimeKeepsDataColumnsWithoutSchemaType() {
+ // These keep the two-argument form, which fails with an error asking for
the explicit data type.
+ for (String column : new String[]{"missingCol", "trim(stringCol)"}) {
+ assertUnchanged("SELECT FIRST_WITH_TIME(" + column + ", timeCol) FROM
testTable");
+ }
+ assertUnchanged("SELECT FIRST_WITH_TIME(stringCol, timeCol, 'STRING'),
LAST_WITH_TIME(intCol, timeCol, 'LONG') "
+ + "FROM testTable");
+ }
+
+ @Test
+ public void testRewritesNestedAggregationsConsistently() {
+ // Aliases, transforms, FILTER, HAVING and ORDER BY wrap the aggregation
in another function.
+ assertRewrite("SELECT intCol, FIRST_WITH_TIME(stringCol, timeCol) AS v1, "
+ + "trim(LAST_WITH_TIME(stringCol, timeCol)) AS v2, "
+ + "FIRST_WITH_TIME(doubleCol, timeCol) FILTER (WHERE boolCol) "
+ + "FROM testTable GROUP BY intCol HAVING FIRST_WITH_TIME(longCol,
timeCol) > 1 "
+ + "ORDER BY LAST_WITH_TIME(timestampCol, timeCol)",
+ "SELECT intCol, FIRST_WITH_TIME(stringCol, timeCol, 'STRING') AS v1, "
+ + "trim(LAST_WITH_TIME(stringCol, timeCol, 'STRING')) AS v2, "
+ + "FIRST_WITH_TIME(doubleCol, timeCol, 'DOUBLE') FILTER (WHERE
boolCol) "
+ + "FROM testTable GROUP BY intCol HAVING FIRST_WITH_TIME(longCol,
timeCol, 'LONG') > 1 "
+ + "ORDER BY LAST_WITH_TIME(timestampCol, timeCol, 'TIMESTAMP')");
+ }
+
+ @Test
+ public void testTypedVariantsRequireQueryOption() {
+ String query = "SELECT MIN(stringCol) AS m, MAX(longCol), SUM(intCol),
SUM(longCol), MIN(doubleCol) "
+ + "FROM testTable ORDER BY MIN(stringCol)";
+ assertUnchanged(query);
+ assertRewrite(AUTO_REWRITE + query, AUTO_REWRITE
+ + "SELECT MINSTRING(stringCol) AS m, MAXLONG(longCol), SUMINT(intCol),
SUMLONG(longCol), MIN(doubleCol) "
+ + "FROM testTable ORDER BY MINSTRING(stringCol)");
+ }
+
+ @Test
+ public void testTypedVariantsRewriteNestedAggregations() {
+ // Aggregations under an alias, a transform and HAVING are rewritten like
top-level ones, so the select list and
+ // HAVING compute the same aggregation.
+ String query = "SELECT intCol, MIN(longCol) AS m, MAX(longCol) + 1 FROM
testTable GROUP BY intCol "
+ + "HAVING MIN(longCol) > 5";
+ assertUnchanged(query);
+ assertRewrite(AUTO_REWRITE + query, AUTO_REWRITE
+ + "SELECT intCol, MINLONG(longCol) AS m, MAXLONG(longCol) + 1 FROM
testTable GROUP BY intCol "
+ + "HAVING MINLONG(longCol) > 5");
+ }
+
+ @Test
+ public void testNoSchema() {
+ PinotQuery query =
+ CalciteSqlParser.compileToPinotQuery("SELECT
FIRST_WITH_TIME(stringCol, timeCol) FROM testTable");
+ PinotQuery original = query.deepCopy();
+ new AggregateFunctionRewriteOptimizer().optimize(query, null);
+ assertEquals(query, original);
+ }
+
+ private static void assertRewrite(String query, String expectedQuery) {
+ PinotQuery actual = CalciteSqlParser.compileToPinotQuery(query);
+ new AggregateFunctionRewriteOptimizer().optimize(actual, SCHEMA);
+ PinotQuery expected = CalciteSqlParser.compileToPinotQuery(expectedQuery);
+ assertEquals(actual.getSelectList(), expected.getSelectList());
+ assertEquals(actual.getHavingExpression(), expected.getHavingExpression());
+ assertEquals(actual.getOrderByList(), expected.getOrderByList());
+ }
+
+ private static void assertUnchanged(String query) {
+ assertRewrite(query, query);
+ }
+}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/FirstWithTimeQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/FirstWithTimeQueriesTest.java
index 768160e1fef..cf1277a0f16 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/FirstWithTimeQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/FirstWithTimeQueriesTest.java
@@ -21,6 +21,8 @@ package org.apache.pinot.queries;
import com.google.common.collect.Lists;
import java.io.File;
import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -28,6 +30,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.UUID;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.response.broker.BrokerResponseNative;
import org.apache.pinot.common.response.broker.ResultTable;
@@ -55,6 +58,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.utils.BytesUtils;
import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.AfterClass;
@@ -90,6 +94,10 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
private static final String STRING_COLUMN = "stringColumn";
private static final String STRING_NO_DICT_COLUMN = "stringNoDictColumn";
private static final String TIME_COLUMN = "timestampColumn";
+ private static final String TIMESTAMP_VALUE_COLUMN = "timestampValueColumn";
+ private static final String BIG_DECIMAL_COLUMN = "bigDecimalColumn";
+ private static final String BYTES_COLUMN = "bytesColumn";
+ private static final String UUID_COLUMN = "uuidColumn";
private static final Schema SCHEMA = new Schema.SchemaBuilder()
.addSingleValueDimension(BOOL_COLUMN, DataType.BOOLEAN)
.addSingleValueDimension(BOOL_NO_DICT_COLUMN, DataType.BOOLEAN)
@@ -104,6 +112,10 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
.addSingleValueDimension(DOUBLE_NO_DICT_COLUMN, DataType.DOUBLE)
.addSingleValueDimension(STRING_COLUMN, DataType.STRING)
.addSingleValueDimension(STRING_NO_DICT_COLUMN, DataType.STRING)
+ .addSingleValueDimension(TIMESTAMP_VALUE_COLUMN, DataType.TIMESTAMP)
+ .addSingleValueDimension(BYTES_COLUMN, DataType.BYTES)
+ .addSingleValueDimension(UUID_COLUMN, DataType.UUID)
+ .addMetric(BIG_DECIMAL_COLUMN, DataType.BIG_DECIMAL)
.addSingleValueDimension(TIME_COLUMN, DataType.LONG).build();
private static final TableConfig TABLE_CONFIG = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
.setNoDictionaryColumns(
@@ -116,12 +128,20 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
private Float _expectedResultFirstFloat;
private Double _expectedResultFirstDouble;
private String _expectedResultFirstString;
+ private Long _expectedResultFirstTimestamp;
+ private BigDecimal _expectedResultFirstBigDecimal;
+ private byte[] _expectedResultFirstBytes;
+ private String _expectedResultFirstUuid;
private Map<Integer, Boolean> _boolGroupValues;
private Map<Integer, Integer> _intGroupValues;
private Map<Integer, Long> _longGroupValues;
private Map<Integer, Float> _floatGroupValues;
private Map<Integer, Double> _doubleGroupValues;
private Map<Integer, String> _stringGroupValues;
+ private Map<Integer, Long> _timestampGroupValues;
+ private Map<Integer, BigDecimal> _bigDecimalGroupValues;
+ private Map<Integer, byte[]> _bytesGroupValues;
+ private Map<Integer, String> _uuidGroupValues;
private IndexSegment _indexSegment;
private List<IndexSegment> _indexSegments;
@@ -152,6 +172,10 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
_floatGroupValues = new HashMap<>();
_doubleGroupValues = new HashMap<>();
_stringGroupValues = new HashMap<>();
+ _timestampGroupValues = new HashMap<>();
+ _bigDecimalGroupValues = new HashMap<>();
+ _bytesGroupValues = new HashMap<>();
+ _uuidGroupValues = new HashMap<>();
for (int i = 0; i < NUM_RECORDS; i++) {
boolean boolValue = RANDOM.nextBoolean();
int intValue = RANDOM.nextInt(MAX_VALUE);
@@ -159,6 +183,11 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
float floatValue = RANDOM.nextFloat();
double doubleValue = RANDOM.nextDouble();
String strValue = String.valueOf(RANDOM.nextDouble());
+ long timestampValue = 1_600_000_000_000L + RANDOM.nextInt(1_000_000_000);
+ BigDecimal bigDecimalValue = BigDecimal.valueOf(RANDOM.nextLong(), 3);
+ byte[] bytesValue = new byte[4];
+ RANDOM.nextBytes(bytesValue);
+ String uuidValue = new UUID(RANDOM.nextLong(),
RANDOM.nextLong()).toString();
GenericRow record = new GenericRow();
record.putValue(BOOL_COLUMN, boolValue);
record.putValue(BOOL_NO_DICT_COLUMN, boolValue);
@@ -173,6 +202,10 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
record.putValue(DOUBLE_NO_DICT_COLUMN, doubleValue);
record.putValue(STRING_COLUMN, strValue);
record.putValue(STRING_NO_DICT_COLUMN, strValue);
+ record.putValue(TIMESTAMP_VALUE_COLUMN, timestampValue);
+ record.putValue(BIG_DECIMAL_COLUMN, bigDecimalValue);
+ record.putValue(BYTES_COLUMN, bytesValue);
+ record.putValue(UUID_COLUMN, uuidValue);
record.putValue(TIME_COLUMN, (long) i);
if (i == 0) {
_expectedResultFirstBoolean = boolValue;
@@ -181,6 +214,10 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
_expectedResultFirstFloat = floatValue;
_expectedResultFirstDouble = doubleValue;
_expectedResultFirstString = strValue;
+ _expectedResultFirstTimestamp = timestampValue;
+ _expectedResultFirstBigDecimal = bigDecimalValue;
+ _expectedResultFirstBytes = bytesValue;
+ _expectedResultFirstUuid = uuidValue;
}
_boolGroupValues.putIfAbsent(intValue, boolValue);
_intGroupValues.putIfAbsent(intValue, intValue);
@@ -188,6 +225,10 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
_floatGroupValues.putIfAbsent(intValue, floatValue);
_doubleGroupValues.putIfAbsent(intValue, doubleValue);
_stringGroupValues.putIfAbsent(intValue, strValue);
+ _timestampGroupValues.putIfAbsent(intValue, timestampValue);
+ _bigDecimalGroupValues.putIfAbsent(intValue, bigDecimalValue);
+ _bytesGroupValues.putIfAbsent(intValue, bytesValue);
+ _uuidGroupValues.putIfAbsent(intValue, uuidValue);
records.add(record);
}
@@ -423,6 +464,113 @@ public class FirstWithTimeQueriesTest extends
BaseQueriesTest {
}
}
+ /// The data type argument can be omitted for a column; the optimizer
supplies it from the schema.
+ @Test
+ public void testAggregationOnlyWithInferredDataType() {
+ String query = "SELECT "
+ + "FIRSTWITHTIME(boolColumn, timestampColumn), "
+ + "FIRSTWITHTIME(intColumn, timestampColumn), "
+ + "FIRSTWITHTIME(longColumn, timestampColumn), "
+ + "FIRSTWITHTIME(floatColumn, timestampColumn), "
+ + "FIRSTWITHTIME(doubleColumn, timestampColumn), "
+ + "FIRSTWITHTIME(stringColumn, timestampColumn), "
+ + "FIRSTWITHTIME(timestampValueColumn, timestampColumn), "
+ + "FIRSTWITHTIME(timestampValueColumn, timestampColumn, 'TIMESTAMP'), "
+ + "FIRSTWITHTIME(bigDecimalColumn, timestampColumn), "
+ + "FIRSTWITHTIME(bytesColumn, timestampColumn), "
+ + "FIRSTWITHTIME(uuidColumn, timestampColumn) "
+ + "FROM testTable";
+ BrokerResponseNative brokerResponse =
getBrokerResponseForOptimizedQuery(query, SCHEMA);
+ // The result columns are named after the explicit form the query is
rewritten to.
+ DataSchema expectedDataSchema = new DataSchema(new String[]{
+ "firstwithtime(boolColumn,timestampColumn,'BOOLEAN')",
+ "firstwithtime(intColumn,timestampColumn,'INT')",
+ "firstwithtime(longColumn,timestampColumn,'LONG')",
+ "firstwithtime(floatColumn,timestampColumn,'FLOAT')",
+ "firstwithtime(doubleColumn,timestampColumn,'DOUBLE')",
+ "firstwithtime(stringColumn,timestampColumn,'STRING')",
+ "firstwithtime(timestampValueColumn,timestampColumn,'TIMESTAMP')",
+ "firstwithtime(timestampValueColumn,timestampColumn,'TIMESTAMP')",
+ "firstwithtime(bigDecimalColumn,timestampColumn,'BIG_DECIMAL')",
+ "firstwithtime(bytesColumn,timestampColumn,'BYTES')",
+ "firstwithtime(uuidColumn,timestampColumn,'UUID')"
+ }, new ColumnDataType[]{
+ ColumnDataType.BOOLEAN,
+ ColumnDataType.INT,
+ ColumnDataType.LONG,
+ ColumnDataType.FLOAT,
+ ColumnDataType.DOUBLE,
+ ColumnDataType.STRING,
+ ColumnDataType.TIMESTAMP,
+ ColumnDataType.TIMESTAMP,
+ ColumnDataType.BIG_DECIMAL,
+ ColumnDataType.BYTES,
+ ColumnDataType.UUID
+ });
+ String expectedTimestamp = new
Timestamp(_expectedResultFirstTimestamp).toString();
+ Object[] expectedResults = new Object[]{
+ _expectedResultFirstBoolean,
+ _expectedResultFirstInt,
+ _expectedResultFirstLong,
+ _expectedResultFirstFloat,
+ _expectedResultFirstDouble,
+ _expectedResultFirstString,
+ expectedTimestamp,
+ expectedTimestamp,
+ // Ingestion strips trailing zeros from BIG_DECIMAL values.
+ _expectedResultFirstBigDecimal.stripTrailingZeros().toPlainString(),
+ BytesUtils.toHexString(_expectedResultFirstBytes),
+ _expectedResultFirstUuid
+ };
+ QueriesTestUtils.testInterSegmentsResult(brokerResponse, 4 * NUM_RECORDS,
0L, 4 * 11 * NUM_RECORDS,
+ 4 * NUM_RECORDS, new ResultTable(expectedDataSchema,
List.<Object[]>of(expectedResults)));
+ }
+
+ @Test
+ public void testAggregationOnlyWithInferredDataTypeAndNoMatchingRows() {
+ // Without matching rows, the results are the default values; for UUID it
is the nil UUID.
+ String query = "SELECT FIRSTWITHTIME(bigDecimalColumn, timestampColumn), "
+ + "FIRSTWITHTIME(bytesColumn, timestampColumn),
FIRSTWITHTIME(uuidColumn, timestampColumn) "
+ + "FROM testTable WHERE timestampColumn < 0";
+ ResultTable resultTable = getBrokerResponseForOptimizedQuery(query,
SCHEMA).getResultTable();
+ assertEquals(resultTable.getRows().get(0), new Object[]{"0", "",
"00000000-0000-0000-0000-000000000000"});
+ }
+
+ @Test
+ public void testAggregationGroupByWithInferredDataType() {
+ String query = "SELECT intColumn AS key, "
+ + "FIRSTWITHTIME(boolColumn, timestampColumn) AS v1, "
+ + "FIRSTWITHTIME(longNoDictColumn, timestampColumn) AS v2, "
+ + "FIRSTWITHTIME(stringColumn, timestampColumn) AS v3, "
+ + "FIRSTWITHTIME(timestampValueColumn, timestampColumn) AS v4, "
+ + "FIRSTWITHTIME(bytesColumn, timestampColumn) AS v5, "
+ + "FIRSTWITHTIME(uuidColumn, timestampColumn) AS v6, "
+ + "FIRSTWITHTIME(bigDecimalColumn, timestampColumn) AS v7 "
+ + "FROM testTable GROUP BY key ORDER BY
FIRSTWITHTIME(longNoDictColumn, timestampColumn)";
+ ResultTable resultTable = getBrokerResponseForOptimizedQuery(query,
SCHEMA).getResultTable();
+ assertEquals(resultTable.getDataSchema(),
+ new DataSchema(new String[]{"key", "v1", "v2", "v3", "v4", "v5", "v6",
"v7"}, new ColumnDataType[]{
+ ColumnDataType.INT, ColumnDataType.BOOLEAN, ColumnDataType.LONG,
ColumnDataType.STRING,
+ ColumnDataType.TIMESTAMP, ColumnDataType.BYTES,
ColumnDataType.UUID, ColumnDataType.BIG_DECIMAL
+ }));
+ List<Object[]> rows = resultTable.getRows();
+ assertEquals(rows.size(), 10);
+ long previous = Long.MIN_VALUE;
+ for (Object[] row : rows) {
+ int key = (Integer) row[0];
+ assertEquals(row[1], _boolGroupValues.get(key));
+ assertEquals(row[2], _longGroupValues.get(key));
+ assertEquals(row[3], _stringGroupValues.get(key));
+ assertEquals(row[4], new
Timestamp(_timestampGroupValues.get(key)).toString());
+ assertEquals(row[5], BytesUtils.toHexString(_bytesGroupValues.get(key)));
+ assertEquals(row[6], _uuidGroupValues.get(key));
+ assertEquals(row[7],
_bigDecimalGroupValues.get(key).stripTrailingZeros().toPlainString());
+ // ORDER BY repeats the call, so it must be rewritten the same way as
the SELECT list.
+ assertTrue((Long) row[2] >= previous);
+ previous = (Long) row[2];
+ }
+ }
+
@AfterClass
public void tearDown()
throws IOException {
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/LastWithTimeQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/LastWithTimeQueriesTest.java
index 2dd14e5455e..dbc8ba0714f 100644
---
a/pinot-core/src/test/java/org/apache/pinot/queries/LastWithTimeQueriesTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/queries/LastWithTimeQueriesTest.java
@@ -21,6 +21,8 @@ package org.apache.pinot.queries;
import com.google.common.collect.Lists;
import java.io.File;
import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
@@ -28,6 +30,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Random;
+import java.util.UUID;
import org.apache.commons.io.FileUtils;
import org.apache.pinot.common.response.broker.BrokerResponseNative;
import org.apache.pinot.common.response.broker.ResultTable;
@@ -55,6 +58,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.utils.BytesUtils;
import org.apache.pinot.spi.utils.ReadMode;
import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
import org.testng.annotations.AfterClass;
@@ -90,6 +94,10 @@ public class LastWithTimeQueriesTest extends BaseQueriesTest
{
private static final String STRING_COLUMN = "stringColumn";
private static final String STRING_NO_DICT_COLUMN = "stringNoDictColumn";
private static final String TIME_COLUMN = "timestampColumn";
+ private static final String TIMESTAMP_VALUE_COLUMN = "timestampValueColumn";
+ private static final String BIG_DECIMAL_COLUMN = "bigDecimalColumn";
+ private static final String BYTES_COLUMN = "bytesColumn";
+ private static final String UUID_COLUMN = "uuidColumn";
private static final Schema SCHEMA = new Schema.SchemaBuilder()
.addSingleValueDimension(BOOL_COLUMN, DataType.BOOLEAN)
.addSingleValueDimension(BOOL_NO_DICT_COLUMN, DataType.BOOLEAN)
@@ -104,6 +112,10 @@ public class LastWithTimeQueriesTest extends
BaseQueriesTest {
.addSingleValueDimension(DOUBLE_NO_DICT_COLUMN, DataType.DOUBLE)
.addSingleValueDimension(STRING_COLUMN, DataType.STRING)
.addSingleValueDimension(STRING_NO_DICT_COLUMN, DataType.STRING)
+ .addSingleValueDimension(TIMESTAMP_VALUE_COLUMN, DataType.TIMESTAMP)
+ .addSingleValueDimension(BYTES_COLUMN, DataType.BYTES)
+ .addSingleValueDimension(UUID_COLUMN, DataType.UUID)
+ .addMetric(BIG_DECIMAL_COLUMN, DataType.BIG_DECIMAL)
.addSingleValueDimension(TIME_COLUMN, DataType.LONG).build();
private static final TableConfig TABLE_CONFIG = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
.setNoDictionaryColumns(
@@ -116,12 +128,20 @@ public class LastWithTimeQueriesTest extends
BaseQueriesTest {
private Float _expectedResultLastFloat;
private Double _expectedResultLastDouble;
private String _expectedResultLastString;
+ private Long _expectedResultLastTimestamp;
+ private BigDecimal _expectedResultLastBigDecimal;
+ private byte[] _expectedResultLastBytes;
+ private String _expectedResultLastUuid;
private Map<Integer, Boolean> _boolGroupValues;
private Map<Integer, Integer> _intGroupValues;
private Map<Integer, Long> _longGroupValues;
private Map<Integer, Float> _floatGroupValues;
private Map<Integer, Double> _doubleGroupValues;
private Map<Integer, String> _stringGroupValues;
+ private Map<Integer, Long> _timestampGroupValues;
+ private Map<Integer, BigDecimal> _bigDecimalGroupValues;
+ private Map<Integer, byte[]> _bytesGroupValues;
+ private Map<Integer, String> _uuidGroupValues;
private IndexSegment _indexSegment;
private List<IndexSegment> _indexSegments;
@@ -152,6 +172,10 @@ public class LastWithTimeQueriesTest extends
BaseQueriesTest {
_floatGroupValues = new HashMap<>();
_doubleGroupValues = new HashMap<>();
_stringGroupValues = new HashMap<>();
+ _timestampGroupValues = new HashMap<>();
+ _bigDecimalGroupValues = new HashMap<>();
+ _bytesGroupValues = new HashMap<>();
+ _uuidGroupValues = new HashMap<>();
for (int i = 0; i < NUM_RECORDS; i++) {
boolean boolValue = RANDOM.nextBoolean();
int intValue = RANDOM.nextInt(MAX_VALUE);
@@ -159,6 +183,11 @@ public class LastWithTimeQueriesTest extends
BaseQueriesTest {
float floatValue = RANDOM.nextFloat();
double doubleValue = RANDOM.nextDouble();
String strValue = String.valueOf(RANDOM.nextDouble());
+ long timestampValue = 1_600_000_000_000L + RANDOM.nextInt(1_000_000_000);
+ BigDecimal bigDecimalValue = BigDecimal.valueOf(RANDOM.nextLong(), 3);
+ byte[] bytesValue = new byte[4];
+ RANDOM.nextBytes(bytesValue);
+ String uuidValue = new UUID(RANDOM.nextLong(),
RANDOM.nextLong()).toString();
GenericRow record = new GenericRow();
record.putValue(BOOL_COLUMN, boolValue);
record.putValue(BOOL_NO_DICT_COLUMN, boolValue);
@@ -173,6 +202,10 @@ public class LastWithTimeQueriesTest extends
BaseQueriesTest {
record.putValue(DOUBLE_NO_DICT_COLUMN, doubleValue);
record.putValue(STRING_COLUMN, strValue);
record.putValue(STRING_NO_DICT_COLUMN, strValue);
+ record.putValue(TIMESTAMP_VALUE_COLUMN, timestampValue);
+ record.putValue(BIG_DECIMAL_COLUMN, bigDecimalValue);
+ record.putValue(BYTES_COLUMN, bytesValue);
+ record.putValue(UUID_COLUMN, uuidValue);
record.putValue(TIME_COLUMN, (long) i);
if (i == NUM_RECORDS - 1) {
_expectedResultLastBoolean = boolValue;
@@ -181,6 +214,10 @@ public class LastWithTimeQueriesTest extends
BaseQueriesTest {
_expectedResultLastFloat = floatValue;
_expectedResultLastDouble = doubleValue;
_expectedResultLastString = strValue;
+ _expectedResultLastTimestamp = timestampValue;
+ _expectedResultLastBigDecimal = bigDecimalValue;
+ _expectedResultLastBytes = bytesValue;
+ _expectedResultLastUuid = uuidValue;
}
_boolGroupValues.put(intValue, boolValue);
_intGroupValues.put(intValue, intValue);
@@ -188,6 +225,10 @@ public class LastWithTimeQueriesTest extends
BaseQueriesTest {
_floatGroupValues.put(intValue, floatValue);
_doubleGroupValues.put(intValue, doubleValue);
_stringGroupValues.put(intValue, strValue);
+ _timestampGroupValues.put(intValue, timestampValue);
+ _bigDecimalGroupValues.put(intValue, bigDecimalValue);
+ _bytesGroupValues.put(intValue, bytesValue);
+ _uuidGroupValues.put(intValue, uuidValue);
records.add(record);
}
@@ -423,6 +464,113 @@ public class LastWithTimeQueriesTest extends
BaseQueriesTest {
}
}
+ /// The data type argument can be omitted for a column; the optimizer
supplies it from the schema.
+ @Test
+ public void testAggregationOnlyWithInferredDataType() {
+ String query = "SELECT "
+ + "LASTWITHTIME(boolColumn, timestampColumn), "
+ + "LASTWITHTIME(intColumn, timestampColumn), "
+ + "LASTWITHTIME(longColumn, timestampColumn), "
+ + "LASTWITHTIME(floatColumn, timestampColumn), "
+ + "LASTWITHTIME(doubleColumn, timestampColumn), "
+ + "LASTWITHTIME(stringColumn, timestampColumn), "
+ + "LASTWITHTIME(timestampValueColumn, timestampColumn), "
+ + "LASTWITHTIME(timestampValueColumn, timestampColumn, 'TIMESTAMP'), "
+ + "LASTWITHTIME(bigDecimalColumn, timestampColumn), "
+ + "LASTWITHTIME(bytesColumn, timestampColumn), "
+ + "LASTWITHTIME(uuidColumn, timestampColumn) "
+ + "FROM testTable";
+ BrokerResponseNative brokerResponse =
getBrokerResponseForOptimizedQuery(query, SCHEMA);
+ // The result columns are named after the explicit form the query is
rewritten to.
+ DataSchema expectedDataSchema = new DataSchema(new String[]{
+ "lastwithtime(boolColumn,timestampColumn,'BOOLEAN')",
+ "lastwithtime(intColumn,timestampColumn,'INT')",
+ "lastwithtime(longColumn,timestampColumn,'LONG')",
+ "lastwithtime(floatColumn,timestampColumn,'FLOAT')",
+ "lastwithtime(doubleColumn,timestampColumn,'DOUBLE')",
+ "lastwithtime(stringColumn,timestampColumn,'STRING')",
+ "lastwithtime(timestampValueColumn,timestampColumn,'TIMESTAMP')",
+ "lastwithtime(timestampValueColumn,timestampColumn,'TIMESTAMP')",
+ "lastwithtime(bigDecimalColumn,timestampColumn,'BIG_DECIMAL')",
+ "lastwithtime(bytesColumn,timestampColumn,'BYTES')",
+ "lastwithtime(uuidColumn,timestampColumn,'UUID')"
+ }, new ColumnDataType[]{
+ ColumnDataType.BOOLEAN,
+ ColumnDataType.INT,
+ ColumnDataType.LONG,
+ ColumnDataType.FLOAT,
+ ColumnDataType.DOUBLE,
+ ColumnDataType.STRING,
+ ColumnDataType.TIMESTAMP,
+ ColumnDataType.TIMESTAMP,
+ ColumnDataType.BIG_DECIMAL,
+ ColumnDataType.BYTES,
+ ColumnDataType.UUID
+ });
+ String expectedTimestamp = new
Timestamp(_expectedResultLastTimestamp).toString();
+ Object[] expectedResults = new Object[]{
+ _expectedResultLastBoolean,
+ _expectedResultLastInt,
+ _expectedResultLastLong,
+ _expectedResultLastFloat,
+ _expectedResultLastDouble,
+ _expectedResultLastString,
+ expectedTimestamp,
+ expectedTimestamp,
+ // Ingestion strips trailing zeros from BIG_DECIMAL values.
+ _expectedResultLastBigDecimal.stripTrailingZeros().toPlainString(),
+ BytesUtils.toHexString(_expectedResultLastBytes),
+ _expectedResultLastUuid
+ };
+ QueriesTestUtils.testInterSegmentsResult(brokerResponse, 4 * NUM_RECORDS,
0L, 4 * 11 * NUM_RECORDS,
+ 4 * NUM_RECORDS, new ResultTable(expectedDataSchema,
List.<Object[]>of(expectedResults)));
+ }
+
+ @Test
+ public void testAggregationOnlyWithInferredDataTypeAndNoMatchingRows() {
+ // Without matching rows, the results are the default values; for UUID it
is the nil UUID.
+ String query = "SELECT LASTWITHTIME(bigDecimalColumn, timestampColumn), "
+ + "LASTWITHTIME(bytesColumn, timestampColumn),
LASTWITHTIME(uuidColumn, timestampColumn) "
+ + "FROM testTable WHERE timestampColumn < 0";
+ ResultTable resultTable = getBrokerResponseForOptimizedQuery(query,
SCHEMA).getResultTable();
+ assertEquals(resultTable.getRows().get(0), new Object[]{"0", "",
"00000000-0000-0000-0000-000000000000"});
+ }
+
+ @Test
+ public void testAggregationGroupByWithInferredDataType() {
+ String query = "SELECT intColumn AS key, "
+ + "LASTWITHTIME(boolColumn, timestampColumn) AS v1, "
+ + "LASTWITHTIME(longNoDictColumn, timestampColumn) AS v2, "
+ + "LASTWITHTIME(stringColumn, timestampColumn) AS v3, "
+ + "LASTWITHTIME(timestampValueColumn, timestampColumn) AS v4, "
+ + "LASTWITHTIME(bytesColumn, timestampColumn) AS v5, "
+ + "LASTWITHTIME(uuidColumn, timestampColumn) AS v6, "
+ + "LASTWITHTIME(bigDecimalColumn, timestampColumn) AS v7 "
+ + "FROM testTable GROUP BY key ORDER BY LASTWITHTIME(longNoDictColumn,
timestampColumn)";
+ ResultTable resultTable = getBrokerResponseForOptimizedQuery(query,
SCHEMA).getResultTable();
+ assertEquals(resultTable.getDataSchema(),
+ new DataSchema(new String[]{"key", "v1", "v2", "v3", "v4", "v5", "v6",
"v7"}, new ColumnDataType[]{
+ ColumnDataType.INT, ColumnDataType.BOOLEAN, ColumnDataType.LONG,
ColumnDataType.STRING,
+ ColumnDataType.TIMESTAMP, ColumnDataType.BYTES,
ColumnDataType.UUID, ColumnDataType.BIG_DECIMAL
+ }));
+ List<Object[]> rows = resultTable.getRows();
+ assertEquals(rows.size(), 10);
+ long previous = Long.MIN_VALUE;
+ for (Object[] row : rows) {
+ int key = (Integer) row[0];
+ assertEquals(row[1], _boolGroupValues.get(key));
+ assertEquals(row[2], _longGroupValues.get(key));
+ assertEquals(row[3], _stringGroupValues.get(key));
+ assertEquals(row[4], new
Timestamp(_timestampGroupValues.get(key)).toString());
+ assertEquals(row[5], BytesUtils.toHexString(_bytesGroupValues.get(key)));
+ assertEquals(row[6], _uuidGroupValues.get(key));
+ assertEquals(row[7],
_bigDecimalGroupValues.get(key).stripTrailingZeros().toPlainString());
+ // ORDER BY repeats the call, so it must be rewritten the same way as
the SELECT list.
+ assertTrue((Long) row[2] >= previous);
+ previous = (Long) row[2];
+ }
+ }
+
@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
new file mode 100644
index 00000000000..aa739af7230
--- /dev/null
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java
@@ -0,0 +1,164 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.calcite.rel.rules;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import javax.annotation.Nullable;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Aggregate;
+import org.apache.calcite.rel.core.AggregateCall;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.hint.RelHint;
+import org.apache.calcite.rel.logical.LogicalAggregate;
+import org.apache.calcite.rel.logical.LogicalProject;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.tools.RelBuilderFactory;
+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')
+///
+/// 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`
+/// 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`
+/// always applies it, and `skipPlannerRules` cannot turn it off.
+public class PinotAggregateTypeArgumentRewriteRule extends RelOptRule {
+ public static final PinotAggregateTypeArgumentRewriteRule INSTANCE =
+ new
PinotAggregateTypeArgumentRewriteRule(PinotRuleUtils.PINOT_REL_FACTORY);
+
+ // Matches with RelOptRule operands, like the other aggregate rewrite rules,
instead of a RelRule config.
+ @SuppressWarnings("deprecation")
+ private PinotAggregateTypeArgumentRewriteRule(RelBuilderFactory factory) {
+ super(operand(LogicalAggregate.class, any()), factory, null);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ // 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);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ Aggregate aggRel = call.rel(0);
+ RelNode input = aggRel.getInput();
+ RelNode unboxedInput = PinotRuleUtils.unboxRel(input);
+ RexBuilder rexBuilder = aggRel.getCluster().getRexBuilder();
+
+ List<RexNode> projects;
+ RelNode projectInput;
+ List<RelHint> hints;
+ if (unboxedInput instanceof Project) {
+ Project project = (Project) unboxedInput;
+ projects = new ArrayList<>(project.getProjects());
+ projectInput = project.getInput();
+ hints = project.getHints();
+ } else {
+ int numFields = input.getRowType().getFieldCount();
+ projects = new ArrayList<>(numFields + 1);
+ for (int i = 0; i < numFields; i++) {
+ projects.add(rexBuilder.makeInputRef(input, i));
+ }
+ projectInput = input;
+ hints = List.of();
+ }
+
+ // 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<>();
+ for (AggregateCall aggCall : orgAggCalls) {
+ String dataType = getDataTypeArgument(aggCall, input);
+ if (dataType == 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;
+ }));
+ // 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. Throws
+ /// for a data argument without an explicit form, such as an array from a
multi-value column.
+ @Nullable
+ private static String getDataTypeArgument(AggregateCall aggCall, RelNode
input) {
+ if (aggCall.getArgList().size() != 2) {
+ return null;
+ }
+ String name =
AggregationFunctionType.getNormalizedAggregationFunctionName(aggCall.getAggregation().getName());
+ if (!name.equals(AggregationFunctionType.FIRSTWITHTIME.name())
+ && !name.equals(AggregationFunctionType.LASTWITHTIME.name())) {
+ return null;
+ }
+ SqlTypeName dataType =
+
input.getRowType().getFieldList().get(aggCall.getArgList().get(0)).getType().getSqlTypeName();
+ switch (dataType) {
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ return "INT";
+ case BIGINT:
+ return "LONG";
+ case REAL:
+ case FLOAT:
+ return "FLOAT";
+ case DOUBLE:
+ return "DOUBLE";
+ case DECIMAL:
+ return "BIG_DECIMAL";
+ case CHAR:
+ case VARCHAR:
+ return "STRING";
+ case BOOLEAN:
+ return "BOOLEAN";
+ case TIMESTAMP:
+ return "TIMESTAMP";
+ case BINARY:
+ case VARBINARY:
+ return "BYTES";
+ case UUID:
+ return "UUID";
+ default:
+ throw new IllegalArgumentException(
+ String.format("Unsupported data type for %s(dataColumn,
timeColumn): %s", name, dataType));
+ }
+ }
+}
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 ce5d25c7558..31531c92139 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
@@ -58,6 +58,7 @@ import org.apache.calcite.tools.RelBuilder;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.pinot.calcite.rel.rules.ImmutablePinotSortExchangeCopyRule;
+import
org.apache.pinot.calcite.rel.rules.PinotAggregateTypeArgumentRewriteRule;
import org.apache.pinot.calcite.rel.rules.PinotImplicitTableHintRule;
import org.apache.pinot.calcite.rel.rules.PinotJoinToDynamicBroadcastRule;
import org.apache.pinot.calcite.rel.rules.PinotRelDistributionTraitRule;
@@ -604,6 +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.
+
hepProgramBuilder.addRuleInstance(PinotAggregateTypeArgumentRewriteRule.INSTANCE);
// ----
// Pushdown filters using a single HepInstruction.
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
index 63eec96d1dc..98d17a494c4 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
@@ -1814,6 +1814,17 @@ public class QueryCompilationTest extends
QueryEnvironmentTestBase {
assertTrue(exception.getMessage().contains(QueryOptionKey.INFER_PARTITION_HINT),
exception.getMessage());
}
+ @Test
+ public void testValueWithTimeRejectsUnsupportedDataArgument() {
+ // The two-argument form infers the data type from the data argument,
which cannot be multi-value.
+ QueryException exception = expectThrows(QueryException.class, () ->
_queryEnvironment.explainQuery(
+ "SELECT FIRST_WITH_TIME(mcol1, ts) FROM e",
RANDOM_REQUEST_ID_GEN.nextLong()));
+ assertEquals(exception.getErrorCode(), QueryErrorCode.QUERY_PLANNING);
+ assertTrue(
+ exception.getMessage().contains("Unsupported data type for
FIRSTWITHTIME(dataColumn, timeColumn): ARRAY"),
+ exception.getMessage());
+ }
+
@DataProvider(name = "testQueryExceptionDataProvider")
private Object[][] provideQueriesWithException() {
return new Object[][]{
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryPlannerRuleOptionsTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryPlannerRuleOptionsTest.java
index be830d02240..7ed615549e2 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryPlannerRuleOptionsTest.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryPlannerRuleOptionsTest.java
@@ -21,6 +21,7 @@ package org.apache.pinot.query;
import java.util.HashMap;
import java.util.Map;
import org.apache.calcite.sql.SqlNode;
+import
org.apache.pinot.calcite.rel.rules.PinotAggregateTypeArgumentRewriteRule;
import org.apache.pinot.common.utils.config.QueryOptionsUtils;
import org.apache.pinot.spi.utils.CommonConstants;
import org.apache.pinot.spi.utils.CommonConstants.Broker.PlannerRuleNames;
@@ -93,6 +94,15 @@ public class QueryPlannerRuleOptionsTest extends
QueryEnvironmentTestBase {
//@formatter:on
}
+ @Test
+ public void testTypeArgumentRewriteIgnoresSkipPlannerRules() {
+ // The servers require the data type argument, so naming the rule in
skipPlannerRules does not turn it off.
+ String query = "EXPLAIN PLAN FOR SELECT FIRST_WITH_TIME(col1, ts) FROM a";
+ String explain = explainQueryWithRuleDisabled(query,
PinotAggregateTypeArgumentRewriteRule.INSTANCE.toString());
+ assertEquals(explain, _queryEnvironment.explainQuery(query,
RANDOM_REQUEST_ID_GEN.nextLong()));
+ assertTrue(explain.contains("FIRSTWITHTIME($0, $1, _UTF-8'STRING')"),
explain);
+ }
+
@Test
public void testDisableReduceFunctions() {
// Tests that when skipPinotAggregateReduceFunctionsRule=true,
diff --git a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
index 364a54fda9d..f1dbfd4813e 100644
--- a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
+++ b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
@@ -158,6 +158,45 @@
"\n"
]
},
+ {
+ "description": "FIRST_WITH_TIME and LAST_WITH_TIME without a data type
infer it from the data argument",
+ "sql": "EXPLAIN PLAN FOR SELECT FIRST_WITH_TIME(a.col1, a.ts),
LAST_WITH_TIME(a.ts_timestamp, a.ts), FIRST_WITH_TIME(a.col5, a.ts),
LAST_WITH_TIME(a.col1, a.ts) FROM a",
+ "output": [
+ "Execution Plan",
+ "\nPinotLogicalAggregate(group=[{}], agg#0=[FIRSTWITHTIME($0, $0,
_UTF-8'STRING')], agg#1=[LASTWITHTIME($1, $1, _UTF-8'TIMESTAMP')],
agg#2=[FIRSTWITHTIME($2, $2, _UTF-8'BOOLEAN')], agg#3=[LASTWITHTIME($3, $3,
_UTF-8'STRING')], aggType=[FINAL])",
+ "\n PinotLogicalExchange(distribution=[hash])",
+ "\n PinotLogicalAggregate(group=[{}], agg#0=[FIRSTWITHTIME($0,
$1, _UTF-8'STRING')], agg#1=[LASTWITHTIME($2, $1, _UTF-8'TIMESTAMP')],
agg#2=[FIRSTWITHTIME($3, $1, _UTF-8'BOOLEAN')], agg#3=[LASTWITHTIME($0, $1,
_UTF-8'STRING')], aggType=[LEAF])",
+ "\n LogicalProject($f0=[$0], $f1=[$7], $f2=[$8], $f3=[$4],
$f4=[_UTF-8'STRING'], $f5=[_UTF-8'TIMESTAMP'], $f6=[_UTF-8'BOOLEAN'])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Inferred data type argument next to another literal
argument in a grouped aggregate",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col2, FIRST_WITH_TIME(a.col3 + 1,
a.ts), PERCENTILE(a.col3, 50) FROM a GROUP BY a.col2",
+ "output": [
+ "Execution Plan",
+ "\nPinotLogicalAggregate(group=[{0}], agg#0=[FIRSTWITHTIME($1, $1,
_UTF-8'INT')], agg#1=[PERCENTILE($2, 50)], aggType=[FINAL])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n PinotLogicalAggregate(group=[{0}], agg#0=[FIRSTWITHTIME($1,
$2, _UTF-8'INT')], agg#1=[PERCENTILE($3, 50)], aggType=[LEAF])",
+ "\n LogicalProject($f0=[$1], $f1=[+($2, 1)], $f2=[$7],
$f3=[$2], $f4=[50], $f5=[_UTF-8'INT'])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "FIRST_WITH_TIME and LAST_WITH_TIME without a data type
infer BIG_DECIMAL and BYTES",
+ "sql": "EXPLAIN PLAN FOR SELECT FIRST_WITH_TIME(a.col4, a.ts),
LAST_WITH_TIME(hexToBytes(a.col1), a.ts) FROM a",
+ "output": [
+ "Execution Plan",
+ "\nPinotLogicalAggregate(group=[{}], agg#0=[FIRSTWITHTIME($0, $0,
_UTF-8'BIG_DECIMAL')], agg#1=[LASTWITHTIME($1, $1, _UTF-8'BYTES')],
aggType=[FINAL])",
+ "\n PinotLogicalExchange(distribution=[hash])",
+ "\n PinotLogicalAggregate(group=[{}], agg#0=[FIRSTWITHTIME($0,
$1, _UTF-8'BIG_DECIMAL')], agg#1=[LASTWITHTIME($2, $1, _UTF-8'BYTES')],
aggType=[LEAF])",
+ "\n LogicalProject($f0=[$3], $f1=[$7], $f2=[HEXTOBYTES($0)],
$f3=[_UTF-8'BIG_DECIMAL'], $f4=[_UTF-8'BYTES'])",
+ "\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/java/org/apache/pinot/query/runtime/queries/QueryRunnerTestBase.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTestBase.java
index ea375c4069d..ddd6070414e 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTestBase.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/queries/QueryRunnerTestBase.java
@@ -73,6 +73,7 @@ import org.apache.pinot.spi.utils.CommonConstants.Broker;
import
org.apache.pinot.spi.utils.CommonConstants.Broker.Request.QueryOptionKey;
import org.apache.pinot.spi.utils.CommonConstants.Query.Request.MetadataKeys;
import org.apache.pinot.spi.utils.StringUtil;
+import org.apache.pinot.spi.utils.UuidUtils;
import org.apache.pinot.sql.parsers.CalciteSqlParser;
import org.apache.pinot.sql.parsers.SqlNodeAndOptions;
import org.h2.jdbc.JdbcArray;
@@ -319,6 +320,8 @@ public abstract class QueryRunnerTestBase extends
QueryTestSet {
assertTrue(value instanceof String,
"Got unexpected value type: " + value.getClass() + " for BYTES
column, expected: String or byte[]");
return value;
+ case UUID:
+ return UuidUtils.toUUID(value).toString();
case INT_ARRAY:
if (value instanceof List) {
return ((List) value).stream().mapToInt(i -> (int) i).toArray();
@@ -427,6 +430,7 @@ public abstract class QueryRunnerTestBase extends
QueryTestSet {
case TIMESTAMP:
case STRING:
case BYTES:
+ case UUID:
return actual.equals(expected);
case INT_ARRAY:
return Arrays.equals((int[]) actual, (int[]) expected);
@@ -523,6 +527,9 @@ public abstract class QueryRunnerTestBase extends
QueryTestSet {
case BYTES:
h2Statement.setBytes(h2Index++, Hex.decodeHex((String) value));
break;
+ case UUID:
+ h2Statement.setObject(h2Index++, UuidUtils.toUUID(value));
+ break;
default:
h2Statement.setObject(h2Index++, value);
break;
@@ -566,6 +573,8 @@ public abstract class QueryRunnerTestBase extends
QueryTestSet {
return "BYTEA";
case TIMESTAMP:
return "TIMESTAMP";
+ case UUID:
+ return "UUID";
default:
throw new UnsupportedOperationException("Unsupported type conversion
to h2 type: " + dataType);
}
diff --git a/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
b/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
index a99ba06694f..ca555e963d8 100644
--- a/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
+++ b/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
@@ -10,16 +10,18 @@
{"name": "string_col", "type": "STRING"},
{"name": "bool_col", "type": "BOOLEAN"},
{"name": "decimal_col", "type": "BIG_DECIMAL"},
- {"name": "timestamp_col", "type": "TIMESTAMP"}
+ {"name": "timestamp_col", "type": "TIMESTAMP"},
+ {"name": "bytes_col", "type": "BYTES"},
+ {"name": "uuid_col", "type": "UUID"}
],
"inputs": [
- [2, 2, 300, 300, "a", true, "1E-307", "2020-01-01 03:10:12"],
- [2, 2, 400, 400, "a", true, "1E+307", "2020-01-01 03:39:12"],
- [3, 3, 100, 100, "b", false, "1E-308", "2020-01-01 08:32:12"],
- [100, 100, 1, 1, "b", false, "1E+308", "2020-02-06 03:32:12"],
- [101, 101, 1.01, 1.01, "c", false, "1E-317", "2020-05-01 03:32:12"],
- [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"]
+ [2, 2, 300, 300, "a", true, "1E-307", "2020-01-01 03:10:12",
"DEADBEEF", "00000000-0000-0000-0000-000000000001"],
+ [2, 2, 400, 400, "a", true, "1E+307", "2020-01-01 03:39:12",
"FDEADBEE", "00000000-0000-0000-0000-000000000002"],
+ [3, 3, 100, 100, "b", false, "1E-308", "2020-01-01 08:32:12",
"EFDEADBE", "00000000-0000-0000-0000-000000000003"],
+ [100, 100, 1, 1, "b", false, "1E+308", "2020-02-06 03:32:12",
"BEEFDEAD", "00000000-0000-0000-0000-000000000004"],
+ [101, 101, 1.01, 1.01, "c", false, "1E-317", "2020-05-01 03:32:12",
"ADBEEFDE", "00000000-0000-0000-0000-000000000005"],
+ [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"]
]
}
},
@@ -48,6 +50,64 @@
"sql": "SELECT /*+ aggOptions(is_skip_leaf_stage_aggregate='true') */
string_col, LASTWITHTIME(float_col, timestamp_col, 'FLOAT'),
LASTWITHTIME(double_col, timestamp_col, 'DOUBLE'), last_with_time(long_col,
CAST(long_col AS TIMESTAMP), 'LONG') FROM {tbl} GROUP BY string_col",
"outputs": [["a", 400, 400, 2], ["b", 1, 1, 100], ["c", 1.01, 1.01,
175]]
},
+ {
+ "comment": "The data type argument of FIRST_WITH_TIME / LAST_WITH_TIME
is inferred when omitted",
+ "sql": "SELECT FIRSTWITHTIME(int_col, timestamp_col),
FIRSTWITHTIME(string_col, timestamp_col), first_with_time(bool_col,
CAST(long_col AS TIMESTAMP)) FROM {tbl}",
+ "outputs": [[2, "a", true]]
+ },
+ {
+ "sql": "SELECT bool_col, FIRSTWITHTIME(float_col, timestamp_col),
FIRSTWITHTIME(double_col, timestamp_col), first_with_time(long_col,
CAST(long_col AS TIMESTAMP)) FROM {tbl} GROUP BY bool_col",
+ "outputs": [[false, 100, 100, 3], [true, 300, 300, 2]]
+ },
+ {
+ "sql": "SELECT /*+ aggOptions(is_skip_leaf_stage_aggregate='true') */
string_col, FIRSTWITHTIME(float_col, timestamp_col), FIRSTWITHTIME(double_col,
timestamp_col), first_with_time(long_col, CAST(long_col AS TIMESTAMP)) FROM
{tbl} GROUP BY string_col",
+ "outputs": [["a", 300, 300, 2], ["b", 100, 100, 3], ["c", 1.75, 1.75,
101]]
+ },
+ {
+ "sql": "SELECT LASTWITHTIME(int_col, timestamp_col),
LASTWITHTIME(string_col, timestamp_col), last_with_time(bool_col, CAST(long_col
AS TIMESTAMP)) FROM {tbl}",
+ "outputs": [[101, "c", true]]
+ },
+ {
+ "sql": "SELECT bool_col, LASTWITHTIME(float_col, timestamp_col),
LASTWITHTIME(double_col, timestamp_col), last_with_time(long_col, CAST(long_col
AS TIMESTAMP)) FROM {tbl} GROUP BY bool_col",
+ "outputs": [[false, 1.01, 1.01, 150], [true, 1.75, 1.75, 175]]
+ },
+ {
+ "sql": "SELECT /*+ aggOptions(is_skip_leaf_stage_aggregate='true') */
string_col, LASTWITHTIME(float_col, timestamp_col), LASTWITHTIME(double_col,
timestamp_col), last_with_time(long_col, CAST(long_col AS TIMESTAMP)) FROM
{tbl} GROUP BY string_col",
+ "outputs": [["a", 400, 400, 2], ["b", 1, 1, 100], ["c", 1.01, 1.01,
175]]
+ },
+ {
+ "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"]]
+ },
+ {
+ "sql": "SELECT FIRST_WITH_TIME(uuid_col, timestamp_col),
LAST_WITH_TIME(uuid_col, timestamp_col) FROM {tbl}",
+ "outputs": [["00000000-0000-0000-0000-000000000001",
"00000000-0000-0000-0000-000000000005"]]
+ },
+ {
+ "sql": "SELECT string_col, FIRST_WITH_TIME(uuid_col, timestamp_col),
LAST_WITH_TIME(uuid_col, timestamp_col) FROM {tbl} GROUP BY string_col",
+ "outputs": [["a", "00000000-0000-0000-0000-000000000001",
"00000000-0000-0000-0000-000000000002"], ["b",
"00000000-0000-0000-0000-000000000003",
"00000000-0000-0000-0000-000000000004"], ["c",
"00000000-0000-0000-0000-000000000007", "00000000-0000-0000-0000-000000000005"]]
+ },
+ {
+ "comment": "SMALLINT and TINYINT data arguments infer 'INT'",
+ "sql": "SELECT FIRST_WITH_TIME(CAST(int_col AS SMALLINT),
timestamp_col), LAST_WITH_TIME(CAST(int_col AS SMALLINT), timestamp_col),
LAST_WITH_TIME(CAST(int_col AS TINYINT), timestamp_col) FROM {tbl}",
+ "outputs": [[2, 101, 101]]
+ },
+ {
+ "sql": "SELECT string_col, FIRST_WITH_TIME(CAST(int_col AS SMALLINT),
timestamp_col), LAST_WITH_TIME(CAST(int_col AS SMALLINT), timestamp_col) FROM
{tbl} GROUP BY string_col",
+ "outputs": [["a", 2, 2], ["b", 3, 100], ["c", 175, 101]]
+ },
+ {
+ "sql": "SELECT 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, 'BYTES') FROM {tbl}",
+ "outputs": [["1E-307", "1E-317", "deadbeef", "adbeefde"]]
+ },
+ {
+ "sql": "SELECT 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"]]
+ },
+ {
+ "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"]]
+ },
{
"sql": "SELECT SUMPRECISION(decimal_col) FROM {tbl}",
"outputs":
[["10000000000100000000110000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000.00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
[...]
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/BigDecimalLongPair.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/BigDecimalLongPair.java
new file mode 100644
index 00000000000..95032c4fe86
--- /dev/null
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/BigDecimalLongPair.java
@@ -0,0 +1,52 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.customobject;
+
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import org.apache.pinot.spi.utils.BigDecimalUtils;
+
+
+/// A BIG_DECIMAL value with its time, for FIRST_WITH_TIME / LAST_WITH_TIME.
+public class BigDecimalLongPair extends ValueLongPair<BigDecimal> {
+
+ public BigDecimalLongPair(BigDecimal value, long time) {
+ super(value, time);
+ }
+
+ public static BigDecimalLongPair fromBytes(byte[] bytes) {
+ return fromByteBuffer(ByteBuffer.wrap(bytes));
+ }
+
+ public static BigDecimalLongPair fromByteBuffer(ByteBuffer byteBuffer) {
+ byte[] valueBytes = new byte[byteBuffer.getInt()];
+ byteBuffer.get(valueBytes);
+ return new BigDecimalLongPair(BigDecimalUtils.deserialize(valueBytes),
byteBuffer.getLong());
+ }
+
+ @Override
+ public byte[] toBytes() {
+ byte[] valueBytes = BigDecimalUtils.serialize(_value);
+ ByteBuffer byteBuffer = ByteBuffer.allocate(Integer.BYTES +
valueBytes.length + Long.BYTES);
+ byteBuffer.putInt(valueBytes.length);
+ byteBuffer.put(valueBytes);
+ byteBuffer.putLong(_time);
+ return byteBuffer.array();
+ }
+}
diff --git
a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/BytesLongPair.java
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/BytesLongPair.java
new file mode 100644
index 00000000000..517eaffdc54
--- /dev/null
+++
b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/customobject/BytesLongPair.java
@@ -0,0 +1,51 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.segment.local.customobject;
+
+import java.nio.ByteBuffer;
+import org.apache.pinot.spi.utils.ByteArray;
+
+
+/// A BYTES value with its time, for FIRST_WITH_TIME / LAST_WITH_TIME. UUID
values use the same pair.
+public class BytesLongPair extends ValueLongPair<ByteArray> {
+
+ public BytesLongPair(ByteArray value, long time) {
+ super(value, time);
+ }
+
+ public static BytesLongPair fromBytes(byte[] bytes) {
+ return fromByteBuffer(ByteBuffer.wrap(bytes));
+ }
+
+ public static BytesLongPair fromByteBuffer(ByteBuffer byteBuffer) {
+ byte[] valueBytes = new byte[byteBuffer.getInt()];
+ byteBuffer.get(valueBytes);
+ return new BytesLongPair(new ByteArray(valueBytes), byteBuffer.getLong());
+ }
+
+ @Override
+ public byte[] toBytes() {
+ byte[] valueBytes = _value.getBytes();
+ ByteBuffer byteBuffer = ByteBuffer.allocate(Integer.BYTES +
valueBytes.length + Long.BYTES);
+ byteBuffer.putInt(valueBytes.length);
+ byteBuffer.put(valueBytes);
+ byteBuffer.putLong(_time);
+ return byteBuffer.array();
+ }
+}
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 06d04efbb00..7534ec713f6 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
@@ -65,10 +65,12 @@ public enum AggregationFunctionType {
AVG("avg", SqlTypeName.OTHER, SqlTypeName.DOUBLE),
MODE("mode", SqlTypeName.OTHER, SqlTypeName.DOUBLE),
ANYVALUE("anyValue", ReturnTypes.ARG0, OperandTypes.ANY, SqlTypeName.OTHER),
- FIRSTWITHTIME("firstWithTime", ReturnTypes.ARG0,
- OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY,
SqlTypeFamily.CHARACTER), SqlTypeName.OTHER),
- LASTWITHTIME("lastWithTime", ReturnTypes.ARG0,
- OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY,
SqlTypeFamily.CHARACTER), SqlTypeName.OTHER),
+ FIRSTWITHTIME("firstWithTime", ReturnTypes.ARG0, OperandTypes.or(
+ OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY),
+ OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY,
SqlTypeFamily.CHARACTER)), SqlTypeName.OTHER),
+ LASTWITHTIME("lastWithTime", ReturnTypes.ARG0, OperandTypes.or(
+ OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY),
+ OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.ANY,
SqlTypeFamily.CHARACTER)), SqlTypeName.OTHER),
MINMAXRANGE("minMaxRange", ReturnTypes.DOUBLE, OperandTypes.ANY,
SqlTypeName.OTHER, SqlTypeName.DOUBLE),
/// for all distinct count family functions:
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
