peter-toth commented on code in PR #56101: URL: https://github.com/apache/spark/pull/56101#discussion_r3681561827
########## sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastNearestByJoinExec.scala: ########## @@ -0,0 +1,186 @@ +/* + * 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. + * Using a case class with primitive `Int` field avoids boxing that `(Int, Any)` tuples incur. + */ +private[joins] case class HeapEntry(index: Int, 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 to all partitions. This operator only fires when + * the right side fits within the broadcast join threshold + * ([[org.apache.spark.sql.internal.SQLConf.autoBroadcastJoinThreshold]]). For right tables + * exceeding this threshold, the existing cross-product + aggregate rewrite is used as + * fallback. Tie-breaking among equal ranking values is non-deterministic (matches the + * existing rewrite behavior). Review Comment: **Finding 5.** Both claims here are now false, and this is the class doc a maintainer reads first. After option (a) the operator fires for *every* `NearestByJoin` once the flag is on, and there is no fallback available at that point: `RewriteNearestByJoin` has already declined to rewrite, so an oversized right side fails the query (`BroadcastExchangeExec`'s row/size caps, or driver OOM below them) where the flag-off path completes. Not re-opening @cloud-fan's link nit on line 45 — that one is fixed; this is about the sentence around it. The same stale claim survives in three more places: - `SparkStrategies.scala:432` — "the size decision is deferred to runtime (the right side is always broadcast)" — and `RewriteNearestByJoin.scala:82` — "The size decision is deferred entirely to the physical operator's runtime." Nothing in `doExecute` decides anything about size; the right side is broadcast unconditionally. These should say there is no size decision at all. - The PR description still lists "Conditional skip of `RewriteNearestByJoin` when the conf is enabled **and right fits in broadcast threshold**", and the test list still claims "broadcast threshold fallback" coverage. Both were true two revisions ago. And the consequence belongs in the conf doc, which is the only place a user looks (`SQLConf.scala:2529`): ```scala .doc("When true, NearestByJoin uses a streaming heap operator instead of the " + "cross-product + aggregate rewrite. The right side is always broadcast, regardless " + "of its size and of spark.sql.autoBroadcastJoinThreshold, so a right side too large " + "to broadcast fails the query instead of falling back to the rewrite.") ``` For the class doc: ```suggestion * The right side is fully broadcast to all partitions, unconditionally: when * `spark.sql.join.nearestBy.broadcast.enabled` is on, `RewriteNearestByJoin` leaves every * `NearestByJoin` intact for this operator, so there is no size test and no fallback to the * cross-product + aggregate rewrite. A right side too large to broadcast fails the query. * Tie-breaking among equal ranking values is non-deterministic (matches the existing * rewrite behavior). ``` ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/NearestByJoin.scala: ########## @@ -20,10 +20,17 @@ package org.apache.spark.sql.catalyst.plans.logical import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression} import org.apache.spark.sql.catalyst.plans.{Inner, JoinType, LeftOuter, NearestByDirection, NearestByJoinValidation} import org.apache.spark.sql.catalyst.trees.TreePattern._ +import org.apache.spark.sql.internal.SQLConf object NearestByJoin { /** @see [[NearestByJoinValidation.MaxNumResults]] */ val MaxNumResults: Int = NearestByJoinValidation.MaxNumResults + + /** Whether the right side of a NearestByJoin is eligible for broadcast execution. */ + def canBroadcastRight(j: NearestByJoin, conf: SQLConf): Boolean = Review Comment: **Finding 6.** Nothing calls this any more — the rewrite guard is flag-only and `NearestByJoinSelection` is unconditional. `grep -rn canBroadcastRight sql/catalyst/src sql/core/src` finds only this definition plus a comment in the test suite. Please delete it together with the now-unused `import org.apache.spark.sql.internal.SQLConf` on line 23. Worth removing now rather than leaving as an unused helper: this is the exact `j.right.stats.sizeInBytes` read that finding 2 was about, still sitting on the logical node, where the next caller from an early optimizer batch reintroduces the DSv2 `INTERNAL_ERROR` and the `Long.MaxValue` partitioned-table estimate. If a size test comes back later it belongs where `JoinSelection` does it — `canBroadcastBySize` on a physical plan. The PR description bullet "Shared broadcast eligibility: `NearestByJoin.canBroadcastRight` (used by both optimizer and planner)" should go with it (it was @cloud-fan's single-source-of-truth ask, which option (a) resolved by removing both call sites). ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteNearestByJoin.scala: ########## @@ -72,7 +73,14 @@ 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. Reading stats here is unsafe: this rule runs in + // FinishAnalysis, before filter/partition pushdown completes, so DSv2 sources + // throw INTERNAL_ERROR and partitioned tables report inflated sizeInBytes. + // The size decision is deferred entirely to the physical operator's runtime. + if !SQLConf.get.nearestByBroadcastEnabled => Review Comment: **Finding 7.** Skipping the rewrite also skips the `Join` node it would have built, so `CheckCartesianProducts` (`Optimizer.scala:2572-2579`) has nothing to match: with the flag on, `spark.sql.crossJoin.enabled=false` no longer rejects NEAREST BY. The comment at `:97-101` in this same rule calls that rejection deliberate: > 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. I think the new behaviour is the better one — the broadcast operator genuinely is not a cross product, it is bounded at k per left row — so the ask is not to restore the rejection but to make the divergence deliberate and pinned, since a perf flag changing which queries are *legal* is surprising: - state it wherever finding 5's contract text lands (operator scaladoc and/or conf doc), and adjust the `:97-101` comment so it scopes its "intentional" claim to the rewrite path; - add one flag-on test that leaves `spark.sql.crossJoin.enabled` at its default. Every flag-on test currently sets `CROSS_JOINS_ENABLED -> "true"` (`BroadcastNearestByJoinExecSuite.scala:34`, `:168`, `:426`, `:472`, `:582`), which is a no-op on this path and hides the change; one test without it pins the contract, and the others could then drop the conf. -- 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]
