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.