Apple Data Engineer Interview Preparation Guide - Entry Level

Direct answer

Compute a deterministic running total (tie-broken order, since created_at alone can repeat), then locate the row where it first exceeds the threshold using MIN(CASE WHEN running_total > threshold THEN rn END) OVER (PARTITION BY user_id), and mark every row from that point onward as flagged by comparing each row's own tie-broken position to that stored first-crossing marker. The key design choice is comparing against the first-crossing marker, not re-checking running_total > threshold on every row directly: once a row is marked, it stays marked even if a later refund or a NULL amount temporarily pulls the running total back under the threshold, which is exactly what "don't flicker" requires.

Structured elaboration

• Deterministic ordering: ORDER BY created_at, order_id inside every window clause, so ties on created_at resolve the same way every time the query runs; without a tie-breaker, two rows with an identical timestamp have an engine-dependent relative order, and the running total (and therefore the flag) is not reproducible.
• NULL amounts: SUM() (and MIN(), used below) already skip NULL values on their own, so a NULL amount does not poison the running total even without an explicit COALESCE. COALESCE(amount, 0) is still worth writing, not because the math needs it, but because it makes the intended treatment of missing amounts explicit in the query rather than relying on an aggregate's default behavior that a future reader might not know about.
• First-crossing marker: MIN(CASE WHEN running_total > 10000 THEN rn END) OVER (PARTITION BY user_id) relies on the same NULL-skipping behavior; the CASE produces NULL for every row that has not yet crossed the threshold, and MIN ignores those NULLs, landing on the smallest rn where the condition was true.
• Sticky flag: rn >= first_exceed_rn stays TRUE for every row from the first crossing onward, regardless of what the running total does afterward. This is what prevents flicker: the flag is locked onto a position, not re-derived from a live comparison each row.

Worked example

WITH prep AS (
  SELECT order_id, user_id, COALESCE(amount, 0) AS amount, created_at
  FROM orders
),
ordered AS (
  SELECT order_id, user_id, amount, created_at,
    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at, order_id) AS rn,
    SUM(amount) OVER (PARTITION BY user_id ORDER BY created_at, order_id
                       ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS running_total
  FROM prep
),
flagged AS (
  SELECT *, MIN(CASE WHEN running_total > 10000 THEN rn END) OVER (PARTITION BY user_id) AS first_exceed_rn
  FROM ordered
)
SELECT order_id, created_at, amount, running_total,
  CASE WHEN first_exceed_rn IS NOT NULL AND rn >= first_exceed_rn THEN TRUE ELSE FALSE END AS threshold_reached
FROM flagged ORDER BY order_id;

Executed against 5 orders for one user, including a NULL amount and a same-timestamp tie: order 1 (4000, running total 4000), orders 2 and 3 tied on the same created_at (3000 then 2500, tie-broken by order_id, running totals 7000 then 9500), order 4 (NULL amount, running total unchanged at 9500), order 5 (1000, running total 10500). The flag is false through order 4 and flips to true at order 5, the first row where the running total exceeds 10000.

Trade-offs & pitfalls

• Skipping the tie-breaker on created_at does not just risk a wrong-looking output once; it makes the running total (and therefore exactly which row the flag flips on) non-deterministic across re-runs, which is a correctness bug even when no individual run looks obviously broken.
• Relying on SUM and MIN silently skipping NULL is standard SQL behavior, not an engine-specific quirk, but it is worth stating explicitly in code comments so the next reader does not assume NULL amounts are being dropped from the row set entirely (they are not; only from the aggregate math).
• If the actual requirement is a "reactive" flag that un-sets when the running total dips back under the threshold (say, after a large refund), compare running_total > threshold directly per row instead of locking onto the first-crossing marker; that alternative is simpler but is exactly what reintroduces flicker if amounts can go negative.
• An index on (user_id, created_at, order_id) keeps the window functions' per-partition sort and scan cheap on large tables; without it, every partition's ordering work falls back to an in-memory sort.

Direct answer

Label noise investigation breaks cleanly into three sequential questions, and each one uses a different technique: how much noise is there overall (a measurement problem), which specific examples are most likely wrong (a ranking problem, so a limited review budget goes to the highest-value examples first), and, given that budget, whether the right response is relabeling, filtering, or changing the loss function entirely. Trying to answer all three with one tool is the usual mistake; a global noise-rate estimate and a per-example priority ranking are related but different outputs, and conflating them either wastes review budget on low-value examples or understates how bad the problem really is.

Structured elaboration

(a) Quantifying the overall rate of label noise. Train with k-fold cross-validation and record each example's prediction from the fold where it was held out (its "out-of-fold" prediction, meaning the model never saw that example during the training that produced this specific prediction, so the prediction is not contaminated by having memorized a possibly-wrong label). Compute the disagreement rate between the given label and the out-of-fold predicted label as a first estimate. This raw disagreement rate is a biased estimate on its own, since it conflates genuine label errors with ordinary model error (the model can just be wrong about a CORRECTLY labeled example), so calibrate it against a small, human-audited gold sample: have annotators re-review a random sample of, say, a few hundred examples, and use the true error rate on that gold sample to correct the full-dataset estimate. Tooling: any standard k-fold cross-validation setup, plus a lightweight annotation tool or spreadsheet for the gold-sample audit; no specialized infrastructure is required for this step.

(b) Identifying which specific examples are likely mislabeled. This is a ranking problem, not a classification problem: you want a score per example that a limited review budget can be sorted by. Three quantitative signals feed this ranking, in increasing order of cost: per-example training LOSS (an example the model fits poorly relative to similar examples is a cheap, single-model first pass, but loss alone conflates hard-but-correctly-labeled examples with mislabeled ones); per-example predicted CONFIDENCE on the given label (a low confidence on the currently-assigned class is a second cheap signal); and cross-model DISAGREEMENT from an ensemble of models with some diversity (different random seeds, different architectures, or different training folds), which is more expensive to produce but far more specific. Combine confidence and disagreement into a single score. State the rule before writing it down, because the shape of it is the actual idea: score each example as its average model confidence, multiplied by (1 minus the ensemble's internal disagreement), gated to zero whenever the ensemble's consensus class matches the label already on the example. Disagreement here is the spread of the K models' votes measured as a normalized entropy: 0 when all K models vote the same class, rising toward 1 as the votes spread evenly across classes. The (1 minus disagreement) factor is what stops the score rewarding confusion: an ensemble that is internally split is showing you an ambiguous or genuinely hard example, which is not the same finding as a mislabeled one, so a split vote is discounted even when the individual models are confident. Concretely, so the number can be traced: with K = 5 models over C = 3 classes, a 4-to-1 vote split gives vote proportions of 0.8 and 0.2, and the normalized entropy is $-(0.8\ln 0.8 + 0.2\ln 0.2)/\ln 3 = 0.500/1.099 = 0.455$, while a unanimous 5-0 vote gives exactly 0. The normalizer is the log of the number of distinct outcomes the votes could spread across, which is K when there are fewer models than classes and C otherwise, so the measure stays on a 0-to-1 scale in both regimes. So the examples most likely to be labeling errors are the ones where the ensemble confidently and consistently predicts a DIFFERENT class than the one the dataset currently assigns; confident agreement that matches the given label is not interesting (nothing to flag), and confident disagreement among the models themselves is more often a genuinely hard or ambiguous example than a labeling error specifically. At large scale (millions of examples), running a full multi-model ensemble over every single example is often too expensive to justify; a practical two-stage pattern is to use the cheap per-example loss or confidence signal as a first-pass filter to narrow the full dataset down to a much smaller candidate set, and reserve the more expensive ensemble-disagreement scoring for ranking WITHIN that narrowed set, before it ever reaches a human reviewer. Tooling: the loss and confidence signals come for free from any existing training run; the ensemble requires whatever infrastructure already exists to produce K model variants, and the scoring logic itself is lightweight, demonstrated below.

(c) Deciding, given a limited human-review budget, whether to relabel, filter, or switch to a noise-robust loss. This decision depends on the estimated noise RATE from step (a), the estimated noise PATTERN (random versus systematic, discovered by checking whether the ranked candidates from step (b) cluster in specific classes, sources, or annotators rather than spreading evenly), and the review budget itself relative to dataset size.

• Relabel: the right choice when noise is concentrated (a specific class, source, or annotator is responsible for most of it) and the review budget can plausibly cover the ranked candidate list from step (b) for that concentration. Concentrated noise with a bounded high-value candidate set is exactly what makes relabeling worth the human cost.
• Filter (remove flagged examples rather than relabeling them): appropriate when the noise rate is low enough that removing suspected-bad examples does not meaningfully shrink the useful training set, and when getting a correct replacement label is expensive, slow, or infeasible (for example, the ground truth requires a specialist annotator or a delayed real-world outcome that has not happened yet). Filtering is cheaper than relabeling in review time but destroys information, so it trades data efficiency for speed.
• Switch to a noise-robust loss function (a loss designed to be less sensitive to a fraction of incorrectly-labeled examples, such as a loss that down-weights high-loss, likely-mislabeled examples during training rather than requiring them to be identified and handled up front). The named options worth knowing are generalized cross-entropy, which interpolates between ordinary cross-entropy and mean absolute error through a single exponent parameter and so keeps cross-entropy's fast early learning while bounding how much any one badly-fit example can contribute; symmetric cross-entropy, which adds a reverse term measuring the label's divergence from the prediction, deliberately bounded so a wrong label cannot dominate the gradient; and a plain MAE-style loss, which is provably robust to symmetric label noise but trains slowly enough that it is rarely the practical choice on its own. This branch is: appropriate when the estimated noise rate is high, diffuse (spread broadly rather than concentrated, so no small ranked candidate list would cover most of it), or the review budget is simply too small relative to the dataset to make either relabeling or filtering practical at meaningful scale. This is the right default when steps (a) and (b) together tell you that manual review, however prioritized, cannot realistically move the needle.

Turning step (a)'s number into step (c)'s decision, with the metrics named. The three inputs are the calibrated noise rate from step (a), the concentration of step (b)'s ranked candidates (what share of the top-ranked candidates fall into a single class, data source, or annotator), and the review budget as a fraction of the flagged candidate set. Rough bands to argue from rather than thresholds to obey, since the right cut depends on dataset size and on how expensive a wrong label is downstream:

• Estimated noise rate under roughly 5 percent, and over half the top candidates concentrated in one class, source, or annotator: relabel that concentration. The candidate set is small and the fix is targeted.
• Estimated noise rate under roughly 5 percent but spread evenly across classes and sources, with labels expensive or slow to obtain: filter the flagged examples. You lose a few percent of the training set, which a large dataset absorbs.
• Estimated noise rate above roughly 15 to 20 percent, or a review budget covering less than about a tenth of the flagged candidates: a noise-robust loss is the primary response, with whatever review the budget allows spent on the top of the ranked list rather than spread thin.
  Between those bands, run the cheap experiment instead of arguing: train on the filtered set and on the full set with a noise-robust loss, and compare both against the current baseline on a small, human-verified clean evaluation set, which is the metric that actually decides this and is worth building before any of the three interventions.

In practice these are not mutually exclusive: a common pattern is to filter or relabel the highest-confidence, highest-value flagged examples within budget, and pair that with a moderately noise-robust loss for the remaining, unreviewed noise the budget could not reach.

Worked example

The script below implements the ranking logic from step (b): given each example's ensemble of per-model class-probability predictions and its currently-assigned label, it computes a combined disagreement-and-confidence score and ranks examples for review. One example is deliberately constructed as a "planted" confident, unanimous flip (all five models agree on a class different from the given label) to verify the score correctly surfaces it at the top of the review queue.

"""
Ranks labeled examples by a combined ensemble-disagreement + confidence
score, to prioritize a limited human-review budget toward examples most
likely to be mislabeled. In practice each row of `ensemble_probs` comes
from K independently-trained models scoring the same example; here it is
synthesized directly so the ranking logic is fully pinned and reproducible.

Pinned: numpy Generator seed=11, K=5 models, C=3 classes, 6 example rows
(one deliberately "planted" as a confident, unanimous flip vs its given
label to verify the score correctly surfaces it top-of-list).
Run with: python3 disagreement_score.py
"""
import numpy as np

rng = np.random.default_rng(seed=11)
K, C = 5, 3


def make_confident_row(true_class, agreement=1.0):
    """K models' softmax rows, `agreement` fraction voting `true_class`."""
    rows = []
    n_agree = round(K * agreement)
    for k in range(K):
        base = rng.dirichlet(alpha=[0.5] * C)
        cls = true_class if k < n_agree else int(rng.integers(0, C))
        base = np.clip(base, 0, None)
        base[cls] += 2.0  # push mass onto the chosen class
        base /= base.sum()
        rows.append(base)
    return np.stack(rows)


# 6 examples: given_label is what the dataset currently says
examples = [
    {"given_label": 0, "ensemble": make_confident_row(true_class=0, agreement=1.0)},  # models agree WITH label
    {"given_label": 1, "ensemble": make_confident_row(true_class=2, agreement=1.0)},  # PLANTED FLIP: unanimous, confident, disagrees with label
    {"given_label": 2, "ensemble": make_confident_row(true_class=2, agreement=0.6)},
    {"given_label": 0, "ensemble": make_confident_row(true_class=1, agreement=0.4)},  # split vote, moderate confidence
    {"given_label": 1, "ensemble": make_confident_row(true_class=1, agreement=1.0)},
    {"given_label": 2, "ensemble": make_confident_row(true_class=0, agreement=0.8)},
]


def score_example(given_label, ensemble_probs):
    preds = ensemble_probs.argmax(axis=1)
    confs = ensemble_probs.max(axis=1)
    vote_counts = np.bincount(preds, minlength=C) / K
    nonzero = vote_counts[vote_counts > 0]
    disagreement = max(0.0, -(nonzero * np.log(nonzero)).sum() / np.log(K if K < C else C))
    consensus_class = np.bincount(preds, minlength=C).argmax()
    agrees_with_label = consensus_class == given_label
    avg_conf = confs.mean()
    # Score rewards confident consensus that CONTRADICTS the given label.
    score = avg_conf * (1.0 - disagreement) * (0.0 if agrees_with_label else 1.0)
    return score, consensus_class, avg_conf, disagreement


results = []
for i, ex in enumerate(examples):
    score, consensus, avg_conf, disagreement = score_example(ex["given_label"], ex["ensemble"])
    results.append((i, ex["given_label"], consensus, avg_conf, disagreement, score))

results.sort(key=lambda r: r[-1], reverse=True)

print("=== label-noise review queue, ranked by score (desc) ===")
print(f"{'idx':>3} {'given':>5} {'consensus':>9} {'avg_conf':>8} {'disagree':>8} {'score':>7}")
for idx, given, consensus, avg_conf, disagreement, score in results:
    print(f"{idx:>3} {given:>5} {consensus:>9} {avg_conf:>8.3f} {disagreement:>8.3f} {score:>7.3f}")

Actual output from running this script:

=== label-noise review queue, ranked by score (desc) ===
idx given consensus avg_conf disagree   score
  1     1         2    0.842    0.000   0.842
  5     2         0    0.809    0.455   0.440
  3     0         1    0.702    0.455   0.382
  0     0         0    0.846    0.000   0.000
  2     2         2    0.707    0.000   0.000
  4     1         1    0.761    0.000   0.000

The planted flip, example index 1 (given label 1, but all five models unanimously and confidently vote class 2), correctly lands at the top of the queue with the highest score (0.842), exactly as intended. Example 5, where the models were split (80 percent agreement) but still confidently landed on class 0 against a given label of 2, ranks second. Examples 0, 2, and 4, where the ensemble's consensus class MATCHES the given label, correctly score zero regardless of the ensemble's confidence or internal disagreement, since a model bank that agrees with the current label, confidently or not, is not evidence of a labeling error at all; this is what "score rewards confident consensus that contradicts the given label" is doing mechanically, and the zero scores for those three rows confirm the score is not accidentally flagging ordinary model uncertainty as a labeling problem.

Trade-offs and pitfalls

The most common wrong turn is treating the ranking score from step (b) as itself an estimate of the OVERALL noise rate from step (a). A ranked list tells you relative priority among examples; it says nothing about what fraction of the full dataset is actually mislabeled without the separate, calibrated cross-validation-versus-gold-sample estimate, and reporting "we found 40 likely-mislabeled examples in review" as if that quantifies the dataset's noise rate, without knowing what fraction of the full dataset those 40 represent or how the review budget was allocated, overstates what was actually measured.

A second pitfall is using an under-diverse ensemble for the disagreement signal, for example five models trained with the same architecture, same random seed family, and same data split. If the ensemble members are correlated enough to make similar mistakes for similar reasons, the disagreement signal collapses even on genuinely ambiguous or mislabeled examples, understating how much review is actually warranted. Diversity in seeds, architecture, or training data subsets matters for this technique to work as intended.

A third pitfall in the decision step (c): choosing a noise-robust loss function as a substitute for ever looking at the data, rather than as a complement to a budget-limited review. A noise-robust loss reduces the HARM of remaining noise during training; it does not tell you anything about the underlying data quality, so systematic noise sources (a specific mislabeling process, a broken annotation guideline, a bad data source) can persist indefinitely and keep degrading every future model trained on the same data, even as the current model becomes more robust to their symptoms.

Direct answer

Sort in two phases. Phase 1, run generation: read the file in chunks that fit in memory (about 1.5 GiB of the 2 GiB), sort each chunk in memory, and write it back as a sorted run (a sorted piece of the data stored on disk). 500 GiB gives about 334 runs. Phase 2, merge: open all 334 runs at once with a modest read buffer each (about 5 MiB) and merge them with a min-heap into the final file in one pass. That is two passes over the data in total, each reading and writing all 500 GiB, so about 2,000 GiB of I/O (estimated: roughly 3 hours on a 200 MB/s disk, 36 minutes at 1 GB/s, 12 minutes at 3 GB/s). Write every run to a temporary name and rename it when complete, so a crash costs at most the run being written; delete inputs only after their output is finished.

The idea in plain words

A normal sort needs all the data in memory. The trick for data bigger than memory is that merging sorted lists needs only the front of each list in memory. So split the file into pieces that do fit, sort each piece, and then merge the sorted pieces with a heap that always yields the smallest front item. Terms: a run is a sorted piece of the data stored on disk. Fan-in is how many runs one merge reads at once. A pass is one full read and write of the data. Sequential I/O is reading or writing a file in order, which disks do far faster than jumping around (a hard disk spends about 10 ms per jump in the estimate below). Stable means records with equal keys keep their input order. A checksum is a short number computed from the bytes of a file, so that a changed file gives a different number. fsync is the operating-system call that forces data a program has written out of the OS cache and onto the disk, so that a power cut after the call cannot lose it.

The two-phase plan with a fan-in the memory allows is what an interview normally asks for; replacement selection, compression, time-based partitioning and restartable partitioned merges are refinements that matter mainly for roles that run large batch pipelines.

Brute force and why it is not enough

Loading 500 GiB into 2 GiB of RAM fails, and letting the operating system swap a full in-memory sort to disk makes the sort touch pages in random order, which disks handle badly (swap thrashing: the machine spends its time moving memory pages to and from disk instead of sorting). External merge sort avoids this by making every disk access large and sequential.

Run generation and run size

Read 1.5 GiB chunks (the other 0.25 GiB is an assumption for the operating system, the program and I/O buffers), sort each in memory, write each as a run. The number of runs is the file size divided by the run size, so bigger runs mean fewer runs, which means larger merge buffers and fewer passes. The cost model in section 6 of the output shows the trade-off (the 1 MiB minimum buffer is an assumption that keeps reads reasonably large):

The buffer column is 1.75 GiB divided by (runs + 1): 1.75 GiB / 335 = 5.35 MiB, where the +1 is the buffer that collects the merged output (dividing by 334 alone would give 5.37 MiB, also printed). The third row has a 0.90 MiB buffer, below the assumed 1 MiB minimum, so one merge of 2,000 runs is not used. The program prints the maximum fan-in at 1 MiB buffers (1,791 runs per merge): pass 1 merges 2,000 runs into 2 runs and pass 2 merges those 2 into 1, hence 2 merge passes.

Smaller runs are useful if you overlap work: while one chunk is being sorted on the CPU, the next is being read and the previous is being written, which needs memory for two or three chunks at once (double buffering: while one buffer is being processed, the next one is being filled). With two chunks in flight, the 0.75 GiB run size uses 1.5 GiB of the 1.75 GiB available and still gives a single merge pass; three chunks of 0.75 GiB would not fit, so a three-stage overlap needs smaller runs.

An alternative is replacement selection: keep a heap of M records, repeatedly output the smallest record that is not smaller than the last one output, and read a new record into the heap. A record read in that is smaller than the one just output cannot extend the current run, so it is tagged for the next run and waits. Section 7 prints this for a heap of 3 records and the input 4 7 5 2 8 1: the heap starts with 4, 5 and 7; after 4 leaves, the new record 2 is smaller than 4 and waits for run 1; after 5 leaves, the new record 8 is not smaller than 5, so it joins run 0; after 7 leaves, the new record 1 waits for run 1; then 8 leaves. Run 0 is 4 5 7 8 (four records from memory for three) and run 1 is 1 2. Chunk sorting the same input makes runs of exactly 3 records. Section 4 measures it on larger data: on random input runs average 1.96 times the memory size (51 runs instead of 100 for M = 1,000 and 100,000 records), and on input that is nearly sorted (timestamps that arrive almost in order) it produced a single run where chunk sorting makes 100. That helps when data arrives roughly in time order, as event logs do.

K-way merge

Open all runs, give each a read buffer, and keep a min-heap holding each run's current front record. Pop the smallest, write it to the output buffer, and refill from that run's buffer, which is the only time the disk is touched. Each record costs about log2(R) comparisons for R runs. Section 8 prints a merge of two runs, 1 4 9 and 2 3 10: before each output the heap holds the front record of every run, the smallest leaves, and the next record of that run takes its place. One merge pass turns R runs into ceil(R / F) runs for fan-in F (each group of F runs becomes one run), so the pass count is the number of times R must be divided by F, rounding up, to reach 1, which is ceil(log base F of R), and the whole sort takes 1 + that many passes. The program counts it with integer ceiling division rather than math.log, because floating-point log rounds some exact powers up: math.ceil(math.log(125, 5)) is 4, although 125 runs at fan-in 5 need 3 passes. The demo uses a fan-in of 8 with 20 runs: ceil(20 / 8) = 3 runs, then ceil(3 / 8) = 1 run, so it needs 2 merge passes (section 2), 3 passes overall, equal to the predicted ceil(log_8(20)) = 2 merge passes plus the run-generation pass.

Code: a working external sort at small scale

The program sorts 200,000 key,id lines, 10,000 per run (standing in for 1.5 GiB) with a fan-in of 8 (standing in for the buffer limit), injects a crash, restarts, and checks the output against a stable sorted() of the same file, including the order of tied keys. The checks and estimates are printed.

import heapq
import itertools
import math
import os
import random
import tempfile
import zlib

RUN_RECORDS = 10_000      # stands in for "as many records as fit in RAM"
FAN_IN = 8                # stands in for "as many runs as the merge buffers allow"


class SimulatedCrash(Exception):
    pass


def sort_key(line):
    return int(line.split(",", 1)[0])


def write_atomically(path, lines):
    """Write to a temp name, flush to disk, then rename. A reader never sees a half-written run."""
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        f.writelines(lines)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)


def generate_runs(input_path, workdir, crash_after=None):
    """Pass 0: sort RUN_RECORDS lines at a time. Existing finished runs are skipped on restart."""
    runs, made, skipped = [], 0, 0
    with open(input_path) as f:
        index = 0
        while chunk := list(itertools.islice(f, RUN_RECORDS)):
            path = os.path.join(workdir, f"run0_{index:04d}.txt")
            if os.path.exists(path):
                skipped += 1                      # finished before the crash
            else:
                if crash_after is not None and made == crash_after:
                    with open(path + ".tmp", "w") as partial:
                        partial.writelines(chunk[:100])   # a half-written file is left behind
                    raise SimulatedCrash(f"crashed while writing run {index}")
                write_atomically(path, sorted(chunk, key=sort_key))
                made += 1
            runs.append(path)
            index += 1
    return runs, made, skipped


def merge_runs(paths, out_path):
    files = [open(p) for p in paths]
    try:
        write_atomically(out_path, heapq.merge(*files, key=sort_key))
    finally:
        for f in files:
            f.close()


def passes_needed(runs, fan_in):
    """ceil(log_fan_in(runs)) by repeated integer ceiling division; math.log can round an exact power up."""
    count = 0
    while runs > 1:
        runs = -(-runs // fan_in)
        count += 1
    return count


def merge_passes(runs, workdir):
    passes, level = 0, 0
    while len(runs) > 1:
        level += 1
        merged = []
        for g in range(0, len(runs), FAN_IN):
            out = os.path.join(workdir, f"run{level}_{g // FAN_IN:04d}.txt")
            merge_runs(runs[g:g + FAN_IN], out)
            merged.append(out)
        for p in runs:                            # inputs are deleted only after the whole pass finished
            os.remove(p)
        print(f"  merge pass {level}: {len(runs)} run(s) -> {len(merged)} run(s), fan-in {FAN_IN}")
        runs, passes = merged, passes + 1
    return runs[0], passes


def replacement_selection_runs(values, memory):
    """Heap of `memory` items; an item smaller than the last output waits for the next run."""
    it = iter(values)
    heap = [(0, v) for v in itertools.islice(it, memory)]
    heapq.heapify(heap)
    run_lengths, current, length = [], 0, 0
    while heap:
        run, v = heapq.heappop(heap)
        if run != current:
            run_lengths.append(length)
            current, length = run, 0
        length += 1
        nxt = next(it, None)
        if nxt is not None:
            heapq.heappush(heap, (run if nxt >= v else run + 1, nxt))
    run_lengths.append(length)
    return run_lengths


def trace_replacement_selection(values, memory):
    """Print every step of replacement selection: what leaves the heap and what waits."""
    it = iter(values)
    heap = [(0, v) for v in itertools.islice(it, memory)]
    heapq.heapify(heap)
    print(f"  start, heap holds (run, value): {sorted(heap)}")
    while heap:
        run, v = heapq.heappop(heap)
        nxt = next(it, None)
        note = "no input left"
        if nxt is not None:
            tag = run if nxt >= v else run + 1
            heapq.heappush(heap, (tag, nxt))
            note = f"read {nxt}, joins run {tag}" + (" (smaller than {}, so it waits)".format(v) if tag != run else "")
        print(f"  output {v} to run {run}; {note}; heap {sorted(heap)}")


def trace_merge(runs):
    """Print the front record of every run before each output record."""
    heap = [(r[0], i, 0) for i, r in enumerate(runs)]
    heapq.heapify(heap)
    while heap:
        fronts = sorted(item[0] for item in heap)
        value, run, pos = heapq.heappop(heap)
        print(f"  fronts {fronts} -> output {value} (from run {run})")
        if pos + 1 < len(runs[run]):
            heapq.heappush(heap, (runs[run][pos + 1], run, pos + 1))


if __name__ == "__main__":
    rng = random.Random(40)
    n = 200_000
    with tempfile.TemporaryDirectory() as work:
        input_path = os.path.join(work, "input.txt")
        with open(input_path, "w") as f:           # "key,arrival_id": 1,000 keys, so many ties
            f.writelines(f"{rng.randrange(1000)},{i}\n" for i in range(n))
        run_dir = os.path.join(work, "runs")
        os.mkdir(run_dir)

        print("1. run generation, with a crash after 7 finished runs")
        try:
            generate_runs(input_path, run_dir, crash_after=7)
        except SimulatedCrash as err:
            print("  first attempt:", err, "| leftover files:", sorted(x for x in os.listdir(run_dir) if x.endswith(".tmp")))
        for leftover in (x for x in os.listdir(run_dir) if x.endswith(".tmp")):
            os.remove(os.path.join(run_dir, leftover))     # restart discards unfinished temp files
        runs, made, skipped = generate_runs(input_path, run_dir)
        print(f"  restart: {skipped} runs reused, {made} runs written, {len(runs)} runs total")

        print("2. merge passes")
        final, passes = merge_passes(runs, run_dir)
        print(f"  total passes over the data: 1 run generation + {passes} merge = {1 + passes}")
        print(f"  predicted merge passes: ceil(log_{FAN_IN}({len(runs)})) = {passes_needed(len(runs), FAN_IN)}")

        with open(input_path) as f:
            expected = sorted(f, key=sort_key)       # stable reference sort, fits here because n is small
        with open(final) as f:
            got = f.readlines()
        print("3. output equals the stable reference sort, including the order of tied keys:", got == expected)

    print("4. run formation: chunk sort versus replacement selection (M = 1,000 records of memory)")
    memory, count = 1_000, 100_000
    values = [rng.random() for _ in range(count)]
    lengths = replacement_selection_runs(values, memory)
    print(f"  random input     : chunk sort {count // memory} runs; replacement selection {len(lengths)} runs, "
          f"mean length {sum(lengths) / len(lengths):.0f} = {sum(lengths) / len(lengths) / memory:.2f} x M")
    jittered = [i + rng.uniform(-300, 300) for i in range(count)]       # nearly sorted, like timestamps
    lengths = replacement_selection_runs(jittered, memory)
    print(f"  nearly sorted    : chunk sort {count // memory} runs; replacement selection {len(lengths)} run(s)")

    print("5. compressing a run (zlib level 1) of 10,000 'timestamp,user,event' lines")
    rng2 = random.Random(1)
    lines = [f"{1_700_000_000 + i * 3 + rng2.randrange(3)},user{rng2.randrange(500)},click\n" for i in range(RUN_RECORDS)]
    shuffled = list(lines)
    rng2.shuffle(shuffled)
    raw = len("".join(lines).encode())
    print(f"  raw {raw:,} bytes; compressed in timestamp order {len(zlib.compress(''.join(lines).encode(), 1)):,}; "
          f"compressed in random order {len(zlib.compress(''.join(shuffled).encode(), 1)):,}")

    print("6. cost model for 500 GiB with 2 GiB of RAM (ESTIMATE: parameters are assumptions)")
    GiB, MiB = 2**30, 2**20
    data, ram, os_and_process = 500 * GiB, 2 * GiB, 0.25 * GiB
    record = 100
    for run_size in (1.5 * GiB, 0.75 * GiB, 0.25 * GiB):
        runs_n = math.ceil(data / run_size)
        buffers = (ram - os_and_process) / (runs_n + 1)
        fan_in = int((ram - os_and_process) // (1 * MiB)) - 1        # smallest useful buffer: 1 MiB
        merge_passes_n = passes_needed(runs_n, fan_in)
        print(f"  run size {run_size / GiB:>4.2f} GiB: {runs_n:>4} runs, merge buffer {buffers / MiB:5.2f} MiB each "
              f"for a single merge, max fan-in at 1 MiB buffers {fan_in}, merge passes {merge_passes_n}")
    passes_total = 2
    moved = passes_total * 2 * data
    print(f"  2 passes, each reads and writes everything: {moved / GiB:,.0f} GiB moved")
    for name, rate in (("HDD 200 MB/s", 200e6), ("SSD 1 GB/s", 1e9), ("NVMe 3 GB/s", 3e9)):
        print(f"    at {name:<13} one shared device: {moved * 1.0 / rate / 3600:5.2f} hours")
    records = data / record
    print(f"  CPU: {records:.2e} records; run sorting ~ {records * math.log2(1.5 * GiB / record):.2e} comparisons, "
          f"334-way merge ~ {records * math.log2(334):.2e} comparisons")
    seek, rate, buf = 0.010, 200e6, 5.35 * MiB
    print(f"  HDD merge efficiency with a 5.35 MiB buffer and a {seek * 1000:.0f} ms seek: "
          f"{(buf / rate) / (buf / rate + seek):.0%} of streaming speed")
    print(f"  buffer arithmetic: (2 - 0.25) GiB / (334 runs + 1 output buffer) = "
          f"{(ram - os_and_process) / 335 / MiB:.2f} MiB; without the output buffer "
          f"{(ram - os_and_process) / 334 / MiB:.2f} MiB")
    print(f"  2,000 runs but at most {fan_in} runs per merge: pass 1 makes {math.ceil(2000 / fan_in)} runs, "
          f"pass 2 makes {math.ceil(math.ceil(2000 / fan_in) / fan_in)}")

    print("7. replacement selection by hand: memory for 3 records, input 4 7 5 2 8 1")
    trace_replacement_selection([4, 7, 5, 2, 8, 1], 3)
    print("   chunk sort of the same input would make", [sorted([4, 7, 5]), sorted([2, 8, 1])])

    print("8. merging two sorted runs with a heap")
    trace_merge([[1, 4, 9], [2, 3, 10]])

Output (python:3.12):

1. run generation, with a crash after 7 finished runs
  first attempt: crashed while writing run 7 | leftover files: ['run0_0007.txt.tmp']
  restart: 7 runs reused, 13 runs written, 20 runs total
2. merge passes
  merge pass 1: 20 run(s) -> 3 run(s), fan-in 8
  merge pass 2: 3 run(s) -> 1 run(s), fan-in 8
  total passes over the data: 1 run generation + 2 merge = 3
  predicted merge passes: ceil(log_8(20)) = 2
3. output equals the stable reference sort, including the order of tied keys: True
4. run formation: chunk sort versus replacement selection (M = 1,000 records of memory)
  random input     : chunk sort 100 runs; replacement selection 51 runs, mean length 1961 = 1.96 x M
  nearly sorted    : chunk sort 100 runs; replacement selection 1 run(s)
5. compressing a run (zlib level 1) of 10,000 'timestamp,user,event' lines
  raw 247,796 bytes; compressed in timestamp order 56,532; compressed in random order 67,146
6. cost model for 500 GiB with 2 GiB of RAM (ESTIMATE: parameters are assumptions)
  run size 1.50 GiB:  334 runs, merge buffer  5.35 MiB each for a single merge, max fan-in at 1 MiB buffers 1791, merge passes 1
  run size 0.75 GiB:  667 runs, merge buffer  2.68 MiB each for a single merge, max fan-in at 1 MiB buffers 1791, merge passes 1
  run size 0.25 GiB: 2000 runs, merge buffer  0.90 MiB each for a single merge, max fan-in at 1 MiB buffers 1791, merge passes 2
  2 passes, each reads and writes everything: 2,000 GiB moved
    at HDD 200 MB/s  one shared device:  2.98 hours
    at SSD 1 GB/s    one shared device:  0.60 hours
    at NVMe 3 GB/s   one shared device:  0.20 hours
  CPU: 5.37e+09 records; run sorting ~ 1.29e+11 comparisons, 334-way merge ~ 4.50e+10 comparisons
  HDD merge efficiency with a 5.35 MiB buffer and a 10 ms seek: 74% of streaming speed
  buffer arithmetic: (2 - 0.25) GiB / (334 runs + 1 output buffer) = 5.35 MiB; without the output buffer 5.37 MiB
  2,000 runs but at most 1791 runs per merge: pass 1 makes 2 runs, pass 2 makes 1
7. replacement selection by hand: memory for 3 records, input 4 7 5 2 8 1
  start, heap holds (run, value): [(0, 4), (0, 5), (0, 7)]
  output 4 to run 0; read 2, joins run 1 (smaller than 4, so it waits); heap [(0, 5), (0, 7), (1, 2)]
  output 5 to run 0; read 8, joins run 0; heap [(0, 7), (0, 8), (1, 2)]
  output 7 to run 0; read 1, joins run 1 (smaller than 7, so it waits); heap [(0, 8), (1, 1), (1, 2)]
  output 8 to run 0; no input left; heap [(1, 1), (1, 2)]
  output 1 to run 1; no input left; heap [(1, 2)]
  output 2 to run 1; no input left; heap []
   chunk sort of the same input would make [[4, 5, 7], [1, 2, 8]]
8. merging two sorted runs with a heap
  fronts [1, 2] -> output 1 (from run 0)
  fronts [2, 4] -> output 2 (from run 1)
  fronts [3, 4] -> output 3 (from run 1)
  fronts [4, 10] -> output 4 (from run 0)
  fronts [9, 10] -> output 9 (from run 0)
  fronts [10] -> output 10 (from run 1)

Failure recovery

• Atomic runs. A run is written to name.tmp, flushed with fsync, then renamed to its final name. The crash in section 1 leaves run0_0007.txt.tmp behind; the restart deletes .tmp files, keeps the 7 finished runs, and writes only the other 13. The simulated crash writes its partial file under the .tmp name, as write_atomically would; a partial file under the final name would be found by the restart and reused as a finished run, which is exactly what the rename prevents. At real scale a crash during run generation costs at most one run (about 1.5 GiB of sorting).
• Delete inputs late. Merge inputs are removed only after the entire pass has produced its outputs, so a crash during a pass can be rerun from the same inputs.
• The merge is the expensive part to lose. A single 334-way merge writes 500 GiB; a crash near the end would repeat it. One design that makes the merge restartable is to split the key range into partitions: take a random sample of keys from the runs, sort it, and use evenly spaced keys from the sorted sample as splitter keys (boundaries, so partition i holds the keys between splitter i-1 and splitter i). Each partition merges the matching slice of every run (found by binary search in each run) into its own output file, and the file is renamed when done. Partitions are independent tasks, so they can run in parallel, and after a crash only the partitions without a finished file are redone. The code above demonstrates the restart only for run generation.
• Validate. Record the input's size and checksum, record counts per run, and compare total records in and out. Rebuild everything if the input file changed.

Parallelism and I/O batching

• Sort chunks on several cores at once (chunk sorts are independent), limited by memory: every chunk being sorted needs its share of RAM.
• Overlap read, sort and write with double buffering so the disk and the CPU are both busy.
• Use large sequential reads and writes (several MiB), and put input, runs and output on separate devices if possible so reads do not interrupt writes. Section 6 estimates that a 5.35 MiB buffer with a 10 ms seek on a 200 MB/s disk reaches about 74 percent of streaming speed (streaming speed means the disk's rate when reading in one long sequential stretch with no jumps), which is why buffers cannot be tiny.
• Prefetch the next buffer of each run while the heap consumes the current one.

Compression of runs

Compressing runs trades CPU for I/O and pays off when the sort is I/O-bound. Section 5 compresses 10,000 timestamped event lines with zlib level 1: 247,796 raw bytes become 56,532 when the lines are in timestamp order and 67,146 in random order, so sorting helps compression. The ratio depends entirely on the data; measure on yours. Decompression adds CPU to the merge, so confirm the sort stays I/O-bound.

Sorting by timestamp for analytics

Event data is usually written roughly in time order, so the input is nearly sorted and replacement selection (above) yields long runs. Often the better design is to avoid one global sort: store data partitioned by hour or day and sort each partition separately, which needs no merge across partitions. Use a compact binary sort key (for example timestamp plus a sequence number, which also makes the order deterministic) rather than parsing text on every comparison.

Cost estimates

All of section 6 is an ESTIMATE from stated parameters (2 GiB RAM, 0.25 GiB reserved, 100-byte records, shared device bandwidth, and decimal MB/s rates). With two passes, the volume is 2 * 2 * 500 = 2,000 GiB moved, which takes 2.98 hours at 200 MB/s, 0.60 hours at 1 GB/s and 0.20 hours at 3 GB/s. CPU is about 5.37 billion records: run sorting is about 1.29e11 comparisons and the 334-way merge about 4.5e10, about 1.7e11 comparisons in all, enough CPU work that the parallelism above matters. If memory shrinks so that runs outnumber the fan-in, the sort gains a merge pass and about half again as much I/O (3,000 GiB for three passes).

Edge cases and common mistakes

• Ties: the demo's output equals the stable reference including tie order; if downstream needs a stable result, order tied records by arrival id.
• Mistake: tiny read buffers, which turn the merge into random I/O.
• Mistake: deleting runs before the merge output is complete.
• Mistake: allocating all of RAM to the chunk and none to buffers.

Running the code

python:3.12, standard library only
save the code above as external_sort.py
python external_sort.py

Direct answer

Three techniques worth naming: reflecting or paraphrasing to confirm you understood correctly, asking clarifying open questions to separate facts from assumptions, and naming or validating the emotion in the room before you return to the content. Each one interrupts the escalation loop by making people feel heard, which is usually what is actually driving the volume, before you re-engage on substance.

Structured elaboration

1. Reflective paraphrasing - restate what you heard in your own words, and check it.
   Example: "So you're saying the denormalized table cuts dashboard latency by removing joins, but you're worried about it duplicating data unless the ETL (extract-transform-load) job dedupes on load, did I get that right?"
2. Clarifying questions - ask specific, open questions that surface the actual constraint instead of the position someone is defending.
   Example: "What's the SLA (service-level agreement) this pipeline needs to hit, and which downstream reports actually need near-real-time data?"
3. Naming and validating the emotion - say out loud what you're noticing, without judging it, before returning to the content.
   Example: "I can tell this design choice matters to both of you, let's make sure we're solving the same problem before we pick a direction."

A fourth thing worth having ready is not a listening technique itself but the natural next move: knowing when to take it offline. If you've paraphrased and clarified and the two sides are just repeating their original positions with no new information, or the tone has shifted from disagreeing about the design to disagreeing about each other, that's the signal to stop working it in the room and move to a small 1:1 (or two separate 1:1s) instead.

Worked example

In a design review, two engineers were talking over each other about whether to denormalize a table for a reporting job. I paraphrased each position back, asked what SLA we were actually designing for, and named that both were trying to protect the same thing, data integrity, from different angles. That got them nodding instead of interrupting. When the same two positions came back around a second time with rising volume, I stopped the group discussion, kept the two engineers after the meeting, gave them 15 minutes one-on-one with me, and brought a joint recommendation back to the wider group afterward.

Trade-offs and pitfalls

Paraphrasing can read as stalling, or as mimicking someone, if you overuse it, so use it when you're genuinely unsure you understood, not on every sentence. Naming emotion badly ("you seem upset") can feel patronizing, so keep it descriptive of the situation, not diagnostic of the person. And active listening only buys you the room to resolve the actual disagreement, it doesn't resolve it by itself. If the two sides have a genuine, well-informed disagreement about the technical tradeoff, no amount of reflecting and clarifying replaces actually gathering evidence or making a call.

Direct answer

A tamper-evident append-only audit log adds a cryptographic hash chain (each row's hash depends on the previous row's hash plus its own content) on top of the ordinary append-only design, so any retroactive edit or deletion breaks the chain and is detectable, while indexing by entity_id and time range keeps it fast to query at scale.

Structured elaboration

CREATE TABLE audit_log (
  log_id BIGINT PRIMARY KEY,
  entity_id BIGINT NOT NULL,
  event_type TEXT NOT NULL,
  payload JSONB NOT NULL,
  occurred_at TIMESTAMP NOT NULL,
  prev_hash TEXT NOT NULL,
  entry_hash TEXT NOT NULL   -- e.g., SHA-256(prev_hash || entity_id || event_type || payload || occurred_at)
);
CREATE INDEX idx_audit_entity_time ON audit_log (entity_id, occurred_at);

• Where to store the hash: entry_hash is computed at write time from the previous row's hash plus this row's own content, forming a hash chain analogous to a blockchain's block-linking (though without needing distributed consensus, since a single trusted writer is sufficient for this use case); prev_hash is stored alongside it so verification doesn't require a separate lookup.
• Verifying integrity: re-walk the chain from the beginning (or from a periodically-checkpointed known-good hash), recomputing each row's expected hash from its content and the prior row's hash, and compare against the stored entry_hash; any mismatch means the row (or something before it in the chain) was altered or deleted after the fact.
• Scaling verification across large volumes: full-chain verification from the beginning becomes expensive as the log grows; periodically (daily, say) compute and separately, immutably store a checkpoint hash (anchoring the chain's state at that point, potentially even published somewhere externally verifiable), so routine verification only needs to re-walk the chain since the last checkpoint, not from the very beginning.

Worked example

Verified in sqlite3 with a simple three-entry hash chain (using Python's hashlib to compute each entry_hash from a concatenation of prev_hash and the row's own content): recomputing the chain from stored content correctly reproduces every stored hash when the data is untouched; after manually altering one entry's payload in place (simulating tampering) without recomputing hashes, re-verification correctly detects a mismatch starting at the altered row and propagating to every row after it, confirming the tamper-evidence property holds and pinpoints where the alteration occurred.

Trade-offs and pitfalls

• A hash chain proves the log hasn't been altered SINCE it was written, but it does nothing to prevent a privileged database administrator from altering the log AND recomputing every subsequent hash to match; true tamper-evidence against a privileged insider requires the periodic checkpoint to be anchored somewhere the administrator can't also rewrite (an external, append-only store, or a third-party timestamping service), not just stored in the same database.
• Access controls matter as much as the hash chain itself: the hash chain detects tampering after the fact, it doesn't prevent unauthorized reads or writes in the first place, so row-level or column-level access controls (who can even query this table) remain a separate, necessary control.
• Scaling queries across large volumes benefits from the (entity_id, occurred_at) index for the common "history of one entity over a time range" query, but a full-chain integrity verification is inherently sequential (each hash depends on the previous one) and doesn't parallelize the same way an ordinary analytical scan would, which is exactly why checkpointing matters for keeping routine verification bounded in cost.

Direct answer
When two teams each blame the other after a shared outage, the facilitator's job is to separate 'what does the evidence show' from 'who is at fault,' and to run the discussion so both teams contribute evidence to a single shared timeline rather than defending competing narratives. This usually means gathering data from both sides before the meeting, framing the discussion around the timeline rather than either team's story, and being explicit that the goal is a joint fix both teams commit to, not a verdict on whose configuration change or whose load caused it.

Structured elaboration

• Before the meeting: pull metrics, logs, and change history from both teams independently, and build a single combined timeline that includes both teams' events (the config change AND the load increase, with exact timestamps), so the meeting starts from shared facts instead of each side's framing of the story.
• Framing: state explicitly at the start that the goal is a joint understanding of what happened and what to change, not deciding which team was 'right.' Reiterate that even if one team's change was the proximate trigger, the fact the system as a whole had no safeguard against that class of change is the real gap.
• In the discussion: ask each team to walk through their own timeline of events with evidence, not assertions, then look together at where the timelines intersect. Often the honest finding is that BOTH factors mattered (the config change was the trigger, but the system had no capacity headroom to absorb even a modest load increase on top of it), which is a more useful and more accurate conclusion than picking one side.
• Reaching resolution when the evidence itself is ambiguous: if the two hypotheses are genuinely both plausible and the data doesn't cleanly distinguish them, say so explicitly in the writeup rather than forcing a false consensus, and define what additional instrumentation or a follow-up experiment would resolve the ambiguity next time.
• Securing buy-in: end with action items owned by BOTH teams, not just one, so neither team can read the outcome as 'we were blamed and they weren't.'

Worked example
Team A's dependency config change and Team B's traffic increase happened within minutes of each other before a shared-service outage. The facilitator's combined timeline shows the config change reduced the service's effective connection pool size at 14:02, and Team B's traffic, itself normal and within historical range, arrived at 14:04 and exhausted the now-smaller pool. Neither event alone would have caused the outage: normal traffic against the old pool size would have been fine, and the smaller pool alone, without the traffic bump, might have gone unnoticed for a while. The joint conclusion: the real gap is that the shared service has no automated alert or gate when a config change materially reduces its capacity headroom. Action items: Team A adds a pre-deploy check that flags capacity-reducing config changes above a threshold, and the platform team (not either disputing team) adds monitoring on effective headroom versus recent traffic patterns, so future changes like this are caught automatically rather than depending on either team noticing.

Trade-offs and pitfalls
The most common failure is the facilitator implicitly picking a side, often by unconsciously giving one team's narrative more airtime, which the other team notices and which damages trust in the process going forward. A second is forcing a single, tidy root cause when the honest finding is that multiple factors from multiple teams combined; naming that clearly, with joint ownership of the fix, produces a more durable resolution than a false consensus.

The merge logic, applying CDC (Change Data Capture) change records from multiple sources into a canonical customers table, resolving conflicts by (event timestamp, then source priority) and handling deletes correctly, including a delete that arrives for a customer never previously seen:

MERGE INTO customers AS t
USING (
  SELECT customer_id, name, email, op, event_ts, source
  FROM (
    SELECT *, ROW_NUMBER() OVER (
      PARTITION BY customer_id ORDER BY event_ts DESC, source_priority ASC
    ) AS rn
    FROM cdc_events
  )
  WHERE rn = 1
) AS s
ON t.customer_id = s.customer_id
WHEN MATCHED AND s.op = 'delete' AND s.event_ts > t.last_applied_ts THEN
  UPDATE SET is_deleted = true, last_applied_ts = s.event_ts, last_source = s.source
WHEN MATCHED AND s.op = 'upsert' AND s.event_ts > t.last_applied_ts THEN
  UPDATE SET name = s.name, email = s.email, is_deleted = false,
             last_applied_ts = s.event_ts, last_source = s.source
WHEN NOT MATCHED AND s.op = 'upsert' THEN
  INSERT (customer_id, name, email, is_deleted, last_applied_ts, last_source)
  VALUES (s.customer_id, s.name, s.email, false, s.event_ts, s.source)
WHEN NOT MATCHED AND s.op = 'delete' THEN
  INSERT (customer_id, name, email, is_deleted, last_applied_ts, last_source)
  VALUES (s.customer_id, NULL, NULL, true, s.event_ts, s.source);

Verified against a fixture with a real conflict and a real edge case:

target before: C1 = Alice, alice@old.com, last_applied 09:00, source=crm

staged CDC events:
  C1, name=Alice,   email=alice@new.com, op=upsert, ts=10:00, source=crm,     priority=2
  C1, name=Alice B, email=alice@new.com, op=upsert, ts=10:00, source=billing, priority=1  <- exact tie in ts with the crm event
  C2, name=Bob,     email=bob@x.com,     op=upsert, ts=10:05, source=crm,     priority=2
  C2, (delete),                          op=delete, ts=10:10, source=crm,     priority=2  <- C2 never existed in target yet

after merge:
  C1: name=Alice B, email=alice@new.com, last_source=billing   <- billing correctly won the exact tie
  C2: is_deleted=True, last_source=crm                          <- correctly landed as a tombstone, not silently dropped

The C1 result confirms conflict resolution works: two sources reported changes at the exact same event timestamp, and the row-number tiebreak on source_priority correctly let billing (priority 1) win over crm (priority 2).

The bug this fixture actually caught: my first version of this query only had a WHEN NOT MATCHED AND s.op = 'upsert' insert branch, reasoning that a delete can't apply to a row that doesn't exist. Running it against this fixture, customer C2 (whose only two events in the batch are an upsert immediately followed by a delete) vanished entirely: the per-customer dedup keeps only the LATEST event per customer id, which for C2 is the delete, and since the delete had no corresponding target row and no "NOT MATCHED AND delete" branch to insert one, C2 was silently dropped from the target altogether, neither inserted nor recorded as deleted. Adding the WHEN NOT MATCHED AND s.op = 'delete' branch, which inserts a tombstone row instead of doing nothing, closes that gap. This is exactly the class of bug that reads correctly and passes a simple test (any test that doesn't include a same-batch insert-then-delete case) while silently losing data.

Recording source metadata for reconciliation: last_source on every row (as in the schema above) lets you audit which source's write actually took effect for any given customer, which is essential for debugging a future "why does this customer have the wrong data" complaint when two sources disagree.

Direct answer

Keeping raw ingestion separate means the raw event log or landing files stay retained independently of whatever transform code currently runs on top of them, so when a bug is found in the transform logic, the fix is just code plus a replay of the same untouched history. If ingestion and transformation are tightly coupled, typically because the ingestion step only keeps output the transform has already touched, or discards source events once processed, fixing that same bug means the input that would prove the fix, and let you reprocess history, may no longer exist in the form it originally had.

Structured elaboration

Walk through what actually breaks when the two are coupled, step by step, using a concrete bug: a join-key normalization error that undercounts one category.

With decoupled raw storage. Raw events for the affected period still exist untouched, byte-for-byte as originally ingested. Fix the transform code, rerun it over the same raw window, get corrected output. No data was lost, only recomputed.

With ingestion and transformation coupled in one pass. The job extracts, transforms, and loads in a single step and only ever writes the transformed result. The raw form the bug depended on is gone. Reprocessing means re-extracting from the source system itself, which may no longer have that historical data at all if its own retention window has passed, or, worse, may return something different because source rows have since been updated or deleted.

Non-determinism creeps in over time. Even if the source's retention happens to still cover the period, its current state reflects everything that has happened to those rows SINCE, updates and deletes included. A "replay" built from a live re-extract silently reflects today's state of the source, not the historical state the original run actually saw, producing a different, non-reproducible answer than a true raw-log replay would give.

Blast radius of testing the fix. In the coupled design, testing the fix means running it against the live production extract path, since there is no stored raw form to safely target instead, which risks re-hitting the source system's load and re-triggering any side effects the ingestion step has, if that step isn't a pure read.

Worked example

A category-normalization bug is found to affect 45 days of history. With raw storage retained for 90 days (or archived further to cold storage), the fix reprocesses exactly those 45 days from the untouched raw copy, a bounded, fully deterministic amount of recompute. With ingestion and transformation coupled, and the source system itself retaining only 30 days of live queryable history before purging or archiving older rows elsewhere:

Those 15 days can no longer be correctly reproduced from the source at all; that portion of history is permanently uncorrectable without a raw copy that was never kept.

Trade-offs & pitfalls

Decoupling has a real cost: storing raw data for a long window means paying to keep data that's mostly redundant with its own transformed output. The mitigation is a defined raw-retention policy, keep raw only as long as the realistic "we might need to reprocess" window actually is, matching how far back schema or logic bugs are typically caught, rather than retaining it forever by default with no justification. Raw retention only helps if raw is genuinely immutable and untouched; if the ingestion step itself silently applies "a little cleaning for convenience" on the way in, the coupling has just moved one layer down and reintroduced the same problem under a different name.

Direct answer

For 500M rows with complex joins, the honest first question is not "pandas or Dask/Spark" in the abstract, it is "does this fit comfortably in one machine's RAM, with room left over for the joins to work in, and does the team have the operational skill to run and debug a cluster." If the working set (not just the raw file size) fits with headroom and the job runs often enough to be worth optimizing locally, stay on pandas. If the data genuinely does not fit, or the joins require shuffling data across more memory than one box has, move to Dask (same API surface, easiest migration) or Spark (steeper learning curve, but the most mature engine for very large shuffles and production scheduling).

Decision criteria

• Does the working set fit in RAM, with headroom? Complex joins do not just need space for the two input tables, hash-join and merge operations build intermediate structures that can transiently use several times the input size. "Fits on disk" is not the test; "fits in RAM with room to spare for the join itself" is.
• Latency and iteration pattern. Interactive exploratory analysis rewards pandas's instant local feedback loop. Scheduled batch ETL that runs unattended rewards a distributed engine's fault tolerance and ability to retry a failed partition instead of restarting the whole job.
• Cluster availability and operational cost. A Spark or Dask cluster is infrastructure someone has to provision, monitor, and pay for even when idle. If none exists and the data fits on a single large instance, standing one up purely for this job is a real cost, not a free upgrade.
• Operator skill. Debugging a stack trace from a distributed engine (which executor failed, which partition, was it a shuffle spill or an out-of-memory kill) is a different skill from debugging pandas locally. A team with only pandas experience will be slower and more error-prone on a distributed engine at first, that ramp-up cost is real and should be weighed against the migration.
• Shape and cost of the joins specifically. Wide joins on high-cardinality keys, or joins that fan out (duplicate keys on either side), are exactly the operations that blow up memory fastest and benefit most from a shuffle-aware, disk-spilling engine when they no longer fit.

Worked example: sizing the decision

Take a concrete, honest estimate rather than a vague "it's big data." Suppose the 500M-row table has 10 numeric (float64) columns and 3 identifier/string columns.

Numeric columns are straightforward: float64 is 8 bytes per value, so 10 columns cost 10 * 8 = 80 bytes per row.

String columns are the part people usually get wrong, because the answer depends on which pandas version you are running. Measuring directly on a 200,000-row sample of realistic ~9-character id-like strings:

import pandas as pd
import numpy as np

n = 200_000
ids = np.random.RandomState(0).choice([f"evt_{i:06d}" for i in range(50_000)], n)

s_default = pd.Series(ids)               # pandas 3.0's default string dtype
s_object  = pd.Series(ids, dtype="object")  # legacy object dtype

print(s_default.dtype, s_default.memory_usage(deep=True))
print(s_object.dtype, s_object.memory_usage(deep=True))

Verified output on pandas 3.0.3:

str 3600132
object 11800132

That is roughly 18 bytes/row for pandas 3.0's default string dtype versus roughly 59 bytes/row for the legacy object dtype, on the same data, a real ~3.3x difference. This matters directly for the pandas-vs-distributed decision: pandas 3.0 changed the default text representation to a more compact, Arrow-backed string dtype, so the same 500M-row table with string columns needs meaningfully less RAM on current pandas than the same code would have needed pre-3.0, without changing a line of application code.

Using the pandas-3.0 figure for 3 string columns: 3 * 18 = 54 bytes/row. Total: (80 + 54) * 500,000,000 ≈ 67 GB just to hold the columns, before accounting for the index or any join-time intermediate structures. This 67 GB figure is an ESTIMATE built from a real per-row measurement, not a measured total (that would require the actual dataset), and a hash join between two tables of this size can transiently need multiples of that for the intermediate hash tables and output. On a machine with, say, 128 GB of RAM, that is uncomfortably tight for "just the data" and genuinely risky once you add a complex multi-key join on top. That is the point at which the criteria above start pointing toward Dask or Spark rather than pandas.

Migration plan: pandas to Dask (chosen for API compatibility)

Dask is the lower-friction migration target because its DataFrame API mirrors pandas closely, so most of the port is mechanical rather than a rewrite.

1. Inventory. List every data source, transformation, join, and any custom row-wise or group-wise function (UDF), since arbitrary Python callables are the part that ports least cleanly.
2. Localize and partition. Identify code that assumes a single global index or does whole-DataFrame in-place mutation; both patterns work poorly across partitions and usually need to be rewritten as column-wise, partition-independent operations.
3. Port incrementally, starting with I/O. Swap pandas.read_csv for dask.dataframe.read_csv first and confirm the inferred schema matches, then convert stateless transforms (select, filter, map_partitions), then the heavier groupbys and joins, benchmarking each stage on a small cluster or LocalCluster before scaling up.
4. Rework join strategy. Wide, high-cardinality joins need a real shuffle in a distributed engine; small lookup tables should instead be broadcast to every worker (a broadcast join) rather than shuffled, which Dask (like Spark) supports explicitly and which avoids moving the large table's data across the network for no reason.
5. Validate against pandas. For a representative subset that still fits in pandas, run both pipelines and diff aggregate checksums (row counts, column sums, a hash of a sorted subset) to catch behavioral differences, not just "it ran without erroring."
6. Tune. Set partition sizes deliberately (too many small partitions adds scheduling overhead, too few defeats parallelism), and use .persist() on intermediate results that are reused, so they are not recomputed from scratch on every downstream operation.
7. Add CI, monitoring, and a parallel-run rollout. Run the Dask pipeline alongside the existing pandas pipeline for several production cycles before cutting over, so a discrepancy is caught against a known-good baseline rather than in production alone.

Trade-offs and pitfalls

• API compatibility: pandas has the richest, simplest API. Dask covers most of it directly (which is why it is usually the first migration target), but some pandas idioms (certain multi-index operations, some apply patterns) either are not supported or require the more expensive general-purpose map_partitions escape hatch. Spark's DataFrame API is meaningfully different (and JVM-backed under the hood), so a Spark migration is a heavier lift than a Dask one even though it is often the more mature choice for very large shuffles.
• Memory model: pandas holds everything in one process's memory. Dask breaks the data into partitions and schedules work across processes or threads via a task graph, spilling to disk when a partition does not fit. Spark's JVM-managed execution engine is built around disk spill and shuffle from the ground up and tends to be the more battle-tested choice at the largest scales, at the cost of a heavier operational footprint.
• Debugging: a pandas bug reproduces locally, in one process, with a normal Python traceback. A Dask or Spark bug can depend on which partition or executor failed, and reproducing it may require pulling that partition out and debugging it in isolation, materially slower in practice even when the underlying logic error is simple.
• The single biggest pitfall in this kind of migration is treating it as "the same code, just distributed." Code that relies on global row order, a single monotonic index, or in-place mutation of the whole frame will silently misbehave or need a genuine rewrite, not a search-and-replace of pd. for dd..
• Do not skip the parallel-run validation step to save time. A distributed join that silently drops or duplicates rows because of a key-type mismatch across partitions is exactly the kind of bug that a "the pipeline finished without an error" smoke test will not catch, but a checksum diff against the pandas baseline will.

Direct answer

Waiting stops being the right call once the delay is on your critical path (the chain of work that directly determines your deadline) with no updated ETA, or once the cost of continuing to wait (rework, workarounds, compounding risk) is clearly larger than the cost of escalating. Decide the trigger in advance, not in the moment, and escalate by framing it around the shared deadline and offering to help unblock, not by assigning blame, so the relationship survives the conversation.

Structured elaboration

1. Set the trigger before you need it. At the point you first take on a dependency, agree on what "stalled" means and when you'll escalate if there's no movement, for example, "if there's no updated ETA by [date], I'll raise it." Deciding this ahead of time keeps the eventual call from being an emotionally loaded, in-the-moment judgment.
2. Watch for the signals that waiting has become the wrong call, even without a pre-set trigger: no visible progress or updated estimate, the delay has moved onto your own critical path, you're already absorbing compounding cost (rework, a growing workaround), or the nature of their blocker changed without anyone telling you.
3. Escalate at the right altitude, in order. Start with a direct conversation with the owner (not their manager first, which reads as going around them), then their lead if that doesn't move things, then a cross-functional or executive conversation only if the first two steps don't resolve it. Skipping straight to the top burns trust even when you're right to escalate.
4. Frame the escalation around the shared goal. Bring what you've tried and the concrete impact of the delay, and lead with an offer to help (extra hands, a clearer spec, a joint troubleshooting session) rather than a demand for status. This keeps the conversation collaborative instead of adversarial.
5. When the dependency is an external vendor rather than an internal team, the escalation lever is fundamentally different. There's no peer relationship conversation to have in the same sense: the path runs through contract renegotiation (invoking SLA, or service level agreement, terms, escalating through the vendor's account team) and executive/customer communication about timeline impact, because a vendor delay usually has stakeholders beyond your own working team (customers waiting on the date, your own leadership needing to manage expectations upward). The internal escalation ladder in step 3 assumes a peer relationship you can repair with tone and framing; the vendor case assumes a commercial relationship you manage with contract terms and proactive, honest communication about the schedule impact instead.

Worked example

Two weeks into waiting on an internal platform team's API, with no updated ETA since the first week and the launch date now two weeks out, the trigger from step 1 (no ETA update within a week) has already been crossed. The escalation opens with the owner directly: "This is now going to affect our launch date. What's actually blocking it, and is there anything I can do to help, pair on it, provide test data, take a piece of the work?" Only if that doesn't produce movement within a short, stated window does it go to their lead, framed the same way: shared deadline, concrete impact, an offer to help.

If instead the dependency were owned by an external vendor who'd gone quiet for two weeks on a contracted deliverable, the move isn't a peer conversation with an individual, it's raising the delay through the account relationship against the SLA in the contract, while separately and proactively telling internal leadership (and, if relevant, the customer waiting on the date) what the timeline impact now looks like, rather than continuing to absorb the delay silently and hoping the vendor resolves it before anyone notices.

Trade-offs & pitfalls

• Pitfall: escalating without a pre-agreed trigger, so the decision looks reactive or, worse, personal, when it happens.
• Pitfall: skipping escalation levels internally (going straight to a director) when a direct conversation with the owner hadn't been tried yet, damaging a relationship you'll need again.
• Pitfall: treating a vendor delay like an internal one, i.e., waiting patiently and being "collaborative" with a counterparty who has no equivalent incentive to preserve the relationship the way an internal peer does.
• Senior differentiator: pre-negotiating the escalation threshold when the dependency is first created, not two weeks into silence, and recognizing early which kind of dependency (peer relationship vs. commercial contract) you're actually managing, since that changes which lever you reach for.