Heaps & Priority Queues

What a heap is

A binary heap is a complete binary tree stored in an array where every parent is ≤ its children (min-heap) or ≥ (max-heap). So the root is always the minimum (or maximum).

Array layout (0-indexed): children of i are 2i + 1 and 2i + 2; parent is (i − 1) // 2.

Using heaps in your language

import heapq

h = []
heapq.heappush(h, 5)
heapq.heappush(h, 1)
smallest = h[0]             # peek
heapq.heappop(h)            # pop min
heapq.heapify(arr)          # O(n) in place
heapq.heappushpop(h, x)     # push then pop (efficient)
heapq.heapreplace(h, x)     # pop then push
heapq.nlargest(k, arr)      # convenience, O(n log k)

# max-heap: negate values
heapq.heappush(h, -x); largest = -h[0]

# tuples compare lexicographically: (priority, tie_breaker, item)
heapq.heappush(h, (dist, node))

#include <queue>

priority_queue<int> maxHeap;                              // max-heap by default!
priority_queue<int, vector<int>, greater<int>> minHeap;   // min-heap
maxHeap.push(5); maxHeap.top(); maxHeap.pop();

// pairs: compared by first, then second
priority_queue<pair<int,int>, vector<pair<int,int>>, greater<>> pq;

// custom comparator (returns true if a has LOWER priority than b)
auto cmp = [](const Node& a, const Node& b) { return a.cost > b.cost; };  // min by cost
priority_queue<Node, vector<Node>, decltype(cmp)> pq2(cmp);

How it works inside

class MinHeap:
    def __init__(self):
        self.a = []

    def push(self, x):
        a = self.a
        a.append(x)
        i = len(a) - 1
        while i > 0 and a[(i - 1) // 2] > a[i]:          # sift up
            p = (i - 1) // 2
            a[i], a[p] = a[p], a[i]
            i = p

    def pop(self):
        a = self.a
        top = a[0]
        last = a.pop()
        if a:
            a[0] = last
            i, n = 0, len(a)
            while True:                                   # sift down
                l, r, m = 2 * i + 1, 2 * i + 2, i
                if l < n and a[l] < a[m]: m = l
                if r < n and a[r] < a[m]: m = r
                if m == i: break
                a[i], a[m] = a[m], a[i]
                i = m
        return top

Heapify is O(n) (not O(n log n)) because most nodes are near the bottom and sift down only a little.

Pattern 1: Top K with a bounded heap

To keep the K largest, use a min-heap of size K: the root is the smallest of the top K, and anything smaller than it can be discarded.

import heapq

def k_largest(nums, k):
    h = []
    for x in nums:
        heapq.heappush(h, x)
        if len(h) > k:
            heapq.heappop(h)        # drop the smallest
    return h                         # h[0] is the k-th largest

int findKthLargest(vector<int>& nums, int k) {
    priority_queue<int, vector<int>, greater<int>> h;
    for (int x : nums) {
        h.push(x);
        if ((int)h.size() > k) h.pop();
    }
    return h.top();
}

O(n log k) time, O(k) space. The rule of thumb: K largest → min-heap; K smallest → max-heap (the opposite of what you might expect).

Top K frequent

from collections import Counter
import heapq

def top_k_frequent(nums, k):
    count = Counter(nums)
    return heapq.nlargest(k, count.keys(), key=count.get)

# O(n) bucket sort alternative: frequencies are bounded by n
def top_k_frequent_bucket(nums, k):
    count = Counter(nums)
    buckets = [[] for _ in range(len(nums) + 1)]
    for x, c in count.items():
        buckets[c].append(x)
    res = []
    for c in range(len(buckets) - 1, 0, -1):
        res.extend(buckets[c])
        if len(res) >= k:
            return res[:k]

Pattern 2: K-way merge

Merge K sorted sequences by keeping one candidate per sequence in a heap.

import heapq

def merge_k_lists(lists):
    h = []
    for i, node in enumerate(lists):
        if node:
            heapq.heappush(h, (node.val, i, node))   # i breaks ties (nodes aren't comparable)
    dummy = tail = ListNode()
    while h:
        _, i, node = heapq.heappop(h)
        tail.next = tail = node
        if node.next:
            heapq.heappush(h, (node.next.val, i, node.next))
    return dummy.next

ListNode* mergeKLists(vector<ListNode*>& lists) {
    auto cmp = [](ListNode* a, ListNode* b) { return a->val > b->val; };
    priority_queue<ListNode*, vector<ListNode*>, decltype(cmp)> pq(cmp);
    for (auto l : lists) if (l) pq.push(l);
    ListNode dummy, *tail = &dummy;
    while (!pq.empty()) {
        ListNode* n = pq.top(); pq.pop();
        tail->next = n; tail = n;
        if (n->next) pq.push(n->next);
    }
    return dummy.next;
}

O(N log K) for N total elements. The same idea generates the k smallest pair sums or the k-th smallest in a sorted matrix: treat each row as a sorted list and lazily push the next candidate.

Pattern 3: Two heaps (running median)

Keep the smaller half in a max-heap and the larger half in a min-heap, sizes balanced within 1. The median is at the tops.

import heapq

class MedianFinder:
    def __init__(self):
        self.low = []    # max-heap (negated)
        self.high = []   # min-heap

    def addNum(self, x):
        heapq.heappush(self.low, -x)
        heapq.heappush(self.high, -heapq.heappop(self.low))   # move the largest low to high
        if len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))

    def findMedian(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2

class MedianFinder {
    priority_queue<int> low;                              // max-heap
    priority_queue<int, vector<int>, greater<int>> high;  // min-heap
public:
    void addNum(int x) {
        low.push(x);
        high.push(low.top()); low.pop();
        if (high.size() > low.size()) { low.push(high.top()); high.pop(); }
    }
    double findMedian() {
        if (low.size() > high.size()) return low.top();
        return (low.top() + (double)high.top()) / 2;
    }
};

Two heaps also solve IPO (available projects vs locked ones) and scheduling problems where items move from “waiting” to “ready”.

Pattern 4: Simulation / event scheduling

Process events in time order when new events are created on the fly.

import heapq

def get_order(tasks):                          # Single-Threaded CPU
    order = sorted(range(len(tasks)), key=lambda i: tasks[i][0])
    ready, res, time, i = [], [], 0, 0
    while len(res) < len(tasks):
        if not ready and time < tasks[order[i]][0]:
            time = tasks[order[i]][0]          # CPU idle: jump ahead
        while i < len(order) and tasks[order[i]][0] <= time:
            j = order[i]
            heapq.heappush(ready, (tasks[j][1], j))
            i += 1
        dur, j = heapq.heappop(ready)
        time += dur
        res.append(j)
    return res

Dijkstra and Prim are also “always expand the cheapest frontier item” — see Shortest Paths.

Lazy deletion

Heaps can't delete arbitrary items efficiently. Instead, mark items as deleted (in a hash map of pending removals or by a version/timestamp) and discard them when they reach the top.

while heap and is_stale(heap[0]):
    heapq.heappop(heap)

Used in Sliding Window Median, the Skyline Problem, and Dijkstra (skip entries whose distance is outdated).

Heap vs other structures

Common mistakes

• Min vs max heap confusion between languages.
• Comparing non-comparable objects in Python tuples (add an index tie-breaker).
• Forgetting heap order is not sorted order — only h[0] is guaranteed.
• Using a heap where sorting once is simpler (no inserts after the start).
• Rebuilding a heap each iteration (O(n) each) instead of pushing/popping.