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: [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]
