Also another clarification requested In the worst case scenario of all partioned keys are required by all the probing nodes, it degenerates to BHJ, right?
democracy of barbarians is worse than dictatorship On Thu, Aug 6, 2026, 2:04 AM Asif Shahid <[email protected]> wrote: > Just went through the doc.. > Based on that it seems that you will be batching to minimize the calls. > Pls correct me if wrong > > democracy of barbarians is worse than dictatorship > > On Thu, Aug 6, 2026, 1:57 AM Asif Shahid <[email protected]> wrote: > >> 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] >>> >>>
