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