# Heap and Priority Queue Interview Patterns: K-th Largest, Merge Streams, and Scheduling (2025)

Source: https://www.techinterview.org/post/3233466504/heap-patterns/
Updated: 2026-04-17 · techinterview.org

## Heap Fundamentals

A heap is a complete binary tree where every parent satisfies the heap property (min-heap: parent = children). The root is always the min (or max). Core operations: insert: O(log n). extract-min/max: O(log n). peek-min/max: O(1). Build heap from array: O(n) (not O(n log n) -- counterintuitively). Python: heapq is a min-heap. For max-heap: negate all values. Key insight: heaps give you the minimum/maximum efficiently but are not sorted -- use only when you need the extremes, not a full sort.

## K-th Largest Element (LC 215) — Min-Heap of Size K


```
import heapq

def find_kth_largest(nums: list[int], k: int) -> int:
    # Maintain a min-heap of size k
    # The root is always the k-th largest
    heap = []
    for n in nums:
        heapq.heappush(heap, n)
        if len(heap) > k:
            heapq.heappop(heap)  # remove the smallest
    return heap[0]  # k-th largest = smallest in the heap of k largest
# O(n log k) time. Better than sorting (O(n log n)) when k << n.
```


## K Closest Points to Origin (LC 973)


```
def k_closest(points: list[list[int]], k: int) -> list[list[int]]:
    # Max-heap of size k (negate distances for max-heap with heapq)
    heap = []
    for x, y in points:
        dist = x*x + y*y  # no need for sqrt -- monotone
        heapq.heappush(heap, (-dist, x, y))
        if len(heap) > k:
            heapq.heappop(heap)
    return [[x, y] for _, x, y in heap]
```


## Merge K Sorted Lists (LC 23)


```
from heapq import heappush, heappop

def merge_k_lists(lists):
    heap = []
    for i, node in enumerate(lists):
        if node:
            heappush(heap, (node.val, i, node))

    dummy = ListNode(0)
    curr = dummy
    while heap:
        val, i, node = heappop(heap)
        curr.next = node
        curr = curr.next
        if node.next:
            heappush(heap, (node.next.val, i, node.next))
    return dummy.next
# O(n log k) where n = total nodes, k = number of lists
```


## Task Scheduler with Heap (LC 621 Alternative)


```
from collections import Counter
import heapq
from collections import deque

def least_interval(tasks: list[str], n: int) -> int:
    freq = Counter(tasks)
    # Max-heap (negate for Python min-heap)
    heap = [-f for f in freq.values()]
    heapq.heapify(heap)

    time = 0
    cooldown = deque()  # (available_at, neg_freq)

    while heap or cooldown:
        time += 1
        if heap:
            freq_remaining = heapq.heappop(heap) + 1  # decrement count
            if freq_remaining < 0:  # still has remaining tasks
                cooldown.append((time + n, freq_remaining))
        if cooldown and cooldown[0][0] == time:
            _, f = cooldown.popleft()
            heapq.heappush(heap, f)
    return time
```


## Find Median from Data Stream (LC 295)


```
class MedianFinder:
    def __init__(self):
        self.small = []  # max-heap (negated): lower half
        self.large = []  # min-heap: upper half

    def addNum(self, num: int) -> None:
        heapq.heappush(self.small, -num)
        # Balance: largest in small  self.large[0]):
            heapq.heappush(self.large, -heapq.heappop(self.small))
        # Size balance: small can have at most 1 more than large
        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 -self.small[0]
        return (-self.small[0] + self.large[0]) / 2.0
```


Meta coding interviews frequently ask heap-based problems like k-th largest and merge k sorted lists. Review patterns for [Meta interview: heap and priority queue patterns](https://www.techinterview.org/post/3233460272/meta-interview-guide-2026-facebook-instagram-whatsapp-engineering/).

Databricks interviews test heap-based streaming algorithms like running median and k-th largest. See problem patterns in [Databricks interview: heap patterns for streaming data](https://www.techinterview.org/post/3233460280/databricks-interview-guide-2026-spark-internals-delta-lake-and-lakehouse-architecture/).

Atlassian coding rounds test heap-based scheduling problems. Review commonly asked patterns for [Atlassian interview: priority queue and scheduling problems](https://www.techinterview.org/post/3233459919/atlassian-interview/).