8. Merge Sorted Experiment Logs by Timestamp

Problem

A distributed training run writes its event log across several shards, one per worker process, each already sorted ascending by timestamp within itself. For debugging, an engineer needs one globally time-ordered log across all shards, and this has to run on every run's postmortem, so re-sorting the full concatenated log from scratch is wasteful when each shard is already sorted.

Given the shards, merge them into one list sorted ascending by timestamp. Ties (equal timestamps) break by shard order: entries from an earlier shard index come first. The whole thing must run in O(N log k) time using a heap over the k shard heads, not a full O(N log N) sort of the concatenated log.

Source: src/8_merge_sorted_experiment_logs.py

def merge_experiment_logs(log_shards: list[list[tuple[int, str]]]) -> list[tuple[int, str]]:
    ...

Examples:
>>> merge_experiment_logs([[(1, "loss=2.3"), (5, "loss=1.9")], [(2, "eval_start"), (5, "ckpt_saved")]])
[(1, "loss=2.3"), (2, "eval_start"), (5, "loss=1.9"), (5, "ckpt_saved")]

>>> merge_experiment_logs([[(0, "start")], [(1, "a")], [(2, "b")]])
[(0, "start"), (1, "a"), (2, "b")]

Step-by-Step Approach

1. Recognize the shape: k already-sorted lists that need to become one sorted list. A full sort of everything concatenated throws away the free ordering you already have within each shard.
2. Maintain a min-heap holding at most one "head" entry per shard — the next not-yet-emitted entry from that shard, keyed by (timestamp, shard_index) so ties resolve deterministically.
3. Seed the heap with the first entry of every non-empty shard.
4. Repeatedly pop the heap's minimum, append it to the result, and if that shard has another entry, push that shard's next entry onto the heap in its place.
5. Stop when the heap is empty — every entry from every shard has then been emitted in globally sorted order.
6. Handle edge cases: an empty list of shards, shards that are themselves empty, and a single shard (which should pass through unchanged).

The key insight is that the heap never holds more than k elements at once, so each of the N total entries costs only O(log k) to place, instead of O(log N) for a full sort — you're paying for how many streams you're interleaving, not how many entries exist across all of them.

Reference solution

import heapq


def merge_experiment_logs(log_shards: list[list[tuple[int, str]]]) -> list[tuple[int, str]]:
    # k-way merge: heap holds one head pointer per shard, O(N log k) time, O(k) space
    heap: list[tuple[int, int, int, str]] = []
    for shard_idx, shard in enumerate(log_shards):
        if shard:
            ts, msg = shard[0]
            # shard_idx as third field breaks timestamp ties by shard order
            heapq.heappush(heap, (ts, shard_idx, 0, msg))
    result: list[tuple[int, str]] = []
    while heap:
        ts, shard_idx, entry_idx, msg = heapq.heappop(heap)
        result.append((ts, msg))
        if entry_idx + 1 < len(log_shards[shard_idx]):
            next_ts, next_msg = log_shards[shard_idx][entry_idx + 1]
            heapq.heappush(heap, (next_ts, shard_idx, entry_idx + 1, next_msg))
    return result

Key Functions & Tricks

• heapq.heappush(heap, item) — push a new shard head, O(log k)
• heapq.heappop(heap) — pop the globally-next entry, O(log k)
• Bounded k-entry heap for k-way merge — heap size never exceeds the number of shards, regardless of total entry count
• (ts, shard_idx, entry_idx, msg) tuple key — timestamp first for ordering, shard_idx second to break ties deterministically
• Pointer-per-shard advancement — only push a shard's next entry after its current head is popped, keeping the heap bounded

How to Recognize This Pattern