Uh oh!
There was an error while loading. Please reload this page.
[SPARK-57487][SQL] Support distributed map join for medium-sized tables via SQL hint - #56542
[SPARK-57487][SQL] Support distributed map join for medium-sized tables via SQL hint#56542yugan95 wants to merge 4 commits into
Conversation
sunchao
left a comment
There was a problem hiding this comment.
Summary
I reviewed the exact current head, 50a75f47be60f89210c1dc2d8974ab63a8298ad1, and then re-validated each candidate finding from the reachable Spark runtime path before posting this review. The overall idea is promising for very large fact-table joins, but the current implementation has correctness, security, fault-tolerance, and resource-lifecycle blockers that need to be addressed before it is safe to merge.
Prior state and problem
Spark currently has a gap between broadcast joins and shuffle joins. A broadcast hash join avoids shuffling the large probe side, but requires the complete build side to fit on every executor. Once the build side becomes too large or risky to broadcast, Spark generally falls back to a shuffle-based join and pays the cost of repartitioning the large probe table.
This PR targets the useful middle ground where the probe side is enormous and the build side is medium-sized: too large to broadcast everywhere, but small enough to partition into a limited number of remotely queryable hash-table shards.
Design approach
The patch adds a new DISTMAPJOIN hint and a shard service integrated with Spark Core and AQE. The build side is hash-partitioned into HashedRelation shards, optionally replicated, and accompanied by a merged Bloom filter. Probe tasks keep their original partitioning, batch lookup keys by shard, and use asynchronous Netty RPCs to fetch matching build rows.
This replaces the probe-side shuffle with a new distributed-state and RPC subsystem. That tradeoff can be valuable, but it also means authentication, ordering metadata, task cancellation, replica durability, response bounds, cleanup, and executor-loss behavior become correctness contracts of the SQL operator.
Correctness / compatibility analysis
I found the following introduced issues:
[P1] Clear the inherited output ordering.
DistributedMapJoinExec.scala:118preserves the probe partitioning but inheritsHashJoin.outputOrdering, which claims the probe ordering is preserved. It is not: a Bloom-negative row is emitted immediately while an earlier positive row remains buffered, and completed lookup batches are consumed in callback order. For example, a sorted left-outer probe can emit a later missing key before an earlier matching key.EnsureRequirements,RemoveRedundantSorts, window operators, and sort-merge joins trust this metadata and can omit a required sort, producing incorrect results. This operator should report no ordering unless responses are explicitly resequenced.[P1] Authenticate the shard lookup transport.
NettyShardLookupService.scala:68constructs a client factory withoutAuthClientBootstrapand deliberately starts the server with an empty bootstrap list at lines 78-84. UnlikeNettyBlockTransferService, it is not given aSecurityManager, does not apply the RPC SSL options, andNettyShardRpcServernever validates itsappId. Thereforespark.authenticate=truedoes not protect this executor-facing port: any peer that can reach it can issue lookup requests and retrieve shard rows or consume executor resources.[P1] Always create a shard exchange for a one-shard build. In
EnsureRequirements.scala:70, a build child withSinglePartitionalready satisfiesShardDistribution(requiredNumPartitions = 1), so the branch that createsShardExchangeExecat line 128 is skipped.DistributedMapJoinExec.resolveShardSetReflater receives the original build plan and throwsIllegalStateExceptionbecause no shard set exists.shard_count=1is accepted by the hint parser and is not covered by the new tests.[P1] Close outstanding lookups before freeing task pages.
DistributedMapJoinExec.scala:416registers onlymap.free()as its task-completion cleanup. Response ownership is transferred to the callback and released only whenBatchMatchReaderis fully exhausted, while callbacks continue enqueueing responses after submission. A normalLIMIT,take, cancellation, or lookup exception can abandon the current reader plus queued and future responses. This leaks pooled response buffers; additionally,Future { batch.wrapKeysBuffer() }may still serialize a request from Tungsten pages after the completion listener has freed those pages. The iterator needs task-scoped cancellation/draining and must release every batch and response before freeing the backing pages.[P1] Do not complete the exchange before requested replicas are installed.
ShardExchangeExec.scala:153requests replicas, but bothShardManagerMaster.installReplicaSetandShardManagerMasterEndpoint.scala:219discard their RPC futures. The exchange nevertheless marks its promise successful at line 174. If the primary executor is lost before a requested replica finishes and reports its location,removeShardManagerremoves the only holder and can no longer seed another copy. Probe-task retries continue using the already-materialized shard stage and cannot recover it. Thus evenreplica_count=2does not currently establish the advertised durability before probe execution begins.[P1] Exclude the driver from replica placement outside local mode. The driver initializes and registers a normal shard manager, while
ShardManagerMasterEndpoint.scala:198selects replica candidates from every registered manager. The suppliedisLocalflag is unused. Consequently, ordinary replica placement can copy and deserialize gigabyte-scale shards onto the driver and make it serve probe traffic. With two executors andreplica_count=3, the driver necessarily receives every shard; with more executors it can still be selected by the scoring function. This creates a direct driver-OOM/application-loss path and contradicts the stated executor-based placement model.[P1] Gate distributed map join on
hashJoinSupported.SparkStrategies.scala:292does not apply thehashJoinSupportedcheck used by broadcast and shuffled hash joins.RewriteCollationJoinhandles direct attribute equality, but non-binary-stable expression keys remain reachable—for example, equality on nestedSTRING COLLATE UTF8_LCASEfields producesGetStructFieldkeys rather thanAttributeReferencekeys. A hinted join on those keys constructsDistributedMapJoinExecand trips the assertions inHashJoin; without those assertions, the Bloom filter and rawUnsafeRowlookup would implement binary rather than collation equality. The hint should be rejected with the existing warning/fallback behavior.[P2] Bound or stream lookup responses.
HashedRelationAdapter.scala:67allocates one pooled directByteBufand appends every matching build row for every key before returning anything.maxBatchSizelimits probe-key count, not response cardinality. A single hot key can make one response approach the whole shard, and the 16 server workers can materialize several such buffers concurrently. This should use chunked/streamed responses or enforce a strict byte bound with backpressure.[P2] Clean shard data on both failed and successful exchange lifecycles.
ShardExchangeExec.scala:115allocates the set before tasks installMEMORY_AND_DISKblocks, but registers cleanup only after collection and Bloom merging succeed at line 175; the catch block only fails the promise. A failed build therefore leaves partially installed shards for the lifetime of the application and dynamic allocation treats their executors as cached. On successful cleanup,ShardManager.unpersistremoves BlockManager data but never removesshardSetInfo,shardSetLocations, or per-manager_shardsentries fromShardManagerMasterEndpoint. Long-lived applications accumulate permanent driver metadata, and later executor removal can try to replicate already-deleted shards.[P2] Enforce the in-flight limit for tail batches and validate it as positive. EOF calls
flushLookupover every per-shard tail batch, andflushLookupsubmits them all before returning. With 100 shards this can launch 100 RPCs per task despitemaxInFlightNum=8. Separately,SQLConf.scala:7491accepts zero or negative values; then line 515 polls an empty queue forever before the first request is submitted. Submission should be throttled centrally and the config should require a value greater than zero.[P2] Release the per-attempt request retain when connection creation fails.
ShardManager.scala:293retains the request once per location attempt and relies on Netty outbound ownership to release it. IfclientFactory.createClientthrows beforesendManagedRpc, the catch reports failure but does not release that retained reference. Retries add further retains, while the final chain releases only the original base reference. Stale locations after executor loss therefore leak one pooled request buffer per failed pre-send attempt.
Current CI independently confirms three additional merge blockers on this head:
- MiMa reports that inserting
shardManagerFactoryinto the publicSparkEnvconstructor removes the previous JVM signature. SparkConfigBindingPolicySuitereports thatspark.shard.enabledand all four new SQL configs lack mandatory binding policies.ContextCleanerSuitefails because insertingshardSetIdsbeforecheckpointIdssilently reinterprets the existing positional call at line 212 as shard set1.
Key design decisions
- Keeping the strategy explicit through a hint is appropriate for a first version because the cost boundary depends on build size, probe size, match rate, fanout, and RPC latency.
- The merged Bloom filter is important: this strategy is most attractive when it can reject many probe keys without an RPC.
- Shard replication is the right mechanism for long-running jobs, but it must be a completed, observable exchange invariant rather than best-effort background work.
- A separate transport is reasonable only if it inherits Spark's authentication, encryption, buffer-ownership, cancellation, and metrics contracts.
Implementation sketch
The physical flow is: build-side hash shuffle -> per-shard HashedRelation materialization -> optional replica installation -> merged Bloom filter -> probe scan without repartitioning -> per-shard batched RPC lookup -> local join result assembly. AQE wraps the build exchange as ShardQueryStageExec and exchange reuse can share the materialized shard set.
The happy-path join semantics receive useful coverage across join types and AQE modes. The missing coverage is concentrated in metadata contracts and adverse lifecycle paths: ordered probes, one-shard input, collated keys, authentication, early termination, executor loss during replica installation, skewed/high-fanout responses, and failed-exchange cleanup.
Behavioral changes worth calling out
- Enabling the feature opens a new executor-facing network port with its own protocol and memory behavior.
- Probe-side row order is not preserved even though the current physical metadata says it is.
- Shard state is not shuffle output, so Spark's normal lost-shuffle recovery does not rebuild it automatically.
- Dynamic allocation must retain live shard holders but release them promptly and completely after both success and failure.
- The strategy can replace shuffle bytes with very high lookup QPS and response traffic; match selectivity and hot-key fanout are essential operational constraints.
Suggested improvements
I suggest addressing this in layers:
- Fix the deterministic correctness and security blockers: output ordering, one-shard planning, key support, and authenticated/encrypted transport.
- Make replica installation part of exchange completion, exclude the driver from placement, and define how a lost last replica invalidates/rebuilds the query stage.
- Give every request, response, batch, and task page an explicit cancellation-safe ownership lifecycle; add response byte limits or streaming.
- Add symmetric cleanup for failed and successful exchanges, including all master metadata, and centralize backpressure/config validation.
- Add targeted tests for every failure path above, then benchmark not only wall time but also RPC QPS/bytes, direct memory, skew behavior, and executor-loss recovery.
The optimization is valuable enough to pursue, but I do not think the current patch is safe to merge until these contracts are closed.
@yugan95 I'm not a PMC member nor a maintainer of SQL area (I mainly deal with Structured Streaming), but given the large scope of change across multiple modules with huge code diff while addressing a specific use case, I wonder we should step back and make a consensus about the direction in prior. Apache Spark has a process for this - https://spark.apache.org/improvement-proposals.html The main purpose is to build a consensus on the community that the improvement is something we want to adopt. The Heilmeier form isn't purposed to bring up detailed design, but high-level design is appreciated (this change obviously warrants it since new distributed data exchange with RPC is introduced). Also probably need a much clearer answer about "when" users will be benefited by this change, especially that this is "opt-in" than opt-out. 5TB vs 2GB example in the JIRA ticket doesn't feel like a very general case, or might need more data about the trade-off between the cost of eliminating shuffle vs retrieving data via remote RPC instead of pre-loading the whole shard after shuffle - simply saying, do we have an answer for this, if you were users which criteria warrants this feature to be enabled? The process requires one PMC member to be a shepherd - if you don't have one to contact and co-work with this process, probably start with dev@ mailing list with empty shepherd, and I assume you can find volunteer as long as your proposal is on consensus to the shape of "good to go". Thanks again! |
Thanks @sunchao for the thorough review! All issues except issue 8 are addressed in the latest push. Issue 1 - outputOrdering — Overrode to Issue 2 - Shard transport auth — Issue 3 - SinglePartition vs ShardDistribution(1) — Added Issue 4 - Task cancellation buffer leak — Added Issue 5 - Replica installation sync — Issue 6 - Driver exclusion — Candidates filter out Issue 7 - hashJoinSupported — Issue 8 - Response size bound — Will address as a follow-up. Adding chunked responses requires protocol changes that deserve a separate review. Issue 9 - Exchange cleanup — Failed exchanges call Issue 10 - EOF flush throttling — Issue 11 - Request retain leak — Added CI fixes (binding policies, MiMa, ContextCleanerSuite, Java checkstyle) also included. |
yugan95
commented
Jun 17, 2026
@HeartSaVioR Thanks for the suggestion on the SPIP process. A couple of clarifications on the scope and entry point: This feature is double-gated — it requires The use case is the gap between broadcast join and shuffle join: the build side is too large to broadcast (tens of GB) but the probe side is very large (PB-scale in our case — a minor correction to the JIRA ticket which mentions 5PB, not 5TB). Shuffling the probe side at that scale is expensive. Distributed map join avoids the probe-side shuffle by partitioning the build side into remotely queryable shards, with a Bloom filter to reduce RPC volume. This has been running stably in our production environment. I'm happy to draft a SPIP proposal if the community feels that's the right process for this scope. Also glad to provide more detailed benchmarks on the trade-off between shuffle cost and RPC cost across different data profiles. Open to either path. |
That sounds to me even bigger scope than what I thought - should the shard infrastructure be also opt-in? What's the downside of making it by default? Is it a separate process than executor or inlined to the executor? If it's a separate process, then how the Spark infra launches them? This obviously needs a design doc (there seems to be meaningful number of design decisions/options you might already went through while someone might want to revisit), and "Since it's opt-in, this doesn't impact others unless they opt-in" does not simply justify an addition of the feature. Again, I'm not an expert of this area, but I feel like folks in SQL area might be a same page that this warrants SPIP. cc. @cloud-fan@dtenedor FYI |
yugan95
commented
Jun 17, 2026
@HeartSaVioR Thanks for the follow-up questions. To clarify the shard infrastructure: it is fully inlined in the executor process — just an additional Netty server and a few RPC endpoints started inside The reason it defaults to off is that the Netty server opens an additional port on each executor. Keeping it opt-in felt more appropriate rather than starting a new listener by default. I understand the concern about scope and I'm open to drafting a SPIP. Will wait to hear from the SQL area maintainers as well. |
f4c6ff1 to
5ebfb07Comparecloud-fan
commented
Jul 1, 2026
This looks like a big feature and it's better to start with a design review. |
Thanks @cloud-fan. I've attached a design document to the JIRA ticket SPARK-57487 covering architecture, data flow, key design decisions, and rejected alternatives. Happy to iterate based on feedback. |
sunchao
left a comment
There was a problem hiding this comment.
Follow-up review of exact head 2f3cd3453959738bc3539fd15c3ae84d39777ee0; verified findings are attached to current diff lines.
| @Override | ||
| public void onSuccess(ByteBuffer response) { | ||
| respond(new RpcResponse(req.requestId, new NioManagedBuffer(response))); | ||
| if (rpcHandler instanceof RpcHandler.ManagedRpcHandler) { |
There was a problem hiding this comment.
[P1] Preserve managed-RPC dispatch through authentication wrappers. With spark.authenticate=true, AuthServerBootstrap wraps NettyShardRpcServer in AuthRpcHandler or SaslRpcHandler, neither of which implements ManagedRpcHandler. This check therefore selects the legacy ByteBuffer path; after authentication, the wrapper delegates to NettyShardRpcServer.receive(ByteBuffer), which unconditionally throws UnsupportedOperationException. Consequently every shard lookup fails on authenticated clusters. Preserve authenticated managed dispatch and add an authentication-enabled integration test.
| _env.blockManager.initialize(_applicationId) | ||
| FallbackStorage.registerBlockManagerIfNeeded( | ||
| _env.blockManager.master, _conf, _hadoopConfiguration) | ||
| _env.initializeShardManager() |
There was a problem hiding this comment.
[P1] Register the driver shard endpoint before executors can start._taskScheduler.start() has already started the cluster backend at line 625 and can launch/register executors before this call creates ShardManagerMaster. Executor construction immediately calls its own initializeShardManager(), synchronously looks up/registers with that driver endpoint, and exits when the endpoint is not present. This makes feature-enabled cluster startup racy, particularly with Kubernetes/YARN or quickly allocated executors. Register the driver endpoint before the backend permits executor registration, or defer/retry executor initialization.
| case scala.util.Failure(e) => | ||
| logWarning(log"Replica installation failed for" + | ||
| log" (${MDC(SHARD_SET_ID, setId)}, ${MDC(SHARD_ID, shardId)})", e) | ||
| context.reply(true) |
There was a problem hiding this comment.
[P1] Propagate replica-installation failures to the build exchange. This failure branch logs an unsuccessful replica RPC but still replies true; consequently ShardManagerMaster.installReplicaSet returns normally and ShardExchangeExec publishes a successful materialized stage without the requested replicas. For replica_count=2, losing the remaining primary then permanently breaks the join. The previous asynchronous-installation concern is therefore not fixed: complete the RPC exceptionally and fail/retry the exchange instead of acknowledging failed replication as success.
| val locations = shardLocations.get(shardId) | ||
| if (locations != null) { | ||
| locations -= shardManagerId | ||
| if (locations.isEmpty) { |
There was a problem hiding this comment.
[P1] Rebuild a materialized shard set when its final replica is lost. The default replica_count=1 means ordinary executor loss reaches this branch after a completely successful exchange, yet it only removes the location and logs. Subsequent lookups fail with an ordinary SparkException, while the AQE shard stage remains materialized with the same now-unrecoverable set ID; every task retry therefore fails rather than rebuilding the build side. Executor loss or finite cached-executor idle timeouts must invalidate/recompute the shard stage, or the strategy needs a stronger durability invariant.
| override def shardSetCleaned(setId: Long): Unit = { | ||
| executors.asScala.foreach { case (_, exec) => | ||
| val toRemove = exec.nonRddCachedBlocks.collect { |
There was a problem hiding this comment.
[P2] Keep shard cleanup on the executor-monitor event thread.ContextCleaner invokes shardSetCleaned on its dedicated cleaner thread, but this callback directly iterates and mutates each executor's mutable nonRddCachedBlocks/cachedBlocks while onBlockUpdated and other listener callbacks mutate the same structures on the listener-bus management thread. This violates the documented event-thread confinement and can corrupt shard tracking or remove/retain the wrong executor. Follow the adjacent shuffleCleaned implementation and post a shard-cleanup event to the listener bus.
| } | ||
| def createDistributedMapJoin(onlyLookingAtHint: Boolean) = { | ||
| if (hashJoinSupport) { |
There was a problem hiding this comment.
[P1] Gate DISTMAPJOIN on initialized shard infrastructure.spark.shard.enabled defaults to false, and SparkEnv.initializeShardManager leaves shardManager null in that configuration, but this planner branch checks only hashJoinSupport before selecting DistributedMapJoinExec for a hint. A hinted query in an otherwise default application therefore reaches ShardExchangeExec and dereferences the null manager instead of ignoring the unavailable strategy/falling back. Check the feature gate before constructing the physical join and add a default-disabled regression test.
| } | ||
| item = lookupQueue.poll() | ||
| } | ||
| map.free() |
There was a problem hiding this comment.
[P1] Quiesce outstanding lookup work before freeing Tungsten pages.ShardManager.fetchRemoteBatch first schedules Future { batch.wrapKeysBuffer() }, which reads addresses from this task's pages, but an early LIMIT, cancellation, or failure can run this completion listener and call map.free() while that serialization future is still queued/running. TaskMemoryManager.freePage clears the page table first, so the later serializer dereferences freed pages; the callback's later cancelled check cannot prevent it. The check-then-enqueue path also races queue drainage. Cancel/await outstanding work before releasing task memory and make completion ownership atomic.
| val manager = SparkEnv.get.shardManager | ||
| implicit val ec: ExecutionContextExecutorService = manager.lookupEc | ||
| while (iter.hasNext && !cancelled) { | ||
| while (numInFlight >= maxInFlightNum && !cancelled) { |
There was a problem hiding this comment.
[P1] Stop flushing once throttling prepares an output row. With maxInFlightNum=1 and three populated shard tail batches A/B/C, this throttle consumes A and iterateLookup() sets prepared=true/nextRowVal=A, but flushLookup continues, submits B, then overwrites currentReader, currentBatchIter, and nextRowVal while consuming B. A's row and any remaining duplicate matches silently disappear, and its reader is abandoned; the default limit of eight reproduces with ten populated shards. Preserve the prepared row/current reader and resume flushing only after the caller consumes them.
| val keyUr = new UnsafeRow(numKeyFields) | ||
| val rel = manager.getLocalValue[HashedRelation](setId, shard).asReadOnlyCopy() | ||
| val valuesBuf = alloc.buffer(INITIAL_RESPONSE_BUFFER_BYTES) |
There was a problem hiding this comment.
[P2] Release the pooled response buffer when lookup construction fails. A 1-MiB pooled buffer is allocated here, but any later exception during filter deserialization/evaluation, malformed-key decoding, or row serialization escapes without releasing it; NettyShardRpcServer catches the failure and releases only the request. For example, a valid existing-shard header followed by one extra byte passes this allocation and then throws from readInt(), leaking at least 1 MiB per failed request/retry. Transfer ownership only on successful return and release the response buffer on every exceptional path.
| val advanceReadKey = UnsafeRowBufCodec.makeAdvanceRead(keyUr, keysBuf) | ||
| val advanceWrite = UnsafeRowBufCodec.makeAdvanceWrite(valuesBuf) | ||
| val predicate = filterCache.computeIfAbsent(setId, _ => { |
There was a problem hiding this comment.
[P1] Do not share one mutable generated predicate across RPC workers.computeIfAbsent stores one BasePredicate per shard set, while NettyShardRpcServer evaluates lookups concurrently on up to 16 worker threads. Generated predicates embed mutable fields; for example, a build-only expression using elt(size(sequence(d.start,d.stop,d.step)), d.keep, d.drop) can remain in the join because sequence(..., step) is throwable, and concurrent elt evaluations race on its generated inputVal field. This includes/excludes the wrong build rows and silently corrupts join results. Create a per-request/per-thread predicate instead of sharing its evaluator.
@sunchao I'd love to hear your opinion about the scope of the change since you are still reviewing this. Do you think this is something one (or a couple of) committer+ can review and merge in? IMHO this looks to touch multiple components and warrants SPIP; he already put the design doc as SPIP format. Instead of keep reviewing this, why not volunteer shepherd role of this SPIP and help getting this through SPIP process? @yugan95 |
@HeartSaVioR yes we should create a SPIP and discuss through it. I'd be happy to help shepherd the process. My Codex bot somehow picked up this PR again and reviewed it. |
Thanks @sunchao for offering to shepherd, and @HeartSaVioR for the guidance on the process. Let's move this to the SPIP track. The design doc attached to SPARK-57487 is already in SPIP format (Heilmeier Q1–Q8 + API/design/rejected-alternatives appendices). @sunchao — would you like to review the doc before I send it to dev@, or shall I go ahead and start the thread and we iterate on the list? Happy to adjust framing based on your preference as shepherd. I'll also fold the Codex review findings into the implementation as the SPIP progresses — thanks for the thorough pass. |
sunchao
commented
Aug 5, 2026
@yugan95 could you convert the SPIP into a Google doc so people can comment on? The overall architecture looks good to me and I'm excited about this feature. Please start a discussion on the dev list! |
Thanks @sunchao! I've put the SPIP into a Google doc (open for comments): https://docs.google.com/document/d/1ioj2vaY1-kgvGX-FCCVA95ybBYjlnc20KePS1GNPbSM Starting the [DISCUSS] thread on dev@ now — let's continue the design discussion there. I'll drop the thread link here once it's up. Started the dev@ discussion: |
FrankPortman
commented
Aug 24, 2026
this feature looks incredibly useful |
What changes were proposed in this pull request?
This PR introduces Distributed Map Join (DMJ), a new join strategy that avoids full probe-side shuffle by building a distributed hash table service for the build side. The probe side performs batched RPC lookups instead of shuffling.
How it works:
HashedRelationon executors, with configurable replica placement for fault tolerance.The strategy is triggered explicitly via SQL hint:
Key components:
ShardManager/ShardManagerMasterShardLookupServicespark.shard.service)ShardExchangeExecDistributedMapJoinExecShardQueryStageExecShardDistribution/ShardPartitioningBufferedShardRowMapResolveHints(extended)DISTMAPJOINSupported join types: Inner, LeftOuter, RightOuter, LeftSemi, LeftAnti, ExistenceJoin.
Why are the changes needed?
Broadcast hash join is limited by driver memory and the executor broadcast threshold — build-side tables between ~200MB and 2GB often fall back to expensive shuffle joins, causing massive data movement on the probe side.
Production benchmark (16k cores, 120TB memory cluster, DRA enabled):
~60% wall-time reduction by eliminating the probe-side shuffle entirely.
This is similar in spirit to Hive's map join but distributed — no single-node bottleneck.
Does this PR introduce any user-facing change?
Yes. Adds a new SQL hint
DISTMAPJOINand the following configs:spark.shard.enabledfalsespark.sql.execution.distributedMapJoin.maxInFlightNum8spark.sql.execution.distributedMapJoin.maxBatchSize1024spark.sql.execution.distributedMapJoin.bloomFilterCapacity5242880spark.sql.execution.distributedMapJoin.exchangeTimeout30minHow was this patch tested?
DistributedMapJoinSuite: end-to-end tests onlocal-cluster[2,1,512]covering inner join, left/right outer join, left semi join, left anti join, with and without join conditions, null key handling, and multi-column join keys.DistributedMapJoinSuiteAE: same test coverage with AQE enabled.ContextCleanerSuite: extended to cover shard-set cleanup.CachedTableSuite: extended withshardSetCleanedlistener stub.