Interesting. One question: Assume that there are say n hash partitions of build side data. On a given node the streaming side( probe side ) has keys corresponding to to those n partitions.. would it mean n calls to fetch the hashed partitions?
Or you plan to group the probe side in batches, identify the needed partitions in a single call As multiple hash partitions may be residing on a single executor. Regards Asif democracy of barbarians is worse than dictatorship On Thu, Aug 6, 2026, 1:44 AM Yu Gan <[email protected]> wrote: > Hi all, > > I'd like to start a discussion on a SPIP for a new opt-in join > strategy: Distributed Map Join (DMJ). > > - Design doc (SPIP format, open for comments): > > https://docs.google.com/document/d/1ioj2vaY1-kgvGX-FCCVA95ybBYjlnc20KePS1GNPbSM > - JIRA: https://issues.apache.org/jira/browse/SPARK-57487 > - PR (reference implementation): > https://github.com/apache/spark/pull/56542 > - Shepherd: Chao Sun (sunchao) > > Problem. There's a gap between broadcast join and shuffle join. When > the build side is medium-sized -- too large to broadcast (roughly > hundreds of MB up to ~10 GB) -- but the probe side is very large > (PB-scale in our production case), Spark falls back to a shuffle join, > and shuffling the probe side becomes the dominant cost of the query. > > Proposal. DMJ avoids the probe-side shuffle. The build side is > hash-partitioned into N remotely-queryable shards held in-memory on > executors; probe tasks send batched RPC lookups to the shard-holding > executors, pre-filtered by a Bloom filter, and stream matched rows > back. No probe-side shuffle. > > Scope / safety. The feature is double-gated and off by default: > - spark.shard.enabled=true starts the shard infrastructure at > application launch (inlined in the executor process -- an additional > Netty server + RPC endpoints in SparkEnv, no separate process or > deployment step). > - An explicit per-query SQL hint /*+ DISTMAPJOIN(table) */ activates > the strategy. > > There is no cost-based selection; the planner never chooses DMJ > automatically. When disabled, no code paths are initialized and there > is zero overhead. > > Production status. Running stably in our 16k-core production > environment for 2+ months on PB-scale workloads (~60% wall-time > reduction on qualifying fact-dimension joins). > > The design doc covers architecture, data flow, the key design > decisions, risks, and rejected alternatives. I'd appreciate feedback > on the overall approach and on the specific design questions raised on > the PR (e.g. whether the shard infrastructure should remain opt-in, > durability/fault-tolerance model, and transport/auth). > > Thanks, > Yu Gan > > --------------------------------------------------------------------- > To unsubscribe e-mail: [email protected] > >
