Heaps, top-K, intervals and the sweep line

What it is

Two patterns that belong on one page because in several problems they are the same algorithm seen from two angles.

A heap is a partially ordered tree that gives you the minimum (or maximum) in $O(1)$ and insert or extract in $O(\log n)$, and gives you nothing else. It cannot search, it cannot tell you the second smallest without extracting the first, and its in-memory order is not sorted. Top-K is the pattern that uses one: keep a bounded heap of size $k$ and evict, so the cost is $O(n \log k)$ and the memory is $O(k)$ rather than $O(n)$.

Interval problems (merge overlapping ranges, insert a range, count overlaps) reduce to sorting by one endpoint and making a single pass. The sweep line is the general version: convert each interval into a start event and an end event, sort all events by time, and walk them while maintaining a running "active set". The heap shows up here because for questions like "which room does each meeting get", the active set has to be a priority queue keyed by end time.

Don't be confused: a heap is not a sorted structure. [1, 8, 2, 9, 10, 3] is a valid min-heap. The only guarantee is that each parent is no greater than its children, so the root is the global minimum and everything else is unordered relative to its siblings. Printing a heap looks like a bug and is not. If you want sorted output you must extract repeatedly, which is heapsort at $O(n \log n)$. The concrete consequence: never index into a heap expecting rank order, and note that top_k in the code below calls sorted() at the end precisely because the heap's internal order is not the answer.

Don't be confused: to find the k largest you use a min-heap, not a max-heap. This inverts in most people's heads. The heap holds the $k$ best candidates so far, and the operation you perform constantly is evicting the worst of them, so the worst must be at the top. A max-heap of size $k$ puts the wrong element within reach.

The problem it solves

"Give me the 10 highest scoring documents out of 50 million" cannot be answered by sorting; that materialises 50 million scored records to throw away all but ten. A bounded min-heap touches each record once, holds ten, and never allocates more. The complexity improvement from $O(n \log n)$ to $O(n \log k)$ is real but secondary. The memory improvement from $O(n)$ to $O(k)$ is what makes it work at all on a stream, and streams are where this pattern actually earns its keep.

Interval problems solve the "overlapping ranges" family that appears in booking systems, resource scheduling, IP address range consolidation, genomic feature sets and log time window analysis. The sweep line answers the harder version, peak concurrency, which is the question behind "how many servers do I need" and "how many meeting rooms".

Mechanics

"""Heaps, top-K, interval merging and the sweep line.

Runnable: `python3 heaps_and_intervals.py`. These sit together because both are
about processing items in an order that is not the input order, and in several
problems (meeting rooms) the heap IS the sweep line's active set.
"""

import heapq


# --- Top-K with a heap -------------------------------------------------------
def top_k(nums, k):
    """The k largest, using a MIN heap of size k.

    The counterintuitive part: to find the largest you keep the smallest at the
    top, because the top is the element you need to evict. Sorting is
    O(n log n) and O(n) space; this is O(n log k) and O(k) space, which is what
    lets it run over a stream you cannot hold in memory.
    """
    heap = []
    for n in nums:
        if len(heap) < k:
            heapq.heappush(heap, n)
        elif n > heap[0]:                 # beats the weakest survivor
            heapq.heapreplace(heap, n)    # one sift, not a pop plus a push
    return sorted(heap, reverse=True)


def k_closest_to(points, k, origin=(0, 0)):
    """Top-K where the comparison key is not the element. Push (key, item) and
    let the heap compare tuples. Negate the key to get a max-heap out of
    Python's min-heap, which is the standard workaround since heapq has no
    reverse option."""
    heap = []
    for x, y in points:
        d = (x - origin[0]) ** 2 + (y - origin[1]) ** 2   # no sqrt: monotonic
        if len(heap) < k:
            heapq.heappush(heap, (-d, (x, y)))
        elif -d > heap[0][0]:
            heapq.heapreplace(heap, (-d, (x, y)))
        # heap holds the k smallest distances, with the LARGEST of them on top
    return [p for _, p in sorted(heap, reverse=True)]


def merge_sorted_streams(streams):
    """K-way merge: the reason heaps exist in most production code. Holds one
    element per stream, so memory is O(k) regardless of stream length. This is
    the merge step of an LSM compaction and of an external sort."""
    heap = [(s[0], i, 0) for i, s in enumerate(streams) if s]
    heapq.heapify(heap)
    out = []
    while heap:
        val, si, idx = heapq.heappop(heap)
        out.append(val)
        if idx + 1 < len(streams[si]):
            heapq.heappush(heap, (streams[si][idx + 1], si, idx + 1))
    return out


# --- Intervals: merging ------------------------------------------------------
def merge_intervals(intervals):
    """Sort by start, then extend or emit. Sorting by START is what makes the
    single pass valid: once sorted, any interval that overlaps the one you are
    building must start before that one ends, so you only ever compare against
    the most recent output."""
    if not intervals:
        return []
    out = []
    for lo, hi in sorted(intervals):
        if out and lo <= out[-1][1]:      # <= merges touching intervals [1,2],[2,3]
            out[-1] = (out[-1][0], max(out[-1][1], hi))
        else:
            out.append((lo, hi))
    return out


def insert_interval(intervals, new):
    """Insert into an already-merged, sorted list. Three phases, in order:
    everything strictly before, everything overlapping (absorbed), everything
    strictly after. Writing it as three loops is clearer than one loop with
    flags, and it is what makes the boundaries checkable."""
    out, i, n = [], 0, len(intervals)
    lo, hi = new
    while i < n and intervals[i][1] < lo:
        out.append(intervals[i])
        i += 1
    while i < n and intervals[i][0] <= hi:
        lo = min(lo, intervals[i][0])
        hi = max(hi, intervals[i][1])
        i += 1
    out.append((lo, hi))
    out.extend(intervals[i:])
    return out


# --- Sweep line --------------------------------------------------------------
def max_overlap(intervals):
    """Peak concurrency: the classic "how many meeting rooms" question.

    Turn each interval into two EVENTS and sort them. The critical detail is the
    tie-break: an END at time t must be processed before a START at time t, or a
    meeting ending exactly when the next begins double-counts a room. Sorting
    (time, delta) with delta -1 for end and +1 for start gets this for free,
    since -1 sorts before +1.
    """
    events = []
    for lo, hi in intervals:
        events.append((lo, +1))
        events.append((hi, -1))
    events.sort()
    cur = best = 0
    at = None
    for t, delta in events:
        cur += delta
        if cur > best:
            best, at = cur, t
    return best, at


def min_rooms_with_heap(intervals):
    """The same answer via a heap, which is the form that also tells you WHICH
    room each meeting goes in. The heap holds the end time of every meeting
    currently running; its top is the soonest to free up."""
    ends = []
    assignment = []
    for lo, hi in sorted(intervals):
        if ends and ends[0][0] <= lo:
            _, room = heapq.heappop(ends)     # reuse the room that just freed
        else:
            room = len(ends)                  # need a new one
        heapq.heappush(ends, (hi, room))
        assignment.append(((lo, hi), room))
    return len(set(r for _, r in assignment)), assignment


if __name__ == "__main__":
    nums = [7, 2, 9, 4, 1, 8, 8, 3]
    print("nums              :", nums)
    print("top_k(3)          :", top_k(nums, 3))
    pts = [(1, 3), (-2, 2), (5, 8), (0, 1), (3, 3)]
    print("points            :", pts)
    print("k_closest_to(2)   :", k_closest_to(pts, 2))
    streams = [[1, 4, 9], [2, 3, 10], [5, 6, 7]]
    print("streams           :", streams)
    print("k-way merge       :", merge_sorted_streams(streams))

    print()
    iv = [(1, 3), (2, 6), (8, 10), (15, 18)]
    print("intervals         :", iv)
    print("merged            :", merge_intervals(iv))
    print("touching [1,2],[2,3] ->", merge_intervals([(1, 2), (2, 3)]))
    base = [(1, 3), (6, 9)]
    print("insert (2,5) into", base, "->", insert_interval(base, (2, 5)))

    print()
    meetings = [(0, 30), (5, 10), (15, 20), (25, 40), (28, 32)]
    print("meetings          :", meetings)
    peak, when = max_overlap(meetings)
    print("sweep line: peak  :", peak, "concurrent, first reached at t =", when)
    rooms, assign = min_rooms_with_heap(meetings)
    print("heap: rooms needed:", rooms)
    for m, r in assign:
        print("   meeting", m, "-> room", r)
    print("back-to-back (0,5),(5,10) needs:", max_overlap([(0, 5), (5, 10)])[0], "room")

Five details that separate a working implementation from a nearly working one.

heapreplace, not heappop then heappush. Both give the same result, but heapreplace performs one sift-down instead of a sift-up and a sift-down. In a hot loop over 50 million records that is a measurable difference, and knowing the API exists is a small signal that you have used heaps outside of interviews.

Push tuples (key, item), and negate the key when you need the other direction. Python's heapq is a min-heap only, with no key= parameter. The standard idiom is to push (-priority, item). The trap: if two keys tie, Python then compares the second tuple element, so the item type must be comparable or you get a TypeError at an unpredictable moment. The production fix is to push (key, counter, item) with a monotonic counter, which both breaks ties deterministically and guarantees the item is never compared.

k_closest_to compares squared distance and never calls sqrt. Square root is monotonic on non-negative inputs, so it cannot change the ordering, and skipping it removes a transcendental function from the inner loop along with the floating point error it introduces. This generalises: whenever you are ranking rather than reporting, strip any monotonic transform from the key.

merge_intervals sorts by start, and that choice is the correctness argument. Once sorted by start, any interval that overlaps the block you are currently building must begin before that block ends, so it suffices to compare against the single most recent output element. Sort by end instead and that property is gone and you would need to look further back. Say this rather than saying "we sort first".

The sweep line's tie-break decides the answer. Events are (time, delta) with +1 for a start and -1 for an end, and the sort puts -1 before +1 at equal times because $-1 < +1$. That means a meeting ending at 10 is processed before one starting at 10, so back-to-back meetings share a room. If your problem treats an interval as closed on both ends, so touching intervals conflict, flip the encoding. This is a one-character change that silently produces an off-by-one in the final answer, and interviewers test it with a back-to-back input.

Worked example

nums              : [7, 2, 9, 4, 1, 8, 8, 3]
top_k(3)          : [9, 8, 8]
points            : [(1, 3), (-2, 2), (5, 8), (0, 1), (3, 3)]
k_closest_to(2)   : [(0, 1), (-2, 2)]
streams           : [[1, 4, 9], [2, 3, 10], [5, 6, 7]]
k-way merge       : [1, 2, 3, 4, 5, 6, 7, 9, 10]

intervals         : [(1, 3), (2, 6), (8, 10), (15, 18)]
merged            : [(1, 6), (8, 10), (15, 18)]
touching [1,2],[2,3] -> [(1, 3)]
insert (2,5) into [(1, 3), (6, 9)] -> [(1, 5), (6, 9)]

meetings          : [(0, 30), (5, 10), (15, 20), (25, 40), (28, 32)]
sweep line: peak  : 3 concurrent, first reached at t = 28
heap: rooms needed: 3
   meeting (0, 30) -> room 0
   meeting (5, 10) -> room 1
   meeting (15, 20) -> room 1
   meeting (25, 40) -> room 1
   meeting (28, 32) -> room 2
back-to-back (0,5),(5,10) needs: 1 room

top_k returning [9, 8, 8] keeps the duplicate, which is correct and is worth checking deliberately, because an implementation built on a set instead of a heap returns [9, 8, 7] and looks plausible. If the problem wants distinct values, that is a different problem and you should ask.

k_closest_to(pts, 2) returns [(0,1), (-2,2)]. The squared distances are $(0,1) \to 1$, $(-2,2) \to 8$, $(1,3) \to 10$, $(3,3) \to 18$, $(5,8) \to 89$, so the two nearest are right. Notice the output is ordered farthest-first inside the sorted call because the heap stores negated keys; the final sorted(heap, reverse=True) on negated keys yields ascending true distance. Sign conventions in a negated heap are the most common source of a silently reversed result, so verify with a hand-checked input like this one every time.

The two meeting-room answers agree at 3, and they are not the same computation. The sweep line answers "what is the peak", and it tells you the peak is first reached at $t = 28$: at that moment (0,30), (25,40) and (28,32) are all running. It does not and cannot tell you which room anything goes in. The heap version answers the assignment question and the room count falls out of it. Pick the sweep line when you need the number and the heap when you need the schedule, and say which the problem asked for.

Read the assignment output closely: room 1 is reused three times, by (5,10), then (15,20), then (25,40). That is the heap doing its job. (15,20) starts at 15, the soonest-ending active meeting is (5,10) which ended at 10, so its room is free and gets reused rather than allocating a fourth. A greedy assignment that always allocated a new room would report 5.

back-to-back (0,5),(5,10) needs 1 room is the tie-break test described above. If the event sort had put +1 before -1, this would print 2, and every other number in the output would still look correct.

Production evidence

Lucene's top-K retrieval. Lucene collects hits into a bounded priority queue (HitQueue, a min-heap of size $k$) and discards any document whose score does not beat the queue's current worst. This is why asking for the top 10 of a 100-million-document index does not allocate 100 million results, and it is the direct production form of top_k. It also enables the WAND and block-max optimisations, which skip entire posting-list blocks whose maximum possible score cannot beat the heap's current threshold.

LSM-tree compaction is a k-way merge. RocksDB and LevelDB compact by opening an iterator over each input SSTable and merging them with a heap that holds one key per input, emitting in sorted order. Memory stays $O(\text{number of files})$ regardless of how large the files are, which is the property that makes compacting terabyte levels possible. External merge sort in a database's sort operator is the same algorithm.

Dijkstra's algorithm. The priority queue is the heap, and the choice of heap changes the published complexity: a binary heap gives $O((V+E) \log V)$, a Fibonacci heap gives $O(E + V \log V)$. In practice the binary heap almost always wins because the Fibonacci heap's constant factors and cache behaviour are poor, which is a useful example of an asymptotically better structure losing in production.

Sweep line in computational geometry. The Bentley-Ottmann algorithm finds all intersections among $n$ line segments in $O((n+k) \log n)$ for $k$ intersections by sweeping a vertical line and maintaining the active segments in a balanced structure ordered by their position along the sweep. It is the same event-sorted, active-set-maintaining shape as max_overlap, generalised from one dimension to two.

The debate

Heap or sorted()[:k] for top-K? For a list already in memory where $k$ is a significant fraction of $n$, just sort. The constant factor on a highly optimised sort (Python's Timsort is C code that exploits existing runs) beats a heap loop in the interpreter, and the code is one line. The heap wins decisively in exactly two situations: $k \ll n$, and the data is a stream that does not fit in memory. My position: state the crossover instead of asserting a winner. "If $k$ is small relative to $n$ or the input is a stream, bounded heap at $O(n \log k)$ and $O(k)$ space. If everything is in memory and $k$ is within an order of magnitude of $n$, sort and slice, because the constants dominate." A third option worth naming: heapq.nlargest in the standard library does exactly the bounded-heap thing, and Quickselect gives expected $O(n)$ if you need the top $k$ unordered and can tolerate a worst case of $O(n^2)$ or the complexity of median-of-medians.

Sweep line or heap for meeting rooms? They are $O(n \log n)$ either way, dominated by the sort. Take the sweep line when the question is a count, because the code is shorter and the events generalise (add a +1 for a start and a -1 for an end and you can mix in other event types for free). Take the heap when you need the assignment, or when intervals arrive online and you cannot sort up front. The sweep line has one property that decides several real cases: it extends to weighted events at no cost. Replace the deltas with arbitrary numbers and "peak concurrent connections" becomes "peak bandwidth" with no other change, which the heap version cannot do.

When are intervals the wrong model entirely? When ranges are dense over a small coordinate space, a difference array or a bitmap over the coordinate space is simpler and faster than sorting intervals, as covered under prefix sums. And when queries are interleaved with updates rather than batched, you want an interval tree or a segment tree, because re-sorting per query is $O(n \log n)$ each time. The honest interview answer names the batch-versus-online distinction as the deciding variable, rather than defending one structure.

Follow-up Q&A

Why is heapify $O(n)$ and not $O(n \log n)$? Because the cost is dominated by the nodes near the leaves, and there are many of those but each sifts down a short distance. At height $h$ from the bottom there are about $n/2^{h+1}$ nodes, each costing $O(h)$, and $\sum_h h/2^{h+1}$ converges to a constant. So building a heap from an existing list is linear, and it is strictly better than $n$ successive pushes. That is why merge_sorted_streams builds the initial list and calls heapify rather than pushing in a loop.

How do you delete an arbitrary element from a heap? You cannot, in $O(\log n)$, without an auxiliary index from element to heap position, which is a real amount of bookkeeping and must be maintained through every sift. The usual production answer is lazy deletion: mark the element dead in a side set, leave it in the heap, and discard it when it surfaces at the top. The cost is that the heap can grow with dead entries, so you either bound it or rebuild when the dead fraction crosses a threshold. This exact pattern is how a sliding window maximum with a heap works, and knowing to reach for lazy deletion rather than claiming an $O(\log n)$ arbitrary delete is the signal.

Your merge_intervals merges [1,2] and [2,3] into [1,3]. Is that right? It depends on whether the intervals are closed or half open, and this is a question to ask rather than assume. For closed intervals representing occupancy, they touch and merging is right, which is why the comparison is lo <= out[-1][1]. For half-open intervals [1,2) and [2,3) they do not overlap, and the comparison should be <. The output line in the run above exists to make the chosen convention visible instead of implicit. In an interview, state the convention before you write the comparison; it takes four seconds and it is a thing interviewers deliberately leave ambiguous.

How would you compute peak concurrency over a billion events that do not fit in memory? The sweep line needs sorted events, and external merge sort handles that in $O(n \log n)$ I/O-bounded passes. But usually you do not need exact event ordering: bucket events by time granularity (per second, per minute), which turns the problem into a difference array over buckets, and that is one streaming pass with $O(\text{buckets})$ memory and is trivially parallel by time range. Give up exactness at sub-bucket resolution and the problem becomes easy. Naming that tradeoff explicitly is better than describing a distributed sort.

Two heaps for a running median: how? Keep a max-heap of the lower half and a min-heap of the upper half, rebalanced after each insert so their sizes differ by at most one. The median is then the top of the larger heap, or the mean of the two tops when the sizes are equal. Each insert is $O(\log n)$. The bug to avoid is rebalancing by size only without checking the ordering invariant across the two heaps, since a new element can belong on the other side; insert by comparing against a top first, then rebalance.

In min_rooms_with_heap, why push (hi, room) rather than just hi? Because the room identity has to travel with the end time, and the heap is ordered by the first tuple element. If two meetings end at the same time, Python compares the second element, which is an integer room number and is therefore safely comparable. Had the payload been a non-comparable object, this would raise a TypeError only on the input where a tie occurs, which is the sort of bug that reaches production.

Common misconceptions

"A heap gives me the k largest in sorted order." It gives you them in heap order. top_k sorts before returning, and that final sort is $O(k \log k)$, which is negligible but is not free and is not automatic.

"heapq has a max-heap." It does not. Negate the key, or use heapq._heapify_max and accept that you are calling a private function. Negation fails on non-numeric keys, which is when you write a wrapper class with __lt__ reversed.

"Sorting by start is arbitrary; I could sort by end." For merging, sorting by start is what makes the one-comparison-back pass valid. Sorting by end is correct for a different problem, the activity selection or "maximum non-overlapping intervals" greedy, where you want to finish as early as possible to leave room. Two interval problems, two different sort keys, and using the wrong one gives a wrong answer rather than a slow one.

"The sweep line needs the intervals sorted." It needs the events sorted, which is twice as many items and a different ordering. Sorting the intervals and then generating events in that order does not produce sorted events, because an early-starting interval can end after a later one starts.

Interview delivery note

The sentence for top-K: "I will keep a min-heap of size k, so the top is the weakest survivor and eviction is $O(\log k)$. That gives $O(n \log k)$ time and, more importantly, $O(k)$ memory, so it works on a stream I cannot hold." Leading with the memory bound rather than the time bound is what marks someone who has actually run this at scale, because at scale the memory is the reason.

The sentence for intervals: "Sort by start. After that, anything overlapping the block I am building must start before that block ends, so I only ever compare against the last output element." State the invariant, then write four lines.

The senior-to-staff separator here is noticing that the two meeting-room formulations answer different questions and asking which one is wanted before coding. A senior candidate implements one and it is usually right. A staff candidate says: "If you want the number of rooms, that is a sweep line and it also gives me the peak time for free, and it generalises if events are weighted. If you want each meeting assigned to a room, that is a heap keyed on end time. Which do you need?" Turning an ambiguous requirement into an explicit choice, before writing code that has to be thrown away, is the behaviour the round is actually measuring.

The Staff & Team Lead Interview Reference