xiangfu0 commented on code in PR #20061:
URL: https://github.com/apache/pinot/pull/20061#discussion_r4405023176
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotAggregateExchangeNodeInsertRule.java:
##########
@@ -271,6 +300,78 @@ private static RelNode createPlan(RelOptRuleCall call,
Aggregate aggRel, boolean
}
}
+ /// Splits a `GROUP BY ARRAY_TO_MV(expr)` aggregate into LEAF
[PinotLogicalMvAggregate] + EXCHANGE + FINAL:
+ /// - The Project under the aggregate gets `expr` (an array) instead of
`ARRAY_TO_MV(expr)` for each MV key.
+ /// - The LEAF MV aggregate groups by every element of the MV keys. Its
row type gives each MV key the component
+ /// type, so it equals the row type of the original aggregate.
+ /// - EXCHANGE: hash by all the group keys, which are scalars now.
+ /// - FINAL: a plain aggregate over the scalar keys.
+ ///
+ /// The hints `is_skip_leaf_stage_group_by` and
`is_partitioned_by_group_by_keys` are ignored: a skip-leaf plan would
+ /// hash whole arrays and a partitioned plan would assume a partitioning by
element that no partition function gives.
+ ///
+ /// Returns null, so that the caller plans a plain aggregate as before, when
an MV key column is also an argument or
+ /// the filter of an aggregate call: rewriting the column would change the
argument from the element type to the
+ /// array type.
+ @Nullable
+ private static RelNode createMvAggregatePlan(Aggregate aggRel, List<Integer>
mvKeys, Map<String, String> hintOptions,
+ @Nullable RelCollation withinGroupCollation, @Nullable
List<RelFieldCollation> collations, int limit) {
+ if (withinGroupCollation != null) {
+ throw new UnsupportedOperationException(
+ "GROUP BY ARRAY_TO_MV() is not supported with WITHIN GROUP ordered
aggregates in the multi-stage query "
+ + "engine");
+ }
+ List<Integer> mvColumns = ArrayToMvGroupKeys.toInputColumns(aggRel,
mvKeys);
+ for (AggregateCall aggCall : aggRel.getAggCallList()) {
+ if (aggCall.filterArg >= 0 && mvColumns.contains(aggCall.filterArg)) {
+ LOGGER.debug("Not planning an MV aggregate: MV key column {} is the
filter of {}", aggCall.filterArg, aggCall);
+ return null;
+ }
+ for (int arg : aggCall.getArgList()) {
+ if (mvColumns.contains(arg)) {
+ LOGGER.debug("Not planning an MV aggregate: MV key column {} is an
argument of {}", arg, aggCall);
+ return null;
+ }
+ }
+ }
+ if (LOGGER.isDebugEnabled() && (Boolean.parseBoolean(
+
hintOptions.get(PinotHintOptions.AggregateOptions.IS_SKIP_LEAF_STAGE_GROUP_BY))
|| Boolean.parseBoolean(
+
hintOptions.get(PinotHintOptions.AggregateOptions.IS_PARTITIONED_BY_GROUP_BY_KEYS))))
{
+ LOGGER.debug("Ignoring the hints {} and {} for the MV aggregate: {}",
+ PinotHintOptions.AggregateOptions.IS_SKIP_LEAF_STAGE_GROUP_BY,
+ PinotHintOptions.AggregateOptions.IS_PARTITIONED_BY_GROUP_BY_KEYS,
aggRel);
+ }
+ boolean leafReturnFinalResult =
+
Boolean.parseBoolean(hintOptions.get(PinotHintOptions.AggregateOptions.IS_LEAF_RETURN_FINAL_RESULT));
+ // ArrayToMvGroupKeys.detect returns keys only when the unboxed input is a
Project.
+ Project project = (Project) PinotRuleUtils.unboxRel(aggRel.getInput());
+ Project newProject = ArrayToMvGroupKeys.rewriteProject(project, mvColumns);
+ PinotLogicalMvAggregate leafAggRel =
+ new PinotLogicalMvAggregate(aggRel.getCluster(), aggRel.getTraitSet(),
aggRel.getHints(), newProject,
+ aggRel.getGroupSet(), null,
+ buildAggCalls(aggRel, newProject, aggRel.getAggCallList(),
AggType.LEAF, leafReturnFinalResult),
+ AggType.LEAF, leafReturnFinalResult, collations, limit, mvKeys);
+ PinotLogicalExchange exchange = PinotLogicalExchange.create(leafAggRel,
+ RelDistributions.hash(ImmutableIntList.range(0,
aggRel.getGroupCount())));
+ return convertAggFromIntermediateInput(aggRel, exchange, AggType.FINAL,
leafReturnFinalResult, collations, limit);
+ }
+
+ /// Whether `GROUP BY ARRAY_TO_MV(col)` is planned as an MV aggregate. The
query option `useMvAggregate` decides when
+ /// the query sets it. Otherwise
[QueryEnvironment.Config#defaultUseMvAggregate()] decides.
+ private static boolean isMvAggregateEnabled(RelOptRuleCall call) {
+ Context genericContext = call.getPlanner().getContext();
+ if (genericContext == null) {
+ return true;
Review Comment:
Minor: with no planner context, or no `QueryEnvironment.Config`, this
returns `true`, which turns the MV aggregate on. That is the opposite of the
gate's default (`DEFAULT_USE_MV_AGGREGATE = false`). The group-trim helpers
below fall back to the `CommonConstants` default. Production always sets a
`PlannerContext`, so this path is unreachable today. Still, a gate that
protects rolling upgrades should fail closed, so consider returning
`CommonConstants.Broker.DEFAULT_USE_MV_AGGREGATE` in both places.
--
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]
