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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
