Heap Data Structure: When to Reach for One in Coding Interviews (and When Not To)

Events arrive forever, you have to keep the 100 largest, and memory is tight. Sorting is off the table.

A heap data structure is a complete binary tree stored in a flat array, where every parent compares less than or equal to its children (min-heap) or greater than or equal to them (max-heap). It hands you the extreme element in O(1), inserts and extracts in O(log n), and builds from an unsorted array in O(n). What you give up is total order: the array is not sorted, never becomes sorted, and cheap access to one end is exactly what you bought by refusing to pay for the rest.

Key Takeaways

• A size-k min-heap is the streaming answer, not the fast answer. Its O(n log k) loses to heapify-all-then-pop-k's O(n + k log n) whenever the data already fits in memory and k is not tiny.
• heapify is O(n), but top-K still is not. Building costs O(n), draining k roots costs another O(k log n), and interviewers push back on candidates who collapse the two terms.
• Always give tuples a tiebreaker. Push (priority, next(counter), payload). Without the counter, a tie sends Python into comparing your payloads and raises TypeError months after the tests passed.
• heapq has no decrease-key and no arbitrary delete. Keep the authoritative value in a dict, let stale heap entries pile up, and discard them when they surface.
• A heap gives you one end and nothing else. Range queries, successor lookups, and ordered scans all want a balanced tree or a sorted container instead.

A heap data structure is an array pretending to be a tree

Asked at Microsoft — Retain Top K Elements Given a list of integers and an integer k, return the list with every element except the k largest removed. Retained values keep their original relative order, duplicates count as separate elements, and when equal values tie at the cutoff you keep the leftmost ones. A k of zero or less returns an empty list; a k at or above the length returns the input unchanged.

There are no node objects and no pointers anywhere in a binary heap. The tree is implied by index arithmetic on a plain list:

• children of i live at 2i + 1 and 2i + 2
• the parent of i lives at (i - 1) // 2

For the array [3, 5, 4, 9, 8, 6]:

Every parent is smaller than both of its children, so this is a valid min-heap. Notice that 4 at index 2 is smaller than 5 at index 1, and that is legal. The invariant binds a parent to its own children and says nothing about siblings, cousins, or left-to-right position.

That is precisely why the Microsoft question is harder than it looks. The heap can tell you which k values survive. It cannot tell you what order they arrived in.

Two operations maintain the invariant. Sift-up pushes an element that is too small for its slot toward the root. Sift-down pushes one that is too large toward the leaves.

def _sift_up(heap, i):
    while i > 0:
        parent = (i - 1) // 2
        if heap[i] < heap[parent]:
            heap[i], heap[parent] = heap[parent], heap[i]
            i = parent
        else:
            return

def _sift_down(heap, i, n):
    while True:
        left = 2 * i + 1
        right = left + 1
        smallest = i
        if left < n and heap[left] < heap[smallest]:
            smallest = left
        if right < n and heap[right] < heap[smallest]:
            smallest = right
        if smallest == i:
            return
        heap[i], heap[smallest] = heap[smallest], heap[i]
        i = smallest

def push(heap, x):
    heap.append(x)
    _sift_up(heap, len(heap) - 1)

def pop(heap):
    top = heap[0]
    last = heap.pop()          # remove the last leaf
    if heap:                   # unless that was the only element
        heap[0] = last
        _sift_down(heap, 0, len(heap))
    return top

Both loops walk a single root-to-leaf path, so both cost O(log n) time and O(1) extra space. pop is where people fumble under pressure: the root's replacement has to be the last leaf, since it is the only element you can remove without leaving a hole in the middle of the array.

So the answer to the Microsoft question is two passes, not one.

import heapq

def retain_top_k(nums, k):
    if k <= 0:
        return []
    if k >= len(nums):
        return list(nums)
    # (value, -index) makes ties resolve toward the leftmost occurrence
    top = heapq.nlargest(k, ((v, -i) for i, v in enumerate(nums)))
    keep = {-neg_i for _, neg_i in top}
    return [v for i, v in enumerate(nums) if i in keep]

O(n log k) time, O(k) extra space beyond the output. The heap picks the survivors; a linear scan over the original list restores their order. Candidates who try to produce the output straight from the heap end up sorting it, and lose the ordering requirement in the process.

One aside that trips up anyone reading CPython's heapq.py: its _siftdown moves an item toward the root and its _siftup moves toward the leaves, inverted relative to most textbooks. _siftup also walks the smaller child all the way down and then sifts back up, spending roughly one comparison per level instead of two. Same asymptotics, fewer comparisons.

Heapify builds in O(n); pushing n times costs O(n log n)

Asked at NVIDIA — Find Top K Frequent Elements Given an integer array, return the k values that occur most often. The interviewer opened by asking only for the top three, then generalised it to an arbitrary k.

Once you have counts, this is a selection problem over the distinct values, and how you build the heap over them fixes the first term of your complexity.

def heapify(a):
    for i in range(len(a) // 2 - 1, -1, -1):
        _sift_down(a, i, len(a))

Start at the last internal node and work backward. A sift-down from height h costs O(h), and a heap of n elements holds at most n / 2^(h+1) nodes at height h. Summing h · n / 2^(h+1) over all heights gives n · Σ h/2^(h+1), and that series converges to 1, which bounds the whole build at n swaps. Each level of a sift-down spends up to two comparisons (left child, right child), so comparisons are bounded by 2n. Both are O(n).

The algebra has a plain-language version. Half the nodes are leaves and cost nothing, a quarter sit one level up and cost at most one swap, and only the root can travel a full log n. The expensive levels are the ones with almost no nodes in them.

None of that makes top-K linear, and the follow-up will test whether you know it. Heapifying is O(n); draining k roots is another O(k log n). "Heapify is O(n), so I can get the top K in O(n)" invites a correction, and it deserves one.

For this particular question there is something better than either. The counts are integers bounded by the array length, so you can bucket by frequency and read the buckets from the top down:

from collections import Counter

def top_k_frequent(nums, k):
    counts = Counter(nums)
    buckets = [[] for _ in range(len(nums) + 1)]
    for value, freq in counts.items():
        buckets[freq].append(value)
    out = []
    for freq in range(len(nums), 0, -1):
        for value in buckets[freq]:
            out.append(value)
            if len(out) == k:
                return out
    return out

O(n) time, O(n) space, no heap at all. Lead with the heap version anyway, because heapq.nlargest(k, counts.items(), key=lambda kv: kv[1]) is one line and it is what most interviewers expect first, then offer the bucket version as the improvement.

A size-k min-heap buys streaming, not speed

Asked at Coinbase — Implement top-K over a stream Design a structure over a high-volume event stream — account IDs from new signups, in the version that was asked — supporting insertion, a query for the current top-K most frequent items, and optionally the top-K over the last T minutes. Expect to defend heaps against bucket counting against a count-min sketch, and to describe how the design partitions, aggregates partially, and merges.

The reflexive answer to "find the K largest" is a max-heap over everything, and it is a good answer: heapify is O(n), draining k roots is O(k log n), and for anything but a tiny k that total beats the size-k heap. What it cannot do is start before the input ends, and it holds all n elements at once. On a stream, neither is available.

The min-heap of size k trades asymptotics for memory and a single pass. Its root is the weakest survivor, so a new element only has to beat one value to earn a place.

import heapq

def top_k(stream, k):
    if k <= 0:
        return []
    heap = []                      # min-heap of the k best seen so far
    for x in stream:
        if len(heap) < k:
            heapq.heappush(heap, x)
        elif x > heap[0]:          # beats the current weakest survivor
            heapq.heapreplace(heap, x)
    return sorted(heap, reverse=True)

O(n log k) time, O(k) space, one pass, and it runs on an iterator that never ends. The k <= 0 guard earns its line: without it, heap[0] on an empty list raises IndexError on the very first element.

Two heapq details show up in that loop. heapreplace(heap, x) pops before it pushes, so it always returns the old root and requires a non-empty heap, while heappushpop(heap, x) pushes first and hands x straight back when x is smaller than the root. Guarded heapreplace and unguarded heappushpop are interchangeable here, and both cost one sift instead of two.

The second is that you are keeping the largest values in a min-heap, an inversion candidates reverse under pressure. Through Python 3.13 the public heapq API is min-heap only, with the max variants present as private helpers (_heapify_max, _heappop_max, _heapreplace_max). Python 3.14 promoted them to public heapify_max, heappush_max, heappop_max, heappushpop_max, and heapreplace_max. On anything older, which covers most interview environments, you push -x and negate on the way out, or wrap the payload in a class with an inverted __lt__ when negation is not defined for it.

The harder half of the Coinbase question is that top-K by frequency over an unbounded stream cannot be exact in bounded memory, because exactness needs a live counter for every distinct key ever seen. Decide out loud whether you are promising an exact answer over a bounded window or an approximate one over all time. Sliding windows, count-min sketches, and per-shard partial aggregates all fall out of that single choice.

Quickselect wins the one-shot; the heap wins the stream

Asked at Meta — Find the kth largest element Given an unsorted array of n integers and a k between 1 and n, return the k-th largest value without sorting the whole array. Two solutions are expected — an expected-linear-time selection algorithm and a heap-based one — with complexity for each, plus how you would adapt either to a data stream.

Sorting, heapifying, and quickselect all return the same number here. The interview is about which one survives the constraint that arrives next.

Quickselect partitions around a random pivot and recurses into one side only, so it never sorts what it does not have to. Average O(n). Adversarial pivots make the worst case quadratic, and median-of-medians repairs that at O(n) worst case with a constant nobody enjoys. It also needs the whole array in memory, it reorders your input, and against a stream it does not apply at all.

When k approaches n, the size-k heap collapses to O(n log n) with worse constants and more code to get wrong. CPython's nlargest short-circuits both ends of that range: k == 1 delegates to max(), and k >= len(iterable) returns sorted(...)[:k]. The second only fires for inputs that have a __len__, so a generator paired with a huge k gets no rescue. Neither behaviour is documented, since the docstring claims only equivalence to sorted(iterable, key=key, reverse=True)[:n], so treat both as implementation detail rather than contract.

Where a heap is the wrong tool

• You need the whole thing sorted anyway. Sort. Heapsort's remaining edge is memory: it is in-place, while sorted() builds a new list and Timsort's merge wants a temp buffer up to n/2. Against an introsort like C++'s std::sort almost nothing is left, since introsort is already in-place and already falls back to heapsort for its O(n log n) guarantee.
• You need range queries, successor lookup, or an ordered scan. A heap gives you one end. Use a balanced BST or a sorted container.
• k is a large fraction of n. O(n log k) degrades to O(n log n) with a worse constant.
• The values are small bounded integers. Counting or bucketing is O(n) and shorter.
• Removals are frequent and arbitrary. Which is the next section.

heapq has no decrease-key, so a dict has to hold the truth

Asked at Applied Intuition — Design event timeout detector Build a timeout detector for a job scheduler. You get a global timeout T and a stream of events, each carrying an id, a type of start, end, or ping, and a timestamp. An event has timed out once it has started, has not ended, and the gap between now and its last update exceeds T; a ping refreshes it and an end retires it. Expose process(event) and get_timed_out(now), and handle duplicates, out-of-order arrivals, pings with no prior start, repeated starts, and a T of zero or less.

A min-heap keyed by deadline is the obvious index. The trouble is that a ping moves a deadline, and heapq offers no way to update or remove an entry that is not at the root. This is the gap that turns "just use a priority queue" into a design conversation.

Lazy deletion is the standard repair. Keep the authoritative deadline in a dict, let stale heap entries accumulate, and discard them when they surface.

import heapq, itertools

class TimeoutDetector:
    def __init__(self, timeout):
        self.timeout = timeout
        self._heap = []                       # (deadline, seq, event_id)
        self._live = {}                       # event_id -> its current deadline
        self._seq = itertools.count()

    def process(self, event):
        eid, kind, ts = event["event_id"], event["type"], event["timestamp"]
        if kind == "end":
            self._live.pop(eid, None)         # heap entry stays, dies on the way out
            return
        if kind == "ping" and eid not in self._live:
            return                            # a ping with no live start is noise
        deadline = ts + self.timeout
        current = self._live.get(eid)
        if current is not None and deadline <= current:
            return                            # duplicate or out-of-order, ignore
        self._live[eid] = deadline
        heapq.heappush(self._heap, (deadline, next(self._seq), eid))

    def get_timed_out(self, now_ts):
        fired = []
        while self._heap and self._heap[0][0] < now_ts:
            deadline, _, eid = heapq.heappop(self._heap)
            if self._live.get(eid) == deadline:
                del self._live[eid]
                fired.append(eid)
        return fired

Three details in there are worth defending before you are asked.

The pop condition is strict. The spec says an event times out when now − last_update > T, and deadline = last_update + T, so an entry fires only when deadline < now_ts. Writing <= reports every event exactly one tick early, and no small test will catch it.

next(self._seq) is load-bearing. Tuples compare lexicographically, so two entries with identical deadlines fall through to comparing the next field. Event ids happen to be orderable strings, but the moment the payload becomes a dict or a custom object, heappush raises TypeError: '<' not supported. Two priorities have to collide before it fires, which is why a five-element test never surfaces it. The counter also buys you FIFO ordering inside a single deadline.

Stale entries are unbounded. One start followed by five pings leaves six entries in the heap for a single live event. A workload that pings constantly inflates the heap without limit, and the repair is either a periodic compaction pass or an indexed heap that supports real deletion.

The same gap appears wherever a ranked value can change after you push it. Maintain top N payers (Meta) asks for the n accounts with the highest cumulative outgoing amounts, with online updates as new debits land. Every new payment invalidates a heap entry you already pushed.

Practice these on PracHub

Treat these as one shape under different constraints rather than nine separate problems.

Selection under a stated constraint

• Find the kth largest element (Meta) — write both the heap and the quickselect version, then defend a choice against "the data streams" and against "the array is read-only."
• Find Top K Frequent Elements (NVIDIA) — counting plus selection, and the case where bucketing beats the heap outright.
• Retain Top K Elements (Microsoft) — the heap gives you a cutoff and a second pass restores the original order. Ask what happens when the k-th and (k+1)-th values tie.

Comparator traps

• Return Top K Relevant Apps (Microsoft, premium) — sum nested keyword scores, rank descending, break ties by app name ascending. Two keys pointing in opposite directions.
• Return Top K Open Businesses (Microsoft, premium) — score descending, distance ascending on ties, filtered to open businesses. reverse=True is no help and negating the whole tuple flips both keys; key=lambda b: (-b.score, b.distance) is the fix.

Values that move after you push them

• Maintain top N payers (Meta) — cumulative totals grow with every debit, so the heap is stale by construction. State your tie-break rule for equal totals.
• Design event timeout detector (Applied Intuition) — lazy deletion with a real spec around it: duplicates, out-of-order events, non-positive timeouts.
• Design autocomplete with Trie (Google) — topK(prefix, k) is the hot path while update and delete mutate weights underneath it. Decide what you cache per node and what you rebuild.

Unbounded input

• Implement top-K over a stream (Coinbase) — top-K by frequency with no array to revisit. Commit to exact-over-a-window or approximate-over-all-time before you write a line.

Every one of them ships a runnable console. Write the sort version first and the heap version second — running both is the fastest way to see which constraint each one breaks.

FAQ

Is a heap the same thing as a priority queue?

No. A priority queue is an interface — insert, extract-highest-priority — and a heap is the data structure that usually implements it. Python's queue.PriorityQueue wraps a heap in locking for thread safety, which makes it slower than raw heapq in single-threaded code. Say "priority queue backed by a binary heap" in an interview and you have named both correctly.

Why is heapify O(n) if each sift-down is O(log n)?

Because almost no nodes sit near the root. Half of them are leaves and need zero work, and only one node can travel a full log n. Summing the per-height cost h · n/2^(h+1) across all heights bounds the build at n swaps, and since sift-down spends up to two comparisons per level, comparisons are bounded by 2n. Building by pushing n times is O(n log n), because then every element starts at a leaf and can climb the entire height.

How do I make a max-heap in Python?

On Python 3.14 and later, heapq exposes public max-heap functions: heapify_max, heappush_max, heappop_max, heappushpop_max, and heapreplace_max. On 3.13 and earlier there is no public max-heap API, so you push -x and negate on the way out for numbers, wrap items in a class with an inverted __lt__ for anything negation does not apply to, or call heapq.nlargest(k, items, key=...) when you only need the top k rather than a live structure.

Should I use a heap or quickselect for the kth largest element?

Quickselect when the data is in memory, you need one answer, and mutating the input is allowed, since it is O(n) on average. A size-k heap when the data streams, does not fit in memory, or will be queried repeatedly as more arrives. When k is close to n, sort.

How do two heaps give you a running median?

Keep the smaller half of the values in a max-heap and the larger half in a min-heap, sized so the two roots straddle the middle. Push every new value into the max-heap, move that heap's root across to the min-heap, then move one back if the min-heap grew larger. The unconditional transfer is what keeps the partition correct without comparison branches. Inserts are O(log n) and the median query is O(1); the follow-up to prepare for is a sliding window, where you also have to remove the element leaving the window and lazy deletion comes back.

Why does heapq.heappush raise TypeError?

You are pushing tuples whose first elements tied, so Python fell through to comparing the second element and that element is not orderable. Insert a monotonically increasing counter as the second field so the payload is never compared. The bug only appears when priorities collide, which is why small tests miss it.