Heaps & priority queues: cheap access to the extreme, nothing more

Sometimes you do not need the data sorted — you only ever need the smallest (or largest) element right now, repeatedly, while still inserting. A binary heap gives you that extreme in O(1) and maintains itself in O(log n) per update. That is the entire value proposition.

Why this data structure

Sorting fully is O(n log n) and overkill; scanning for the min is O(n) each time. A binary heap gives you the extreme in O(1) and repairs itself in O(log n) per insert or remove. That is the whole pitch: a "priority queue" where pop always returns the highest-priority item.

What a binary heap is (first principles)

A complete binary tree (every level full except possibly the last, filled left to right) obeying the heap invariant:

MIN-HEAP: every parent <= both its children.
(so the global minimum is necessarily at the root)

This is weaker than a BST: there is no left/right ordering between siblings, only the parent-child relation. That weaker promise is exactly why inserts are cheap — you repair only one root-to-leaf path, not the whole structure.

The array representation — the trick that makes it fast

Because the tree is complete, you can store it in a flat array with no pointers, using index arithmetic (0-based):

node at index i:
    parent      = (i - 1) // 2
    left child  = 2*i + 1
    right child = 2*i + 2

array:  [1, 3, 2, 7, 4, 9]
tree:        1
            / \
           3   2
          / \  /
         7  4 9

Contiguous memory means cache-friendly access and no per-node allocation.

Why push and pop are O(log n)

• push: append to the end, then "sift up" — swap with the parent while it is smaller, walking at most the tree height = O(log n).
• pop: the min is at index 0. Move the last element to the root, shrink, then "sift down" — swap with the smaller child while it is bigger, again at most the height = O(log n).

Both touch a single root-to-leaf path; a complete tree's height is floor(log2 n).

Python's heapq — the practical API

heapq is a min-heap operating on an ordinary list in place:

There is no max-heap. Two standard workarounds:

1. Negate values: push -x; the smallest -x is the largest x.
2. Tuples for priority: the heap orders by the tuple lexicographically, so (priority, payload) pops the lowest priority first.

Kth largest element (LeetCode 215)

Keep a min-heap of size k holding the k largest seen so far. The heap's root is the smallest of those k — i.e. exactly the kth largest overall once all elements are processed. Push each number; if the heap exceeds k, pop the smallest (it cannot be in the top k).

Why a min-heap for "largest"? Because the element most at risk of eviction from the top-k club is the smallest of the club, and a min-heap exposes that in O(1).

def kth_largest(nums: List[int], k: int) -> int:
    heap: List[int] = []
    for x in nums:
        heapq.heappush(heap, x)
        if len(heap) > k:
            heapq.heappop(heap)      # drop the smallest, keep k largest
    return heap[0]                   # smallest of the k largest = kth largest

Top k frequent elements (LeetCode 347)

Count frequencies in O(n), then keep a size-k min-heap keyed by frequency. The least-frequent of the current top k sits at the root and is evicted first.

def top_k_frequent(nums: List[int], k: int) -> List[int]:
    counts = Counter(nums)
    heap: List = []                  # (freq, value)
    for value, freq in counts.items():
        heapq.heappush(heap, (freq, value))
        if len(heap) > k:
            heapq.heappop(heap)      # remove the least frequent so far
    return [value for (freq, value) in heap]

Merge k sorted lists (LeetCode 23) — where the counter trick is mandatory

The next element of the merged output is the smallest among the current fronts of the k lists. A heap of those k fronts gives it in O(log k). Pop the smallest, append it, and push the next element from the list it came from.

Here the counter trick is not optional. We store (value, counter, list_index, elem_index). If two values tie, Python next compares counters — which are unique — so it never tries to compare the payload indices in a way that matters, and never compares non-comparable payloads.

def merge_k_sorted(lists: List[List[int]]) -> List[int]:
    heap: List = []
    counter = itertools.count()      # unique, increasing tie-breaker
    # seed with the first element of each non-empty list
    for li, lst in enumerate(lists):
        if lst:
            heapq.heappush(heap, (lst[0], next(counter), li, 0))
    out: List[int] = []
    while heap:
        val, _, li, ei = heapq.heappop(heap)
        out.append(val)
        if ei + 1 < len(lists[li]):                  # more in that list?
            nxt = lists[li][ei + 1]
            heapq.heappush(heap, (nxt, next(counter), li, ei + 1))
    return out

Find median from a data stream (LeetCode 295) — the two-heap pattern

Split the numbers into a lower half and an upper half:

• lo = max-heap of the smaller half (root = largest of the small ones)
• hi = min-heap of the larger half (root = smallest of the big ones)

Keep them balanced (sizes differ by at most 1). Then the median is either the root of the bigger heap (odd count) or the average of the two roots (even count) — both O(1). Python has only a min-heap, so the max-heap lo stores negated values: the smallest negative is the largest original.

class MedianFinder:
    def __init__(self) -> None:
        self.lo: List[int] = []   # max-heap via negation (smaller half)
        self.hi: List[int] = []   # min-heap (larger half)

    def add_num(self, num: int) -> None:
        # push onto lo (as a negative), then move lo's max over to hi so that
        # lo's max stays <= hi's min.
        heapq.heappush(self.lo, -num)
        heapq.heappush(self.hi, -heapq.heappop(self.lo))
        # rebalance: lo is allowed to be at most one larger than hi.
        if len(self.hi) > len(self.lo):
            heapq.heappush(self.lo, -heapq.heappop(self.hi))

    def find_median(self) -> float:
        if len(self.lo) > len(self.hi):
            return float(-self.lo[0])            # odd count -> lo has the extra
        return (-self.lo[0] + self.hi[0]) / 2.0  # even -> average of roots

Meeting rooms II — heap of end times (LeetCode 253)

Sort meetings by start time. Maintain a min-heap of the end times of meetings currently occupying a room. For each new meeting, if the earliest-ending current meeting (the heap root) is already over by the time the new one starts, that room frees up — reuse it (pop). Otherwise we need a new room. The heap size at its peak is the answer.

Why a heap of end times? Among all in-progress meetings, the only one that can possibly free up for the next meeting is the one ending soonest — exactly the min-heap root in O(1).

def min_meeting_rooms(intervals: List[List[int]]) -> int:
    if not intervals:
        return 0
    intervals = sorted(intervals, key=lambda iv: iv[0])
    end_times: List[int] = []          # min-heap of end times
    for start, end in intervals:
        if end_times and end_times[0] <= start:
            heapq.heappop(end_times)   # earliest-ending room is free; reuse
        heapq.heappush(end_times, end)
    return len(end_times)              # rooms in simultaneous use at the peak

Dijkstra preview — heaps power weighted shortest paths

A teaser that connects this lesson to the next: a heap lets you always expand the closest-so-far node next. graph[u] is a list of (neighbor, weight); a min-heap keyed by tentative distance always hands us the closest unfinalized node, the greedy choice Dijkstra relies on. The full graph framing lives in lesson 12.

def dijkstra_preview(graph: dict, source: int) -> dict:
    dist = {source: 0}
    heap = [(0, source)]               # (distance_so_far, node)
    while heap:
        d, u = heapq.heappop(heap)
        if d > dist.get(u, float("inf")):
            continue                   # stale entry; a better one was processed
        for v, w in graph.get(u, []):
            nd = d + w
            if nd < dist.get(v, float("inf")):
                dist[v] = nd
                heapq.heappush(heap, (nd, v))
    return dist