Topic 10: Heap / Priority Queue

Key-Intuition: Heaps efficiently maintain order-statistics in dynamic datasets, great for priority-based retrieval and merging sorted sequences.

We can expect them to be especially useful when used as aux datastructures, in conjunction with other topic areas.

Canonical Questions #

Top-K and Median Finding #

• Objective
  • Efficiently find the k-th largest, smallest, or median elements from streaming data or static arrays.

Problems #

1. Kth Largest Element (215)

   Oh wait we can do this via heapq, just have to use the api correctly else will be too slow. Since we need kth largest, we can effectively keep a fixed capacity in our heap by checking the size of it and if it’s already k elements large, we can do heapreplace instead of heappush. Other attempts to manually push and pop from the heap manually will have ridiculous performance overheads.

   1 2 3 4 5 6 7 8 9 10 11 12

   import heapq class Solution: def findKthLargest(self, nums, k): heap = [] for num in nums: if len(heap) < k: heapq.heappush(heap, num) else: if num > heap[0]: heapq.heapreplace(heap, num) return heap[0]

   Code Snippet 1: Kth Largest Element (215) – using heappush, heapreplace

   Alternatively, for this question, we can use the quickselect algo, though that is more manual.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43

   import random from typing import List class Solution: def findKthLargest(self, nums: List[int], k: int) -> int: def quickselect(left, right, target): if left == right: return nums[left] pivot_index = random.randint(left, right) # Partition around pivot and return final pivot position new_pivot = partition(left, right, pivot_index) if new_pivot == target: return nums[new_pivot] elif new_pivot < target: return quickselect(new_pivot + 1, right, target) else: return quickselect(left, new_pivot - 1, target) def partition(left, right, pivot_index): """ Returns the pivot idx after paritioning left and right segments based on the element values. The property it maintains thereafter is the all to the right of pivot are STRICTLY MORE THAN pivot. All the left of hte pivot are STRICTLY LESS THAN pivot. """ pivot_value = nums[pivot_index] nums[pivot_index], nums[right] = nums[right], nums[pivot_index] store_index = left for i in range(left, right): if nums[i] < pivot_value: nums[store_index], nums[i] = nums[i], nums[store_index] store_index += 1 # move pivot to its final place nums[store_index], nums[right] = nums[right], nums[store_index] return store_index n = len(nums) target = n - k # index of k-th largest in sorted order return quickselect(0, n - 1, target)

   Code Snippet 2: Kth Largest Element (215) – using quickselect approach
   Pattern Extraction

   Core Insight: Maintain a min-heap of size \(k\); after processing all elements, the heap root is the $k$-th largest. Alternatively, use Quickselect for expected \(O(n)\). Pattern Name: Top-K with Min-Heap of Size K Canonical Section: Heap > Top-K Selection Recognition Signals:

   • “$k$-th largest/smallest” element
   • Don’t need the full sorted order, just one element
   • \(O(n \log k)\) is acceptable (better than \(O(n \log n)\) full sort)

   Anti-Signals:

   • Need the top \(k\) elements sorted – heap gives them unsorted; need a final sort
   • Need online/streaming $k$-th element – heap naturally supports this
   • \(k\) is close to \(n\) – sorting might be simpler

   Decision Point: Heap gives guaranteed \(O(n \log k)\). Quickselect gives expected \(O(n)\) but worst-case \(O(n^2)\). For interviews, heap is safer and cleaner to implement.

   Pitfalls and Misconceptions

   1. Trap: Using a max-heap of all \(n\) elements and popping \(k\) times. Why: Building a max-heap is \(O(n)\), but popping \(k\) times is \(O(k \log n)\). For large \(n\) and small \(k\), the min-heap of size \(k\) (\(O(n \log k)\)) is better. Correction: Use a min-heap of size \(k\). Push each element; if heap size exceeds \(k\), pop the minimum. The root is the $k$-th largest.
   2. Trap: Forgetting that Python’s heapq is a min-heap, not max-heap. Why: For $k$-th largest, you want a min-heap (the root is the smallest of the top \(k\), which is the $k$-th largest). Negating values for a “max-heap” is unnecessary here. Correction: For $k$-th largest, min-heap of size \(k\) is correct as-is.
   Problem Mutations

   1. What if elements arrived in a stream? (stress-tests: static array) Same min-heap of size \(k\). Each new element: push, pop if size > \(k\). The root is always the current $k$-th largest. Connects to LC 703 Kth Largest Element in a Stream.
   2. What if you needed the median instead of $k$-th? (stress-tests: fixed \(k\) vs dynamic) Use two heaps: max-heap for lower half, min-heap for upper half. Balance sizes. Connects to LC 295 Find Median from Data Stream.
   3. What if you needed expected \(O(n)\)? (stress-tests: \(O(n \log k)\)) Quickselect (partition-based selection) achieves expected \(O(n)\). But worst-case \(O(n^2)\) unless randomised.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Standardised annotation dropdowns for Kth Largest Element in an Array generated via batch canonical enrichment pass. Review against personal solving experience and adjust.
   

2. Kth Largest Element in a Stream (703)

   the only thing that matters here is that we keep a fixed capacity heap “manually” and it’s good enough for us. There’s no need to negate this because we let the structure handle the order, don’t short circuit that part.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

   import heapq class KthLargest: def __init__(self, k: int, nums: list[int]): self.k = k self.heap = nums heapq.heapify(self.heap) # prune it to only keep top k elements while len(self.heap) > self.k: heapq.heappop(self.heap) def add(self, val: int) -> int: heapq.heappush(self.heap, val) if len(self.heap) > self.k: heapq.heappop(self.heap) return self.heap[0]

   Code Snippet 3: Kth Largest Element in a Stream (703)

   Here I use heapreplace to make the fixed capacity handling a little more idiomatic:

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17

   import heapq class KthLargest: def __init__(self, k, nums): self.k = k self.heap = nums heapq.heapify(self.heap) while len(self.heap) > k: heapq.heappop(self.heap) def add(self, val): if len(self.heap) < self.k: heapq.heappush(self.heap, val) else: # Only pushpop if val is bigger, otherwise kth largest does not change if val > self.heap[0]: heapq.heapreplace(self.heap, val) return self.heap[0]

   Code Snippet 4: Kth Largest Element in a Stream (703)
   Pattern Extraction

   Core Insight: Maintain a min-heap of size \(k\). The root is always the $k$-th largest. For each new element: if larger than root, push and pop to maintain size \(k\). Pattern Name: Online Top-K with Min-Heap Canonical Section: Heap > Top-K Selection (Streaming) Recognition Signals:

   • “$k$-th largest in a stream” (online/dynamic)
   • \(O(\log k)\) per insertion
   • Heap size = \(k\)

   Anti-Signals:

   • Static array – just sort or use quickselect

   Decision Point: This is the streaming version of LC 215. Same min-heap of size \(k\).

   Pitfalls and Misconceptions

   1. Trap: Using a max-heap or maintaining all elements. Why: Max-heap of all elements gives \(O(n)\) space and \(O(k \log n)\) per query. Min-heap of size \(k\) is \(O(k)\) space and \(O(\log k)\) per insertion. Correction: Min-heap of size \(k\). Root = $k$-th largest.
   Problem Mutations

   1. What if you needed the $k$-th smallest? (stress-tests: largest vs smallest) Use a max-heap of size \(k\). Root = $k$-th smallest.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Batch annotation for Kth Largest Element in a Stream.
   

3. ⭐️ Find Median from Data Stream (295)

   Streams usually we can consider using a balancing approach a max and a min heap (to get the median).

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

   import heapq class MedianFinder: def __init__(self): self.small = [] # Max heap (as negative numbers) self.large = [] # Min heap def addNum(self, num: int) -> None: # Python heapq is minheap, so push negative for maxheap heapq.heappush(self.small, -num) # All numbers in small must be <= numbers in large, so we take the largest number and shift it to the large heap heapq.heappush(self.large, -heapq.heappop(self.small)) # Balance: small can be larger by 1, so we rebalance if large is greater than small if len(self.large) > len(self.small): heapq.heappush(self.small, -heapq.heappop(self.large)) def findMedian(self) -> float: # odd numbers encountered: if len(self.small) > len(self.large): return -self.small[0] # even numbers encountered: return (self.large[0] - self.small[0]) / 2.0

   Code Snippet 5: Find Median from Data Stream (295)

   Suppose we know that the range is a bounded range, then we have the option to just keep track of the median position based on the bounded range.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32

   class MedianFinder: def __init__(self): self.freq = [0] * 101 # freq[i]: count of number i in [0,100] self.count = 0 # total number of inserted values def addNum(self, num: int) -> None: self.freq[num] += 1 self.count += 1 def findMedian(self) -> float: total = self.count # Find the median index(es) (0-based in sorted stream) if total % 2 == 1: median_pos1 = median_pos2 = total // 2 else: median_pos1 = total // 2 - 1 median_pos2 = total // 2 # helps us account for the even cases where we'll have 2 values result = [] running_count = 0 for value in range(101): # Loop over all possible values running_count += self.freq[value] # Add value to result if we cross the median positions if len(result) == 0 and running_count > median_pos1: result.append(value) if len(result) == 1 and running_count > median_pos2: result.append(value) break return sum(result) / len(result)

   Code Snippet 6: Find Median from Data Stream (295) – if it was a bounded range
   Pattern Extraction

   Core Insight: Split the stream into two halves: a max-heap for the lower half and a min-heap for the upper half. The median is the top of the max-heap (odd count) or the average of both tops (even count). Rebalance after each insertion. Pattern Name: Two-Heap Median Maintenance Canonical Section: Heap > Streaming Statistics Recognition Signals:

   • “Find median” from a data stream
   • Need \(O(\log n)\) insertion and \(O(1)\) median query
   • Data arrives one element at a time

   Anti-Signals:

   • Static array – just sort and index at \(n/2\)
   • Need other percentiles (not just median) – might need an order-statistic tree
   • Sliding window median (not cumulative) – needs augmented balanced BST or two-heap with lazy deletion. Connects to LC 480

   Decision Point: Two heaps for cumulative median. For sliding window median, the same idea plus lazy deletion or a balanced BST. Interviewer Comms: “I maintain two heaps: a max-heap for the smaller half and a min-heap for the larger half. On each insertion, I add to the appropriate heap and rebalance so sizes differ by at most 1. Median is the max-heap top (odd count) or average of both tops (even count). \(O(\log n)\) insert, \(O(1)\) query.”

   Pitfalls and Misconceptions

   1. Trap: Inserting into the wrong heap without rebalancing. Why: The invariant is: all elements in max-heap \(\leq\) all elements in min-heap, AND sizes differ by at most 1. Inserting without checking this invariant can place elements in the wrong half. Correction: Always insert into one heap (e.g., max-heap), then move the max to the min-heap if it exceeds the min-heap’s minimum. Then rebalance sizes.
   2. Trap: Forgetting to negate values for Python’s max-heap simulation. Why: Python’s heapq is min-heap only. Simulating max-heap requires negating values. Correction: Push -val to the max-heap; when reading the top, negate again: -max_heap[0].
   Problem Mutations

   1. What if you needed the sliding window median (window of size \(k\))? (stress-tests: cumulative vs windowed) Two heaps with lazy deletion (mark expired elements, clean on access). Connects to LC 480 Sliding Window Median.
   2. What if you needed an arbitrary percentile, not just the median? (stress-tests: 50th percentile) Generalise by adjusting the size ratio between the two heaps. For the $p$-th percentile, maintain \(p\%\) of elements in the lower half.
   3. What if elements could be removed as well as added? (stress-tests: insert-only) Lazy deletion: mark elements as removed; when they appear at the top of a heap, discard them. Track effective sizes separately from heap sizes.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Standardised annotation dropdowns for Find Median from Data Stream generated via batch canonical enrichment pass. Review against personal solving experience and adjust.
   

4. K Closest Points to Origin (973) It’s a euclidean distance shortcut where we don’t do the sqrt-ing. We judge using a valid proxy metric that is easier to calculate.

   Also, it’s so much simpler to just use the heapq.nsmallest() api directly for this question.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

   import heapq class Solution: def kClosest(self, points: list[list[int]], k: int) -> list[list[int]]: get_euclidean_dist = lambda p: p[0]**2 + p[1]**2 return heapq.nsmallest(k, points, key=get_euclidean_dist) # just for demo, here's a more handwritten one: import heapq class Solution: def kClosest(self, points: list[list[int]], k: int) -> list[list[int]]: # Maintain a max-heap of size k (heap stores -(distance, point)) heap = [] for p in points: dist = -(p[0] ** 2 + p[1] ** 2) if len(heap) < k: heapq.heappush(heap, (dist, p)) else: heapq.heappushpop(heap, (dist, p)) return [item[1] for item in heap]

   Code Snippet 7: K Closest Points to Origin (973)
   Pattern Extraction

   Core Insight: Use a max-heap of size \(k\) storing negative distances (Python min-heap trick). After processing all points, the heap contains the \(k\) closest. Pattern Name: Top-K with Max-Heap of Size K Canonical Section: Heap > Top-K Selection Recognition Signals:

   • “\(k\) closest/smallest” elements
   • Custom comparison (distance to origin)
   • Don’t need sorted order in output

   Anti-Signals:

   • Need sorted output – sort the heap results
   • \(k\) is close to \(n\) – sorting might be simpler

   Decision Point: \(k\) closest = max-heap of size \(k\) (or min-heap of all + \(k\) pops, or quickselect).

   Pitfalls and Misconceptions

   1. Trap: Computing actual Euclidean distance (with sqrt). Why: Sqrt is unnecessary for comparison. \(x^2 + y^2\) preserves ordering. Correction: Use squared distance: dist = x*x + y*y.
   2. Trap: Using a min-heap and popping \(k\) times instead of a max-heap of size \(k\). Why: Min-heap of all \(n\) is \(O(n + k \log n)\). Max-heap of size \(k\) is \(O(n \log k)\). For small \(k\), max-heap is better. Correction: Use max-heap of size \(k\) for better constant factors.
   Problem Mutations

   1. What if points were in 3D? (stress-tests: 2D) Same algorithm, add \(z^2\) to distance. No structural change.
   2. What if you needed the \(k\) farthest? (stress-tests: closest) Use a min-heap of size \(k\). Flip the comparison.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Batch annotation for K Closest Points to Origin.
   

Merging Sorted Lists or Arrays #

• Objective
  • Merge multiple sorted sequences or intervals efficiently using min-heaps.

Problems #

1. Merge K Sorted Lists (23)

   here, the heapq is used as an aux ds to help make it easier to choose which is the next provider (linked list) to take from. Here’s both the solutions.

   Same concept as 2 sorted lists, just maintain k providers. We have two solutions, one that is space-optimal \(O(N / logk)\) time \(O(1)\) space and one that is not \(O(N / logk)\) time \(O(k)\) space .

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

   class Solution: def mergeKLists(self, lists: List[Optional[ListNode]]) -> Optional[ListNode]: if not lists: return None n = len(lists) interval = 1 while interval < n: for i in range(0, n - interval, interval * 2): lists[i] = self.merge2Lists(lists[i], lists[i + interval]) interval *= 2 return lists[0] if n > 0 else None def merge2Lists(self, l1, l2): dummy = curr = ListNode() while l1 and l2: if l1.val < l2.val: curr.next, l1 = l1, l1.next else: curr.next, l2 = l2, l2.next # advance curr = curr.next curr.next = l1 or l2 return dummy.next

   Code Snippet 8: Merge K Sorted Lists (23)

   The easiest way imo to get the best value each time is to just use a priority queue. I prefer the PQ solution.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

   import heapq # Definition for singly-linked list. # class ListNode: # def __init__(self, val=0, next=None): # self.val = val # self.next = next class Solution: def mergeKLists(self, lists: List[Optional[ListNode]]) -> Optional[ListNode]: # filters away the empty lists: # NOTE: the use of id(ls) is to give a tie-breaker in the case that two lists have the same value. # this id usage is an easy way to avoid compiler issues providers = [(ls.val, id(ls), ls) for ls in lists if ls] heapq.heapify(providers) pre_head = dummy = ListNode() while providers: (val, _, ls) = heapq.heappop(providers) dummy.next = ls if ls.next: ls = ls.next new_val = ls.val heapq.heappush(providers, (new_val, id(ls), ls)) dummy = dummy.next return pre_head.next

2. Merge Intervals (56) with heap variant

   The optimal solution is typically a linear scan but we could rely on a heap version of this solution as well.

   It’s a linear scan alright (after sorting intervals using the start time), we have a working accumulator that we keep making reference to.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

   from typing import List class Solution: def merge(self, intervals: List[List[int]]) -> List[List[int]]: if not intervals: return [] # Sort intervals based on start time intervals.sort(key=lambda x: x[0]) merged_intervals = [intervals[0]] for start, end in intervals[1:]: last_merged_start, last_merged_end = merged_intervals[-1] if start <= last_merged_end: # Overlapping intervals merged_intervals[-1][1] = max(last_merged_end, end) # Merge else: merged_intervals.append([start, end]) # No overlap, add new return merged_intervals

   Code Snippet 9: Merge Intervals (56) with heap variant

   In the heap-based approach, intervals can be pushed into a min-heap based on their start time, and then repeatedly pop and merge the top intervals.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

   import heapq class Solution: def merge(self, intervals: list[list[int]]) -> list[list[int]]: if not intervals: return [] # Build a min-heap by the start time heap = [(start, end) for start, end in intervals] heapq.heapify(heap) result = [] # Pop the first interval prev_start, prev_end = heapq.heappop(heap) while heap: curr_start, curr_end = heapq.heappop(heap) if curr_start <= prev_end: # Merge intervals prev_end = max(prev_end, curr_end) else: # No overlap, push previous interval to result result.append([prev_start, prev_end]) prev_start, prev_end = curr_start, curr_end # Don't forget the last interval result.append([prev_start, prev_end]) return result

   This approach is not more optimal than the linear scan (after sorting), but it is instructive as an alternative and can be more useful if the input is streaming or not already sorted.

Frequency and Top-K Elements #

• Objective
  • Identify the top k most frequent elements in arrays or strings using priority queues for efficient counts management.

Problems #

1. Top K Frequent Elements (347)

   There’s a couple of ways to do this since we’re asked for frequency counts.

   The most optimal approach that runs in \(O(n)\) time is BucketSorting which looks like this:

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

   from collections import Counter class Solution: def topKFrequent(self, nums: List[int], k: int) -> List[int]: count = Counter(nums) # Bucket sort: index = frequency, value = list of numbers with that frequency max_possible_frequency = len(nums) buckets = [[] for _ in range(max_possible_frequency + 1)] for num, freq in count.items(): buckets[freq].append(num) # Flatten buckets from high to low frequency res = [] for freq in range(len(buckets) - 1, 0, -1): for num in buckets[freq]: res.append(num) if len(res) == k: return res

   Code Snippet 10: Top K Frequent Elements (347) – optimal bucketsort in \(O(n)\) time

   The other solutions that are suboptimal will run in \(O(n log n)\) time. Firstly we consider the heap based solution: A heap-based approach is \(O(n log k)\) where \(k\) is the char count

   1 2 3 4 5 6 7

   import heapq from collections import Counter class Solution: def topKFrequent(self, nums: List[int], k: int) -> List[int]: count = Counter(nums) return [item for item, _ in heapq.nlargest(k, count.items(), key=lambda x: x[1])]

   Code Snippet 11: Top K Frequent Elements (347) – suboptimal heap-based (easy) in \(O(n log n)\) time and space

   A counter based solution runs in \(O(nlogn)\) :

   1 2 3 4 5

   from collections import Counter class Solution: def topKFrequent(self, nums: List[int], k: int) -> List[int]: return [elem for [ elem, _ ] in Counter(nums).most_common(k)]

   Code Snippet 12: Top K Frequent Elements (347) – counter-based solution in \(O(nlogn)\)
   Pattern Extraction

   Core Insight: Count frequencies with Counter, then select top \(k\) using a min-heap of size \(k\) or bucket sort (frequency as index). Pattern Name: Top-K with Frequency Counting Canonical Section: Heap > Top-K Selection Recognition Signals:

   • “Top \(k\) most frequent”
   • Frequency counting + selection

   Anti-Signals:

   • Need ALL elements sorted by frequency – full sort

   Decision Point: Heap: \(O(n \log k)\). Bucket sort: \(O(n)\). Bucket sort uses frequency as the array index.

   Pitfalls and Misconceptions

   1. Trap: Using Counter.most_common(k) without understanding the underlying algorithm. Why: In an interview, you should implement the heap approach yourself. Correction: Count frequencies, then use heap or bucket sort.
   Problem Mutations

   1. What if frequencies updated dynamically? (stress-tests: static) Need LFU-style structure: hash map + frequency buckets. Connects to LC 460.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Batch annotation for Top K Frequent Elements.
   

Scheduling and Task Management #

• Objective
  • Use priority queues for simulating task scheduling, managing intervals, and handling dynamic priorities.
• Classic Examples (often conceptual or system design related)

Problems #

1. Last Stone Weight (1046)

   This is a parable to actual scheduling algos, the idea is just to maintain a running schedule and just retrieve from there.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14

   import heapq class Solution: def lastStoneWeight(self, stones: list[int]) -> int: stones = [-s for s in stones] # API is a minheap, hence the negation heapq.heapify(stones) while len(stones) > 1: first = -heapq.heappop(stones) # Largest second = -heapq.heappop(stones) # Next largest if first != second: heapq.heappush(stones, -(first - second)) return -stones[0] if stones else 0

   Code Snippet 13: Last Stone Weight (1046)
   Pattern Extraction

   Core Insight: Max-heap. Repeatedly pop the two heaviest, push back their difference (if non-zero). Continue until one or zero stones remain. Pattern Name: Greedy Simulation with Max-Heap Canonical Section: Heap > Simulation Recognition Signals:

   • “Smash two heaviest stones” repeatedly
   • Need efficient max-extraction

   Anti-Signals:

   • Need to smash specific stones (not the heaviest) – different selection strategy

   Decision Point: Repeated max-extraction + insertion = heap. Greedy (always process the two largest) is provably optimal here.

   Pitfalls and Misconceptions

   1. Trap: Using a list and sorting each iteration (\(O(n^2 \log n)\) total). Why: Heap operations are \(O(\log n)\) per extraction/insertion. Re-sorting is \(O(n \log n)\) per iteration. Correction: Use heapq with negated values (Python min-heap as max-heap).
   Problem Mutations

   1. What if you needed the minimum possible last stone weight? (stress-tests: greedy selection) NP-hard in general (partition problem variant). The greedy approach doesn’t give the minimum for all inputs. Connects to LC 1049 Last Stone Weight II.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Batch annotation for Last Stone Weight.
   

2. Task Scheduling variations

3. CPU scheduling simulation problems

4. Task Scheduler (621)

   This can do a time-simulation approach.

   We use two things: a max heap for schedule-able tasks and a cooldown QUEUE for the tasks that are currently in cooldown. Cooldowns happen in FIFO hence the use of the deque. I prefer this approach to the problem over the purely mathematical approach.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

   import heapq from collections import Counter, deque # TIME SIMULATION APPROACH class Solution: def leastInterval(self, tasks: List[str], n: int) -> int: freqs = Counter(tasks) pq = [(-cnt, task) for task, cnt in freqs.items()] heapq.heapify(pq) cooldown = deque() # (ready_time, -cnt, task) time = 0 while pq or cooldown: time += 1 if pq: cnt, task = heapq.heappop(pq) if cnt + 1: # more left cooldown.append((time + n, cnt + 1, task)) if cooldown and cooldown[0][0] == time: _, next_cnt, next_task = cooldown.popleft() heapq.heappush(pq, (next_cnt, next_task)) return time

   Code Snippet 14: Task Scheduler (621) – maxheap for tasks, cooldown queue

   This particular question has a math approach too. Realise that the tasks with max frequency are the bottleneck in the schedule. We need to ensure at least n intervals between repeats of the most common task. We fill all cooldown slots with other tasks if we can else we must idle. So the formula to compute the time needed if we strictly alternate between maxtasks and idleslots, possibly filling them with other tasks, otherwise falling back to the total number of tasks.

   1 2 3 4 5 6 7 8 9 10 11

   from collections import Counter class Solution: def leastInterval(self, tasks: List[str], n: int) -> int: freqs = Counter(tasks) f_max = max(freqs.values()) n_max = sum(1 for freq in freqs.values() if freq == f_max) part_len = f_max - 1 empty_slots = part_len * (n + 1) res = part_len * (n + 1) + n_max return max(res, len(tasks))

   Code Snippet 15: Task Scheduler (621) – math-based approach
   Pattern Extraction

   Core Insight: The most frequent task determines idle time. Formula: (max_freq - 1) * (n + 1) + count_of_max_freq_tasks. Answer = max(formula, total_tasks). Pattern Name: Greedy Scheduling with Idle Slot Formula Canonical Section: Heap > Greedy Scheduling Recognition Signals:

   • “Schedule tasks with cooldown \(n\) between same tasks”
   • Minimise total time (including idle slots)

   Anti-Signals:

   • No cooldown – answer is just total tasks
   • Tasks have different durations – more complex

   Decision Point: Formula approach is \(O(n)\) and cleaner. Heap simulation also works.

   Pitfalls and Misconceptions

   1. Trap: Simulating step-by-step without the formula. Why: More complex. Formula gives direct answer. Correction: idle_time = (max_freq - 1) * (n + 1) + count_max. Answer = max(idle_time, len(tasks)).
   2. Trap: Forgetting max(formula, total_tasks). Why: When cooldown is short, no idle slots needed. Answer is just total tasks. Correction: return max(formula_result, len(tasks)).
   Problem Mutations

   1. What if you needed the actual schedule? (stress-tests: time vs schedule) Simulation with heap + cooldown queue. Pop most frequent, schedule, add to cooldown.
   2. What if tasks had priorities? (stress-tests: no priorities) Schedule highest-priority first within cooldown constraints.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Batch annotation for Task Scheduler.
   

Online Median / Running Statistics #

• Objective
  • Maintain running medians or generalized order-statistics efficiently on dynamically inserting and removing elements.

Problems #

1. TODO Sliding Window Median (480)
2. Data Stream Median or Quantile problems

Heap Sort and Selection Problems #

• Objective
  • Use heaps for efficient sorting or selecting the nth smallest/largest element.

Problems #

1. Heap Sort algorithms

2. Kth Smallest Element In a Sorted Matrix

   The heap implementation of this involves us keeping a heap and popping from it for (k - 1) times. We keep adding to the heap if possible. The init version of the heap is just all the first elements from each row.

   We do this popping because we keep a minheap that takes in values from cells within the matrix.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

   import heapq class Solution: def kthSmallest(self, matrix: List[List[int]], k: int) -> int: n = len(matrix) # we init the heap with the first element from each row: heap = [ (matrix[row][0], 0, row) for row in range(n) ] # (elem, column idx, row idx) heapq.heapify(heap) # now we extract from the min heap for (k - 1) times so that we can pop out the kth smallest element: for _ in range(k - 1): elem, col, row = heapq.heappop(heap) # add next element to heap, if exists: if col < (n - 1): new_elem = matrix[row][col + 1] heapq.heappush(heap, (new_elem, col + 1, row)) return heap[0][0]

   Code Snippet 16: Kth Smallest Element In a Sorted Matrix

Interval and Range Merging with Heaps #

• Objective
  • Maintain earliest ending intervals or ranges for merging or scheduling problems using min-heaps.

Problems #

1. Meeting Rooms II (253) (NC link)

   We have 2 choices: we can use a min heap and play with the intervals OR we can do the sweep line algo Here’s my version with PQs:

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28

   import heapq class Solution: def minMeetingRooms(self, intervals: List[Interval]) -> int: if not intervals: return 0 # Sort intervals by start time intervals.sort(key=lambda x: x.start) # Min-heap to track end times of meetings currently occupying rooms min_heap = [] # Add the first meeting's end time heapq.heappush(min_heap, intervals[0].end) for interval in intervals[1:]: start, end = interval.start, interval.end # If the current meeting starts after or when a room is freed if start >= min_heap[0]: heapq.heappop(min_heap) # Reuse room # Assign new room (or reused room with updated end time) heapq.heappush(min_heap, end) # Number of rooms = size of min-heap return len(min_heap)

   Code Snippet 17: Meeting Rooms II (253) – my version with PQs

   And here’s the usage of the sweep line algo

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30

   # SWEEP-LINE ALGO def minMeetingRooms(intervals): if not intervals: return 0 starts = sorted(i[0] for i in intervals) ends = sorted(i[1] for i in intervals) s, e = 0, 0 rooms = 0 max_rooms = 0 while s < len(intervals): if starts[s] < ends[e]: rooms += 1 max_rooms = max(max_rooms, rooms) s += 1 else: rooms -= 1 e += 1 return max_rooms # - Explanation: # - Separate all start times and end times into two sorted lists. # - Use pointers to iterate through both arrays, increment a counter when a meeting starts, decrement when one ends. # - Track the max counter during traversal → max concurrency → min rooms required. # - This approach avoids heaps and uses two sorted arrays, can be slightly faster for certain use-cases.

   Code Snippet 18: Meeting Rooms II (253) – sweep-line algo version
   Pattern Extraction

   Core Insight: Sort by start time. Min-heap tracks end times of active meetings. For each meeting: if it starts after the earliest ending, reuse that room (pop heap). Otherwise, new room (push). Heap size = answer. Pattern Name: Sweep Line with Min-Heap Canonical Section: Heap > Interval Scheduling Recognition Signals:

   • “Minimum rooms for overlapping meetings”
   • Sort by start, heap of end times

   Anti-Signals:

   • Just check if any overlap – Meeting Rooms I (single comparison)
   • Need to assign specific rooms – more complex

   Decision Point: Min-heap of end times tracks “soonest-freeing room.” Alternatively: event-based counting (+1 at start, -1 at end, max running count).

   Pitfalls and Misconceptions

   1. Trap: Sorting by end time instead of start time. Why: Process meetings chronologically (by arrival). Start-time sort ensures this. Correction: Sort by interval[0].
   2. Trap: Comparing new meeting’s start with the latest ending meeting instead of earliest. Why: Want to reuse the room that frees soonest. Heap root = earliest end. Correction: if intervals[i].start > heap[0]: heappop(heap)=.
   Problem Mutations

   1. What if meetings had different room requirements (sizes)? (stress-tests: uniform rooms) Assignment problem: match meeting sizes to room capacities. More complex.
   2. What if meetings could be rescheduled? (stress-tests: fixed schedule) This IS the minimum – the algorithm already finds optimal room count.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Batch annotation for Meeting Rooms II.
   

2. Car Fleet (853)

   Merging from one-side problem. Can view the problem as creating contiguous segments of the cars array based on arrival times after merging. Each segment maps to exactly one car fleet. Top of the stack will be the next arriving fleet, nearest to the destination, we have to just check if the car in consideration is going to be part of that fleet or will start a new fleet.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16

   class Solution: def carFleet(self, target: int, position: List[int], speed: List[int]) -> int: cars = sorted(list(zip(position, speed)), reverse=True) arrival_time_stack = [] for pos, s in cars: arrival_time = (target - pos) / s if not arrival_time_stack: # empty stack arrival_time_stack.append(arrival_time) continue # if slower than the prev arrival, then it's a new fleet, else ignore this: if arrival_time_stack and arrival_time > arrival_time_stack[-1]: arrival_time_stack.append(arrival_time) return len(arrival_time_stack)

   Code Snippet 19: Car Fleet (853)

   for completeness, here’s a heap-based approach that is similar

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25

   import heapq from typing import List class Solution: def carFleet(self, target: int, position: List[int], speed: List[int]) -> int: # Build max-heap based on position (negate to simulate max-heap with Python's min-heap) heap = [(-pos, speed[i]) for i, pos in enumerate(position)] heapq.heapify(heap) fleets = 0 last_arrival_time = 0.0 # Process cars from closest to target (largest position) to farthest while heap: pos, spd = heapq.heappop(heap) pos = -pos # revert negation to original position arrival_time = (target - pos) / spd # If current car's arrival time is greater than last fleet's time, it forms a new fleet if arrival_time > last_arrival_time: fleets += 1 last_arrival_time = arrival_time # else, it merges into the fleet represented by last_arrival_time return fleets

   Code Snippet 20: Car Fleet (853)

Graph Algorithms using Heaps #

• Objective

  • Use heaps for efficient priority-based processing in graph algorithms like Dijkstra’s or Prim’s.

    Refer directly to the Advanced Graph Sections below.

Problems #

1. Network Delay Time (743)

   This is just a classic dijkstra, single source we want to find the longest branch form this single source to every other node. Applies for network related problem domains as well. Note the use of the 1-idx shimming and splicing here.

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26

   from collections import defaultdict import heapq class Solution: def networkDelayTime(self, times: List[List[int]], n: int, k: int) -> int: graph = defaultdict(list) for u, v, w in times: graph[u].append((v, w)) # we're going to shim the 0th idx to handle the 1-indexing, then later ignore it dist = [float('inf')] * (n + 1) dist[k] = 0 heap = [(0, k)] # current dist, node (min-pq by distance) while heap: curr_dist, node = heapq.heappop(heap) if curr_dist > dist[node]: continue for nei, w in graph[node]: # if can relax: if curr_dist + w < dist[nei]: dist[nei] = curr_dist + w heapq.heappush(heap, (dist[nei], nei)) max_dist = max(dist[1:]) return max_dist if max_dist != float('inf') else -1

   Code Snippet 21: Network Delay Time (743)

2. Minimum Spanning Tree algorithms

Custom Comparator / Complex Priority Logic #

• Objective
  • Implement priority queues with custom ordering or multi-key priorities.

Problems #

1. Any problem requiring lex order or multi-field ordering for tasks or objects

2. Design Twitter (355)

   The idea here is to maintain multiple PQs, one for each user then we can merge them in using heapq.merge()

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24

   from collections import defaultdict import heapq class Twitter: def __init__(self): self.tweet_count = 0 self.uid_to_tweets = defaultdict(list) self.uid_to_following = defaultdict(set) def postTweet(self, userId: int, tweetId: int) -> None: self.tweet_count += 1 self.uid_to_tweets[userId].append((self.tweet_count, tweetId, userId)) def getNewsFeed(self, userId: int) -> list[int]: posts = heapq.merge(*(self.uid_to_tweets[uid] for uid in self.uid_to_following[userId] | {userId})) return [post[1] for post in heapq.nlargest(10, posts)] def follow(self, followerId: int, followeeId: int) -> None: self.uid_to_following[followerId].add(followeeId) def unfollow(self, followerId: int, followeeId: int) -> None: self.uid_to_following[followerId].discard(followeeId)

   Code Snippet 22: Design Twitter (355) – (suboptimal) multiple PQs, merged via heapq.merge()

   However, the optimal approach is really to implement k-way merging:

   1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39

   from collections import defaultdict import heapq class Twitter: def __init__(self): self.time = 0 self.user_tweets = defaultdict(list) self.following = defaultdict(set) def postTweet(self, userId: int, tweetId: int) -> None: self.time += 1 self.user_tweets[userId].append((self.time, tweetId)) def getNewsFeed(self, userId: int) -> list[int]: result = [] # We'll build a heap of (timestamp, tweetId, whose_list, index) heap = [] users = self.following[userId].copy() users.add(userId) for uid in users: tweets = self.user_tweets[uid] if tweets: # push last tweet (most recent), track index ts, tid = tweets[-1] heapq.heappush(heap, (-ts, tid, uid, len(tweets) - 1)) while heap and len(result) < 10: ts, tid, uid, idx = heapq.heappop(heap) result.append(tid) if idx > 0: prev_ts, prev_tid = self.user_tweets[uid][idx - 1] heapq.heappush(heap, (-prev_ts, prev_tid, uid, idx - 1)) return result def follow(self, followerId: int, followeeId: int): if followerId != followeeId: self.following[followerId].add(followeeId) def unfollow(self, followerId: int, followeeId: int): self.following[followerId].discard(followeeId)

   Code Snippet 23: Design Twitter (355) – (optimal)
   Pattern Extraction

   Core Insight: Hash maps for follow relationships. Each user has a list of tweets with timestamps. getNewsFeed merges the \(k\) most recent tweets from the user and all followees using a max-heap (merge-K-sorted-lists pattern). Pattern Name: Merge-K-Sorted-Lists with Social Graph Canonical Section: Heap > Design Problems Recognition Signals:

   • “Design a simplified Twitter”
   • Follow/unfollow + post tweet + get news feed
   • News feed = top-K merge from multiple sorted streams

   Anti-Signals:

   • Need real-time feed updates – push-based vs pull-based design

   Decision Point: The getNewsFeed operation is the interesting part. It’s merge-K-sorted-lists on each followee’s tweet list.

   Pitfalls and Misconceptions

   1. Trap: Sorting all tweets from all followees every time getNewsFeed is called. Why: \(O(n \log n)\) per call. Heap-based merge of pre-sorted lists is more efficient. Correction: Push most recent tweet from each followee into a heap. Pop 10 times.
   2. Trap: Not handling self-follows (user sees their own tweets). Why: A user should always see their own tweets in the feed. Correction: Include the user’s own tweet list in the merge, or always include self in followees.
   Problem Mutations

   1. What if the feed needed to be real-time (push-based)? (stress-tests: pull-based) Fan-out on write: when a user tweets, push to all followers’ feed caches. System design territory.
   AI usage disclosure Claude Opus 4.6 · content enrichment

   Batch annotation for Design Twitter.
   

Approximate / Streaming Data Structures with Heaps #

• Objective
  • Use approximate counting and priority heuristics for large or streaming datasets.

Problems #

1. Count-Min Sketch with priority queues (more advanced/outside standard Leetcode scope)

SOE #

• Reversed or wrong comparator in heap (min vs max). The default is min heap ordedring, so if we want to keep max heaps, then we have to negate the values.

• Forgetting lazy deletion or invalidation in PQ operations

• Improper heapify or failure to maintain heap invariant

  Remember that we can technically peek by just checking the first element in the heap, that’s guaranteed to the min element in the heap.

• Sometimes a heap is overkill if we can get away with doing a quickselect, since that would avoid completely sorting the input

• forgetting the heap invariants (should only use the heap API for our needs, we shouldn’t accidentally think that it’s a sorted array)

Decision Flow #

• Need streaming top-k? -> Sliding window or heap
• Merging sorted data? -> Min-heap merging
• Frequency based selections? -> Heap top-k strategy
• Need min AND max simultaneously? -> Dual heap (max-heap for lower half, min-heap for upper half)
• Need kth element in stream? -> Size-bounded heap of size k
• Sometimes a heap is overkill -> if we can get away with quickselect, that avoids full sorting

Styles, Tricks and Boilerplate #

Boilerplate #

Quickselect Algo #

The main part is the partition function, then the rest is the intuitive understanding of the invariants / properties that lead us to the reason why partitioning it k times is good enough.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

import random
from typing import List

class Solution:
    def findKthLargest(self, nums: List[int], k: int) -> int:
        def quickselect(left, right, target):
            if left == right:
                return nums[left]
            pivot_index = random.randint(left, right)
            # Partition around pivot and return final pivot position
            new_pivot = partition(left, right, pivot_index)
            if new_pivot == target:
                return nums[new_pivot]
            elif new_pivot < target:
                return quickselect(new_pivot + 1, right, target)
            else:
                return quickselect(left, new_pivot - 1, target)


        def partition(left, right, pivot_index):
            """
            Returns the pivot idx after paritioning left and right segments based on the element values.

            The property it maintains thereafter is the all to the right of pivot are STRICTLY MORE THAN pivot.

            All the left of hte pivot are STRICTLY LESS THAN pivot.
            """
            pivot_value = nums[pivot_index]
            nums[pivot_index], nums[right] = nums[right], nums[pivot_index]
            store_index = left
            for i in range(left, right):
                if nums[i] < pivot_value:
                    nums[store_index], nums[i] = nums[i], nums[store_index]
                    store_index += 1
            # move pivot to its final place
            nums[store_index], nums[right] = nums[right], nums[store_index]

            return store_index

        n = len(nums)
        target = n - k  # index of k-th largest in sorted order

        return quickselect(0, n - 1, target)

Styles #

1. let the heap do its job when you’re maintaining order statistics, just prune AFTER the heap does its job (e.g. after it gets added).

Tricks #

1. Double heap approach is optimal for general data-streams. We can keep balancing them and we can rebalance when we add / remove to the overall system.

2. MATH TRICK: when dealing with Euclidean distances, we can just use the squared distances and avoid having to do the sqrt.

3. For stream median finding, if it’s a bounded range, representing the domain via a fixed size array that keeps frequency counts, then just adjusting a median pointer is a great idea.

Python Recipes: #

1. Heap order statistics functions:

   • We can pass in a lambda for the nsmallest comparator.

     E.g. return heapq.nsmallest(k, points, key=lambda p: p[0]**2 + p[1]**2)

   • copying a heap preserves its heap properties because it’s “just a list”

   • heapq.heappushpop() and heapq.heapreplace is very useful:

     1 2 3 4 5 6 7 8 9 10 11 12 13

     import heapq heap = [5, 7, 9] heapq.heapify(heap) heapq.heappush(heap, 3) # Add element smallest = heapq.heappop(heap) # Remove smallest element result = heapq.heappushpop(heap, 4) # Push 4 then pop smallest result2 = heapq.heapreplace(heap, 6) # Pop smallest then push 6 top3 = heapq.nlargest(3, [1,5,2,9,7]) merged = heapq.merge([1,3,5], [2,4,6])

TODO KIV #

[ ] Existing Canonicals #

• Online Median / Running Statistics:: TODO Sliding Window Median (480)

[ ] Fresh #

• see problems from leetcode’s list of questions here