Heaps and Graphs

📖 Introduction

Heaps and graphs are two fundamental non-linear data structures that appear constantly in real-world applications. Every time I use a task scheduler with priorities, implement a social network feature, or work with network routing, I'm leveraging these structures.

Heaps gave me an elegant way to always access the "most important" element efficiently. Graphs taught me to model relationships between entities—from dependencies in build systems to recommendations in e-commerce.

---

🏔️ Heaps

A heap is a complete binary tree that satisfies the heap property.

Heap Property

# Min Heap: parent ≤ children (root is minimum)
# Max Heap: parent ≥ children (root is maximum)

# Array representation (0-indexed):
# Parent of i: (i - 1) // 2
# Left child of i: 2 * i + 1
# Right child of i: 2 * i + 2

Heap Implementation

from typing import List, Optional

class MinHeap:
    """
    Min heap implementation.
    """
    
    def __init__(self):
        self.heap: List[int] = []
    
    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 swap(self, i: int, j: int) -> None:
        self.heap[i], self.heap[j] = self.heap[j], self.heap[i]
    
    def push(self, val: int) -> None:
        """
        Insert element. O(log n).
        Add to end, then bubble up.
        """
        self.heap.append(val)
        self._bubble_up(len(self.heap) - 1)
    
    def _bubble_up(self, i: int) -> None:
        """Move element up until heap property restored."""
        while i > 0 and self.heap[self.parent(i)] > self.heap[i]:
            self.swap(i, self.parent(i))
            i = self.parent(i)
    
    def pop(self) -> Optional[int]:
        """
        Remove and return minimum. O(log n).
        Move last to root, then bubble down.
        """
        if not self.heap:
            return None
        
        if len(self.heap) == 1:
            return self.heap.pop()
        
        min_val = self.heap[0]
        self.heap[0] = self.heap.pop()
        self._bubble_down(0)
        
        return min_val
    
    def _bubble_down(self, i: int) -> None:
        """Move element down until heap property restored."""
        size = len(self.heap)
        
        while True:
            smallest = i
            left = self.left_child(i)
            right = self.right_child(i)
            
            if left < size and self.heap[left] < self.heap[smallest]:
                smallest = left
            
            if right < size and self.heap[right] < self.heap[smallest]:
                smallest = right
            
            if smallest == i:
                break
            
            self.swap(i, smallest)
            i = smallest
    
    def peek(self) -> Optional[int]:
        """Return minimum without removing. O(1)."""
        return self.heap[0] if self.heap else None
    
    def __len__(self) -> int:
        return len(self.heap)


# Usage
heap = MinHeap()
for val in [4, 1, 7, 3, 8, 5]:
    heap.push(val)

while heap:
    print(heap.pop())  # 1, 3, 4, 5, 7, 8

Heapify: Building a Heap

def heapify(arr: List[int]) -> None:
    """
    Convert array to min heap in-place.
    O(n) - not O(n log n)!
    """
    n = len(arr)
    
    # Start from last non-leaf and bubble down
    for i in range(n // 2 - 1, -1, -1):
        _bubble_down(arr, i, n)


def _bubble_down(arr: List[int], i: int, size: int) -> None:
    while True:
        smallest = i
        left = 2 * i + 1
        right = 2 * i + 2
        
        if left < size and arr[left] < arr[smallest]:
            smallest = left
        if right < size and arr[right] < arr[smallest]:
            smallest = right
        
        if smallest == i:
            break
        
        arr[i], arr[smallest] = arr[smallest], arr[i]
        i = smallest

Python's heapq Module

import heapq

# heapq provides min heap operations on a list

nums = [4, 1, 7, 3, 8, 5]

# Convert list to heap in-place O(n)
heapq.heapify(nums)
print(nums)  # [1, 3, 5, 4, 8, 7]

# Push element O(log n)
heapq.heappush(nums, 2)

# Pop minimum O(log n)
min_val = heapq.heappop(nums)  # 1

# Push and pop in one operation
heapq.heappushpop(nums, 0)  # Push 0, return min

# Pop and push in one operation
heapq.heapreplace(nums, 10)  # Return min, then push 10

# Get n smallest/largest
smallest_3 = heapq.nsmallest(3, nums)
largest_3 = heapq.nlargest(3, nums)


# Max heap trick: negate values
max_heap = []
heapq.heappush(max_heap, -5)
heapq.heappush(max_heap, -3)
heapq.heappush(max_heap, -8)
max_val = -heapq.heappop(max_heap)  # 8

Heap Time Complexity

Operation	Time
peek	O(1)
push	O(log n)
pop	O(log n)
heapify	O(n)
search	O(n)

---

📊 Heap Problems

Kth Largest Element

import heapq

def find_kth_largest(nums: list[int], k: int) -> int:
    """
    Find kth largest element.
    Time: O(n log k), Space: O(k)
    """
    # Min heap of size k
    min_heap = nums[:k]
    heapq.heapify(min_heap)
    
    for num in nums[k:]:
        if num > min_heap[0]:
            heapq.heapreplace(min_heap, num)
    
    return min_heap[0]


# Alternative: use max heap
def find_kth_largest_max_heap(nums: list[int], k: int) -> int:
    """Time: O(n + k log n)."""
    max_heap = [-num for num in nums]
    heapq.heapify(max_heap)
    
    for _ in range(k - 1):
        heapq.heappop(max_heap)
    
    return -heapq.heappop(max_heap)

Top K Frequent Elements

from collections import Counter
import heapq

def top_k_frequent(nums: list[int], k: int) -> list[int]:
    """
    Return k most frequent elements.
    Time: O(n log k), Space: O(n)
    """
    count = Counter(nums)
    
    # Min heap by frequency
    heap = []
    
    for num, freq in count.items():
        heapq.heappush(heap, (freq, num))
        if len(heap) > k:
            heapq.heappop(heap)
    
    return [num for freq, num in heap]


# Using nlargest
def top_k_frequent_nlargest(nums: list[int], k: int) -> list[int]:
    count = Counter(nums)
    return [num for num, freq in heapq.nlargest(k, count.items(), key=lambda x: x[1])]

Merge K Sorted Lists

import heapq
from typing import List, Optional

class ListNode:
    def __init__(self, val=0, next=None):
        self.val = val
        self.next = next


def merge_k_lists(lists: List[Optional[ListNode]]) -> Optional[ListNode]:
    """
    Merge k sorted linked lists.
    Time: O(n log k), Space: O(k)
    """
    # Min heap with (value, index, node)
    # Index used for tiebreaker
    heap = []
    
    for i, node in enumerate(lists):
        if node:
            heapq.heappush(heap, (node.val, i, node))
    
    dummy = ListNode()
    current = dummy
    
    while heap:
        val, i, node = heapq.heappop(heap)
        current.next = node
        current = current.next
        
        if node.next:
            heapq.heappush(heap, (node.next.val, i, node.next))
    
    return dummy.next

Find Median from Data Stream

import heapq

class MedianFinder:
    """
    Find median of stream efficiently.
    Two heaps: max heap for lower half, min heap for upper half.
    """
    
    def __init__(self):
        self.lower = []  # Max heap (negated)
        self.upper = []  # Min heap
    
    def add_num(self, num: int) -> None:
        """O(log n)."""
        # Add to max heap (lower half)
        heapq.heappush(self.lower, -num)
        
        # Move max of lower to upper
        heapq.heappush(self.upper, -heapq.heappop(self.lower))
        
        # Balance: lower can have at most 1 more than upper
        if len(self.upper) > len(self.lower):
            heapq.heappush(self.lower, -heapq.heappop(self.upper))
    
    def find_median(self) -> float:
        """O(1)."""
        if len(self.lower) > len(self.upper):
            return -self.lower[0]
        
        return (-self.lower[0] + self.upper[0]) / 2


# Usage
mf = MedianFinder()
mf.add_num(1)
mf.add_num(2)
print(mf.find_median())  # 1.5
mf.add_num(3)
print(mf.find_median())  # 2

---

🕸️ Graphs

A graph consists of vertices (nodes) and edges connecting them.

Graph Terminology

# Directed: edges have direction (A → B)
# Undirected: edges go both ways (A — B)
# Weighted: edges have values (A --5--> B)
# Cycle: path that starts and ends at same vertex
# DAG: Directed Acyclic Graph
# Connected: path exists between every pair of vertices
# Sparse: few edges (E << V²)
# Dense: many edges (E ≈ V²)

Graph Representations

from collections import defaultdict
from typing import List, Dict, Set, Tuple

# 1. Adjacency List (Most common)
# Good for sparse graphs
# Space: O(V + E)
graph_adj_list: Dict[str, List[str]] = {
    'A': ['B', 'C'],
    'B': ['A', 'C', 'D'],
    'C': ['A', 'B'],
    'D': ['B']
}

# Using defaultdict
graph = defaultdict(list)
edges = [('A', 'B'), ('A', 'C'), ('B', 'C'), ('B', 'D')]
for u, v in edges:
    graph[u].append(v)
    graph[v].append(u)  # Undirected


# 2. Adjacency Matrix
# Good for dense graphs
# Space: O(V²)
# Access: O(1)
vertices = ['A', 'B', 'C', 'D']
v_to_idx = {v: i for i, v in enumerate(vertices)}

matrix = [[0] * 4 for _ in range(4)]
edges = [('A', 'B'), ('A', 'C'), ('B', 'C'), ('B', 'D')]
for u, v in edges:
    i, j = v_to_idx[u], v_to_idx[v]
    matrix[i][j] = 1
    matrix[j][i] = 1  # Undirected


# 3. Edge List
# Simple but less efficient for traversal
edges: List[Tuple[str, str, int]] = [
    ('A', 'B', 5),
    ('A', 'C', 3),
    ('B', 'D', 2),
]


# 4. Adjacency Set (for fast edge lookup)
graph_set: Dict[str, Set[str]] = defaultdict(set)
for u, v in [('A', 'B'), ('A', 'C')]:
    graph_set[u].add(v)
    graph_set[v].add(u)

# Check edge exists: O(1)
has_edge = 'B' in graph_set['A']

Comparison

Representation	Space	Check Edge	Get Neighbors	Add Edge
Adjacency List	O(V+E)	O(degree)	O(1)	O(1)
Adjacency Matrix	O(V²)	O(1)	O(V)	O(1)
Edge List	O(E)	O(E)	O(E)	O(1)

---

🔍 Graph Traversals

Breadth-First Search (BFS)

Explore level by level. Use queue.

from collections import deque
from typing import Dict, List, Set

def bfs(graph: Dict[str, List[str]], start: str) -> List[str]:
    """
    BFS traversal.
    Time: O(V + E), Space: O(V)
    """
    visited = {start}
    queue = deque([start])
    result = []
    
    while queue:
        node = queue.popleft()
        result.append(node)
        
        for neighbor in graph[node]:
            if neighbor not in visited:
                visited.add(neighbor)
                queue.append(neighbor)
    
    return result


def bfs_shortest_path(
    graph: Dict[str, List[str]], 
    start: str, 
    end: str
) -> List[str]:
    """
    Find shortest path (unweighted).
    """
    if start == end:
        return [start]
    
    visited = {start}
    queue = deque([(start, [start])])
    
    while queue:
        node, path = queue.popleft()
        
        for neighbor in graph[node]:
            if neighbor == end:
                return path + [neighbor]
            
            if neighbor not in visited:
                visited.add(neighbor)
                queue.append((neighbor, path + [neighbor]))
    
    return []  # No path found


def bfs_distance(
    graph: Dict[str, List[str]], 
    start: str
) -> Dict[str, int]:
    """
    Find distance to all nodes from start.
    """
    distances = {start: 0}
    queue = deque([start])
    
    while queue:
        node = queue.popleft()
        
        for neighbor in graph[node]:
            if neighbor not in distances:
                distances[neighbor] = distances[node] + 1
                queue.append(neighbor)
    
    return distances

Depth-First Search (DFS)

Explore as deep as possible. Use recursion or stack.

from typing import Dict, List, Set

def dfs_recursive(
    graph: Dict[str, List[str]], 
    start: str,
    visited: Set[str] = None
) -> List[str]:
    """
    DFS using recursion.
    Time: O(V + E), Space: O(V)
    """
    if visited is None:
        visited = set()
    
    visited.add(start)
    result = [start]
    
    for neighbor in graph[start]:
        if neighbor not in visited:
            result.extend(dfs_recursive(graph, neighbor, visited))
    
    return result


def dfs_iterative(graph: Dict[str, List[str]], start: str) -> List[str]:
    """
    DFS using explicit stack.
    """
    visited = set()
    stack = [start]
    result = []
    
    while stack:
        node = stack.pop()
        
        if node not in visited:
            visited.add(node)
            result.append(node)
            
            # Add neighbors in reverse for same order as recursive
            for neighbor in reversed(graph[node]):
                if neighbor not in visited:
                    stack.append(neighbor)
    
    return result


def dfs_all_paths(
    graph: Dict[str, List[str]], 
    start: str, 
    end: str
) -> List[List[str]]:
    """
    Find all paths from start to end.
    """
    paths = []
    
    def backtrack(node, path):
        if node == end:
            paths.append(path[:])
            return
        
        for neighbor in graph[node]:
            if neighbor not in path:
                path.append(neighbor)
                backtrack(neighbor, path)
                path.pop()
    
    backtrack(start, [start])
    return paths

BFS vs DFS

Aspect	BFS	DFS
Data Structure	Queue	Stack/Recursion
Shortest Path	Yes (unweighted)	No
Memory	O(width)	O(height)
Use Cases	Level-order, shortest path	Cycle detection, topological sort

---

📊 Common Graph Problems

Number of Islands

from typing import List

def num_islands(grid: List[List[str]]) -> int:
    """
    Count connected components of '1's.
    Time: O(m*n), Space: O(m*n) for recursion
    """
    if not grid:
        return 0
    
    rows, cols = len(grid), len(grid[0])
    count = 0
    
    def dfs(r, c):
        if (r < 0 or r >= rows or 
            c < 0 or c >= cols or 
            grid[r][c] != '1'):
            return
        
        grid[r][c] = '#'  # Mark visited
        
        dfs(r + 1, c)
        dfs(r - 1, c)
        dfs(r, c + 1)
        dfs(r, c - 1)
    
    for r in range(rows):
        for c in range(cols):
            if grid[r][c] == '1':
                count += 1
                dfs(r, c)
    
    return count

Clone Graph

from typing import Optional, Dict

class Node:
    def __init__(self, val=0, neighbors=None):
        self.val = val
        self.neighbors = neighbors if neighbors else []


def clone_graph(node: Optional[Node]) -> Optional[Node]:
    """
    Deep copy a graph.
    Time: O(V + E), Space: O(V)
    """
    if not node:
        return None
    
    clones: Dict[Node, Node] = {}
    
    def dfs(original):
        if original in clones:
            return clones[original]
        
        copy = Node(original.val)
        clones[original] = copy
        
        for neighbor in original.neighbors:
            copy.neighbors.append(dfs(neighbor))
        
        return copy
    
    return dfs(node)

Course Schedule (Cycle Detection)

from collections import defaultdict, deque
from typing import List

def can_finish_dfs(num_courses: int, prerequisites: List[List[int]]) -> bool:
    """
    Detect if cycle exists (DFS with colors).
    Time: O(V + E), Space: O(V + E)
    """
    graph = defaultdict(list)
    for course, prereq in prerequisites:
        graph[prereq].append(course)
    
    # 0: unvisited, 1: visiting, 2: visited
    state = [0] * num_courses
    
    def has_cycle(course):
        if state[course] == 1:  # Cycle detected
            return True
        if state[course] == 2:  # Already processed
            return False
        
        state[course] = 1  # Mark as visiting
        
        for next_course in graph[course]:
            if has_cycle(next_course):
                return True
        
        state[course] = 2  # Mark as visited
        return False
    
    for course in range(num_courses):
        if has_cycle(course):
            return False
    
    return True


def can_finish_bfs(num_courses: int, prerequisites: List[List[int]]) -> bool:
    """
    Kahn's algorithm (BFS topological sort).
    """
    graph = defaultdict(list)
    in_degree = [0] * num_courses
    
    for course, prereq in prerequisites:
        graph[prereq].append(course)
        in_degree[course] += 1
    
    # Start with nodes that have no prerequisites
    queue = deque([c for c in range(num_courses) if in_degree[c] == 0])
    processed = 0
    
    while queue:
        course = queue.popleft()
        processed += 1
        
        for next_course in graph[course]:
            in_degree[next_course] -= 1
            if in_degree[next_course] == 0:
                queue.append(next_course)
    
    return processed == num_courses

Word Ladder

from collections import deque
from typing import List, Set

def ladder_length(
    begin_word: str, 
    end_word: str, 
    word_list: List[str]
) -> int:
    """
    Find shortest transformation sequence.
    Time: O(m² * n) where m = word length, n = words
    """
    word_set = set(word_list)
    
    if end_word not in word_set:
        return 0
    
    queue = deque([(begin_word, 1)])
    
    while queue:
        word, length = queue.popleft()
        
        if word == end_word:
            return length
        
        # Try changing each character
        for i in range(len(word)):
            for c in 'abcdefghijklmnopqrstuvwxyz':
                next_word = word[:i] + c + word[i+1:]
                
                if next_word in word_set:
                    word_set.remove(next_word)
                    queue.append((next_word, length + 1))
    
    return 0

Bipartite Graph

from collections import deque
from typing import List

def is_bipartite(graph: List[List[int]]) -> bool:
    """
    Check if graph can be 2-colored.
    Time: O(V + E), Space: O(V)
    """
    n = len(graph)
    color = [0] * n  # 0: uncolored, 1: color A, -1: color B
    
    for start in range(n):
        if color[start] != 0:
            continue
        
        queue = deque([start])
        color[start] = 1
        
        while queue:
            node = queue.popleft()
            
            for neighbor in graph[node]:
                if color[neighbor] == 0:
                    color[neighbor] = -color[node]
                    queue.append(neighbor)
                elif color[neighbor] == color[node]:
                    return False
    
    return True

---

📝 Practice Exercises

Exercise 1: K Closest Points to Origin

def k_closest(points: List[List[int]], k: int) -> List[List[int]]:
    """
    Find k closest points to origin (0, 0).
    Use Euclidean distance.
    """
    pass

Solution

import heapq

def k_closest(points: List[List[int]], k: int) -> List[List[int]]:
    # Max heap with size k
    heap = []
    
    for x, y in points:
        dist = x * x + y * y
        heapq.heappush(heap, (-dist, [x, y]))
        
        if len(heap) > k:
            heapq.heappop(heap)
    
    return [point for _, point in heap]

Exercise 2: Surrounded Regions

def solve(board: List[List[str]]) -> None:
    """
    Capture all regions surrounded by 'X'.
    'O' on border cannot be captured.
    Modify board in-place.
    
    Example:
    X X X X      X X X X
    X O O X  ->  X X X X
    X X O X      X X X X
    X O X X      X O X X
    """
    pass

Solution

def solve(board: List[List[str]]) -> None:
    if not board:
        return
    
    rows, cols = len(board), len(board[0])
    
    def dfs(r, c):
        if (r < 0 or r >= rows or 
            c < 0 or c >= cols or 
            board[r][c] != 'O'):
            return
        
        board[r][c] = 'T'  # Temporary mark
        dfs(r + 1, c)
        dfs(r - 1, c)
        dfs(r, c + 1)
        dfs(r, c - 1)
    
    # Mark 'O's connected to border
    for r in range(rows):
        dfs(r, 0)
        dfs(r, cols - 1)
    
    for c in range(cols):
        dfs(0, c)
        dfs(rows - 1, c)
    
    # Capture surrounded and restore border-connected
    for r in range(rows):
        for c in range(cols):
            if board[r][c] == 'O':
                board[r][c] = 'X'
            elif board[r][c] == 'T':
                board[r][c] = 'O'

Exercise 3: Network Delay Time

def network_delay_time(times: List[List[int]], n: int, k: int) -> int:
    """
    Signal sent from node k. Return time for all nodes to receive.
    times[i] = [source, target, time]
    Return -1 if not all nodes receive.
    """
    pass

Solution

import heapq
from collections import defaultdict

def network_delay_time(times: List[List[int]], n: int, k: int) -> int:
    # Dijkstra's algorithm
    graph = defaultdict(list)
    for u, v, w in times:
        graph[u].append((v, w))
    
    dist = {k: 0}
    heap = [(0, k)]
    
    while heap:
        d, node = heapq.heappop(heap)
        
        if d > dist.get(node, float('inf')):
            continue
        
        for neighbor, weight in graph[node]:
            new_dist = d + weight
            if new_dist < dist.get(neighbor, float('inf')):
                dist[neighbor] = new_dist
                heapq.heappush(heap, (new_dist, neighbor))
    
    if len(dist) != n:
        return -1
    
    return max(dist.values())

---

🔑 Key Takeaways

1. Heaps provide O(log n) insert/remove and O(1) peek for min/max

2. Python's heapq is a min heap—negate for max heap

3. Graphs model relationships; choose representation by use case

4. BFS finds shortest path in unweighted graphs

5. DFS is better for exploring all paths, cycle detection

6. Grid problems are often graph problems in disguise

---

🚀 What's Next?

In the next article, we'll explore sorting algorithms:

• Comparison-based sorts

• Stable vs unstable sorting

• Python's Timsort

• When to use which algorithm

Next: Article 10 - Sorting

---

📚 References

• CLRS Chapter 6: Heapsort

• CLRS Chapter 22: Graph Algorithms

• Python heapq module documentation