peter-toth commented on code in PR #56101:
URL: https://github.com/apache/spark/pull/56101#discussion_r3672570540
##########
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 =
+ conf.nearestByBroadcastEnabled &&
+ j.right.stats.sizeInBytes >= 0 &&
Review Comment:
**Finding 2.** (Carried from round 1, escalated to Blocking — continues the
round-1 thread on `RewriteNearestByJoin.scala:83`.)
This predicate is fine when `NearestByJoinSelection` calls it, but
`RewriteNearestByJoin` calls it from the `FinishAnalysis` batch — the
optimizer's *first* batch. `Optimizer.scala:239-242` states the invariant:
> This batch pushes filters and projections into scan nodes. Before this
batch, the logical plan may contain nodes that do not report stats. **Anything
that uses stats must run after this batch.**
Your round-1 reply argued the early read is conservative because estimates
only shrink. That's true of the *direction*, but the problem is that some
relations cannot answer at all this early:
**1. DSv2 right side throws.** `DataSourceV2Relation.computeStats`
(`DataSourceV2Relation.scala:86-103`) throws when `Utils.isTesting`, precisely
to catch this:
```
org.apache.spark.SparkException: [INTERNAL_ERROR] BUG: computeStats called
before
pushdown on DSv2 relation: testcat.rightTbl SQLSTATE: XX000
```
Reproduced on this head — the same query passes with the flag off:
```scala
sql("CREATE TABLE testcat.rightTbl (rid INT, y DOUBLE) USING foo")
sql("INSERT INTO testcat.rightTbl VALUES (10, 1.0), (11, 2.0), (12, 3.0)")
Seq((1, 2.5)).toDF("id", "x").nearestByJoin(
spark.table("testcat.rightTbl"), abs($"x" - $"y"),
numResults = 2, mode = "exact", direction = "distance").collect()
// flag on -> INTERNAL_ERROR above
// flag off -> [1,2.5,12,3.0], [1,2.5,11,2.0]
```
Outside testing it takes the `else` branch, which builds a throwaway scan
(`table.asReadable.newScanBuilder(options).build()`) inside the optimizer's
first batch and returns full-table/all-columns stats — the comment there says
"bad stats are better than failing a query". So on DSv2 this is either a hard
failure or a decision made on a number the file itself calls bad, plus
per-query scan-build cost.
**2. Partitioned file tables never fire.** Their pre-pushdown `sizeInBytes`
is `defaultSizeInBytes` (`Long.MaxValue`), so the predicate is false at any
threshold:
```
right = spark.table("rightPart").filter($"p" === 0) // partitioned parquet
analyzed sizeInBytes = 9223372036854775807
optimized sizeInBytes = 9398
BroadcastNearestByJoin fired = false
```
A 9 KB right side takes the cross-product path. (Non-partitioned v1 tables
and in-memory DataFrames are unaffected, which is why the suite is green.)
**Fix.** Broadcast sizing belongs where `JoinSelection` does it — the
planner. Keep this predicate as-is for the strategy and make the rewrite gate
stats-free:
```scala
// RewriteNearestByJoin
case j @ NearestByJoin(...) if !SQLConf.get.nearestByBroadcastEnabled =>
```
That leaves the planner obligated to handle every `NearestByJoin` when the
flag is on, which is the "no strategy handles it" hazard your code comment
calls out. Two ways to discharge it:
- **(a)** Drop the size test from `NearestByJoinSelection` too, so the flag
means "always use the operator". For an `.internal()`, default-off flag that's
a reasonable contract, and it makes the two call sites agree by construction.
It gives up the automatic large-right fallback.
- **(b)** Keep the size test in the strategy and, when it fails, plan the
rewrite from there — `planLater(RewriteNearestByJoin.rewriteOne(j))` after
factoring the rewrite body out of the rule. This preserves today's semantics
with valid stats; the cost is that the rewritten subtree misses the
operator-optimization batches. This is essentially the two-pass approach you
deferred.
Either way, please add a DSv2 right-side test with the flag on — the repro
above is about 10 lines with `InMemoryTableCatalog`, and it's the shape most
users on Iceberg/Delta will hit first.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -178,6 +178,21 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
* Supports both equi-joins and non-equi-joins.
* Supports only inner like joins.
*/
+
+ object NearestByJoinSelection extends Strategy {
Review Comment:
**Finding 4.** The new strategy was inserted between the strategy-selection
scaladoc block (ending `*/` at line 180) and `JoinSelection`, so that block —
which documents equi-join/non-equi-join support, broadcast hash join build-side
preferences, and "Shuffle-and-replicate nested loop join ... Supports only
inner like joins" — now attaches to `NearestByJoinSelection`, which does none
of those things. `JoinSelection` is left undocumented.
Moving the object below `JoinSelection` restores both (and keeps the
join-strategy doc adjacent to the code it describes):
```scala
object JoinSelection extends Strategy with JoinSelectionHelper {
...
}
object NearestByJoinSelection extends Strategy {
...
}
```
Placing it above the scaladoc block would work equally well.
--
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]
