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