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 fa3fec3591b Don't push non-deterministic or volatile filters below a
join (#19216)
fa3fec3591b is described below
commit fa3fec3591b6288a2d5a7c8d7e0fa454fbdd0ee1
Author: Yash Mayya <[email protected]>
AuthorDate: Tue Aug 18 14:49:03 2026 -0400
Don't push non-deterministic or volatile filters below a join (#19216)
---
.../pinot/common/function/FunctionRegistry.java | 16 +-
.../common/function/sql/PinotSqlFunction.java | 26 ++-
.../pinot/common/function/FunctionUtilsTest.java | 34 ++++
.../calcite/rel/rules/PinotFilterJoinRule.java | 30 +++
.../pinot/calcite/rel/rules/PinotRuleUtils.java | 48 +++++
.../pinot/calcite/sql/fun/PinotOperatorTable.java | 5 +-
.../calcite/rel/rules/PinotRuleUtilsTest.java | 159 +++++++++++++++
.../apache/pinot/query/QueryCompilationTest.java | 28 +++
.../src/test/resources/queries/JoinPlans.json | 220 +++++++++++++++++++++
9 files changed, 563 insertions(+), 3 deletions(-)
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java
index f88b73cf740..43b08241967 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/FunctionRegistry.java
@@ -39,6 +39,7 @@ import org.apache.calcite.sql.type.SqlTypeFamily;
import org.apache.commons.lang3.StringUtils;
import org.apache.pinot.common.function.sql.PinotSqlFunction;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
+import org.apache.pinot.spi.annotations.FunctionVolatility;
import org.apache.pinot.spi.annotations.ScalarFunction;
import org.apache.pinot.spi.utils.PinotReflectionUtils;
import org.slf4j.Logger;
@@ -249,7 +250,8 @@ public class FunctionRegistry {
@Override
public PinotSqlFunction toPinotSqlFunction() {
- return new PinotSqlFunction(_mainName, getReturnTypeInference(),
getOperandTypeChecker(), isDeterministic());
+ return new PinotSqlFunction(_mainName, getReturnTypeInference(),
getOperandTypeChecker(), isDeterministic(),
+ isVolatile());
}
private SqlReturnTypeInference getReturnTypeInference() {
@@ -328,6 +330,18 @@ public class FunctionRegistry {
return true;
}
+ /// Conservatively reports the function as volatile if any registered
overload is, matching [#isDeterministic()].
+ /// Both are operator-level properties, whereas the operand types that
would select a single overload are only
+ /// known per call site.
+ private boolean isVolatile() {
+ for (FunctionInfo functionInfo : _functionInfoMap.values()) {
+ if (functionInfo.getVolatility() == FunctionVolatility.VOLATILE) {
+ return true;
+ }
+ }
+ return false;
+ }
+
@Override
public String getScalarFunctionId() {
if (_functionInfoMap.size() == 1) {
diff --git
a/pinot-common/src/main/java/org/apache/pinot/common/function/sql/PinotSqlFunction.java
b/pinot-common/src/main/java/org/apache/pinot/common/function/sql/PinotSqlFunction.java
index 554389947e1..f18bfb079a9 100644
---
a/pinot-common/src/main/java/org/apache/pinot/common/function/sql/PinotSqlFunction.java
+++
b/pinot-common/src/main/java/org/apache/pinot/common/function/sql/PinotSqlFunction.java
@@ -28,12 +28,25 @@ import org.apache.calcite.sql.type.SqlReturnTypeInference;
/// Pinot custom SqlFunction to be registered into SqlOperatorTable.
public class PinotSqlFunction extends SqlFunction {
private final boolean _deterministic;
+ private final boolean _isVolatile;
public PinotSqlFunction(String name, SqlReturnTypeInference
returnTypeInference,
- SqlOperandTypeChecker operandTypeChecker, boolean deterministic) {
+ SqlOperandTypeChecker operandTypeChecker, boolean deterministic, boolean
isVolatile) {
super(name.toUpperCase(), SqlKind.OTHER_FUNCTION, returnTypeInference,
null, operandTypeChecker,
SqlFunctionCategory.USER_DEFINED_FUNCTION);
_deterministic = deterministic;
+ _isVolatile = isVolatile;
+ }
+
+ /// Derives volatility from determinism: a non-deterministic function is
always volatile, and a deterministic one is
+ /// assumed not to be.
+ ///
+ /// Only the second half is an assumption -- a deterministic function can
still be
+ /// `FunctionVolatility.VOLATILE` (that is exactly what `now()` is). Use the
constructor above to say so explicitly;
+ /// this overload is for operators that are plain immutable functions of
their arguments.
+ public PinotSqlFunction(String name, SqlReturnTypeInference
returnTypeInference,
+ SqlOperandTypeChecker operandTypeChecker, boolean deterministic) {
+ this(name, returnTypeInference, operandTypeChecker, deterministic,
!deterministic);
}
public PinotSqlFunction(String name, SqlReturnTypeInference
returnTypeInference,
@@ -45,4 +58,15 @@ public class PinotSqlFunction extends SqlFunction {
public boolean isDeterministic() {
return _deterministic;
}
+
+ /// Whether the function is `FunctionVolatility.VOLATILE`, i.e. its result
can change on every invocation or it has
+ /// side effects.
+ ///
+ /// This is independent of [#isDeterministic()], which is Pinot's
compile-time-evaluation hint: `now()` is
+ /// deterministic (so it can be constant-folded once at plan time) but
volatile (so it must not be re-evaluated at a
+ /// different point in the plan). `FunctionVolatility.STABLE` is not
reported here, since a stable function is
+ /// constant within a single query and is therefore safe to relocate.
+ public boolean isVolatile() {
+ return _isVolatile;
+ }
}
diff --git
a/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java
b/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java
index a653a91ecc7..b99e4b737e3 100644
---
a/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java
+++
b/pinot-common/src/test/java/org/apache/pinot/common/function/FunctionUtilsTest.java
@@ -25,6 +25,7 @@ import java.time.LocalTime;
import java.util.HashMap;
import java.util.List;
import java.util.UUID;
+import org.apache.pinot.common.function.sql.PinotSqlFunction;
import org.apache.pinot.common.utils.DataSchema.ColumnDataType;
import org.apache.pinot.spi.annotations.FunctionVolatility;
import org.apache.pinot.spi.annotations.ScalarFunction;
@@ -125,6 +126,39 @@ public class FunctionUtilsTest {
assertEquals(workerId.getVolatility(), FunctionVolatility.VOLATILE);
}
+ /// The [FunctionInfo] volatility above is per-arity, but the
[PinotSqlFunction] the planner sees is per-operator.
+ /// Both determinism and volatility are therefore aggregated conservatively
across every registered overload.
+ @Test
+ public void testOperatorLevelVolatilityAggregatesAcrossOverloads() {
+ // rand() is VOLATILE and non-deterministic at 0 args, IMMUTABLE and
deterministic at 1 arg. The single operator
+ // has to report the more restrictive of each.
+ PinotSqlFunction rand = toSqlFunction("rand");
+ assertFalse(rand.isDeterministic());
+ assertTrue(rand.isVolatile());
+
+ // now() stays deterministic so it can be constant-folded, but must still
report volatile.
+ PinotSqlFunction now = toSqlFunction("now");
+ assertTrue(now.isDeterministic());
+ assertTrue(now.isVolatile());
+
+ // STABLE is constant within a query, so it must NOT be reported as
volatile.
+ PinotSqlFunction reqId = toSqlFunction("reqid");
+ assertTrue(reqId.isDeterministic());
+ assertFalse(reqId.isVolatile());
+
+ PinotSqlFunction upper = toSqlFunction("upper");
+ assertTrue(upper.isDeterministic());
+ assertFalse(upper.isVolatile());
+ }
+
+ private static PinotSqlFunction toSqlFunction(String canonicalName) {
+ PinotScalarFunction scalarFunction =
FunctionRegistry.getFunctions().get(canonicalName);
+ assertNotNull(scalarFunction, "Failed to find function: " + canonicalName);
+ PinotSqlFunction sqlFunction = scalarFunction.toPinotSqlFunction();
+ assertNotNull(sqlFunction, "Function is not registered as a
PinotSqlFunction: " + canonicalName);
+ return sqlFunction;
+ }
+
@Test
public void testFunctionVolatilityResolution()
throws NoSuchMethodException {
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotFilterJoinRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotFilterJoinRule.java
index e350d045a5f..398eb7b9386 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotFilterJoinRule.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotFilterJoinRule.java
@@ -48,9 +48,39 @@ public abstract class PinotFilterJoinRule<C extends
FilterJoinRule.Config> exten
}
// Following code are copy-pasted from Calcite, and modified to not push
down filter into right side of lookup join.
+ // SYNCED WITH Calcite 1.42.0 FilterJoinRule#perform -- re-diff this method
body against upstream on every
+ // calcite.version bump. The intended deviations are the canPushRight
lookup-join restriction and the volatility half
+ // of the isRelocatable guard, both marked PINOT MODIFICATION below.
+ //
+ // Known outstanding drift: upstream's RexUtil.containsCorrelation
partitioning of aboveFilters and the variablesSet
+ // argument on the final RelBuilder#filter (CALCITE-7319) are not ported.
Pinot decorrelates in
+ // QueryEnvironment#toRelation before these rules run, and the
LogicalCorrelate shapes that do survive it
+ // (UNNEST / CROSS JOIN UNNEST) have an Uncollect right input, so a Filter
carrying $cor never sits directly above a
+ // Join here. Revisit if Pinot ever retains a Correlate over a Join.
//@formatter:off
@Override
protected void perform(RelOptRuleCall call, @Nullable Filter filter, Join
join) {
+ // From CALCITE-7373: a non-deterministic conjunct such as rand() < 0.1
has an empty input bitmap, so
+ // classifyFilters would treat it as pushable and relocate it below the
join, evaluating it per input row instead
+ // of per join-output row. Like upstream, this bails on the whole
condition rather than per conjunct, so a
+ // deterministic conjunct sharing the WHERE clause also stays above the
join.
+ // PINOT MODIFICATION to also skip volatile conditions. Upstream uses
RexUtil.isDeterministic, which only covers
+ // @ScalarFunction(isDeterministic = false). Pinot's separate
FunctionVolatility.VOLATILE axis (now(), ago(),
+ // stageId(), ...) keeps isDeterministic() == true so
PinotEvaluateLiteralRule can still fold it once at plan time,
+ // so it needs the wider PinotRuleUtils.isRelocatable check here.
+ // Skip non-deterministic or volatile filter condition
+ if (filter != null &&
!PinotRuleUtils.isRelocatable(filter.getCondition())) {
+ return;
+ }
+ // Skip non-deterministic or volatile join condition.
+ // NOTE: for an INNER join this guard is largely moot for anything
referencing a single side, because
+ // RelOptUtil.pushDownJoinConditions already hoisted such a call into that
input's Project during sql-to-rel,
+ // before any rule ran; the condition seen here is then a bare
RexInputRef. It is still load-bearing for outer
+ // joins, where the ON clause is preserved. See JoinPlans.json for both
shapes.
+ if (!PinotRuleUtils.isRelocatable(join.getCondition())) {
+ return;
+ }
+
List<RexNode> joinFilters =
RelOptUtil.conjunctions(join.getCondition());
final List<RexNode> origJoinFilters = List.copyOf(joinFilters);
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 4e876392d8d..b7ad7a4edb2 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
@@ -38,14 +38,19 @@ 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;
import org.apache.calcite.sql.SqlAggFunction;
import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
import org.apache.calcite.sql2rel.SqlToRelConverter;
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.common.function.sql.PinotSqlFunction;
public class PinotRuleUtils {
@@ -132,6 +137,49 @@ public class PinotRuleUtils {
return funcSqlKind == SqlKind.OTHER_FUNCTION ?
function.getOperator().getName() : funcSqlKind.name();
}
+ /// Returns whether `node` evaluates to the same result no matter where in
the plan it sits, and can therefore be
+ /// relocated -- pushed below a join, duplicated onto another input, and so
on.
+ ///
+ /// 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.
+ /// - [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.
+ ///
+ /// 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
+ /// query, so moving it is safe.
+ ///
+ /// 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())) {
+ throw Util.FoundOne.NULL;
+ }
+ return super.visitCall(call);
+ }
+ });
+ return true;
+ } catch (Util.FoundOne e) {
+ Util.swallow(e, null);
+ return false;
+ }
+ }
+
public static class WindowUtils {
// Supported window functions
// OTHER_FUNCTION supported are: BOOL_AND, BOOL_OR
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 773386badca..e05ee827547 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
@@ -324,7 +324,10 @@ public class PinotOperatorTable implements
SqlOperatorTable {
List.of(SqlTypeFamily.CHARACTER, SqlTypeFamily.CHARACTER,
SqlTypeFamily.CHARACTER, SqlTypeFamily.ANY),
i -> i > 1)),
- new PinotSqlFunction("NOW", ReturnTypes.TIMESTAMP, OperandTypes.NILADIC)
+ // Deterministic so PinotEvaluateLiteralRule folds it once at plan time,
but volatile so it is never
+ // re-evaluated at a different point in the plan. This entry shadows the
FunctionRegistry one (see
+ // registerScalarFunctions), so the volatility has to be repeated here.
+ new PinotSqlFunction("NOW", ReturnTypes.TIMESTAMP, OperandTypes.NILADIC,
true, true)
);
private static final List<Pair<SqlOperator, List<String>>>
PINOT_OPERATORS_WITH_ALIASES = List.of(
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
new file mode 100644
index 00000000000..30fba46e27f
--- /dev/null
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotRuleUtilsTest.java
@@ -0,0 +1,159 @@
+/**
+ * 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 org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.pinot.calcite.sql.fun.PinotOperatorTable;
+import org.apache.pinot.common.function.FunctionRegistry;
+import org.apache.pinot.common.function.PinotScalarFunction;
+import org.apache.pinot.common.function.sql.PinotSqlFunction;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertFalse;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+
+/// Tests [PinotRuleUtils#isRelocatable], which decides whether an expression
may be moved to a different position in
+/// the plan. It must reject all three variability axes: `isDeterministic =
false`, Calcite's `isDynamicFunction()`,
+/// and Pinot's `FunctionVolatility.VOLATILE`.
+public class PinotRuleUtilsTest {
+
+ private final RelDataTypeFactory _typeFactory = new JavaTypeFactoryImpl();
+ private final RexBuilder _rexBuilder = new RexBuilder(_typeFactory);
+
+ private PinotSqlFunction registryFunction(String name) {
+ PinotScalarFunction scalarFunction =
FunctionRegistry.getFunctions().get(FunctionRegistry.canonicalize(name));
+ assertNotNull(scalarFunction, "Failed to find function: " + name);
+ PinotSqlFunction sqlFunction = scalarFunction.toPinotSqlFunction();
+ assertNotNull(sqlFunction, "Function is not registered as a
PinotSqlFunction: " + name);
+ return sqlFunction;
+ }
+
+ /// Resolves the operator a query would actually bind to, which is not
always the [FunctionRegistry] entry --
+ /// `PinotOperatorTable#registerScalarFunctions` skips names already present
in its hard-coded list.
+ private SqlOperator resolvedOperator(String name) {
+ List<SqlOperator> matches = new ArrayList<>();
+ PinotOperatorTable.instance(false).lookupOperatorOverloads(new
SqlIdentifier(name, SqlParserPos.ZERO),
+ SqlFunctionCategory.USER_DEFINED_FUNCTION, SqlSyntax.FUNCTION,
matches, null);
+ assertFalse(matches.isEmpty(), "Failed to resolve operator: " + name);
+ return matches.get(0);
+ }
+
+ private RexNode call(SqlOperator operator, RexNode... operands) {
+ RelDataType returnType = _typeFactory.createSqlType(SqlTypeName.BIGINT);
+ return _rexBuilder.makeCall(returnType, operator, List.of(operands));
+ }
+
+ private RexNode literal(int value) {
+ return _rexBuilder.makeLiteral(value,
_typeFactory.createSqlType(SqlTypeName.INTEGER));
+ }
+
+ @Test
+ public void testLiteralAndInputRefAreRelocatable() {
+ assertTrue(PinotRuleUtils.isRelocatable(literal(1)));
+ assertTrue(PinotRuleUtils.isRelocatable(
+
_rexBuilder.makeInputRef(_typeFactory.createSqlType(SqlTypeName.INTEGER), 0)));
+ }
+
+ @Test
+ public void testDeterministicCallIsRelocatable() {
+ assertTrue(PinotRuleUtils.isRelocatable(call(SqlStdOperatorTable.PLUS,
literal(1), literal(2))));
+ }
+
+ @Test
+ public void testImmutableFunctionIsRelocatable() {
+ PinotSqlFunction upper = registryFunction("upper");
+ assertTrue(upper.isDeterministic());
+ assertFalse(upper.isVolatile());
+ assertTrue(PinotRuleUtils.isRelocatable(call(upper,
_rexBuilder.makeLiteral("x"))));
+ }
+
+ @Test
+ public void testNonDeterministicFunctionIsNotRelocatable() {
+ // rand() is @ScalarFunction(isDeterministic = false); the operator-level
flag is shared with the seeded overload.
+ PinotSqlFunction rand = registryFunction("rand");
+ assertFalse(rand.isDeterministic());
+ assertFalse(PinotRuleUtils.isRelocatable(call(rand)));
+ }
+
+ @Test
+ public void testVolatileFunctionIsNotRelocatable() {
+ // stageId() stays deterministic so it can be constant-folded, but is
VOLATILE, so it must not be relocated.
+ PinotSqlFunction stageId = registryFunction("stageId");
+ assertTrue(stageId.isDeterministic(), "stageId should stay deterministic
for compile-time evaluation");
+ assertTrue(stageId.isVolatile());
+ assertFalse(PinotRuleUtils.isRelocatable(call(stageId, literal(0))));
+ }
+
+ /// `FunctionVolatility.STABLE` is constant within a single query, so it is
safe to relocate. `reqId` gets STABLE
+ /// from the class-level annotation on `InternalFunctions`, which also
covers annotation inheritance.
+ @Test
+ public void testStableFunctionIsRelocatable() {
+ PinotSqlFunction reqId = registryFunction("reqId");
+ assertTrue(reqId.isDeterministic());
+ assertFalse(reqId.isVolatile(), "STABLE must not be reported as volatile");
+ assertTrue(PinotRuleUtils.isRelocatable(call(reqId, literal(0))));
+ }
+
+ /// Calcite's own dynamic functions carry the same "fold once, never
re-evaluate" contract via a different flag.
+ @Test
+ public void testCalciteDynamicFunctionIsNotRelocatable() {
+ assertTrue(SqlStdOperatorTable.CURRENT_TIMESTAMP.isDynamicFunction());
+ assertTrue(SqlStdOperatorTable.CURRENT_TIMESTAMP.isDeterministic(),
+ "guarding on isDeterministic alone would miss this");
+
assertFalse(PinotRuleUtils.isRelocatable(call(SqlStdOperatorTable.CURRENT_TIMESTAMP)));
+ }
+
+ /// Both the registry entry and the hard-coded [PinotOperatorTable] entry
that shadows it must agree that `now()` is
+ /// volatile, otherwise the guard's answer depends on which one a query
happens to bind to.
+ @Test
+ public void testNowIsVolatileOnBothRegistrations() {
+ PinotSqlFunction registryNow = registryFunction("now");
+ assertTrue(registryNow.isDeterministic(), "now() must stay deterministic
so it is folded once at plan time");
+ assertTrue(registryNow.isVolatile());
+ assertFalse(PinotRuleUtils.isRelocatable(call(registryNow)));
+
+ SqlOperator resolvedNow = resolvedOperator("NOW");
+ assertTrue(resolvedNow instanceof PinotSqlFunction, "expected a
PinotSqlFunction, got: " + resolvedNow.getClass());
+ assertTrue(((PinotSqlFunction) resolvedNow).isVolatile(),
+ "the operator NOW() actually binds to must also report volatile");
+ assertFalse(PinotRuleUtils.isRelocatable(call(resolvedNow)));
+ }
+
+ @Test
+ public void testNestedVolatileOperandIsDetected() {
+ // The visitor must recurse into operands, not just inspect the top-level
operator.
+ RexNode nested = call(SqlStdOperatorTable.PLUS, literal(1),
call(registryFunction("stageId"), literal(0)));
+ assertFalse(PinotRuleUtils.isRelocatable(nested));
+ }
+}
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
index 4416e47bb52..54e916982c5 100644
---
a/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/query/QueryCompilationTest.java
@@ -434,6 +434,34 @@ public class QueryCompilationTest extends
QueryEnvironmentTestBase {
}
}
+ /// [org.apache.pinot.calcite.rel.rules.PinotFilterJoinRule] refuses to push
a volatile filter below a join. `now()`
+ /// is volatile, but
[org.apache.pinot.calcite.rel.rules.PinotEvaluateLiteralRule] folds it to a
literal first, so
+ /// the common time-filter-over-a-join pattern must still reach the leaf
scan.
+ ///
+ /// Asserted here rather than in JoinPlans.json because the folded epoch
literal differs on every run.
+ @Test
+ public void testVolatileNowFilterIsStillPushedBelowJoin() {
+ String query =
+ "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a JOIN b ON a.col1 =
b.col1 WHERE a.ts > now() - 86400000";
+
+ String explain = _queryEnvironment.explainQuery(query,
RANDOM_REQUEST_ID_GEN.nextLong());
+ // now() folds to the current epoch millis, so mask the literal before
comparing.
+ String normalized = explain.replaceAll("(?<=\\$7, )\\d+", "<EPOCH>");
+ //@formatter:off
+ assertEquals(normalized,
+ "Execution Plan\n"
+ + "LogicalProject(col1=[$0], col2=[$2])\n"
+ + " LogicalJoin(condition=[=($0, $1)], joinType=[inner])\n"
+ + " PinotLogicalExchange(distribution=[hash[0]])\n"
+ + " LogicalProject(col1=[$0])\n"
+ + " LogicalFilter(condition=[>($7, <EPOCH>)])\n"
+ + " PinotLogicalTableScan(table=[[default, a]])\n"
+ + " PinotLogicalExchange(distribution=[hash[0]])\n"
+ + " LogicalProject(col1=[$0], col2=[$1])\n"
+ + " PinotLogicalTableScan(table=[[default, b]])\n");
+ //@formatter:on
+ }
+
@Test
public void testAggregateCaseToFilter() {
// Tests that queries like "SELECT SUM(CASE WHEN col1 = 'a' THEN 1 ELSE 0
END) FROM a" are rewritten to
diff --git a/pinot-query-planner/src/test/resources/queries/JoinPlans.json
b/pinot-query-planner/src/test/resources/queries/JoinPlans.json
index eee3a32cf66..a96520741ea 100644
--- a/pinot-query-planner/src/test/resources/queries/JoinPlans.json
+++ b/pinot-query-planner/src/test/resources/queries/JoinPlans.json
@@ -664,6 +664,210 @@
"\n PinotLogicalTableScan(table=[[default, b]])",
"\n"
]
+ },
+ {
+ "description": "Non-deterministic WHERE filter is not pushed below the
join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a JOIN b ON a.col1
= b.col1 WHERE rand() < 0.1",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalFilter(condition=[<(RAND(), 0.1E0)])",
+ "\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], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Non-deterministic WHERE filter is not pushed below a
left join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a LEFT JOIN b ON
a.col1 = b.col1 WHERE rand() < 0.1",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalFilter(condition=[<(RAND(), 0.1E0)])",
+ "\n LogicalJoin(condition=[=($0, $1)], joinType=[left])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Non-deterministic WHERE filter on the null-generating
side does not simplify the left join to inner",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a LEFT JOIN b ON
a.col1 = b.col1 WHERE b.col3 > 100 * rand()",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalFilter(condition=[>(CAST($3):DOUBLE, *(100, RAND()))])",
+ "\n LogicalJoin(condition=[=($0, $1)], joinType=[left])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col2=[$1], col3=[$2])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Non-deterministic WHERE filter on the null-generating
side does not simplify the full join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a FULL JOIN b ON
a.col1 = b.col1 WHERE b.col3 > 100 * rand()",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalFilter(condition=[>(CAST($3):DOUBLE, *(100, RAND()))])",
+ "\n LogicalJoin(condition=[=($0, $1)], joinType=[full])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col2=[$1], col3=[$2])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Non-deterministic ON condition is not pushed below the
join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a JOIN b ON a.col1
= b.col1 AND rand() < 0.1",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalJoin(condition=[AND(=($0, $1), <(RAND(), 0.1E0))],
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], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Non-deterministic UUID filter is not pushed below the
join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a JOIN b ON a.col1
= b.col1 WHERE uuid_v4() <> 'x'",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalFilter(condition=[<>(CAST(UUIDV4()):CHAR(1) CHARACTER
SET \"UTF-8\" NOT NULL, _UTF-8'x')])",
+ "\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], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Non-deterministic filter blocks the semi join rewrite:
the IN sub-query is planned as an inner join over a distinct aggregate, with
the filter kept above the join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1 FROM a WHERE a.col1 IN (SELECT
b.col1 FROM b) AND rand() < 0.1",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0])",
+ "\n LogicalFilter(condition=[<(RAND(), 0.1E0)])",
+ "\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 PinotLogicalAggregate(group=[{0}], aggType=[FINAL])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n PinotLogicalAggregate(group=[{0}], aggType=[LEAF])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "A non-deterministic conjunct holds the whole WHERE
filter above the join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a JOIN b ON a.col1
= b.col1 WHERE a.col3 > 5 AND rand() < 0.1",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$3])",
+ "\n LogicalFilter(condition=[AND(>($1, 5), <(RAND(), 0.1E0))])",
+ "\n LogicalJoin(condition=[=($0, $2)], joinType=[inner])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col3=[$2])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Volatile WHERE filter is not pushed below the join,
nor duplicated onto both inputs",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a JOIN b ON a.col1
= b.col1 WHERE stageId(a.col1) >= 0",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalFilter(condition=[>=(STAGEID($0), 0)])",
+ "\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], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Volatile ON condition is kept in the join condition
for an outer join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a LEFT JOIN b ON
a.col1 = b.col1 AND stageId(b.col3) >= 0",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalJoin(condition=[AND(=($0, $1), >=(STAGEID($3), 0))],
joinType=[left])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col2=[$1],
EXPR$0=[CAST($2):VARCHAR CHARACTER SET \"UTF-8\" NOT NULL])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "KNOWN GAP: a volatile ON conjunct referencing one side
of an INNER join is hoisted into that input by
RelOptUtil.pushDownJoinConditions during sql-to-rel, before any rule runs, so
the join-condition guard never sees it",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a JOIN b ON a.col1
= b.col1 AND stageId(a.col3) >= 0",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalJoin(condition=[=($0, $1)], joinType=[inner])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0])",
+ "\n LogicalFilter(condition=[>=(STAGEID(CAST($2):VARCHAR
CHARACTER SET \"UTF-8\" NOT NULL), 0)])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
+ {
+ "description": "Deterministic WHERE filter is still pushed below the
join",
+ "sql": "EXPLAIN PLAN FOR SELECT a.col1, b.col2 FROM a JOIN b ON a.col1
= b.col1 WHERE a.col3 > 5",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalJoin(condition=[=($0, $1)], joinType=[inner])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0])",
+ "\n LogicalFilter(condition=[>($2, 5)])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n PinotLogicalExchange(distribution=[hash[0]])",
+ "\n LogicalProject(col1=[$0], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
}
]
},
@@ -684,6 +888,22 @@
"\n"
]
},
+ {
+ "description": "Non-deterministic WHERE filter is not pushed below a
lookup join",
+ "sql": "EXPLAIN PLAN FOR SELECT /*+ joinOptions(join_strategy =
'lookup') */ a.col1, b.col2 FROM a JOIN b ON a.col1 = b.col1 WHERE rand() <
0.1",
+ "output": [
+ "Execution Plan",
+ "\nLogicalProject(col1=[$0], col2=[$2])",
+ "\n LogicalFilter(condition=[<(RAND(), 0.1E0)])",
+ "\n LogicalJoin(condition=[=($0, $1)], joinType=[inner])",
+ "\n PinotLogicalExchange(distribution=[single])",
+ "\n LogicalProject(col1=[$0])",
+ "\n PinotLogicalTableScan(table=[[default, a]])",
+ "\n LogicalProject(col1=[$0], col2=[$1])",
+ "\n PinotLogicalTableScan(table=[[default, b]])",
+ "\n"
+ ]
+ },
{
"description": "Lookup join with filter on left table",
"sql": "EXPLAIN PLAN FOR SELECT /*+ joinOptions(join_strategy =
'lookup') */ a.col1, b.col2 FROM a JOIN b ON a.col1 = b.col1 WHERE a.col2 =
'foo'",
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
