TL;DR
Size-k min-heap: add is O(log k) time, O(k) space.
Approach 1 — Brute force (re-sort on every add)
Store every value; on each add, sort descending and return index k - 1.
from typing import List
class KthLargest:
def __init__(self, k: int, nums: List[int]):
self.k = k
self.nums = list(nums)
def add(self, val: int) -> int:
self.nums.append(val)
self.nums.sort(reverse=True)
return self.nums[self.k - 1]
Each add costs O(n log n) with n the stream length so far. With up to 10^4 adds on a stream that grows to 2·10^4 elements, that is hundreds of millions of comparison steps, which the constraints are set up to rule out.
Approach 2 — Min-heap of size k
The insight: the k-th largest element is the minimum of the k largest. Keep only the k largest values seen so far in a min-heap. Any value that drops out of the top k can never re-enter it, so it is safe to discard. The heap root is the answer at all times.
A binary min-heap is a complete binary tree (stored as an array, Python’s heapq) where every parent is ≤ its children, giving O(1) peek-min and O(log size) push/pop. The root holds the smallest of the k retained values, which is exactly the k-th largest of the stream:
graph TD
A["4 (root = 3rd largest)"] --> B["5"]
A --> C["8"]
import heapq
from typing import List
class KthLargest:
def __init__(self, k: int, nums: List[int]):
self.k = k
self.heap = list(nums)
heapq.heapify(self.heap)
while len(self.heap) > 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]
Walkthrough of KthLargest(3, [4, 5, 8, 2]):
- Init: heapify
[4, 5, 8, 2], then pop once (size 4 > 3) removing 2 → heap holds {4, 5, 8}, root 4.
add(3): push 3 → {3, 4, 5, 8} → pop min 3 → root 4. Return 4.add(5): push 5 → {4, 5, 5, 8} → pop 4 → {5, 5, 8}. Return 5.add(10): push 10 → pop 5 → {5, 8, 10}. Return 5.add(9): push 9 → pop 5 → {8, 9, 10}. Return 8.add(4): 4 is pushed then immediately popped (it is the min) → {8, 9, 10}. Return 8.
Matches the expected outputs 4, 5, 5, 8, 8.
Complexity: __init__ O(n log n) (or O(n + (n−k) log n) with heapify-then-pop); each add O(log k). Space O(k).
Approach 3 — Slightly slicker add (guarded push)
The insight: once the heap already has k elements, a new value only matters if it beats the current root; heapq.heappushpop does push-then-pop in one sift, and skipping values <= root avoids even that.
import heapq
from typing import List
class KthLargest:
def __init__(self, k: int, nums: List[int]):
self.k = k
self.heap = heapq.nlargest(k, nums) # at most k items
heapq.heapify(self.heap)
def add(self, val: int) -> int:
if len(self.heap) < self.k:
heapq.heappush(self.heap, val)
elif val > self.heap[0]:
heapq.heappushpop(self.heap, val)
return self.heap[0]
Same asymptotics — O(log k) per add, O(k) space — but each add does at most one sift instead of two, and often zero. On the walkthrough above the behavior is identical (e.g. add(4) is rejected up front because 4 <= 8).
Common pitfalls
- Using a max-heap of everything: peek gives the 1st largest, not the k-th, and finding the k-th costs k pops per query.
- Forgetting that
nums may start with fewer than k elements — don’t index heap[0] before trimming logic runs; the guarantee is only that add is never asked before k elements exist.
- Treating “k-th largest” as k-th distinct value; duplicates each occupy a slot (
[7, 7, 7] with k = 2 → 7).
- Rebuilding or re-sorting per call — the whole point of the streaming setup is amortized O(log k) updates.
Pattern takeaway
When a stream question asks for “the k-th largest/smallest so far,” keep the opposite-ordered heap capped at size k: a min-heap for k-largest (root = answer), a max-heap for k-smallest. Elements evicted from the top-k can never return, so the cap is safe, and every update is O(log k) regardless of stream length.