AmouAI Hub/Courses/Data Structures & Algorithms/Chapter 17

Part 4 · Trees · Chapter 17

Heaps and Priority Queues

A hospital triage desk, not a queue: whoever is most urgent goes next. A tree that hides inside an array, with arithmetic instead of pointers.

Reading

CLRS Ch. 6

Focus

One extreme, cheaply

Cost

push/pop O(log n)

Powers

Dijkstra · schedulers

By the end of this chapter you can

State the heap property and explain how it differs from the BST property
Map a complete binary tree onto an array using the 2i+1 / 2i+2 index arithmetic
Implement sift-up and sift-down, and justify their O(log n) cost
Explain why bottom-up heapify is O(n) rather than O(n log n)
Use heapq correctly, including max-heaps, tuples and the top-k pattern

1Triage, not a queue

A queue is fair. Sometimes fair is wrong — you want whoever is most urgent, and you want it now.

The emergency department

A queue would serve people in arrival order. An emergency department does not: a nurse assesses everyone and the most urgent case goes next, regardless of who arrived first.

Crucially, the waiting room is not sorted. Nobody maintains a full ranking of forty patients. There is just enough organisation to answer one question instantly: who is most urgent?

That “just enough” is exactly what a heap maintains, and it is why a heap is cheaper than a sorted structure.

A priority queue supports three operations:

push(item, priority) — add an item
pop() — remove and return the highest-priority item
peek() — look at it without removing

You could implement that three obvious ways, and all three are worse:

Implementation	push	pop min	Why it is wrong
Unsorted list	O(1)	O(n)	Every pop scans everything
Sorted list	O(n)	O(1)	Every push shifts elements
Balanced BST	O(log n)	O(log n)	Correct, but it maintains a full ordering you never asked for
Heap	O(log n)	O(log n)	Maintains exactly enough order and no more

The heap property

In a min-heap: every node is less than or equal to both of its children. That is the whole rule.

Note what it does not say. There is no relationship between siblings, no left-versus-right meaning, and no sorted order anywhere. The only guarantee is that the minimum is at the root — and that is precisely the guarantee you asked for.

Compare with the BST property, which is far stronger:

	BST	Heap
Rule	left < node < right, for entire subtrees	node ≤ both children
Find the minimum	O(log n) — walk left	O(1) — it is the root
Find an arbitrary value	O(log n)	O(n) — no idea where it is
Sorted traversal	O(n) in-order	O(n log n) — pop everything
Shape	Any	Always complete
Storage	Pointers	A flat array

A heap gives up general search entirely, and in exchange gets a much weaker invariant that is much cheaper to maintain. That is the trade, and it is the most focused one in the course.

2A tree with no pointers

Because a heap is always a complete tree, its shape is fully determined by its size — so the pointers can be replaced by arithmetic.

A complete binary tree has every level full except possibly the last, which fills left to right. That constraint means there is exactly one shape for any number of nodes — and so the tree can be written down as a flat array in level order.

index_arithmetic.py

#            1              index 0
#          /   \
#         3     2           indices 1, 2
#        / \   /
#       7  5  4             indices 3, 4, 5
#
#  array:  [1, 3, 2, 7, 5, 4]
#           0  1  2  3  4  5

left_child(i)  = 2*i + 1
right_child(i) = 2*i + 2
parent(i)      = (i - 1) // 2

# Node 1 (value 3): children at 3 and 4 → values 7 and 5.  ✓
# Node 4 (value 5): parent at (4-1)//2 = 1 → value 3.      ✓

Why this is worth so much

No pointers. A million-element heap of integers is a million integers, not a million objects each with two references.
Cache locality. A parent and its children are near each other in memory, so traversing a heap is fast in a way that traversing a pointer tree is not.
No allocation. Push and pop are array writes and one length change.

This only works because the tree is complete. A BST can be any shape, so it cannot be stored this way without wasting exponential space on the gaps.

3Sift up, sift down

Every heap operation is: break the property in one place, then walk one path to repair it. Both walks are O(log n) because a path is the height.

Push — sift up

Put the new value in the first free slot at the end of the array, which keeps the tree complete. That almost certainly breaks the heap property, so compare with the parent and swap upward until it does not.

push.py

def push(heap, x):
    heap.append(x)                    # keeps the tree complete
    i = len(heap) - 1
    while i > 0:
        parent = (i - 1) // 2
        if heap[parent] <= heap[i]:
            break                     # property restored — stop early
        heap[parent], heap[i] = heap[i], heap[parent]
        i = parent

Pop — sift down

The minimum is at index 0, so reading it is free. The problem is the hole it leaves. The fix: move the last element into the root — removing from the end is free and keeps the tree complete — then sink it down, swapping with its smaller child until it settles.

pop.py

def pop(heap):
    if not heap:
        raise IndexError("pop from an empty heap")
    smallest = heap[0]
    last = heap.pop()                 # O(1) — removing from the END
    if heap:
        heap[0] = last
        sift_down(heap, 0)
    return smallest


def sift_down(heap, i):
    n = len(heap)
    while True:
        left, right = 2*i + 1, 2*i + 2
        best = i
        if left  < n and heap[left]  < heap[best]: best = left
        if right < n and heap[right] < heap[best]: best = right
        if best == i:
            return                    # already smaller than both children
        heap[i], heap[best] = heap[best], heap[i]
        i = best

Swap with the SMALLER child

Sinking into the larger child is the classic bug. If you swap with the larger of the two, the value you promote is not necessarily smaller than its new sibling, and the heap property breaks silently.

The symptom is nasty: the heap still looks fine, pops mostly return plausible values, and occasionally you get one out of order. Always compare against both children and take the smaller.

peek

O(1)

push

O(log n)

pop

O(log n)

heapify

O(n)

space

O(n)

extra space

O(1)

4Heapify in O(n)

Turning an arbitrary array into a heap looks like it should cost n log n. It does not, and the reason is a nice piece of counting.

The obvious approach is n pushes: O(n log n). There is a better way, and it works backwards.

Every element in the second half of the array is a leaf — and a single leaf is already a valid heap. So half the array needs no work at all. Start at the last non-leaf and sift down, moving backwards to the root.

heapify.py

def heapify(arr):
    # everything from n//2 onward is a leaf, hence already a valid heap
    for i in range(len(arr) // 2 - 1, -1, -1):
        sift_down(arr, i)
    return arr

Why this is O(n) and not O(n log n)

The cost of sifting down is bounded by the node's height, not the tree's. And most nodes are near the bottom, where the height is tiny:

n/2 nodes are leaves — height 0, cost 0
n/4 nodes have height 1
n/8 nodes have height 2
… and exactly 1 node has height log n

Total work is the sum of (number at height h) × h, which is n × ∑(h / 2^h+1). That series converges to 1, so the total is less than n.

The intuition: the expensive nodes are rare and the common nodes are cheap. Doing n pushes gets this exactly backwards — it makes every element climb from the bottom.

Heapsort

Heapify, then repeatedly swap the root to the end of the array and shrink the heap by one. The sorted region grows from the right, and nothing is allocated.

heapsort.py

def heapsort(arr):
    heapify(arr)                       # O(n)
    end = len(arr)
    while end > 1:
        arr[0], arr[end-1] = arr[end-1], arr[0]   # smallest to the back
        end -= 1
        sift_down(arr, 0, end)                    # restore over the smaller heap
    return arr

# O(n log n) WORST CASE with O(1) extra space — the only common sort
# that guarantees both. Merge sort needs O(n) space; quicksort has an
# O(n²) worst case.
#
# So why is nothing sorted with it? Cache locality. Sifting down jumps
# between indices i, 2i+1, 4i+3... which scatters across memory, while
# quicksort scans linearly. In practice quicksort wins by a wide margin,
# and heapsort survives as introsort's safety net.

5heapq, and the patterns worth knowing

Python's heapq is a set of functions operating on a plain list. Three idioms cover almost everything you will do with it.

heapq does not give you a class. It gives you functions that treat any list as a min-heap, which means no wrapper object and no conversion cost.

heapq_basics.py

import heapq

h = []
heapq.heappush(h, 5)
heapq.heappush(h, 1)
heapq.heappush(h, 3)
h[0]                      # 1 — peek is just index 0
heapq.heappop(h)          # 1

heapq.heapify(existing_list)          # in place, O(n)
heapq.heappushpop(h, x)               # push then pop — one sift, not two
heapq.heapreplace(h, x)               # pop then push — also one sift
heapq.nlargest(3, iterable)           # top 3
heapq.nsmallest(3, iterable)

Idiom 1 — a max-heap

heapq is min-only. Negate on the way in and on the way out.

maxheap.py

# max-heap via negation
h = []
for x in [5, 1, 8, 3]:
    heapq.heappush(h, -x)
largest = -heapq.heappop(h)     # 8

# For objects you cannot negate, wrap in a tuple with a negated key:
heapq.heappush(h, (-priority, task))

Idiom 2 — tuples for priority, and the tie-break trap

tuples.py

# Tuples compare element by element, so (priority, item) works...
heapq.heappush(h, (2, "write report"))
heapq.heappush(h, (1, "fix outage"))
heapq.heappop(h)                    # (1, C)

# ...until two priorities tie, at which point Python compares the SECOND
# element — and if that is an object with no ordering, it raises TypeError:
heapq.heappush(h, (1, some_object))
heapq.heappush(h, (1, another_object))     # TypeError: D not supported

# The fix: a monotonically increasing counter as a tie-breaker.
import itertools
counter = itertools.count()
heapq.heappush(h, (priority, next(counter), task))

# It never ties (the counter is unique), and as a bonus it makes the
# queue stable: equal priorities come out in insertion order.

Idiom 3 — top k without sorting

top_k.py

def top_k(stream, k):
    """The k largest items, using O(k) memory. The stream can be huge."""
    h = []
    for x in stream:
        if len(h) < k:
            heapq.heappush(h, x)
        elif x > h[0]:                    # bigger than the smallest kept
            heapq.heapreplace(h, x)       # one sift instead of two
    return sorted(h, reverse=True)

# O(n log k) time and O(k) space.
# sorted(stream)[:k] is O(n log n) time and O(n) space — and impossible
# if the stream does not fit in memory.

Where heaps show up	What is prioritised
Dijkstra's algorithm (Ch. 22)	The closest unfinished vertex
*A search** (Ch. 22)	Estimated total path cost
Prim's MST (Ch. 23)	The cheapest edge leaving the tree
Huffman coding (Ch. 24)	The two least frequent symbols
OS schedulers	Process priority
Event simulation	The next event by timestamp
Merging k sorted streams	The smallest unconsumed head
Rate limiters, timers	The soonest deadline

How to recognise a heap problem

The tell is: you repeatedly need the extreme element from a collection that keeps changing.

If the collection were static you would sort it once. If you only needed the extreme once you would call min(). It is the combination — repeated extraction and ongoing insertion — that makes a heap the right answer.

What to carry forward

A heap keeps only enough order to know the minimum: every node ≤ both children, and nothing else. That weak invariant is why it is cheap.
Because a heap is a complete tree, it lives in a flat array: children at 2i+1 and 2i+2, parent at (i−1)//2. No pointers, good cache behaviour.
Push = append then sift up. Pop = take the root, move the last element in, sift down. Both walk one root-to-leaf path, so both are O(log n).
Always sift down into the smaller child. Choosing the larger breaks the heap silently.
Bottom-up heapify is O(n), not O(n log n) — most nodes are near the bottom and barely move.
In Python: heapq on a plain list. Negate for a max-heap, add a counter to break ties, and use heapreplace for the top-k pattern.

>_Playground

A complete heap implementation, then the heapq idioms. Compare heapify against n pushes and see the O(n) claim.

scratch.pypython not loaded

import heapq, random, time

# ---- from scratch ----
def sift_up(h, i):
    while i > 0:
        p = (i - 1) // 2
        if h[p] <= h[i]:
            return
        h[p], h[i] = h[i], h[p]
        i = p

def sift_down(h, i, n=None):
    n = len(h) if n is None else n
    while True:
        l, r, best = 2*i + 1, 2*i + 2, i
        if l < n and h[l] < h[best]: best = l
        if r < n and h[r] < h[best]: best = r
        if best == i:
            return
        h[i], h[best] = h[best], h[i]
        i = best

def push(h, x):
    h.append(x)
    sift_up(h, len(h) - 1)

def pop(h):
    smallest = h[0]
    last = h.pop()
    if h:
        h[0] = last
        sift_down(h, 0)
    return smallest

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

def is_heap(h):
    return all(h[i] <= h[c]
               for i in range(len(h))
               for c in (2*i+1, 2*i+2) if c < len(h))

h = []
for x in [5, 3, 8, 1, 9, 2]:
    push(h, x)
    print(f"push {x}: {h}")
print("valid heap?", is_heap(h))
print("popping everything:", [pop(h) for _ in range(6)])
print()

# ---- heapify vs n pushes: count the swaps ----
SWAPS = 0
def sift_down_counting(a, i, n):
    global SWAPS
    while True:
        l, r, best = 2*i+1, 2*i+2, i
        if l < n and a[l] < a[best]: best = l
        if r < n and a[r] < a[best]: best = r
        if best == i: return
        a[i], a[best] = a[best], a[i]
        SWAPS += 1
        i = best

def sift_up_counting(a, i):
    global SWAPS
    while i > 0:
        p = (i-1)//2
        if a[p] <= a[i]: return
        a[p], a[i] = a[i], a[p]
        SWAPS += 1
        i = p

random.seed(3)
for n in [1000, 10000, 100000]:
    data = [random.randint(0, 10**6) for _ in range(n)]

a = data[:]
    SWAPS = 0
    for i in range(len(a)//2 - 1, -1, -1):
        sift_down_counting(a, i, len(a))
    bottom_up = SWAPS

b = []
    SWAPS = 0
    for x in data:
        b.append(x)
        sift_up_counting(b, len(b)-1)
    n_pushes = SWAPS

print(f"n={n:7}  bottom-up heapify: {bottom_up:9,} swaps"
          f"   ({bottom_up/n:.2f} per element)")
    print(f"          n pushes         : {n_pushes:9,} swaps"
          f"   ({n_pushes/n:.2f} per element)")
print()
print("Bottom-up stays under ~1 swap per element as n grows: that is the O(n).")
print("n pushes grows with log n per element: that is the O(n log n).")
print()

# ---- heapq idioms ----
import itertools
counter = itertools.count()
pq = []
for pri, task in [(2, "write report"), (1, "fix outage"), (1, "answer email")]:
    heapq.heappush(pq, (pri, next(counter), task))
print("priority queue, ties broken by arrival:")
while pq:
    pri, _, task = heapq.heappop(pq)
    print(f"  [{pri}] {task}")

Real Python, running in your browser. Nothing is installed or uploaded.

Output appears here. The first run takes a few seconds while Python loads.

✓Exercises

Checked automatically the moment you submit. Work top to bottom — each one assumes the last. Your answers are saved in this browser.

All Warm-up Core Challenge Reset chapter

Chapter 18 — Tries and Prefix Trees

The structure behind autocomplete and spellcheck. Lookup cost stops depending on how many words you stored.

Continue →