sunchao commented on code in PR #58404:
URL: https://github.com/apache/spark/pull/58404#discussion_r3887065977
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -341,7 +341,7 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
}
case j @ ExtractSingleColumnNullAwareAntiJoin(leftKeys, rightKeys)
- if canBroadcastBySize(j.right, conf) =>
+ if canPlanAsBroadcastHashJoin(j, conf) =>
Review Comment:
[P1] Normalize floating keys before enabling the expanded hash path
With nullable DOUBLE keys, a known RHS row count,
`spark.sql.autoBroadcastJoinThreshold=0`, and AQE disabled, this changes a
previously correct nested-loop join into a null-aware hash join. For left
values `[-0.0, 2.0, NULL]` and right values `[+0.0, 1.0]`, `NOT IN` should
return only `2.0`, but the hash lookup also returns `-0.0`.
`NormalizeFloatingNumbers` only handles `ExtractEquiJoinKeys`, so it skips
the NAAJ `Or(EqualTo, IsNull(EqualTo))` shape. `HashJoin.normalizeJoinKeys`
only injects collation keys, leaving these floating keys unchanged, and
`UnsafeHashedRelation` compares their raw bytes. The missing normalization
predates this PR, but the formerly correct threshold-disabled execution changes
here. Please normalize these keys before expanding hash eligibility, or retain
the nested-loop path for affected types.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:
##########
@@ -289,6 +291,20 @@ case object BuildLeft extends BuildSide
trait JoinSelectionHelper extends Logging {
+ // Keep this synchronized with BroadcastExchangeExec's non-Long hashed
relation limit.
+ private val maxBroadcastHashRows = (BytesToBytesMap.MAX_CAPACITY /
1.5).toLong
+
+ private def canBuildNullAwareAntiJoinHashRelation(
+ rightKeys: Seq[Expression],
+ right: LogicalPlan,
+ conf: SQLConf): Boolean = {
+ val usesLongHashedRelation =
+ rightKeys.length == 1 &&
rightKeys.head.dataType.isInstanceOf[IntegralType]
+ usesLongHashedRelation ||
right.stats.rowCount.fold(canBroadcastBySize(right, conf)) {
+ _ < maxBroadcastHashRows
+ }
Review Comment:
[P2] Account for hash storage exceeding the broadcast byte limit
Row capacity does not establish that the hashed representation fits the
broadcast byte limit. With `spark.sql.autoBroadcastJoinThreshold=0` and
`spark.sql.maxBroadcastTableSize=1MB`, even a tiny nonempty RHS with nullable
integral keys and no actual null values now fails: `LongHashedRelation`
allocates at least a 1 MiB page plus its index, whereas the previous identity
broadcast contains only the raw rows and fits.
`BroadcastExchangeExec` checks `HashedRelation.estimatedSize` for hashing
but sums raw `UnsafeRow` sizes for identity broadcasts, so selecting the same
build side does not preserve byte-limit behavior. The same representation
overhead matters for larger inputs under the default limit. Please account for
this capacity difference or preserve a fallback when hash storage exceeds the
byte limit.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PushDownLeftSemiAntiJoin.scala:
##########
@@ -75,7 +74,11 @@ object PushDownLeftSemiAntiJoin extends Rule[LogicalPlan]
val makeJoinCondition = (predicates: Seq[Expression]) => {
replaceAlias(predicates.reduce(And), aliasMap)
}
- pushDownJoin(join, canPushDownPredicate, makeJoinCondition)
+ pushDownJoin(
+ join,
+ canPushDownPredicate,
+ makeJoinCondition,
+ canPlanAsBroadcastHashJoin(_, conf))
Review Comment:
[P2] Preserve the original aggregate join eligibility check
Checking only the rewritten join loses the original `BuildLeft` fallback
when CBO estimates that an aggregate reduces a large input to a few groups. For
an explicit `LEFT ANTI JOIN` above the aggregate with a nullable integral
condition `g.k = r.k OR isnull(g.k = r.k)`, the aggregate can be below the
broadcast threshold while both its child and the RHS exceed it. The original
plan broadcasts the small aggregate; replacing it with its large child makes
the rewritten join choose `BuildRight`, so this callback permits pushdown and
forces the large RHS into a hash broadcast.
At 512,000,000 RHS rows, that broadcast necessarily fails the exchange row
limit even though the original `BuildLeft` join can stream the RHS. Smaller
inputs can exceed the byte limit or driver memory. Please retain the original
eligibility guard as well as validating the rewritten join.
--
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]
