Sorting and Searching Algorithms Questions
Comparison and non-comparison sorting (quicksort with its pivot and partition choices and hybrid cutoffs, mergesort, heapsort as a sort, insertion/selection/bubble, counting/radix/bucket), their stability, in-place behavior, stack depth and lower bounds, and choosing a sort from input constraints, including near-sorted data, library sorts (Timsort, introsort, language defaults), comparator design and expensive comparisons or copies. Also selection (quickselect, median-of-medians, nth_element), external merge sorting of data larger than memory (run generation and merge passes), and binary search with its variants: first/last occurrence, lower/upper bound, rotated arrays with or without duplicates, peak and mountain arrays, sorted matrices, unbounded search, loop-invariant correctness and off-by-one pitfalls, and binary search on an answer space (discrete or continuous, including k-th smallest by value). Includes divide-and-conquer reasoning for sorts (merge step, merging sorted arrays, median of two sorted arrays, inversion counting). Excludes heap data-structure mechanics and heap-based top-k and k-way merging, trees and tries, general graph search, distributed data-processing systems, approximate nearest-neighbor and streaming-sketch algorithms, and abstract Big-O comparison.
Design an external merge sort to sort a 500GB file on disk given only 2GB of RAM. Describe run generation, run size selection, k-way merge strategy, disk I/O optimization, and how to recover from failures mid-sort. Provide rough cost estimates for passes over the data.
Sample Answer
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):
| Run size | Runs | Merge buffer per run (one merge of all runs, 1.75 GiB shared by the runs plus one output buffer) | Merge passes |
|---|---|---|---|
| 1.50 GiB | 334 | 5.35 MiB | 1 |
| 0.75 GiB | 667 | 2.68 MiB | 1 |
| 0.25 GiB | 2,000 | 0.90 MiB | 2 |
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 withfsync, then renamed to its final name. The crash in section 1 leavesrun0_0007.txt.tmpbehind; the restart deletes.tmpfiles, keeps the 7 finished runs, and writes only the other 13. The simulated crash writes its partial file under the.tmpname, aswrite_atomicallywould; 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
That is every published Sorting and Searching Algorithms question for Data Engineer so far. Browse the other topics in this category, or practice this one interactively.