Introduction
A priority queue is an abstract data type that operates similarly to a regular queue or stack, but where each element has a distinct 'priority' associated with it. In a priority queue, an element with high priority is served before an element with low priority. While binary search trees and balanced trees like AVL or Red-Black trees can implement priority queues, heaps offer optimal performance profiles for core operations without the memory overhead of node pointers.
This article examines heap data structures from the ground up. We will explore the array representation of complete binary trees, derive the arithmetic for parent-child navigation, implement core structural invariants (sift_up and sift_down), build an $O(N)$ bottom-up heap constructor, and apply these concepts to solve the classic 'Top-K Elements' streaming problem.
Array Representation of Complete Binary Trees
A binary heap is a complete binary tree: every level of the tree, except possibly the last, is completely filled, and all nodes in the last level are as far left as possible. Because of this structural rigidity, heaps do not require pointer-based node allocations. Instead, they map directly to contiguous memory locations (arrays or dynamic lists).
For any node located at index $i$ (0-indexed):
- Left Child Index: $2i + 1$
- Right Child Index: $2i + 2$
- Parent Index: $\lfloor \frac{i - 1}{2} \rfloor$
Mathematical Verification
Let us verify this mapping visually using a small array:
Index: 0 1 2 3 4 5
Value: [3, 9, 5, 12, 10, 7]
Tree Topology:
(3) [i=0]
/ \
(9)[i=1] (5)[i=2]
/ \\ \
(12)[i=3] (10)[i=4] (7)[i=5]
- For node at $i = 1$ (value
9): Left child is $2(1) + 1 = 3$ (value12), right child is $2(1) + 2 = 4$ (value10). Correct. - For node at $i = 4$ (value
10): Parent is $\lfloor \frac{4 - 1}{2} \rfloor = \lfloor 1.5 \rfloor = 1$ (value9). Correct.
Heap Invariants and Core Operations
A Min-Heap maintains the invariant that the value of each node is less than or equal to the values of its children. Consequently, the minimum element is always at the root index $0$.
When we insert or remove elements, we temporarily violate this invariant. We restore it using two foundational mechanics: Bubble-Up (sift_up) and Bubble-Down (sift_down).
1. Insertion and Bubble-Up (sift_up)
To insert an element, append it to the end of the array (preserving the complete binary tree shape property). This may violate the min-heap invariant if the new element is smaller than its parent. We fix this by comparing the node with its parent and swapping them if necessary, repeating the process up to the root.
2. Extraction and Bubble-Down (sift_down)
To extract the minimum element (the root), we cannot simply delete index $0$ without breaking the array. Instead, we swap the root with the last element in the array, pop the last element, and then restore the heap invariant by bubbling the new root down. At each step, we compare the parent with its smallest child and swap if the parent is larger.
Python Implementation of Min-Heap
class MinHeap:
def __init__(self):
self.heap = []
def _parent(self, i: int) -> int:
return (i - 1) // 2
def _left_child(self, i: int) -> int:
return 2 * i + 1
def _right_child(self, i: int) -> int:
return 2 * i + 2
def peek(self):
if not self.heap:
raise IndexError("Peek from empty heap")
return self.heap[0]
def push(self, val):
self.heap.append(val)
self._sift_up(len(self.heap) - 1)
def pop(self):
if not self.heap:
raise IndexError("Pop from empty heap")
if len(self.heap) == 1:
return self.heap.pop()
root = self.heap[0]
# Move last element to root
self.heap[0] = self.heap.pop()
self._sift_down(0)
return root
def _sift_up(self, i: int):
while i > 0 and self.heap[i] < self.heap[self._parent(i)]:
p = self._parent(i)
self.heap[i], self.heap[p] = self.heap[p], self.heap[i]
i = p
def _sift_down(self, i: int):
size = len(self.heap)
while self._left_child(i) < size:
left = self._left_child(i)
right = self._right_child(i)
smallest = i
if self.heap[left] < self.heap[smallest]:
smallest = left
if right < size and self.heap[right] < self.heap[smallest]:
smallest = right
if smallest != i:
self.heap[i], self.heap[smallest] = self.heap[smallest], self.heap[i]
i = smallest
else:
break
Heapify in $O(N)$ Time
An intuitive approach to building a heap from an arbitrary array is to start with an empty heap and call push() $N$ times. Since each insertion takes $O(\log K)$ time where $K$ is the current heap size, the total time complexity of this naive approach is $O(N \log N)$.
However, we can build a heap in $O(N)$ time using a bottom-up approach known as Heapify (or Floyd's Building Algorithm).
Why is Heapify $O(N)$?
In a complete binary tree containing $N$ nodes:
- Approximately $\frac{N}{2}$ nodes are leaves (at the bottom level). Leaves require 0 operations to sift down.
- Approximately $\frac{N}{4}$ nodes are at the level right above the leaves. They can travel at most 1 level down.
- Approximately $\frac{N}{8}$ nodes can travel at most 2 levels down.
Summing the maximum work across all levels yields the series:
$$\sum_{h=0}^{\log N} \frac{N}{2^{h+1}} \cdot O(h) = O(N \sum_{h=0}^{\log N} \frac{h}{2^h}) = O(N)$$
Because the infinite series $\sum \frac{h}{2^h}$ converges to 2, the asymptotic complexity is linear $O(N)$.
Implementing Heapify
To build a heap from an unsorted list, we locate the last non-leaf node, which is always located at index $\lfloor \frac{N - 1}{2} \rfloor$, and iterate backward down to index $0$, calling sift_down() on each.
@classmethod
def heapify(cls, arr: list) -> 'MinHeap':
instance = cls()
instance.heap = list(arr) # Shallow copy
n = len(instance.heap)
# Start from the last non-leaf node and sift down
for i in range((n - 2) // 2, -1, -1):
instance._sift_down(i)
return instance
Bounded Priority Queues for Streaming Data
A common interview and production pattern is solving the Top-K Elements problem. Given a data stream of unknown (or massive) length $N$, find the $K$ largest elements.
Naive Approaches vs. Heaps
- Full Sort: Store all incoming elements, sort the array, and take the last $K$ elements. Time Complexity: $O(N \log N)$, Space Complexity: $O(N)$. This fails when $N$ exceeds available memory.
-
Min-Heap Approach: Maintain a bounded min-heap of size $K$. For every incoming element:
- If the heap has fewer than $K$ elements, push the element.
- If the heap has $K$ elements and the incoming element is larger than the heap's minimum element (the root), pop the root and push the new element.
Complexity Analysis of Bounded Min-Heap
- Time Complexity: Processing $N$ elements where each insertion/deletion costs $O(\log K)$ results in $O(N \log K)$. If $K \ll N$, this approaches $O(N)$ time.
- Space Complexity: Strictly $O(K)$, irrespective of how massive the data stream $N$ becomes.
Python Implementation: Top-K Stream Processor
class TopKStreamProcessor:
def __init__(self, k: int):
self.k = k
self.min_heap = []
def add_element(self, val):
if len(self.min_heap) < self.k:
# Python's heapq is a min-heap by default
import heapq
heapq.heappush(self.min_heap, val)
elif val > self.min_heap[0]:
heapq.heapreplace(self.min_heap, val)
def get_top_k(self):
# Return sorted descending for presentation
return sorted(self.min_heap, reverse=True)
# Usage Example:
stream = [45, 12, 89, 33, 22, 90, 10, 5, 77, 102]
processor = TopKStreamProcessor(k=3)
for num in stream:
processor.add_element(num)
print(processor.get_top_k()) # Output: [102, 90, 89]
Conclusion
Heaps and priority queues provide structural elegance without pointer overhead, making them ideal for high-throughput, latency-sensitive applications. By leveraging array index arithmetic, maintaining structural invariants via sift_up and sift_down, building structures in linear time via bottom-up heapification, and capping memory footprints with bounded priority queues, engineers can build robust systems capable of processing unbounded streaming data efficiently.
Top comments (0)