[SPARK-36082][SQL][FOLLOWUP] Restore NAAJ broadcast hash join by default #58631
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7366,15 +7366,34 @@ object SQLConf { | |
| val OPTIMIZE_NULL_AWARE_ANTI_JOIN = | ||
| buildConf("spark.sql.optimizeNullAwareAntiJoin") | ||
| .internal() | ||
| .doc("When true, NULL-aware anti join execution can be planned as " + | ||
| "BroadcastHashJoinExec with flag isNullAwareAntiJoin enabled, " + | ||
| "optimized from O(M*N) calculation into O(M) calculation " + | ||
| "using Hash lookup instead of Looping lookup. " + | ||
| "Only support for singleColumn NAAJ for now. The optimization is also controlled by " + | ||
| "spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold.") | ||
| .version("3.1.0") | ||
| .booleanConf | ||
| .createWithDefault(true) | ||
|
|
||
| val NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD = | ||
| buildConf("spark.sql.optimizeNullAwareAntiJoin.broadcastThreshold") | ||
| .internal() | ||
| .doc("Configures the maximum estimated size in bytes of the right side of a " + | ||
|
Contributor
There was a problem hiding this comment. The doc only says the fallback "may still use a broadcast nested loop join", which reads as if lowering this config can stop a large right side from being broadcast. With no broadcast hint there is nothing else to pick for this shape: the condition is non-equi so no SMJ/SHJ is available, The PR description explains this; the config doc, which is usually all an operator sees via
Contributor
Author
There was a problem hiding this comment. Updated the config doc in 0b7c77b. It now explains that the fallback may broadcast the right side using a nested-loop representation that consumes more memory and runs in |
||
| "single-column null-aware anti join for which Spark uses the broadcast hash join " + | ||
| "optimization. This configuration takes effect only when " + | ||
| "spark.sql.optimizeNullAwareAntiJoin is enabled. A negative value allows the " + | ||
| "optimization regardless of the estimated size, while zero disables it. If the " + | ||
| "estimated size exceeds a positive value, Spark falls back to regular join planning. " + | ||
| "The fallback may still broadcast the right side with a nested-loop representation " + | ||
| "that uses more memory and runs in O(M * N) time. Join hints do not override this " + | ||
| "configuration when the broadcast hash optimization is selected. This configuration " + | ||
| "also controls whether a null-aware anti join can be pushed below an aggregate.") | ||
| .version("4.2.1") | ||
| .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) | ||
| .bytesConf(ByteUnit.BYTE) | ||
| .createWithDefault(-1) | ||
|
|
||
| val LEGACY_DUPLICATE_BETWEEN_INPUT = | ||
| buildConf("spark.sql.legacy.duplicateBetweenInput") | ||
| .internal() | ||
|
|
@@ -9743,6 +9762,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf { | |
| def optimizeNullAwareAntiJoin: Boolean = | ||
| getConf(SQLConf.OPTIMIZE_NULL_AWARE_ANTI_JOIN) | ||
|
|
||
| def nullAwareAntiJoinBroadcastThreshold: Long = | ||
| getConf(SQLConf.NULL_AWARE_ANTI_JOIN_BROADCAST_THRESHOLD) | ||
|
|
||
| def legacyDuplicateBetweenInput: Boolean = | ||
| getConf(SQLConf.LEGACY_DUPLICATE_BETWEEN_INPUT) | ||
|
|
||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Negative meaning unlimited inverts the broadcast-threshold family:
autoBroadcastJoinThresholdand its adaptive twin both document-1as the way to disable broadcasting, andcanBroadcastBySize(joins.scala:370) refuses any negative size. An operator whose driver is dying on a NAAJ broadcast will set-1and get the size limit removed instead.bytesConfstrips the sign, so-2glands in the same bucket.createOptionalavoids the collision: unset means no limit, a set value is a real limit, and disabling stays withspark.sql.optimizeNullAwareAntiJoin. One reader to update, plus the-2assertion inJoinSelectionHelperSuite. If the current semantics stay, the doc should say they read the opposite way fromautoBroadcastJoinThreshold.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for raising this. I prefer to keep the current
-1semantics. This threshold is not intended to followspark.sql.autoBroadcastJoinThreshold: disabling the NAAJ optimization is already controlled byspark.sql.optimizeNullAwareAntiJoin=false, while this threshold needs an unbounded default to restore the original behavior.Long.MaxValueis not sufficient becausesizeInBytesis aBigInt; thethreshold < 0branch is genuinely unbounded and covers estimates aboveLong.MaxValue.createOptionalwould encode the same unbounded state asNone, but would not improve correctness here. The config is internal and its documentation explicitly calls out the negative-value semantics, so I prefer to keep-1.