cloud-fan commented on code in PR #56101:
URL: https://github.com/apache/spark/pull/56101#discussion_r3756249274
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala:
##########
@@ -710,7 +710,12 @@ trait CheckAnalysis extends LookupCatalog with
QueryErrorsBase with PlanToString
errorClass = "NEAREST_BY_JOIN.STREAMING_NOT_SUPPORTED",
messageParameters = Map.empty)
- case j: NearestByJoin if !conf.crossJoinEnabled =>
+ case j: NearestByJoin if !conf.crossJoinEnabled &&
+ !conf.nearestByBroadcastEnabled =>
+ // The cross-join check only applies on the REWRITE path (flag
OFF) where
+ // a synthetic unconditioned Join node is created. On the OPERATOR
path
+ // (flag ON), no Join node is built and the operator produces at
most k rows
Review Comment:
**Nit:**
Flag-on is no longer always the operator path: a cross-child scalar Python
UDF is rewritten and builds the synthetic Join. Please qualify this as the
surviving operator path (or mention the UDF fallback).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -424,6 +424,28 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
}
}
+ /**
+ * Plans NearestByJoin as a BroadcastNearestByJoinExec when the broadcast
flag is ON.
+ * The optimizer's RewriteNearestByJoin leaves the NearestByJoin node intact
when the flag
Review Comment:
**Nit:**
The optimizer no longer preserves every enabled NearestByJoin: cross-child
scalar Python UDF rankings deliberately take the rewrite path. Please document
this as applying only to nodes that reach the strategy.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteNearestByJoin.scala:
##########
@@ -85,11 +120,13 @@ object RewriteNearestByJoin extends Rule[LogicalPlan] {
// are dropped. When `right` is non-empty every left row already has
right-row
// pairings, so `LEFT OUTER` and `INNER` are equivalent in that case.
//
- // This synthetic join is an unconditioned cross-product, so `NEAREST
BY` queries
- // are subject to `CheckCartesianProducts` and will be rejected when
the user has
- // set `spark.sql.crossJoin.enabled = false`. That is intentional: if
the user has
- // opted out of cross-products, the NEAREST BY rewrite -- which is
itself a bounded
- // cross-product today -- should not silently bypass that choice.
+ // This synthetic join is an unconditioned cross-product, so on the
REWRITE path
+ // (flag OFF) `NEAREST BY` queries are subject to
`CheckCartesianProducts` and will
+ // be rejected when the user has set `spark.sql.crossJoin.enabled =
false`. That is
+ // intentional: if the user has opted out of cross-products, the
rewrite -- which is
+ // itself a bounded cross-product today -- should not silently bypass
that choice.
+ // (On the OPERATOR path -- flag ON -- no Join node is built, so
Review Comment:
**Nit:**
Flag-on is no longer synonymous with this operator path: the cross-child
Python UDF fallback enters this rewrite and creates a Join. Please scope this
parenthetical to nodes that survive the rewrite.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNearestByJoinExec.scala:
##########
@@ -0,0 +1,225 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.joins
+
+import java.util.{Comparator, PriorityQueue => JPriorityQueue}
+
+import org.apache.spark.SparkException
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.{InnerLike, JoinType, LeftOuter,
NearestByDirection, NearestByDistance}
+import org.apache.spark.sql.catalyst.plans.physical._
+import org.apache.spark.sql.catalyst.util.TypeUtils
+import org.apache.spark.sql.execution.{ExplainUtils, SparkPlan}
+import org.apache.spark.sql.execution.metric.SQLMetrics
+
+/**
+ * Heap entry storing an index into the broadcast array alongside its ranking
value.
+ * Mutable so that evicted entries can be reused: once the heap reaches
capacity k,
+ * new candidates reuse the polled (worst) entry instead of allocating,
bounding total
+ * heap-entry allocations to k per left row regardless of right-side size.
+ */
+private[joins] class HeapEntry(var index: Int, var rankingValue: Any)
+
+/**
+ * Physical operator for NearestByJoin that avoids materializing the full
cross product.
+ * For each left row, iterates all broadcast right rows maintaining a bounded
priority
+ * queue of size k, then emits the top-k matches directly.
+ *
+ * The right side is fully broadcast unconditionally when
+ * `spark.sql.join.nearestBy.broadcast.enabled` is on.
+ * [[org.apache.spark.sql.catalyst.optimizer.RewriteNearestByJoin]] leaves
every
+ * [[org.apache.spark.sql.catalyst.plans.logical.NearestByJoin]] intact for
this operator;
Review Comment:
**Nit:**
The new Python-UDF guard rewrites two-sided scalar UDF rankings even when
the flag is enabled, so not every NearestByJoin is left intact for this
operator. Please qualify this statement with that exception.
--
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]
