Spark / execution / joins
Only One Side of a Sort-Merge Join Streams
That a sort-merge join sorts both tables and then streams through them, so it costs a sort and uses bounded memory. All three parts are wrong. The Sort that EnsureRequirements inserts is written `SortExec(requiredOrdering, global = false, child)` — it sorts within each shuffle partition and produces no globally ordered result. The expensive part is the Exchange, not the Sort, and both disappear entirely when a child's outputPartitioning already satisfies ClusteredDistribution(keys), which is why a sort-merge join can legitimately have zero shuffles in its plan. And only the streamed side streams: SortMergeJoinScanner.bufferMatchingRows() copies every buffered-side row carrying the current join key into an ExternalAppendOnlyUnsafeRowArray, whose spark.sql.sortMergeJoinExec.buffer.in.memory.threshold defaults to ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH — Integer.MAX_VALUE minus 15, or 2,147,483,632 rows — so the documented in-memory guarantee is no bound at all. Which side gets buffered is decided by joinType alone: right for inner and left outer, left for right outer.
A sort-merge join asks its children for ClusteredDistribution on the join keys and an ascending per-partition ordering, then walks two iterators in lockstep — streaming one side row by row and buffering, in an ExternalAppendOnlyUnsafeRowArray, every row of the current key from the other side, where which side is buffered is fixed by the join type and never by size.
Where this is already explained
- Dynamic File Pruning
That a selective join produces a selective scan: if the dimension filter matches four thousand rows out of two hundred thousand, dynamic file pruning will read roughly four thousand rows' worth of files. What is pushed down is a set of key values, and a file survives if any one of them falls between that file's recorded minimum and maximum for the join key. Four thousand values scattered evenly across the key domain therefore keep every file alive on a perfectly Z-ordered table, while the same four thousand values in one contiguous block keep five. The two failures that follow are both silent: when the join key sits past delta.dataSkippingNumIndexedCols there is no min/max to test and nothing can be excluded, and when the build side outgrows spark.sql.autoBroadcastJoinThreshold Spark replaces the pruning subquery with a literal true. In both cases the physical plan still shows a dynamicpruning expression on the scan and the query profile still reports the feature as applied.
- Spark Broadcast Hash Join
That broadcast() is a free speedup for any small-looking table and that raising spark.sql.autoBroadcastJoinThreshold makes more joins fast. The planner tests a compressed on-disk estimate, the driver pays the decompressed in-memory price, and every executor holds a full copy for the life of the query — so the setting that looks like a speed dial is really a driver-heap and cluster-memory dial.
- Spark Table Statistics
That Spark knows how big your data is, so a table small enough to broadcast will be broadcast. Spark plans with an estimate that is the compressed on-disk size, is unchanged by any WHERE clause by default, and is the product of both sides above a join — so it is routinely wrong by 100x in both directions, and the join strategy follows the estimate rather than the data.
3 published lessons depend on this concept, which is what moves it up the writing queue. Nothing is hidden behind this page — it has not been written.