[jira] [Created] (SPARK-57487) Support distributed map join for medium-sized build tables via SQL hint

["Yu Gan (Jira)" <ji...@apache.org>]

Yu Gan created SPARK-57487:
------------------------------

             Summary: Support distributed map join for medium-sized build tables via SQL hint
                 Key: SPARK-57487
                 URL: https://issues.apache.org/jira/browse/SPARK-57487
             Project: Spark
          Issue Type: New Feature
          Components: SQL
    Affects Versions: 5.0.0
            Reporter: Yu Gan


Broadcast hash join is limited by driver memory and executor broadcast threshold — tables between ~200MB and 2GB often fall back to expensive shuffle joins, causing unnecessary data movement on the probe side.

  This PR introduces *Distributed Map Join (DMJ)*, a new join strategy that avoids full shuffle by:
  1. Hash-partitioning the build side into a fixed number of shards across executors
  2. Building a HashedRelation per shard with replica placement for fault tolerance
  3. Probe side performs batched RPC lookups (with bloom filter pre-filtering) instead of shuffling

  The strategy is triggered explicitly via SQL hint:
  \{code}
  SELECT /*+ DISTMAPJOIN(build_table(shard_count=5, replica_count=3)) */ ...
  FROM probe_table JOIN build_table ON ...
  \{code}

  h3. Production benchmark

  A LEFT JOIN workload in our production environment (16k cores, 120TB memory):
  - Probe side: ~5PB
  - Build side: ~2GB
  - Before (shuffle hash join): probe-side shuffle write + join stages took *~5 hours*
  - After (distributed map join): single probe-side join stage took *~2 hours*
  - *~60% wall-time reduction* by eliminating the probe-side shuffle entirely

  h3. Supported join types

  Inner, LeftOuter, RightOuter, LeftSemi, LeftAnti, ExistenceJoin.

  h3. Key components

  - *ShardManager / ShardManagerMaster* (core): manages shard lifecycle, location tracking, and replica placement
  - *ShardExchangeExec*: physical exchange operator that partitions build side into shards
  - *DistributedMapJoinExec*: physical join operator with async batched RPC lookups
  - *ShardLookupService*: Netty-based RPC layer for shard data transfer (pluggable via \{{spark.shard.service}})
  - Full AQE integration via ShardQueryStageExec

  This is similar in spirit to Hive's map join but distributed — no single-node bottleneck.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org