This is an automated email from the ASF dual-hosted git repository.
yashmayya 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 22470be7320 Add a cluster config to rewrite exact DISTINCTCOUNT and
PERCENTILE into approximations (#19510)
22470be7320 is described below
commit 22470be732070645e59611cc73fd4a6948bd02a3
Author: Yash Mayya <[email protected]>
AuthorDate: Wed Sep 16 18:21:16 2026 -0400
Add a cluster config to rewrite exact DISTINCTCOUNT and PERCENTILE into
approximations (#19510)
---
.../broker/broker/helix/BaseBrokerStarter.java | 10 +
.../apache/pinot/broker/querylog/QueryLogger.java | 6 +
.../ApproximateFunctionOverrideProvider.java | 161 ++++++++++++++++
.../requesthandler/BaseBrokerRequestHandler.java | 10 +
.../BaseSingleStageBrokerRequestHandler.java | 98 +++++++---
.../MultiStageBrokerRequestHandler.java | 18 ++
.../pinot/broker/querylog/QueryLoggerTest.java | 1 +
.../ApproximateFunctionOverrideProviderTest.java | 134 ++++++++++++++
.../broker/requesthandler/QueryOverrideTest.java | 56 +++++-
.../apache/pinot/common/metrics/BrokerMeter.java | 5 +
.../pinot/common/response/BrokerResponse.java | 13 ++
.../response/broker/BrokerResponseNative.java | 19 +-
.../response/broker/BrokerResponseNativeV2.java | 18 +-
.../response/broker/CursorResponseNative.java | 5 +-
.../common/utils/config/QueryOptionsUtils.java | 8 +
...istinctCountSmartSketchAggregationFunction.java | 77 ++++++--
.../DistinctCountSmartHLLAggregationFunction.java | 19 +-
...stinctCountSmartHLLPlusAggregationFunction.java | 19 +-
.../DistinctCountSmartULLAggregationFunction.java | 19 +-
.../PercentileSmartTDigestAggregationFunction.java | 69 +++++--
...stinctCountSmartHLLAggregationFunctionTest.java | 126 +++++++++++++
...centileSmartTDigestAggregationFunctionTest.java | 76 ++++++++
...ApproximateFunctionOverrideIntegrationTest.java | 202 +++++++++++++++++++++
.../PinotApproximateAggregateRewriteRule.java | 191 +++++++++++++++++++
.../calcite/rel/rules/PinotQueryRuleSets.java | 4 +
.../org/apache/pinot/query/QueryEnvironment.java | 19 ++
.../apache/pinot/query/context/PlannerContext.java | 14 ++
.../query/ApproximateAggregateRewriteTest.java | 162 +++++++++++++++++
.../pinot/segment/spi/AggregationFunctionType.java | 23 ++-
.../apache/pinot/spi/utils/CommonConstants.java | 35 +++-
30 files changed, 1527 insertions(+), 90 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 826c732d4d5..ebdad1eaf06 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
@@ -651,6 +651,16 @@ 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.
+ // 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());
+ if (multiStageBrokerRequestHandler != null) {
+ _clusterConfigChangeHandler.registerClusterConfigChangeListener(
+
multiStageBrokerRequestHandler.getApproximateFunctionOverrideProvider());
+ }
_brokerRequestHandler.start();
String controllerUrl = _brokerConf.getProperty(Broker.CONTROLLER_URL);
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/querylog/QueryLogger.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/querylog/QueryLogger.java
index 6276ddeb265..e4be84010fa 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/querylog/QueryLogger.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/querylog/QueryLogger.java
@@ -436,6 +436,12 @@ public class QueryLogger {
builder.append(params._response.getRLSFiltersApplied());
}
},
+ APPROXIMATE_FUNCTION_APPLIED("approximateFunctionApplied") {
+ @Override
+ void doFormat(StringBuilder builder, QueryLogger logger, QueryLogParams
params) {
+ builder.append(params._response.isApproximateFunctionApplied());
+ }
+ },
WORKLOAD_NAME("workloadName") {
@Override
void doFormat(StringBuilder builder, QueryLogger logger, QueryLogParams
params) {
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ApproximateFunctionOverrideProvider.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ApproximateFunctionOverrideProvider.java
new file mode 100644
index 00000000000..fa94214e316
--- /dev/null
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ApproximateFunctionOverrideProvider.java
@@ -0,0 +1,161 @@
+/**
+ * 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.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Consumer;
+import javax.annotation.Nullable;
+import org.apache.pinot.common.request.Literal;
+import org.apache.pinot.common.request.context.ExpressionContext;
+import
org.apache.pinot.core.query.aggregation.function.DistinctCountSmartHLLAggregationFunction;
+import
org.apache.pinot.core.query.aggregation.function.PercentileSmartTDigestAggregationFunction;
+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 defaults for the approximate function rewrite, from the
broker conf and from the Helix cluster config.
+/// The cluster config wins where it sets a key, and it is applied live:
registering this as a
+/// [PinotClusterConfigChangeListener] means a change takes effect on the next
query without a broker restart.
+///
+/// Thread-safe. [#getSettings()] hands out one immutable snapshot, so a query
that reads it once cannot see a config
+/// change half applied.
+public class ApproximateFunctionOverrideProvider implements
PinotClusterConfigChangeListener {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(ApproximateFunctionOverrideProvider.class);
+
+ private static final ExpressionContext PROBE_COLUMN =
ExpressionContext.forIdentifier("__probe__");
+ private static final ExpressionContext PROBE_PERCENTILE =
ExpressionContext.forLiteral(Literal.doubleValue(50.0));
+
+ private final Settings _brokerConfSettings;
+ private volatile Settings _settings;
+
+ public ApproximateFunctionOverrideProvider(PinotConfiguration config) {
+ _brokerConfSettings = new Settings(
+ config.getProperty(Broker.USE_APPROXIMATE_FUNCTION,
Broker.DEFAULT_USE_APPROXIMATE_FUNCTION),
+
validated(config.getProperty(Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS,
+ Broker.DEFAULT_APPROXIMATE_FUNCTION_PARAMS),
Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS,
+ Broker.DEFAULT_APPROXIMATE_FUNCTION_PARAMS,
ApproximateFunctionOverrideProvider::probeDistinctCount),
+
validated(config.getProperty(Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
+ Broker.DEFAULT_APPROXIMATE_FUNCTION_PARAMS),
Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
+ Broker.DEFAULT_APPROXIMATE_FUNCTION_PARAMS,
ApproximateFunctionOverrideProvider::probePercentile));
+ _settings = _brokerConfSettings;
+ }
+
+ @Override
+ public void onChange(Set<String> changedConfigs, Map<String, String>
clusterConfigs) {
+ if (!changedConfigs.contains(Broker.USE_APPROXIMATE_FUNCTION)
+ &&
!changedConfigs.contains(Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS)
+ &&
!changedConfigs.contains(Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS)) {
+ return;
+ }
+ // A key the cluster config does not set, or that was removed from it,
falls back to the broker conf value. The
+ // fallback is the broker conf rather than the last good live value, so a
broker that restarts resolves the same.
+ Settings updated = new Settings(
+ validatedEnabled(clusterConfigs.get(Broker.USE_APPROXIMATE_FUNCTION)),
+
validated(clusterConfigs.get(Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS),
+ Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS,
_brokerConfSettings._distinctCountParams,
+ ApproximateFunctionOverrideProvider::probeDistinctCount),
+
validated(clusterConfigs.get(Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS),
+ Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
_brokerConfSettings._percentileParams,
+ ApproximateFunctionOverrideProvider::probePercentile));
+ _settings = updated;
+ LOGGER.info("Updated approximate function override: enabled: {},
distinctCountParams: '{}', percentileParams: '{}'",
+ updated._enabled, updated._distinctCountParams,
updated._percentileParams);
+ }
+
+ /// Returns the current defaults. Read it once per query, so that the
enabled flag and the parameters a query uses
+ /// come from the same version of the config.
+ public Settings getSettings() {
+ return _settings;
+ }
+
+ /// Anything that is not true or false keeps the broker conf value. Silently
reading a typo as `false` would turn
+ /// the guard rail off, which is the worse direction to fail in.
+ private boolean validatedEnabled(@Nullable String enabled) {
+ if (enabled == null) {
+ return _brokerConfSettings._enabled;
+ }
+ if ("true".equalsIgnoreCase(enabled.trim())) {
+ return true;
+ }
+ if ("false".equalsIgnoreCase(enabled.trim())) {
+ return false;
+ }
+ LOGGER.error("Ignoring invalid value '{}' for {}, falling back to '{}'",
enabled,
+ Broker.USE_APPROXIMATE_FUNCTION, _brokerConfSettings._enabled);
+ return _brokerConfSettings._enabled;
+ }
+
+ /// Parses the value with the aggregation function itself, so an unknown key
or a non-numeric value is caught here
+ /// instead of failing every rewritten query on the servers. A value the
parser accepts but the sketch rejects, such
+ /// as an out-of-range `log2m`, still fails on the servers.
+ ///
+ /// @param params the configured value, or `null` if unset
+ /// @param fallback the value to use if `params` is unset or invalid
+ private static String validated(@Nullable String params, String key, String
fallback, Consumer<String> probe) {
+ if (params == null) {
+ return fallback;
+ }
+ if (params.isEmpty()) {
+ return params;
+ }
+ try {
+ probe.accept(params);
+ return params;
+ } catch (Exception e) {
+ LOGGER.error("Ignoring invalid value '{}' for {}, falling back to '{}'",
params, key, fallback, e);
+ return fallback;
+ }
+ }
+
+ private static void probeDistinctCount(String params) {
+ new DistinctCountSmartHLLAggregationFunction(
+ List.of(PROBE_COLUMN,
ExpressionContext.forLiteral(Literal.stringValue(params))), false);
+ }
+
+ private static void probePercentile(String params) {
+ new PercentileSmartTDigestAggregationFunction(
+ List.of(PROBE_COLUMN, PROBE_PERCENTILE,
ExpressionContext.forLiteral(Literal.stringValue(params))), false);
+ }
+
+ /// The resolved defaults. Fields are read directly by the request handlers
in this package.
+ public static class Settings {
+ final boolean _enabled;
+ final String _distinctCountParams;
+ final String _percentileParams;
+
+ Settings(boolean enabled, String distinctCountParams, String
percentileParams) {
+ _enabled = enabled;
+ _distinctCountParams = distinctCountParams;
+ _percentileParams = percentileParams;
+ }
+
+ /// Applies the precedence query option > table config > cluster config >
broker conf. `null` means unset.
+ boolean isEnabled(@Nullable Boolean queryOptionOverride, @Nullable Boolean
tableConfigOverride) {
+ if (queryOptionOverride != null) {
+ return queryOptionOverride;
+ }
+ return tableConfigOverride != null ? tableConfigOverride : _enabled;
+ }
+ }
+}
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 605fcda0ce8..742cf545d02 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
@@ -106,6 +106,9 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
protected final Map<Long, String> _queriesById;
/// Maps broker-generated query id to client-provided query id.
protected final Map<Long, String> _clientQueryIds;
+ /// 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;
public BaseBrokerRequestHandler(PinotConfiguration config, String brokerId,
BrokerRequestIdGenerator requestIdGenerator, RoutingManager
routingManager,
@@ -120,6 +123,7 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
_tableCache = tableCache;
_threadAccountant = threadAccountant;
_multiClusterRoutingContext = multiClusterRoutingContext;
+ _approximateFunctionOverrideProvider = new
ApproximateFunctionOverrideProvider(config);
_brokerMetrics = BrokerMetrics.get();
_brokerQueryEventListener =
BrokerQueryEventListenerFactory.getBrokerQueryEventListener();
_trackedHeaders = BrokerQueryEventListenerFactory.getTrackedHeaders();
@@ -147,6 +151,12 @@ public abstract class BaseBrokerRequestHandler implements
BrokerRequestHandler {
}
}
+ /// Returns the provider that resolves the approximate-function rewrite
defaults. The broker starter registers it
+ /// with the cluster config change handler so that the defaults reload
without a restart.
+ public ApproximateFunctionOverrideProvider
getApproximateFunctionOverrideProvider() {
+ return _approximateFunctionOverrideProvider;
+ }
+
@Override
public BrokerResponse handleRequest(JsonNode request, @Nullable
SqlNodeAndOptions sqlNodeAndOptions,
@Nullable RequesterIdentity requesterIdentity, RequestContext
requestContext, @Nullable HttpHeaders httpHeaders)
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 b61abf35f08..a099861f951 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
@@ -160,7 +160,6 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
@Nullable
protected final MaterializedViewHandler _materializedViewHandler;
protected final boolean _disableGroovy;
- protected final boolean _useApproximateFunction;
protected final int _defaultHllLog2m;
protected final boolean _enableQueryLimitOverride;
protected final boolean _enableDistinctCountBitmapOverride;
@@ -200,7 +199,6 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
_materializedViewHandler = materializedViewHandler;
_serverAdminAuthProvider = AuthProviderUtils.extractAuthProvider(config,
Broker.SERVER_ADMIN_AUTH_PREFIX);
_disableGroovy = _config.getProperty(Broker.DISABLE_GROOVY,
Broker.DEFAULT_DISABLE_GROOVY);
- _useApproximateFunction =
_config.getProperty(Broker.USE_APPROXIMATE_FUNCTION, false);
_defaultHllLog2m =
_config.getProperty(CommonConstants.Helix.DEFAULT_HYPERLOGLOG_LOG2M_KEY,
CommonConstants.Helix.DEFAULT_HYPERLOGLOG_LOG2M);
_enableQueryLimitOverride =
_config.getProperty(Broker.CONFIG_OF_ENABLE_QUERY_LIMIT_OVERRIDE, false);
@@ -673,10 +671,16 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
QueryConfig realtimeTableQueryConfig =
routeInfo.getRealtimeTableQueryConfig();
TimeBoundaryInfo timeBoundaryInfo = routeInfo.getTimeBoundaryInfo();
- HandlerContext handlerContext = getHandlerContext(offlineTableQueryConfig,
realtimeTableQueryConfig);
+ HandlerContext handlerContext =
+ getHandlerContext(offlineTableQueryConfig, realtimeTableQueryConfig,
pinotQuery.getQueryOptions());
validateGroovyScript(serverPinotQuery, handlerContext._disableGroovy);
+ boolean approximateFunctionApplied = false;
if (handlerContext._useApproximateFunction) {
- handleApproximateFunctionOverride(serverPinotQuery);
+ approximateFunctionApplied =
handleApproximateFunctionOverride(serverPinotQuery,
+ handlerContext._distinctCountParams,
handlerContext._percentileParams);
+ if (approximateFunctionApplied) {
+ _brokerMetrics.addMeteredTableValue(rawTableName,
BrokerMeter.APPROXIMATE_FUNCTION_OVERRIDES, 1);
+ }
}
// Validate the request
@@ -1082,6 +1086,7 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
}
brokerResponse.setRLSFiltersApplied(rlsFiltersApplied.get());
+ brokerResponse.setApproximateFunctionApplied(approximateFunctionApplied);
// Record per-server stats on the SSE BrokerResponse so downstream
consumers can read it.
brokerResponse.setServerStats(serverStats.getServerStats());
@@ -1862,7 +1867,7 @@ public abstract class BaseSingleStageBrokerRequestHandler
extends BaseBrokerRequ
}
private HandlerContext getHandlerContext(@Nullable QueryConfig
offlineTableQueryConfig,
- @Nullable QueryConfig realtimeTableQueryConfig) {
+ @Nullable QueryConfig realtimeTableQueryConfig, @Nullable Map<String,
String> queryOptions) {
Boolean disableGroovyOverride = null;
Boolean useApproximateFunctionOverride = null;
if (offlineTableQueryConfig != null) {
@@ -1897,18 +1902,31 @@ public abstract class
BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ
}
boolean disableGroovy = disableGroovyOverride != null ?
disableGroovyOverride : _disableGroovy;
+ // Precedence: query option > table config > cluster config > broker conf.
One snapshot, so that the flag and the
+ // parameters this query uses come from the same version of the config.
+ Boolean queryOptionOverride =
+ queryOptions != null ?
QueryOptionsUtils.isUseApproximateFunction(queryOptions) : null;
+ ApproximateFunctionOverrideProvider.Settings approximateFunctionSettings =
+ _approximateFunctionOverrideProvider.getSettings();
boolean useApproximateFunction =
- useApproximateFunctionOverride != null ?
useApproximateFunctionOverride : _useApproximateFunction;
- return new HandlerContext(disableGroovy, useApproximateFunction);
+ approximateFunctionSettings.isEnabled(queryOptionOverride,
useApproximateFunctionOverride);
+ return new HandlerContext(disableGroovy, useApproximateFunction,
+ approximateFunctionSettings._distinctCountParams,
approximateFunctionSettings._percentileParams);
}
private static class HandlerContext {
final boolean _disableGroovy;
final boolean _useApproximateFunction;
+ /// Parameters for the rewritten calls, empty when the aggregation
function defaults apply.
+ final String _distinctCountParams;
+ final String _percentileParams;
- HandlerContext(boolean disableGroovy, boolean useApproximateFunction) {
+ HandlerContext(boolean disableGroovy, boolean useApproximateFunction,
String distinctCountParams,
+ String percentileParams) {
_disableGroovy = disableGroovy;
_useApproximateFunction = useApproximateFunction;
+ _distinctCountParams = distinctCountParams;
+ _percentileParams = percentileParams;
}
}
@@ -1972,57 +1990,79 @@ public abstract class
BaseSingleStageBrokerRequestHandler extends BaseBrokerRequ
/// Rewrites potential expensive functions to their approximation
counterparts.
/// - DISTINCT_COUNT -> DISTINCT_COUNT_SMART_HLL
/// - PERCENTILE -> PERCENTILE_SMART_TDIGEST
+ ///
+ /// The rewritten functions stay exact until an accumulator crosses their
conversion threshold, which the
+ /// parameters carry. An empty parameter string appends nothing, so the
aggregation function defaults apply.
+ ///
+ /// @return true if at least one function was rewritten, which makes the
result of the query approximate
@VisibleForTesting
- static void handleApproximateFunctionOverride(PinotQuery pinotQuery) {
+ static boolean handleApproximateFunctionOverride(PinotQuery pinotQuery,
String distinctCountParams,
+ String percentileParams) {
+ boolean applied = false;
for (Expression expression : pinotQuery.getSelectList()) {
- handleApproximateFunctionOverride(expression);
+ applied |= handleApproximateFunctionOverride(expression,
distinctCountParams, percentileParams);
}
List<Expression> orderByExpressions = pinotQuery.getOrderByList();
if (orderByExpressions != null) {
for (Expression expression : orderByExpressions) {
// NOTE: Order-by is always a Function with the ordering of the
Expression
-
handleApproximateFunctionOverride(expression.getFunctionCall().getOperands().get(0));
+ applied |=
handleApproximateFunctionOverride(expression.getFunctionCall().getOperands().get(0),
+ distinctCountParams, percentileParams);
}
}
Expression havingExpression = pinotQuery.getHavingExpression();
if (havingExpression != null) {
- handleApproximateFunctionOverride(havingExpression);
+ applied |= handleApproximateFunctionOverride(havingExpression,
distinctCountParams, percentileParams);
}
+ return applied;
}
- private static void handleApproximateFunctionOverride(Expression expression)
{
+ private static boolean handleApproximateFunctionOverride(Expression
expression, String distinctCountParams,
+ String percentileParams) {
Function function = expression.getFunctionCall();
if (function == null) {
- return;
+ return false;
}
String functionName = function.getOperator();
if (functionName.equals("distinctcount") ||
functionName.equals("distinctcountmv")) {
function.setOperator("distinctcountsmarthll");
+ appendParams(function, distinctCountParams, 1);
+ return true;
} else if (functionName.startsWith("percentile")) {
- String remainingFunctionName = functionName.substring(10);
- if (remainingFunctionName.isEmpty() ||
remainingFunctionName.equals("mv")) {
+ String suffix = functionName.substring(10);
+ if (suffix.isEmpty() || suffix.equals("mv")) {
function.setOperator("percentilesmarttdigest");
- } else if (remainingFunctionName.matches("\\d+")) {
+ } else if (suffix.matches("\\d+(mv)?")) {
+ // The percentile is in the function name, so it becomes an explicit
argument.
+ String digits = suffix.endsWith("mv") ? suffix.substring(0,
suffix.length() - 2) : suffix;
+ int percentile;
try {
- int percentile = Integer.parseInt(remainingFunctionName);
- function.setOperator("percentilesmarttdigest");
-
function.addToOperands(RequestUtils.getLiteralExpression(percentile));
- } catch (Exception e) {
- throw new BadQueryRequestException("Illegal function name: " +
functionName);
- }
- } else if (remainingFunctionName.matches("\\d+mv")) {
- try {
- int percentile = Integer.parseInt(remainingFunctionName.substring(0,
remainingFunctionName.length() - 2));
- function.setOperator("percentilesmarttdigest");
-
function.addToOperands(RequestUtils.getLiteralExpression(percentile));
+ percentile = Integer.parseInt(digits);
} catch (Exception e) {
throw new BadQueryRequestException("Illegal function name: " +
functionName);
}
+ function.setOperator("percentilesmarttdigest");
+ function.addToOperands(RequestUtils.getLiteralExpression(percentile));
+ } else {
+ // An already approximate variant, e.g. PERCENTILE_TDIGEST. Nothing to
rewrite.
+ return false;
}
+ appendParams(function, percentileParams, 2);
+ return true;
} else {
+ boolean applied = false;
for (Expression operand : function.getOperands()) {
- handleApproximateFunctionOverride(operand);
+ applied |= handleApproximateFunctionOverride(operand,
distinctCountParams, percentileParams);
}
+ return applied;
+ }
+ }
+
+ /// Appends the parameters as the trailing argument of a rewritten call,
unless the call has an unexpected arity,
+ /// in which case the function defaults apply rather than a call the server
cannot construct.
+ private static void appendParams(Function function, String params, int
expectedNumOperands) {
+ if (!params.isEmpty() && function.getOperandsSize() ==
expectedNumOperands) {
+ function.addToOperands(RequestUtils.getLiteralExpression(params));
}
}
diff --git
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
index 08005e3be51..299be308c2c 100644
---
a/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
+++
b/pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java
@@ -575,6 +575,12 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
CommonConstants.Broker.DEFAULT_UNNEST_COLUMN_PRUNING);
WorkerManager workerManager =
QueryOptionsUtils.isMultiClusterRoutingEnabled(queryOptions, false)
? _multiClusterWorkerManager : _workerManager;
+ // Unlike the single-stage engine there is no table-level layer here,
because a multi-stage query can span tables.
+ // Precedence is therefore query option > cluster config > broker conf.
+ ApproximateFunctionOverrideProvider.Settings approximateFunctionSettings =
+ _approximateFunctionOverrideProvider.getSettings();
+ boolean useApproximateFunction =
+
approximateFunctionSettings.isEnabled(QueryOptionsUtils.isUseApproximateFunction(queryOptions),
null);
return QueryEnvironment.configBuilder()
.requestId(requestId)
.database(database)
@@ -587,6 +593,9 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
.defaultUnnestColumnPruning(defaultUnnestColumnPruning)
.defaultUseLeafServerForIntermediateStage(defaultUseLeafServerForIntermediateStage)
.defaultEnableGroupTrim(defaultEnableGroupTrim)
+ .useApproximateFunction(useApproximateFunction)
+
.approximateFunctionDistinctCountParams(approximateFunctionSettings._distinctCountParams)
+
.approximateFunctionPercentileParams(approximateFunctionSettings._percentileParams)
.defaultEnableDynamicFilteringSemiJoin(defaultEnableDynamicFilteringSemiJoin)
.defaultUsePhysicalOptimizer(defaultUsePhysicalOptimizer)
.defaultUseLiteMode(defaultUseLiteMode)
@@ -908,6 +917,15 @@ public class MultiStageBrokerRequestHandler extends
BaseBrokerRequestHandler {
// set if rls (row level security) filters have been applied on the query
brokerResponse.setRLSFiltersApplied(rlsFiltersApplied);
+ // set if an exact aggregation was rewritten into its approximate
counterpart, which makes the result
+ // approximate rather than exact
+ if (query.getPlannerContext().isApproximateFunctionApplied()) {
+ brokerResponse.setApproximateFunctionApplied(true);
+ for (String tableName : tableNames) {
+ _brokerMetrics.addMeteredTableValue(tableName,
BrokerMeter.APPROXIMATE_FUNCTION_OVERRIDES, 1);
+ }
+ }
+
// Log query and stats
_queryLogger.logQueryCompleted(
new QueryLogger.QueryLogParams(requestContext,
tableNames.toString(), brokerResponse,
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/querylog/QueryLoggerTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/querylog/QueryLoggerTest.java
index 58dd30cc3e8..eb2f32d62de 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/querylog/QueryLoggerTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/querylog/QueryLoggerTest.java
@@ -137,6 +137,7 @@ public class QueryLoggerTest {
+ "realtimeMemAllocatedBytes(total/thread/resSer):0/0/0,"
+ "pools=[],"
+ "rlsFiltersApplied=true,"
+ + "approximateFunctionApplied=false,"
+ "workloadName=workloadName,"
+ "query=SELECT * FROM foo");
//@formatter:on
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ApproximateFunctionOverrideProviderTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ApproximateFunctionOverrideProviderTest.java
new file mode 100644
index 00000000000..9935b1b565b
--- /dev/null
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/ApproximateFunctionOverrideProviderTest.java
@@ -0,0 +1,134 @@
+/**
+ * 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 org.apache.pinot.spi.env.PinotConfiguration;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+public class ApproximateFunctionOverrideProviderTest {
+
+ @Test
+ public void testBrokerConfProvidesTheDefault() {
+ ApproximateFunctionOverrideProvider unset = newProvider(Map.of());
+ assertFalse(unset.getSettings().isEnabled(null, null));
+ assertEquals(unset.getSettings()._distinctCountParams, "");
+ assertEquals(unset.getSettings()._percentileParams, "");
+
+ ApproximateFunctionOverrideProvider provider = newProvider(Map.of(
+ Broker.USE_APPROXIMATE_FUNCTION, "true",
+ Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS,
"threshold=10;log2m=8",
+ Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
"threshold=20;compression=50"));
+ assertTrue(provider.getSettings().isEnabled(null, null));
+ assertEquals(provider.getSettings()._distinctCountParams,
"threshold=10;log2m=8");
+ assertEquals(provider.getSettings()._percentileParams,
"threshold=20;compression=50");
+ }
+
+ @Test
+ public void testPrecedenceIsQueryOptionThenTableThenDefault() {
+ ApproximateFunctionOverrideProvider disabledByDefault =
newProvider(Map.of());
+ assertTrue(disabledByDefault.getSettings().isEnabled(Boolean.TRUE, null));
+ assertTrue(disabledByDefault.getSettings().isEnabled(null, Boolean.TRUE));
+ // The query option outranks the table config in both directions.
+ assertFalse(disabledByDefault.getSettings().isEnabled(Boolean.FALSE,
Boolean.TRUE));
+
+ ApproximateFunctionOverrideProvider enabledByDefault =
+ newProvider(Map.of(Broker.USE_APPROXIMATE_FUNCTION, "true"));
+ assertFalse(enabledByDefault.getSettings().isEnabled(Boolean.FALSE, null));
+ assertFalse(enabledByDefault.getSettings().isEnabled(null, Boolean.FALSE));
+ assertTrue(enabledByDefault.getSettings().isEnabled(Boolean.TRUE,
Boolean.FALSE));
+ }
+
+ @Test
+ public void testClusterConfigWinsOverBrokerConfAndReloads() {
+ ApproximateFunctionOverrideProvider provider = newProvider(Map.of(
+ Broker.USE_APPROXIMATE_FUNCTION, "false",
+ Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS, "threshold=20"));
+
+ Map<String, String> clusterConfigs = Map.of(
+ Broker.USE_APPROXIMATE_FUNCTION, "true",
+ Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
"threshold=5;compression=200");
+ provider.onChange(clusterConfigs.keySet(), clusterConfigs);
+
+ assertTrue(provider.getSettings().isEnabled(null, null));
+ assertEquals(provider.getSettings()._percentileParams,
"threshold=5;compression=200");
+
+ // Removing the keys from the cluster config falls back to the broker conf.
+ provider.onChange(clusterConfigs.keySet(), Map.of());
+ assertFalse(provider.getSettings().isEnabled(null, null));
+ assertEquals(provider.getSettings()._percentileParams, "threshold=20");
+ }
+
+ @Test
+ public void testUnrelatedClusterConfigChangeIsIgnored() {
+ ApproximateFunctionOverrideProvider provider =
+ newProvider(Map.of(Broker.USE_APPROXIMATE_FUNCTION, "true"));
+ provider.onChange(Map.of("some.other.key", "1").keySet(),
Map.of("some.other.key", "1"));
+ assertTrue(provider.getSettings().isEnabled(null, null));
+ }
+
+ /// A typo in the cluster config would otherwise fail every rewritten query
on the servers, so a bad value is
+ /// rejected in favour of the broker conf value. Falling back to the broker
conf rather than to the last good live
+ /// value is what makes a broker that restarts resolve the same value as one
that stayed up.
+ @Test
+ public void testInvalidParamsAreRejectedInFavourOfTheBrokerConfValue() {
+ ApproximateFunctionOverrideProvider provider = newProvider(Map.of(
+ Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS, "threshold=10",
+ Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS, "threshold=10"));
+
+ onChange(provider, Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS,
"thresold=10");
+ assertEquals(provider.getSettings()._distinctCountParams, "threshold=10");
+
+ onChange(provider, Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
"threshold=abc");
+ assertEquals(provider.getSettings()._percentileParams, "threshold=10");
+
+ // The percentile parameters must not be accepted for distinct count, and
the other way round.
+ onChange(provider, Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS,
"compression=50");
+ assertEquals(provider.getSettings()._distinctCountParams, "threshold=10");
+
+ // A good live value does not become the fallback for a later bad one.
+ onChange(provider, Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
"threshold=500");
+ assertEquals(provider.getSettings()._percentileParams, "threshold=500");
+ onChange(provider, Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
"threshold=nope");
+ assertEquals(provider.getSettings()._percentileParams, "threshold=10");
+ }
+
+ /// A broker conf that is invalid at startup must not leave the broker
emitting calls the servers reject.
+ @Test
+ public void testInvalidBrokerConfParamsFallBackToTheFunctionDefaults() {
+ ApproximateFunctionOverrideProvider provider =
+ newProvider(Map.of(Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS,
"threshold=10;log2m=8"));
+ assertEquals(provider.getSettings()._percentileParams, "");
+ }
+
+ private static void onChange(ApproximateFunctionOverrideProvider provider,
String key, String value) {
+ Map<String, String> clusterConfigs = Map.of(key, value);
+ provider.onChange(clusterConfigs.keySet(), clusterConfigs);
+ }
+
+ private static ApproximateFunctionOverrideProvider newProvider(Map<String,
Object> properties) {
+ return new ApproximateFunctionOverrideProvider(new
PinotConfiguration(properties));
+ }
+}
diff --git
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryOverrideTest.java
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryOverrideTest.java
index 6647bd20ff5..b60107c65b6 100644
---
a/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryOverrideTest.java
+++
b/pinot-broker/src/test/java/org/apache/pinot/broker/requesthandler/QueryOverrideTest.java
@@ -20,12 +20,15 @@ package org.apache.pinot.broker.requesthandler;
import com.google.common.collect.ImmutableSet;
import java.util.Arrays;
+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.sql.parsers.CalciteSqlParser;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
public class QueryOverrideTest {
@@ -70,7 +73,7 @@ public class QueryOverrideTest {
String query = "SELECT DISTINCT_COUNT(col1) FROM myTable GROUP BY col2
HAVING DISTINCT_COUNT(col1) > 10 "
+ "ORDER BY DISTINCT_COUNT(col1) DESC";
PinotQuery pinotQuery = CalciteSqlParser.compileToPinotQuery(query);
-
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery);
+
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"", "");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperator(),
"distinctcountsmarthll");
assertEquals(
pinotQuery.getOrderByList().get(0).getFunctionCall().getOperands().get(0).getFunctionCall().getOperator(),
@@ -81,22 +84,22 @@ public class QueryOverrideTest {
query = "SELECT DISTINCT_COUNT_MV(col1) FROM myTable";
pinotQuery = CalciteSqlParser.compileToPinotQuery(query);
-
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery);
+
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"", "");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperator(),
"distinctcountsmarthll");
query = "SELECT DISTINCT col1 FROM myTable";
pinotQuery = CalciteSqlParser.compileToPinotQuery(query);
-
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery);
+
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"", "");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperator(),
"distinct");
query = "SELECT DISTINCT_COUNT_HLL(col1) FROM myTable";
pinotQuery = CalciteSqlParser.compileToPinotQuery(query);
-
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery);
+
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"", "");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperator(),
"distinctcounthll");
query = "SELECT DISTINCT_COUNT_BITMAP(col1) FROM myTable";
pinotQuery = CalciteSqlParser.compileToPinotQuery(query);
-
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery);
+
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"", "");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperator(),
"distinctcountbitmap");
}
@@ -104,7 +107,7 @@ public class QueryOverrideTest {
"SELECT PERCENTILE_MV(col1, 95) FROM myTable", "SELECT
PERCENTILE95(col1) FROM myTable",
"SELECT PERCENTILE95MV(col1) FROM myTable")) {
PinotQuery pinotQuery = CalciteSqlParser.compileToPinotQuery(query);
-
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery);
+
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"", "");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperator(),
"percentilesmarttdigest");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperands().get(1),
RequestUtils.getLiteralExpression(95));
@@ -112,13 +115,50 @@ public class QueryOverrideTest {
{
String query = "SELECT PERCENTILE_TDIGEST(col1, 95) FROM myTable";
PinotQuery pinotQuery = CalciteSqlParser.compileToPinotQuery(query);
-
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery);
+
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"", "");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperator(),
"percentiletdigest");
query = "SELECT PERCENTILE_EST(col1, 95) FROM myTable";
pinotQuery = CalciteSqlParser.compileToPinotQuery(query);
-
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery);
+
BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"", "");
assertEquals(pinotQuery.getSelectList().get(0).getFunctionCall().getOperator(),
"percentileest");
}
}
+
+ @Test
+ public void testApproximateFunctionOverrideReportsWhetherItRewroteAnything()
{
+ PinotQuery rewritten = CalciteSqlParser.compileToPinotQuery("SELECT
DISTINCT_COUNT(col1) FROM myTable");
+
assertTrue(BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(rewritten,
"", ""));
+
+ PinotQuery untouched = CalciteSqlParser.compileToPinotQuery("SELECT
SUM(col1) FROM myTable");
+
assertFalse(BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(untouched,
"", ""));
+
+ // An already approximate variant is left alone, so the query keeps
reporting exact results.
+ PinotQuery approximate = CalciteSqlParser.compileToPinotQuery("SELECT
PERCENTILE_TDIGEST(col1, 95) FROM myTable");
+
assertFalse(BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(approximate,
"", ""));
+ }
+
+ @Test
+ public void testApproximateFunctionOverrideAppendsConfiguredParams() {
+ PinotQuery pinotQuery = CalciteSqlParser.compileToPinotQuery("SELECT
DISTINCT_COUNT(col1) FROM myTable");
+
assertTrue(BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(pinotQuery,
"threshold=10", ""));
+ Function distinctCount =
pinotQuery.getSelectList().get(0).getFunctionCall();
+ assertEquals(distinctCount.getOperator(), "distinctcountsmarthll");
+ assertEquals(distinctCount.getOperands().size(), 2);
+ assertEquals(distinctCount.getOperands().get(1),
RequestUtils.getLiteralExpression("threshold=10"));
+
+ // The percentile parameters go after the percentile itself, whichever
spelling the query used.
+ for (String query : Arrays.asList("SELECT PERCENTILE(col1, 95) FROM
myTable",
+ "SELECT PERCENTILE95(col1) FROM myTable")) {
+ PinotQuery percentileQuery = CalciteSqlParser.compileToPinotQuery(query);
+
assertTrue(BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride(percentileQuery,
"",
+ "threshold=20;compression=50"));
+ Function percentile =
percentileQuery.getSelectList().get(0).getFunctionCall();
+ assertEquals(percentile.getOperator(), "percentilesmarttdigest");
+ assertEquals(percentile.getOperands().size(), 3);
+ assertEquals(percentile.getOperands().get(1),
RequestUtils.getLiteralExpression(95));
+ assertEquals(percentile.getOperands().get(2),
+ RequestUtils.getLiteralExpression("threshold=20;compression=50"));
+ }
+ }
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
index 18fa4545f91..bc2a8b354e9 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/metrics/BrokerMeter.java
@@ -42,6 +42,11 @@ public class BrokerMeter implements AbstractMetrics.Meter {
///
/// At this moment this counter does not include queries executed in
multi-stage mode.
public static final BrokerMeter QUERIES = create("QUERIES", "queries",
false);
+
+ /// Queries in which the broker rewrote at least one exact aggregation into
its approximate counterpart, because
+ /// `pinot.broker.use.approximate.function` was on. The results of those
queries are approximate.
+ public static final BrokerMeter APPROXIMATE_FUNCTION_OVERRIDES =
+ create("APPROXIMATE_FUNCTION_OVERRIDES", "queries", false);
/// Number of single-stage queries that have been started.
///
/// Unlike [#QUERIES], this metric is global and not attached to a
particular table.
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
index 83e2d945dfc..98362b1b4e2 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java
@@ -311,6 +311,19 @@ public interface BrokerResponse {
/// @return true if RLS filters were applied, false otherwise
boolean getRLSFiltersApplied();
+ /// Set whether the broker rewrote an exact aggregation into its approximate
counterpart, for example
+ /// `DISTINCT_COUNT` into `DISTINCT_COUNT_SMART_HLL`. The result is then
approximate, not exact.
+ /// The default is a no-op so that implementations which do not track this
need no change.
+ /// @param approximateFunctionApplied true if at least one function was
rewritten
+ default void setApproximateFunctionApplied(boolean
approximateFunctionApplied) {
+ }
+
+ /// Get whether the broker rewrote an exact aggregation into its approximate
counterpart.
+ /// @return true if at least one function was rewritten, false otherwise
+ default boolean isApproximateFunctionApplied() {
+ return false;
+ }
+
/// Get the materialized view table name that was hit (used) for this query,
or `null`
/// if no materialized view was used. The default returns `null` so impls
that do not track MV
/// rewrite (e.g. MSE response paths) need no explicit override; the
matching *setter* is
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
index a414f565934..b38bae56605 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNative.java
@@ -42,7 +42,8 @@ import org.apache.pinot.spi.utils.JsonUtils;
/// This class can be used to serialize/deserialize the broker response.
@JsonPropertyOrder({
"resultTable", "numRowsResultSet", "partialResult", "exceptions",
"numGroupsLimitReached",
- "numGroupsWarningLimitReached", "maxRowsInDistinctReached",
"maxRowsWithoutChangeInDistinctReached",
+ "numGroupsWarningLimitReached", "maxRowsInDistinctReached",
+ "maxRowsWithoutChangeInDistinctReached",
"maxExecutionTimeInDistinctReached", "timeUsedMs",
"requestId", "clientRequestId", "brokerId", "numDocsScanned", "totalDocs",
"numEntriesScannedInFilter",
@@ -56,7 +57,8 @@ import org.apache.pinot.spi.utils.JsonUtils;
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tablesQueried",
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
"realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes",
- "pools", "rlsFiltersApplied", "groupsTrimmed", "materializedViewQueried",
"serverStats"
+ "pools", "rlsFiltersApplied", "approximateFunctionApplied",
"groupsTrimmed", "materializedViewQueried",
+ "serverStats"
})
@JsonIgnoreProperties(ignoreUnknown = true)
public class BrokerResponseNative implements BrokerResponse {
@@ -120,6 +122,7 @@ public class BrokerResponseNative implements BrokerResponse
{
private Set<Integer> _pools = Set.of();
private boolean _rlsFiltersApplied = false;
+ private boolean _approximateFunctionApplied = false;
@Nullable
private String _materializedViewQueried;
@@ -636,6 +639,18 @@ public class BrokerResponseNative implements
BrokerResponse {
return _rlsFiltersApplied;
}
+ @JsonProperty("approximateFunctionApplied")
+ @Override
+ public void setApproximateFunctionApplied(boolean
approximateFunctionApplied) {
+ _approximateFunctionApplied = approximateFunctionApplied;
+ }
+
+ @JsonProperty("approximateFunctionApplied")
+ @Override
+ public boolean isApproximateFunctionApplied() {
+ return _approximateFunctionApplied;
+ }
+
@JsonProperty("materializedViewQueried")
public void setMaterializedViewQueried(@Nullable String
materializedViewQueried) {
_materializedViewQueried = materializedViewQueried;
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
index efbb57ead5a..9919057b565 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/BrokerResponseNativeV2.java
@@ -40,7 +40,8 @@ import org.apache.pinot.common.response.ProcessingException;
/// TODO: Currently this class cannot be used to deserialize the JSON response.
@JsonPropertyOrder({
"resultTable", "numRowsResultSet", "partialResult", "exceptions",
"numGroupsLimitReached",
- "numGroupsWarningLimitReached", "numGroups", "earlyTerminationReasons",
"maxRowsInJoinReached",
+ "numGroupsWarningLimitReached", "numGroups", "earlyTerminationReasons",
+ "maxRowsInJoinReached",
"maxRowsInJoin", "maxRowsInWindowReached", "maxRowsInWindow",
"timeUsedMs", "stageStats", "streamStatsCoverage",
"maxRowsInOperator", "requestId", "clientRequestId", "brokerId",
"numDocsScanned", "totalDocs",
"numEntriesScannedInFilter", "numEntriesScannedPostFilter",
"numServersQueried", "numServersResponded",
@@ -53,7 +54,7 @@ import org.apache.pinot.common.response.ProcessingException;
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tablesQueried",
"offlineThreadMemAllocatedBytes", "realtimeThreadMemAllocatedBytes",
"offlineResponseSerMemAllocatedBytes",
"realtimeResponseSerMemAllocatedBytes", "offlineTotalMemAllocatedBytes",
"realtimeTotalMemAllocatedBytes",
- "pools", "rlsFiltersApplied", "groupsTrimmed",
+ "pools", "rlsFiltersApplied", "approximateFunctionApplied",
"groupsTrimmed",
"mseLiteLeafStageLimitReached", "mseLiteLeafStageEffectiveLimit",
"mseLiteFanOutAdjustedLimitApplied",
"responseMetadata"
})
@@ -91,6 +92,7 @@ public class BrokerResponseNativeV2 implements BrokerResponse
{
private Set<Integer> _pools = Set.of();
private boolean _rlsFiltersApplied = false;
+ private boolean _approximateFunctionApplied = false;
@Nullable
private Integer _mseLiteLeafStageEffectiveLimit;
@Nullable
@@ -520,6 +522,18 @@ public class BrokerResponseNativeV2 implements
BrokerResponse {
return _rlsFiltersApplied;
}
+ @JsonProperty("approximateFunctionApplied")
+ @Override
+ public void setApproximateFunctionApplied(boolean
approximateFunctionApplied) {
+ _approximateFunctionApplied = approximateFunctionApplied;
+ }
+
+ @JsonProperty("approximateFunctionApplied")
+ @Override
+ public boolean isApproximateFunctionApplied() {
+ return _approximateFunctionApplied;
+ }
+
public void addBrokerStats(StatMap<StatKey> brokerStats) {
_brokerStats.merge(brokerStats);
}
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
index e1ba761767d..96565b68cad 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/response/broker/CursorResponseNative.java
@@ -25,7 +25,8 @@ import org.apache.pinot.common.response.CursorResponse;
@JsonPropertyOrder({
"resultTable", "numRowsResultSet", "partialResult", "exceptions",
"numGroupsLimitReached",
- "numGroupsWarningLimitReached", "timeUsedMs", "requestId", "brokerId",
"numDocsScanned", "totalDocs",
+ "numGroupsWarningLimitReached", "timeUsedMs", "requestId", "brokerId",
+ "numDocsScanned", "totalDocs",
"numEntriesScannedInFilter", "numEntriesScannedPostFilter",
"numServersQueried", "numServersResponded",
"numSegmentsQueried", "numSegmentsProcessed", "numSegmentsMatched",
"numConsumingSegmentsQueried",
"numConsumingSegmentsProcessed", "numConsumingSegmentsMatched",
"minConsumingFreshnessTimeMs",
@@ -34,7 +35,7 @@ import org.apache.pinot.common.response.CursorResponse;
"offlineSystemActivitiesCpuTimeNs", "realtimeSystemActivitiesCpuTimeNs",
"offlineResponseSerializationCpuTimeNs",
"realtimeResponseSerializationCpuTimeNs", "offlineTotalCpuTimeNs",
"realtimeTotalCpuTimeNs",
"explainPlanNumEmptyFilterSegments",
"explainPlanNumMatchAllFilterSegments", "traceInfo", "tableQueries",
- "groupsTrimmed",
+ "groupsTrimmed", "approximateFunctionApplied",
// Fields specific to CursorResponse
"offset", "numRows", "cursorResultWriteTimeMs", "cursorFetchTimeMs",
"submissionTimeMs", "expirationTimeMs",
"brokerHost", "brokerPort", "bytesWritten"
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 a48c88cd146..8a9bdb8639b 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
@@ -437,6 +437,14 @@ public class QueryOptionsUtils {
return new HashSet<>(List.of(useRules));
}
+ /// Returns the per-query override of the approximate-function rewrite, or
`null` if the query does not set one, in
+ /// which case the table config and then the cluster/broker default decide.
+ @Nullable
+ public static Boolean isUseApproximateFunction(Map<String, String>
queryOptions) {
+ return checkedParseBooleanNullable(QueryOptionKey.USE_APPROXIMATE_FUNCTION,
+ queryOptions.get(QueryOptionKey.USE_APPROXIMATE_FUNCTION));
+ }
+
@Nullable
public static Boolean isUseFixedReplica(Map<String, String> queryOptions) {
String useFixedReplica =
queryOptions.get(CommonConstants.Broker.Request.QueryOptionKey.USE_FIXED_REPLICA);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java
index fd3fbe0bbc8..181fc11297b 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java
@@ -61,6 +61,10 @@ abstract class
BaseDistinctCountSmartSketchAggregationFunction
protected abstract Object convertSetToSketch(Set valueSet, DataType
storedType);
+ /// Adds every value of the set into a sketch that [#convertSetToSketch]
already created. Used to fold the values
+ /// that arrive after a group has been converted, without building a second
sketch.
+ protected abstract void addSetToSketch(Object sketch, Set valueSet, DataType
storedType);
+
protected abstract Object convertToSketch(DictIdsWrapper dictIdsWrapper);
protected abstract IllegalStateException
getIllegalDataTypeException(DataType dataType, boolean singleValue);
@@ -586,9 +590,16 @@ abstract class
BaseDistinctCountSmartSketchAggregationFunction
if (result instanceof DictIdsWrapper) {
return convertToValueSet((DictIdsWrapper) result);
- } else {
- return result;
}
+ if (result instanceof SketchWithPendingValues) {
+ SketchWithPendingValues converted = (SketchWithPendingValues) result;
+ if (!converted._pendingValues.isEmpty()) {
+ addSetToSketch(converted._sketch, converted._pendingValues,
converted._storedType);
+ converted._pendingValues.clear();
+ }
+ return converted._sketch;
+ }
+ return result;
}
/// Returns the dictionary id bitmap from the result holder or creates a new
one if it does not exist.
@@ -643,14 +654,38 @@ abstract class
BaseDistinctCountSmartSketchAggregationFunction
return dictIdsWrapper._dictIdBitmap;
}
- /// Returns the value set for the given group key or creates a new one if it
does not exist.
- protected static Set getValueSet(GroupByResultHolder groupByResultHolder,
int groupKey, DataType valueType) {
- Set valueSet = groupByResultHolder.getResult(groupKey);
- if (valueSet == null) {
- valueSet = getValueSet(valueType);
+ /// Returns the value set for the given group key, creating one if it does
not exist, and converting the group to a
+ /// sketch once its set has grown past the threshold. A group that already
converted returns its pending set, which
+ /// this folds into the sketch. Without any of this the per-group sets of a
non-dictionary column grow for the whole
+ /// segment, because the only other check runs at merge time.
+ ///
+ /// The check runs here, on a set that has just been fetched anyway, rather
than as a second pass over the batch. It
+ /// therefore converts on the first touch after the threshold is crossed,
which bounds a group at the threshold plus
+ /// one batch of values.
+ protected final Set getValueSet(GroupByResultHolder groupByResultHolder, int
groupKey, DataType valueType) {
+ Object result = groupByResultHolder.getResult(groupKey);
+ if (result == null) {
+ Set valueSet = getValueSet(valueType);
groupByResultHolder.setValueForKey(groupKey, valueSet);
+ return valueSet;
}
- return valueSet;
+ int threshold = getThreshold();
+ if (result instanceof SketchWithPendingValues) {
+ SketchWithPendingValues converted = (SketchWithPendingValues) result;
+ if (converted._pendingValues.size() > threshold) {
+ addSetToSketch(converted._sketch, converted._pendingValues, valueType);
+ converted._pendingValues.clear();
+ }
+ return converted._pendingValues;
+ }
+ Set valueSet = (Set) result;
+ if (valueSet.size() <= threshold) {
+ return valueSet;
+ }
+ SketchWithPendingValues converted =
+ new SketchWithPendingValues(convertSetToSketch(valueSet, valueType),
getValueSet(valueType), valueType);
+ groupByResultHolder.setValueForKey(groupKey, converted);
+ return converted._pendingValues;
}
/// Helper method to set dictionary id for the given group keys into the
result holder.
@@ -662,42 +697,42 @@ abstract class
BaseDistinctCountSmartSketchAggregationFunction
}
/// Helper method to set INT value for the given group keys into the result
holder.
- protected static void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, int value) {
+ protected final void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, int value) {
for (int groupKey : groupKeys) {
((IntOpenHashSet) getValueSet(groupByResultHolder, groupKey,
DataType.INT)).add(value);
}
}
/// Helper method to set LONG value for the given group keys into the result
holder.
- protected static void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, long value) {
+ protected final void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, long value) {
for (int groupKey : groupKeys) {
((LongOpenHashSet) getValueSet(groupByResultHolder, groupKey,
DataType.LONG)).add(value);
}
}
/// Helper method to set FLOAT value for the given group keys into the
result holder.
- protected static void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, float value) {
+ protected final void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, float value) {
for (int groupKey : groupKeys) {
((FloatOpenHashSet) getValueSet(groupByResultHolder, groupKey,
DataType.FLOAT)).add(value);
}
}
/// Helper method to set DOUBLE value for the given group keys into the
result holder.
- protected static void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, double value) {
+ protected final void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, double value) {
for (int groupKey : groupKeys) {
((DoubleOpenHashSet) getValueSet(groupByResultHolder, groupKey,
DataType.DOUBLE)).add(value);
}
}
/// Helper method to set STRING value for the given group keys into the
result holder.
- protected static void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, String value) {
+ protected final void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys, String value) {
for (int groupKey : groupKeys) {
((ObjectOpenHashSet<String>) getValueSet(groupByResultHolder, groupKey,
DataType.STRING)).add(value);
}
}
/// Helper method to set BYTES value for the given group keys into the
result holder.
- protected static void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys,
+ protected final void setValueForGroupKeys(GroupByResultHolder
groupByResultHolder, int[] groupKeys,
ByteArray value) {
for (int groupKey : groupKeys) {
((ObjectOpenHashSet<ByteArray>) getValueSet(groupByResultHolder,
groupKey, DataType.BYTES)).add(value);
@@ -810,6 +845,20 @@ abstract class
BaseDistinctCountSmartSketchAggregationFunction
}
}
+ /// Per-group state after a group converted to a sketch. New values keep
landing in a plain set, which the next
+ /// threshold check folds into the sketch, so the typed aggregation loops
never see a sketch.
+ private static final class SketchWithPendingValues {
+ final Object _sketch;
+ final DataType _storedType;
+ final Set _pendingValues;
+
+ SketchWithPendingValues(Object sketch, Set pendingValues, DataType
storedType) {
+ _sketch = sketch;
+ _pendingValues = pendingValues;
+ _storedType = storedType;
+ }
+ }
+
/// Wrapper of dictionary and dict-id bitmap used during aggregation.
protected static final class DictIdsWrapper {
final Dictionary _dictionary;
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunction.java
index e3ffccf2cc7..d40686e8d2f 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunction.java
@@ -44,8 +44,9 @@ import org.roaringbitmap.RoaringBitmap;
/// The `DistinctCountSmartHLLAggregationFunction` calculates the number of
distinct values for a given expression
/// (both single-valued and multi-valued are supported).
///
-/// For aggregation-only queries, the distinct values are stored in a Set
initially. Once the number of distinct values
-/// exceeds a threshold, the Set will be converted into a HyperLogLog, and
approximate result will be returned.
+/// The distinct values are stored in a Set initially. Once the number of
distinct values exceeds a threshold, the
+/// Set will be converted into a HyperLogLog, and approximate result will be
returned. The threshold is applied per
+/// accumulator, which means per group for a group-by query.
///
/// The function takes an optional second argument for parameters:
/// - threshold: Threshold of the number of distinct values to trigger the
conversion, 100_000 by default. Non-positive
@@ -425,6 +426,20 @@ public class DistinctCountSmartHLLAggregationFunction
extends BaseDistinctCountS
return hyperLogLog;
}
+ @Override
+ protected void addSetToSketch(Object sketch, Set valueSet, DataType
storedType) {
+ HyperLogLog hll = (HyperLogLog) sketch;
+ if (storedType == DataType.BYTES) {
+ for (Object value : valueSet) {
+ hll.offer(((ByteArray) value).getBytes());
+ }
+ } else {
+ for (Object value : valueSet) {
+ hll.offer(value);
+ }
+ }
+ }
+
@Override
protected Object convertSetToSketch(Set valueSet, DataType storedType) {
return convertSetToHLL(valueSet, storedType);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
index f974c54f13a..a2d11c9351e 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLPlusAggregationFunction.java
@@ -44,8 +44,9 @@ import org.roaringbitmap.RoaringBitmap;
/// The `DistinctCountSmartHLLPlusAggregationFunction` calculates the number
of distinct values for a given
/// expression (both single-valued and multi-valued are supported).
///
-/// For aggregation-only queries, the distinct values are stored in a Set
initially. Once the number of distinct values
-/// exceeds a threshold, the Set will be converted into a HyperLogLogPlus, and
approximate result will be returned.
+/// The distinct values are stored in a Set initially. Once the number of
distinct values exceeds a threshold, the
+/// Set will be converted into a HyperLogLogPlus, and approximate result will
be returned. The threshold is applied per
+/// accumulator, which means per group for a group-by query.
///
/// The function takes an optional second argument for parameters:
/// - threshold: Threshold of the number of distinct values to trigger the
conversion, 100_000 by default. Non-positive
@@ -373,6 +374,20 @@ public class DistinctCountSmartHLLPlusAggregationFunction
extends BaseDistinctCo
return hllPlus;
}
+ @Override
+ protected void addSetToSketch(Object sketch, Set valueSet, DataType
storedType) {
+ HyperLogLogPlus hllPlus = (HyperLogLogPlus) sketch;
+ if (storedType == DataType.BYTES) {
+ for (Object value : valueSet) {
+ hllPlus.offer(((ByteArray) value).getBytes());
+ }
+ } else {
+ for (Object value : valueSet) {
+ hllPlus.offer(value);
+ }
+ }
+ }
+
@Override
protected Object convertSetToSketch(Set valueSet, DataType storedType) {
return convertSetToHLLPlus(valueSet, storedType);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartULLAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartULLAggregationFunction.java
index 23243762790..a3634cf4006 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartULLAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartULLAggregationFunction.java
@@ -44,8 +44,9 @@ import org.roaringbitmap.RoaringBitmap;
/// The `DistinctCountSmartULLAggregationFunction` calculates the number of
distinct values for a given expression
/// (both single-valued and multi-valued are supported).
///
-/// For aggregation-only queries, the distinct values are stored in a Set
initially. Once the number of distinct values
-/// exceeds a threshold, the Set will be converted into an UltraLogLog, and
approximate result will be returned.
+/// The distinct values are stored in a Set initially. Once the number of
distinct values exceeds a threshold, the
+/// Set will be converted into an UltraLogLog, and approximate result will be
returned. The threshold is applied per
+/// accumulator, which means per group for a group-by query.
///
/// The function takes an optional second argument for parameters:
/// - threshold: Threshold of the number of distinct values to trigger the
conversion, 100_000 by default. Non-positive
@@ -446,6 +447,20 @@ public class DistinctCountSmartULLAggregationFunction
extends BaseDistinctCountS
// threshold accessor for base class is provided by getThreshold()
+ @Override
+ protected void addSetToSketch(Object sketch, Set valueSet, DataType
storedType) {
+ UltraLogLog ull = (UltraLogLog) sketch;
+ if (storedType == DataType.BYTES) {
+ for (Object value : valueSet) {
+ UltraLogLogUtils.hashObject(((ByteArray)
value).getBytes()).ifPresent(ull::add);
+ }
+ } else {
+ for (Object value : valueSet) {
+ UltraLogLogUtils.hashObject(value).ifPresent(ull::add);
+ }
+ }
+ }
+
@Override
protected Object convertSetToSketch(Set valueSet, DataType storedType) {
return convertSetToULL(valueSet, storedType);
diff --git
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java
index 6ada67ef34f..062c1ca9252 100644
---
a/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java
+++
b/pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunction.java
@@ -43,9 +43,9 @@ import org.apache.pinot.spi.data.FieldSpec.DataType;
/// The `PercentileSmartTDigestAggregationFunction` calculates the percentile
of the values for a given
/// expression (both single-valued and multi-valued are supported).
///
-/// For aggregation-only queries, the values are stored in a [DoubleArrayList]
initially. Once the number of
-/// values exceeds a threshold, the list will be converted into a [TDigest],
and approximate result will be
-/// returned.
+/// The values are stored in a [DoubleArrayList] initially. Once the number of
values exceeds a threshold, the list
+/// will be converted into a [TDigest], and approximate result will be
returned. The threshold is applied per
+/// accumulator, which means per group for a group-by query.
///
/// The function takes an optional third argument for parameters:
/// - threshold: Threshold of the number of values to trigger the conversion,
100_000 by default. Non-positive value
@@ -201,30 +201,19 @@ public class PercentileSmartTDigestAggregationFunction
extends BaseSingleInputAg
double[] doubleValues = blockValSet.getDoubleValuesSV();
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- DoubleArrayList valueList = getValueList(groupByResultHolder,
groupKeyArray[i]);
- valueList.add(doubleValues[i]);
+ addValueForGroup(groupByResultHolder, groupKeyArray[i],
doubleValues[i], _threshold);
}
});
} else {
double[][] doubleValues = blockValSet.getDoubleValuesMV();
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
- DoubleArrayList valueList = getValueList(groupByResultHolder,
groupKeyArray[i]);
- valueList.addElements(valueList.size(), doubleValues[i]);
+ addValuesForGroup(groupByResultHolder, groupKeyArray[i],
doubleValues[i], _threshold);
}
});
}
}
- private static DoubleArrayList getValueList(GroupByResultHolder
groupByResultHolder, int groupKey) {
- DoubleArrayList valueList = groupByResultHolder.getResult(groupKey);
- if (valueList == null) {
- valueList = new DoubleArrayList();
- groupByResultHolder.setValueForKey(groupKey, valueList);
- }
- return valueList;
- }
-
@Override
public void aggregateGroupByMV(int length, int[][] groupKeysArray,
GroupByResultHolder groupByResultHolder,
Map<ExpressionContext, BlockValSet> blockValSetMap) {
@@ -235,7 +224,7 @@ public class PercentileSmartTDigestAggregationFunction
extends BaseSingleInputAg
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
for (int groupKey : groupKeysArray[i]) {
- getValueList(groupByResultHolder, groupKey).add(doubleValues[i]);
+ addValueForGroup(groupByResultHolder, groupKey, doubleValues[i],
_threshold);
}
}
});
@@ -244,14 +233,56 @@ public class PercentileSmartTDigestAggregationFunction
extends BaseSingleInputAg
forEachNotNull(length, blockValSet, (from, to) -> {
for (int i = from; i < to; i++) {
for (int groupKey : groupKeysArray[i]) {
- DoubleArrayList valueList = getValueList(groupByResultHolder,
groupKey);
- valueList.addElements(valueList.size(), doubleValues[i]);
+ addValuesForGroup(groupByResultHolder, groupKey, doubleValues[i],
_threshold);
}
}
});
}
}
+ /// Adds one value into the accumulator of the given group, converting it
once it holds more values than the
+ /// threshold. The accumulator is a [DoubleArrayList] until then and a
[TDigest] after. Without this the per-group
+ /// lists grow for the whole segment: it is the group-by counterpart of the
check in [#aggregateIntoValueList].
+ private void addValueForGroup(GroupByResultHolder groupByResultHolder, int
groupKey, double value, int threshold) {
+ Object result = groupByResultHolder.getResult(groupKey);
+ if (result instanceof TDigest) {
+ ((TDigest) result).add(value);
+ return;
+ }
+ DoubleArrayList valueList = getOrCreateValueList(groupByResultHolder,
groupKey, (DoubleArrayList) result);
+ valueList.add(value);
+ if (valueList.size() > threshold) {
+ groupByResultHolder.setValueForKey(groupKey,
convertValueListToTDigest(valueList));
+ }
+ }
+
+ /// As [#addValueForGroup], for every value of a multi-valued entry.
+ private void addValuesForGroup(GroupByResultHolder groupByResultHolder, int
groupKey, double[] values,
+ int threshold) {
+ Object result = groupByResultHolder.getResult(groupKey);
+ if (result instanceof TDigest) {
+ TDigest tDigest = (TDigest) result;
+ for (double value : values) {
+ tDigest.add(value);
+ }
+ return;
+ }
+ DoubleArrayList valueList = getOrCreateValueList(groupByResultHolder,
groupKey, (DoubleArrayList) result);
+ valueList.addElements(valueList.size(), values);
+ if (valueList.size() > threshold) {
+ groupByResultHolder.setValueForKey(groupKey,
convertValueListToTDigest(valueList));
+ }
+ }
+
+ private static DoubleArrayList getOrCreateValueList(GroupByResultHolder
groupByResultHolder, int groupKey,
+ @Nullable DoubleArrayList valueList) {
+ if (valueList == null) {
+ valueList = new DoubleArrayList();
+ groupByResultHolder.setValueForKey(groupKey, valueList);
+ }
+ return valueList;
+ }
+
@Override
public Object extractAggregationResult(AggregationResultHolder
aggregationResultHolder) {
return aggregationResultHolder.getResult();
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunctionTest.java
index 4655ab6a95a..69d2274f0b6 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/DistinctCountSmartHLLAggregationFunctionTest.java
@@ -20,9 +20,12 @@ package org.apache.pinot.core.query.aggregation.function;
import com.clearspring.analytics.stream.cardinality.HyperLogLog;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.common.utils.DataSchema;
+import org.apache.pinot.core.common.BlockValSet;
+import org.apache.pinot.core.common.SyntheticBlockValSets;
import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
import org.apache.pinot.spi.data.FieldSpec.DataType;
import org.testng.annotations.Test;
@@ -33,6 +36,7 @@ import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
+@SuppressWarnings("rawtypes")
public class DistinctCountSmartHLLAggregationFunctionTest {
@Test
@@ -170,6 +174,128 @@ public class DistinctCountSmartHLLAggregationFunctionTest
{
assertTrue(cardinality >= 90 && cardinality <= 110, "Cardinality out of
range: " + cardinality);
}
+ @Test
+ public void testRawValueGroupBySvStaysExactBelowThreshold() {
+ DistinctCountSmartHLLAggregationFunction function =
newFunctionWithThreshold(100);
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10, 10);
+ function.aggregateGroupBySV(3, new int[]{0, 0, 1}, holder,
+ Map.of(ExpressionContext.forIdentifier("col"),
SyntheticBlockValSets.Int.create(null, new int[]{1, 2, 3})));
+
+ assertTrue(function.extractGroupByResult(holder, 0) instanceof Set);
+ assertTrue(function.extractGroupByResult(holder, 1) instanceof Set);
+ }
+
+ /// Values that arrive after a group converted must reach that group's
sketch, not a fresh value set.
+ @Test
+ public void testRawValueGroupBySvKeepsFeedingTheSketchAfterConversion() {
+ DistinctCountSmartHLLAggregationFunction function =
newFunctionWithThreshold(4);
+ int[] groupKeys = {0, 0, 0};
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10, 10);
+ ExpressionContext column = ExpressionContext.forIdentifier("col");
+
+ function.aggregateGroupBySV(3, groupKeys, holder,
+ Map.of(column, SyntheticBlockValSets.Int.create(null, new int[]{1, 2,
3})));
+ assertTrue(function.extractGroupByResult(holder, 0) instanceof Set);
+
+ function.aggregateGroupBySV(3, groupKeys, holder,
+ Map.of(column, SyntheticBlockValSets.Int.create(null, new int[]{4, 5,
6})));
+ assertTrue(function.extractGroupByResult(holder, 0) instanceof
HyperLogLog);
+
+ function.aggregateGroupBySV(3, groupKeys, holder,
+ Map.of(column, SyntheticBlockValSets.Int.create(null, new int[]{7, 8,
9})));
+ Object result = function.extractGroupByResult(holder, 0);
+ assertTrue(result instanceof HyperLogLog);
+ assertEquals(function.extractFinalResult(result), 9);
+ }
+
+ @Test
+ public void testRawValueGroupByMvConvertsEveryGroupTheRowBelongsTo() {
+ DistinctCountSmartHLLAggregationFunction function =
newFunctionWithThreshold(3);
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10, 10);
+ int[][] groupKeysArray = {{0, 1}, {0, 1}, {0, 1}, {0, 1}};
+ ExpressionContext column = ExpressionContext.forIdentifier("col");
+
+ function.aggregateGroupByMV(4, groupKeysArray, holder,
+ Map.of(column, SyntheticBlockValSets.Int.create(null, new int[]{1, 2,
3, 4})));
+ function.aggregateGroupByMV(4, groupKeysArray, holder,
+ Map.of(column, SyntheticBlockValSets.Int.create(null, new int[]{5, 6,
7, 8})));
+
+ for (int groupKey = 0; groupKey < 2; groupKey++) {
+ Object result = function.extractGroupByResult(holder, groupKey);
+ assertTrue(result instanceof HyperLogLog, "Group " + groupKey + " was
not converted: " + result);
+ assertEquals(function.extractFinalResult(result), 8);
+ }
+ }
+
+ /// BYTES values are the one branch of `addSetToSketch` that unwraps a
[org.apache.pinot.spi.utils.ByteArray] before
+ /// hashing, so it has to fold post-conversion values into the sketch the
same way the initial conversion did.
+ @Test
+ public void testRawValueGroupByConvertsBytesColumn() {
+ DistinctCountSmartHLLAggregationFunction function =
newFunctionWithThreshold(2);
+ int[] groupKeys = {0, 0, 0};
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10, 10);
+ ExpressionContext column = ExpressionContext.forIdentifier("col");
+
+ function.aggregateGroupBySV(3, groupKeys, holder, Map.of(column,
+ SyntheticBlockValSets.Bytes.create(null, new byte[][]{{1}, {2},
{3}})));
+
+ // The conversion happens on the first touch after the threshold is
crossed, and values that arrive after it must
+ // reach the same sketch, hashed the same way.
+ function.aggregateGroupBySV(3, groupKeys, holder, Map.of(column,
+ SyntheticBlockValSets.Bytes.create(null, new byte[][]{{3}, {4},
{5}})));
+ Object result = function.extractGroupByResult(holder, 0);
+ assertTrue(result instanceof HyperLogLog);
+ assertEquals(function.extractFinalResult(result), 5);
+ }
+
+ /// The other two sketch families have their own hashing, and UltraLogLog
drops values its hasher rejects, so a
+ /// post-conversion value has to be seen to land. STRING is the other
ObjectOpenHashSet branch. The counts are large
+ /// enough that a dropped batch would halve the estimate, well outside the
error of these sketches.
+ @Test
+ public void
testRawValueGroupByConvertsForEverySketchFamilyAndStringColumns() {
+ ExpressionContext column = ExpressionContext.forIdentifier("col");
+ ExpressionContext params = ExpressionContext.forLiteral(DataType.STRING,
"threshold=10");
+ List<AggregationFunction> functions = List.of(
+ new DistinctCountSmartHLLAggregationFunction(List.of(column, params),
false),
+ new DistinctCountSmartHLLPlusAggregationFunction(List.of(column,
params), false),
+ new DistinctCountSmartULLAggregationFunction(List.of(column, params),
false));
+ int[] first = new int[50];
+ int[] second = new int[50];
+ String[] firstStrings = new String[50];
+ String[] secondStrings = new String[50];
+ for (int i = 0; i < 50; i++) {
+ first[i] = i;
+ second[i] = 50 + i;
+ firstStrings[i] = "v" + i;
+ secondStrings[i] = "v" + (50 + i);
+ }
+ int[] groupKeys = new int[50];
+
+ for (AggregationFunction function : functions) {
+ for (BlockValSet[] batches : List.of(
+ new BlockValSet[]{
+ SyntheticBlockValSets.Int.create(null, first),
SyntheticBlockValSets.Int.create(null, second)},
+ new BlockValSet[]{
+ SyntheticBlockValSets.Str.create(null, firstStrings),
+ SyntheticBlockValSets.Str.create(null, secondStrings)})) {
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10,
10);
+ function.aggregateGroupBySV(50, groupKeys, holder, Map.of(column,
batches[0]));
+ function.aggregateGroupBySV(50, groupKeys, holder, Map.of(column,
batches[1]));
+
+ Object result = function.extractGroupByResult(holder, 0);
+ assertFalse(result instanceof Set, function.getType() + " did not
convert: " + result);
+ int estimate = ((Number)
function.extractFinalResult(result)).intValue();
+ assertTrue(estimate >= 90 && estimate <= 110, function.getType() + "
estimated " + estimate + ", expected 100");
+ }
+ }
+ }
+
+ private static DistinctCountSmartHLLAggregationFunction
newFunctionWithThreshold(int threshold) {
+ return new DistinctCountSmartHLLAggregationFunction(
+ List.of(ExpressionContext.forIdentifier("col"),
+ ExpressionContext.forLiteral(DataType.STRING, "threshold=" +
threshold)), false);
+ }
+
@Test
public void testAdaptiveConversion() {
// Test adaptive conversion enabled by default (100K threshold)
diff --git
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunctionTest.java
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunctionTest.java
index 324e736d266..256ef347a29 100644
---
a/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunctionTest.java
+++
b/pinot-core/src/test/java/org/apache/pinot/core/query/aggregation/function/PercentileSmartTDigestAggregationFunctionTest.java
@@ -32,6 +32,7 @@ import
org.apache.pinot.common.request.context.ExpressionContext;
import org.apache.pinot.core.common.ObjectSerDeUtils;
import org.apache.pinot.core.common.SyntheticBlockValSets;
import org.apache.pinot.core.query.aggregation.AggregationResultHolder;
+import
org.apache.pinot.core.query.aggregation.groupby.ObjectGroupByResultHolder;
import org.testng.annotations.Test;
import static org.testng.Assert.assertEquals;
@@ -178,6 +179,81 @@ public class PercentileSmartTDigestAggregationFunctionTest
{
assertDuplicateInfinityResult((TDigest) merged, 2L * values.length);
}
+ @Test
+ public void testGroupBySvStaysExactBelowThreshold() {
+ PercentileSmartTDigestAggregationFunction function = newFunction(100);
+ int[] groupKeys = {0, 0, 1};
+
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10, 10);
+ function.aggregateGroupBySV(3, groupKeys, holder,
+ Map.of(EXPRESSION, SyntheticBlockValSets.Double.create(null, new
double[]{1.0, 2.0, 3.0})));
+
+ assertTrue(function.extractGroupByResult(holder, 0) instanceof
DoubleArrayList);
+ assertTrue(function.extractGroupByResult(holder, 1) instanceof
DoubleArrayList);
+ }
+
+ /// The accumulator of a converted group must keep taking values from later
batches, and it must take them into the
+ /// digest instead of into a new value list.
+ @Test
+ public void testGroupBySvKeepsFeedingTheDigestAfterConversion() {
+ PercentileSmartTDigestAggregationFunction function = newFunction(4);
+ int[] groupKeys = {0, 0, 0};
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10, 10);
+
+ function.aggregateGroupBySV(3, groupKeys, holder,
+ Map.of(EXPRESSION, SyntheticBlockValSets.Double.create(null, new
double[]{1.0, 2.0, 3.0})));
+ assertTrue(function.extractGroupByResult(holder, 0) instanceof
DoubleArrayList);
+
+ function.aggregateGroupBySV(3, groupKeys, holder,
+ Map.of(EXPRESSION, SyntheticBlockValSets.Double.create(null, new
double[]{4.0, 5.0, 6.0})));
+ Object converted = function.extractGroupByResult(holder, 0);
+ assertTrue(converted instanceof TDigest);
+ assertEquals(((TDigest) converted).size(), 6L);
+
+ function.aggregateGroupBySV(3, groupKeys, holder,
+ Map.of(EXPRESSION, SyntheticBlockValSets.Double.create(null, new
double[]{7.0, 8.0, 9.0})));
+ Object stillConverted = function.extractGroupByResult(holder, 0);
+ assertTrue(stillConverted instanceof TDigest);
+ assertEquals(((TDigest) stillConverted).size(), 9L);
+ assertEquals(function.extractFinalResult(stillConverted), 5.0, 1.0);
+ }
+
+ @Test
+ public void testGroupBySvConvertsForMultiValuedColumn() {
+ PercentileSmartTDigestAggregationFunction function = newFunction(4);
+ int[] groupKeys = {0, 0};
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10, 10);
+
+ function.aggregateGroupBySV(2, groupKeys, holder, Map.of(EXPRESSION,
+ SyntheticBlockValSets.DoubleMV.create(null, new double[][]{{1.0, 2.0,
3.0}, {4.0, 5.0, 6.0}})));
+
+ Object result = function.extractGroupByResult(holder, 0);
+ assertTrue(result instanceof TDigest);
+ assertEquals(((TDigest) result).size(), 6L);
+ }
+
+ @Test
+ public void testGroupByMvConvertsEveryGroupTheRowBelongsTo() {
+ PercentileSmartTDigestAggregationFunction function = newFunction(4);
+ int[][] groupKeysArray = {{0, 1}, {0, 1}, {0, 1}, {0, 1}, {0, 1}};
+ ObjectGroupByResultHolder holder = new ObjectGroupByResultHolder(10, 10);
+
+ function.aggregateGroupByMV(5, groupKeysArray, holder,
+ Map.of(EXPRESSION, SyntheticBlockValSets.Double.create(null, new
double[]{1.0, 2.0, 3.0, 4.0, 5.0})));
+
+ for (int groupKey = 0; groupKey < 2; groupKey++) {
+ Object result = function.extractGroupByResult(holder, groupKey);
+ assertTrue(result instanceof TDigest, "Group " + groupKey + " was not
converted");
+ assertEquals(((TDigest) result).size(), 5L);
+ }
+ }
+
+ private static PercentileSmartTDigestAggregationFunction newFunction(int
threshold) {
+ return new PercentileSmartTDigestAggregationFunction(
+ List.of(EXPRESSION,
ExpressionContext.forLiteral(Literal.doubleValue(50.0)),
+ ExpressionContext.forLiteral(Literal.stringValue("THRESHOLD=" +
threshold + ";COMPRESSION=20"))), false);
+ }
+
private static PercentileSmartTDigestAggregationFunction newFunction() {
return new PercentileSmartTDigestAggregationFunction(
List.of(EXPRESSION,
ExpressionContext.forLiteral(Literal.doubleValue(50.0)),
diff --git
a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java
new file mode 100644
index 00000000000..67596a1072f
--- /dev/null
+++
b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ApproximateFunctionOverrideIntegrationTest.java
@@ -0,0 +1,202 @@
+/**
+ * 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.integration.tests;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.File;
+import java.util.List;
+import org.apache.commons.io.FileUtils;
+import org.apache.helix.model.HelixConfigScope;
+import org.apache.helix.model.builder.HelixConfigScopeBuilder;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants.Broker;
+import org.apache.pinot.util.TestUtils;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+
+/// End-to-end test of the cluster config that rewrites exact aggregations
into their approximate counterparts. It
+/// covers the two properties unit tests cannot show: the config takes effect
on a running broker without a restart,
+/// and the rewrite is invisible to the caller apart from the flag on the
response.
+public class ApproximateFunctionOverrideIntegrationTest extends
BaseClusterIntegrationTestSet {
+ private static final String DISTINCT_COUNT_QUERY =
+ "SELECT Carrier, DISTINCTCOUNT(AirlineID) FROM mytable GROUP BY Carrier
ORDER BY Carrier LIMIT 100";
+ private static final String COUNT_DISTINCT_QUERY =
+ "SELECT Carrier, COUNT(DISTINCT AirlineID) FROM mytable GROUP BY Carrier
ORDER BY Carrier LIMIT 100";
+ private static final String PERCENTILE_QUERY =
+ "SELECT Carrier, PERCENTILE(ArrDelay, 90) FROM mytable GROUP BY Carrier
ORDER BY Carrier LIMIT 100";
+
+ private HelixConfigScope _clusterScope;
+
+ @BeforeClass
+ public void setUp()
+ throws Exception {
+ TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir);
+
+ startZk();
+ startController();
+ startBroker();
+ startServer();
+ _clusterScope =
+ new
HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER).forCluster(getHelixClusterName())
+ .build();
+
+ Schema schema = createSchema();
+ addSchema(schema);
+ TableConfig tableConfig = createOfflineTableConfig();
+ addTableConfig(tableConfig);
+
+ List<File> avroFiles = unpackAvroData(_tempDir);
+ ClusterIntegrationTestUtils.buildSegmentsFromAvro(avroFiles, tableConfig,
schema, 0, _segmentDir, _tarDir);
+ uploadSegments(getTableName(), _tarDir);
+ waitForAllDocsLoaded(600_000L);
+ }
+
+ /// Clears the config after every method, so that a method that fails part
way cannot make the next one fail too.
+ @AfterMethod
+ public void clearOverride() {
+ clearApproximateFunctionOverride();
+ waitForOverride(false);
+ }
+
+ @AfterClass
+ public void tearDown()
+ throws Exception {
+ dropOfflineTable(getTableName());
+ stopServer();
+ stopBroker();
+ stopController();
+ stopZk();
+ FileUtils.deleteDirectory(_tempDir);
+ }
+
+ @Test(dataProvider = "useBothQueryEngines")
+ public void testClusterConfigAppliesWithoutRestart(boolean
useMultiStageQueryEngine)
+ throws Exception {
+ setUseMultiStageQueryEngine(useMultiStageQueryEngine);
+
+ JsonNode exactDistinctCount = postQuery(DISTINCT_COUNT_QUERY);
+ JsonNode exactPercentile = postQuery(PERCENTILE_QUERY);
+ assertApplied(exactDistinctCount, false);
+ assertApplied(exactPercentile, false);
+
+ // A threshold far above the cardinality of the test data keeps the
rewritten functions exact, which is what makes
+ // the answers comparable before and after the flip.
+ setApproximateFunctionOverride("threshold=1000000");
+ waitForOverride(true);
+
+ JsonNode distinctCount = postQuery(DISTINCT_COUNT_QUERY);
+ assertApplied(distinctCount, true);
+ assertSameResult(exactDistinctCount, distinctCount);
+ JsonNode percentile = postQuery(PERCENTILE_QUERY);
+ assertApplied(percentile, true);
+ assertSameResult(exactPercentile, percentile);
+
+ // COUNT(DISTINCT x) is the standard SQL spelling and must be rewritten
too.
+ assertApplied(postQuery(COUNT_DISTINCT_QUERY), true);
+
+ // The query option is the escape hatch back to exact results.
+ JsonNode forcedExact = postQuery("SET useApproximateFunction = false; " +
DISTINCT_COUNT_QUERY);
+ assertApplied(forcedExact, false);
+ assertSameResult(exactDistinctCount, forcedExact);
+
+ // A threshold below the cardinality has to make the answer approximate
rather than fail, keeping the same shape.
+ // This is also what proves the parameters reach the servers and parse
there.
+ setApproximateFunctionOverride("threshold=1");
+ waitForOverride(true);
+ for (JsonNode exact : List.of(exactDistinctCount, exactPercentile)) {
+ String query = exact == exactDistinctCount ? DISTINCT_COUNT_QUERY :
PERCENTILE_QUERY;
+ JsonNode approximate = postQuery(query);
+ assertApplied(approximate, true);
+ assertEquals(approximate.get("exceptions").size(), 0,
approximate.toString());
+ assertEquals(columnDataTypes(approximate), columnDataTypes(exact));
+ assertEquals(rows(approximate).size(), rows(exact).size());
+ }
+ }
+
+ /// Pins the one visible difference the rewrite makes to a single-stage
response, so that changing it is a
+ /// deliberate decision. The multi-stage engine keeps the Calcite-derived
column name.
+ @Test
+ public void testSingleStageRenamesTheColumn()
+ throws Exception {
+ setUseMultiStageQueryEngine(false);
+ assertEquals(columnName(postQuery(DISTINCT_COUNT_QUERY)),
"distinctcount(AirlineID)");
+
+ setApproximateFunctionOverride("threshold=1000000");
+ waitForOverride(true);
+ assertEquals(columnName(postQuery(DISTINCT_COUNT_QUERY)),
"distinctcountsmarthll(AirlineID)");
+
+ // An explicit alias is the way to keep the old name.
+ assertEquals(columnName(postQuery(
+ "SELECT Carrier, DISTINCTCOUNT(AirlineID) AS dc FROM mytable GROUP BY
Carrier ORDER BY Carrier LIMIT 100")),
+ "dc");
+ }
+
+ /// Compares the part of the result the rewrite must not change: the values
and their types. The column name is
+ /// excluded on purpose, see [#testSingleStageRenamesTheColumn].
+ private static void assertSameResult(JsonNode expected, JsonNode actual) {
+ assertEquals(columnDataTypes(actual), columnDataTypes(expected));
+ assertEquals(rows(actual), rows(expected));
+ }
+
+ private static void assertApplied(JsonNode response, boolean expected) {
+ assertEquals(response.get("approximateFunctionApplied").asBoolean(),
expected, response.toString());
+ }
+
+ private static JsonNode columnDataTypes(JsonNode response) {
+ return
response.get("resultTable").get("dataSchema").get("columnDataTypes");
+ }
+
+ private static JsonNode rows(JsonNode response) {
+ return response.get("resultTable").get("rows");
+ }
+
+ private static String columnName(JsonNode response) {
+ return
response.get("resultTable").get("dataSchema").get("columnNames").get(1).asText();
+ }
+
+ private void setApproximateFunctionOverride(String params) {
+ _helixManager.getConfigAccessor().set(_clusterScope,
Broker.USE_APPROXIMATE_FUNCTION, "true");
+ _helixManager.getConfigAccessor().set(_clusterScope,
Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS, params);
+ _helixManager.getConfigAccessor().set(_clusterScope,
Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS, params);
+ }
+
+ /// Removes the keys, so the broker falls back to its own conf.
+ private void clearApproximateFunctionOverride() {
+ _helixManager.getConfigAccessor().remove(_clusterScope,
Broker.USE_APPROXIMATE_FUNCTION);
+ _helixManager.getConfigAccessor().remove(_clusterScope,
Broker.APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS);
+ _helixManager.getConfigAccessor().remove(_clusterScope,
Broker.APPROXIMATE_FUNCTION_PERCENTILE_PARAMS);
+ }
+
+ /// The broker picks the change up from a ZooKeeper watch, so the test waits
for it instead of restarting anything.
+ private void waitForOverride(boolean expected) {
+ TestUtils.waitForCondition(aVoid -> {
+ try {
+ return
postQuery(DISTINCT_COUNT_QUERY).get("approximateFunctionApplied").asBoolean()
== expected;
+ } catch (Exception e) {
+ return false;
+ }
+ }, 100L, 60_000L, "Broker did not pick up the approximate function cluster
config");
+ }
+}
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java
new file mode 100644
index 00000000000..5f2bf560704
--- /dev/null
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java
@@ -0,0 +1,191 @@
+/**
+ * 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.List;
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.calcite.plan.Context;
+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.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.SqlAggFunction;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.ReturnTypes;
+import org.apache.calcite.tools.RelBuilderFactory;
+import org.apache.pinot.common.function.sql.PinotSqlAggFunction;
+import org.apache.pinot.query.QueryEnvironment;
+import org.apache.pinot.query.context.PlannerContext;
+import org.apache.pinot.segment.spi.AggregationFunctionType;
+import org.apache.pinot.spi.utils.CommonConstants.Broker.PlannerRuleNames;
+
+
+/// Rewrites exact aggregations into their threshold-based approximate
counterparts, so that one cluster config can
+/// stop unbounded per-group accumulators from taking servers down:
+/// - `DISTINCT_COUNT(x)` and `COUNT(DISTINCT x)` ->
`DISTINCT_COUNT_SMART_HLL(x)`
+/// - `PERCENTILE(x, p)` -> `PERCENTILE_SMART_TDIGEST(x, p)`
+///
+/// This is the multi-stage counterpart of
`BaseSingleStageBrokerRequestHandler.handleApproximateFunctionOverride`,
+/// gated by the same resolved setting, read from
[QueryEnvironment.Config#useApproximateFunction()]. The rewrite keeps
+/// the original return type, so switching the config on does not change the
result schema.
+///
+/// It must run before [PinotAggregateExchangeNodeInsertRule], which derives
the leaf-to-final intermediate result
+/// format from the function name, hence `Phase.BASIC` rather than that rule's
`POST_LOGICAL`. That is also why the
+/// runtime is not a valid place to do this rewrite.
+public class PinotApproximateAggregateRewriteRule extends RelOptRule {
+ public static final PinotApproximateAggregateRewriteRule INSTANCE =
+ new
PinotApproximateAggregateRewriteRule(PinotRuleUtils.PINOT_REL_FACTORY);
+
+ private PinotApproximateAggregateRewriteRule(RelBuilderFactory factory) {
+ super(operand(LogicalAggregate.class, any()), factory,
PlannerRuleNames.APPROXIMATE_AGGREGATE_REWRITE);
+ }
+
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ QueryEnvironment.Config envConfig = envConfig(call);
+ if (envConfig == null || !envConfig.useApproximateFunction()) {
+ return false;
+ }
+ // Requiring a rewritable call is also what makes the rule terminate: the
rewritten names no longer match.
+ Aggregate aggRel = call.rel(0);
+ return aggRel.getAggCallList().stream().anyMatch(aggCall ->
targetOf(aggCall) != null);
+ }
+
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ Aggregate aggRel = call.rel(0);
+ QueryEnvironment.Config envConfig =
Objects.requireNonNull(envConfig(call));
+ RelNode input = aggRel.getInput();
+ int numInputFields = input.getRowType().getFieldCount();
+
+ boolean hasDistinctCount = false;
+ boolean hasPercentile = false;
+ for (AggregateCall aggCall : aggRel.getAggCallList()) {
+ AggregationFunctionType target = targetOf(aggCall);
+ hasDistinctCount |= target ==
AggregationFunctionType.DISTINCTCOUNTSMARTHLL;
+ hasPercentile |= target ==
AggregationFunctionType.PERCENTILESMARTTDIGEST;
+ }
+ String distinctCountParams = hasDistinctCount ?
envConfig.approximateFunctionDistinctCountParams() : "";
+ String percentileParams = hasPercentile ?
envConfig.approximateFunctionPercentileParams() : "";
+
+ // The parameters must be input fields, because an aggregate call holds
field indices rather than inline literals,
+ // so they are projected underneath the aggregate. With no parameters
configured the plan keeps the shape it had
+ // before, with no extra project.
+ //
+ // Invariant: this project has to stay directly beneath the aggregate.
Both PinotAggregateExchangeNodeInsertRule
+ // and AggregatePushdownRule inline the literal only through
findImmediateProjects(input), so anything that puts a
+ // node between them would leave the leaf with a column reference where
the function expects a parameter string.
+ // Nothing in the later phases does: AggregateProjectMergeRule needs an
all-input-ref project and the literals
+ // block it, and ProjectMergeRule collapses into a single project that
stays adjacent.
+ RelNode newInput = input;
+ int distinctCountParamsIndex = -1;
+ int percentileParamsIndex = -1;
+ if (!distinctCountParams.isEmpty() || !percentileParams.isEmpty()) {
+ RexBuilder rexBuilder = aggRel.getCluster().getRexBuilder();
+ List<RexNode> projects = new ArrayList<>(numInputFields + 2);
+ for (int i = 0; i < numInputFields; i++) {
+ projects.add(rexBuilder.makeInputRef(input, i));
+ }
+ if (!distinctCountParams.isEmpty()) {
+ distinctCountParamsIndex = projects.size();
+ projects.add(rexBuilder.makeLiteral(distinctCountParams));
+ }
+ if (!percentileParams.isEmpty()) {
+ percentileParamsIndex = projects.size();
+ projects.add(rexBuilder.makeLiteral(percentileParams));
+ }
+ newInput = LogicalProject.create(input, List.of(), projects,
(List<String>) null);
+ }
+
+ List<AggregateCall> rewrittenAggCalls = new
ArrayList<>(aggRel.getAggCallList().size());
+ for (AggregateCall aggCall : aggRel.getAggCallList()) {
+ AggregationFunctionType target = targetOf(aggCall);
+ if (target == null) {
+ rewrittenAggCalls.add(aggCall);
+ continue;
+ }
+ int paramsIndex = target == AggregationFunctionType.DISTINCTCOUNTSMARTHLL
+ ? distinctCountParamsIndex : percentileParamsIndex;
+ List<Integer> argList = aggCall.getArgList();
+ if (paramsIndex >= 0) {
+ argList = new ArrayList<>(argList);
+ argList.add(paramsIndex);
+ }
+ rewrittenAggCalls.add(rewrite(aggCall, target, argList, newInput,
aggRel.getGroupCount()));
+ }
+
+ PlannerContext plannerContext =
call.getPlanner().getContext().unwrap(PlannerContext.class);
+ if (plannerContext != null) {
+ plannerContext.setApproximateFunctionApplied();
+ }
+ call.transformTo(
+ aggRel.copy(aggRel.getTraitSet(), newInput, aggRel.getGroupSet(),
aggRel.getGroupSets(), rewrittenAggCalls));
+ }
+
+ /// Returns the approximate function this call should become, or `null` when
the call must be left alone.
+ @Nullable
+ private static AggregationFunctionType targetOf(AggregateCall aggCall) {
+ SqlAggFunction aggFunction = aggCall.getAggregation();
+ if (aggCall.isDistinct()) {
+ // COUNT(DISTINCT x) is the standard SQL spelling of DISTINCT_COUNT, and
PinotAggregateExchangeNodeInsertRule
+ // only renames it in POST_LOGICAL, after this rule, so it has to be
matched on the kind here. Multi-argument
+ // COUNT(DISTINCT a, b) means something else and is left alone.
+ return aggFunction.getKind() == SqlKind.COUNT &&
aggCall.getArgList().size() == 1
+ ? AggregationFunctionType.DISTINCTCOUNTSMARTHLL : null;
+ }
+ // The smart functions take multi-valued input themselves, so the MV
spellings map onto the same targets, which
+ // keeps this in step with the single-stage rewrite.
+ String name =
AggregationFunctionType.getNormalizedAggregationFunctionName(aggFunction.getName());
+ if (name.equals(AggregationFunctionType.DISTINCTCOUNT.name())
+ || name.equals(AggregationFunctionType.DISTINCTCOUNTMV.name())) {
+ return AggregationFunctionType.DISTINCTCOUNTSMARTHLL;
+ }
+ if (name.equals(AggregationFunctionType.PERCENTILE.name())
+ || name.equals(AggregationFunctionType.PERCENTILEMV.name())) {
+ return AggregationFunctionType.PERCENTILESMARTTDIGEST;
+ }
+ return null;
+ }
+
+ private static AggregateCall rewrite(AggregateCall aggCall,
AggregationFunctionType target, List<Integer> argList,
+ RelNode input, int groupCount) {
+ // Pinning the original type keeps the rewrite invisible to the caller:
PERCENTILE infers ARG0 while
+ // PERCENTILE_SMART_TDIGEST infers DOUBLE, and a silent cluster-wide
rewrite must not change the result schema.
+ SqlAggFunction newAggFunction = new PinotSqlAggFunction(target.name(),
SqlKind.OTHER_FUNCTION,
+ ReturnTypes.explicit(aggCall.getType()),
target.getOperandTypeChecker(),
+ SqlFunctionCategory.USER_DEFINED_FUNCTION);
+ return AggregateCall.create(newAggFunction, false,
aggCall.isApproximate(), aggCall.ignoreNulls(), argList,
+ aggCall.filterArg, aggCall.distinctKeys, aggCall.getCollation(),
groupCount, input, aggCall.getType(),
+ aggCall.getName());
+ }
+
+ @Nullable
+ private static QueryEnvironment.Config envConfig(RelOptRuleCall call) {
+ Context context = call.getPlanner().getContext();
+ return context != null ? context.unwrap(QueryEnvironment.Config.class) :
null;
+ }
+}
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotQueryRuleSets.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotQueryRuleSets.java
index 38822d4dfad..f62294c58ef 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotQueryRuleSets.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotQueryRuleSets.java
@@ -175,6 +175,10 @@ public class PinotQueryRuleSets {
PinotAggregateFunctionRewriteRule
.instanceWithDescription(PlannerRuleNames.AGGREGATE_FUNCTION_REWRITE),
+ // Must stay ahead of PinotAggregateExchangeNodeInsertRule
(POST_LOGICAL), which fixes the leaf-to-final
+ // intermediate result format from the function name.
+ PinotApproximateAggregateRewriteRule.INSTANCE,
+
// convert CASE-style filtered aggregates into true filtered aggregates
// put it after AGGREGATE_REDUCE_FUNCTIONS where SUM is converted to SUM0
AggregateCaseToFilterRule.Config.DEFAULT
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 3f30e35547a..d07f2089082 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
@@ -805,6 +805,25 @@ public class QueryEnvironment {
return CommonConstants.Broker.DEFAULT_MSE_ENABLE_GROUP_TRIM;
}
+ /// Whether to rewrite exact aggregations into their approximate
counterparts, already resolved from the query
+ /// option and the defaults. See
[CommonConstants.Broker#USE_APPROXIMATE_FUNCTION].
+ @Value.Default
+ default boolean useApproximateFunction() {
+ return CommonConstants.Broker.DEFAULT_USE_APPROXIMATE_FUNCTION;
+ }
+
+ /// Parameters appended to the rewritten calls, empty for the aggregation
function defaults.
+ /// See
[CommonConstants.Broker#APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS].
+ @Value.Default
+ default String approximateFunctionDistinctCountParams() {
+ return CommonConstants.Broker.DEFAULT_APPROXIMATE_FUNCTION_PARAMS;
+ }
+
+ @Value.Default
+ default String approximateFunctionPercentileParams() {
+ return CommonConstants.Broker.DEFAULT_APPROXIMATE_FUNCTION_PARAMS;
+ }
+
@Value.Default
default boolean defaultEnableDynamicFilteringSemiJoin() {
return CommonConstants.Broker.DEFAULT_ENABLE_DYNAMIC_FILTERING_SEMI_JOIN;
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/context/PlannerContext.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/context/PlannerContext.java
index 34f1c4c6965..f5748199acb 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/context/PlannerContext.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/context/PlannerContext.java
@@ -61,6 +61,10 @@ public class PlannerContext implements AutoCloseable,
Context {
private final SqlExplainFormat _sqlExplainFormat;
@Nullable
private final PhysicalPlannerContext _physicalPlannerContext;
+ /// Set by the approximate aggregation rewrite rule when it rewrites at
least one aggregation, so that the broker
+ /// can report on the response that the results are approximate. Written
during planning, which is single threaded
+ /// for a query, and read after planning completes.
+ private boolean _approximateFunctionApplied;
public PlannerContext(FrameworkConfig config, Prepare.CatalogReader
catalogReader, RelDataTypeFactory typeFactory,
HepProgram optProgram, HepProgram traitProgram, Map<String, String>
options, QueryEnvironment.Config envConfig,
@@ -123,6 +127,16 @@ public class PlannerContext implements AutoCloseable,
Context {
return _envConfig;
}
+ /// Records that an exact aggregation was rewritten into its approximate
counterpart.
+ public void setApproximateFunctionApplied() {
+ _approximateFunctionApplied = true;
+ }
+
+ /// Returns whether any exact aggregation was rewritten into its approximate
counterpart while planning this query.
+ public boolean isApproximateFunctionApplied() {
+ return _approximateFunctionApplied;
+ }
+
/// Unwraps this context. Returns `this` when asked for [PlannerContext] or
/// [Context], and delegates to [#_envConfig] when asked for
/// [QueryEnvironment.Config] so that existing rules remain compatible.
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/ApproximateAggregateRewriteTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/ApproximateAggregateRewriteTest.java
new file mode 100644
index 00000000000..7a93ab1966e
--- /dev/null
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/ApproximateAggregateRewriteTest.java
@@ -0,0 +1,162 @@
+/**
+ * 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.query;
+
+import java.util.Map;
+import org.apache.pinot.core.routing.MockRoutingManagerFactory;
+import org.apache.pinot.core.routing.RoutingManager;
+import org.apache.pinot.query.routing.WorkerManager;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.utils.CommonConstants;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests the multi-stage counterpart of the single-stage approximate function
override.
+public class ApproximateAggregateRewriteTest extends QueryEnvironmentTestBase {
+
+ @Test
+ public void testDisabledByDefault() {
+ String plan = explain(newQueryEnvironment(false, "", ""), "SELECT
DISTINCTCOUNT(col1) FROM a");
+ assertTrue(plan.toLowerCase().contains("distinctcount"), plan);
+ assertFalse(plan.toLowerCase().contains("distinctcountsmarthll"), plan);
+ }
+
+ @Test
+ public void testRewritesDistinctCountAndPercentile() {
+ QueryEnvironment env = newQueryEnvironment(true, "", "");
+
+ assertRewritten(explain(env, "SELECT DISTINCTCOUNT(col1) FROM a"),
"distinctcountsmarthll");
+ assertRewritten(explain(env, "SELECT col2, DISTINCTCOUNT(col1) FROM a
GROUP BY col2"), "distinctcountsmarthll");
+ assertRewritten(explain(env, "SELECT PERCENTILE(col3, 90) FROM a"),
"percentilesmarttdigest");
+ assertRewritten(explain(env, "SELECT col2, PERCENTILE(col3, 90) FROM a
GROUP BY col2"), "percentilesmarttdigest");
+
+ // COUNT(DISTINCT x) is the standard SQL spelling, and the plan only
renames it to DISTINCTCOUNT after this rule
+ // has run, so the rule has to recognise it by kind.
+ assertRewritten(explain(env, "SELECT COUNT(DISTINCT col1) FROM a"),
"distinctcountsmarthll");
+ assertRewritten(explain(env, "SELECT col2, COUNT(DISTINCT col1) FROM a
GROUP BY col2"), "distinctcountsmarthll");
+
+ // The multi-valued spellings too, so that the two engines agree.
+ assertRewritten(explain(env, "SELECT DISTINCTCOUNTMV(mcol1) FROM e"),
"distinctcountsmarthll");
+ assertRewritten(explain(env, "SELECT PERCENTILEMV(mcol2, 90) FROM e"),
"percentilesmarttdigest");
+ }
+
+ @Test
+ public void testLeavesOtherAggregationsAlone() {
+ QueryEnvironment env = newQueryEnvironment(true, "", "");
+ assertNotRewritten(explain(env, "SELECT DISTINCTCOUNTHLL(col1) FROM a"));
+ assertNotRewritten(explain(env, "SELECT PERCENTILETDIGEST(col3, 90) FROM
a"));
+ assertNotRewritten(explain(env, "SELECT SUM(col3), COUNT(*) FROM a"));
+ // COUNT(DISTINCT a, b) means something different from DISTINCT_COUNT.
+ assertNotRewritten(explain(env, "SELECT COUNT(DISTINCT col1, col2) FROM
a"));
+ }
+
+ @Test
+ public void testAppendsConfiguredParams() {
+ QueryEnvironment env = newQueryEnvironment(true, "threshold=17",
"threshold=23;compression=50");
+
+ // Separate queries, so that each parameter string is seen to reach only
its own function.
+ String distinctCount = explain(env, "SELECT col2, DISTINCTCOUNT(col1) FROM
a GROUP BY col2");
+ assertRewritten(distinctCount, "distinctcountsmarthll");
+ assertTrue(distinctCount.contains("threshold=17"), distinctCount);
+ assertFalse(distinctCount.contains("threshold=23"), distinctCount);
+
+ String percentile = explain(env, "SELECT col2, PERCENTILE(col3, 90) FROM a
GROUP BY col2");
+ assertRewritten(percentile, "percentilesmarttdigest");
+ assertTrue(percentile.contains("threshold=23;compression=50"), percentile);
+
+ // Both in one query needs two literals in the same project.
+ String both = explain(env, "SELECT col2, DISTINCTCOUNT(col1),
PERCENTILE(col3, 90) FROM a GROUP BY col2");
+ assertRewritten(both, "distinctcountsmarthll");
+ assertRewritten(both, "percentilesmarttdigest");
+ assertTrue(both.contains("threshold=17"), both);
+ assertTrue(both.contains("threshold=23;compression=50"), both);
+ }
+
+ /// The parameter literal has to stay in the project directly beneath the
aggregate, or the split rules cannot
+ /// inline it. These shapes put a join and a subquery under the aggregate,
unlike the single-table cases above.
+ @Test
+ public void testAppendsParamsUnderAJoinAndASubquery() {
+ QueryEnvironment env = newQueryEnvironment(true, "threshold=17",
"threshold=23");
+
+ String join = explain(env,
+ "SELECT a.col2, DISTINCTCOUNT(a.col1) FROM a JOIN b ON a.col1 = b.col1
GROUP BY a.col2");
+ assertRewritten(join, "distinctcountsmarthll");
+ assertTrue(join.contains("threshold=17"), join);
+
+ String subquery = explain(env,
+ "SELECT col2, PERCENTILE(col3, 90) FROM (SELECT col2, col3 FROM a
WHERE col3 > 0) GROUP BY col2");
+ assertRewritten(subquery, "percentilesmarttdigest");
+ assertTrue(subquery.contains("threshold=23"), subquery);
+ }
+
+ /// The physical optimizer splits aggregates in its own rule, which also
keys the leaf-to-final intermediate format
+ /// off the function name, so the rewrite has to hold on that path too.
+ @Test
+ public void testRewritesUnderThePhysicalOptimizer() {
+ QueryEnvironment env = newQueryEnvironment(true, "threshold=17",
"threshold=23", true);
+ String distinctCount = explain(env, "SELECT col2, DISTINCTCOUNT(col1) FROM
a GROUP BY col2");
+ assertRewritten(distinctCount, "distinctcountsmarthll");
+ assertTrue(distinctCount.contains("threshold=17"), distinctCount);
+
+ String percentile = explain(env, "SELECT col2, PERCENTILE(col3, 90) FROM a
GROUP BY col2");
+ assertRewritten(percentile, "percentilesmarttdigest");
+ assertTrue(percentile.contains("threshold=23"), percentile);
+ }
+
+ private static void assertRewritten(String plan, String expectedFunction) {
+ assertTrue(plan.toLowerCase().contains(expectedFunction), "Expected " +
expectedFunction + " in:\n" + plan);
+ }
+
+ private static void assertNotRewritten(String plan) {
+ assertFalse(plan.toLowerCase().contains("smart"), "Unexpected rewrite
in:\n" + plan);
+ }
+
+ private static String explain(QueryEnvironment env, String query) {
+ return env.explainQuery("EXPLAIN PLAN FOR " + query,
RANDOM_REQUEST_ID_GEN.nextLong());
+ }
+
+ private static QueryEnvironment newQueryEnvironment(boolean
useApproximateFunction, String distinctCountParams,
+ String percentileParams) {
+ return newQueryEnvironment(useApproximateFunction, distinctCountParams,
percentileParams, false);
+ }
+
+ private static QueryEnvironment newQueryEnvironment(boolean
useApproximateFunction, String distinctCountParams,
+ String percentileParams, boolean usePhysicalOptimizer) {
+ MockRoutingManagerFactory factory = new MockRoutingManagerFactory(1, 2);
+ for (Map.Entry<String, Schema> entry : TABLE_SCHEMAS.entrySet()) {
+ factory.registerTable(entry.getValue(), entry.getKey());
+ }
+ SERVER1_SEGMENTS.forEach((table, segments) -> segments.forEach(s ->
factory.registerSegment(1, table, s)));
+ SERVER2_SEGMENTS.forEach((table, segments) -> segments.forEach(s ->
factory.registerSegment(2, table, s)));
+ RoutingManager routingManager = factory.buildRoutingManager(null);
+ return new QueryEnvironment(QueryEnvironment.configBuilder()
+ .requestId(-1L)
+ .database(CommonConstants.DEFAULT_DATABASE)
+ .tableCache(factory.buildTableCache())
+ .workerManager(new WorkerManager("Broker_localhost", "localhost", 3,
routingManager))
+ .useApproximateFunction(useApproximateFunction)
+ .approximateFunctionDistinctCountParams(distinctCountParams)
+ .approximateFunctionPercentileParams(percentileParams)
+ .defaultUsePhysicalOptimizer(usePhysicalOptimizer)
+ .build());
+ }
+}
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 080138b3a32..06d04efbb00 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
@@ -90,9 +90,11 @@ public enum AggregationFunctionType {
DISTINCTCOUNTRAWHLL("distinctCountRawHLL", ReturnTypes.VARCHAR,
OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.INTEGER), i
-> i == 1), SqlTypeName.OTHER),
DISTINCTCOUNTSMARTHLL("distinctCountSmartHLL", ReturnTypes.BIGINT,
- OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER),
i -> i == 1), SqlTypeName.OTHER),
+ OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER),
i -> i == 1), SqlTypeName.OTHER,
+ SqlTypeName.INTEGER),
DISTINCTCOUNTSMARTHLLPLUS("distinctCountSmartHLLPlus", ReturnTypes.BIGINT,
- OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER),
i -> i == 1), SqlTypeName.OTHER),
+ OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER),
i -> i == 1), SqlTypeName.OTHER,
+ SqlTypeName.INTEGER),
@Deprecated FASTHLL("fastHLL"),
DISTINCTCOUNTHLLPLUS("distinctCountHLLPlus", ReturnTypes.BIGINT,
OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.INTEGER), i
-> i == 1), SqlTypeName.OTHER),
@@ -103,7 +105,8 @@ public enum AggregationFunctionType {
DISTINCTCOUNTRAWULL("distinctCountRawULL", ReturnTypes.VARCHAR,
OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.INTEGER), i
-> i == 1), SqlTypeName.OTHER),
DISTINCTCOUNTSMARTULL("distinctCountSmartULL", ReturnTypes.BIGINT,
- OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER),
i -> i == 1), SqlTypeName.OTHER),
+ OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER),
i -> i == 1), SqlTypeName.OTHER,
+ SqlTypeName.INTEGER),
DISTINCTCOUNTTHETASKETCH("distinctCountThetaSketch", ReturnTypes.BIGINT,
OperandTypes.ONE_OR_MORE, SqlTypeName.OTHER),
DISTINCTCOUNTRAWTHETASKETCH("distinctCountRawThetaSketch",
ReturnTypes.VARCHAR, OperandTypes.ONE_OR_MORE,
SqlTypeName.OTHER),
@@ -129,9 +132,12 @@ public enum AggregationFunctionType {
PERCENTILERAWTDIGEST("percentileRawTDigest", ReturnTypes.VARCHAR,
OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.NUMERIC,
SqlTypeFamily.INTEGER), i -> i == 2),
SqlTypeName.OTHER),
+ // The final return type is redundant with the standard return type above,
and is declared because
+ // PinotApproximateAggregateRewriteRule pins the rewritten call to
PERCENTILE's type, which is ARG0. Without it a
+ // leaf that returns the final result is typed from that pinned type and
truncates on a non-DOUBLE column.
PERCENTILESMARTTDIGEST("percentileSmartTDigest", ReturnTypes.DOUBLE,
OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.NUMERIC,
SqlTypeFamily.CHARACTER), i -> i == 2),
- SqlTypeName.OTHER),
+ SqlTypeName.OTHER, SqlTypeName.DOUBLE),
PERCENTILEKLL("percentileKLL", ReturnTypes.DOUBLE,
OperandTypes.family(List.of(SqlTypeFamily.ANY, SqlTypeFamily.NUMERIC,
SqlTypeFamily.INTEGER), i -> i == 2),
SqlTypeName.OTHER),
@@ -371,6 +377,13 @@ public enum AggregationFunctionType {
public static AggregationFunctionType getAggregationFunctionType(String
functionName) {
String normalizedFunctionName =
getNormalizedAggregationFunctionName(functionName);
if (normalizedFunctionName.regionMatches(false, 0, "PERCENTILE", 0, 10)) {
+ // A canonical name wins over the numeric-suffix spellings below.
Without this, names that are canonical but
+ // are not one of those spellings, such as PERCENTILESMARTTDIGEST, would
fall through and be rejected.
+ try {
+ return AggregationFunctionType.valueOf(normalizedFunctionName);
+ } catch (IllegalArgumentException ignored) {
+ // Not a canonical name, so try the numeric-suffix spellings.
+ }
// This style of aggregation functions is not supported in the
multistage engine
String remainingFunctionName =
normalizedFunctionName.substring(10).toUpperCase();
if (remainingFunctionName.isEmpty() ||
remainingFunctionName.matches("\\d+")) {
@@ -400,7 +413,7 @@ public enum AggregationFunctionType {
} else if (remainingFunctionName.equals("KLLMV") ||
remainingFunctionName.matches("KLL\\d+MV")) {
return PERCENTILEKLLMV;
} else if (remainingFunctionName.equals("RAWKLLMV") ||
remainingFunctionName.matches("RAWKLL\\d+MV")) {
- return PERCENTILEKLLMV;
+ return PERCENTILERAWKLLMV;
} else {
throw new IllegalArgumentException("Invalid aggregation function name:
" + functionName);
}
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 bb29493d497..7fbf935c687 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
@@ -541,10 +541,34 @@ public class CommonConstants {
public static final String DISABLE_GROOVY =
"pinot.broker.disable.query.groovy";
public static final boolean DEFAULT_DISABLE_GROOVY = true;
- // Rewrite potential expensive functions to their approximation
counterparts
- // - DISTINCT_COUNT -> DISTINCT_COUNT_SMART_HLL
- // - PERCENTILE -> PERCENTILE_SMART_TDIGEST
+ /// Rewrite potential expensive functions to their approximation
counterparts, in both query engines:
+ /// - DISTINCT_COUNT and COUNT(DISTINCT) -> DISTINCT_COUNT_SMART_HLL
+ /// - PERCENTILE -> PERCENTILE_SMART_TDIGEST
+ ///
+ /// The rewritten functions stay exact until an accumulator exceeds their
conversion threshold, so this bounds
+ /// server memory without changing the answer for low-cardinality inputs.
+ ///
+ /// Settable in the broker conf, read once at startup, or in the Helix
cluster config, which wins and which
+ /// brokers pick up without a restart. Both are defaults: the
`useApproximateFunction` query option overrides
+ /// them, and so does `QueryConfig.useApproximateFunction`, though only in
the single-stage engine, because a
+ /// multi-stage query can span tables and so resolves the setting before
it knows the table set.
public static final String USE_APPROXIMATE_FUNCTION =
"pinot.broker.use.approximate.function";
+ public static final boolean DEFAULT_USE_APPROXIMATE_FUNCTION = false;
+
+ /// Parameters passed verbatim as the trailing argument of the calls the
rewrite produces, for example
+ /// `threshold=10000;log2m=12;dictThreshold=10000` and
`threshold=1000;compression=100`. Empty means no argument
+ /// is added, so the aggregation function defaults apply. The two
functions reject each other's parameter names,
+ /// hence one key each.
+ ///
+ /// Before conversion a group holds up to `threshold` values; after it, a
sketch whose registers are allocated
+ /// eagerly, around 2.7 KB at `log2m=12`. A low threshold therefore trades
raw values for sketches and can raise
+ /// group-by memory rather than lower it, so size `threshold` together
with `log2m` or `compression` and with
+ /// `pinot.server.query.executor.num.groups.limit` rather than in
isolation.
+ public static final String APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS =
+ "pinot.broker.approximate.function.distinct.count.params";
+ public static final String APPROXIMATE_FUNCTION_PERCENTILE_PARAMS =
+ "pinot.broker.approximate.function.percentile.params";
+ public static final String DEFAULT_APPROXIMATE_FUNCTION_PARAMS = "";
public static final String CONTROLLER_URL = "pinot.broker.controller.url";
@@ -762,6 +786,9 @@ public class CommonConstants {
public static class QueryOptionKey {
public static final String TIMEOUT_MS = "timeoutMs";
+ /// Per-query override of
[CommonConstants.Broker#USE_APPROXIMATE_FUNCTION], outranking both the table
config
+ /// and the cluster or broker default. `false` forces exact results,
`true` opts one expensive query in.
+ public static final String USE_APPROXIMATE_FUNCTION =
"useApproximateFunction";
/// Broker-internal marker set on the rewritten server-side PinotQuery
after a FULL_REWRITE
/// materialized-view rewrite. Read by BrokerReduceService to
distinguish MV-rewritten
/// queries from gapfill / future federated paths without relying on a
brittle structural
@@ -1157,6 +1184,8 @@ public class CommonConstants {
public static final String AGGREGATE_UNION_TRANSPOSE =
"AggregateUnionTranspose";
public static final String AGGREGATE_REDUCE_FUNCTIONS =
"AggregateReduceFunctions";
public static final String AGGREGATE_FUNCTION_REWRITE =
"AggregateFunctionRewrite";
+ /// Inert unless [CommonConstants.Broker#USE_APPROXIMATE_FUNCTION] is
on; name it here to switch it off entirely.
+ public static final String APPROXIMATE_AGGREGATE_REWRITE =
"ApproximateAggregateRewrite";
public static final String AGGREGATE_CASE_TO_FILTER =
"AggregateCaseToFilter";
public static final String PROJECT_FILTER_TRANSPOSE =
"ProjectFilterTranspose";
public static final String PROJECT_MERGE = "ProjectMerge";
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
