gortiz commented on code in PR #19510:
URL: https://github.com/apache/pinot/pull/19510#discussion_r4028124996
##########
pinot-segment-spi/src/main/java/org/apache/pinot/segment/spi/AggregationFunctionType.java:
##########
@@ -400,7 +410,7 @@ public static AggregationFunctionType
getAggregationFunctionType(String function
} 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;
Review Comment:
**The one pre-merge ask.**
This changes `percentileRawKLLMV` / `percentileRawKLL<n>MV` from DOUBLE to
VARCHAR for existing users. Worth the `backward-incompat` label plus a release
note.
One detail worth stating explicitly in the note so readers don't have to
derive it: the single-stage engine was already correct here —
`AggregationFunctionFactory` dispatches `RAWKLLMV` to
`PercentileRawKLLMVAggregationFunction` — so the blast radius is the
multi-stage planner.
Separately, neither `getAggregationFunctionType` fix is pinned by a test. A
parameterized one over `values()` asserting
`getAggregationFunctionType(t.getName()) == t` would lock the whole table down
and would have caught this years ago. Follow-up is fine.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * 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. Placing them in the
immediate project is what lets
+ // PinotAggregateExchangeNodeInsertRule inline them back into the
pushed-down call. With no parameters configured
+ // the plan keeps the shape it had before, with no extra project.
+ 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);
Review Comment:
*Optional.*
Worth a comment naming the contract: both
`PinotAggregateExchangeNodeInsertRule.buildAggCalls` and
`AggregatePushdownRule` inline this literal only via
`findImmediateProjects(input)`, so the project has to stay directly beneath the
aggregate or the leaf gets a column reference where the function expects a
parameter string.
I convinced myself nothing in BASIC/LOGICAL breaks it —
`AggregateProjectMergeRule` needs an all-input-ref project, so the literals
block it, and `ProjectMergeRule` collapses into one project that stays adjacent
— but it's an invariant held implicitly across two phases with nothing stating
it. The planner tests are all single-table `GROUP BY col` with no intervening
project; one over a subquery or join would stress it.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotApproximateAggregateRewriteRule.java:
##########
@@ -0,0 +1,186 @@
+/**
+ * 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. Placing them in the
immediate project is what lets
+ // PinotAggregateExchangeNodeInsertRule inline them back into the
pushed-down call. With no parameters configured
+ // the plan keeps the shape it had before, with no extra project.
+ 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,
Review Comment:
*Optional.*
Pinning is right, but it quietly invalidates an assumption `buildAggCall`
depends on: that fallback does `returnType = orgAggCall.getType()`, which is
only sound because Calcite normally derived the call's type *from* that
function's own `returnTypeInference`. After the rewrite, the call carries
`PERCENTILE`'s `ARG0` while the function is `PERCENTILESMARTTDIGEST`.
Concretely, with `/*+ aggOptions(is_leaf_return_final_result='true') */`
over an INT column, the LEAF agg is typed INTEGER instead of DOUBLE and
`TypeUtils.convertRow` does `((Number) value).intValue()` — a silent truncation
at the leaf, where unrewritten `PERCENTILE` carries the DOUBLE across the
exchange and coerces only at FINAL. Opt-in hint only, non-DOUBLE columns only,
no crash.
Cheapest fix: declare `SqlTypeName.DOUBLE` on `PERCENTILESMARTTDIGEST` (a
no-op for direct callers, whose call type is already DOUBLE) with a comment
saying it's there because of this pinning. Note it's redundant under that
field's documented convention — `PERCENTILESMARTTDIGEST`'s standard return type
already *is* DOUBLE — which is exactly why it reads as deliberate rather than
missing.
The root-cause fix (having that fallback consult
`functionType.getReturnTypeInference()`) would immunise the next rule that does
this, but it re-derives types for every aggregate in both split rules, so it
deserves its own diff rather than riding along here.
##########
pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java:
##########
@@ -500,10 +500,32 @@ public static class Broker {
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.
+ ///
+ /// Worst-case memory for a group-by is `threshold` values per group, so
size this together with
+ /// `pinot.server.query.executor.num.groups.limit` rather than in
isolation.
+ public static final String APPROXIMATE_FUNCTION_DISTINCT_COUNT_PARAMS =
Review Comment:
*Optional, docs.*
The "worst case is `threshold` values per group" guidance describes only the
pre-conversion state. After conversion a group holds a sketch whose register
set is allocated eagerly (~2.7 KB at `log2m=12`), so a **low** threshold can
increase group-by memory: at `threshold=1000` with a 100k `numGroupsLimit`
that's ~270 MB of sketches, unconditionally.
Worth putting `log2m` / `compression` in the same sentence as `threshold`,
since operators tuning this down is the whole point of the key.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/ApproximateFunctionOverrideProvider.java:
##########
@@ -0,0 +1,145 @@
+/**
+ * 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.
+ String enabled = clusterConfigs.get(Broker.USE_APPROXIMATE_FUNCTION);
Review Comment:
*Nit.*
The params get careful validation; the enable flag gets none. `ture`, `1`,
`yes`, `TRUE ` (trailing space) all become `false` — and for an OOM guardrail,
silently-off is the worse failure direction. Rejecting anything that isn't
`equalsIgnoreCase` true/false and logging it would match the care taken just
below.
##########
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 {
Review Comment:
*Follow-up.*
The standalone cluster is justified here — a CLUSTER-scoped Helix config
that changes every query would leak into a shared fixture, and the
`@AfterMethod` + `waitForOverride` shape is right.
Two gaps worth a follow-up: nothing covers the table-level
`QueryConfig.useApproximateFunction` layer, including the documented "MSE
ignores it" asymmetry (the most surprising thing in the feature), and the
`APPROXIMATE_FUNCTION_OVERRIDES` meter is never asserted in either engine.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java:
##########
@@ -61,6 +61,10 @@ protected
BaseDistinctCountSmartSketchAggregationFunction(ExpressionContext expr
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);
Review Comment:
*Optional, test coverage.*
Three implementations, only the HLL one is exercised.
`DistinctCountSmartULL` routes through
`UltraLogLogUtils.hashObject(...).ifPresent(...)`, which silently drops values
the hasher rejects — worth one test that a post-conversion value actually lands.
Also missing: a STRING column (the other `ObjectOpenHashSet` branch) and a
null-handling-enabled case, given the new pass iterates the whole batch rather
than the non-null ranges.
##########
pinot-core/src/main/java/org/apache/pinot/core/query/aggregation/function/BaseDistinctCountSmartSketchAggregationFunction.java:
##########
@@ -383,6 +387,7 @@ public final void aggregateGroupBySV(int length, int[]
groupKeyArray, GroupByRes
throw getIllegalDataTypeException(valueType, false);
}
}
+ checkAndConvertValueSetsForGroups(groupByResultHolder, groupKeyArray,
length, storedType);
Review Comment:
*Optional, performance.*
The dictionary branch just above deliberately collects `modifiedGroups` in
an `IntOpenHashSet` so the cardinality check runs once per group per batch.
This one re-checks once per **row**: for a 10k-row block that's ~10k extra
`getResult()` + `instanceof`, on top of the `getResult()` that `getValueSet()`
already does per row. It also walks `0..length` rather than the
`forEachNotNull` ranges, so all-null rows are visited too.
That cost lands on every existing user of these four functions, not just
those who enable the new config.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/BaseBrokerRequestHandler.java:
##########
@@ -120,6 +123,7 @@ public BaseBrokerRequestHandler(PinotConfiguration config,
String brokerId,
_tableCache = tableCache;
_threadAccountant = threadAccountant;
_multiClusterRoutingContext = multiClusterRoutingContext;
+ _approximateFunctionOverrideProvider = new
ApproximateFunctionOverrideProvider(config);
Review Comment:
*Follow-up.*
`TimeSeriesRequestHandler` also extends this, so it builds a third
`ApproximateFunctionOverrideProvider` that `BaseBrokerStarter` never registers
as a listener. Harmless today since it's unused, but the next handler to read
the setting silently gets broker-conf-only values with no live reload, and
nothing in the code says so.
Building one in the starter, registering it once, and passing it into the
handlers would make the wiring impossible to get wrong (and would halve the
parse work on every cluster config change).
##########
pinot-common/src/main/java/org/apache/pinot/common/response/BrokerResponse.java:
##########
@@ -311,6 +311,15 @@ default long getRealtimeTotalMemAllocatedBytes() {
/// @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.
+ /// @param approximateFunctionApplied true if at least one function was
rewritten
+ void setApproximateFunctionApplied(boolean approximateFunctionApplied);
Review Comment:
*Nit.*
`getMaterializedViewQueried` a few lines down is a `default` precisely so
implementations that don't track it need no change. Only two classes implement
this in-tree, but `pinot-common` is on out-of-tree compile classpaths —
`default boolean isApproximateFunctionApplied() { return false; }` keeps it
additive at no cost.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
