Skip to content

[SPARK-56065][SQL] Add AQE fallback from failed broadcast joins to shuffle joins #54925

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sunchao wants to merge 8 commits into
base: master
Choose a base branch
from
+638 −6
Open

[SPARK-56065][SQL] Add AQE fallback from failed broadcast joins to shuffle joins #54925

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
db01dd4
initial commit
sunchao Mar 5, 2026
0795b86
Avoid mutating session SQLConf in AQE fallback replanning
sunchao Mar 19, 2026
d062883
Tighten failed broadcast relation matching in AQE fallback
sunchao Mar 19, 2026
02c9133
Preserve SessionState planner behavior in AQE fallback replanning
sunchao Mar 19, 2026
599c609
Use cloneSession in AQE fallback replanning
sunchao Mar 19, 2026
b165ab2
Remove failed broadcast logical query stages before AQE fallback
sunchao Mar 19, 2026
d930686
Restrict AQE root broadcast fallback exemption to root stage
sunchao Mar 19, 2026
2fd2cff
Address comments, use NO_BROADCAST_HASH hint, and simplify code
sunchao Mar 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1229,6 +1229,14 @@ object SQLConf {
.bytesConf(ByteUnit.BYTE)
.createOptional

val ADAPTIVE_BROADCAST_JOIN_FALLBACK_TO_SHUFFLE_ENABLED =
buildConf("spark.sql.adaptive.broadcastJoin.fallbackToShuffle.enabled")
.doc("When true, adaptive execution retries with broadcast joins disabled if a broadcast " +
"query stage fails because it exceeds broadcast table row or size limits.")
.version("3.5.0")
.booleanConf
.createWithDefault(false)

val ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD =
buildConf("spark.sql.adaptive.maxShuffledHashJoinLocalMapThreshold")
.doc("Configures the maximum size in bytes per partition that can be allowed to build " +
Expand Down Expand Up @@ -7425,6 +7433,9 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
def nonEmptyPartitionRatioForBroadcastJoin: Double =
getConf(NON_EMPTY_PARTITION_RATIO_FOR_BROADCAST_JOIN)

def adaptiveBroadcastJoinFallbackToShuffleEnabled: Boolean =
getConf(ADAPTIVE_BROADCAST_JOIN_FALLBACK_TO_SHUFFLE_ENABLED)

def coalesceShufflePartitionsEnabled: Boolean = getConf(COALESCE_PARTITIONS_ENABLED)

def minBatchesToRetain: Int = getConf(MIN_BATCHES_TO_RETAIN)
Expand Down
Loading