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 e574c52eff6 Push ARRAY_TO_MV filters below joins again in MSE (#19973)
e574c52eff6 is described below
commit e574c52eff6fc6c1544e8270f465565c8e51e03b
Author: Yash Mayya <[email protected]>
AuthorDate: Wed Oct 21 11:49:24 2026 -0700
Push ARRAY_TO_MV filters below joins again in MSE (#19973)
---
.../pinot/calcite/rel/rules/PinotRuleUtils.java | 18 +++----
.../pinot/calcite/sql/fun/PinotOperatorTable.java | 32 ++++++++-----
.../validation/ArrayToMvValidationVisitor.java | 3 +-
.../calcite/rel/rules/PinotRuleUtilsTest.java | 19 ++++++++
.../src/test/resources/queries/JoinPlans.json | 17 +++++++
.../resources/queries/ValidationErrorPlan.json | 5 ++
.../src/test/resources/queries/Joins.json | 56 ++++++++++++++++++++++
7 files changed, 128 insertions(+), 22 deletions(-)
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtils.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtils.java
index bb7f98a24c4..6bbb888ebab 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtils.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtils.java
@@ -42,7 +42,6 @@ import org.apache.calcite.rex.RexCall;
import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.rex.RexUtil;
import org.apache.calcite.rex.RexVisitorImpl;
import org.apache.calcite.rex.RexWindowBound;
import org.apache.calcite.rex.RexWindowBounds;
@@ -54,6 +53,7 @@ import org.apache.calcite.tools.RelBuilder;
import org.apache.calcite.tools.RelBuilderFactory;
import org.apache.calcite.util.Util;
import org.apache.pinot.calcite.rel.hint.PinotHintStrategyTable;
+import org.apache.pinot.calcite.sql.fun.PinotOperatorTable;
import org.apache.pinot.common.function.sql.PinotSqlFunction;
@@ -147,13 +147,17 @@ public class PinotRuleUtils {
/// An expression must be clear of three axes of variability:
///
/// - [SqlOperator#isDeterministic()] -- `false` for `rand()`, `UUID_V4`,
`UUID_V7` and Calcite's own `RAND` /
- /// `RAND_INTEGER`. Delegated to `RexUtil#isDeterministic` so this half
tracks upstream automatically.
+ /// `RAND_INTEGER`.
/// - [SqlOperator#isDynamicFunction()] -- Calcite's own "fold once per
query, never re-evaluate" marker, used by
/// `CURRENT_TIMESTAMP` and friends.
/// - [PinotSqlFunction#isVolatile()] -- Pinot's equivalent marker, `true`
for `FunctionVolatility.VOLATILE`
/// functions such as `now()`, `ago()` and `stageId()`. These deliberately
stay `isDeterministic() == true` so that
/// [PinotEvaluateLiteralRule] can still fold them once at plan time,
which is precisely why
- /// `RexUtil#isDeterministic` alone does not catch them.
+ /// [SqlOperator#isDeterministic()] alone does not catch them.
+ ///
+ /// [PinotOperatorTable#ARRAY_TO_MV] is the one exception. It is marked
non-deterministic (and so volatile) only to
+ /// stop Calcite from simplifying MV predicates such as `ARRAY_TO_MV(col) =
'a' AND ARRAY_TO_MV(col) = 'b'` to
+ /// `false`. Its value depends only on its row, so its own flags are ignored
here. Its operands are still checked.
///
/// Relocating an expression that fails this check changes how many times,
and in what context, it is evaluated --
/// which changes query results. `FunctionVolatility.STABLE` deliberately
passes: it is constant within a single
@@ -162,16 +166,14 @@ public class PinotRuleUtils {
/// Note this is a predicate that callers must apply; it is not enforced
globally. Only [PinotFilterJoinRule]
/// consults it today, so other rules that relocate expressions can still
move volatile ones.
public static boolean isRelocatable(RexNode node) {
- if (!RexUtil.isDeterministic(node)) {
- return false;
- }
try {
node.accept(new RexVisitorImpl<Void>(true) {
@Override
public Void visitCall(RexCall call) {
SqlOperator operator = call.getOperator();
- if (operator.isDynamicFunction()
- || (operator instanceof PinotSqlFunction && ((PinotSqlFunction)
operator).isVolatile())) {
+ boolean isVariable = !operator.isDeterministic() ||
operator.isDynamicFunction()
+ || (operator instanceof PinotSqlFunction && ((PinotSqlFunction)
operator).isVolatile());
+ if (isVariable && operator != PinotOperatorTable.ARRAY_TO_MV) {
throw Util.FoundOne.NULL;
}
return super.visitCall(call);
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java
index c6750013e74..c0bcb0996e3 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/sql/fun/PinotOperatorTable.java
@@ -126,6 +126,24 @@ public class PinotOperatorTable implements
SqlOperatorTable {
InferTypes.VARCHAR_1024,
OperandTypes.ANY);
+ /// Placeholder for special functions to handle MV.
+ ///
+ /// NOTE: ARRAY_TO_MV is not deterministic.
+ /// We need to explicitly set it as not deterministic in order to do not let
Calcite optimize expressions like
+ /// `ARRAY_TO_MV(RandomAirports) = 'MFR' and ARRAY_TO_MV(RandomAirports) =
'GTR'` as `false`.
+ /// If the function were deterministic, its value would never be MFR and GTR
at the same time, so Calcite is
+ /// smart enough to know there is no value that satisfies the condition.
+ /// In fact what ARRAY_TO_MV does is just to trick Calcite typesystem, but
then what the leaf stage executor
+ /// receives is `RandomAirports = 'MFR' and RandomAirports = 'GTR'`, which
in the V1 semantics means:
+ /// true if and only if RandomAirports contains a value equal to 'MFR' and
RandomAirports contains a value equal
+ /// to 'GTR'.
+ ///
+ /// [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(),
+ OperandTypes.ARRAY, false);
+
/// This list includes the supported standard [SqlOperator]s defined in
[SqlStdOperatorTable].
/// NOTE: The operator order follows the same order as defined in
[SqlStdOperatorTable] for easier search.
/// Some operators are commented out and re-declared in
[#STANDARD_OPERATORS_WITH_ALIASES].
@@ -305,19 +323,7 @@ public class PinotOperatorTable implements
SqlOperatorTable {
new PinotSqlFunction("IN_SUBQUERY", ReturnTypes.BOOLEAN_NULLABLE,
OperandTypes.family(SqlTypeFamily.ANY, SqlTypeFamily.CHARACTER)),
- // Placeholder for special functions to handle MV
- // NOTE:
- // ARRAY_TO_MV is not deterministic.
- // We need to explicitly set it as not deterministic in order to do not
let Calcite optimize expressions like
- // `ARRAY_TO_MV(RandomAirports) = 'MFR' and ARRAY_TO_MV(RandomAirports)
= 'GTR'` as `false`.
- // If the function were deterministic, its value would never be MFR and
GTR at the same time, so Calcite is
- // smart enough to know there is no value that satisfies the condition.
- // In fact what ARRAY_TO_MV does is just to trick Calcite typesystem,
but then what the leaf stage executor
- // receives is `RandomAirports = 'MFR' and RandomAirports = 'GTR'`,
which in the V1 semantics means:
- // true if and only if RandomAirports contains a value equal to 'MFR'
and RandomAirports contains a value equal
- // to 'GTR'
- new PinotSqlFunction("ARRAY_TO_MV", opBinding ->
opBinding.getOperandType(0).getComponentType(),
- OperandTypes.ARRAY, false),
+ ARRAY_TO_MV,
// SqlStdOperatorTable.COALESCE without rewrite
new SqlFunction("COALESCE", SqlKind.COALESCE,
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/validation/ArrayToMvValidationVisitor.java
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/validation/ArrayToMvValidationVisitor.java
index 6653f1f22fc..246a9aa2119 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/validation/ArrayToMvValidationVisitor.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/query/planner/validation/ArrayToMvValidationVisitor.java
@@ -115,7 +115,8 @@ public class ArrayToMvValidationVisitor implements
PlanNodeVisitor<Void, Boolean
if (containsArrayToMv(node.getProjects())) {
throw new QueryException(QueryErrorCode.QUERY_PLANNING,
"Function 'ArrayToMv' is not supported in PROJECT " +
(isIntermediateStage ? "Intermediate Stage"
- : "Leaf Stage"));
+ : "Leaf Stage") + ". To get one row for each value of a
multi-value column, use "
+ + "'CROSS JOIN UNNEST(mvCol) AS t(val)' instead");
}
node.getInputs().forEach(e -> e.visit(this, isIntermediateStage));
return null;
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtilsTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtilsTest.java
index 30fba46e27f..5447ac59dd2 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtilsTest.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtilsTest.java
@@ -40,6 +40,7 @@ import org.testng.annotations.Test;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertSame;
import static org.testng.Assert.assertTrue;
@@ -156,4 +157,22 @@ public class PinotRuleUtilsTest {
RexNode nested = call(SqlStdOperatorTable.PLUS, literal(1),
call(registryFunction("stageId"), literal(0)));
assertFalse(PinotRuleUtils.isRelocatable(nested));
}
+
+ /// `ARRAY_TO_MV` is flagged non-deterministic only to stop Calcite from
folding MV predicates, so it must still be
+ /// relocatable (otherwise MV filters stay above joins and fail validation).
Its operands must still be checked.
+ @Test
+ public void testArrayToMvIsRelocatable() {
+ SqlOperator arrayToMv = resolvedOperator("arrayToMv");
+ assertSame(arrayToMv, PinotOperatorTable.ARRAY_TO_MV, "isRelocatable
matches ARRAY_TO_MV by identity");
+ assertFalse(arrayToMv.isDeterministic(), "ARRAY_TO_MV is expected to be
flagged non-deterministic");
+ RexNode mvCol = _rexBuilder.makeInputRef(
+
_typeFactory.createArrayType(_typeFactory.createSqlType(SqlTypeName.VARCHAR),
-1), 0);
+ RexNode predicate = _rexBuilder.makeCall(SqlStdOperatorTable.EQUALS,
+ _rexBuilder.makeCall(_typeFactory.createSqlType(SqlTypeName.VARCHAR),
arrayToMv, List.of(mvCol)),
+ _rexBuilder.makeLiteral("a"));
+ assertTrue(PinotRuleUtils.isRelocatable(predicate));
+
+ assertFalse(PinotRuleUtils.isRelocatable(call(arrayToMv,
call(registryFunction("rand")))));
+ assertFalse(PinotRuleUtils.isRelocatable(call(arrayToMv,
call(registryFunction("stageId"), literal(0)))));
+ }
}
diff --git a/pinot-query-planner/src/test/resources/queries/JoinPlans.json
b/pinot-query-planner/src/test/resources/queries/JoinPlans.json
index b68b0d4a0a5..3e5f6e90631 100644
--- a/pinot-query-planner/src/test/resources/queries/JoinPlans.json
+++ b/pinot-query-planner/src/test/resources/queries/JoinPlans.json
@@ -882,6 +882,23 @@
"\n PinotLogicalTableScan(table=[[default, b]])",
"\n"
]
+ },
+ {
+ "description": "ARRAY_TO_MV WHERE filter is pushed below the join
together with the other conjuncts",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, e.col3 FROM a JOIN e ON a.col1
= e.col1 WHERE e.col2 = 'x' AND ARRAY_TO_MV(e.mcol1) = 'a'",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col3=[$2])",
+ "\n LogicalJoin(condition=[=($0, $1)], joinType=[inner])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col3=[$2])",
+ "\n LogicalFilter(condition=[AND(=($1, _UTF-8'x'),
=(ARRAY_TO_MV($7), _UTF-8'a'))])",
+ "\n PinotLogicalTableScan(table=[[default, e]])",
+ "\n"
+ ]
}
]
},
diff --git
a/pinot-query-planner/src/test/resources/queries/ValidationErrorPlan.json
b/pinot-query-planner/src/test/resources/queries/ValidationErrorPlan.json
index 494f1cf138f..c7cf334b334 100644
--- a/pinot-query-planner/src/test/resources/queries/ValidationErrorPlan.json
+++ b/pinot-query-planner/src/test/resources/queries/ValidationErrorPlan.json
@@ -16,6 +16,11 @@
"sql": "SET usePhysicalOptimizer=true; EXPLAIN PLAN FOR SELECT
SUM(a.col3) as sumCol3, arrayToMv(e.mcol1), a.col2 FROM a JOIN e on
a.col1=e.col1 GROUP BY arrayToMv(e.mcol1), a.col2",
"expectedException": ".*'ArrayToMv' is not supported.*"
},
+ {
+ "description": "arrayToMV in a PROJECT error suggests CROSS JOIN
UNNEST",
+ "sql": "EXPLAIN PLAN FOR SELECT arrayToMv(e.mcol1), a.col2 FROM a JOIN
e ON a.col1 = e.col1",
+ "expectedException": ".*'ArrayToMv' is not supported in PROJECT.*CROSS
JOIN UNNEST.*"
+ },
{
"description": "Select * with negative limit -1",
"sql": "EXPLAIN PLAN FOR SELECT * FROM d LIMIT -1",
diff --git a/pinot-query-runtime/src/test/resources/queries/Joins.json
b/pinot-query-runtime/src/test/resources/queries/Joins.json
index 50ecf1cc976..f523163a2b9 100644
--- a/pinot-query-runtime/src/test/resources/queries/Joins.json
+++ b/pinot-query-runtime/src/test/resources/queries/Joins.json
@@ -108,5 +108,61 @@
"sql": "SELECT c.name, o.total_amount FROM {customers} c JOIN (SELECT
customer_id, SUM(amount) as total_amount FROM {orders} GROUP BY customer_id) o
ON c.customer_id = o.customer_id ORDER BY c.name"
}
]
+ },
+ "join_with_mv_filter": {
+ "comment": "ARRAY_TO_MV filters in WHERE must be pushed below the join to
the leaf stage, the only stage that can run them. An MV predicate matches a row
if any value matches.",
+ "tables": {
+ "orders": {
+ "schema": [
+ {"name": "order_id", "type": "STRING"},
+ {"name": "customer_id", "type": "STRING"},
+ {"name": "status", "type": "STRING"},
+ {"name": "tags", "type": "STRING", "isSingleValue": false}
+ ],
+ "inputs": [
+ ["o1", "c1", "open", ["urgent", "vip"]],
+ ["o2", "c1", "open", ["urgent"]],
+ ["o3", "c2", "closed", ["urgent"]],
+ ["o4", "c2", "open", ["vip"]],
+ ["o5", "c3", "open", ["normal"]]
+ ]
+ },
+ "customers": {
+ "schema": [
+ {"name": "customer_id", "type": "STRING"},
+ {"name": "name", "type": "STRING"}
+ ],
+ "inputs": [
+ ["c1", "Alice"],
+ ["c2", "Bob"],
+ ["c3", "Charlie"]
+ ]
+ }
+ },
+ "queries": [
+ {
+ "description": "ARRAY_TO_MV equality with another conjunct on the same
side",
+ "sql": "SELECT o.order_id, c.name FROM {orders} o JOIN {customers} c
ON o.customer_id = c.customer_id WHERE o.status = 'open' AND
ARRAY_TO_MV(o.tags) = 'urgent' ORDER BY o.order_id",
+ "outputs": [
+ ["o1", "Alice"],
+ ["o2", "Alice"]
+ ]
+ },
+ {
+ "description": "Two ARRAY_TO_MV equalities on the same column are not
folded to false",
+ "sql": "SELECT o.order_id, c.name FROM {orders} o JOIN {customers} c
ON o.customer_id = c.customer_id WHERE ARRAY_TO_MV(o.tags) = 'urgent' AND
ARRAY_TO_MV(o.tags) = 'vip' ORDER BY o.order_id",
+ "outputs": [
+ ["o1", "Alice"]
+ ]
+ },
+ {
+ "description": "ARRAY_TO_MV IN filter with a filter on the other side
and an aggregate above the join",
+ "sql": "SELECT c.name, COUNT(*) FROM {orders} o JOIN {customers} c ON
o.customer_id = c.customer_id WHERE ARRAY_TO_MV(o.tags) IN ('urgent', 'vip')
AND c.name <> 'Charlie' GROUP BY c.name ORDER BY c.name",
+ "outputs": [
+ ["Alice", 2],
+ ["Bob", 2]
+ ]
+ }
+ ]
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
