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 b58776bca94 Return the input's logical type from ANY_VALUE (#19990)
b58776bca94 is described below
commit b58776bca94f1686b501025cd092c3104cc2cf33
Author: Xiang Fu <[email protected]>
AuthorDate: Mon Oct 26 20:46:21 2026 -0700
Return the input's logical type from ANY_VALUE (#19990)
Adds an opt-in aggregate rewrite that preserves ANY_VALUE's logical input
type in both query engines. The rewrite defaults off for rolling upgrades and
can be enabled through the inferAnyValueType query option or live cluster
config. Covers empty input, null handling, schema evolution and efficient
per-block reads.
---
.../broker/broker/helix/BaseBrokerStarter.java | 12 +-
.../requesthandler/BaseBrokerRequestHandler.java | 14 ++
.../BaseSingleStageBrokerRequestHandler.java | 4 +
.../requesthandler/InferAnyValueTypeProvider.java | 70 ++++++
.../InferAnyValueTypeProviderTest.java | 69 ++++++
.../common/utils/config/QueryOptionsUtils.java | 5 +
.../function/AnyValueAggregationFunction.java | 257 +++++++++++++--------
.../AggregateFunctionRewriteOptimizer.java | 39 ++--
.../function/AnyValueAggregationFunctionTest.java | 147 +++++++++++-
.../AggregateFunctionRewriteOptimizerTest.java | 48 ++++
.../apache/pinot/queries/AnyValueQueriesTest.java | 230 ++++++++++++++++++
.../PinotAggregateTypeArgumentRewriteRule.java | 76 +++++-
.../org/apache/pinot/query/QueryEnvironment.java | 20 +-
.../pinot/query/QueryPlannerRuleOptionsTest.java | 18 +-
.../src/test/resources/queries/AggregatePlans.json | 72 ++++++
.../src/test/resources/queries/UDFAggregates.json | 66 ++++++
.../pinot/segment/spi/AggregationFunctionType.java | 4 +-
.../apache/pinot/spi/utils/CommonConstants.java | 8 +
18 files changed, 1022 insertions(+), 137 deletions(-)
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
index 02e3634354e..cbfbd27d8ac 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/broker/helix/BaseBrokerStarter.java
@@ -231,11 +231,14 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
// Remove all white-spaces from the list of zkServers (if any).
_zkServers =
brokerConf.getProperty(Helix.CONFIG_OF_ZOOKEEPER_SERVER).replaceAll("\\s+", "");
_clusterName = brokerConf.getProperty(Helix.CONFIG_OF_CLUSTER_NAME);
- // Keep the component fallback separate from cluster values: the live
listener applies the cluster override and
+ // Keep the component fallback separate from cluster values: the live
listeners apply the cluster override and
// must restore this value when that override is removed, including when
it was already present at startup.
_brokerConf.setProperty(Broker.CONFIG_OF_BROKER_ENABLE_QUERY_FINGERPRINTING,
_brokerConf.getProperty(Broker.CONFIG_OF_BROKER_ENABLE_QUERY_FINGERPRINTING,
Broker.DEFAULT_BROKER_ENABLE_QUERY_FINGERPRINTING));
+ _brokerConf.setProperty(Broker.CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE,
+
_brokerConf.getProperty(Broker.CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE,
+ Broker.DEFAULT_BROKER_QUERY_INFER_ANY_VALUE_TYPE));
ServiceStartableUtils.applyClusterConfig(_brokerConf, _zkServers,
_clusterName, ServiceRole.BROKER);
applyCustomConfigs(brokerConf);
TableSamplerFactory.init(_brokerConf.subset(CommonConstants.Broker.TABLE_SAMPLER_CONFIG_PREFIX));
@@ -661,17 +664,22 @@ public abstract class BaseBrokerStarter implements
ServiceStartable {
_brokerRequestHandler =
new BrokerRequestHandlerDelegate(singleStageBrokerRequestHandler,
multiStageBrokerRequestHandler,
timeSeriesRequestHandler, _responseStore);
- // Lets the approximate-function rewrite defaults be changed from the
cluster config without a broker restart.
+ // Lets the approximate-function rewrite and inferAnyValueType defaults be
changed from the cluster config without a
+ // broker restart.
// Registering here is a no-op in itself, because the cluster config
handler is not wired to Helix yet, so the
// snapshot it hands the listener is empty. The real values arrive from
that handler's first Helix callback,
// which is set up below and still runs before the broker starts serving
traffic.
_clusterConfigChangeHandler.registerClusterConfigChangeListener(
singleStageBrokerRequestHandler.getApproximateFunctionOverrideProvider());
_clusterConfigChangeHandler.registerClusterConfigChangeListener(singleStageBrokerRequestHandler);
+ _clusterConfigChangeHandler.registerClusterConfigChangeListener(
+ singleStageBrokerRequestHandler.getInferAnyValueTypeProvider());
if (multiStageBrokerRequestHandler != null) {
_clusterConfigChangeHandler.registerClusterConfigChangeListener(
multiStageBrokerRequestHandler.getApproximateFunctionOverrideProvider());
_clusterConfigChangeHandler.registerClusterConfigChangeListener(multiStageBrokerRequestHandler);
+ _clusterConfigChangeHandler.registerClusterConfigChangeListener(
+ multiStageBrokerRequestHandler.getInferAnyValueTypeProvider());
}
_brokerRequestHandler.start();
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java
index 3ba4405c37f..8f4b8520913 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java
@@ -110,6 +110,9 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler,
/// Resolves the approximate-function rewrite defaults, live from the Helix
cluster config. Registered as a
/// listener by the broker starter through
[#getApproximateFunctionOverrideProvider()].
protected final ApproximateFunctionOverrideProvider
_approximateFunctionOverrideProvider;
+ /// Resolves the default of the `inferAnyValueType` query option, live from
the Helix cluster config. Registered as a
+ /// listener by the broker starter through [#getInferAnyValueTypeProvider()].
+ protected final InferAnyValueTypeProvider _inferAnyValueTypeProvider;
public BaseBrokerRequestHandler(PinotConfiguration config, String brokerId,
BrokerRequestIdGenerator requestIdGenerator, RoutingManager
routingManager,
@@ -125,6 +128,7 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler,
_threadAccountant = threadAccountant;
_multiClusterRoutingContext = multiClusterRoutingContext;
_approximateFunctionOverrideProvider = new
ApproximateFunctionOverrideProvider(config);
+ _inferAnyValueTypeProvider = new InferAnyValueTypeProvider(config);
_brokerMetrics = BrokerMetrics.get();
_brokerQueryEventListener =
BrokerQueryEventListenerFactory.getBrokerQueryEventListener();
_trackedHeaders = BrokerQueryEventListenerFactory.getTrackedHeaders();
@@ -156,6 +160,12 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler,
return _approximateFunctionOverrideProvider;
}
+ /// Returns the provider that resolves the default of the
`inferAnyValueType` query option. The broker starter
+ /// registers it with the cluster config change handler so that the default
reloads without a restart.
+ public InferAnyValueTypeProvider getInferAnyValueTypeProvider() {
+ return _inferAnyValueTypeProvider;
+ }
+
/// Applies the live cluster override. Removing it restores the broker
configuration (enabled by default).
/// Literal-value SQL redaction always requires fingerprinting, including
when the override is `false`.
@Override
@@ -253,6 +263,10 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler,
.putIfAbsent(QueryOptionKey.AUTO_REWRITE_AGGREGATION_TYPE,
_enableAutoRewriteAggregationType);
}
+ if (_inferAnyValueTypeProvider.isEnabled()) {
+
sqlNodeAndOptions.getOptions().putIfAbsent(QueryOptionKey.INFER_ANY_VALUE_TYPE,
"true");
+ }
+
if (_regexDictSizeThreshold != null) {
sqlNodeAndOptions.getOptions().putIfAbsent(QueryOptionKey.REGEX_DICT_SIZE_THRESHOLD,
_regexDictSizeThreshold);
}
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
index c0aada9c719..07156bccaaf 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseSingleStageBrokerRequestHandler.java
@@ -1651,6 +1651,10 @@ public abstract class
BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ
.putIfAbsent(QueryOptionKey.AUTO_REWRITE_AGGREGATION_TYPE,
_enableAutoRewriteAggregationType);
}
+ if (_inferAnyValueTypeProvider.isEnabled()) {
+
sqlNodeAndOptions.getOptions().putIfAbsent(QueryOptionKey.INFER_ANY_VALUE_TYPE,
"true");
+ }
+
if (_regexDictSizeThreshold != null) {
sqlNodeAndOptions.getOptions().putIfAbsent(QueryOptionKey.REGEX_DICT_SIZE_THRESHOLD,
_regexDictSizeThreshold);
}
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/InferAnyValueTypeProvider.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/InferAnyValueTypeProvider.java
new file mode 100644
index 00000000000..bd680bb37de
--- /dev/null
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/InferAnyValueTypeProvider.java
@@ -0,0 +1,70 @@
+/**
+ * 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.broker.requesthandler;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.spi.config.provider.PinotClusterConfigChangeListener;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/// Resolves the default of the `inferAnyValueType` query option from
+/// [Broker#CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE] in the broker conf
and in the Helix cluster config. The cluster
+/// config wins where it sets the key, and it is applied live: registered as a
[PinotClusterConfigChangeListener], a
+/// change takes effect on the next query without a broker restart.
+///
+/// Thread-safe.
+public class InferAnyValueTypeProvider implements
PinotClusterConfigChangeListener {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(InferAnyValueTypeProvider.class);
+
+ private final boolean _brokerConfEnabled;
+ private volatile boolean _enabled;
+
+ public InferAnyValueTypeProvider(PinotConfiguration config) {
+ _brokerConfEnabled =
config.getProperty(Broker.CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE,
+ Broker.DEFAULT_BROKER_QUERY_INFER_ANY_VALUE_TYPE);
+ _enabled = _brokerConfEnabled;
+ }
+
+ @Override
+ public void onChange(Set<String> changedConfigs, Map<String, String>
clusterConfigs) {
+ if
(!changedConfigs.contains(Broker.CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE)) {
+ return;
+ }
+ // A removed or invalid value falls back to the broker conf, so that a
broker that restarts resolves the same.
+ String value =
clusterConfigs.get(Broker.CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE);
+ boolean enabled = _brokerConfEnabled;
+ // Accept only true/false so a typo restores the fallback rather than
silently disabling the rewrite.
+ if (value != null && ("true".equalsIgnoreCase(value.trim()) ||
"false".equalsIgnoreCase(value.trim()))) {
+ enabled = Boolean.parseBoolean(value.trim());
+ } else if (value != null) {
+ LOGGER.error("Ignoring invalid value '{}' for {}, falling back to '{}'",
value,
+ Broker.CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE,
_brokerConfEnabled);
+ }
+ _enabled = enabled;
+ LOGGER.info("Updated {} to {}",
Broker.CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE, enabled);
+ }
+
+ public boolean isEnabled() {
+ return _enabled;
+ }
+}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/InferAnyValueTypeProviderTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/InferAnyValueTypeProviderTest.java
new file mode 100644
index 00000000000..7945691339d
--- /dev/null
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/InferAnyValueTypeProviderTest.java
@@ -0,0 +1,69 @@
+/**
+ * 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.broker.requesthandler;
+
+import java.util.Map;
+import java.util.Set;
+import org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+public class InferAnyValueTypeProviderTest {
+ private static final String KEY =
Broker.CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE;
+
+ @Test
+ public void testBrokerConfProvidesTheDefault() {
+ assertFalse(newProvider(Map.of()).isEnabled());
+ assertTrue(newProvider(Map.of(KEY, "true")).isEnabled());
+ }
+
+ @Test
+ public void testClusterConfigWinsOverBrokerConfAndReloads() {
+ InferAnyValueTypeProvider provider = newProvider(Map.of());
+ provider.onChange(Set.of(KEY), Map.of(KEY, "true"));
+ assertTrue(provider.isEnabled());
+
+ // Another key changing leaves the value as it is.
+ provider.onChange(Set.of("pinot.broker.other"), Map.of());
+ assertTrue(provider.isEnabled());
+
+ // Removing the key from the cluster config falls back to the broker conf.
+ provider.onChange(Set.of(KEY), Map.of());
+ assertFalse(provider.isEnabled());
+
+ provider = newProvider(Map.of(KEY, "true"));
+ provider.onChange(Set.of(KEY), Map.of(KEY, " FALSE "));
+ assertFalse(provider.isEnabled());
+ }
+
+ @Test
+ public void testInvalidClusterConfigFallsBackToBrokerConf() {
+ InferAnyValueTypeProvider provider = newProvider(Map.of(KEY, "true"));
+ provider.onChange(Set.of(KEY), Map.of(KEY, "flase"));
+ assertTrue(provider.isEnabled());
+ }
+
+ private static InferAnyValueTypeProvider newProvider(Map<String, Object>
properties) {
+ return new InferAnyValueTypeProvider(new PinotConfiguration(properties));
+ }
+}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
index 936c968aaf0..d02fe34c761 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/utils/config/QueryOptionsUtils.java
@@ -394,6 +394,11 @@ public class QueryOptionsUtils {
return value == null || Boolean.parseBoolean(value);
}
+ /// Returns whether ANY_VALUE(expr) is rewritten to take the data type of
`expr`. Defaults to `false`.
+ public static boolean isInferAnyValueType(@Nullable Map<String, String>
queryOptions) {
+ return queryOptions != null &&
Boolean.parseBoolean(queryOptions.get(QueryOptionKey.INFER_ANY_VALUE_TYPE));
+ }
+
public static boolean isSkipUpsertView(Map<String, String> queryOptions) {
return
Boolean.parseBoolean(queryOptions.get(QueryOptionKey.SKIP_UPSERT_VIEW));
}
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
index c984363d2c5..34747dc3eb6 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunction.java
@@ -18,12 +18,16 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Suppliers;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
+import java.util.function.IntFunction;
+import java.util.function.Supplier;
import javax.annotation.Nullable;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
@@ -37,6 +41,7 @@ import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.exception.BadQueryRequestException;
import org.apache.pinot.spi.utils.ByteArray;
@@ -47,6 +52,10 @@ import org.apache.pinot.spi.utils.ByteArray;
/// only until it finds the first non-null value in the current batch for each
group/key. This makes it O(n) over the
/// input once per group until the first value is set, with early-exit fast
paths when there are no nulls.
///
+/// `ANY_VALUE(expr, 'dataType')` fixes the result type at construction and
reads every block as that type. Without
+/// the data type argument, the result type is the stored type of the first
block, or STRING before any block.
+/// The single-stage engine accepts convertible explicit types; the
multi-stage engine requires the inferred type.
+///
/// **Example:**
///
/// ```
@@ -58,12 +67,52 @@ import org.apache.pinot.spi.utils.ByteArray;
/// ```
public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunction<Object, Comparable<?>> {
private static final DataType[] DATA_TYPE_VALUES = DataType.values();
- // Result type is determined at runtime based on input expression type
+ // Data type argument, or null when the result type is determined at runtime
from the input blocks
+ @Nullable
+ private final DataType _dataType;
private ColumnDataType _resultType;
public AnyValueAggregationFunction(List<ExpressionContext> arguments,
boolean nullHandlingEnabled) {
- super(verifySingleArgument(arguments, "ANY_VALUE"), nullHandlingEnabled);
- _resultType = null;
+ super(getValueArgument(arguments), nullHandlingEnabled);
+ if (arguments.size() == 2) {
+ ExpressionContext dataTypeArgument = arguments.get(1);
+ Preconditions.checkArgument(dataTypeArgument.getType() ==
ExpressionContext.Type.LITERAL,
+ "ANY_VALUE expects the 2nd argument to be literal, got: %s",
dataTypeArgument.getType());
+ String dataType = dataTypeArgument.getLiteral().getStringValue();
+ try {
+ _dataType = DataType.valueOf(dataType.toUpperCase());
+ } catch (IllegalArgumentException e) {
+ throw new BadQueryRequestException("Unsupported data type for
ANY_VALUE: " + dataType, e);
+ }
+ _resultType = getResultType(_dataType);
+ } else {
+ _dataType = null;
+ }
+ }
+
+ private static ExpressionContext getValueArgument(List<ExpressionContext>
arguments) {
+ int numArguments = arguments.size();
+ Preconditions.checkArgument(numArguments == 1 || numArguments == 2,
"ANY_VALUE expects 1 or 2 arguments, got: %s",
+ numArguments);
+ return arguments.get(0);
+ }
+
+ private static ColumnDataType getResultType(DataType dataType) {
+ switch (dataType) {
+ case INT:
+ case LONG:
+ case FLOAT:
+ case DOUBLE:
+ case BIG_DECIMAL:
+ case BOOLEAN:
+ case TIMESTAMP:
+ case STRING:
+ case BYTES:
+ case UUID:
+ return ColumnDataType.fromDataTypeSV(dataType);
+ default:
+ throw new IllegalArgumentException("Unsupported data type for
ANY_VALUE: " + dataType);
+ }
}
@Override
@@ -71,6 +120,14 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
return AggregationFunctionType.ANYVALUE;
}
+ @Override
+ public String getResultColumnName() {
+ if (_dataType == null) {
+ return super.getResultColumnName();
+ }
+ return getType().getName().toLowerCase() + "(" + _expression + ",'" +
_dataType + "')";
+ }
+
@Override
public AggregationResultHolder createAggregationResultHolder() {
return new ObjectAggregationResultHolder();
@@ -85,7 +142,7 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
public ColumnDataType getIntermediateResultColumnType() {
// Default to STRING if result type is not yet determined
// TODO: See if UNKNOWN can be used instead
- return _resultType != null ? _resultType : ColumnDataType.STRING;
+ return _resultType != null ? _resultType.getStoredType() :
ColumnDataType.STRING;
}
@Override
@@ -124,7 +181,14 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
}
BlockValSet bvs = blockValSetMap.get(_expression);
ensureResultType(bvs);
- aggregateHelper(length, bvs, (i, value) -> {
+ aggregateHelper(length, bvs, (i, values) -> {
+ if (holder.getResult() != null) {
+ return true;
+ }
+ Object value = values.apply(i);
+ if (value == null) {
+ return false;
+ }
holder.setValue(value);
return true; // Stop after first value found
});
@@ -135,10 +199,13 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
Map<ExpressionContext, BlockValSet> map) {
BlockValSet bvs = map.get(_expression);
ensureResultType(bvs);
- aggregateHelper(length, bvs, (i, value) -> {
+ aggregateHelper(length, bvs, (i, values) -> {
int g = groupKeys[i];
if (holder.getResult(g) == null) {
- holder.setValueForKey(g, value);
+ Object value = values.apply(i);
+ if (value != null) {
+ holder.setValueForKey(g, value);
+ }
}
return false; // Continue processing for other groups
});
@@ -149,10 +216,16 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
Map<ExpressionContext, BlockValSet> map) {
BlockValSet bvs = map.get(_expression);
ensureResultType(bvs);
- aggregateHelper(length, bvs, (i, value) -> {
- int[] keys = groupKeysArray[i];
- for (int g : keys) {
+ aggregateHelper(length, bvs, (i, values) -> {
+ Object value = null;
+ for (int g : groupKeysArray[i]) {
if (holder.getResult(g) == null) {
+ if (value == null) {
+ value = values.apply(i);
+ if (value == null) {
+ break;
+ }
+ }
holder.setValueForKey(g, value);
}
}
@@ -178,78 +251,85 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
}
@FunctionalInterface
- private interface ValueProcessor<T> {
- boolean process(int index, T value); // Returns true to stop processing,
false to continue
+ private interface ValueProcessor {
+ /// Returns true to stop processing, false to continue. `values` reads the
value of a row, so that a row is read
+ /// only when it is needed.
+ boolean process(int index, IntFunction<Object> values);
}
/// Generic helper for processing values with dictionary optimization for
all supported data types
- private void aggregateHelper(int length, BlockValSet bvs,
ValueProcessor<Object> processor) {
- // Use dictionary-based access for efficiency when available
- if (bvs.isDictionaryEncoded()) {
- final int[] dictIds = bvs.getDictionaryIdsSV();
- final Dictionary dict = bvs.getDictionary();
- forEachNotNull(length, bvs, (from, to) -> {
- for (int i = from; i < to; i++) {
- Object value = getDictionaryValue(dict, dictIds[i],
bvs.getValueType().getStoredType());
- if (processor.process(i, value)) {
- break;
- }
- }
- });
- } else {
- // Fall back to direct value access based on type
- forEachNotNull(length, bvs, (from, to) -> {
- for (int i = from; i < to; i++) {
- Object value = getDirectValue(bvs, i);
- if (value != null && processor.process(i, value)) {
- break;
- }
+ private void aggregateHelper(int length, BlockValSet bvs, ValueProcessor
processor) {
+ // Reading through the bound type keeps values consistent when segments
store the column as different types
+ DataType storedType = _dataType != null ? _dataType.getStoredType() :
bvs.getValueType().getStoredType();
+ // Created for the first non-null range, so that a block whose rows are
all null never evaluates the expression
+ Supplier<IntFunction<Object>> reader = Suppliers.memoize(() ->
getValueReader(bvs, storedType));
+ forEachNotNull(length, bvs, (from, to) -> {
+ IntFunction<Object> values = reader.get();
+ for (int i = from; i < to; i++) {
+ if (processor.process(i, values)) {
+ break;
}
- });
- }
+ }
+ });
}
- /// Get value from dictionary based on data type
- private Object getDictionaryValue(Dictionary dict, int dictId, DataType
storedType) {
- switch (storedType) {
- case INT:
- return dict.getIntValue(dictId);
- case LONG:
- return dict.getLongValue(dictId);
- case FLOAT:
- return dict.getFloatValue(dictId);
- case DOUBLE:
- return dict.getDoubleValue(dictId);
- case STRING:
- return dict.getStringValue(dictId);
- case BIG_DECIMAL:
- return dict.getBigDecimalValue(dictId);
- case BYTES:
- return dict.getByteArrayValue(dictId);
- default:
- throw new IllegalStateException("Unsupported dictionary type: " +
storedType);
+ /// Returns a reader of the values in the block. The values are fetched once
per block, because some block value sets
+ /// build the value array on each call.
+ private static IntFunction<Object> getValueReader(BlockValSet bvs, DataType
storedType) {
+ // Use dictionary-based access for efficiency when available
+ if (bvs.isDictionaryEncoded()) {
+ int[] dictIds = bvs.getDictionaryIdsSV();
+ Dictionary dictionary = bvs.getDictionary();
+ switch (storedType) {
+ case INT:
+ return i -> dictionary.getIntValue(dictIds[i]);
+ case LONG:
+ return i -> dictionary.getLongValue(dictIds[i]);
+ case FLOAT:
+ return i -> dictionary.getFloatValue(dictIds[i]);
+ case DOUBLE:
+ return i -> dictionary.getDoubleValue(dictIds[i]);
+ case STRING:
+ return i -> dictionary.getStringValue(dictIds[i]);
+ case BIG_DECIMAL:
+ return i -> dictionary.getBigDecimalValue(dictIds[i]);
+ case BYTES:
+ return i -> dictionary.getByteArrayValue(dictIds[i]);
+ default:
+ throw new IllegalStateException("Unsupported dictionary type: " +
storedType);
+ }
}
- }
-
- /// Get value directly from BlockValSet based on data type
- private Object getDirectValue(BlockValSet bvs, int index) {
- switch (bvs.getValueType().getStoredType()) {
- case INT:
- return bvs.getIntValuesSV()[index];
- case LONG:
- return bvs.getLongValuesSV()[index];
- case FLOAT:
- return bvs.getFloatValuesSV()[index];
- case DOUBLE:
- return bvs.getDoubleValuesSV()[index];
- case STRING:
- return bvs.getStringValuesSV()[index];
- case BIG_DECIMAL:
- return bvs.getBigDecimalValuesSV()[index];
- case BYTES:
- return new ByteArray(bvs.getBytesValuesSV()[index]);
+ switch (storedType) {
+ case INT: {
+ int[] values = bvs.getIntValuesSV();
+ return i -> values[i];
+ }
+ case LONG: {
+ long[] values = bvs.getLongValuesSV();
+ return i -> values[i];
+ }
+ case FLOAT: {
+ float[] values = bvs.getFloatValuesSV();
+ return i -> values[i];
+ }
+ case DOUBLE: {
+ double[] values = bvs.getDoubleValuesSV();
+ return i -> values[i];
+ }
+ case STRING: {
+ String[] values = bvs.getStringValuesSV();
+ return i -> values[i];
+ }
+ case BIG_DECIMAL: {
+ BigDecimal[] values = bvs.getBigDecimalValuesSV();
+ return i -> values[i];
+ }
+ case BYTES: {
+ byte[][] values = bvs.getBytesValuesSV();
+ return i -> new ByteArray(values[i]);
+ }
default:
- throw new IllegalStateException("Unsupported direct access type: " +
bvs.getValueType().getStoredType());
+ throw new IllegalStateException("Unsupported direct access type: " +
storedType);
}
}
@@ -326,33 +406,8 @@ public class AnyValueAggregationFunction extends
BaseSingleInputAggregationFunct
}
private void ensureResultType(BlockValSet bvs) {
- if (_resultType != null) {
- return;
- }
- switch (bvs.getValueType().getStoredType()) {
- case INT:
- _resultType = ColumnDataType.INT;
- return;
- case LONG:
- _resultType = ColumnDataType.LONG;
- return;
- case FLOAT:
- _resultType = ColumnDataType.FLOAT;
- return;
- case DOUBLE:
- _resultType = ColumnDataType.DOUBLE;
- return;
- case STRING:
- _resultType = ColumnDataType.STRING;
- return;
- case BIG_DECIMAL:
- _resultType = ColumnDataType.BIG_DECIMAL;
- return;
- case BYTES:
- _resultType = ColumnDataType.BYTES;
- return;
- default:
- throw new IllegalStateException("ANY_VALUE unsupported type: " +
bvs.getValueType());
+ if (_resultType == null) {
+ _resultType = getResultType(bvs.getValueType().getStoredType());
}
}
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 518fefc9881..398e5d53b9b 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,6 +23,7 @@ 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.config.QueryOptionsUtils;
import org.apache.pinot.common.utils.request.RequestUtils;
import org.apache.pinot.segment.spi.AggregationFunctionType;
import org.apache.pinot.spi.data.FieldSpec;
@@ -38,6 +39,9 @@ import org.apache.pinot.spi.utils.CommonConstants;
/// - LAST_WITH_TIME(dataColumn, timeColumn) -> LAST_WITH_TIME(dataColumn,
timeColumn, 'dataType')
/// - MODE(stringColumn[, 'MIN'|'MAX']) -> MODE(stringColumn, 'MIN'|'MAX',
'STRING')
///
+/// Applied only when enabled through the `inferAnyValueType` query option:
+/// - ANY_VALUE(column) -> ANY_VALUE(column, 'dataType')
+///
/// Applied only when enabled through the `autoRewriteAggregationType` query
option:
/// - MIN(stringType) -> MINSTRING
/// - MAX(stringType) -> MAXSTRING
@@ -57,22 +61,25 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
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);
+ boolean inferAnyValueType =
QueryOptionsUtils.isInferAnyValueType(pinotQuery.getQueryOptions());
+ rewrite(pinotQuery.getSelectList(), schema, rewriteToTypedVariants,
inferAnyValueType);
+ rewrite(pinotQuery.getGroupByList(), schema, rewriteToTypedVariants,
inferAnyValueType);
+ rewrite(pinotQuery.getOrderByList(), schema, rewriteToTypedVariants,
inferAnyValueType);
+ rewrite(pinotQuery.getFilterExpression(), schema, rewriteToTypedVariants,
inferAnyValueType);
+ rewrite(pinotQuery.getHavingExpression(), schema, rewriteToTypedVariants,
inferAnyValueType);
}
- private static void rewrite(@Nullable List<Expression> expressions, Schema
schema, boolean rewriteToTypedVariants) {
+ private static void rewrite(@Nullable List<Expression> expressions, Schema
schema, boolean rewriteToTypedVariants,
+ boolean inferAnyValueType) {
if (expressions != null) {
for (Expression expression : expressions) {
- rewrite(expression, schema, rewriteToTypedVariants);
+ rewrite(expression, schema, rewriteToTypedVariants, inferAnyValueType);
}
}
}
- private static void rewrite(@Nullable Expression expression, Schema schema,
boolean rewriteToTypedVariants) {
+ private static void rewrite(@Nullable Expression expression, Schema schema,
boolean rewriteToTypedVariants,
+ boolean inferAnyValueType) {
if (expression == null || !expression.isSetFunctionCall()) {
return;
}
@@ -80,7 +87,7 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
Function function = expression.getFunctionCall();
String functionName = function.getOperator();
if (!AggregationFunctionType.isAggregationFunction(functionName)) {
- rewrite(function.getOperands(), schema, rewriteToTypedVariants);
+ rewrite(function.getOperands(), schema, rewriteToTypedVariants,
inferAnyValueType);
return;
}
@@ -96,7 +103,13 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
String normalizedName =
AggregationFunctionType.getNormalizedAggregationFunctionName(functionName);
if ((normalizedName.equals(AggregationFunctionType.FIRSTWITHTIME.name())
|| normalizedName.equals(AggregationFunctionType.LASTWITHTIME.name()))
&& function.getOperandsSize() == 2) {
- addDataTypeArgument(function, normalizedName, fieldSpec);
+ addDataTypeArgument(function, normalizedName + "(dataColumn,
timeColumn)", fieldSpec);
+ return;
+ }
+ if (normalizedName.equals(AggregationFunctionType.ANYVALUE.name()) &&
function.getOperandsSize() == 1) {
+ if (inferAnyValueType) {
+ addDataTypeArgument(function, normalizedName + "(expr)", fieldSpec);
+ }
return;
}
if (normalizedName.equals(AggregationFunctionType.MODE.name()) &&
function.getOperandsSize() <= 2) {
@@ -112,7 +125,7 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
/// 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) {
+ private static void addDataTypeArgument(Function function, String call,
FieldSpec fieldSpec) {
if (fieldSpec.isSingleValueField()) {
switch (fieldSpec.getDataType()) {
case INT:
@@ -135,8 +148,8 @@ public class AggregateFunctionRewriteOptimizer implements
StatementOptimizer {
break;
}
}
- throw new BadQueryRequestException(String.format("Unsupported data type
for %s(dataColumn, timeColumn): %s%s",
- functionName, fieldSpec.isSingleValueField() ? "" : "multi-value ",
fieldSpec.getDataType()));
+ throw new BadQueryRequestException(String.format("Unsupported data type
for %s: %s%s", call,
+ fieldSpec.isSingleValueField() ? "" : "multi-value ",
fieldSpec.getDataType()));
}
/// Appends the default tie reducer when missing, then STRING for a STRING
or JSON column.
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
index 8ae6327b6c6..dc5dc0e504b 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/AnyValueAggregationFunctionTest.java
@@ -18,32 +18,41 @@
*/
package org.apache.pinot.core.query.aggregation.function;
+import java.math.BigDecimal;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.Map;
import org.apache.pinot.common.CustomObject;
import org.apache.pinot.common.request.context.ExpressionContext;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.core.common.BlockValSet;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
import org.apache.pinot.core.query.aggregation.groupby.GroupByResultHolder;
import org.apache.pinot.queries.FluentQueryTest;
import org.apache.pinot.segment.spi.index.reader.Dictionary;
import org.apache.pinot.spi.data.FieldSpec.DataType;
+import org.apache.pinot.spi.exception.BadQueryRequestException;
import org.apache.pinot.spi.utils.ByteArray;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertThrows;
public class AnyValueAggregationFunctionTest extends
AbstractAggregationFunctionTest {
- @DataProvider(name = "bytesValueSources")
- Object[][] bytesValueSources() {
+ private static final ExpressionContext EXPRESSION =
ExpressionContext.forIdentifier("myField");
+
+ @DataProvider(name = "dictionaryEncoded")
+ Object[][] dictionaryEncoded() {
return new Object[][]{{false}, {true}};
}
- @Test(dataProvider = "bytesValueSources")
+ @Test(dataProvider = "dictionaryEncoded")
void testBytesIntermediateResultRoundTrip(boolean dictionaryEncoded) {
ExpressionContext expression = ExpressionContext.forIdentifier("myField");
AnyValueAggregationFunction function = new
AnyValueAggregationFunction(List.of(expression), true);
@@ -240,6 +249,23 @@ public class AnyValueAggregationFunctionTest extends
AbstractAggregationFunction
validateAnyValueBehavior(result, true); // Should return non-null
(unique_value exists)
}
+ @Test
+ void testAllNullSegmentDoesNotEvaluateExpression() {
+ // Casting the null placeholder of a STRING column to INT throws, so a
segment whose rows are all null must not
+ // evaluate the expression.
+ new DataTypeScenario(DataType.STRING).getDeclaringTable(true)
+ .onFirstInstance("myField", "null", "null")
+ .andSegment("myField", "5", "5")
+ .whenQuery("select anyValue(cast(myField as int)) from testTable")
+ .thenResultIs("STRING", "5")
+ .whenQuery("select 'k', anyValue(cast(myField as int)) from testTable
group by 'k'")
+ .thenResultIs("STRING | STRING", "k | 5")
+ .whenQuery("select anyValue(cast(myField as int), 'INT') from
testTable")
+ .thenResultIs("INT", "5")
+ .whenQuery("select 'k', anyValue(cast(myField as int), 'INT') from
testTable group by 'k'")
+ .thenResultIs("STRING | INT", "k | 5");
+ }
+
// GROUP BY tests with different data types
@Test
void testGroupByWithMultipleGroups() {
@@ -357,4 +383,119 @@ public class AnyValueAggregationFunctionTest extends
AbstractAggregationFunction
queryResult.thenResultIs(EXPECTED_COLUMN_TYPES, EXPECTED_NULL_RESULT);
}
}
+
+ @DataProvider(name = "dataTypes")
+ Object[][] dataTypes() {
+ return new Object[][]{
+ {"INT", ColumnDataType.INT, ColumnDataType.INT},
+ {"LONG", ColumnDataType.LONG, ColumnDataType.LONG},
+ {"FLOAT", ColumnDataType.FLOAT, ColumnDataType.FLOAT},
+ {"DOUBLE", ColumnDataType.DOUBLE, ColumnDataType.DOUBLE},
+ {"BIG_DECIMAL", ColumnDataType.BIG_DECIMAL,
ColumnDataType.BIG_DECIMAL},
+ {"BOOLEAN", ColumnDataType.INT, ColumnDataType.BOOLEAN},
+ {"TIMESTAMP", ColumnDataType.LONG, ColumnDataType.TIMESTAMP},
+ {"STRING", ColumnDataType.STRING, ColumnDataType.STRING},
+ {"BYTES", ColumnDataType.BYTES, ColumnDataType.BYTES},
+ {"uuid", ColumnDataType.BYTES, ColumnDataType.UUID}
+ };
+ }
+
+ @Test(dataProvider = "dataTypes")
+ void testDataTypeArgumentFixesResultTypes(String dataType, ColumnDataType
intermediateResultType,
+ ColumnDataType finalResultType) {
+ // The types are known before any input, so an empty result is typed as
well.
+ AnyValueAggregationFunction function = createFunction(dataType);
+ assertEquals(function.getIntermediateResultColumnType(),
intermediateResultType);
+ assertEquals(function.getFinalResultColumnType(), finalResultType);
+ assertEquals(function.getResultColumnName(), "anyvalue(myField,'" +
dataType.toUpperCase() + "')");
+ assertEquals(function.getInputExpressions(), List.of(EXPRESSION));
+
assertNull(function.extractAggregationResult(function.createAggregationResultHolder()));
+ }
+
+ @Test
+ void testDataTypeArgumentRejectsInvalidArguments() {
+ assertThrows(IllegalArgumentException.class, () -> createFunction("JSON"));
+ assertThrows(BadQueryRequestException.class, () ->
createFunction("NOT_A_TYPE"));
+ assertThrows(IllegalArgumentException.class,
+ () -> new AnyValueAggregationFunction(List.of(EXPRESSION,
ExpressionContext.forIdentifier("INT")), false));
+ assertThrows(IllegalArgumentException.class, () -> new
AnyValueAggregationFunction(
+ List.of(EXPRESSION, ExpressionContext.forLiteral(DataType.STRING,
"INT"), EXPRESSION), false));
+ }
+
+ @Test(dataProvider = "dictionaryEncoded")
+ void testDataTypeArgumentReadsBoundStoredType(boolean dictionaryEncoded) {
+ // A segment that still stores the column as INT after the schema widened
it to LONG
+ AnyValueAggregationFunction function = createFunction("LONG");
+ BlockValSet valueSet = mock(BlockValSet.class);
+ when(valueSet.getValueType()).thenReturn(DataType.INT);
+ if (dictionaryEncoded) {
+ Dictionary dictionary = mock(Dictionary.class);
+ when(dictionary.getLongValue(0)).thenReturn(5L);
+ when(valueSet.isDictionaryEncoded()).thenReturn(true);
+ when(valueSet.getDictionary()).thenReturn(dictionary);
+ when(valueSet.getDictionaryIdsSV()).thenReturn(new int[]{0});
+ } else {
+ when(valueSet.getLongValuesSV()).thenReturn(new long[]{5L});
+ }
+ Map<ExpressionContext, BlockValSet> values = Map.of(EXPRESSION, valueSet);
+
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(1, holder, values);
+ assertEquals(function.extractAggregationResult(holder), 5L);
+ GroupByResultHolder groupHolder = function.createGroupByResultHolder(1, 1);
+ function.aggregateGroupBySV(1, new int[]{0}, groupHolder, values);
+ assertEquals(function.extractGroupByResult(groupHolder, 0), 5L);
+ assertEquals(function.getIntermediateResultColumnType(),
ColumnDataType.LONG);
+ }
+
+ @Test
+ void testGroupByReadsValuesOncePerBlock() {
+ AnyValueAggregationFunction function = createFunction("LONG");
+ BlockValSet valueSet = mock(BlockValSet.class);
+ when(valueSet.getValueType()).thenReturn(DataType.LONG);
+ when(valueSet.getLongValuesSV()).thenReturn(new long[]{1L, 2L, 3L});
+ GroupByResultHolder holder = function.createGroupByResultHolder(2, 2);
+ function.aggregateGroupBySV(3, new int[]{0, 1, 0}, holder,
Map.of(EXPRESSION, valueSet));
+ assertEquals(function.extractGroupByResult(holder, 0), 1L);
+ assertEquals(function.extractGroupByResult(holder, 1), 2L);
+ // Some block value sets build the value array on each call
+ verify(valueSet, times(1)).getLongValuesSV();
+ }
+
+ @Test
+ void testDataTypeArgumentSerializedMerge() {
+ // One server found a value and the other did not.
+ AnyValueAggregationFunction function = createFunction("BOOLEAN");
+ BlockValSet valueSet = mock(BlockValSet.class);
+ when(valueSet.getValueType()).thenReturn(DataType.BOOLEAN);
+ when(valueSet.getIntValuesSV()).thenReturn(new int[]{1});
+ AggregationResultHolder holder = function.createAggregationResultHolder();
+ function.aggregate(1, holder, Map.of(EXPRESSION, valueSet));
+ CustomObject found = serialize(function,
function.extractAggregationResult(holder));
+ CustomObject empty =
+ serialize(function,
function.extractAggregationResult(function.createAggregationResultHolder()));
+
+ Object merged = AggregationFunctionUtils.mergeSerialized(function, null,
empty);
+ assertNull(merged);
+ merged = AggregationFunctionUtils.mergeSerialized(function, merged, found);
+ merged = AggregationFunctionUtils.mergeSerialized(function, merged, empty);
+ assertEquals(merged, 1);
+
assertEquals(function.getFinalResultColumnType().convert(function.extractFinalResult(merged)),
true);
+
+ AnyValueAggregationFunction bigDecimalFunction =
createFunction("BIG_DECIMAL");
+ BigDecimal value = new BigDecimal("12345678901234567890.123");
+
assertEquals(bigDecimalFunction.merge(bigDecimalFunction.deserializeIntermediateResult(
+ serialize(bigDecimalFunction, value)), new BigDecimal("1")), value);
+ }
+
+ private static AnyValueAggregationFunction createFunction(String dataType) {
+ return new AnyValueAggregationFunction(List.of(EXPRESSION,
ExpressionContext.forLiteral(DataType.STRING, dataType)),
+ true);
+ }
+
+ private static CustomObject serialize(AnyValueAggregationFunction function,
Object intermediateResult) {
+ AggregationFunction.SerializedIntermediateResult serialized =
+ function.serializeIntermediateResult(intermediateResult);
+ return new CustomObject(serialized.getType(),
ByteBuffer.wrap(serialized.getBytes()));
+ }
}
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java
index ffef0397a46..dc58ca7bac4 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/optimizer/statement/AggregateFunctionRewriteOptimizerTest.java
@@ -50,6 +50,7 @@ public class AggregateFunctionRewriteOptimizerTest {
.addSingleValueDimension("timeCol", DataType.LONG)
.build();
private static final String AUTO_REWRITE = "SET
autoRewriteAggregationType=true; ";
+ private static final String INFER_ANY_VALUE_TYPE = "SET
inferAnyValueType=true; ";
@DataProvider
public Object[][] dataTypes() {
@@ -102,6 +103,53 @@ public class AggregateFunctionRewriteOptimizerTest {
+ "ORDER BY LAST_WITH_TIME(timestampCol, timeCol, 'TIMESTAMP')");
}
+ @Test(dataProvider = "dataTypes")
+ public void testAnyValueInfersDataType(String column, String dataType) {
+ for (String function : new String[]{"ANY_VALUE", "anyValue"}) {
+ assertRewrite(INFER_ANY_VALUE_TYPE + "SELECT " + function + "(" + column
+ ") FROM testTable",
+ INFER_ANY_VALUE_TYPE + "SELECT " + function + "(" + column + ", '" +
dataType + "') FROM testTable");
+ }
+ }
+
+ @Test
+ public void testAnyValueRejectsMultiValueColumn() {
+ PinotQuery query =
+ CalciteSqlParser.compileToPinotQuery(INFER_ANY_VALUE_TYPE + "SELECT
ANY_VALUE(mvCol) FROM testTable");
+ BadQueryRequestException exception =
expectThrows(BadQueryRequestException.class,
+ () -> new AggregateFunctionRewriteOptimizer().optimize(query, SCHEMA));
+ assertEquals(exception.getMessage(), "Unsupported data type for
ANYVALUE(expr): multi-value INT");
+ assertUnchanged("SELECT ANY_VALUE(mvCol) FROM testTable");
+ }
+
+ @Test
+ public void testAnyValueKeepsArgumentsWithoutSchemaType() {
+ for (String column : new String[]{"missingCol", "trim(stringCol)"}) {
+ assertUnchanged(INFER_ANY_VALUE_TYPE + "SELECT ANY_VALUE(" + column + ")
FROM testTable");
+ }
+ assertUnchanged(INFER_ANY_VALUE_TYPE + "SELECT ANY_VALUE(intCol, 'LONG')
FROM testTable");
+ }
+
+ @Test
+ public void testRewritesNestedAnyValueConsistently() {
+ assertRewrite(INFER_ANY_VALUE_TYPE + "SELECT intCol, ANY_VALUE(stringCol)
AS v1, ANY_VALUE(doubleCol) FILTER "
+ + "(WHERE boolCol) FROM testTable GROUP BY intCol HAVING
ANY_VALUE(longCol) > 1 "
+ + "ORDER BY ANY_VALUE(timestampCol)",
+ INFER_ANY_VALUE_TYPE + "SELECT intCol, ANY_VALUE(stringCol, 'STRING')
AS v1, ANY_VALUE(doubleCol, 'DOUBLE') "
+ + "FILTER (WHERE boolCol) FROM testTable GROUP BY intCol HAVING
ANY_VALUE(longCol, 'LONG') > 1 "
+ + "ORDER BY ANY_VALUE(timestampCol, 'TIMESTAMP')");
+ }
+
+ @Test
+ public void testAnyValueDataTypeRequiresQueryOption() {
+ String query = "SELECT ANY_VALUE(intCol), FIRST_WITH_TIME(intCol, timeCol)
FROM testTable";
+ String expectedQuery = "SELECT ANY_VALUE(intCol, 'INT'),
FIRST_WITH_TIME(intCol, timeCol, 'INT') FROM testTable";
+ assertRewrite(INFER_ANY_VALUE_TYPE + query, INFER_ANY_VALUE_TYPE +
expectedQuery);
+ // The option only applies to ANY_VALUE.
+ String unchangedAnyValue = "SELECT ANY_VALUE(intCol),
FIRST_WITH_TIME(intCol, timeCol, 'INT') FROM testTable";
+ assertRewrite(query, unchangedAnyValue);
+ assertRewrite("SET inferAnyValueType=false; " + query, "SET
inferAnyValueType=false; " + unchangedAnyValue);
+ }
+
@Test
public void testModeInfersNonNumericDataType() {
assertRewrite("SELECT MODE(stringCol), MODE(jsonCol), MODE(boolCol,
'MAX'), MODE(timestampCol, 'AVG') "
diff --git
a/pinot-core/src/test/java/org/apache/pinot/queries/AnyValueQueriesTest.java
b/pinot-core/src/test/java/org/apache/pinot/queries/AnyValueQueriesTest.java
new file mode 100644
index 00000000000..a8d6916f133
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/queries/AnyValueQueriesTest.java
@@ -0,0 +1,230 @@
+/**
+ * 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.queries;
+
+import java.io.File;
+import java.math.BigDecimal;
+import java.sql.Timestamp;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.common.response.broker.BrokerResponseNative;
+import org.apache.pinot.common.response.broker.ResultTable;
+import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import
org.apache.pinot.segment.local.indexsegment.immutable.ImmutableSegmentLoader;
+import
org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl;
+import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader;
+import org.apache.pinot.segment.spi.ImmutableSegment;
+import org.apache.pinot.segment.spi.IndexSegment;
+import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+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.ReadMode;
+import org.apache.pinot.spi.utils.builder.TableConfigBuilder;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertTrue;
+
+
+/// Queries test for ANY_VALUE, whose data type argument the optimizer
supplies from the schema when the
+/// `inferAnyValueType` query option is set. All the rows with the same key
hold the same values, so that any value is
+/// the expected one.
+public class AnyValueQueriesTest extends BaseQueriesTest {
+ private static final File INDEX_DIR = new File(FileUtils.getTempDirectory(),
"AnyValueQueriesTest");
+ private static final String RAW_TABLE_NAME = "testTable";
+ private static final String SEGMENT_NAME = "testSegment";
+ private static final int NUM_RECORDS = 10;
+
+ private static final Schema SCHEMA = new
Schema.SchemaBuilder().setSchemaName(RAW_TABLE_NAME)
+ .addSingleValueDimension("keyCol", DataType.INT)
+ .addSingleValueDimension("intCol", DataType.INT)
+ .addSingleValueDimension("longCol", DataType.LONG)
+ .addSingleValueDimension("floatCol", DataType.FLOAT)
+ .addSingleValueDimension("doubleCol", DataType.DOUBLE)
+ .addMetric("bigDecimalCol", DataType.BIG_DECIMAL)
+ .addSingleValueDimension("boolCol", DataType.BOOLEAN)
+ .addSingleValueDimension("timestampCol", DataType.TIMESTAMP)
+ .addSingleValueDimension("stringCol", DataType.STRING)
+ .addSingleValueDimension("jsonCol", DataType.JSON)
+ .addSingleValueDimension("bytesCol", DataType.BYTES)
+ .addSingleValueDimension("uuidCol", DataType.UUID)
+ .build();
+ private static final TableConfig TABLE_CONFIG = new
TableConfigBuilder(TableType.OFFLINE).setTableName(RAW_TABLE_NAME)
+ .setNoDictionaryColumns(List.of("longCol", "doubleCol",
"bytesCol")).build();
+ private static final String INFER_ANY_VALUE_TYPE = "SET
inferAnyValueType=true; ";
+ private static final String ALL_COLUMNS = "ANY_VALUE(intCol),
ANY_VALUE(longCol), ANY_VALUE(floatCol), "
+ + "ANY_VALUE(doubleCol), ANY_VALUE(bigDecimalCol), ANY_VALUE(boolCol),
ANY_VALUE(timestampCol), "
+ + "ANY_VALUE(stringCol), ANY_VALUE(jsonCol), ANY_VALUE(bytesCol),
ANY_VALUE(uuidCol)";
+
+ private IndexSegment _indexSegment;
+ private List<IndexSegment> _indexSegments;
+
+ @Override
+ protected String getFilter() {
+ return "";
+ }
+
+ @Override
+ protected IndexSegment getIndexSegment() {
+ return _indexSegment;
+ }
+
+ @Override
+ protected List<IndexSegment> getIndexSegments() {
+ return _indexSegments;
+ }
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ FileUtils.deleteDirectory(INDEX_DIR);
+
+ List<GenericRow> records = new ArrayList<>(NUM_RECORDS);
+ for (int i = 0; i < NUM_RECORDS; i++) {
+ int key = i % 2;
+ GenericRow record = new GenericRow();
+ record.putValue("keyCol", key);
+ record.putValue("intCol", 10 + key);
+ record.putValue("longCol", 10_000_000_000L + key);
+ record.putValue("floatCol", 0.5f + key);
+ record.putValue("doubleCol", 0.25 + key);
+ record.putValue("bigDecimalCol", new
BigDecimal("12345678901234567890.5").add(BigDecimal.valueOf(key)));
+ record.putValue("boolCol", key == 0);
+ record.putValue("timestampCol", 1_600_000_000_000L + key);
+ record.putValue("stringCol", "s" + key);
+ record.putValue("jsonCol", "{\"k\":" + key + "}");
+ record.putValue("bytesCol", new byte[]{(byte) key, 1});
+ record.putValue("uuidCol", "123e4567-e89b-12d3-a456-42661417400" + key);
+ records.add(record);
+ }
+
+ SegmentGeneratorConfig segmentGeneratorConfig = new
SegmentGeneratorConfig(TABLE_CONFIG, SCHEMA);
+ segmentGeneratorConfig.setTableName(RAW_TABLE_NAME);
+ segmentGeneratorConfig.setSegmentName(SEGMENT_NAME);
+ segmentGeneratorConfig.setOutDir(INDEX_DIR.getPath());
+ SegmentIndexCreationDriverImpl driver = new
SegmentIndexCreationDriverImpl();
+ driver.init(segmentGeneratorConfig, new GenericRowRecordReader(records));
+ driver.build();
+
+ ImmutableSegment immutableSegment = ImmutableSegmentLoader.load(new
File(INDEX_DIR, SEGMENT_NAME), ReadMode.mmap);
+ _indexSegment = immutableSegment;
+ _indexSegments = List.of(immutableSegment, immutableSegment);
+ }
+
+ @Test
+ public void testAggregationOnly() {
+ BrokerResponseNative brokerResponse = getBrokerResponseForOptimizedQuery(
+ INFER_ANY_VALUE_TYPE + "SELECT " + ALL_COLUMNS + " FROM testTable
WHERE keyCol = 0", SCHEMA);
+ DataSchema expectedDataSchema = new DataSchema(new String[]{
+ "anyvalue(intCol,'INT')", "anyvalue(longCol,'LONG')",
"anyvalue(floatCol,'FLOAT')",
+ "anyvalue(doubleCol,'DOUBLE')",
"anyvalue(bigDecimalCol,'BIG_DECIMAL')", "anyvalue(boolCol,'BOOLEAN')",
+ "anyvalue(timestampCol,'TIMESTAMP')", "anyvalue(stringCol,'STRING')",
"anyvalue(jsonCol,'STRING')",
+ "anyvalue(bytesCol,'BYTES')", "anyvalue(uuidCol,'UUID')"
+ }, new ColumnDataType[]{
+ ColumnDataType.INT, ColumnDataType.LONG, ColumnDataType.FLOAT,
ColumnDataType.DOUBLE,
+ ColumnDataType.BIG_DECIMAL, ColumnDataType.BOOLEAN,
ColumnDataType.TIMESTAMP, ColumnDataType.STRING,
+ ColumnDataType.STRING, ColumnDataType.BYTES, ColumnDataType.UUID
+ });
+ Object[] expectedRow = new Object[]{
+ 10, 10_000_000_000L, 0.5f, 0.25, "12345678901234567890.5", true,
+ new Timestamp(1_600_000_000_000L).toString(), "s0", "{\"k\":0}",
"0001", "123e4567-e89b-12d3-a456-426614174000"
+ };
+ QueriesTestUtils.testInterSegmentsResult(brokerResponse,
+ new ResultTable(expectedDataSchema, List.<Object[]>of(expectedRow)));
+ }
+
+ @Test
+ public void testGroupBy() {
+ ResultTable resultTable = getBrokerResponseForOptimizedQuery(
+ INFER_ANY_VALUE_TYPE + "SELECT keyCol, ANY_VALUE(intCol) AS v1,
ANY_VALUE(boolCol) AS v2, "
+ + "ANY_VALUE(timestampCol) AS v3, "
+ + "ANY_VALUE(uuidCol) AS v4 FROM testTable GROUP BY keyCol ORDER
BY ANY_VALUE(longCol) DESC",
+ SCHEMA).getResultTable();
+ assertEquals(resultTable.getDataSchema(), new DataSchema(new
String[]{"keyCol", "v1", "v2", "v3", "v4"},
+ new ColumnDataType[]{
+ ColumnDataType.INT, ColumnDataType.INT, ColumnDataType.BOOLEAN,
ColumnDataType.TIMESTAMP,
+ ColumnDataType.UUID
+ }));
+ List<Object[]> rows = resultTable.getRows();
+ assertEquals(rows.size(), 2);
+ assertEquals(rows.get(0), new Object[]{
+ 1, 11, false, new Timestamp(1_600_000_000_001L).toString(),
"123e4567-e89b-12d3-a456-426614174001"
+ });
+ assertEquals(rows.get(1), new Object[]{
+ 0, 10, true, new Timestamp(1_600_000_000_000L).toString(),
"123e4567-e89b-12d3-a456-426614174000"
+ });
+ }
+
+ @Test
+ public void testEmptyInput() {
+ ResultTable resultTable = getBrokerResponseForOptimizedQuery(
+ INFER_ANY_VALUE_TYPE + "SELECT " + ALL_COLUMNS + " FROM testTable
WHERE keyCol < 0", SCHEMA).getResultTable();
+ assertEquals(resultTable.getDataSchema().getColumnDataTypes(), new
ColumnDataType[]{
+ ColumnDataType.INT, ColumnDataType.LONG, ColumnDataType.FLOAT,
ColumnDataType.DOUBLE,
+ ColumnDataType.BIG_DECIMAL, ColumnDataType.BOOLEAN,
ColumnDataType.TIMESTAMP, ColumnDataType.STRING,
+ ColumnDataType.STRING, ColumnDataType.BYTES, ColumnDataType.UUID
+ });
+ assertEquals(resultTable.getRows().size(), 1);
+ assertEquals(resultTable.getRows().get(0), new Object[11]);
+
+ resultTable = getBrokerResponseForOptimizedQuery(
+ INFER_ANY_VALUE_TYPE + "SELECT keyCol, ANY_VALUE(boolCol),
ANY_VALUE(timestampCol) FROM testTable "
+ + "WHERE keyCol < 0 GROUP BY keyCol",
+ SCHEMA).getResultTable();
+ assertEquals(resultTable.getDataSchema().getColumnDataTypes(),
+ new ColumnDataType[]{ColumnDataType.INT, ColumnDataType.BOOLEAN,
ColumnDataType.TIMESTAMP});
+ assertTrue(resultTable.getRows().isEmpty());
+ }
+
+ @Test
+ public void testExplicitDataType() {
+ // SSE accepts convertible explicit types; MSE requires the inferred type.
+ ResultTable resultTable = getBrokerResponseForOptimizedQuery(
+ "SELECT ANY_VALUE(intCol, 'LONG'), ANY_VALUE(intCol, 'STRING'),
ANY_VALUE(timestampCol, 'LONG') "
+ + "FROM testTable WHERE keyCol = 0", SCHEMA).getResultTable();
+ assertEquals(resultTable.getDataSchema().getColumnDataTypes(),
+ new ColumnDataType[]{ColumnDataType.LONG, ColumnDataType.STRING,
ColumnDataType.LONG});
+ assertEquals(resultTable.getRows().get(0), new Object[]{10L, "10",
1_600_000_000_000L});
+ }
+
+ @Test
+ public void testWithoutInferredDataType() {
+ ResultTable resultTable = getBrokerResponseForOptimizedQuery(
+ "SELECT ANY_VALUE(intCol), ANY_VALUE(boolCol), ANY_VALUE(timestampCol)
"
+ + "FROM testTable WHERE keyCol = 0", SCHEMA).getResultTable();
+ assertEquals(resultTable.getDataSchema(),
+ new DataSchema(new String[]{"anyvalue(intCol)", "anyvalue(boolCol)",
"anyvalue(timestampCol)"},
+ new ColumnDataType[]{ColumnDataType.STRING, ColumnDataType.STRING,
ColumnDataType.STRING}));
+ assertEquals(resultTable.getRows().get(0), new Object[]{"10", "1",
"1600000000000"});
+ }
+
+ @AfterClass
+ public void tearDown()
+ throws Exception {
+ _indexSegment.destroy();
+ FileUtils.deleteDirectory(INDEX_DIR);
+ }
+}
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java
index a2b2c2cccd4..edbc553d07f 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateTypeArgumentRewriteRule.java
@@ -19,6 +19,7 @@
package org.apache.pinot.calcite.rel.rules;
import java.util.ArrayList;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -33,10 +34,14 @@ 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.rel.type.RelDataType;
import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.tools.RelBuilderFactory;
+import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.query.planner.logical.RelToPlanNodeConverter;
import org.apache.pinot.segment.spi.AggregationFunctionType;
@@ -44,21 +49,36 @@ import org.apache.pinot.segment.spi.AggregationFunctionType;
/// - FIRST_WITH_TIME(dataColumn, timeColumn) -> FIRST_WITH_TIME(dataColumn,
timeColumn, 'dataType')
/// - LAST_WITH_TIME(dataColumn, timeColumn) -> LAST_WITH_TIME(dataColumn,
timeColumn, 'dataType')
/// - MODE(stringColumn) -> MODE(stringColumn, 'MIN', 'STRING')
+/// - ANY_VALUE(expr) -> ANY_VALUE(expr, 'dataType'), except by
[#WITHOUT_ANY_VALUE]
///
/// The result type of the call is unchanged. The literals are projected
directly beneath the aggregate, appended to
/// the existing project when there is one, because
[PinotAggregateExchangeNodeInsertRule] and `AggregatePushdownRule`
/// inline literal arguments only from the immediate project.
///
-/// The servers require the data type argument, so the rule is required rather
than an optimization: `QueryEnvironment`
-/// always applies it, and `skipPlannerRules` cannot turn it off.
+/// The servers require the data type argument of FIRST_WITH_TIME /
LAST_WITH_TIME, so the rule is required rather than
+/// an optimization: `QueryEnvironment` always applies [#WITHOUT_ANY_VALUE],
or [#INSTANCE] when the `inferAnyValueType`
+/// query option is true, and `skipPlannerRules` cannot turn it off.
public class PinotAggregateTypeArgumentRewriteRule extends RelOptRule {
+ private static final Set<ColumnDataType> ANY_VALUE_DATA_TYPES =
+ EnumSet.of(ColumnDataType.INT, ColumnDataType.LONG,
ColumnDataType.FLOAT, ColumnDataType.DOUBLE,
+ ColumnDataType.BIG_DECIMAL, ColumnDataType.BOOLEAN,
ColumnDataType.TIMESTAMP, ColumnDataType.STRING,
+ ColumnDataType.BYTES, ColumnDataType.UUID);
+
public static final PinotAggregateTypeArgumentRewriteRule INSTANCE =
- new
PinotAggregateTypeArgumentRewriteRule(PinotRuleUtils.PINOT_REL_FACTORY);
+ new PinotAggregateTypeArgumentRewriteRule(true,
PinotRuleUtils.PINOT_REL_FACTORY, null);
+ /// Leaves ANY_VALUE without the data type argument.
+ public static final PinotAggregateTypeArgumentRewriteRule WITHOUT_ANY_VALUE =
+ new PinotAggregateTypeArgumentRewriteRule(false,
PinotRuleUtils.PINOT_REL_FACTORY,
+ "PinotAggregateTypeArgumentRewriteRuleWithoutAnyValue");
+
+ private final boolean _rewriteAnyValue;
// 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);
+ private PinotAggregateTypeArgumentRewriteRule(boolean rewriteAnyValue,
RelBuilderFactory factory,
+ @Nullable String description) {
+ super(operand(LogicalAggregate.class, any()), factory, description);
+ _rewriteAnyValue = rewriteAnyValue;
}
@Override
@@ -121,7 +141,7 @@ public class PinotAggregateTypeArgumentRewriteRule extends
RelOptRule {
/// Returns the literal arguments to append, or `null` when the call already
has them or is not such a call.
@Nullable
- private static List<String> getLiteralArguments(AggregateCall aggCall,
RelNode input) {
+ private List<String> getLiteralArguments(AggregateCall aggCall, RelNode
input) {
List<Integer> argList = aggCall.getArgList();
String name =
AggregationFunctionType.getNormalizedAggregationFunctionName(aggCall.getAggregation().getName());
if (name.equals(AggregationFunctionType.MODE.name())) {
@@ -133,11 +153,22 @@ public class PinotAggregateTypeArgumentRewriteRule
extends RelOptRule {
|| name.equals(AggregationFunctionType.LASTWITHTIME.name()))) {
return List.of(getDataTypeArgument(name, getSqlTypeName(input,
argList.get(0))));
}
+ if (name.equals(AggregationFunctionType.ANYVALUE.name())) {
+ if (argList.size() == 2) {
+ checkAnyValueDataTypeArgument(name, argList, input);
+ } else if (_rewriteAnyValue) {
+ return List.of(getAnyValueDataType(name + "(expr)",
getArgumentType(input, argList.get(0))));
+ }
+ }
return null;
}
+ private static RelDataType getArgumentType(RelNode input, int argIndex) {
+ return input.getRowType().getFieldList().get(argIndex).getType();
+ }
+
private static SqlTypeName getSqlTypeName(RelNode input, int fieldIndex) {
- return
input.getRowType().getFieldList().get(fieldIndex).getType().getSqlTypeName();
+ return getArgumentType(input, fieldIndex).getSqlTypeName();
}
/// Returns the data type argument of FIRST_WITH_TIME / LAST_WITH_TIME.
Throws for a data argument without an
@@ -174,4 +205,35 @@ public class PinotAggregateTypeArgumentRewriteRule extends
RelOptRule {
String.format("Unsupported data type for %s(dataColumn,
timeColumn): %s", name, dataType));
}
}
+
+ /// Returns the type the multi-stage engine reads the argument as, e.g.
DOUBLE for a DECIMAL expression with a scale,
+ /// so that the values are read without a conversion that can fail, such as
an infinite DOUBLE to BIG_DECIMAL.
+ private static String getAnyValueDataType(String call, RelDataType
relDataType) {
+ ColumnDataType dataType =
RelToPlanNodeConverter.convertToColumnDataType(relDataType);
+ if (!ANY_VALUE_DATA_TYPES.contains(dataType)) {
+ throw new IllegalArgumentException(
+ String.format("Unsupported data type for %s: %s", call,
relDataType.getSqlTypeName()));
+ }
+ return dataType.name();
+ }
+
+ /// The result type of ANY_VALUE follows its data argument, so an explicit
data type argument must be the one this
+ /// rule supplies for it.
+ private static void checkAnyValueDataTypeArgument(String name, List<Integer>
argList, RelNode input) {
+ RelDataType relDataType = getArgumentType(input, argList.get(0));
+ String dataType = getAnyValueDataType(name + "(expr, 'dataType')",
relDataType);
+ RelNode unboxedInput = PinotRuleUtils.unboxRel(input);
+ RexNode dataTypeArgument =
+ unboxedInput instanceof Project ? ((Project)
unboxedInput).getProjects().get(argList.get(1)) : null;
+ if (!(dataTypeArgument instanceof RexLiteral)) {
+ throw new IllegalArgumentException(
+ String.format("%s expects a literal data type argument, got: %s",
name, dataTypeArgument));
+ }
+ String explicitDataType = ((RexLiteral)
dataTypeArgument).getValueAs(String.class);
+ if (!dataType.equalsIgnoreCase(explicitDataType)) {
+ throw new IllegalArgumentException(
+ String.format("Unsupported data type for %s(expr, 'dataType') with
expr of type %s: %s", name,
+ relDataType.getSqlTypeName(), explicitDataType != null ?
explicitDataType : dataTypeArgument));
+ }
+ }
}
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 da842624029..8b50aea94f6 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
@@ -147,6 +147,8 @@ public class QueryEnvironment {
private final FrameworkConfig _config;
private final CalciteCatalogReader _catalogReader;
private final HepProgram _optProgram;
+ // Same as _optProgram, but rewrites ANY_VALUE(expr) to ANY_VALUE(expr,
'dataType') as well
+ private final HepProgram _inferAnyValueTypeOptProgram;
private final Config _envConfig;
private final PinotCatalog _catalog;
private final Set<String> _defaultDisabledPlannerRules;
@@ -171,7 +173,9 @@ public class QueryEnvironment {
rootSchema, List.of(database), _typeFactory, CONNECTION_CONFIG,
config.isCaseSensitive());
_defaultDisabledPlannerRules = _envConfig.defaultDisabledPlannerRules();
// default optProgram with no skip rule options and no use rule options
- _optProgram = getOptProgram(_envConfig.getRuleSet(), Set.of(), Set.of(),
_defaultDisabledPlannerRules);
+ _optProgram = getOptProgram(_envConfig.getRuleSet(), Set.of(), Set.of(),
_defaultDisabledPlannerRules, false);
+ _inferAnyValueTypeOptProgram =
+ getOptProgram(_envConfig.getRuleSet(), Set.of(), Set.of(),
_defaultDisabledPlannerRules, true);
_multiClusterRoutingContext = multiClusterRoutingContext;
}
@@ -198,7 +202,8 @@ public class QueryEnvironment {
_catalog.configureVirtualColumnExclusion(true);
}
- HepProgram optProgram = _optProgram;
+ boolean inferAnyValueType = QueryOptionsUtils.isInferAnyValueType(options);
+ HepProgram optProgram = inferAnyValueType ? _inferAnyValueTypeOptProgram :
_optProgram;
Set<String> useRuleSet = new
HashSet<>(QueryOptionsUtils.getUsePlannerRules(options));
if
(Boolean.parseBoolean(options.get(QueryOptionKey.AUTO_REWRITE_AGGREGATION_TYPE)))
{
useRuleSet.add(CommonConstants.Broker.PlannerRuleNames.AGGREGATE_FUNCTION_REWRITE);
@@ -207,7 +212,8 @@ public class QueryEnvironment {
Set<String> skipRuleSet = QueryOptionsUtils.getSkipPlannerRules(options);
if (!skipRuleSet.isEmpty() || !useRuleSet.isEmpty()) {
// dynamically create optProgram according to rule options
- optProgram = getOptProgram(_envConfig.getRuleSet(), skipRuleSet,
useRuleSet, _defaultDisabledPlannerRules);
+ optProgram = getOptProgram(_envConfig.getRuleSet(), skipRuleSet,
useRuleSet, _defaultDisabledPlannerRules,
+ inferAnyValueType);
}
}
int sortExchangeCopyLimit =
QueryOptionsUtils.getSortExchangeCopyThreshold(options,
@@ -579,9 +585,10 @@ public class QueryEnvironment {
/// @param skipRuleSet parsed skipped rule name set from query options
/// @param useRuleSet parsed use rule set from query options
/// @param defaultDisabledRuleSet parsed default disabled rule set from
broker config
+ /// @param inferAnyValueType whether ANY_VALUE gets the data type argument
of its argument's type
/// @return HepProgram that performs logical transformations
private static HepProgram getOptProgram(PinotRuleSet ruleSet, Set<String>
skipRuleSet, Set<String> useRuleSet,
- Set<String> defaultDisabledRuleSet) {
+ Set<String> defaultDisabledRuleSet, boolean inferAnyValueType) {
HepProgramBuilder hepProgramBuilder = new HepProgramBuilder();
// Set the match order as DEPTH_FIRST. The default is arbitrary which
works the same as DEPTH_FIRST, but it's
// best to be explicit.
@@ -605,9 +612,10 @@ public class QueryEnvironment {
for (RelOptRule relOptRule : basicRules) {
hepProgramBuilder.addRuleInstance(relOptRule);
}
- // Required to plan FIRST_WITH_TIME / LAST_WITH_TIME without a data type
and MODE on strings, so the rule options
+ // Required for FIRST_WITH_TIME / LAST_WITH_TIME without a data type and
MODE on strings, so the rule options
// do not filter it.
-
hepProgramBuilder.addRuleInstance(PinotAggregateTypeArgumentRewriteRule.INSTANCE);
+ hepProgramBuilder.addRuleInstance(inferAnyValueType ?
PinotAggregateTypeArgumentRewriteRule.INSTANCE
+ : PinotAggregateTypeArgumentRewriteRule.WITHOUT_ANY_VALUE);
// ----
// Pushdown filters using a single HepInstruction.
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 7ed615549e2..27e466cc001 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
@@ -36,8 +36,9 @@ import static org.testng.Assert.*;
public class QueryPlannerRuleOptionsTest extends QueryEnvironmentTestBase {
private String explainQueryWithRuleDisabled(String query, String ruleToSkip)
{
- SqlNode sqlNode =
CalciteSqlParser.compileToSqlNodeAndOptions(query).getSqlNode();
- Map<String, String> options = new HashMap<>();
+ SqlNodeAndOptions parsed =
CalciteSqlParser.compileToSqlNodeAndOptions(query);
+ SqlNode sqlNode = parsed.getSqlNode();
+ Map<String, String> options = new HashMap<>(parsed.getOptions());
// disable rule
options.put(CommonConstants.Broker.Request.QueryOptionKey.SKIP_PLANNER_RULES,
ruleToSkip);
SqlNodeAndOptions sqlNodeAndOptions =
@@ -97,10 +98,19 @@ public class QueryPlannerRuleOptionsTest extends
QueryEnvironmentTestBase {
@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());
+ String query = "EXPLAIN PLAN FOR SELECT FIRST_WITH_TIME(col1, ts),
ANY_VALUE(col3) FROM a";
+ String explain =
+ explainQueryWithRuleDisabled(query,
PinotAggregateTypeArgumentRewriteRule.WITHOUT_ANY_VALUE.toString());
+ assertEquals(explain, _queryEnvironment.explainQuery(query,
RANDOM_REQUEST_ID_GEN.nextLong()));
+ assertTrue(explain.contains("FIRSTWITHTIME($0, $1, _UTF-8'STRING')"),
explain);
+ assertTrue(explain.contains("ANYVALUE($2)"), explain);
+
+ // With inferAnyValueType, the program runs the rule that also rewrites
ANY_VALUE
+ query = "SET inferAnyValueType=true; " + query;
+ 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);
+ assertTrue(explain.contains("ANYVALUE($2, _UTF-8'INT')"), explain);
}
@Test
diff --git a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
index 3be6c9e0d64..d025fa29406 100644
--- a/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
+++ b/pinot-query-planner/src/test/resources/queries/AggregatePlans.json
@@ -223,6 +223,78 @@
"\n"
]
},
+ {
+ "description": "ANY_VALUE without a data type infers it from its
argument",
+ "sql": "SET inferAnyValueType=true; EXPLAIN PLAN FOR SELECT
ANY_VALUE(a.col1), ANY_VALUE(a.col3), ANY_VALUE(a.col4), ANY_VALUE(a.col5),
ANY_VALUE(a.col7), ANY_VALUE(a.ts_timestamp) FROM a",
+ "output": [
+ "Execution Plan",
+ "\nPinotLogicalAggregate(group=[{}], agg#0=[ANYVALUE($0,
_UTF-8'STRING')], agg#1=[ANYVALUE($1, _UTF-8'INT')], agg#2=[ANYVALUE($2,
_UTF-8'BIG_DECIMAL')], agg#3=[ANYVALUE($3, _UTF-8'BOOLEAN')],
agg#4=[ANYVALUE($4, _UTF-8'LONG')], agg#5=[ANYVALUE($5, _UTF-8'TIMESTAMP')],
aggType=[FINAL])",
+ "\n PinotLogicalExchange(distribution=[hash])",
+ "\n PinotLogicalAggregate(group=[{}], agg#0=[ANYVALUE($0,
_UTF-8'STRING')], agg#1=[ANYVALUE($1, _UTF-8'INT')], agg#2=[ANYVALUE($2,
_UTF-8'BIG_DECIMAL')], agg#3=[ANYVALUE($3, _UTF-8'BOOLEAN')],
agg#4=[ANYVALUE($4, _UTF-8'LONG')], agg#5=[ANYVALUE($5, _UTF-8'TIMESTAMP')],
aggType=[LEAF])",
+ "\n LogicalProject($f0=[$0], $f1=[$2], $f2=[$3], $f3=[$4],
$f4=[$6], $f5=[$8], $f6=[_UTF-8'STRING'], $f7=[_UTF-8'INT'],
$f8=[_UTF-8'BIG_DECIMAL'], $f9=[_UTF-8'BOOLEAN'], $f10=[_UTF-8'LONG'],
$f11=[_UTF-8'TIMESTAMP'])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Inferred ANY_VALUE data type for a transform and a
filtered aggregate in a grouped aggregate",
+ "sql": "SET inferAnyValueType=true; EXPLAIN PLAN FOR SELECT a.col2,
ANY_VALUE(a.col3 + 1), ANY_VALUE(a.col5) FILTER (WHERE a.col3 > 0),
ANY_VALUE(a.col1, 'STRING') FROM a GROUP BY a.col2",
+ "output": [
+ "Execution Plan",
+ "\nPinotLogicalAggregate(group=[{0}], agg#0=[ANYVALUE($1,
_UTF-8'INT')], agg#1=[ANYVALUE($2, _UTF-8'BOOLEAN')], agg#2=[ANYVALUE($3,
_UTF-8'STRING')], aggType=[FINAL])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n PinotLogicalAggregate(group=[{0}], agg#0=[ANYVALUE($1,
_UTF-8'INT')], agg#1=[ANYVALUE($2, _UTF-8'BOOLEAN') FILTER $3],
agg#2=[ANYVALUE($4, _UTF-8'STRING')], aggType=[LEAF])",
+ "\n LogicalProject($f0=[$1], $f1=[+($2, 1)], $f2=[$4],
$f3=[>($2, 0)], $f4=[$0], $f5=[_UTF-8'STRING'], $f6=[_UTF-8'INT'],
$f7=[_UTF-8'BOOLEAN'])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "ANY_VALUE over a DECIMAL expression takes the type the
expression is read as",
+ "sql": "SET inferAnyValueType=true; EXPLAIN PLAN FOR SELECT
ANY_VALUE(a.col3 * 1.5), ANY_VALUE(CAST(a.col3 AS DECIMAL(10, 0))) FROM a",
+ "output": [
+ "Execution Plan",
+ "\nPinotLogicalAggregate(group=[{}], agg#0=[ANYVALUE($0,
_UTF-8'DOUBLE')], agg#1=[ANYVALUE($1, _UTF-8'INT')], aggType=[FINAL])",
+ "\n PinotLogicalExchange(distribution=[hash])",
+ "\n PinotLogicalAggregate(group=[{}], agg#0=[ANYVALUE($0,
_UTF-8'DOUBLE')], agg#1=[ANYVALUE($1, _UTF-8'INT')], aggType=[LEAF])",
+ "\n LogicalProject($f0=[*($2, 1.5:DECIMAL(2, 1))],
$f1=[CAST($2):DECIMAL(10, 0) NOT NULL], $f2=[_UTF-8'DOUBLE'],
$f3=[_UTF-8'INT'])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "ANY_VALUE without a data type rejects a multi-value
argument",
+ "sql": "SET inferAnyValueType=true; EXPLAIN PLAN FOR SELECT
ANY_VALUE(e.mcol1) FROM e",
+ "expectedException": ".*Unsupported data type for ANYVALUE\\(expr\\):
ARRAY.*"
+ },
+ {
+ "description": "ANY_VALUE rejects a data type that does not match its
argument",
+ "sql": "EXPLAIN PLAN FOR SELECT ANY_VALUE(a.col3, 'STRING') FROM a",
+ "expectedException": ".*Unsupported data type for ANYVALUE\\(expr,
'dataType'\\) with expr of type INTEGER: STRING.*"
+ },
+ {
+ "description": "ANY_VALUE rejects a data type that does not match its
argument when inferAnyValueType is true",
+ "sql": "SET inferAnyValueType=true; EXPLAIN PLAN FOR SELECT
ANY_VALUE(a.col5, 'INT') FROM a",
+ "expectedException": ".*Unsupported data type for ANYVALUE\\(expr,
'dataType'\\) with expr of type BOOLEAN: INT.*"
+ },
+ {
+ "description": "ANY_VALUE rejects a data type argument that is not a
literal",
+ "sql": "EXPLAIN PLAN FOR SELECT ANY_VALUE(a.col3, a.col1) FROM a",
+ "expectedException": ".*ANYVALUE expects a literal data type argument,
got: \\$.*"
+ },
+ {
+ "description": "ANY_VALUE keeps the form without a data type by
default",
+ "sql": "EXPLAIN PLAN FOR SELECT ANY_VALUE(a.col5),
FIRST_WITH_TIME(a.col5, a.ts) FROM a",
+ "output": [
+ "Execution Plan",
+ "\nPinotLogicalAggregate(group=[{}], agg#0=[ANYVALUE($0)],
agg#1=[FIRSTWITHTIME($1, $1, _UTF-8'BOOLEAN')], aggType=[FINAL])",
+ "\n PinotLogicalExchange(distribution=[hash])",
+ "\n PinotLogicalAggregate(group=[{}], agg#0=[ANYVALUE($0)],
agg#1=[FIRSTWITHTIME($0, $1, _UTF-8'BOOLEAN')], aggType=[LEAF])",
+ "\n LogicalProject($f0=[$4], $f1=[$7], $f2=[_UTF-8'BOOLEAN'])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n"
+ ]
+ },
{
"description": "MIN on STRING column rewritten to MINSTRING",
"sql": "SET autoRewriteAggregationType=true; EXPLAIN PLAN FOR SELECT
MIN(a.col1) FROM a",
diff --git a/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
b/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
index 816e9e2b637..68813c3d637 100644
--- a/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
+++ b/pinot-query-runtime/src/test/resources/queries/UDFAggregates.json
@@ -149,6 +149,31 @@
"sql": "SELECT /*+ aggOptions(is_leaf_return_final_result='true') */
{partitioned_tbl}.num, MODE({partitioned_tbl}.name),
MODE({partitioned_tbl}.flag), MODE({partitioned_tbl}.event_ts) FROM
{partitioned_tbl} /*+ tableOptions(partition_function='hashcode',
partition_key='num', partition_size='4') */ GROUP BY {partitioned_tbl}.num",
"outputs": [[1, "a", true, "2020-01-01 00:00:00.0"], [3, "c", false,
"2020-01-02 00:00:00.0"], [4, "e", false, "2020-01-04 00:00:00.0"], [7, "d",
true, "2020-01-06 00:00:00.0"]]
},
+ {
+ "comment": "The data type argument of ANY_VALUE is inferred when
omitted and inferAnyValueType is true",
+ "sql": "SET inferAnyValueType=true; SELECT ANY_VALUE(int_col),
ANY_VALUE(long_col), ANY_VALUE(float_col), ANY_VALUE(double_col),
ANY_VALUE(string_col), ANY_VALUE(bool_col), ANY_VALUE(timestamp_col) FROM {tbl}
WHERE int_col = 175",
+ "outputs": [[175, 175, 1.75, 1.75, "c", true, "2020-01-02 03:32:12.0"]]
+ },
+ {
+ "sql": "SET inferAnyValueType=true; SELECT int_col,
ANY_VALUE(string_col), ANY_VALUE(bool_col), ANY_VALUE(timestamp_col) FROM {tbl}
WHERE int_col IN (3, 175) GROUP BY int_col",
+ "outputs": [[3, "b", false, "2020-01-01 08:32:12.0"], [175, "c", true,
"2020-01-02 03:32:12.0"]]
+ },
+ {
+ "sql": "SET inferAnyValueType=true; SELECT /*+
aggOptions(is_skip_leaf_stage_aggregate='true') */ int_col,
ANY_VALUE(string_col), ANY_VALUE(bool_col), ANY_VALUE(timestamp_col) FROM {tbl}
WHERE int_col IN (3, 175) GROUP BY int_col",
+ "outputs": [[3, "b", false, "2020-01-01 08:32:12.0"], [175, "c", true,
"2020-01-02 03:32:12.0"]]
+ },
+ {
+ "sql": "SELECT ANY_VALUE(int_col, 'INT'), ANY_VALUE(bool_col,
'BOOLEAN'), ANY_VALUE(timestamp_col, 'TIMESTAMP') FROM {tbl} WHERE int_col =
175",
+ "outputs": [[175, true, "2020-01-02 03:32:12.0"]]
+ },
+ {
+ "sql": "SET inferAnyValueType=true; SELECT ANY_VALUE(bool_col),
ANY_VALUE(timestamp_col) FROM {tbl} WHERE int_col < 0",
+ "outputs": [[null, null]]
+ },
+ {
+ "sql": "SELECT int_col, ANY_VALUE(bool_col), ANY_VALUE(timestamp_col)
FROM {tbl} WHERE int_col = 175 GROUP BY int_col",
+ "outputs": [[175, true, "2020-01-02 03:32:12.0"]]
+ },
{
"sql": "SELECT SUMPRECISION(decimal_col) FROM {tbl}",
"outputs":
[["10000000000100000000110000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000.00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000
[...]
@@ -199,6 +224,47 @@
}
]
},
+ "pinot_any_value_tests": {
+ "tables": {
+ "tbl": {
+ "schema": [
+ {"name": "int_col", "type": "INT"},
+ {"name": "decimal_col", "type": "BIG_DECIMAL"},
+ {"name": "bytes_col", "type": "BYTES"},
+ {"name": "uuid_col", "type": "UUID"}
+ ],
+ "inputs": [
+ [1, "1.5", "DEADBEEF", "123e4567-e89b-12d3-a456-426614174000"],
+ [1, "1.5", "DEADBEEF", "123e4567-e89b-12d3-a456-426614174000"],
+ [2, "2.5", "CAFE", "123e4567-e89b-12d3-a456-426614174001"]
+ ]
+ }
+ },
+ "queries": [
+ {
+ "comment": "ANY_VALUE over the types that only ANY_VALUE infers a data
type for",
+ "sql": "SET inferAnyValueType=true; SELECT ANY_VALUE(decimal_col),
ANY_VALUE(bytes_col), ANY_VALUE(uuid_col), ANY_VALUE(int_col * 1.5) FROM {tbl}
WHERE int_col = 1",
+ "outputs": [["1.5", "deadbeef",
"123e4567-e89b-12d3-a456-426614174000", 1.5]]
+ },
+ {
+ "sql": "SET inferAnyValueType=true; SELECT int_col,
ANY_VALUE(decimal_col), ANY_VALUE(bytes_col), ANY_VALUE(uuid_col),
ANY_VALUE(int_col * 1.5) FROM {tbl} GROUP BY int_col",
+ "outputs": [[1, "1.5", "deadbeef",
"123e4567-e89b-12d3-a456-426614174000", 1.5], [2, "2.5", "cafe",
"123e4567-e89b-12d3-a456-426614174001", 3.0]]
+ },
+ {
+ "sql": "SET inferAnyValueType=true; SELECT /*+
aggOptions(is_skip_leaf_stage_aggregate='true') */ int_col,
ANY_VALUE(decimal_col), ANY_VALUE(bytes_col), ANY_VALUE(uuid_col),
ANY_VALUE(int_col * 1.5) FROM {tbl} GROUP BY int_col",
+ "outputs": [[1, "1.5", "deadbeef",
"123e4567-e89b-12d3-a456-426614174000", 1.5], [2, "2.5", "cafe",
"123e4567-e89b-12d3-a456-426614174001", 3.0]]
+ },
+ {
+ "comment": "The infinite value of int_col = 1 is in the same block as
the returned group",
+ "sql": "SET inferAnyValueType=true; SELECT int_col, ANY_VALUE(int_col
/ (int_col - 1.0)) FROM {tbl} GROUP BY int_col HAVING COUNT(*) = 1",
+ "outputs": [[2, 2.0]]
+ },
+ {
+ "sql": "SET inferAnyValueType=true; SELECT /*+
aggOptions(is_skip_leaf_stage_aggregate='true') */ int_col, ANY_VALUE(int_col /
(int_col - 1.0)) FROM {tbl} GROUP BY int_col HAVING COUNT(*) = 1",
+ "outputs": [[2, 2.0]]
+ }
+ ]
+ },
"pinot_percentile_agg_udf_tests": {
"tables": {
"tbl": {
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 06907a7b730..e4daf33989c 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
@@ -64,7 +64,9 @@ public enum AggregationFunctionType {
SUMPRECISION("sumPrecision", ReturnTypes.explicit(SqlTypeName.DECIMAL),
OperandTypes.ANY, SqlTypeName.OTHER),
AVG("avg", SqlTypeName.OTHER, SqlTypeName.DOUBLE),
MODE("mode", null, null, ReturnTypes.explicit(SqlTypeName.OTHER), new
ModeFinalReturnTypeInference()),
- ANYVALUE("anyValue", ReturnTypes.ARG0, OperandTypes.ANY, SqlTypeName.OTHER),
+ ANYVALUE("anyValue", ReturnTypes.ARG0,
+ OperandTypes.or(OperandTypes.ANY, OperandTypes.family(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),
diff --git
a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index 9e2a6ff7b20..0b494ba02b5 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -800,6 +800,10 @@ public class CommonConstants {
public static final String
CONFIG_OF_BROKER_QUERY_ENABLE_AUTO_REWRITE_AGGREGATION_TYPE =
"pinot.broker.query.enable.auto.rewrite.aggregation.type";
+ /// Default of the [Request.QueryOptionKey#INFER_ANY_VALUE_TYPE] query
option, read from the broker conf and, live,
+ /// from the cluster config.
+ public static final String CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE =
"pinot.broker.query.infer.any.value.type";
+ public static final boolean DEFAULT_BROKER_QUERY_INFER_ANY_VALUE_TYPE =
false;
/// Config for sort exchange copy threshold in multi-stage engine.
///
@@ -1219,6 +1223,10 @@ public class CommonConstants {
// MAX(stringCol) -> MAXSTRING(stringCol)
// SUM(intCol) -> SUMINT(intCol)
public static final String AUTO_REWRITE_AGGREGATION_TYPE =
"autoRewriteAggregationType";
+ /// Whether ANY_VALUE(expr) is rewritten to ANY_VALUE(expr,
'dataType') to return the type of `expr`, instead of
+ /// the legacy result type, which is STRING in the single-stage
engine. Defaults to
+ /// [Broker#CONFIG_OF_BROKER_QUERY_INFER_ANY_VALUE_TYPE].
+ public static final String INFER_ANY_VALUE_TYPE = "inferAnyValueType";
// When enabled, allows multi cluster/federated queries to be executed.
public static final String ENABLE_MULTI_CLUSTER_ROUTING =
"enableMultiClusterRouting";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
