gortiz commented on code in PR #20061:
URL: https://github.com/apache/pinot/pull/20061#discussion_r4415229177
##########
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:
Done. The check moved to `PinotRuleUtils.isMvAggregateEnabled`, which
`PinotProjectJoinTransposeRule` also uses. Without a context it returns false
now, and without a `PlannerContext` it plans the MV aggregate only when the
config mode is `true`.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java:
##########
@@ -141,9 +142,21 @@ public static PinotOperatorTable instance(boolean
nullHandlingEnabled) {
/// [org.apache.pinot.calcite.rel.rules.PinotRuleUtils#isRelocatable]
ignores this flag, so ARRAY_TO_MV filters are
/// still pushed below joins.
public static final PinotSqlFunction ARRAY_TO_MV =
- new PinotSqlFunction("ARRAY_TO_MV", opBinding ->
opBinding.getOperandType(0).getComponentType(),
+ new PinotSqlFunction(ArrayToMvFunctionName.NAME,
+ opBinding -> arrayToMvReturnType(opBinding.getTypeFactory(),
opBinding.getOperandType(0)),
OperandTypes.ARRAY, false);
+ /// The type of `ARRAY_TO_MV(array)`: the component type of `arrayType`,
nullable when the array or its component
+ /// is nullable. The MV aggregate gives its MV keys the same type, so that
it can replace a plain aggregate over
+ /// `ARRAY_TO_MV(array)` without changing the row type.
+ public static RelDataType arrayToMvReturnType(RelDataTypeFactory
typeFactory, RelDataType arrayType) {
+ RelDataType componentType = arrayType.getComponentType();
+ Preconditions.checkArgument(componentType != null, "ARRAY_TO_MV requires
an array, got: %s",
+ arrayType.getFullTypeString());
+ return typeFactory.createTypeWithNullability(componentType,
+ arrayType.isNullable() || componentType.isNullable());
+ }
Review Comment:
Fixed. `ARRAY_TO_MV` has the component type of the array again, as on
master, in every mode. `auto` does not need a different type, because the
second pass reuses the converted plan. I added runtime cases with column-based
null handling for `COALESCE(ARRAY_TO_MV(col), ...)` in a group key (scan and
above a join) and in a filter, and `IS NOT NULL` is back to the master answer.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotProjectJoinTransposeRule.java:
##########
@@ -21,17 +21,28 @@
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.rules.ProjectJoinTransposeRule;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
import org.apache.pinot.calcite.rel.hint.PinotHintOptions;
+import org.apache.pinot.calcite.sql.fun.ArrayToMvFunctionName;
/// Similar to [ProjectJoinTransposeRule] but do not transpose project into
right side of lookup join.
///
+/// An expression that contains `ARRAY_TO_MV` stays above the join; only the
input references it uses are pushed down.
+/// `GROUP BY ARRAY_TO_MV(col)` above a join is planned as an MV aggregate
only when the group expression is in the
+/// Project under the aggregate (see [ArrayToMvGroupKeys]). Below the join,
the expression would be in a leaf-stage
+/// Project, where `ARRAY_TO_MV` is not supported.
+///
/// TODO: Allow transposing project into left side of lookup join.
public class PinotProjectJoinTransposeRule extends ProjectJoinTransposeRule {
- public static final PinotProjectJoinTransposeRule INSTANCE = new
PinotProjectJoinTransposeRule(Config.DEFAULT);
+ private static final Config DEFAULT_CONFIG =
Config.DEFAULT.withPreserveExprCondition(
+ expr -> Config.DEFAULT.preserveExprCondition().test(expr) &&
!containsArrayToMv(expr));
Review Comment:
Fixed as in your prototype: the rule keeps `ARRAY_TO_MV` above the join only
while the MV aggregate is on. With `false`, and in the first pass of `auto`, it
pushes the expression down as before.
`MvAggregateGateTest.testGateOffPushesArrayToMvBelowTheSemiJoin` checks your
repro. You are right about the PR text too. I removed the claim that these
queries failed before.
##########
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,
Review Comment:
Fixed with your suggestion. `FilterAggregateTransposeRule` now has an
operand predicate that skips an aggregate whose group keys include
`ARRAY_TO_MV(expr)` (`ArrayToMvGroupKeys.detect`). It applies in every mode, so
it also fixes the single-table case on master. The plan uses only existing
nodes, so old servers can run it. The outer-query filter goes through the same
rule. `MvAggregate.json` has the `HAVING` case above a self-join, the
single-table case with the gate off, and the outer filter.
##########
pinot-broker/src/main/java/org/apache/pinot/broker/requesthandler/MultiStageBrokerRequestHandler.java:
##########
@@ -215,6 +219,9 @@ public MultiStageBrokerRequestHandler(PinotConfiguration
config, String brokerId
boolean dispatchKeepAliveWithoutCalls = config.getProperty(
CommonConstants.MultiStageQueryRunner.KEY_OF_DISPATCH_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS,
CommonConstants.MultiStageQueryRunner.DEFAULT_OF_DISPATCH_CHANNEL_KEEP_ALIVE_WITHOUT_CALLS);
+ _useMvAggregateBrokerConf =
_config.getProperty(CommonConstants.Broker.CONFIG_OF_USE_MV_AGGREGATE,
+ CommonConstants.Broker.DEFAULT_USE_MV_AGGREGATE);
Review Comment:
Fixed. `BaseBrokerStarter` now pins the broker value of
`pinot.broker.multistage.use.mv.aggregate` before `applyClusterConfig`, next to
the fingerprinting and `inferAnyValueType` keys. I moved the three pins into
`pinLiveClusterConfigDefaults` so `BaseBrokerStarterTest` can check it without
ZooKeeper.
--
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]
