EF-T12: Merge timestamped streams

Exercise contract#

Merge k finite streams of records that are each sorted by nondecreasing signed integer timestamp. Return payloads ordered first by timestamp, then by source index, then by ordinal within that source. Streams and individual sources may be empty. Reject an input source whose timestamps decrease. Payload strings are opaque and do not participate in ordering.

Constraints

• 0 <= k <= 10,000 and total records N <= 200,000.
• Timestamps fit a signed 64-bit integer.
• Each source's record order defines ordinal 0,1,2,... and is preserved for equal timestamps.
• Payloads are returned unchanged; duplicate payload values are allowed.

Examples

[[(1,"a"),(4,"b")],[(1,"c"),(3,"d")]]

["a","c","d","b"]The timestamp-1 tie goes to source 0, then each pop exposes that source's next record.

[[],[(2,"only")],[]]

["only"]Empty sources add no frontier entry.

Derive the approach#

Baseline: Concatenate all N records with source and ordinal metadata, then sort the combined list by the three-key contract. This costs O(N log N) comparisons and stores all decorated records.

1. The smallest unconsumed record in any source must be at that source's frontier because every source is already sorted.
2. Put the first record from each nonempty source into a min-heap keyed by (timestamp, source index, ordinal).
3. Pop the smallest frontier, append its payload, then push the next record from the same source if one exists.
4. The heap never holds more than one record per source, and the complete key implements deterministic ties without comparing payloads.

Time

O(N log k)

Space

O(k) auxiliary heap space plus O(N) returned payload references

Assumptions

N is total records, k is the number of sources, each source has been validated in O(N), timestamp/source/ordinal comparison is constant-time, and payload copying is not part of comparison.

Three-stage hint ladder#

Reveal only the smallest useful step. Any revealed hint should be recorded as assistance.

Reference solutions#

Open these after an attempt or when working in Learn mode. Viewing either solution is not independent-success evidence.

import heapq

def merge_streams(streams):
    for stream in streams:
        for index in range(1, len(stream)):
            if stream[index][0] < stream[index - 1][0]:
                raise ValueError("each stream must be timestamp-sorted")
    frontier = []
    for source, stream in enumerate(streams):
        if stream:
            timestamp, payload = stream[0]
            heapq.heappush(frontier, (timestamp, source, 0, payload))
    output = []
    while frontier:
        _, source, ordinal, payload = heapq.heappop(frontier)
        output.append(payload)
        next_ordinal = ordinal + 1
        if next_ordinal < len(streams[source]):
            timestamp, next_payload = streams[source][next_ordinal]
            heapq.heappush(frontier, (timestamp, source, next_ordinal, next_payload))
    return output

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.PriorityQueue;

record StreamRecord(long timestamp, String payload) {}
record Cursor(long timestamp, int source, int ordinal, String payload) {}

static List<String> mergeStreams(List<List<StreamRecord>> streams) {
    for (List<StreamRecord> stream : streams) {
        for (int index = 1; index < stream.size(); index++) if (stream.get(index).timestamp() < stream.get(index - 1).timestamp()) throw new IllegalArgumentException("stream not sorted");
    }
    PriorityQueue<Cursor> heap = new PriorityQueue<>(Comparator.comparingLong(Cursor::timestamp).thenComparingInt(Cursor::source).thenComparingInt(Cursor::ordinal));
    for (int source = 0; source < streams.size(); source++) if (!streams.get(source).isEmpty()) {
        StreamRecord record = streams.get(source).get(0);
        heap.offer(new Cursor(record.timestamp(), source, 0, record.payload()));
    }
    List<String> output = new ArrayList<>();
    while (!heap.isEmpty()) {
        Cursor current = heap.poll();
        output.add(current.payload());
        int nextOrdinal = current.ordinal() + 1;
        if (nextOrdinal < streams.get(current.source()).size()) {
            StreamRecord next = streams.get(current.source()).get(nextOrdinal);
            heap.offer(new Cursor(next.timestamp(), current.source(), nextOrdinal, next.payload()));
        }
    }
    return output;
}

Deterministic test plan#

Transfer and follow-ups#

• Return decorated records instead of payloads while keeping the same tie policy.
• Discuss a lazy iterator form and what ownership/lifetime contract it would need.

State why one candidate per source is sufficient and include every observable tie-breaker in the heap key.

Authorship and review#

Original instructional record around the standard k-way heap merge; it is not production-streaming guidance or an employer-derived task.

Engineering Foundry original · Codex self-review on 2026-09-05 · independent human editorial review is not claimed.

Evidence IDs: RA-DSA-CURRICULUM-EXPANSION-2026-09