Heaps: Min-Heaps, Max-Heaps, and Priority Queues

Introduction

A heap is a complete binary tree where every parent node satisfies the heap property: either every parent is less than or equal to its children (min-heap) or every parent is greater than or equal to its children (max-heap). This constraint gives O(1) access to the minimum/maximum element while maintaining O(log n) insertion and deletion.

Heaps are the backbone of priority queues, Dijkstra’s algorithm, heap sort, and many other algorithms. The tree is stored as an array, using index arithmetic to navigate parent-child relationships.

Heap Implementation

class MinHeap:
    """
    Min-heap implementation using array representation.
    Parent at index i → children at 2i+1 and 2i+2
    Child at index i → parent at (i-1) // 2
    """

    def __init__(self):
        self.heap = []

    def _parent(self, i):
        return (i - 1) // 2

    def _left_child(self, i):
        return 2 * i + 1

    def _right_child(self, i):
        return 2 * i + 2

    def _swap(self, i, j):
        self.heap[i], self.heap[j] = self.heap[j], self.heap[i]

    def _sift_up(self, i):
        """Move element up to maintain heap property."""
        while i > 0 and self.heap[self._parent(i)] > self.heap[i]:
            self._swap(i, self._parent(i))
            i = self._parent(i)

    def _sift_down(self, i):
        """Move element down to maintain heap property."""
        n = len(self.heap)
        while self._left_child(i) < n:
            smallest = self._left_child(i)

            if (self._right_child(i) < n and
                self.heap[self._right_child(i)] < self.heap[smallest]):
                smallest = self._right_child(i)

            if self.heap[i] <= self.heap[smallest]:
                break

            self._swap(i, smallest)
            i = smallest

    def push(self, val):
        """O(log n) insert."""
        self.heap.append(val)
        self._sift_up(len(self.heap) - 1)

    def pop(self):
        """O(log n) extract minimum."""
        if not self.heap:
            raise IndexError("Heap is empty")

        min_val = self.heap[0]
        last = self.heap.pop()

        if self.heap:
            self.heap[0] = last
            self._sift_down(0)

        return min_val

    def peek(self):
        """O(1) view minimum."""
        if not self.heap:
            raise IndexError("Heap is empty")
        return self.heap[0]

    def __len__(self):
        return len(self.heap)


## Heap Structure and Array Mapping

```mermaid
graph TD
    subgraph Tree["Heap as Tree"]
        direction TB
        R["[0] 10"] --> L["[1] 20"]
        R --> M["[2] 30"]
        L --> LL["[3] 40"]
        L --> LM["[4] 50"]
        M --> ML["[5] 60"]
        M --> MR["[6] 70"]
    end

    subgraph Array["Array: [10, 20, 30, 40, 50, 60, 70]"]
        direction LR
        I0["[0]"] --> I1["[1]"] --> I2["[2]"] --> I3["[3]"] --> I4["[4]"] --> I5["[5]"] --> I6["[6]"]
    end

Parent at index i has children at 2i+1 and 2i+2. Child at index i has parent at (i-1) // 2. This array representation makes heaps cache-friendly compared to pointer-based trees.

Min-Heap vs Max-Heap Operations

graph LR
    subgraph MinHeap["Min-Heap (parent ≤ children)"]
        direction TB
        MRoot["[0] 10 (min)"]
        MRoot --> MLeft["[1] 20"]
        MRoot --> MRight["[2] 30"]
        MLeft --> MLL["[3] 40"]
        MLeft --> MLM["[4] 50"]

        MOp1["push(5)"]
        MOp2["sifts up to root"]
        MOp3["pop() → 5 (minimum)"]
    end

    subgraph MaxHeap["Max-Heap (parent ≥ children)"]
        direction TB
        XRoot["[0] 70 (max)"]
        XRoot --> XLeft["[1] 50"]
        XRoot --> XRight["[2] 40"]
        XLeft --> XLL["[3] 20"]
        XLeft --> XLM["[4] 10"]

        XOp1["push(80)"]
        XOp2["sifts up to root"]
        XOp3["pop() → 80 (maximum)"]
    end

class MaxHeap:
    """Max-heap: parent is always greater than or equal to children."""

    def __init__(self):
        self.heap = []

    def _parent(self, i):
        return (i - 1) // 2

    def _left_child(self, i):
        return 2 * i + 1

    def _right_child(self, i):
        return 2 * i + 2

    def _swap(self, i, j):
        self.heap[i], self.heap[j] = self.heap[j], self.heap[i]

    def _sift_up(self, i):
        while i > 0 and self.heap[self._parent(i)] < self.heap[i]:
            self._swap(i, self._parent(i))
            i = self._parent(i)

    def _sift_down(self, i):
        n = len(self.heap)
        while self._left_child(i) < n:
            largest = self._left_child(i)

            if (self._right_child(i) < n and
                self.heap[self._right_child(i)] > self.heap[largest]):
                largest = self._right_child(i)

            if self.heap[i] >= self.heap[largest]:
                break

            self._swap(i, largest)
            i = largest

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

    def pop(self):
        if not self.heap:
            raise IndexError("Heap is empty")

        max_val = self.heap[0]
        last = self.heap.pop()

        if self.heap:
            self.heap[0] = last
            self._sift_down(0)

        return max_val

    def peek(self):
        if not self.heap:
            raise IndexError("Heap is empty")
        return self.heap[0]

Priority Queue Using Heap

import heapq
from dataclasses import dataclass
from typing import Any

@dataclass
class PriorityItem:
    priority: int
    value: Any

class PriorityQueue:
    """
    Priority queue using min-heap.
    Lower priority number = higher priority (popped first).
    """

    def __init__(self):
        self._heap = []
        self._counter = 0  # Tiebreaker for same priority

    def enqueue(self, item, priority=0):
        """O(log n) insert."""
        heapq.heappush(self._heap, (priority, self._counter, item))
        self._counter += 1

    def dequeue(self):
        """O(log n) extract highest priority."""
        if not self._heap:
            raise IndexError("Priority queue is empty")
        return heapq.heappop(self._heap)[2]

    def peek(self):
        """O(1) view highest priority."""
        if not self._heap:
            raise IndexError("Priority queue is empty")
        return self._heap[0][2]

    def __len__(self):
        return len(self._heap)


# For max priority (larger number = higher priority), negate priority:
# enqueue(item, priority=-priority)
# dequeue returns item with highest original priority

Heap Sort

def heap_sort(arr):
    """
    Heap sort: O(n log n) worst case, O(1) space (in-place).
    Not stable but efficient.
    """
    n = len(arr)

    # Build max heap (rearrange array)
    def sift_down(start, end):
        while True:
            parent = start
            left = 2 * parent + 1

            if left > end:
                break

            # Find larger child
            right = left + 1
            if right <= end and arr[right] > arr[left]:
                left = right

            # If child > parent, swap and continue
            if arr[left] > arr[parent]:
                arr[left], arr[parent] = arr[parent], arr[left]
                parent = left
            else:
                break

    # Build max heap: sift down from last non-leaf node to root
    for start in range(n // 2 - 1, -1, -1):
        sift_down(start, n - 1)

    # Extract elements from heap: swap root with end, restore heap
    for end in range(n - 1, 0, -1):
        arr[0], arr[end] = arr[end], arr[0]  # Move max to end
        sift_down(0, end - 1)  # Restore heap on remaining

    return arr

Common Heap Problems

# Kth largest element in stream
class KthLargest:
    def __init__(self, k, nums):
        self.k = k
        self.heap = []
        for num in nums:
            self.add(num)

    def add(self, val):
        heapq.heappush(self.heap, val)
        if len(self.heap) > self.k:
            heapq.heappop(self.heap)
        return self.heap[0]


# Merge k sorted arrays
import heapq

def merge_sorted_arrays(sorted_arrays):
    result = []
    heap = []

    # Initialize with first element from each array
    for i, arr in enumerate(sorted_arrays):
        if arr:
            heapq.heappush(heap, (arr[0], i, 0))

    while heap:
        val, arr_idx, elem_idx = heapq.heappop(heap)
        result.append(val)

        # Add next from same array
        if elem_idx + 1 < len(sorted_arrays[arr_idx]):
            next_val = sorted_arrays[arr_idx][elem_idx + 1]
            heapq.heappush(heap, (next_val, arr_idx, elem_idx + 1))

    return result


# Median finder using two heaps
class MedianFinder:
    """
    O(log n) add, O(1) find median.
    max_heap stores smaller half, min_heap stores larger half.
    """

    def __init__(self):
        self.small = []  # Max heap (inverted)
        self.large = []  # Min heap

    def add_num(self, num):
        # Max heap: negate for max-heap behavior
        heapq.heappush(self.small, -num)

        # Balance: move largest from small to large
        heapq.heappush(self.large, -heapq.heappop(self.small))

        # Ensure small has >= large elements
        if len(self.large) > len(self.small):
            heapq.heappush(self.small, -heapq.heappop(self.large))

    def find_median(self):
        if len(self.small) > len(self.large):
            return -self.small[0]
        return (-self.small[0] + self.large[0]) / 2.0

When to Use Heaps

Use Case	Heap Type	Why
Priority queue	Min or Max	O(1) access to extreme
Dijkstra’s algorithm	Min	Always process shortest distance first
Top-k elements	Min of size k	Only keep k smallest/largest
Median finder	Two heaps	Balance to get middle elements
Merge sorted streams	Min	Always get globally minimum

Heap vs Binary Search Tree

Aspect	Heap	BST
Insert	O(log n)	O(log n)
Delete min/max	O(log n)	O(log n)
Search arbitrary	O(n)	O(log n)
Find min/max	O(1)	O(log n)*
Ordered traversal	No	Yes
Balanced required	No**	Yes

*BST if balanced **Height can degrade to O(n) if inserted in sorted order

Production Failure Scenarios and Mitigations

Scenario 1: Priority Inversion in Task Scheduling

Priority inversion is one of those bugs that looks impossible on paper. A low-priority task holds a lock, a high-priority task needs that lock, and an medium-priority task runs in between—which means the high-priority task sits blocked while the medium task wastes CPU cycles. Classic.

NASA’s Opportunity rover ran into this during wheel control. The system would get stuck, requiring restarts. The fix was priority inheritance: the low-priority task temporarily bumps up to the blocked task’s priority so it finishes and releases the resource. Most real-time schedulers (POSIX, FreeRTOS) support this, but you have to enable it explicitly.

Python’s heapq won’t save you here—it doesn’t handle priority inheritance at all. For production work with dependencies, look at libraries like TaskTiger that track task relationships.

Scenario 2: Heap Corruption Under Concurrency

Two threads pushing to a shared heap without locks. The heap array ends up with duplicates, missing elements, or a violated parent-child relationship. The worst part is the corruption is often silent—you don’t get an exception, you just get wrong answers later.

I once debugged a payment service that crashed randomly under load. Turns out the priority queue was shared across threads without synchronization. Once we added a mutex wrapper, the crashes stopped.

import threading
import heapq

class ThreadSafeHeap:
    def __init__(self):
        self._heap = []
        self._lock = threading.Lock()

    def push(self, item):
        with self._lock:
            heapq.heappush(self._heap, item)

    def pop(self):
        with self._lock:
            return heapq.heappop(self._heap)

Python’s queue.PriorityQueue handles locking internally. Rust’s binary_heap needs an explicit mutex if you’re sharing across threads.

Scenario 3: Unbounded Queue Growth

Producers outpacing consumers is a classic scaling problem. The queue keeps growing, memory fills up, and eventually the system starts swapping or dies.

A distributed indexing service I heard about had this happen during a traffic spike. The priority queue ballooned to 40GB before things went sideways. The fix: bounded queues with backpressure. Either reject new items, block producers, or drop low-priority items when at capacity.

def enqueue_bounded(self, item, priority, max_size=10000):
    with self._lock:
        if len(self._heap) >= max_size:
            # If the new item can't beat the worst item, reject it
            if priority < self._heap[0]:
                return False
            heapq.heappop(self._heap)  # Make room
        heapq.heappush(self._heap, (priority, self._counter, item))
        self._counter += 1
        return True

Scenario 4: Fibonacci Heap Lazy Deletion Problem

Fibonacci heaps look fantastic in algorithm textbooks. O(1) amortized insert! What could go wrong?

The “lazy deletion” part. Deleted nodes stick around marked as “removed” but not physically removed from the structure. Under heavy delete-min workload, you end up with a heap full of dead weight. Performance stops being O(1) amortized and starts looking more like O(n).

NetworkX, the Python graph library, actually falls back to binary heap for certain operations precisely because of this. Graph algorithms tend to delete a lot.

The fix: track deleted-node count and trigger compaction when the ratio gets too high. Or just use binary heap and save yourself the debugging headaches.

Trade-off Analysis

Heap Type	Push	Pop	Peek	Memory	Best for
Binary Heap	O(log n)	O(log n)	O(1)	O(n)	Most everyday priority queues
Fibonacci Heap	O(1) amort.	O(log n)*	O(1)	O(n)	Graph algos with decrease-key
Binomial Heap	O(log n)	O(log n)	O(1)	O(n)	When you need mergeable queues
d-ary Heap	O(log_d n)	O(d log_d n)	O(1)	O(n)	Cache-sensitive workloads
Pairing Heap	O(1)	O(log n)**	O(1)	O(n)	Practical when Fibonacci feels slow
Skew Heap	O(1) amort.	O(log n)*	O(1)	O(n)	Simple code, decent bounds

*Amortized bound **Pairing heaps aren’t formally bounded but perform well in practice

For most use cases, binary heap is the answer. Fibonacci heaps have real constant-factor overhead that eats the theoretical advantage. d-ary heaps (d=4 or d=8) are worth knowing about when cache locality matters.

Observability Checklist

• Track heap size and alert on unexpected growth
• Monitor push/pop latency—spikes point to contention or heap degradation
• Watch memory usage, especially with millions of items queued
• Log out-of-memory rejections on bounded queues
• Track priority inversion wait times if your scheduler supports it
• Measure cache miss rate—high L3 misses suggest poor heap locality
• Monitor lock contention on thread-safe heaps
• Alert when queue depth approaches configured limits
• Log dropped low-priority items—means producers are overwhelming consumers

Security Notes

Heap overflows in C/C++: The array underlying a heap can overflow if you’re not careful, corrupting adjacent memory. Use safe string functions and consider a memory-safe language if starting fresh.

Priority manipulation in multi-tenant systems: A bad actor might flood high-priority items to starve everyone else. Rate-limit enqueues per user and cap priority values.

Timing attacks: Heap operation timing varies with tree structure, which can leak information about data. If you’re in a security-sensitive context, use constant-time comparisons.

DoS via heap growth: An attacker who can trigger unbounded push can exhaust memory. Always enforce size limits on external input.

Observability data: Heap dumps and profiler output sometimes contain sensitive queued data. Sanitize before exporting.

Quick Recap Checklist

• Heap is complete binary tree with heap property
• Min-heap: parent ≤ children, O(1) peek at minimum
• Max-heap: parent ≥ children, O(1) peek at maximum
• Stored as array: parent at (i-1)//2, children at 2i+1 and 2i+2
• Push/pop: O(log n) via sift up/down
• Heap sort: O(n log n), in-place, not stable

Interview Questions

Further Reading

Binary Heaps and Priority Queues

• Introduction to Algorithms (CLRS), Chapter 6 — The definitive reference on binary heaps, heap sort, and priority queues. Covers correctness proofs and complexity analysis in depth.
• Algorithms (Sedgewick & Wayne), Chapter 2.4 — Practical coverage of priority queues with Java implementations, including indexed heaps used in Dijkstra’s algorithm.

Advanced Heap Variants

• Fibonacci Heaps — Fredman & Tarjan’s original paper (1987) on Fibonacci heaps with O(1) amortized decrease-key. Essential reading for understanding amortized analysis and advanced data structures.
• Binomial Heaps — CLRS Chapter 19 covers binomial heaps, a mergeable heap variant supporting O(log n) union operations.
• Pairing Heaps — Fredman et al. (1986). Simpler than Fibonacci heaps with excellent practical performance. Used in some implementations of Prim’s and Dijkstra’s algorithms.
• d-ary Heaps — Johnson (1975). Generalization where each node has d children. Higher d yields shorter trees but more expensive sift-down. Sweet spot is often d=4 or d=8 for cache-conscious workloads.

Applications Deep Dives

• Dijkstra’s Algorithm with Priority Queue — The classic shortest-path algorithm uses a min-heap to always process the closest unvisited node. With a binary heap, complexity is O((V+E) log V). With a Fibonacci heap (supporting O(1) decrease-key), it improves to O(V log V + E), though high constants often make binary heap preferable in practice.
• Heap Sort vs. Quick Sort — Both are O(n log n) on average, but heap sort is not stable and has worse cache behavior. Quick sort’s divide-and-conquer pattern leverages cache better for most inputs. Heap sort’s key advantage is its guaranteed O(n log n) worst case with O(1) extra space.
• Task Scheduling with Priority Queues — Operating systems use priority queues extensively for process scheduling. Real-time systems add priority inheritance to avoid inversion. See SCHED_FIFO and SCHED_RR in the Linux kernel scheduler for real-world examples.

Conclusion

Heaps give you O(1) access to the min or max with O(log n) inserts and deletes. That’s why they’re the natural choice for priority queues, Dijkstra’s, and heap sort. The key insight is that a heap is just an array pretending to be a tree—parent-child navigation is pure index math, no pointers needed. For Python priority queues, heapq with a tiebreaker counter handles same-priority ordering cleanly. If you’re building systems that need top-k tracking, medians, or sorted stream merging, heaps are usually the right call.