IntermediateStacks, Queues, Heaps

Priority Queue / Heap

Use heaps for top-k, k-way merge, scheduling, and streaming medians

Time: O(log n) ops
Space: O(n)
Pattern Overview

Core Idea

Heaps maintain partial order to get min/max quickly; support merging/scheduling efficiently.

When to Use

Use when repeatedly extracting min/max, merging k sorted inputs, or balancing two halves for median.

General Recognition Cues
General indicators that this pattern might be applicable
  • Top-k
  • k-way merge
  • Streaming median
  • Greedy scheduling
Pattern Variants & Approaches
Different methodologies and implementations for this pattern
1

Top-k

Size-k heap to keep best candidates

When to Use This Variant
  • Keep k best
Use Case
Top-k elements/frequencies
Advantages
  • O(n log k)
  • Streaming friendly
Implementation
import heapq
def top_k(nums, k):
    heap = []
    for x in nums:
        if len(heap) < k: heapq.heappush(heap, x)
        elif x > heap[0]: heapq.heapreplace(heap, x)
    return sorted(heap, reverse=True)
2

k-Way Merge

Push head of each list/array; pop/push next

When to Use This Variant
  • Merge sorted inputs
Use Case
Merging streams/lists
Advantages
  • O(n log k)
  • Simple
Implementation
import heapq
def merge_k_sorted(arrs):
    heap = []
    for i, arr in enumerate(arrs):
        if arr: heapq.heappush(heap, (arr[0], i, 0))
    res = []
    while heap:
        val, i, j = heapq.heappop(heap)
        res.append(val)
        if j+1 < len(arrs[i]): heapq.heappush(heap, (arrs[i][j+1], i, j+1))
    return res
3

Task Scheduling

Greedy by frequency with cooldown

When to Use This Variant
  • Cooldown or spacing
Use Case
Schedule tasks
Advantages
  • Minimize idle
  • Optimal spacing
Implementation
import heapq
from collections import Counter, deque
def least_interval(tasks, n):
    freq = Counter(tasks)
    heap = [-c for c in freq.values()]
    heapq.heapify(heap)
    time = 0
    q = deque()  # (ready_time, count)
    while heap or q:
        time += 1
        if heap:
            cnt = 1 + heapq.heappop(heap)  # negative counts
            if cnt != 0:
                q.append((time + n, cnt))
        if q and q[0][0] == time:
            heapq.heappush(heap, q.popleft()[1])
    return time
4

Two Heaps

Max-heap left + min-heap right for median

When to Use This Variant
  • Online median
Use Case
Median stream
Advantages
  • O(log n) update
  • O(1) query
Implementation
import heapq
class MedianFinder:
    def __init__(self):
        self.small = []  # max-heap via negatives
        self.large = []  # min-heap
    def addNum(self, num: int) -> None:
        if not self.small or num <= -self.small[0]:
            heapq.heappush(self.small, -num)
        else:
            heapq.heappush(self.large, num)
        # balance
        if len(self.small) > len(self.large) + 1:
            heapq.heappush(self.large, -heapq.heappop(self.small))
        if len(self.large) > len(self.small):
            heapq.heappush(self.small, -heapq.heappop(self.large))
    def findMedian(self) -> float:
        if len(self.small) > len(self.large):
            return float(-self.small[0])
        return (-self.small[0] + self.large[0]) / 2.0
Common Pitfalls
Watch out for these common mistakes
  • Lazy deletion bookkeeping
  • Comparator direction
  • Heapifying cost and updates
Code Templates
Compare implementations across different languages
def template(*args, **kwargs):
    # General template placeholder
    pass
Example Problems
Classic LeetCode problems using this pattern
Complexity Analysis
Time Complexity
O(log n) ops

Push/pop heap operations are logarithmic.

Space Complexity
O(n)

Heap stores elements.

Ready to test your knowledge?

Test your pattern recognition with interactive questions