41. Merge Sorted Pod Logs

Problem

Several service pods each emit their own timestamp-sorted stream of log entries. To debug a customer conversation that touched multiple pods, those k per-pod streams need to be merged into one globally time-sorted timeline.

Do this with recursive divide-and-conquer: split the k logs in half, recursively merge each half, then merge the two resulting sorted lists pairwise. A naive concatenate-then-sort would throw away the fact that each individual pod's stream is already ordered, so the intended solution keeps to O(N log k) time (N = total entries across all logs, k = number of logs) instead of the O(N log N) a full re-sort would cost.

Source: src/41_merge_sorted_pod_logs.py

def merge_sorted_logs(logs: list[list[tuple[float, str]]]) -> list[tuple[float, str]]:
    ...

>>> merge_sorted_logs([[(1.0, "a"), (3.0, "b")], [(2.0, "c")], [(0.5, "d"), (4.0, "e")]])
[(0.5, 'd'), (1.0, 'a'), (2.0, 'c'), (3.0, 'b'), (4.0, 'e')]

>>> merge_sorted_logs([])
[]

>>> merge_sorted_logs([[], [(1.0, "x")], []])
[(1.0, 'x')]

Step-by-Step Approach

1. Handle the base cases first: an empty list of logs merges to an empty result, and a single log is already merged (just return a copy of it).
2. Otherwise, split the k logs into two halves at the midpoint index.
3. Recursively call the merge function on each half. Each recursive call itself splits further until it bottoms out at zero or one logs, so this builds a balanced binary recursion tree of depth O(log k).
4. Once both halves come back as single sorted lists, merge those two sorted lists together with a standard two-pointer merge: walk both lists, always taking the smaller-timestamp head, and append whatever remains once one list is exhausted.
5. Return the merged result up the recursion, where it becomes one half of the next level's two-way merge.

The key insight is that merging is associative and a two-way merge of two sorted lists is O(n) — so recursively halving k lists and merging pairs on the way back up does the same total amount of two-way-merge work as flat pairwise merging, but in O(log k) levels instead of O(k) sequential merges, giving O(N log k) total instead of O(N * k).

Reference solution

def _merge_two(a: list[tuple[float, str]], b: list[tuple[float, str]]) -> list[tuple[float, str]]:
    result: list[tuple[float, str]] = []
    i = j = 0
    while i < len(a) and j < len(b):
        if a[i][0] <= b[j][0]:
            result.append(a[i])
            i += 1
        else:
            result.append(b[j])
            j += 1
    result.extend(a[i:])
    result.extend(b[j:])
    return result


def merge_sorted_logs(logs: list[list[tuple[float, str]]]) -> list[tuple[float, str]]:
    # recursive mergesort over the k logs, then pairwise 2-way merge, O(N log k) time
    if not logs:
        return []
    if len(logs) == 1:
        # shallow copy so callers can't mutate the original
        return list(logs[0])
    # floor division for the midpoint index
    mid = len(logs) // 2
    # slices allocate new lists, safe to recurse on
    left = merge_sorted_logs(logs[:mid])
    right = merge_sorted_logs(logs[mid:])
    return _merge_two(left, right)

Key Functions & Tricks

• len(logs) // 2 — floor division for the divide-step midpoint index.
• logs[:mid] / logs[mid:] — list slicing, allocates new lists so recursive calls can't mutate each other's input.
• list(logs[0]) — shallow copy so the caller never mutates the original log.
• result.append(a[i]) / result.extend(a[i:]) — append one element vs. bulk-copy the rest of a list once the other input is exhausted.
• a[i][0] <= b[j][0] — compares by tuple field 0 (timestamp), ignoring the message.
• Two-pointer merge — i/j track consumption of each list; smaller head advances, keeping output sorted in O(n).

How to Recognize This Pattern