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