Joins at Scale

All the measurements of the previous lesson were made on a single data set: 2200 rows were partitioned, repartitioned, and the shuffled rows were counted. Most distributed jobs, however, bring together not one set but two: a large fact side and a dimension side attached to it. For a join to work, the two rows that are to match must sit on the same worker, otherwise they cannot meet.

There are two ways to ensure this, and their costs are of different kinds. The small side can be copied to every worker; then no row is shuffled, but the small side is multiplied by the number of workers. Or both sides are shuffled by the join key; then there are no copies, but the rows of both sides cross the network. This lesson compares the two strategies as the dimension side grows and looks for the point where the winner changes.

• PF10. The source, the consumption rows, the dimension side and the cluster are fictional and are generated within the lesson. No distributed processing framework, query engine or file system is called. The seed is 20260218.
• PF11. The fact side is the 2200 consumption rows coming out of the pipeline, and it sits partitioned by the region key, in the most unbalanced layout of the previous lesson. The join key is subscriber. The distinction between the fact and dimension relations was established in the Relational Database Administration course of the Databases curriculum; it is not repeated.
• PF12. The dimension side is a subscriber index: keys starting from A000. Its size is swept; an index of 50 rows represents a subset of the subscribers, and an index of 4000 rows a register far wider than the pipeline sees.
• PF13. Broadcast join: the dimension side is copied to every worker. Shuffled rows 0, copies dimension x workers. The cost of a copy is one unit and is not multiplied by the network cost; this is a model choice and it directly determines the result of the lesson.
• PF14. Partitioned join: both sides are shuffled by the join key. Copies 0, and each shuffled row writes a cost of NETWORK_COST times.
• PF15. The dimension side enters the job on a single worker, unpartitioned. In the partitioned strategy, therefore, its rows that change workers are added to the shuffled rows.
• PF16. Makespan is the work of the busiest worker, total work is counted separately, and makespan is the model’s own unit. TASK_OVERHEAD = 50 is added equally to both strategies at the same number of workers, so it does not enter the comparison.
• PF17. The baseline is a single worker: makespan 2250, total work 2200.
• PF18. The two strategies match the same rows at every dimension size; the only thing measured is cost.

Two Strategies, Two Kinds of Cost

The block below partitions the fact side by region, generates the dimension side at six different sizes and writes the cost of the two strategies side by side at each size. The matched rows column is the same for both strategies and is printed once.

# FICTIONAL. Meter readings, consumption rows and the cluster are generated within the lesson;
# no distributed processing framework or file system is called. Seed 20260218.
SEED, M32 = 20260218, 0xFFFFFFFF


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


REGION = ["dogu", "bati", "kuzey", "guney", "merkez"]
TYPE = ["residential", "commercial"]
PERIODS = [f"2026-{a:02d}" for a in range(1, 13)]
WORKERS, NETWORK_COST, TASK_OVERHEAD = 4, 3, 50
SUBSCRIBERS = {}
for i in range(200):
    r = rng(SEED + 37 * i)
    SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)],
                                    "start": 10000 + int(r() * 60000)}


def rows():
    """Consumption rows coming out of the pipeline (the fact side). On a sound source the screen
    and dedupe stages drop no rows; consumption is found from the period difference."""
    raw = []
    for i, (sub, a) in enumerate(SUBSCRIBERS.items()):
        e = a["start"]
        for p in PERIODS:
            r = rng(SEED + 101 * i + 7 * PERIODS.index(p))
            e += 4 + int(r() * 46)
            raw.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"],
                        "index": e, "unit": "m3"})
    table, output = {(k["subscriber"], k["period"]): k for k in raw}, []
    for k in raw:
        i = PERIODS.index(k["period"])
        prev = table.get((k["subscriber"], PERIODS[i - 1])) if i else None
        if prev is not None:
            output.append(dict(k, m3=k["index"] - prev["index"]))
    return output


def trace_id(m):
    h = 2166136261
    for c in str(m):
        h = ((h ^ ord(c)) * 16777619) & 0xFFFFFFFF
    return h


def distribute(records, key, workers=WORKERS):
    """Assigns rows to workers by key. The same key always goes to the same worker."""
    buckets = {i: [] for i in range(workers)}
    for k in records:
        buckets[trace_id(k[key]) % workers].append(k)
    return buckets


def shuffle(buckets, new_key, workers=WORKERS):
    """Repartition. A row that changes workers crosses the network."""
    new, moved = {i: [] for i in range(workers)}, 0
    for old, records in buckets.items():
        for k in records:
            target = trace_id(k[new_key]) % workers
            if target != old:
                moved += 1
            new[target].append(k)
    return new, moved


def broadcast_join(buckets, dimension, workers=WORKERS):
    """The dimension is copied to every worker. No shuffle , but there are copies."""
    copies = len(dimension) * workers
    n = [len(v) for v in buckets.values()]
    return {"moved": 0, "copies": copies,
            "duration": max(n) + copies // workers + TASK_OVERHEAD,
            "total_work": sum(n) + copies}


def partitioned_join(buckets, dimension, key="subscriber", workers=WORKERS):
    """Both sides are shuffled by key."""
    _, moved = shuffle(buckets, key, workers)
    dimension_moved = sum(1 for a in dimension if trace_id(a) % workers != 0)
    n = [len(v) for v in buckets.values()]
    t = moved + dimension_moved
    return {"moved": t, "copies": 0,
            "duration": max(n) + t * NETWORK_COST // workers + TASK_OVERHEAD,
            "total_work": sum(n) + t * NETWORK_COST}


def generate_dimension(n):
    """Subscriber index: n keys. Sits on a single worker , unpartitioned."""
    return [f"A{i:03d}" for i in range(n)]


ALL = rows()
BUCKETS = distribute(ALL, "region")
print(f"fact side {len(ALL)} rows , partitioned by the region key ,"
      f" worker load {[len(v) for v in BUCKETS.values()]}")
print(f"single-worker baseline: makespan {len(ALL) + TASK_OVERHEAD} , total work {len(ALL)}")
print(f"join key subscriber , workers {WORKERS} , network cost {NETWORK_COST}")
print("\n size  matched copies  bc.dur  bc.total     moved  pt.dur  pt.total   winner")
for n in (50, 200, 400, 800, 1600, 4000):
    d = generate_dimension(n)
    b, p = broadcast_join(BUCKETS, d), partitioned_join(BUCKETS, d)
    m = sum(1 for k in ALL if k["subscriber"] in set(d))
    print(f"{n:>5} {m:>8} {b['copies']:>6} {b['duration']:>7} {b['total_work']:>9}"
          f" {p['moved']:>9} {p['duration']:>7} {p['total_work']:>9}"
          f"   {'broadcast' if b['duration'] < p['duration'] else 'partitioned'}")

fact side 2200 rows , partitioned by the region key , worker load [374, 0, 0, 1826]
single-worker baseline: makespan 2250 , total work 2200
join key subscriber , workers 4 , network cost 3

 size  matched copies  bc.dur  bc.total     moved  pt.dur  pt.total   winner
   50      539    200    1926      2400      1775    3207      7525   broadcast
  200     2189    800    2076      3000      1888    3292      7864   broadcast
  400     2200   1600    2276      3800      2038    3404      8314   broadcast
  800     2200   3200    2676      5400      2338    3629      9214   broadcast
 1600     2200   6400    3476      8600      2938    4079     11014   broadcast
 4000     2200  16000    5876     18200      4738    5429     16414   partitioned

The matched rows column is the precondition of the table. At the same dimension size the two strategies match the same rows: 539 in the index of 50 rows, 2189 in the one of 200, and 2200 after 400. The column grows with the size because the index covers more and more subscribers; after 400 rows it is fixed because all the subscribers are covered. What is measured is not correctness, because correctness is the same in both strategies.

The first row gives the most important property of the partitioned strategy: it has a floor. Even if the dimension is only 50 rows, the partitioned join has to move 1738 rows of the fact side, because the fact side is partitioned by region while the join is made over subscriber. The partitioned makespan therefore has a floor: 3179 even if the dimension fell to zero, and 3207 at 50 rows, the smallest size in the table. The broadcast strategy has no floor; as the dimension shrinks its cost approaches zero, and at 50 rows the makespan is only 1926. The broadcast join winning at a small dimension is not a tendency but a structural result.

The growth rate of the two columns decides the matter. In the broadcast strategy each row of the dimension is copied as many times as there are workers, and because the copies are divided among the workers, it adds one unit per row to the makespan: 1876 plus the dimension rows. In the partitioned strategy three quarters of the dimension changes workers, each shuffled row is multiplied by the network cost and divided among the workers, and what is added to the makespan is nine sixteenths of a unit per row. The broadcast line is steeper but starts from zero; the partitioned line grows more slowly but from a height of 1303 units. The two lines are bound to cross, and the place where they cross is the number this lesson is looking for.

The total work column tells the same story in another unit. At a dimension of 1600 rows the broadcast join does 8600 units of work and the partitioned join 11014; at 4000 the order reverses (18200 against 16414). The rule of the course says that makespan and total work can go in different directions; in this comparison they do not, because the makespan of each strategy is its total work divided by the number of workers, and the divisor is the same in both. The meaning of the rule here is this: that the two agree must also be shown, otherwise it is not known which one was measured.

The Turning Point

Six sizes say that the turn lies between 1600 and 4000, but they do not say where. The block below sweeps the dimension size one by one. To make the sweep fast, the shuffled rows of the dimension side are kept as a prefix sum; the first line verifies that this shortcut gives the same result as the functions in the first block.

# This block continues the definitions of trace_id(), distribute(), shuffle(), broadcast_join(),
# partitioned_join(), generate_dimension(), ALL and BUCKETS from the first block.
PREFIX = [0]                     # PREFIX[n]: how many of the first n dimension rows change workers
for a in generate_dimension(4000):
    PREFIX.append(PREFIX[-1] + (1 if trace_id(a) % WORKERS else 0))
_, FACT_MOVED = shuffle(BUCKETS, "subscriber")
LOAD = [len(v) for v in BUCKETS.values()]


def measure(n, network=NETWORK_COST):
    """The same computation as the two functions in the first block; the only differences are that
    the network cost is passed in from outside and the dimension side is sped up with a prefix sum.
    Returns: broadcast makespan , partitioned makespan , broadcast total work , partitioned total work."""
    t = FACT_MOVED + PREFIX[n]
    return (max(LOAD) + n + TASK_OVERHEAD, max(LOAD) + t * network // WORKERS + TASK_OVERHEAD,
            sum(LOAD) + n * WORKERS, sum(LOAD) + t * network)


d = generate_dimension(4000)
print(f"check , dimension 4000: with the functions {broadcast_join(BUCKETS, d)['duration']}"
      f"/{partitioned_join(BUCKETS, d)['duration']} , with the prefix sum"
      f" {measure(4000)[0]}/{measure(4000)[1]}")
print("\nboth sides of the turning point at network cost 3")
for n in (2978, 2979, 2980, 2981):
    b, p, _, _ = measure(n)
    print(f"dimension {n}   broadcast {b}   partitioned {p}   "
          f"{'broadcast' if b < p else 'partitioned' if p < b else 'equal'}")
first_total = next(n for n in range(1, 4001) if measure(n)[3] < measure(n)[2])
print(f"\nfirst dimension size at which the partitioned join pulls ahead on total work: {first_total}")
print("\nnetwork cost   first dimension size at which the partitioned join pulls ahead on makespan (up to 4000)")
for network in (1, 3, 6, 12):
    first = next((n for n in range(1, 4001) if measure(n, network)[1] < measure(n, network)[0]), None)
    print(f"{network:>12}   {first if first else 'none'}")

check , dimension 4000: with the functions 5876/5429 , with the prefix sum 5876/5429

both sides of the turning point at network cost 3
dimension 2978   broadcast 4854   partitioned 4854   equal
dimension 2979   broadcast 4855   partitioned 4855   equal
dimension 2980   broadcast 4856   partitioned 4855   partitioned
dimension 2981   broadcast 4857   partitioned 4856   partitioned

first dimension size at which the partitioned join pulls ahead on total work: 2980

network cost   first dimension size at which the partitioned join pulls ahead on makespan (up to 4000)
           1   535
           3   2980
           6   none
          12   none

The crossover point is 2980. At 2978 and 2979 the two strategies are equal, and at 2980 the partitioned join pulls ahead by one unit. That the crossing is not sharp comes from the integer division of the model: the broadcast makespan grows by a whole unit per row and the partitioned makespan by nine sixteenths of a unit, so the two lines meet at the same integer for a few rows. The practical reading of the decision is not “the strategy changes as the dimension approaches three thousand” but “around three thousand the difference between the two strategies becomes unmeasurable”; the only places where the difference is meaningful are the extremes. At 50 rows the broadcast join is ahead by 1281 units, at 4000 rows the partitioned join by 447.

The total work row confirms that the turn is at the same place: 2980. This is no coincidence. The total work of the broadcast strategy grows by four units per dimension row and its makespan by one; the total work of the partitioned strategy grows by 2.25 units per row and its makespan by one quarter of that. The ratio of the two measures is the number of workers in both, so their orderings turn at the same point. This is not an exception to the rule of the course but proof of why the rule had to be written: that they agree is known only when both are measured.

The last table shows what the decision hangs on, and it is the harshest finding of this lesson. When the network cost is 1, the crossover point drops to 535; a decision made for an index of 2980 rows reverses, with the same data and the same number of workers, only because one constant changed. When the network cost is 6 or 12, the partitioned join cannot pull ahead at any of the swept dimension sizes. The reason can be read off: the partitioned makespan grows by three sixteenths of the network cost per dimension row, the broadcast makespan by one unit; as long as the network cost stays at five, the partitioned join grows more slowly, and when it rises to six it grows faster and can never catch up again. So the sentence “at a large dimension the partitioned join wins” is true for a network cost of 3 and false for 6.

The rule that follows is this: a join strategy cannot be chosen by looking at the data. Both sides match the same rows, give the same result, and the correctness check passes both. The choice is made by looking at the size of the dimension side, at the key by which the fact side is partitioned, and at the network cost; a decision given without writing down one of the three inputs counts as unmeasured.

Summary

• The broadcast join copies the dimension side to every worker (shuffled 0), and the partitioned join shuffles both sides (copies 0). The costs are of different kinds: one is copies, the other shuffled rows.
• The partitioned strategy has a floor: because the fact side is partitioned by region while the join is made over subscriber, 1738 rows are shuffled under every condition and the makespan does not fall below 3179; at a dimension of 50 rows it is 3207. The broadcast join has no floor; at a dimension of 50 rows the makespan is 1926.
• The broadcast makespan grows by one unit per dimension row, the partitioned makespan by nine sixteenths of a unit. The two lines cross: the makespans are equal at 2978 and 2979, and the partitioned join pulls ahead at 2980. At 4000 the gap is 5876 against 5429.
• When total work is measured, the crossover point again comes out as 2980, because the ratio of the two measures is the number of workers in both strategies. That they agree is known only when both are measured.
• The decision hangs on the network cost: at a cost of 1 the crossover point is 535, at 3 it is 2980, and at 6 and 12 the partitioned join cannot pull ahead at any size. Because both strategies match the same rows at every size, the correctness check shows none of this difference.

Next Step

In all the tables of this lesson a number stood there silently: the worker load of the fact side had been printed as [374, 0, 0, 1826]. Two of the four workers had received no rows at all, and all the makespan calculations had been made over the 1826 rows of the busiest worker. The floor of 1876 units of the broadcast strategy and the floor of 3207 of the partitioned strategy were both fed by this one number. For two lessons the source of the slowness was sought in movement; yet in this layout half of the workers were not working at all, and that is a loss independent of movement. The next lesson looks at the partitioning itself: it measures how unbalanced the spread of a key across the workers is as skew, partitions the same job by the right and the wrong key, and counts what adding workers does for unbalanced partitioning.