This is an automated email from the ASF dual-hosted git repository.
Jackie-Jiang 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 19dec509431 Reject the join distribution types that answer the join
wrong (#19862)
19dec509431 is described below
commit 19dec509431436925b33d7b981fe489c02cb7f17
Author: RAGHVENDRA KUMAR YADAV <[email protected]>
AuthorDate: Mon Oct 26 16:24:32 2026 -0700
Reject the join distribution types that answer the join wrong (#19862)
---
.../rel/rules/PinotJoinExchangeNodeInsertRule.java | 134 ++++++++++++++++++-
.../rules/PinotJoinExchangeNodeInsertRuleTest.java | 143 +++++++++++++++++++++
.../test/resources/queries/NonEquiOuterJoins.json | 58 +++++++++
.../src/test/resources/queries/QueryHints.json | 81 ++++++++++++
4 files changed, 412 insertions(+), 4 deletions(-)
diff --git
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java
index 86ff8603d90..c9ff53023a0 100644
---
a/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java
+++
b/pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java
@@ -18,6 +18,7 @@
*/
package org.apache.pinot.calcite.rel.rules;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.List;
import java.util.Map;
@@ -26,12 +27,17 @@ import org.apache.calcite.plan.RelOptRule;
import org.apache.calcite.plan.RelOptRuleCall;
import org.apache.calcite.rel.RelDistributions;
import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Calc;
+import org.apache.calcite.rel.core.Filter;
import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.JoinInfo;
import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.logical.LogicalAsofJoin;
import org.apache.calcite.tools.RelBuilderFactory;
import org.apache.pinot.calcite.rel.hint.PinotHintOptions;
+import org.apache.pinot.calcite.rel.hint.PinotHintStrategyTable;
import org.apache.pinot.calcite.rel.logical.PinotLogicalExchange;
@@ -88,14 +94,15 @@ public class PinotJoinExchangeNodeInsertRule extends
RelOptRule {
// Hash join
// Force pre-partitioned exchange when colocated join hint is provided
Boolean prePartitioned =
PinotHintOptions.JoinHintOptions.isColocatedByJoinKeys(join);
- // TODO: Validate if the configured distribution types are valid
if (leftDistributionType == null) {
leftDistributionType = inferLeftDistributionType(joinInfo,
join.getJoinType());
}
- newLeft = createExchangeForHashJoin(leftDistributionType,
joinInfo.leftKeys, left, prePartitioned);
if (rightDistributionType == null) {
rightDistributionType = inferRightDistributionType(joinInfo,
join.getJoinType());
}
+ validateDistributionTypes(leftDistributionType, rightDistributionType,
isReplicatedTableScan(left),
+ isReplicatedTableScan(right), joinInfo, join.getJoinType());
+ newLeft = createExchangeForHashJoin(leftDistributionType,
joinInfo.leftKeys, left, prePartitioned);
newRight = createExchangeForHashJoin(rightDistributionType,
joinInfo.rightKeys, right, prePartitioned);
}
@@ -127,11 +134,129 @@ public class PinotJoinExchangeNodeInsertRule extends
RelOptRule {
}
}
+ /// Rejects the pairs of distribution types that cannot produce the join's
rows, which are otherwise silently
+ /// answered wrong. A pair is workable only if every pair of rows that could
match meets on one worker, and only if
+ /// each row is emitted by one worker.
+ ///
+ /// Broadcasting a side gives every worker all of its rows, so it stands in
for any partitioning of the other side. A
+ /// LOCAL scan of a table hinted `is_replicated='true'` does the same, since
its leaf hands every worker the whole
+ /// table. Two such sides give every worker all the rows of both, and each
joined row is then emitted once per worker,
+ /// whether the sides are broadcast, replicated tables read locally, or one
of each. Broadcasting neither leaves both
+ /// sides partitioned, and they only line up when both are partitioned the
same way on the join keys, which also
+ /// requires the join to have keys to partition on. LOCAL on both sides is
accepted on the user's word that the two
+ /// inputs are co-partitioned, like the colocated join hint: nothing here
can check it.
+ ///
+ /// A join that decides a left row's output from the right rows it saw --
LEFT emits the left rows that matched
+ /// nothing, SEMI and ANTI emit a left row on the presence or absence of a
match, and ASOF / LEFT_ASOF pick the
+ /// closest match -- can only decide on a worker that holds every right row.
So the broadcast side must be the right
+ /// one; RIGHT is the mirror image. SEMI is the exception when the right
side is hashed on the join keys: every right
+ /// row sharing a left row's key is on the one worker that key hashes to, so
only that worker can emit the left row,
+ /// and it does exactly when a match exists. ANTI and LEFT get no such
exception, since every other worker would emit
+ /// the left row as unmatched. A FULL join emits the unmatched rows of both
sides, so with keys it cannot use a
+ /// broadcast side at all, and without keys it runs on a single worker,
where only an exchange that delivers every
+ /// row there (RANDOM or BROADCAST) is workable.
+ @VisibleForTesting
+ static void validateDistributionTypes(PinotHintOptions.DistributionType
leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, boolean
leftReplicated, boolean rightReplicated,
+ JoinInfo joinInfo, JoinRelType joinType) {
+ if (joinType == JoinRelType.FULL && joinInfo.leftKeys.isEmpty()) {
+ // DispatchablePlanVisitor gives a FULL join without equi keys a single
worker, so every row reaches it once.
+ Preconditions.checkArgument(isSingletonFeed(leftDistributionType) &&
isSingletonFeed(rightDistributionType),
+ "FULL join without equi keys runs on a single worker, so each side
must be distributed RANDOM or BROADCAST "
+ + "to reach it, got left: %s, right: %s", leftDistributionType,
rightDistributionType);
+ return;
+ }
+ boolean leftWhole = holdsWholeSide(leftDistributionType, leftReplicated);
+ boolean rightWhole = holdsWholeSide(rightDistributionType,
rightReplicated);
+ Preconditions.checkArgument(!(leftWhole && rightWhole),
+ "Cannot give every worker the whole of both sides of the join
(broadcast, or a LOCAL scan of a table hinted "
+ + "is_replicated='true'): every worker would emit each joined row,
repeating the join result once per "
+ + "worker. At most one side can be whole, got left: %s, right:
%s", leftDistributionType,
+ rightDistributionType);
+ if (leftWhole || rightWhole) {
+ switch (joinType) {
+ case SEMI:
+ if (rightDistributionType == PinotHintOptions.DistributionType.HASH
&& !joinInfo.leftKeys.isEmpty()) {
+ // Every right row with a given key is on one worker, so a left
row can only be emitted there.
+ break;
+ }
+ Preconditions.checkArgument(rightWhole,
+ "SEMI join requires every right row on each worker (broadcast
the right side, read a replicated right "
+ + "table locally, or hash the right side on the join keys):
a worker decides each left row from "
+ + "the right rows it holds, got left: %s, right: %s",
leftDistributionType, rightDistributionType);
+ break;
+ case LEFT:
+ case ANTI:
+ case ASOF:
+ case LEFT_ASOF:
+ case LEFT_MARK:
+ Preconditions.checkArgument(rightWhole,
+ "%s join requires every right row on each worker (broadcast the
right side, or read a replicated right "
+ + "table locally): a worker decides each left row from the
right rows it holds, got left: %s, "
+ + "right: %s", joinType, leftDistributionType,
rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftWhole,
+ "RIGHT join requires every left row on each worker (broadcast
the left side, or read a replicated left "
+ + "table locally): a worker can only tell that a right row
matched nothing if it holds every left "
+ + "row, got left: %s, right: %s", leftDistributionType,
rightDistributionType);
+ break;
+ case FULL:
+ throw new IllegalArgumentException(
+ "FULL join cannot broadcast a side: both sides emit the rows
that matched nothing, so neither side can "
+ + "be the partitioned one. Distribute both sides by hash on
the join keys.");
+ case INNER:
+ break;
+ default:
+ // A join type not listed above may be asymmetric, so reject it
rather than accept a whole side unchecked
+ throw new IllegalArgumentException("Unsupported join type for a
broadcast or replicated side: " + joinType);
+ }
+ return;
+ }
+ Preconditions.checkArgument(!joinInfo.leftKeys.isEmpty(),
+ "Join without equi keys requires one side to be broadcast: with both
sides partitioned, a pair of rows that "
+ + "could match may land on different workers, got left: %s, right:
%s", leftDistributionType,
+ rightDistributionType);
+ Preconditions.checkArgument(
+ leftDistributionType == rightDistributionType
+ && leftDistributionType !=
PinotHintOptions.DistributionType.RANDOM,
+ "Both sides of the join must be distributed the same way on the join
keys when neither side is broadcast, "
+ + "otherwise rows sharing a key land on different workers, got
left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ }
+
+ /// Whether every worker of the join ends up with every row of this side.
+ private static boolean holdsWholeSide(PinotHintOptions.DistributionType
distributionType, boolean replicated) {
+ return distributionType == PinotHintOptions.DistributionType.BROADCAST
+ || (replicated && distributionType ==
PinotHintOptions.DistributionType.LOCAL);
+ }
+
+ /// Whether this exchange delivers every row of its side to a single
receiving worker.
+ private static boolean isSingletonFeed(PinotHintOptions.DistributionType
distributionType) {
+ return distributionType == PinotHintOptions.DistributionType.RANDOM
+ || distributionType == PinotHintOptions.DistributionType.BROADCAST;
+ }
+
+ /// Whether `input` is a scan of a table hinted `is_replicated='true'`,
reached only through projects, filters and
+ /// calcs.
+ /// Anything else in between (an exchange, an aggregate, another join) is
not assumed to keep the whole table.
+ private static boolean isReplicatedTableScan(RelNode input) {
+ RelNode node = PinotRuleUtils.unboxRel(input);
+ // A Sort is not unwrapped: with a LIMIT or OFFSET it no longer keeps the
whole table
+ while (node instanceof Project || node instanceof Filter || node
instanceof Calc) {
+ node = PinotRuleUtils.unboxRel(node.getInput(0));
+ }
+ return node instanceof TableScan && Boolean.parseBoolean(
+ PinotHintStrategyTable.getHintOption(((TableScan) node).getHints(),
PinotHintOptions.TABLE_HINT_OPTIONS,
+ PinotHintOptions.TableHintOptions.IS_REPLICATED));
+ }
+
/// For non-equi RIGHT JOINs (no equi keys), the default
RANDOM(left)+BROADCAST(right) is incorrect because each
/// worker independently tracks matched right rows via a local BitSet.
Workers that receive only a subset of left rows
/// will incorrectly emit right rows as unmatched. Invert to
BROADCAST(left)+RANDOM(right) so each worker has the
/// full left table and can correctly determine which of its local right
rows are unmatched.
- private static PinotHintOptions.DistributionType
inferLeftDistributionType(JoinInfo joinInfo,
+ @VisibleForTesting
+ static PinotHintOptions.DistributionType inferLeftDistributionType(JoinInfo
joinInfo,
JoinRelType joinType) {
if (!joinInfo.leftKeys.isEmpty()) {
return PinotHintOptions.DistributionType.HASH;
@@ -142,7 +267,8 @@ public class PinotJoinExchangeNodeInsertRule extends
RelOptRule {
return PinotHintOptions.DistributionType.RANDOM;
}
- private static PinotHintOptions.DistributionType
inferRightDistributionType(JoinInfo joinInfo,
+ @VisibleForTesting
+ static PinotHintOptions.DistributionType inferRightDistributionType(JoinInfo
joinInfo,
JoinRelType joinType) {
if (!joinInfo.rightKeys.isEmpty()) {
return PinotHintOptions.DistributionType.HASH;
diff --git
a/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRuleTest.java
b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRuleTest.java
new file mode 100644
index 00000000000..d83b3aa6745
--- /dev/null
+++
b/pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRuleTest.java
@@ -0,0 +1,143 @@
+/**
+ * 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 org.apache.calcite.rel.core.JoinInfo;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.util.ImmutableIntList;
+import org.apache.pinot.calcite.rel.hint.PinotHintOptions.DistributionType;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static
org.apache.pinot.calcite.rel.hint.PinotHintOptions.DistributionType.BROADCAST;
+import static
org.apache.pinot.calcite.rel.hint.PinotHintOptions.DistributionType.HASH;
+import static
org.apache.pinot.calcite.rel.hint.PinotHintOptions.DistributionType.LOCAL;
+import static
org.apache.pinot.calcite.rel.hint.PinotHintOptions.DistributionType.RANDOM;
+import static org.testng.Assert.assertThrows;
+
+
+/// Covers the distribution type pairs a join can and cannot be answered with,
including join types that SQL hints do
+/// not reach today (ANTI, ASOF and LEFT_ASOF are produced or hinted outside
the path the query tests exercise).
+public class PinotJoinExchangeNodeInsertRuleTest {
+ private static final JoinInfo EQUI = JoinInfo.of(ImmutableIntList.of(0),
ImmutableIntList.of(0));
+ private static final JoinInfo NON_EQUI = JoinInfo.of(ImmutableIntList.of(),
ImmutableIntList.of());
+
+ /// {join type, join info, left, right, left replicated, right replicated}
+ @DataProvider
+ public static Object[][] accepted() {
+ return new Object[][]{
+ {JoinRelType.INNER, EQUI, HASH, HASH, false, false},
+ {JoinRelType.INNER, EQUI, LOCAL, BROADCAST, false, false},
+ {JoinRelType.LEFT, EQUI, LOCAL, BROADCAST, false, false},
+ {JoinRelType.SEMI, EQUI, LOCAL, BROADCAST, false, false},
+ {JoinRelType.ANTI, EQUI, RANDOM, BROADCAST, false, false},
+ {JoinRelType.ASOF, EQUI, HASH, BROADCAST, false, false},
+ {JoinRelType.LEFT_ASOF, EQUI, LOCAL, BROADCAST, false, false},
+ {JoinRelType.LEFT_MARK, EQUI, LOCAL, BROADCAST, false, false},
+ {JoinRelType.RIGHT, EQUI, BROADCAST, LOCAL, false, false},
+ // A SEMI join emits a left row only on the worker its key hashes to,
which holds every right row with that key.
+ {JoinRelType.SEMI, EQUI, BROADCAST, HASH, false, false},
+ {JoinRelType.SEMI, EQUI, LOCAL, HASH, true, false},
+ // A LOCAL scan of a replicated table gives every worker the whole
table, like a broadcast.
+ {JoinRelType.INNER, NON_EQUI, LOCAL, LOCAL, false, true},
+ {JoinRelType.LEFT, NON_EQUI, LOCAL, LOCAL, false, true},
+ {JoinRelType.INNER, EQUI, HASH, LOCAL, false, true},
+ {JoinRelType.RIGHT, EQUI, LOCAL, HASH, true, false},
+ // A FULL join without equi keys runs on a single worker, which RANDOM
and BROADCAST both deliver every row to.
+ {JoinRelType.FULL, NON_EQUI, RANDOM, BROADCAST, false, false},
+ {JoinRelType.FULL, NON_EQUI, RANDOM, RANDOM, false, false},
+ {JoinRelType.FULL, NON_EQUI, BROADCAST, BROADCAST, false, false},
+ };
+ }
+
+ @Test(dataProvider = "accepted")
+ public void testAccepted(JoinRelType joinType, JoinInfo joinInfo,
DistributionType left, DistributionType right,
+ boolean leftReplicated, boolean rightReplicated) {
+ PinotJoinExchangeNodeInsertRule.validateDistributionTypes(left, right,
leftReplicated, rightReplicated, joinInfo,
+ joinType);
+ }
+
+ /// {join type, join info, left, right, left replicated, right replicated}
+ @DataProvider
+ public static Object[][] rejected() {
+ return new Object[][]{
+ // Each of these decides a left row from the right rows a worker
holds, so the left cannot be the whole side.
+ {JoinRelType.LEFT, EQUI, BROADCAST, LOCAL, false, false},
+ {JoinRelType.SEMI, EQUI, BROADCAST, LOCAL, false, false},
+ {JoinRelType.SEMI, EQUI, BROADCAST, RANDOM, false, false},
+ {JoinRelType.SEMI, NON_EQUI, BROADCAST, HASH, false, false},
+ {JoinRelType.SEMI, EQUI, LOCAL, LOCAL, true, false},
+ {JoinRelType.ANTI, EQUI, BROADCAST, HASH, false, false},
+ {JoinRelType.ANTI, EQUI, LOCAL, HASH, true, false},
+ {JoinRelType.LEFT, EQUI, BROADCAST, HASH, false, false},
+ {JoinRelType.ASOF, EQUI, BROADCAST, HASH, false, false},
+ {JoinRelType.LEFT_ASOF, EQUI, BROADCAST, HASH, false, false},
+ {JoinRelType.LEFT_MARK, EQUI, BROADCAST, HASH, false, false},
+ {JoinRelType.LEFT, EQUI, LOCAL, LOCAL, true, false},
+ {JoinRelType.RIGHT, EQUI, LOCAL, BROADCAST, false, false},
+ {JoinRelType.FULL, EQUI, LOCAL, BROADCAST, false, false},
+ // Two sides that each give every worker the whole side repeat the
join result once per worker.
+ {JoinRelType.INNER, EQUI, BROADCAST, BROADCAST, false, false},
+ {JoinRelType.INNER, NON_EQUI, LOCAL, LOCAL, true, true},
+ {JoinRelType.INNER, EQUI, LOCAL, LOCAL, true, true},
+ {JoinRelType.INNER, EQUI, LOCAL, BROADCAST, true, false},
+ {JoinRelType.INNER, NON_EQUI, BROADCAST, LOCAL, false, true},
+ {JoinRelType.LEFT, EQUI, LOCAL, LOCAL, true, true},
+ {JoinRelType.SEMI, EQUI, BROADCAST, LOCAL, false, true},
+ {JoinRelType.INNER, EQUI, RANDOM, RANDOM, false, false},
+ {JoinRelType.INNER, EQUI, HASH, RANDOM, false, false},
+ {JoinRelType.INNER, NON_EQUI, LOCAL, LOCAL, false, false},
+ {JoinRelType.FULL, NON_EQUI, LOCAL, LOCAL, false, false},
+ {JoinRelType.FULL, NON_EQUI, RANDOM, LOCAL, false, false},
+ };
+ }
+
+ /// {join type, join info}
+ @DataProvider
+ public static Object[][] joinTypes() {
+ JoinRelType[] joinTypes = {
+ JoinRelType.INNER, JoinRelType.LEFT, JoinRelType.RIGHT,
JoinRelType.FULL, JoinRelType.SEMI, JoinRelType.ANTI,
+ JoinRelType.ASOF, JoinRelType.LEFT_ASOF, JoinRelType.LEFT_MARK
+ };
+ Object[][] cases = new Object[joinTypes.length * 2][];
+ for (int i = 0; i < joinTypes.length; i++) {
+ cases[2 * i] = new Object[]{joinTypes[i], EQUI};
+ cases[2 * i + 1] = new Object[]{joinTypes[i], NON_EQUI};
+ }
+ return cases;
+ }
+
+ /// The distribution types picked when the query has no hint must always
pass validation, so a change to the
+ /// inferred defaults cannot start rejecting ordinary queries.
+ @Test(dataProvider = "joinTypes")
+ public void testInferredDefaultsAccepted(JoinRelType joinType, JoinInfo
joinInfo) {
+ PinotJoinExchangeNodeInsertRule.validateDistributionTypes(
+ PinotJoinExchangeNodeInsertRule.inferLeftDistributionType(joinInfo,
joinType),
+ PinotJoinExchangeNodeInsertRule.inferRightDistributionType(joinInfo,
joinType), false, false, joinInfo,
+ joinType);
+ }
+
+ @Test(dataProvider = "rejected")
+ public void testRejected(JoinRelType joinType, JoinInfo joinInfo,
DistributionType left, DistributionType right,
+ boolean leftReplicated, boolean rightReplicated) {
+ assertThrows(IllegalArgumentException.class,
+ () -> PinotJoinExchangeNodeInsertRule.validateDistributionTypes(left,
right, leftReplicated, rightReplicated,
+ joinInfo, joinType));
+ }
+}
diff --git
a/pinot-query-runtime/src/test/resources/queries/NonEquiOuterJoins.json
b/pinot-query-runtime/src/test/resources/queries/NonEquiOuterJoins.json
index 5716b3497c8..aef2f7873c7 100644
--- a/pinot-query-runtime/src/test/resources/queries/NonEquiOuterJoins.json
+++ b/pinot-query-runtime/src/test/resources/queries/NonEquiOuterJoins.json
@@ -58,6 +58,64 @@
[12, "d", null, null],
[null, null, 3, "x"]
]
+ },
+ {
+ "description": "FULL OUTER JOIN with only non-equi condition, random
left and broadcast right: one worker receives every row",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'random',
right_distribution_type = 'broadcast') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} FULL OUTER JOIN {tbl2} ON {tbl1}.num < {tbl2}.num",
+ "outputs": [
+ [5, "a", 7, "y"],
+ [5, "a", 10, "z"],
+ [5, "a", 11, "w"],
+ [6, "b", 7, "y"],
+ [6, "b", 10, "z"],
+ [6, "b", 11, "w"],
+ [8, "c", 10, "z"],
+ [8, "c", 11, "w"],
+ [12, "d", null, null],
+ [null, null, 3, "x"]
+ ],
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "FULL OUTER JOIN with only non-equi condition, random
left and random right: one worker receives every row",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'random',
right_distribution_type = 'random') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} FULL OUTER JOIN {tbl2} ON {tbl1}.num < {tbl2}.num",
+ "outputs": [
+ [5, "a", 7, "y"],
+ [5, "a", 10, "z"],
+ [5, "a", 11, "w"],
+ [6, "b", 7, "y"],
+ [6, "b", 10, "z"],
+ [6, "b", 11, "w"],
+ [8, "c", 10, "z"],
+ [8, "c", 11, "w"],
+ [12, "d", null, null],
+ [null, null, 3, "x"]
+ ],
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "FULL OUTER JOIN with only non-equi condition,
broadcast left and broadcast right: one worker receives every row",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'broadcast',
right_distribution_type = 'broadcast') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} FULL OUTER JOIN {tbl2} ON {tbl1}.num < {tbl2}.num",
+ "outputs": [
+ [5, "a", 7, "y"],
+ [5, "a", 10, "z"],
+ [5, "a", 11, "w"],
+ [6, "b", 7, "y"],
+ [6, "b", 10, "z"],
+ [6, "b", 11, "w"],
+ [8, "c", 10, "z"],
+ [8, "c", 11, "w"],
+ [12, "d", null, null],
+ [null, null, 3, "x"]
+ ],
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "FULL OUTER JOIN with only non-equi condition cannot
take a local side on its single worker",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'local') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} FULL OUTER JOIN {tbl2} ON {tbl1}.num < {tbl2}.num",
+ "outputs": [],
+ "expectedException": ".*FULL join without equi keys runs on a single
worker.*",
+ "ignoreV2Optimizer": true
}
]
}
diff --git a/pinot-query-runtime/src/test/resources/queries/QueryHints.json
b/pinot-query-runtime/src/test/resources/queries/QueryHints.json
index 05cd92da01e..1bbf8754e02 100644
--- a/pinot-query-runtime/src/test/resources/queries/QueryHints.json
+++ b/pinot-query-runtime/src/test/resources/queries/QueryHints.json
@@ -127,6 +127,63 @@
"description": "Colocated JOIN with partition column and group by
non-partitioned column with stage parallelism",
"sql": "SET stageParallelism=2; SELECT {tbl1}.name, SUM({tbl2}.num)
FROM {tbl1} /*+ tableOptions(partition_function='hashcode',
partition_key='num', partition_size='4') */ JOIN {tbl2} /*+
tableOptions(partition_function='hashcode', partition_key='num',
partition_size='4') */ ON {tbl1}.num = {tbl2}.num GROUP BY {tbl1}.name"
},
+ {
+ "description": "Both sides distributed at random drops rows that share
a key but land on different workers",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'random',
right_distribution_type = 'random') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} JOIN {tbl2} ON {tbl1}.num = {tbl2}.num",
+ "expectedException": ".*must be distributed the same way on the join
keys.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "One side hashed and the other at random drops rows
that share a key",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'hash',
right_distribution_type = 'random') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} JOIN {tbl2} ON {tbl1}.num = {tbl2}.num",
+ "expectedException": ".*must be distributed the same way on the join
keys.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "Broadcasting both sides repeats the join result once
per worker",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'broadcast',
right_distribution_type = 'broadcast') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} JOIN {tbl2} ON {tbl1}.num = {tbl2}.num",
+ "expectedException": ".*Cannot give every worker the whole of both
sides of the join.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "Join without equi keys and no broadcast side cannot
bring matching rows together",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'local') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} JOIN {tbl2} ON {tbl1}.num > {tbl2}.num",
+ "expectedException": ".*requires one side to be broadcast.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "LEFT join broadcasting the left side emits left rows
as unmatched on workers holding part of the right side",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'broadcast',
right_distribution_type = 'local') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} LEFT JOIN {tbl2} ON {tbl1}.num = {tbl2}.num",
+ "expectedException": ".*LEFT join requires every right row on each
worker.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "RIGHT join broadcasting the right side emits right
rows as unmatched on workers holding part of the left side",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'broadcast') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} RIGHT JOIN {tbl2} ON {tbl1}.num = {tbl2}.num",
+ "expectedException": ".*RIGHT join requires every left row on each
worker.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "RIGHT join without equi keys broadcasting the right
side emits right rows as unmatched",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'random',
right_distribution_type = 'broadcast') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} RIGHT JOIN {tbl2} ON {tbl1}.num > {tbl2}.num",
+ "expectedException": ".*RIGHT join requires every left row on each
worker.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "SEMI join broadcasting the left side emits a left row
once per worker holding a match",
+ "sql": "SELECT /*+ joinOptions(join_strategy = 'hash',
left_distribution_type = 'broadcast', right_distribution_type = 'local') */
{tbl1}.num, {tbl1}.name FROM {tbl1} WHERE {tbl1}.num IN (SELECT {tbl2}.num FROM
{tbl2})",
+ "expectedException": ".*SEMI join requires every right row on each
worker.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "SEMI join broadcasting the left side and hashing the
right side on the join keys: only the worker a key hashes to holds its right
rows",
+ "sql": "SELECT /*+ joinOptions(join_strategy = 'hash',
left_distribution_type = 'broadcast', right_distribution_type = 'hash') */
{tbl1}.num, {tbl1}.name FROM {tbl1} WHERE {tbl1}.num IN (SELECT {tbl2}.num FROM
{tbl2})",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "LEFT join broadcasting the right side is the workable
direction",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'broadcast') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} LEFT JOIN {tbl2} ON {tbl1}.num = {tbl2}.num"
+ },
{
"description": "Broadcast JOIN",
"sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'broadcast') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} JOIN {tbl2} ON {tbl1}.num = {tbl2}.num"
@@ -721,6 +778,30 @@
"description": "Local replicated JOIN with partition hint and
parallelism",
"sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'local') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} /*+ tableOptions(partition_function='hashcode',
partition_key='num', partition_size='4', partition_parallelism='2') */ JOIN
{tbl2} /*+ tableOptions(is_replicated='true') */ ON {tbl1}.num = {tbl2}.num"
},
+ {
+ "description": "Local JOIN without equi keys against a replicated table:
every worker holds the whole right table",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'local') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} JOIN {tbl2} /*+ tableOptions(is_replicated='true') */ ON
{tbl1}.num > {tbl2}.num"
+ },
+ {
+ "description": "Local LEFT JOIN without equi keys against a replicated
table",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'local') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} LEFT JOIN {tbl2} /*+ tableOptions(is_replicated='true')
*/ ON {tbl1}.num > {tbl2}.num"
+ },
+ {
+ "description": "Hash on the left and local on a replicated right table",
+ "sql": "SELECT /*+ joinOptions(right_distribution_type = 'local') */
{tbl1}.num, {tbl1}.name, {tbl2}.num, {tbl2}.val FROM {tbl1} JOIN {tbl2} /*+
tableOptions(is_replicated='true') */ ON {tbl1}.num = {tbl2}.num"
+ },
+ {
+ "description": "Local JOIN between two replicated tables repeats the
join result once per worker",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'local') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} /*+ tableOptions(is_replicated='true') */ JOIN {tbl2}
/*+ tableOptions(is_replicated='true') */ ON {tbl1}.num > {tbl2}.num",
+ "expectedException": ".*Cannot give every worker the whole of both sides
of the join.*",
+ "ignoreV2Optimizer": true
+ },
+ {
+ "description": "Local scan of a replicated left table and a broadcast
right side repeats the join result once per worker",
+ "sql": "SELECT /*+ joinOptions(left_distribution_type = 'local',
right_distribution_type = 'broadcast') */ {tbl1}.num, {tbl1}.name, {tbl2}.num,
{tbl2}.val FROM {tbl1} /*+ tableOptions(is_replicated='true') */ JOIN {tbl2} ON
{tbl1}.num = {tbl2}.num",
+ "expectedException": ".*Cannot give every worker the whole of both sides
of the join.*",
+ "ignoreV2Optimizer": true
+ },
{
"description": "Window PARTITION BY the table's partition column with
is_partitioned_by_window_keys hint forces a pre-partitioned (direct) exchange;
tbl1 is partitioned by num so results stay correct",
"sql": "SELECT /*+ windowOptions(is_partitioned_by_window_keys='true')
*/ {tbl1}.num, SUM({tbl1}.num) OVER (PARTITION BY {tbl1}.num) FROM {tbl1}"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
