Forcing and reading a `SortMergeJoin`

Description

Lesson 3 confirmed that Spark, left to its default criterion, chooses BroadcastHashJoin for the match between fact_orders_at_scale and dim_store — the correct decision, given that dim_store.csv weighs barely 79 bytes. This lesson does something that's never a good idea in a real pipeline, but is exactly what's needed to see the other side of the coin with evidence: it disables the threshold (spark.sql.autoBroadcastJoinThreshold = -1), on the same JOIN, and reads the resulting physical plan. The plan's shape changes — SortMergeJoin shows up, with Exchange on both sides — and comparing the two plans side by side is the most direct piece of evidence in this whole module about what a broadcast join really does for you.

Connection to the module. This lesson closes the arc from lessons 2 through 4: the criterion (lesson 2), the default behavior (lesson 3), and now the forced behavior, on the same JOIN, with the same data. Starting in lesson 5, the module changes topic entirely — window functions — so this is the last piece of the joins half.

An analogy: the explicit order "don't photocopy anything"

Go back to lessons 2 and 3's analogy. The logistics manager, seeing that the customer directory is easily small enough to photocopy, lets every truck get its copy — the efficient decision, made automatically. This lesson is the equivalent of that same manager giving an explicit, deliberately suboptimal order: "don't photocopy the directory, no matter how small it is — every truck stops and reorganizes its cargo, as if the directory weighed tons." Nobody would give that order in a real operation — it would be unnecessary work — but giving it to yourself, in a controlled experiment, is the only way to see with your own eyes what would have happened if the directory had really been too large to photocopy.

Worked example: the same join, with the threshold switched off

Step 1 — Pick back up fact_orders_at_scale and dim_store

# sortmerge_forced.py
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import (
    StructType, StructField, StringType, IntegerType, DoubleType, TimestampType,
)

spark = SparkSession.builder.appName("kiosko-spark").master("local[*]").getOrCreate()

scale_schema = StructType([
    StructField("order_id", StringType(), False),
    StructField("franchise_id", IntegerType(), False),
    StructField("store_id", StringType(), False),
    StructField("product_id", StringType(), False),
    StructField("quantity", IntegerType(), False),
    StructField("unit_price", DoubleType(), False),
    StructField("order_ts", TimestampType(), False),
])
dim_store_schema = StructType([
    StructField("store_id", StringType(), False),
    StructField("store_name", StringType(), False),
    StructField("city", StringType(), False),
])

orders_at_scale_df = spark.read.csv(
    "kiosko_orders_at_scale.csv", schema=scale_schema, header=True, enforceSchema=False,
)
dim_store_df = spark.read.csv("dim_store.csv", schema=dim_store_schema, header=True, enforceSchema=False)

fact_orders_at_scale_df = orders_at_scale_df.withColumn(
    "revenue", F.col("quantity") * F.col("unit_price")
)

Step 2 — Disable the threshold, and run the same JOIN

print(f"BEFORE -- spark.sql.autoBroadcastJoinThreshold = {spark.conf.get('spark.sql.autoBroadcastJoinThreshold')}")

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
print(f"AFTER -- spark.sql.autoBroadcastJoinThreshold = {spark.conf.get('spark.sql.autoBroadcastJoinThreshold')}")

joined_forced = fact_orders_at_scale_df.join(dim_store_df, "store_id")

print("\n=== physical plan: fact_orders_at_scale join dim_store, broadcast disabled ===")
joined_forced.explain()

count_after = joined_forced.count()
print(f"\ncount after the join (forced SortMergeJoin) = {count_after}")
assert count_after == 10_000_000
print("Verification: forced SortMergeJoin doesn't lose or duplicate rows -> OK")

spark.stop()

What to expect. Running python3 sortmerge_forced.py, the output is exactly this (executed in this run, PySpark 4.2.0):

BEFORE -- spark.sql.autoBroadcastJoinThreshold = 10485760b
AFTER -- spark.sql.autoBroadcastJoinThreshold = -1

=== physical plan: fact_orders_at_scale join dim_store, broadcast disabled ===
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [store_id#2, order_id#0, franchise_id#1, product_id#3, quantity#4, unit_price#5, order_ts#6, revenue#11, store_name#8, city#9]
   +- SortMergeJoin [store_id#2], [store_id#7], Inner
      :- Sort [store_id#2 ASC NULLS FIRST], false, 0
      :  +- Exchange hashpartitioning(store_id#2, 200), ENSURE_REQUIREMENTS, [plan_id=27]
      :     +- Project [order_id#0, franchise_id#1, store_id#2, product_id#3, quantity#4, unit_price#5, order_ts#6, (cast(quantity#4 as double) * unit_price#5) AS revenue#11]
      :        +- Filter isnotnull(store_id#2)
      :           +- FileScan csv [order_id#0,franchise_id#1,store_id#2,product_id#3,quantity#4,unit_price#5,order_ts#6] Batched: false, DataFilters: [isnotnull(store_id#2)], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/.../kiosko_orders_at_scale.csv], PartitionFilters: [], PushedFilters: [IsNotNull(store_id)], ReadSchema: struct<order_id:string,franchise_id:int,store_id:string,product_id:string,quantity:int,unit_price...
      +- Sort [store_id#7 ASC NULLS FIRST], false, 0
         +- Exchange hashpartitioning(store_id#7, 200), ENSURE_REQUIREMENTS, [plan_id=28]
            +- Filter isnotnull(store_id#7)
               +- FileScan csv [store_id#7,store_name#8,city#9] Batched: false, DataFilters: [isnotnull(store_id#7)], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/.../dim_store.csv], PartitionFilters: [], PushedFilters: [IsNotNull(store_id)], ReadSchema: struct<store_id:string,store_name:string,city:string>

count after the join (forced SortMergeJoin) = 10000000
Verification: forced SortMergeJoin doesn't lose or duplicate rows -> OK

Now, yes — two Exchange hashpartitioning(store_id, 200) nodes — one for each side of the JOIN, including dim_store_df, which still has only three rows. Spark reorganizes even the tiny table, by the same store_id key, to guarantee rows with the same value end up in the same partition on both sides — the mechanism that makes a SortMergeJoin possible. Also notice the Sort node, above each Exchange: a SortMergeJoin, as its name suggests, needs both sides sorted by the JOIN key before it can combine them efficiently, a step a BroadcastHashJoin never needs.

Diagram: the two plans, side by side

flowchart TD
    subgraph Default["By default -- lesson 3"]
        direction TB
        A1["fact_orders_at_scale\n10M rows, 12 partitions\n(not reorganized)"] --> A2["BroadcastHashJoin"]
        A3["dim_store\ncopied whole\n3 rows"] -->|"BroadcastExchange"| A2
        A2 --> A4["joined_store\n10M rows"]
    end

    subgraph Forced["Threshold disabled -- this lesson"]
        direction TB
        B1["fact_orders_at_scale\n10M rows"] -->|"Exchange hashpartitioning\n(store_id, 200)"| B2["Sort by store_id"]
        B3["dim_store\n3 rows"] -->|"Exchange hashpartitioning\n(store_id, 200)"| B4["Sort by store_id"]
        B2 --> B5["SortMergeJoin"]
        B4 --> B5
        B5 --> B6["joined_forced\n10M rows"]
    end

    style A2 fill:#9c6,stroke:#333
    style B5 fill:#f96,stroke:#333

Both plans produce exactly the same result — ten million rows, with the same values — and that's precisely the most important part of this comparison: the result's correctness doesn't depend on which strategy got chosen. The only thing that changes is how Spark gets there, and that "how" does carry a measurable cost: the forced plan moves the complete dim_store through an Exchange — three rows, an insignificant cost in this particular case — but, more importantly, it also reorganizes the ten-million-row side by the store_id key, something lesson 3's BroadcastHashJoin never needed to do.

Going deeper: why the result is identical, even though the path is different

It's worth confirming, not just assuming, that both plans produce exactly the same data — not just the same row count. BroadcastHashJoin and SortMergeJoin are two different algorithms for solving the same logical problem: finding, for every row on the large side, the row (or rows) on the small side that share the same join-key value. The first solves this with a direct lookup in a hash table every partition already has a copy of; the second solves this by sorting both sides by the key and walking through them in parallel, like someone comparing two already-alphabetized lists to find matches. Neither algorithm changes the JOIN's meaning — an INNER JOIN is still an INNER JOIN, with the same rules for which rows match — only the physical mechanism Spark uses to execute it changes.

This is, essentially, the same separation Catalyst makes throughout its entire design: the logical plan describes what gets asked for (an INNER JOIN between fact_orders_at_scale and dim_store by store_id), and the physical plan describes how that request is going to get executed (with which algorithm, in what order). The logical plan for both this lesson's experiment and the previous one's is identical; only the physical plan changed, because you, on purpose, took away Catalyst's option to choose the more efficient strategy. Module 6 develops this distinction between logical and physical plans in much more depth, showing Catalyst's complete phases.

Common mistakes

Forgetting to restore spark.sql.autoBroadcastJoinThreshold after this experiment. What happens: someone runs this lesson's code inside a SparkSession they keep using for other work afterward, and is surprised when a completely unrelated JOIN, later in the same script, also starts using SortMergeJoin for no apparent reason. Why it happens: spark.conf.set() changes the setting for the rest of that SparkSession's life, not just for the immediately following query — an easy detail to forget after a one-off experiment. How to spot it: if a JOIN that "should" be a broadcast (a clearly small table) shows up with SortMergeJoin in a later plan in the same script, check whether an earlier experiment left spark.sql.autoBroadcastJoinThreshold at -1 without restoring it. How to fix it: in a real script, explicitly restore the setting after an experiment like this one (spark.conf.set("spark.sql.autoBroadcastJoinThreshold", 10 * 1024 * 1024), the default value), or use JOIN hints (.hint("broadcast")) to force the strategy for a single query without touching the session's global configuration.

Thinking forcing SortMergeJoin over such small data reveals some kind of Spark bug. What happens: someone sees this lesson's forced plan — with Exchange hashpartitioning moving a three-row table — and concludes this proves Spark "isn't that smart" when forced into a suboptimal strategy. Why it happens: watching a system deliberately do something inefficient can feel like a system failure, instead of a user-controlled experiment. How to spot it: if your takeaway from this lesson is about "Spark's limits" instead of "what lesson 2's criterion decides when you deliberately disable it," you're missing the central pedagogical point. How to fix it: Spark did exactly what it was asked to do — broadcast join was explicitly disabled with spark.sql.autoBroadcastJoinThreshold = -1, so SortMergeJoin is the only remaining option, no matter how small the table is. It isn't an optimizer failure; it's the deliberate absence of the more efficient option.

Confusing the Sort in SortMergeJoin's plan with the .orderBy() you already saw in module 4. What happens: someone, seeing the Sort [store_id#2 ASC NULLS FIRST] node in this lesson's plan, assumes joined_forced's final result comes sorted by store_id, and relies on that order without an explicit .orderBy(). Why it happens: the word "Sort" in the plan seems to promise a sorted result, just like the .orderBy() you already know. How to spot it: if your code depends on joined_forced's rows coming out in a specific order without an explicit .orderBy() at the end, you risk that order changing between runs or Spark versions. How to fix it: the Sort inside a SortMergeJoin's plan is an internal step of the algorithm — necessary for the JOIN to work, not a guarantee about the final result's order; any later operation (another JOIN, a groupBy, even the final Project) can reorganize the rows again. If you need a guaranteed order in the result, always use an explicit .orderBy() — never infer the order from an internal Sort in the plan.

Exercises

Exercise 1 — Confirm the per-store count is still correct with a forced SortMergeJoin. Using joined_forced from the worked example, group by store_id and confirm with assert that the per-store row count is identical to what you already verified with BroadcastHashJoin in lesson 3.

per_store_forced = joined_forced.groupBy("store_id").count().orderBy("store_id")
per_store_forced.show()

counts = {r["store_id"]: r["count"] for r in per_store_forced.collect()}
assert counts == {"S01": 4_000_000, "S02": 3_250_000, "S03": 2_750_000}
print("Verification: SortMergeJoin and BroadcastHashJoin produce exactly the same per-store breakdown -> OK")

+--------+-------+
|store_id|  count|
+--------+-------+
|     S01|4000000|
|     S02|3250000|
|     S03|2750000|
+--------+-------+

Verification: SortMergeJoin and BroadcastHashJoin produce exactly the same per-store breakdown -> OK

Exercise 2 — Force SortMergeJoin over the chained join (against dim_store and dim_product), and count how many Exchange nodes show up in total. Extend this lesson's experiment to the double JOIN (dim_store and dim_product), and count how many distinct Exchange hashpartitioning nodes show up in the complete plan.

dim_product_schema = StructType([
    StructField("product_id", StringType(), False),
    StructField("product_name", StringType(), False),
    StructField("category", StringType(), False),
    StructField("unit_cost", DoubleType(), False),
])
dim_product_df = spark.read.csv("dim_product.csv", schema=dim_product_schema, header=True, enforceSchema=False)

joined_double_forced = fact_orders_at_scale_df.join(dim_store_df, "store_id").join(dim_product_df, "product_id")
joined_double_forced.explain()
count_double = joined_double_forced.count()
assert count_double == 10_000_000
print(f"count = {count_double}")

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- Project [...]
   +- SortMergeJoin [product_id#3], [product_id#N], Inner
      :- Sort [...]
      :  +- Exchange hashpartitioning(product_id#3, 200), ...   -- 1
      :     +- SortMergeJoin [store_id#2], [store_id#M], Inner
      :        :- Sort [...]
      :        :  +- Exchange hashpartitioning(store_id#2, 200), ...   -- 2
      :        +- Sort [...]
      :           +- Exchange hashpartitioning(store_id#M, 200), ...   -- 3
      +- Sort [...]
         +- Exchange hashpartitioning(product_id#N, 200), ...   -- 4

count = 10000000

Exercise 3 — Explain, without code, why forcing SortMergeJoin over dim_store still produces a correct result, despite being an inefficient decision. In 2-3 sentences, explain why "inefficient" and "incorrect" are different things in this lesson's context.

Summary and next step

This lesson forced, with spark.sql.autoBroadcastJoinThreshold = -1, the same JOIN lesson 3 resolved with BroadcastHashJoin by default — and the plan changed completely: SortMergeJoin, with Exchange hashpartitioning(store_id, 200) on both sides, including the three-row table. The result — ten million rows, same per-store breakdown — was identical in both experiments, confirming the strategy choice affects performance, never correctness.

Before moving on you should be able to: force and restore spark.sql.autoBroadcastJoinThreshold with confidence; compare two .explain() plans and name exactly what changed between them; and explain why a SortMergeJoin's internal Sort doesn't guarantee the final result's order.

With lessons 2 through 4 closed, this module changes topic. Lesson 5 introduces Window.partitionBy().orderBy() from scratch, over Kiosko's forty real rows, before applying it to lessons 6 and 7's two business questions.

Resources

• Apache Spark — SQL Performance Tuning, spark.sql.autoBroadcastJoinThreshold section and JOIN strategies (SortMergeJoin as the strategy used when broadcast doesn't apply or is disabled). spark.apache.org/docs/latest/sql-performance-tuning.html.
• Apache Spark — RDD Programming Guide, "Shuffle operations" section (the definition of shuffle, already cited in module 4, explaining why this lesson's SortMergeJoin pays the cost of an Exchange). spark.apache.org/docs/latest/rdd-programming-guide.html.
• This guide's DESIGN doc (spark-and-distributed-processing-guide/DISENO.md) — this lesson's exact specification: lesson 3's same JOIN, forced to SortMergeJoin for comparison.