yashmayya commented on code in PR #20061:
URL: https://github.com/apache/pinot/pull/20061#discussion_r4406726392
##########
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:
**Blocker: queries that work today now fail or change results, also with the
gate off.**
`ARRAY_TO_MV(col)` is now nullable, whatever `useMvAggregate` says. On
master it is NOT NULL, so Calcite removes `COALESCE` and `CASE` around it and
folds `IS [NOT] NULL`. Now these reach the leaf stage.
Table with column-based null handling (MV columns `strMv`, `intMv` with
`notNull: false`, one null row). Master vs this PR, with the gate off and on,
`enableNullHandling` true and false, in v1, v2 and lite:
| Query | master | this PR |
|---|---|---|
| `SELECT COALESCE(ARRAY_TO_MV(strMv), 'none') v, COUNT(*) FROM t GROUP BY
1` | works | fails: `GroupByOperator: null` |
| `SELECT COUNT(*) FROM t WHERE COALESCE(ARRAY_TO_MV(strMv), 'x') = 'a'` |
works | fails: `AggregationOperator: null` |
| `GROUP BY CASE WHEN ARRAY_TO_MV(strMv) IS NULL THEN 'none' ELSE
ARRAY_TO_MV(strMv) END` | works | fails: `Unsupported multi-value expression in
THEN/ELSE clause` |
| The same shapes with `COALESCE(ARRAY_TO_MV(intMv), -1)` | works | fails |
With `enableNullHandling=true`, results also change: `WHERE
ARRAY_TO_MV(strMv) IS NOT NULL` now drops the null row (4 rows become 3), and
`IS NULL` now returns it (0 rows become 1).
Please keep the NOT NULL type when the gate is off, for example by choosing
the operator from the effective gate, as for `PINOT_IS_NULL`. When the gate is
on, the shapes above also need to keep working.
##########
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:
**Blocker: plans change with the gate off.**
This exclusion also applies when `useMvAggregate=false`. `GROUP BY
ARRAY_TO_MV(...)` with an `IN` or `EXISTS` subquery works on master: the
semi-join becomes a dynamic broadcast, and `ARRAY_TO_MV` goes below it into the
leaf. Now it stays above the join, so the plan changes and the leaf stage gets
an extra PROJECT. The results are the same in our runs.
Repro: `SET useMvAggregate=false; EXPLAIN PLAN FOR SELECT
ARRAY_TO_MV(mcol1), COUNT(*) FROM e WHERE col1 IN (SELECT col1 FROM a WHERE
col2 = 'x') GROUP BY ARRAY_TO_MV(mcol1)`
- master: `Aggregate(group={1})` <- semi `Join` <- `Project(col1,
ARRAY_TO_MV($7))`
- this PR: `Aggregate(group={0})` <- `Project(ARRAY_TO_MV($1))` <- semi
`Join` <- `Project(col1, mcol1)`
The PR text says that these queries failed before. That is not true for this
shape. Please apply this exclusion only when the gate is on.
##########
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:
**Blocker: removing the cluster key does not turn the gate off on brokers
that restarted while the key was set.**
`BaseBrokerStarter.init()` copies the cluster config keys into `_brokerConf`
(`ServiceStartableUtils.applyClusterConfig`, add if absent) before this handler
is created. So a broker that starts while the cluster key is `true` reads
`true` here as its broker value. When the operator then removes the key,
`onChange` falls back to `true` on that broker, and to `false` on brokers that
did not restart.
This breaks the rollback in the docs: remove the key, roll back a server,
and the restarted brokers keep sending `MvAggregateNode` to it, so their
queries fail.
Fix: pin the broker value before `applyClusterConfig`, as `init()` already
does for `CONFIG_OF_BROKER_ENABLE_QUERY_FINGERPRINTING`. A test can start from
a config that has the cluster key set to `true` and no broker value, remove the
key, and expect the gate to be `false`.
##########
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:
**Blocker: `HAVING` on an MV key returns wrong groups when the gate is on.**
Data: `orders` has `(o1, c1, tags=[urgent, vip])` and `(o4, c2,
tags=[vip])`. `customers` has `c1` and `c2`.
```sql
SELECT ARRAY_TO_MV(o.tags), COUNT(*)
FROM orders o JOIN customers c ON o.customer_id = c.customer_id
GROUP BY ARRAY_TO_MV(o.tags)
HAVING ARRAY_TO_MV(o.tags) = 'vip'
```
- Expected: `[vip, 2]`
- This PR: `[urgent, 1], [vip, 2]`
- Master: fails with an error. So this is new wrong output, not an old one.
Cause: `FilterAggregateTransposeRule` pushes the `HAVING` condition below
the aggregate and then into the leaf `WHERE`. There, an MV predicate keeps the
whole row when any value matches, so the other values of that row also get
groups. Master has the same bug on a single table (`GROUP BY ARRAY_TO_MV(tags)
HAVING ARRAY_TO_MV(tags) = 'vip'` also returns `urgent`). This PR makes it
reachable above joins and subqueries. A filter on the key in an outer query
(`SELECT * FROM (... GROUP BY ARRAY_TO_MV(tags)) WHERE tag = 'vip'`) has the
same problem.
A fix that worked in my test: do not push a filter below an aggregate whose
input project computes a group key with `ARRAY_TO_MV`, using an operand
predicate on `FilterAggregateTransposeRule`. No existing plan snapshot changed.
--
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]
