cloud-fan commented on code in PR #56101:
URL: https://github.com/apache/spark/pull/56101#discussion_r3752338980
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteNearestByJoin.scala:
##########
@@ -72,7 +72,12 @@ object RewriteNearestByJoin extends Rule[LogicalPlan] {
private lazy val random = new scala.util.Random()
def apply(plan: LogicalPlan): LogicalPlan = plan.transformUp {
- case j @ NearestByJoin(left, right, joinType, _, numResults,
rankingExpression, direction) =>
+ case j @ NearestByJoin(left, right, joinType, _, numResults,
rankingExpression, direction)
+ // When the broadcast flag is ON the NearestByJoin node is left intact
for the
+ // planner's NearestByJoinSelection strategy, which unconditionally plans
+ // BroadcastNearestByJoinExec. There is no size decision; the right side
is
+ // broadcast unconditionally regardless of
spark.sql.autoBroadcastJoinThreshold.
+ if !conf.nearestByBroadcastEnabled =>
Review Comment:
Please keep rankings containing a Python UDF on a plan shape where the UDF
can be evaluated after both inputs are joined. With this gate enabled, a UDF
such as `udf((x, y) => ...)(left.x, right.y)` remains on the binary
`NearestByJoin`; `ExtractPythonUDFs` cannot place it under either child and
throws `Invalid PythonUDF ... requires attributes from more than one child`.
The flag-off rewrite works because it creates the Join before UDF extraction.
Please route this case through the rewrite (or introduce a joined evaluation
node) and add a two-sided Python UDF parity test.
--
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]
