Sorting and Searching Algorithms Questions

Direct answer

Do not sort the objects. Sort a small array of indices by the objects' keys with a stable sort, then rearrange the objects once, in place, by following the cycles of that permutation. The objects move at most n times in total (each at most once) and the only extra object alive at any moment is a single temporary; the extra memory is about n small integers (plus whatever the stable sort uses internally). Stability comes from the sort of indices: equal keys keep their input order. The trade-off is that comparisons go through an indirection (look up the key by index), which is less friendly to the CPU cache than comparing values stored side by side, and the in-place permutation pass jumps around memory.

The idea in plain words

Sorting objects directly with a merge sort copies each object about 2 log2 n times (once into the buffer and once back, per level) and needs a buffer as large as the data. If an object is expensive to copy, that is both a time cost and a memory cost, and they are different quantities: the total copying work is the sum over the whole run (2 n log2 n object writes here, shown on 4 objects in the output below), while the peak extra memory is the most that is alive at once (a buffer of n objects here). Indirection attacks both: sort cheap indices, so the heavy objects are not touched until the final order is known, and then move each object once.

Terms: indirection means reaching the object through a small stand-in (here its position number) instead of holding the object itself. Argsort is a sort that returns the positions of the items in sorted order instead of the sorted items. Cache is the small, fast memory next to the CPU: it loads a block of neighbouring memory at once, so reading elements that sit next to each other is fast, while jumping around memory (a cache miss, when the needed bytes are not in the cache yet) is slow. A permutation is a rearrangement of positions; it splits into disjoint cycles (slot 0 wants the item from slot 1, slot 1 wants the item from slot 3, slot 3 wants the item from slot 0, so the three items rotate among those slots). Stable means equal keys keep input order. Fixed point is a slot already holding the right object.

Approach

1. Extract the keys into a list keys (one cheap value per object; use the object field directly if comparison is cheap).
2. order = sorted(range(n), key=keys.__getitem__). Python's sort is stable, so order[i] is the old position of the object that belongs in slot i, and objects with equal keys appear in input order.
3. For each slot start not yet finished: hold store[start] in one temporary, then repeatedly do store[j] = store[order[j]] along the cycle, mark each finished slot (order[j] = j, which makes the pass need no separate visited array), and finally put the held object into the last slot. The loop body is two separate statements, nxt = order[j] (remember where the cycle goes next) and order[j] = j (mark this slot finished), so the order is explicit: read first, then overwrite.
4. Fixed points cost nothing. The number of object writes is n minus the number of fixed points.
5. If the caller only needs to iterate in sorted order, skip step 3 and read store[order[i]] directly: zero object moves.

Code

Store is a list that counts slot assignments, so the write counts below are tallied by the same functions that do the work. merge_sort_objects is the comparison baseline: a stable merge sort that moves the objects themselves with one buffer, using index ranges instead of slices.

import random


class Store(list):
    """A list that counts slot assignments, standing in for 'moving an expensive object'."""
    writes = 0

    def __setitem__(self, i, value):
        Store.writes += 1
        super().__setitem__(i, value)


def sorted_order(keys):
    """Stable argsort: positions of the items in sorted order. Equal keys keep their input order."""
    return sorted(range(len(keys)), key=keys.__getitem__)


def apply_order_in_place(store, order, trace=False):
    """Rearrange store so that new[i] = old[order[i]], following permutation cycles.

    Holds one object in a temporary at a time. Consumes `order` (marks finished slots with order[j] = j).
    """
    for start in range(len(store)):
        if order[start] == start:
            continue
        held = store[start]                  # the only extra object alive
        j = start
        while order[j] != start:
            store[j] = store[order[j]]
            nxt = order[j]                   # where the next slot in this cycle is
            order[j] = j                     # mark slot j finished
            j = nxt
        store[j] = held
        order[j] = j
        if trace:
            print(f"  cycle from slot {start}: {[o[0] + o[1] for o in store]}")


def merge_sort_objects(store, key, trace=False):
    """Stable top-down merge sort that moves the objects themselves, using one buffer of len(store)."""
    n = len(store)
    buf = Store([None] * n)

    def sort(lo, hi):
        if hi - lo < 2:
            return
        mid = (lo + hi) // 2
        sort(lo, mid)
        sort(mid, hi)
        i, j = lo, mid
        for k in range(lo, hi):
            if j >= hi or (i < mid and key(store[i]) <= key(store[j])):   # <= keeps ties in input order
                buf[k] = store[i]; i += 1
            else:
                buf[k] = store[j]; j += 1
        for k in range(lo, hi):
            store[k] = buf[k]
        if trace:
            print(f"  merge slots {lo}..{hi - 1}: total object writes so far {Store.writes}")
    sort(0, n)


def make_records(n, rng, distinct):
    return Store((f"k{rng.randrange(distinct):03d}", f"#{i}") for i in range(n))     # (key, tag)


if __name__ == "__main__":
    rng = random.Random(3)

    print("trace: 6 records as key+tag, sorted by key")
    recs = Store([("b", "1"), ("a", "2"), ("b", "3"), ("a", "4"), ("c", "5"), ("a", "6")])
    keys = [r[0] for r in recs]
    order = sorted_order(keys)
    print("  before:", [r[0] + r[1] for r in recs])
    print("  order (old position of each sorted slot):", order)
    Store.writes = 0
    apply_order_in_place(recs, list(order), trace=True)
    print("  after: ", [r[0] + r[1] for r in recs], " object writes:", Store.writes)

    print("\nmerge sort on 4 objects, object writes after each merge (each merge writes its slots twice: into the buffer, then back)")
    four = Store([("d", "0"), ("c", "1"), ("b", "2"), ("a", "3")])
    Store.writes = 0
    merge_sort_objects(four, key=lambda r: r[0], trace=True)
    print("  2 * n * log2(n) for n = 4:", 2 * 4 * 2)

    print("\nedge cases (object writes)")
    for name, rows in (("empty", []), ("one", [("a", "0")]), ("all equal", [("a", str(i)) for i in range(5)]),
                       ("already sorted", [(c, "0") for c in "abcde"]), ("reversed", [(c, "0") for c in "edcba"])):
        s = Store(rows); Store.writes = 0
        apply_order_in_place(s, sorted_order([r[0] for r in s]))
        print(f"  {name:15} writes={Store.writes:2d} result={[r[0] + r[1] for r in s]}")

    print("\nobject writes for n records (keys from 50 distinct values)")
    print("      n   merge sort on objects   index sort + cycles   2 n log2 n")
    for n in (8, 64, 1024):
        base = make_records(n, rng, 50)
        a, b = Store(base), Store(base)
        Store.writes = 0; merge_sort_objects(a, key=lambda r: r[0]); w_merge = Store.writes
        Store.writes = 0; apply_order_in_place(b, sorted_order([r[0] for r in b])); w_idx = Store.writes
        assert list(a) == list(b) == sorted(base, key=lambda r: r[0])
        lg = n.bit_length() - 1
        print(f"  {n:5d}   {w_merge:19d}   {w_idx:19d}   {2 * n * lg:10d}")

    print("\nrandom check: 300 inputs of up to 40 records with many ties, vs sorted(key=...)")
    for _ in range(300):
        rows = make_records(rng.randrange(0, 41), rng, 4)
        want = sorted(rows, key=lambda r: r[0])
        got = Store(rows)
        apply_order_in_place(got, sorted_order([r[0] for r in got]))
        assert list(got) == want
    print("  all match, ties in input order")

Run with Python 3.12, it prints:

trace: 6 records as key+tag, sorted by key
  before: ['b1', 'a2', 'b3', 'a4', 'c5', 'a6']
  order (old position of each sorted slot): [1, 3, 5, 0, 2, 4]
  cycle from slot 0: ['a2', 'a4', 'b3', 'b1', 'c5', 'a6']
  cycle from slot 2: ['a2', 'a4', 'a6', 'b1', 'b3', 'c5']
  after:  ['a2', 'a4', 'a6', 'b1', 'b3', 'c5']  object writes: 6

merge sort on 4 objects, object writes after each merge (each merge writes its slots twice: into the buffer, then back)
  merge slots 0..1: total object writes so far 4
  merge slots 2..3: total object writes so far 8
  merge slots 0..3: total object writes so far 16
  2 * n * log2(n) for n = 4: 16

edge cases (object writes)
  empty           writes= 0 result=[]
  one             writes= 0 result=['a0']
  all equal       writes= 0 result=['a0', 'a1', 'a2', 'a3', 'a4']
  already sorted  writes= 0 result=['a0', 'b0', 'c0', 'd0', 'e0']
  reversed        writes= 4 result=['a0', 'b0', 'c0', 'd0', 'e0']

object writes for n records (keys from 50 distinct values)
      n   merge sort on objects   index sort + cycles   2 n log2 n
      8                    48                     7           48
     64                   768                    64          768
   1024                 20480                  1023        20480

random check: 300 inputs of up to 40 records with many ties, vs sorted(key=...)
  all match, ties in input order

Reading the trace

On 4 objects in reverse key order, the merge sort writes 4 object slots in the first merge of two slots (2 into the buffer, 2 back), 8 after the second, and 16 after the last merge, which equals 2 * n * log2 n = 16: every level of merging writes all n slots twice, and there are log2 n levels.

The six records sort by key to a2 a4 a6 b1 b3 c5 (b1 means key b, tag 1). order = [1, 3, 5, 0, 2, 4] says slot 0 wants the object now at position 1, slot 1 wants position 3, and so on.

• Cycle from slot 0: hold b1; slot 0 takes a2, slot 1 takes a4, and slot 3's value comes from the held b1. Array: a2 a4 b3 b1 c5 a6. Three writes.
• Cycle from slot 2: hold b3; slot 2 takes a6, slot 5 takes c5, slot 4 gets the held b3. Three writes.

Total 6 writes for 6 objects, and the equal keys a2 a4 a6 and b1 b3 kept their input order. The edge-case rows show 0 writes for empty, one-element, all-equal and already sorted input (every slot is a fixed point), and 4 writes for the reversed 5-item input (two swaps).

Complexity, with the reason

• Time: O(n log n) key comparisons for the argsort, plus O(n) for the cycle pass. Object writes: at most n (the table shows 7, 64 and 1023 against 48, 768 and 20480 for merge sort on the objects; 20480 is exactly 2 n log2 n for n = 1024).
• Peak extra memory: the order list (n index entries), one temporary object, and the working memory of the sort itself. CPython's list sort may need a temporary array of up to N//2 pointers during merges (from the implementation notes in CPython's listsort.txt), and a key-function sort also keeps its computed keys; all of those are pointer-sized, not object-sized. Direct merge sort on objects needs a buffer of n objects.
• Cache behaviour: keys[i] lookups and the cycle walk are random accesses, so each is likely to miss the cache on large n; sorting contiguous (key, index) pairs trades 2x key memory for sequential access during comparison. This is a qualitative trade-off; the run counts moves, not time.

If even the index array is too large, or the data exceeds memory

In an interview, the index sort followed by the cycle pass is the expected answer, and mentioning an external sort for data beyond memory is usual; in-place stable block sorts are specialist material that comes up when the interviewer pushes on O(1) extra memory.

• Stable in-place sorts exist (block merge sorts: they merge in small fixed-size blocks inside the array itself, and WikiSort and GrailSort are two published implementations) that use O(1) extra memory with O(n log n) time (the Wikipedia article on block sort gives O(1) auxiliary space in its stated model), but they are intricate and have larger constants; use a tested library implementation rather than writing one.
• C++ std::stable_sort tries to allocate a temporary buffer the size of the data; if that allocation fails, the implementation falls back to a slower algorithm, documented as O(N log^2 N) comparisons instead of O(N log N) (cppreference), that is N times the square of log N, a factor of log N more than the normal bound. In a memory-constrained process that fallback can be what actually runs, so sorting indices is the safer plan.
• External sort: if the data does not fit in memory, sort memory-sized runs of (key, record id), write them to disk, and merge the runs with a k-way merge (repeatedly take the smallest front item among k sorted runs); the permutation is then applied by reading records in sorted order.

Edge cases and common mistakes

• Empty and one-element inputs run without error; all-equal input moves nothing.
• Using an unstable sort of indices (or a comparator that treats ties as unordered) loses stability; the argsort here is stable, and the random check with only four distinct keys confirms equal keys keep input order.
• Applying the permutation the wrong way round (new[order[i]] = old[i] instead of new[i] = old[order[i]]) gives the inverse permutation, which is a different, wrong arrangement; the tests compare against sorted(..., key=...).
• Forgetting to mark finished slots makes the cycle loop revisit them and corrupt the data.
• The cycle pass destroys order; copy it first (the driver passes list(order)) if it is needed afterwards.

The idea in plain words

Versions 1 to n were released in order, and once a version is bad every later one is bad too (the API is monotonic: its answers go False, False, ..., True, True). You want the first True. The list of answers is a sorted yes/no sequence, so you can binary search it: ask about the middle version, and the answer tells you which half to throw away. Each API call is expensive in the story (it builds and tests a release), so the goal is the fewest calls, not the fewest lines.

The brute force asks about version 1, 2, 3, ... and makes up to n calls. Binary search makes at most ceil(log2 n) calls. Here log2 n is the number of times n can be halved before reaching 1 (log2 8 = 3), and ceil rounds up to a whole number of calls. For n = 10 each probe leaves at most the larger half, so the window shrinks 10, 5, 3, 2, 1: four calls, and log2 10 = 3.32 rounds up to 4 (the first printed line shows these sizes).

The approach

Keep a window lo..hi with this invariant, in words: the first bad version is somewhere in lo..hi, both ends included. The window starts as 1..n, which is valid when version n is bad (the problem's premise: a precondition, meaning a fact the caller must guarantee before calling, so that the search is allowed to assume it). Probe mid:

• If mid is bad, then the first bad version is mid itself or earlier, so hi = mid. You cannot write mid - 1, because mid might be the answer.
• If mid is good, then every version up to mid is good, so lo = mid + 1.

The loop runs while lo < hi. When lo == hi the window has one version in it and the invariant says that is the first bad one, so no further call is needed. Each step shrinks the window (mid < hi, and mid + 1 > lo), so it terminates.

The code

from math import ceil, log2


def first_bad_version(n, is_bad_version):
    """Smallest v in 1..n with is_bad_version(v) True. Assumes version n is bad."""
    lo, hi = 1, n                       # invariant: the first bad version is in lo..hi
    while lo < hi:
        mid = lo + (hi - lo) // 2
        if is_bad_version(mid):
            hi = mid                    # mid is bad, so the first bad is mid or earlier
        else:
            lo = mid + 1                # mid is good, so every version up to mid is good
    return lo


def first_bad_or_none(n, is_bad_version):
    """Same search, but returns None when no version is bad (costs one extra API call)."""
    if n < 1 or not is_bad_version(n):
        return None
    return first_bad_version(n, is_bad_version)


class Api:
    """Fake API: versions first_bad..n are bad. Records every call."""
    def __init__(self, first_bad):
        self.first_bad, self.calls = first_bad, []

    def __call__(self, version):
        self.calls.append(version)
        return version >= self.first_bad


def mutant_first_bad(n, is_bad_version):
    """Wrong on purpose: 'hi = mid - 1' skips the answer when mid is the first bad version."""
    lo, hi = 1, n
    while lo < hi:
        mid = lo + (hi - lo) // 2
        if is_bad_version(mid):
            hi = mid - 1
        else:
            lo = mid + 1
    return lo


size, sizes = 10, [10]
while size > 1:
    size = (size + 1) // 2               # the larger half can remain, so round up
    sizes.append(size)
print("window sizes for n=10 when halving and rounding up:", sizes, "->", len(sizes) - 1,
      "halvings; log2(10) =", round(log2(10), 2), "and ceil =", ceil(log2(10)))

api = Api(first_bad=7)
print("n=10, first bad 7 ->", first_bad_version(10, api), "calls:", api.calls)

print("\nn     worst calls  allowed ceil(log2 n)")
for n in (1, 2, 3, 5, 8, 100, 1000):
    worst = 0
    for first in range(1, n + 1):
        api = Api(first)
        assert first_bad_version(n, api) == first
        assert len(set(api.calls)) == len(api.calls)    # never asks about the same version twice
        worst = max(worst, len(api.calls))
    print(f"{n:<6}{worst:<13}{ceil(log2(n))}")

checked = 0
for n in range(1, 130):
    for first in range(1, n + 1):
        api = Api(first)
        assert first_bad_version(n, api) == first and len(api.calls) <= ceil(log2(n))
        checked += 1
print("\nexhaustive check passed for", checked, "(n, first bad) pairs with n up to 129")

n = 2 ** 31 - 1
api = Api(first_bad=n - 3)
print("n = 2**31 - 1, first bad", n - 3, "->", first_bad_version(n, api), "in", len(api.calls), "calls")
api = Api(first_bad=1)
print("everything bad ->", first_bad_version(10, api), "after", len(api.calls), "calls")
api = Api(first_bad=11)
print("nothing bad (first_bad_version, wrong precondition) ->", first_bad_version(10, api))
api = Api(first_bad=11)
print("nothing bad (first_bad_or_none) ->", first_bad_or_none(10, api), "after", len(api.calls), "call")

caught = 0
for n in range(1, 30):
    for first in range(1, n + 1):
        caught += mutant_first_bad(n, Api(first)) != first
print("mutant with hi = mid - 1 is wrong on", caught, "pairs")

for n in range(1, 30):
    first = next((f for f in range(1, n + 1) if mutant_first_bad(n, Api(f)) != f), None)
    if first is not None:
        api = Api(first)
        print(f"smallest failing input: n={n}, first bad {first}: mutant returns {mutant_first_bad(n, api)}, "
              f"probes {api.calls}")
        break

The output below is the complete output of that program.

window sizes for n=10 when halving and rounding up: [10, 5, 3, 2, 1] -> 4 halvings; log2(10) = 3.32 and ceil = 4
n=10, first bad 7 -> 7 calls: [5, 8, 7, 6]

n     worst calls  allowed ceil(log2 n)
1     0            0
2     1            1
3     2            2
5     3            3
8     3            3
100   7            7
1000  10           10

exhaustive check passed for 8385 (n, first bad) pairs with n up to 129
n = 2**31 - 1, first bad 2147483644 -> 2147483644 in 31 calls
everything bad -> 1 after 4 calls
nothing bad (first_bad_version, wrong precondition) -> 10
nothing bad (first_bad_or_none) -> None after 1 call
mutant with hi = mid - 1 is wrong on 161 pairs
smallest failing input: n=3, first bad 2: mutant returns 1, probes [2]

Reading the trace for n = 10, first bad version 7

The calls were 5, 8, 7, 6:

1. lo=1, hi=10, probe 5: good, so lo=6.
2. lo=6, hi=10, probe 8: bad, so hi=8.
3. lo=6, hi=8, probe 7: bad, so hi=7.
4. lo=6, hi=7, probe 6: good, so lo=7.

Now lo == hi == 7 and the function returns 7 after 4 calls, which equals ceil(log2 10).

Complexity

• API calls: at most ceil(log2 n). The table above is measured from the code: the worst case over every possible first-bad position is 0, 1, 2, 3, 3, 7 and 10 calls for n = 1, 2, 3, 5, 8, 100 and 1000, equal to ceil(log2 n) each time. For n = 2^31 - 1 it took 31 calls. The program also asserts that no version is ever probed twice.
• Time and space: O(log n) time and O(1) extra space, because it is a loop with two integers.

Edge cases, tests and pitfalls

• One version (n = 1): the loop never runs and the answer is 1 with zero calls, relying on the premise that it is bad.
• First bad is version 1, or the last version: both are in the exhaustive check (every (n, first bad) pair for n up to 129, 8,385 pairs, each also asserting the call count).
• No bad version at all: first_bad_version assumes the premise and returns n (printed above for first_bad=11, which is the wrong answer to the wrong question). If the caller cannot guarantee a bad version, first_bad_or_none first asks about version n, one extra call, and returns None.
• Large n: lo + (hi - lo) // 2 avoids the lo + hi sum that overflows a 32-bit signed integer when n is near 2^31 - 1. Python would not overflow, but the habit carries over to C++ and Java.
• Off-by-one traps: hi = mid - 1 (the shipped mutant, wrong on 161 of the 435 pairs with n from 1 to 29; the last printed line shows the smallest failing input, where the probe lands exactly on the first bad version, the mutant then discards it and returns the wrong version), looping on lo <= hi while setting hi = mid (never ends when lo == hi), and returning lo - 1 or hi + 1.
• Never ask about the same version twice. When each call is slow, a repeated probe is wasted money; the loop never repeats one, and the program asserts it for every case in the table.

Running the code

docker run --rm -v "$PWD":/w -w /w python:3.12 python first_bad_version_demo.py

Save the program as first_bad_version_demo.py. It uses only the standard library, and the output above is from Python 3.12.

The idea in plain words

Binary search works on any question of the form "is length L good enough?" as long as the answer flips from no to yes exactly once and never flips back (monotone). The property "some substring of length L has at least K distinct characters" is monotone: if a window of length L has K distinct characters, then widening it by one character (when the string is long enough) still contains those K, so every longer length also works. So the smallest good L can be found by binary search on L, and each yes/no check is a fixed-size sliding window (a window of constant length moved one step at a time while a running count is updated) that costs O(n).

Brute force and why it is not enough

Try every start and every length and count distinct characters with set(...): O(n^3) naive, or O(n^2) with a running set per start. Trying lengths from 1 upward with a sliding window check is O(n^2). Binary search on the length cuts the number of checks from n to about log2 n.

Approach

1. If K is larger than the whole string's distinct count (checked by testing length n), return -1: no length works.
2. Otherwise the answer lies in [K, n], since at least K characters are needed to hold K distinct ones, and length n is known to work. Invariant: the answer is in [lo, hi] and hi works.
3. While lo < hi: mid = lo + (hi - lo) // 2. If a window of length mid has at least K distinct characters, the answer is <= mid, so hi = mid; otherwise lo = mid + 1.
4. Check by sliding window: keep a count per character and the number of characters with a non-zero count (distinct). Add the entering character; once the window is longer than mid, remove the leaving character; whenever the window is full and distinct >= K, answer yes.

Code, trace and example

Take s = "aabbcc" and K = 3. The only way to see three different letters is to span a boundary on both sides, so the shortest window is abbc, length 4. The program prints the binary search trace, the lengths that work, the impossible case, an empty string, and compares against brute force on every string over abc of length 0 to 8 and on 500 random strings. The last four lines show why the monotone requirement matters.

import random
from collections import defaultdict
from itertools import product


def window_has_k_distinct(s, length, k, trace=False):
    """Sliding window of fixed size: does any window of this length hold >= k distinct characters?"""
    counts = defaultdict(int)
    distinct = 0
    for i, ch in enumerate(s):
        counts[ch] += 1
        if counts[ch] == 1:
            distinct += 1
        if i >= length:                       # drop the character leaving the window
            out = s[i - length]
            counts[out] -= 1
            if counts[out] == 0:
                distinct -= 1
        if trace and i >= length - 1:
            print(f"      window {s[i - length + 1:i + 1]!r}: counts {dict((c, v) for c, v in counts.items() if v)}, distinct = {distinct}")
        if i >= length - 1 and distinct >= k:
            return True
    return False


def min_length_with_k_distinct(s, k, trace=None):
    """Smallest L such that some substring of length L has >= k distinct characters, else -1."""
    n = len(s)
    if k <= 0:
        return 0
    if k > n or not window_has_k_distinct(s, n, k):
        return -1                              # even the whole string is not enough
    lo, hi = k, n                              # invariant: the answer lies in [lo, hi]; hi always works
    while lo < hi:
        mid = lo + (hi - lo) // 2
        ok = window_has_k_distinct(s, mid, k)
        if trace is not None:
            trace.append(f"lo={lo} mid={mid} hi={hi} some window of length {mid} has >= {k} distinct: {ok}")
        if ok:
            hi = mid
        else:
            lo = mid + 1
    return lo


s, k = "aabbcc", 3
trace = []
print(f"s={s!r} k={k} -> {min_length_with_k_distinct(s, k, trace)}")
for line in trace:
    print("   ", line)
for length in (3, 4):
    print(f"  sliding window of length {length} over {s!r}:")
    print(f"    -> {window_has_k_distinct(s, length, k, trace=True)}")
print("window lengths 1..6 that work:", [L for L in range(1, 7) if window_has_k_distinct(s, L, k)])
print("s='aabb', k=3 ->", min_length_with_k_distinct("aabb", 3))
print("s='', k=1 ->", min_length_with_k_distinct("", 1))
print("s='z', k=1 ->", min_length_with_k_distinct("z", 1))


def brute(s, k):
    for length in range(max(k, 1), len(s) + 1):
        if any(len(set(s[i:i + length])) >= k for i in range(len(s) - length + 1)):
            return length
    return -1


for n in range(0, 9):
    for chars in product("abc", repeat=n):
        text = "".join(chars)
        for k in (1, 2, 3, 4):
            assert min_length_with_k_distinct(text, k) == brute(text, k), (text, k)
rng = random.Random(9)
for _ in range(500):
    text = "".join(rng.choice("abcdef") for _ in range(rng.randint(0, 40)))
    k = rng.randint(1, 6)
    assert min_length_with_k_distinct(text, k) == brute(text, k)
print("exhaustive strings over abc (length 0..8) and 500 random strings match brute force")


# The binary search needs a monotone check. "Exactly k distinct" is not monotone:
def window_has_exactly_k_distinct(s, length, k):
    return any(len(set(s[i:i + length])) == k for i in range(len(s) - length + 1))

print("exactly 2 distinct in 'abc':", [(L, window_has_exactly_k_distinct("abc", L, 2)) for L in (1, 2, 3)])
print("at least 2 distinct in 'abc':", [(L, window_has_k_distinct("abc", L, 2)) for L in (1, 2, 3)])

# The same binary search driven by the non-monotone check, on 'abcab' (lengths 1..5):
text = "abcab"
print("exactly 2 distinct in 'abcab':", [window_has_exactly_k_distinct(text, L, 2) for L in range(1, 6)])
lo, hi = 1, len(text)
while lo < hi:
    mid = lo + (hi - lo) // 2
    if window_has_exactly_k_distinct(text, mid, 2):
        hi = mid
    else:
        lo = mid + 1
print("binary search on that check returns", lo, "but the smallest length that works is 2")

Output:

s='aabbcc' k=3 -> 4
    lo=3 mid=4 hi=6 some window of length 4 has >= 3 distinct: True
    lo=3 mid=3 hi=4 some window of length 3 has >= 3 distinct: False
  sliding window of length 3 over 'aabbcc':
      window 'aab': counts {'a': 2, 'b': 1}, distinct = 2
      window 'abb': counts {'a': 1, 'b': 2}, distinct = 2
      window 'bbc': counts {'b': 2, 'c': 1}, distinct = 2
      window 'bcc': counts {'b': 1, 'c': 2}, distinct = 2
    -> False
  sliding window of length 4 over 'aabbcc':
      window 'aabb': counts {'a': 2, 'b': 2}, distinct = 2
      window 'abbc': counts {'a': 1, 'b': 2, 'c': 1}, distinct = 3
    -> True
window lengths 1..6 that work: [4, 5, 6]
s='aabb', k=3 -> -1
s='', k=1 -> -1
s='z', k=1 -> 1
exhaustive strings over abc (length 0..8) and 500 random strings match brute force
exactly 2 distinct in 'abc': [(1, False), (2, True), (3, False)]
at least 2 distinct in 'abc': [(1, False), (2, True), (3, True)]
exactly 2 distinct in 'abcab': [False, True, False, False, False]
binary search on that check returns 5 but the smallest length that works is 2

Reading the sliding window: with length 3 the window slides over aab, abb, bbc, bcc; each step adds the new character to the counts and, once the window is full, removes the one that left. distinct is the number of letters with a non-zero count, here 2 every time, so length 3 fails. With length 4 the first window aabb has 2 distinct letters and the next window abbc has counts a:1, b:2, c:1, so distinct = 3 and the check says yes at once. (defaultdict(int) is a dictionary that treats a missing letter as count 0.)

Reading the trace: the search starts with the range [3, 6]. Length 4 works, so hi = 4. Length 3 fails (every window of 3 letters, such as aab or bbc, has only two distinct letters), so lo = 4 and the answer is 4. The list [4, 5, 6] confirms the yes/no pattern is a block of no's followed by a block of yes's.

Complexity

Each check is O(n) time: one pass with O(1) updates to a dictionary of counts. The binary search makes at most ceil(log2(n - K + 1)) checks plus the one initial check at length n; each check costs at most n window steps, so the total is n * (log2(n - K + 1) + 1) steps, which is O(n log n). For aabbcc with K = 3: n - K + 1 = 4, ceil(log2 4) = 2 checks plus the initial one, 3 passes over 6 letters, and the trace shows exactly 2 search checks. Space is O(sigma), where sigma (the Greek letter, used as the name for the alphabet size) is the number of distinct characters that can appear, because the dictionary holds at most that many entries. For this particular property, a single two-pointer pass (grow the right end until K distinct, shrink from the left) gives O(n) directly. Binary search on the length earns its place when the check is easy but no simple greedy exists (a greedy method builds the answer by always taking the locally best step), or when the check is a black box, a function you can call for a yes or no but cannot look inside.

When the length is not monotone

"At least K distinct" works, and so does "at most K distinct" (longer windows can only add characters, so good lengths shrink the other way: binary search for the longest good length). "Exactly K distinct" does not: in abc with K = 2 the run shows lengths 1, 2, 3 giving no, yes, no, and in abcab the pattern is no, yes, no, no, no. A binary search on that predicate lands on 5, although length 2 works, so the property must be rephrased as "at least" or "at most" first.

Edge cases and common mistakes

• Empty string, or K larger than the number of distinct characters: -1, as the run shows for aabb with K = 3 and the empty string.
• K = 1 on a non-empty string: the answer is 1 (the search range is [1, n]).
• K <= 0: the code returns 0 because the empty window already qualifies; ask the interviewer whether that input is allowed.
• Off-by-one in the window: remove the leaving character before checking. Checking first looks at one character too many, and the brute-force comparison above fails on it (abbc with K = 3 returns 3 instead of 4).
• Setting lo = 1 instead of K is correct but wastes checks; setting hi = mid - 1 after a success discards the answer (aabbc with K = 3 returns 3 instead of 4).
• Forgetting the pre-check at length n leaves the search returning n for impossible inputs.

Running the code

Save as min_length_distinct.py and run with Python 3.12 (standard library only):

python3 min_length_distinct.py

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

The idea in plain words

A rotated sorted array is a sorted array whose front part was moved to the back, such as [4, 5, 6, 7, 0, 1, 2]. The rotation point is the one place where the values drop (here 7 is followed by 0). Binary search still works on it because at any probe mid, at least one of the two halves lo..mid or mid..hi is a plain sorted run. You find the sorted half, check whether the target lies inside its value range, and keep that half or discard it.

Here is the picture for [4, 5, 6, 7, 0, 1, 2] with lo=0, mid=3, hi=6:

index:  0  1  2  3 | 4  5  6
value:  4  5  6  7 | 0  1  2
        lo      mid         hi

The rotation point sits between index 3 and index 4, so it falls in the right half. The left half 4, 5, 6, 7 has no drop, so it is a plain sorted run: nums[lo] = 4 <= nums[mid] = 7. If the rotation point had fallen inside the left half, then nums[mid] would already be on the low side of the drop, below nums[lo]. So nums[lo] <= nums[mid] means "left half sorted", and nums[lo] > nums[mid] means "right half sorted".

With distinct values that test identifies the sorted half. With duplicates it can fail to: if nums[lo], nums[mid] and nums[hi] are all equal (a tie), you cannot tell where the rotation point is, so the comparison is ambiguous: it does not say which half is sorted. The only safe move is to give up a single element from each end, and when that keeps happening the search degrades to a scan: O(n) in the worst case.

The brute force is target in nums, O(n). The rotated binary search is O(log n) when ties are rare and O(n) in the worst case.

Why duplicates blur which half is sorted

[1, 0, 1, 1, 1] and [1, 1, 1, 0, 1] are two rotations of [0, 1, 1, 1, 1]. At the first probe (lo=0, mid=2, hi=4) both show 1, 1, 1: the three values the algorithm reads are identical, yet the 0 sits in the left half of one array and the right half of the other. Any rule that keeps only one half is wrong for one of them. Dropping nums[lo] and nums[hi] is safe, because both equal nums[mid], and nums[mid] is not the target (it was just compared), so neither dropped element could be the target.

When the three values are not all equal, the usual test is still correct. Suppose nums[lo] <= nums[mid] and yet the rotation point lay inside the left half. Then three things would hold at once:

1. nums[mid] sits after the drop, in the low run that continues to hi, so nums[mid] <= nums[hi].
2. Every value in that low run is at most every value before the drop, so nums[hi] <= nums[lo].
3. We assumed nums[lo] <= nums[mid].

Chaining them gives nums[lo] <= nums[mid] <= nums[hi] <= nums[lo], which forces all three values to be equal. That is the tie the code handles first, so outside a tie the rotation point cannot be in the left half and the left half is sorted. In the [4, 5, 6, 7, 0, 1, 2] picture the chain would need 4 <= 7 <= 2 <= 4, which is false, as it should be.

The code

import itertools
import random
from typing import List


def _search(nums: List[int], target: int, trace=None):
    """Return (found, loop steps)."""
    lo, hi = 0, len(nums) - 1
    steps = 0
    while lo <= hi:
        steps += 1
        mid = lo + (hi - lo) // 2
        if trace is not None:
            trace.append((lo, hi, mid, nums[lo], nums[mid], nums[hi]))
        if nums[mid] == target:
            return True, steps
        if nums[lo] == nums[mid] == nums[hi]:
            lo += 1                          # ambiguous: cannot tell which half is sorted.
            hi -= 1                          # Neither end is the target, so drop both.
        elif nums[lo] <= nums[mid]:          # left half lo..mid is sorted
            if nums[lo] <= target < nums[mid]:
                hi = mid - 1
            else:
                lo = mid + 1
        else:                                # right half mid..hi is sorted
            if nums[mid] < target <= nums[hi]:
                lo = mid + 1
            else:
                hi = mid - 1
    return False, steps


def search_rotated_with_duplicates(nums: List[int], target: int) -> bool:
    return _search(nums, target)[0]


def search_without_tie_rule(nums, target):
    """The textbook distinct-values version: it never handles nums[lo] == nums[mid] == nums[hi]."""
    lo, hi = 0, len(nums) - 1
    while lo <= hi:
        mid = lo + (hi - lo) // 2
        if nums[mid] == target:
            return True
        if nums[lo] <= nums[mid]:
            if nums[lo] <= target < nums[mid]:
                hi = mid - 1
            else:
                lo = mid + 1
        else:
            if nums[mid] < target <= nums[hi]:
                lo = mid + 1
            else:
                hi = mid - 1
    return False


def show(nums, target):
    trace = []
    found, steps = _search(nums, target, trace)
    print(f"{nums} target={target} -> {found} in {steps} steps")
    for lo, hi, mid, a, b, c in trace:
        print(f"  lo={lo} hi={hi} mid={mid}  nums[lo], nums[mid], nums[hi] = {a}, {b}, {c}")


# No ties: the three probe values show which half is sorted.
show([4, 5, 6, 7, 0, 1, 2], 0)
print()

# Two different rotations of [0, 1, 1, 1, 1] look identical at the first probe.
show([1, 0, 1, 1, 1], 0)
show([1, 1, 1, 0, 1], 0)
print()
show([2, 5, 6, 0, 0, 1, 2], 0)
show([2, 5, 6, 0, 0, 1, 2], 3)

# Exhaustive check against 'target in nums' on every rotation of every non-decreasing list.
print("without the tie rule:", search_without_tie_rule([1, 0, 1, 1, 1], 0),
      search_without_tie_rule([1, 1, 1, 0, 1], 0), "(both should be True)")
checked = 0
wrong_without_tie_rule = 0
for length in range(8):
    for combo in itertools.combinations_with_replacement(range(4), length):
        base = list(combo)
        for k in range(max(length, 1)):
            rotated = base[k:] + base[:k]
            for t in range(-1, 5):
                assert search_rotated_with_duplicates(rotated, t) == (t in rotated)
                wrong_without_tie_rule += search_without_tie_rule(rotated, t) != (t in rotated)
                checked += 1
print("\nexhaustive check passed:", checked, "cases;",
      "the version without the tie rule is wrong on", wrong_without_tie_rule)

print("\nworst case: n equal values plus one 0 hidden somewhere, searching for 0")
print("n     most steps over all rotations   steps on an all-equal list without 0   distinct values, most steps")
for n in (16, 64, 256, 1024):
    most = 0
    for pos in range(n):
        rotated = [1] * n
        rotated[pos] = 0
        most = max(most, _search(rotated, 0)[1])
    absent = _search([1] * n, 0)[1]
    distinct = 0
    for k in range(n):
        base = list(range(n))
        rotated = base[k:] + base[:k]
        for t in (0, n // 2, n - 1, n):
            distinct = max(distinct, _search(rotated, t)[1])
    print(f"{n:<6}{most:<32}{absent:<40}{distinct}")

print("\naverage steps over 2000 random cases (seed 5), n = 1024, random rotation, random target (often absent)")
rng = random.Random(5)
for label, top in (("values 0..3  ", 3), ("values 0..9999", 9999)):
    total = 0
    for _ in range(2000):
        base = sorted(rng.randint(0, top) for _ in range(1024))
        k = rng.randrange(1024)
        rotated = base[k:] + base[:k]
        total += _search(rotated, rng.randint(-1, top + 1))[1]
    print(f"  {label}: {total / 2000:.2f} steps on average")

The output below is the complete output of that program.

[4, 5, 6, 7, 0, 1, 2] target=0 -> True in 3 steps
  lo=0 hi=6 mid=3  nums[lo], nums[mid], nums[hi] = 4, 7, 2
  lo=4 hi=6 mid=5  nums[lo], nums[mid], nums[hi] = 0, 1, 2
  lo=4 hi=4 mid=4  nums[lo], nums[mid], nums[hi] = 0, 0, 0

[1, 0, 1, 1, 1] target=0 -> True in 3 steps
  lo=0 hi=4 mid=2  nums[lo], nums[mid], nums[hi] = 1, 1, 1
  lo=1 hi=3 mid=2  nums[lo], nums[mid], nums[hi] = 0, 1, 1
  lo=1 hi=1 mid=1  nums[lo], nums[mid], nums[hi] = 0, 0, 0
[1, 1, 1, 0, 1] target=0 -> True in 3 steps
  lo=0 hi=4 mid=2  nums[lo], nums[mid], nums[hi] = 1, 1, 1
  lo=1 hi=3 mid=2  nums[lo], nums[mid], nums[hi] = 1, 1, 0
  lo=3 hi=3 mid=3  nums[lo], nums[mid], nums[hi] = 0, 0, 0

[2, 5, 6, 0, 0, 1, 2] target=0 -> True in 1 steps
  lo=0 hi=6 mid=3  nums[lo], nums[mid], nums[hi] = 2, 0, 2
[2, 5, 6, 0, 0, 1, 2] target=3 -> False in 3 steps
  lo=0 hi=6 mid=3  nums[lo], nums[mid], nums[hi] = 2, 0, 2
  lo=0 hi=2 mid=1  nums[lo], nums[mid], nums[hi] = 2, 5, 6
  lo=0 hi=0 mid=0  nums[lo], nums[mid], nums[hi] = 2, 2, 2
without the tie rule: False True (both should be True)

exhaustive check passed: 11094 cases; the version without the tie rule is wrong on 84

worst case: n equal values plus one 0 hidden somewhere, searching for 0
n     most steps over all rotations   steps on an all-equal list without 0   distinct values, most steps
16    9                               8                                       5
64    33                              32                                      7
256   129                             128                                     9
1024  513                             512                                     11

average steps over 2000 random cases (seed 5), n = 1024, random rotation, random target (often absent)
  values 0..3  : 5.61 steps on average
  values 0..9999: 9.92 steps on average

Reading the traces

• [4, 5, 6, 7, 0, 1, 2], target 0: the first probe reads 4, 7, 2. There is no tie and 4 <= 7, so the left half is sorted; the target 0 is not in [4, 7), so lo=4. The second probe reads 0, 1, 2 at mid=5: the left half 0, 1 is sorted but 0 is in [0, 1), so hi=4, and the third probe finds it.
• [1, 0, 1, 1, 1], target 0: the first probe reads 1, 1, 1 (ambiguous), so lo=1, hi=3. The second probe reads 0, 1, 1: the left half is sorted (0 <= 1) and the target lies in [0, 1), so hi=1. The third probe finds the 0. Three steps.
• [1, 1, 1, 0, 1], target 0: the same ambiguous first step, then 1, 1, 0: the left half is judged sorted and the target is not in [1, 1), so lo=3, and the third step finds it.
• [2, 5, 6, 0, 0, 1, 2], target 0 is found at the first probe; for target 3 the search ends after three steps with False. The first probe reads 2, 0, 2: nums[mid] is smaller than nums[lo], so the right half is sorted and 3 is outside (0, 2], so it goes left.
• Dropping the tie rule breaks it: the textbook version returns False for the first array and True for the second, and it is wrong on 84 of the 11,094 cases in the exhaustive check, while the shipped version matches target in nums on all of them (every rotation of every non-decreasing list of length 0 to 7 over four values, targets -1 to 4).

Why the worst case is O(n), and the typical case is not

The table counts loop steps from the shipped function. An array of n equal values plus one hidden 0 (every one of the n possible positions tried) needs n/2 + 1 steps at worst (9, 33, 129, 513 for n = 16, 64, 256, 1024), and an all-equal list that has no 0 at all takes n/2 steps (8, 32, 128, 512). Each ambiguous step removes only two elements. Distinct values, by contrast, need only 5, 7, 9, 11 steps at worst (log2 n + 1).

This is not just a weakness of this code. Take n = 5. The arrays [0,1,1,1,1], [1,0,1,1,1], [1,1,0,1,1], [1,1,1,0,1] and [1,1,1,1,0] are all valid rotations of the same sorted list, and [1,1,1,1,1] is a valid input too. Picture an adversary: someone who chooses the array after seeing which positions you read, and answers every read with 1. If your method stops after reading only four of the five positions, the adversary still has two legal arrays to pick from, one with a 0 in the unread slot and one without, and the right answer for target 0 differs between them. So on the all-ones input any method that reads positions one at a time must read every position, which is n reads. No such method beats O(n) for this input family.

On random data with duplicates the average is far better: over 2,000 seeded cases with n = 1024, random rotations and random targets, arrays with values 0 to 3 averaged 5.61 steps and arrays with values 0 to 9999 averaged 9.92. The slow case needs long runs of equal values that sit exactly at lo, mid and hi. So the honest summary is: O(log n) typically, O(n) worst case, O(1) extra space.

Edge cases and common mistakes

• Empty list: the loop never runs, the result is False.
• One element: lo == mid == hi, so the answer comes from the first comparison, or the tie branch empties the window.
• All elements equal: only the tie branch runs, and the cost is n/2 steps.
• Not rotated at all (rotation by 0): nums[lo] <= nums[mid] holds throughout and it behaves as an ordinary binary search.
• Using nums[lo] < nums[mid] (strict) instead of <=: when the window has two elements (lo == mid) the left run is not recognised as sorted, and on [3, 1] with target 1 the search returns False.
• Shrinking only one end is also correct in the tie case, but dropping both ends halves the number of steps on an all-equal list.
• Skipping the tie rule: the examples above show exactly how it fails.

Running the code

docker run --rm -v "$PWD":/w -w /w python:3.12 python rotated_search_demo.py

Save the program as rotated_search_demo.py. It uses only the standard library, and the output above is from Python 3.12.