You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "raghavyadav01 (via GitHub)" <gi...@apache.org> on 2026/10/14 05:08:08 UTC
[PR] Reject the join distribution types that answer the join wrong [pinot]
raghavyadav01 opened a new pull request, #19862:
URL: https://github.com/apache/pinot/pull/19862
`PinotJoinExchangeNodeInsertRule` turns `left_distribution_type` / `right_distribution_type` into the exchanges that route each side's rows, and carries a `// TODO: Validate if the configured distribution types are valid`. Until it does, several pairs a user can ask for are accepted and answered wrong, with no error and no warning.
Measured against H2 on the two-worker `ResourceBasedQueriesTest` setup, with the hint as the only difference between the query and a correct one:
| hint | join | H2 | returned |
| --- | --- | ---: | ---: |
| `random` + `random` | inner, equi | 10 | 8 |
| `hash` + `random` | inner, equi | 10 | 2 |
| `broadcast` + `broadcast` | inner, equi | 10 | 20 |
| `broadcast` on the left | LEFT, equi | 13 | 23 |
| `broadcast` on the right | RIGHT, equi | 11 | 19 |
| `random` + `broadcast` | RIGHT, non-equi | 39 | 47 |
Two conditions explain all six:
1. A pair of rows that could match has to meet on one worker, and each joined row has to be emitted once. Broadcasting one side gives every worker all of its rows, so it stands in for any partitioning of the other. Broadcasting both gives every worker all of both, and the join result repeats once per worker. Broadcasting neither leaves both sides partitioned, which lines up only when both are partitioned the same way on the join keys β and that needs keys to partition on.
2. An outer join also emits the rows of its outer side that matched nothing, which a worker can only determine once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one. A FULL join is outer on both sides and cannot have a broadcast side at all.
This PR replaces the TODO with a check of the resolved pair against both conditions, throwing at plan time with the reason. The pairs `inferLeftDistributionType` / `inferRightDistributionType` produce already satisfy them β including the RIGHT-join inversion they document β so a query that does not hint a distribution type plans exactly as before.
The comment above `inferLeftDistributionType` already notes that `RANDOM(left) + BROADCAST(right)` is incorrect for a non-equi RIGHT JOIN, which is why the inference inverts it. The last row of the table is that same combination supplied by hand, where nothing inverted it.
### Testing
The six combinations above are added to `QueryHints.json` with the error they must now raise, plus the no-equi-keys case with neither side broadcast, and `LEFT JOIN` with the right side broadcast as the workable direction that must keep planning. Each one returns the wrong row count on master and fails to plan with this change.
All 3901 cases in `ResourceBasedQueriesTest` and 2665 in `pinot-query-planner` pass, so no existing plan or result moves. The new error cases carry `ignoreV2Optimizer`, because the physical optimizer does not build its exchanges through this rule and already answers all six correctly β this is the old planner path only.
π€ Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19862:
URL: https://github.com/apache/pinot/pull/19862#discussion_r4278286461
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ return;
+ }
+ Preconditions.checkArgument(!joinInfo.leftKeys.isEmpty(),
Review Comment:
Minor: a FULL join with no equi keys and non-broadcast hints (e.g. `local`+`local`) reaches this check and throws "requires one side to be broadcast" β but the FULL case above forbids broadcasting either side, so the message points the user at something impossible. A FULL-specific message would be clearer here.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19862:
URL: https://github.com/apache/pinot/pull/19862#discussion_r4278285941
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
Review Comment:
SEMI and ANTI joins fall into this `default` and are accepted with any broadcast side, but they're asymmetric like LEFT: a worker can only decide a left row has (ANTI) no match / (SEMI) a match if it holds *every* right row. So `broadcast(left) + local(right)` on an ANTI/SEMI join passes here yet answers wrong β the exact class of bug this rule rejects.
ANTI never matches `PinotJoinToDynamicBroadcastRule` (SEMI-only), and SEMI escapes it for non-equi / multi-key / hash-strategy cases, so both can reach this rule. The inferred defaults stay safe (right side ends up BROADCAST), so no regression β but a hand-hinted broadcast-left isn't caught. Consider treating SEMI/ANTI like LEFT (require right broadcast).
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ return;
+ }
+ Preconditions.checkArgument(!joinInfo.leftKeys.isEmpty(),
Review Comment:
Minor: a FULL join with no equi keys and non-broadcast hints (e.g. `local`+`local`) reaches this check and throws "requires one side to be broadcast" β but the FULL case above forbids broadcasting either side, so the message points the user at something impossible. A FULL-specific message would be clearer here.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
Review Comment:
This frames completeness around "outer" joins, but SEMI/ANTI share the same asymmetry (they emit left rows based on presence/absence of a right match) without being outer. Either cover them in the switch below or note here that SEMI/ANTI are intentionally out of scope, so the contract doesn't over-promise what it checks.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ return;
+ }
+ Preconditions.checkArgument(!joinInfo.leftKeys.isEmpty(),
Review Comment:
Minor: a FULL join with no equi keys and non-broadcast hints (e.g. `local`+`local`) reaches this check and throws "requires one side to be broadcast" β but the FULL case above forbids broadcasting either side, so the message points the user at something impossible. A FULL-specific message would be clearer here.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
Review Comment:
This frames completeness around "outer" joins, but SEMI/ANTI share the same asymmetry (they emit left rows based on presence/absence of a right match) without being outer. Either cover them in the switch below or note here that SEMI/ANTI are intentionally out of scope, so the contract doesn't over-promise what it checks.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
Review Comment:
SEMI and ANTI joins fall into this `default` and are accepted with any broadcast side, but they're asymmetric like LEFT: a worker can only decide a left row has (ANTI) no match / (SEMI) a match if it holds *every* right row. So `broadcast(left) + local(right)` on an ANTI/SEMI join passes here yet answers wrong β the exact class of bug this rule rejects.
ANTI never matches `PinotJoinToDynamicBroadcastRule` (SEMI-only), and SEMI escapes it for non-equi / multi-key / hash-strategy cases, so both can reach this rule. The inferred defaults stay safe (right side ends up BROADCAST), so no regression β but a hand-hinted broadcast-left isn't caught. Consider treating SEMI/ANTI like LEFT (require right broadcast).
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19862:
URL: https://github.com/apache/pinot/pull/19862#discussion_r4361635501
##########
pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRuleTest.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * 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() {
Review Comment:
The unit tests only feed explicit distribution pairs. It'd be worth adding cases that run the inferred defaults (null hints -> inferLeft/RightDistributionType) through validateDistributionTypes for each join type, so a future tweak to the inference defaults that produces a now-rejected pair gets caught here rather than surfacing as a regression on ordinary no-hint queries.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +133,117 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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:
+ 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.");
+ default:
+ break;
+ }
+ 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 and filters.
+ /// 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);
+ while (node instanceof Project || node instanceof Filter) {
Review Comment:
Unwrapping only Project/Filter is a safe conservative choice, but note it can false-negative: a replicated scan wrapped in, say, a Calc or Sort that still preserves the whole table gets treated as not-whole and the LEFT/SEMI path then rejects an actually-valid plan. Probably fine for the shapes that reach this rule today β just flagging that the conservative direction here is a rejection, not a wrong answer.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +133,117 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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:
+ 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.");
+ default:
Review Comment:
The `default` branch accepts any unlisted join type with a whole side, but LEFT_MARK is asymmetric like SEMI/ANTI β a worker decides the mark from the right rows it holds, so `broadcast(left)` would emit a mark per worker. LEFT_MARK does flow through `convertLogicalJoin` today (see `SearchSealer`), so it's worth either handling it alongside SEMI/ANTI or narrowing the `default` to throw on unknown types rather than waving them through.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +133,117 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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:
+ 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.");
+ default:
+ break;
+ }
+ 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 and filters.
+ /// 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);
+ while (node instanceof Project || node instanceof Filter) {
Review Comment:
Unwrapping only Project/Filter is a safe conservative choice, but note it can false-negative: a replicated scan wrapped in, say, a Calc or Sort that still preserves the whole table gets treated as not-whole and the LEFT/SEMI path then rejects an actually-valid plan. Probably fine for the shapes that reach this rule today β just flagging that the conservative direction here is a rejection, not a wrong answer.
##########
pinot-query-planner/src/test/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRuleTest.java:
##########
@@ -0,0 +1,116 @@
+/**
+ * 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() {
Review Comment:
The unit tests only feed explicit distribution pairs. It'd be worth adding cases that run the inferred defaults (null hints -> inferLeft/RightDistributionType) through validateDistributionTypes for each join type, so a future tweak to the inference defaults that produces a now-rejected pair gets caught here rather than surfacing as a regression on ordinary no-hint queries.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +133,117 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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:
+ 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.");
+ default:
Review Comment:
The `default` branch accepts any unlisted join type with a whole side, but LEFT_MARK is asymmetric like SEMI/ANTI β a worker decides the mark from the right rows it holds, so `broadcast(left)` would emit a mark per worker. LEFT_MARK does flow through `convertLogicalJoin` today (see `SearchSealer`), so it's worth either handling it alongside SEMI/ANTI or narrowing the `default` to throw on unknown types rather than waving them through.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang merged PR #19862:
URL: https://github.com/apache/pinot/pull/19862
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on PR #19862:
URL: https://github.com/apache/pinot/pull/19862#issuecomment-6430420263
Documentation follow-up merged: https://github.com/pinot-contrib/pinot-docs/pull/1148 β explains distribution validation, replicated LOCAL scans, join-type restrictions and the single-worker FULL non-equi exception.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19862:
URL: https://github.com/apache/pinot/pull/19862#discussion_r4289849137
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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,
Review Comment:
The `default: break` here silently lets `SEMI`, `ANTI` and `LEFT_ASOF` skip the direction check. Those three decide on the local right table (ANTI/LEFT_ASOF emit left rows with no local match, SEMI keeps left rows with a local match), so broadcasting the left side while the right is partitioned answers them wrong β the same bug this PR fixes for LEFT. Defaults infer HASH/HASH so it only bites with explicit hints, but it's worth handling `ANTI`/`LEFT_ASOF` like `LEFT` and `SEMI` too.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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
Review Comment:
Minor/back-compat note: a user who was running one of these broken hint combinations used to silently get wrong rows, and will now get a hard query failure instead. That's the right trade since the old answer was incorrect, but it is a behavior change worth a line in the release notes so people know why a previously-accepted query now errors.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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,
Review Comment:
Minor: `local`/`local` passes here on trust, without checking the two tables are actually co-partitioned or one is replicated. That matches the existing colocated-hint philosophy of trusting the user, so no change needed β just calling out that this pair is accepted on faith while the others are reasoned about.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
Review Comment:
SEMI and ANTI joins fall into this `default` and are accepted with any broadcast side, but they're asymmetric like LEFT: a worker can only decide a left row has (ANTI) no match / (SEMI) a match if it holds *every* right row. So `broadcast(left) + local(right)` on an ANTI/SEMI join passes here yet answers wrong β the exact class of bug this rule rejects.
ANTI never matches `PinotJoinToDynamicBroadcastRule` (SEMI-only), and SEMI escapes it for non-equi / multi-key / hash-strategy cases, so both can reach this rule. The inferred defaults stay safe (right side ends up BROADCAST), so no regression β but a hand-hinted broadcast-left isn't caught. Consider treating SEMI/ANTI like LEFT (require right broadcast).
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
Review Comment:
This frames completeness around "outer" joins, but SEMI/ANTI share the same asymmetry (they emit left rows based on presence/absence of a right match) without being outer. Either cover them in the switch below or note here that SEMI/ANTI are intentionally out of scope, so the contract doesn't over-promise what it checks.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19862:
URL: https://github.com/apache/pinot/pull/19862#issuecomment-6162149583
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19862?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 100.00%. Comparing base ([`b72a7b8`](https://app.codecov.io/gh/apache/pinot/commit/b72a7b8ff44bad07612ef2033449eeba8b6d1b7d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`1654cfb`](https://app.codecov.io/gh/apache/pinot/commit/1654cfbf555a3091ad9c892274f63dd9eab2828b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19862 +/- ##
==============================================
+ Coverage 68.96% 100.00% +31.03%
+ Complexity 1491 6 -1485
==============================================
Files 3533 3 -3530
Lines 231734 6 -231728
Branches 36979 0 -36979
==============================================
- Hits 159824 6 -159818
+ Misses 59502 0 -59502
+ Partials 12408 0 -12408
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Ξ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ΓΈ> (ΓΈ)` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ΓΈ> (ΓΈ)` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ΓΈ> (+31.03%)` | :arrow_up: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ΓΈ> (ΓΈ)` | |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ΓΈ> (ΓΈ)` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `100.00% <ΓΈ> (+31.03%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19862/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/19862?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19862:
URL: https://github.com/apache/pinot/pull/19862#discussion_r4289850325
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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
Review Comment:
Minor/back-compat note: a user who was running one of these broken hint combinations used to silently get wrong rows, and will now get a hard query failure instead. That's the right trade since the old answer was incorrect, but it is a behavior change worth a line in the release notes so people know why a previously-accepted query now errors.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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,
Review Comment:
Minor: `local`/`local` passes here on trust, without checking the two tables are actually co-partitioned or one is replicated. That matches the existing colocated-hint philosophy of trusting the user, so no change needed β just calling out that this pair is accepted on faith while the others are reasoned about.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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
Review Comment:
Minor/back-compat note: a user who was running one of these broken hint combinations used to silently get wrong rows, and will now get a hard query failure instead. That's the right trade since the old answer was incorrect, but it is a behavior change worth a line in the release notes so people know why a previously-accepted query now errors.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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,
Review Comment:
Minor: `local`/`local` passes here on trust, without checking the two tables are actually co-partitioned or one is replicated. That matches the existing colocated-hint philosophy of trusting the user, so no change needed β just calling out that this pair is accepted on faith while the others are reasoned about.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #19862:
URL: https://github.com/apache/pinot/pull/19862#discussion_r4300489933
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
Review Comment:
[P1] Validate LEFT_ASOF and ANTI before accepting the broadcast pair
Both join types fall through this default and accept left BROADCAST/right HASH, although they need the same preserved-left-side restriction as LEFT. For example, with two workers, a LEFT ASOF JOIN on `l.k = r.k` with `MATCH_CONDITION(l.ts >= r.ts)`, `l=(a,10)` and `r=(a,9)` emits the matched row on the worker holding right key `a`, plus a spurious null-extended left row on the other worker. `AsofJoinOperator` decides unmatched rows from its local right table. ANTI similarly emits a false unmatched row on workers without the right key. There is no singleton repair for these join types.
The underlying execution bug predates this PR, but the new validator still accepts this wrong-result combination. Please handle LEFT_ASOF and ANTI with the LEFT restriction and add corresponding regression cases.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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);
Review Comment:
[P2] Account for replicated tables before requiring a broadcast exchange
A LOCAL input with `tableOptions(is_replicated='true')` already supplies the entire table to each join worker. At default parallelism, `WorkerManager` copies the local peer's worker map and assigns all replicated segments to each corresponding worker; `MailboxAssignmentVisitor` connects them 1:1 without requiring join keys. Consequently, an INNER or LEFT join using LOCAL/LOCAL against a replicated right table remains correct with a non-equi predicate such as `l.num > r.num`, but this check now rejects it.
The following equality check also rejects a valid equi-join variant: specify only `right_distribution_type='local'` on a replicated right table, and the left side defaults to HASH. Please account for effective replication when validating the pair, and extend `hint_option_replicated_table` with the non-equi LOCAL/LOCAL and equi HASH/LOCAL cases.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
Review Comment:
[P2] Preserve singleton execution for non-equi FULL joins
A non-equi FULL JOIN with explicit RANDOM/BROADCAST distributions was already correct: `DispatchablePlanVisitor.visitJoin` requires singleton execution whenever a FULL join has no equi keys, and `WorkerManager` selects one receiver. Both exchanges therefore deliver each input row exactly once to that receiver. This new unconditional rejection makes those previously correct queries fail; the suggested hash distribution is also unavailable because there are no join keys.
RANDOM/RANDOM is likewise valid on that singleton path but is rejected by the subsequent no-equi-key check. Please preserve the existing singleton execution path for these hints and add explicit-distribution variants of the FULL-join cases in `NonEquiOuterJoins.json`.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #19862:
URL: https://github.com/apache/pinot/pull/19862#discussion_r4289849137
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +127,62 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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.
+ /// Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once per
+ /// worker. 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.
+ ///
+ /// An outer join additionally emits the rows of its outer side that matched nothing, which a worker can only tell
+ /// once it holds every row of the opposite side. So the broadcast side must be the side opposite the outer one, and
+ /// a FULL join, whose two sides are both outer, cannot be answered with a broadcast side at all.
+ private static void validateDistributionTypes(PinotHintOptions.DistributionType leftDistributionType,
+ PinotHintOptions.DistributionType rightDistributionType, JoinInfo joinInfo, JoinRelType joinType) {
+ boolean leftBroadcast = leftDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ boolean rightBroadcast = rightDistributionType == PinotHintOptions.DistributionType.BROADCAST;
+ if (leftBroadcast && rightBroadcast) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ if (leftBroadcast || rightBroadcast) {
+ switch (joinType) {
+ case LEFT:
+ Preconditions.checkArgument(rightBroadcast,
+ "LEFT join requires the right side to be the broadcast one: a worker can only tell that a left row "
+ + "matched nothing if it holds every right row, got left: %s, right: %s", leftDistributionType,
+ rightDistributionType);
+ break;
+ case RIGHT:
+ Preconditions.checkArgument(leftBroadcast,
+ "RIGHT join requires the left side to be the broadcast one: 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.");
+ default:
+ break;
+ }
+ 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,
Review Comment:
The `default: break` here silently lets `SEMI`, `ANTI` and `LEFT_ASOF` skip the direction check. Those three decide on the local right table (ANTI/LEFT_ASOF emit left rows with no local match, SEMI keeps left rows with a local match), so broadcasting the left side while the right is partitioned answers them wrong β the same bug this PR fixes for LEFT. Defaults infer HASH/HASH so it only bites with explicit hints, but it's worth handling `ANTI`/`LEFT_ASOF` like `LEFT` and `SEMI` too.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Reject the join distribution types that answer the join wrong [pinot]
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on code in PR #19862:
URL: https://github.com/apache/pinot/pull/19862#discussion_r4300815019
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +133,106 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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. Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once
+ /// per worker. 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. 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;
+ }
+ if (leftDistributionType == PinotHintOptions.DistributionType.BROADCAST
+ && rightDistributionType == PinotHintOptions.DistributionType.BROADCAST) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ boolean leftWhole = holdsWholeSide(leftDistributionType, leftReplicated);
+ boolean rightWhole = holdsWholeSide(rightDistributionType, rightReplicated);
+ if (leftWhole || rightWhole) {
+ switch (joinType) {
+ case LEFT:
+ case SEMI:
+ case ANTI:
+ case ASOF:
+ case LEFT_ASOF:
+ Preconditions.checkArgument(rightWhole,
Review Comment:
[P2] Preserve valid broadcast/hash SEMI joins
SEMI does not need the same restriction as LEFT when the right side is hashed on the equi-join keys. With left BROADCAST/right HASH, all right rows sharing a key reach exactly one worker. HashJoinOperator emits a matching left row only on that worker; the other workers emit nothing, so each left row retains its correct multiplicity. The new unconditional `rightWhole` requirement rejects this previously correct query:
```sql
SELECT /*+ joinOptions(join_strategy='hash',
left_distribution_type='broadcast',
right_distribution_type='hash') */
l.num, l.name
FROM l
WHERE l.num IN (SELECT r.num FROM r)
```
This follows the same SQL-to-SEMI path as the new negative query test. Explicit `join_strategy='hash'` prevents the later dynamic-broadcast rewrite from replacing these distributions. Please permit this keyed SEMI combination and add a positive regression test; retain rejection when the right-side distribution can split matching keys across workers.
##########
pinot-query-planner/src/main/java/org/apache/pinot/calcite/rel/rules/PinotJoinExchangeNodeInsertRule.java:
##########
@@ -127,6 +133,106 @@ private static PinotLogicalExchange createExchangeForLookupJoin(PinotHintOptions
}
}
+ /// 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. Broadcasting both sides gives every worker all the rows of both, and each joined row is then emitted once
+ /// per worker. 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. 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;
+ }
+ if (leftDistributionType == PinotHintOptions.DistributionType.BROADCAST
+ && rightDistributionType == PinotHintOptions.DistributionType.BROADCAST) {
+ throw new IllegalArgumentException(
+ "Cannot broadcast both sides of the join: every worker would hold every row of both sides and emit each "
+ + "joined row, repeating the join result once per worker. Broadcast at most one side.");
+ }
+ boolean leftWhole = holdsWholeSide(leftDistributionType, leftReplicated);
+ boolean rightWhole = holdsWholeSide(rightDistributionType, rightReplicated);
+ if (leftWhole || rightWhole) {
Review Comment:
[P1] Reject both sides being effectively broadcast
The duplicate-result guard above checks only literal BROADCAST/BROADCAST. LOCAL/LOCAL with both tables hinted `is_replicated='true'`, or LOCAL(replicated)/BROADCAST, produces `leftWhole && rightWhole` but is accepted here for INNER joins. With two workers at default parallelism, WorkerManager assigns all replicated segments to each corresponding leaf worker, and the local exchanges deliver both complete inputs to each join worker. Each worker therefore emits the full join, duplicating the result. A non-equi INNER join between two replicated tables with LOCAL/LOCAL demonstrates this newly accepted case.
Please reject two effective-whole inputs after the singleton FULL exception, rather than checking only the literal distribution enums, and add both combinations to the validation tests plus a two-replicated-table query regression.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org