Heap & Priority QueueHeaps & Priority Queues
Mục lục
- Tổng quan
- Kiến thức nền tảng
- Tính chất heap
- Mảng ngầm định — một cái cây không có pointer
- Sift-up và sift-down
- Vì sao heapify là O(n) chứ không phải O(n log n)
- Bảng tổng hợp độ phức tạp
- Khái niệm chính
- heapq của Python
- Max-heap trong Python: mẹo đổi dấu
- ADT priority queue
- Heapsort
- Pattern two-heaps: running median
- Tìm phần tử lớn thứ k
- Gộp k-way
- Best Practices
- Tài liệu tham khảo
Table of contents
- Overview
- Fundamentals
- The heap property
- The implicit array — a tree with no pointers
- Sift-up and sift-down
- Why heapify is O(n), not O(n log n)
- Complexity summary
- Key Concepts
- Python’s heapq
- Max-heaps in Python: the negation trick
- The priority queue ADT
- Heapsort
- The two-heaps pattern: running median
- Finding the k-th largest element
- k-way merge
- Best Practices
- References
Thuộc bộ kiến thức Data Structures & Algorithms Roadmap.
Tổng quan
Mảng đã sắp xếp trả lời “phần tử nhỏ nhất là gì?” trong O(1) nhưng tốn O(n) để insert. Một list thường insert trong O(1) nhưng tốn O(n) để tìm minimum. Balanced BST làm cả hai trong O(log n), nhưng bạn đang trả tiền cho một thứ tự toàn phần mà bạn không hề cần tới.
Heap là cấu trúc dành cho trường hợp bạn chỉ cần một đầu của thứ tự: phần tử nhỏ nhất, hoặc lớn nhất, lặp đi lặp lại. Nó cho bạn O(1) để peek phần tử cực trị, O(log n) để insert, O(log n) để extract, và — phần khiến người ta bất ngờ — O(n) để dựng một heap từ mảng chưa sắp xếp. Nó làm tất cả những điều đó hoàn toàn không cần pointer, lưu trong một mảng phẳng, khiến nó là một trong những cấu trúc thân thiện với cache nhất trong toàn bộ knowledge base này.
Thao tác trừu tượng mà heap cài đặt là priority queue: một ADT trong đó phần tử được lấy ra theo thứ tự ưu tiên chứ không phải thứ tự insert. ADT đó có mặt khắp nơi trong hệ thống thật:
- Thuật toán Dijkstra và A* liên tục lấy ra node chưa thăm gần nhất — xem ./14-shortest-path-algorithms.md.
- Thuật toán Prim liên tục lấy ra edge rẻ nhất cắt ngang cut — xem ./15-minimum-spanning-trees.md.
- Scheduler của OS chọn task có độ ưu tiên cao nhất đang sẵn sàng chạy.
- Mô phỏng sự kiện xử lý sự kiện kế tiếp theo timestamp.
- Huffman coding liên tục gộp hai ký hiệu có tần suất thấp nhất.
- Rate limiter, timer, job queue kích hoạt phần tử tới hạn sớm nhất.
heapq.merge, external sorting, compaction của LSM-tree gộpkdãy đã sắp xếp bằng cách liên tục lấy đầu nhỏ nhất.- Streaming top-k và running median giữ một cửa sổ có chặn các phần tử đáng quan tâm.
Một nguồn nhầm lẫn phổ biến cần nói trước: “heap” ở đây không liên quan gì tới “the heap” trong quản lý bộ nhớ. Vùng cấp phát động và cấu trúc dữ liệu này trùng tên vì lý do lịch sử, ngoài ra không liên quan.
Kiến thức nền tảng
Tính chất heap
Binary heap là một complete binary tree (xem ./10-tree-data-structures.md) thoả một bất biến cục bộ duy nhất:
- Min-heap: key của mọi node đều
≤key của cả hai child. Do đó minimum nằm ở root. - Max-heap: key của mọi node đều
≥key của cả hai child. Maximum nằm ở root.
Đó là toàn bộ bất biến. Chú ý nó yếu tới mức nào so với BST:
Một min-heap hợp lệ: KHÔNG phải BST, và điều đó không sao:
(1) - 3 nằm trong subtree trái của 1 và 8 nằm
/ \ trong subtree phải, nhưng 3 > ... không gì cả.
(3) (2) - Không có thứ tự nào giữa hai sibling.
/ \ / \ - In-order traversal cho 7 3 5 1 8 2 9,
(7) (5) (8) (9) không hề được sắp xếp.
Heap chỉ có thứ tự một phần: nó biết quan hệ dọc theo đường root-tới-leaf và không biết gì về quan hệ trái-phải. Đó chính xác là lý do nó rẻ hơn BST — nó làm ít việc hơn vì nó hứa ít hơn. Bạn không thể tìm một key bất kỳ trong heap nhanh hơn O(n), và bạn không thể duyệt nó theo thứ tự sắp xếp mà không phá huỷ nó.
Mảng ngầm định — một cái cây không có pointer
Vì heap là complete binary tree, các node của nó có thể được đánh số theo từng level mà không có lỗ hổng, và lưu trong một mảng phẳng. Quan hệ parent/child trở thành phép tính số học:
index 0
(1)
/ \
index 1 index 2
(3) (2)
/ \ / \
(7) (5) (8) (9)
3 4 5 6
mảng: [ 1, 3, 2, 7, 5, 8, 9 ]
index: 0 1 2 3 4 5 6
Với chỉ số 0-based:
| Quan hệ | Công thức |
|---|---|
Left child của i | 2i + 1 |
Right child của i | 2i + 2 |
Parent của i | (i − 1) // 2 |
i có phải leaf? | 2i + 1 >= n |
| Chỉ số leaf đầu tiên | n // 2 |
| Chỉ số non-leaf cuối cùng | n // 2 − 1 |
(Một số giáo trình dùng chỉ số 1-based, khi đó công thức gọn hơn một chút: 2i, 2i+1, và i // 2. Luôn kiểm tra nguồn dùng quy ước nào trước khi chép lại.)
Cách bố trí này là lợi thế lớn thứ hai của heap sau độ phức tạp:
- Không tốn pointer. Một node BST tốn một giá trị cộng hai pointer (16 byte overhead trên máy 64-bit); một phần tử heap chỉ tốn đúng giá trị đó.
- Cache locality tuyệt vời. Vài level trên cùng của heap — những level bị chạm tới ở mọi thao tác — nằm ở vài chục ô mảng đầu tiên, luôn ở lại trong L1 cache. Cây dựa trên pointer thì rải các node đó khắp không gian địa chỉ.
- Không cấp phát riêng cho từng phần tử. Tăng trưởng là append mảng amortized
O(1)(xem ./04-arrays.md).
Điểm gài là nó chỉ hoạt động vì cây là complete. Một binary tree tổng quát lưu theo cách này sẽ cần 2^h ô cho h level, đó là lý do mẹo này chỉ dành riêng cho heap.
Sift-up và sift-down
Mọi thao tác heap đều là một trong hai thủ tục sửa chữa. Cả hai đều đi dọc một đường root-tới-leaf duy nhất, nên cả hai đều O(log n).
Sift-up (còn gọi là bubble-up, percolate-up) sửa một node có thể quá nhỏ so với vị trí của nó — dùng sau khi append phần tử mới vào cuối:
push 0 vào [1, 3, 2, 7, 5, 8, 9]:
append tại index 7: 0 < parent 7 -> đổi chỗ: 0 < parent 3 -> đổi chỗ:
(1) (1) (1)
/ \ / \ / \
(3) (2) (3) (2) (0) (2)
/ \ / \ / \ / \ / \ / \
(7) (5) (8) (9) (0) (5) (8) (9) (3) (5) (8) (9)
/ / /
(0) (7) (7)
0 < parent 1 -> đổi chỗ -> xong
Sift-down (bubble-down, percolate-down, heapify) sửa một node có thể quá lớn so với vị trí của nó — dùng sau khi chuyển phần tử cuối lên root trong lúc pop:
pop từ [1, 3, 2, 7, 5, 8, 9]: lấy 1, chuyển phần tử cuối (9) lên root, sift down.
(9) 9 > min(3,2)=2 (2) 9 > min(8,9)=8 (2)
/ \ -> đổi với 2 / \ -> đổi với 8 / \
(3) (2) ============> (3) (9) ===========> (3) (8)
/ \ / / \ / / \ /
(7) (5) (8) (7) (5) (8) (7) (5) (9)
Chi tiết then chốt trong sift-down: bạn phải so sánh với child NHỎ HƠN trong hai child (với min-heap). Đổi chỗ với child lớn hơn sẽ đặt một giá trị lên trên một giá trị nhỏ hơn và phá vỡ bất biến ngay lập tức. Đây là bug phổ biến nhất trong một heap tự viết tay.
class MinHeap:
"""Binary min-heap trên một list Python phẳng. Không pointer, không node."""
def __init__(self):
self.a = []
def __len__(self):
return len(self.a)
def peek(self):
"""Minimum. O(1) — nó chính là a[0]."""
return self.a[0]
def push(self, item):
"""Append vào cuối, rồi sift up. O(log n)."""
self.a.append(item)
self._sift_up(len(self.a) - 1)
def pop(self):
"""Bỏ và trả về minimum. O(log n)."""
last = self.a.pop()
if not self.a:
return last # heap chỉ có đúng một phần tử
top = self.a[0]
self.a[0] = last # chuyển leaf cuối lên root...
self._sift_down(0) # ...rồi để nó chìm về đúng tầng của nó
return top
def _sift_up(self, i):
"""Đẩy a[i] lên trong khi nó còn nhỏ hơn parent."""
item = self.a[i]
while i > 0:
parent = (i - 1) // 2
if self.a[parent] <= item:
break # tính chất heap đã được khôi phục
self.a[i] = self.a[parent] # dịch parent xuống; không cần swap đầy đủ
i = parent
self.a[i] = item # ghi một lần duy nhất tại vị trí cuối cùng
def _sift_down(self, i):
"""Đẩy a[i] xuống trong khi nó còn lớn hơn child NHỎ NHẤT của nó."""
n = len(self.a)
item = self.a[i]
while True:
left = 2 * i + 1
if left >= n:
break # a[i] là leaf
right = left + 1
child = left
if right < n and self.a[right] < self.a[left]:
child = right # chọn child nhỏ hơn — làm sai chỗ này
# sẽ âm thầm phá vỡ bất biến
if item <= self.a[child]:
break
self.a[i] = self.a[child]
i = child
self.a[i] = item
@classmethod
def heapify(cls, items):
"""Dựng heap từ một iterable bất kỳ trong O(n) — xem bên dưới."""
h = cls()
h.a = list(items)
for i in range(len(h.a) // 2 - 1, -1, -1):
h._sift_down(i)
return h
Chú ý kỹ thuật “lỗ trống” trong _sift_up và _sift_down: thay vì swap ở mỗi bước (ba lần ghi mỗi bước), code dịch phần tử bị đẩy vào chỗ và chỉ ghi phần tử đang di chuyển đúng một lần ở cuối. Cùng độ phức tạp, nhưng lưu lượng bộ nhớ chỉ còn khoảng một phần ba. heapq của CPython dùng đúng mẹo này.
Vì sao heapify là O(n) chứ không phải O(n log n)
Cách ngây thơ để dựng heap từ n phần tử là n lần push, mỗi lần O(log n), cho ra O(n log n). Hàm heapify ở trên làm tốt hơn: nó gọi _sift_down trên mọi node non-leaf, từ node cuối cùng ngược về root, và chạy trong O(n).
Điều bất ngờ được giải thích ngay khi bạn nhận ra rằng chi phí của _sift_down phụ thuộc vào height của node, không phải của cả cây, và gần như mọi node đều nằm gần đáy.
Trong một heap n node:
| Level (tính từ đáy) | Số node | Khoảng sift-down tối đa | Tổng công |
|---|---|---|---|
| 0 (leaf) | ~n/2 | 0 | 0 |
| 1 | ~n/4 | 1 | n/4 |
| 2 | ~n/8 | 2 | 2n/8 |
| 3 | ~n/16 | 3 | 3n/16 |
| … | … | … | … |
h (root) | 1 | h | h |
Cộng lại:
Tổng = Σ (n / 2^(k+1)) · k với k = 0 .. h
= n · Σ k / 2^(k+1)
∞
và chuỗi Σ k / 2^(k+1) hội tụ về 1
k=0
Tổng ≈ n · 1 = O(n)
Trực giác không cần đại số: một nửa số node là leaf và làm việc bằng không; một phần tư làm nhiều nhất một lần đổi chỗ; một phần tám làm nhiều nhất hai lần. Những node có thể tốn kém lại chính là những node gần như không có mấy. Chuỗi 1/2 + 2/4 + 3/8 + 4/16 + … hội tụ, nên tổng là tuyến tính.
Đối lập với dựng heap bằng sift-up (for i in range(n): sift_up(i)). Cách đó là O(n log n) và không tuyến tính, vì chi phí sift-up phụ thuộc vào depth, và một nửa số node nằm ở depth lớn nhất. Cùng một cây, ngược chiều, khác độ phức tạp. Đây là một cặp ví dụ thật sự có tính giáo dục — sự bất đối xứng hoàn toàn nằm ở chỗ đám đông tập trung ở đầu nào của cây.
Về mặt thực dụng: heapq.heapify(a) là O(n); một vòng lặp heapq.heappush là O(n log n). Khi bạn đã có sẵn toàn bộ dữ liệu, luôn dùng heapify.
Bảng tổng hợp độ phức tạp
| Thao tác | Binary heap | Mảng đã sắp xếp | Mảng chưa sắp xếp | Balanced BST |
|---|---|---|---|---|
peek min/max | O(1) | O(1) | O(n) | O(log n) |
push | O(log n) | O(n) | O(1) amortized | O(log n) |
pop cực trị | O(log n) | O(1) (từ cuối) | O(n) | O(log n) |
Dựng từ n phần tử | O(n) | O(n log n) | O(n) | O(n log n) |
| Tìm key bất kỳ | O(n) | O(log n) | O(n) | O(log n) |
| Xoá key bất kỳ | O(n) để tìm + O(log n) | O(n) | O(n) | O(log n) |
| Decrease-key (biết chỉ số) | O(log n) | O(n) | O(1) | O(log n) |
| Duyệt có thứ tự | O(n log n) (phá huỷ) | O(n) | O(n log n) | O(n) |
| Overhead bộ nhớ | không có | không có | không có | 2 pointer/node |
| Gộp hai cấu trúc | O(n) | O(n) | O(1) | O(n) |
Dòng quyết định phần lớn các thiết kế là tìm kiếm: heap không thể tìm một phần tử bất kỳ nhanh hơn quét tuyến tính. Nếu workload của bạn cần cả “cho tôi minimum” lẫn “xoá đúng phần tử này”, heap trần là lựa chọn sai — bạn cần heap kèm một index map, hoặc balanced BST, hoặc pattern lazy deletion mô tả bên dưới.
Khái niệm chính
heapq của Python
Python không có class heap; heapq là một tập hàm thao tác trên một list thông thường được diễn giải như min-heap. Đây là bản cài đặt bạn nên dùng trong production — nó viết bằng C, được test kỹ, và dùng đúng các thuật toán ở trên.
import heapq
a = [5, 7, 9, 1, 3]
heapq.heapify(a) # O(n) — tại chỗ, a giờ là min-heap hợp lệ
heapq.heappush(a, 4) # O(log n)
smallest = heapq.heappop(a) # O(log n) -> 1
peek = a[0] # O(1) — minimum luôn nằm ở index 0
# Thao tác gộp: rẻ hơn hai lời gọi riêng, và giữ nguyên kích thước
heapq.heappushpop(a, 6) # push rồi pop — một lần sift-down, không tăng size
heapq.heapreplace(a, 6) # pop rồi push — item KHÔNG được so sánh trước
# Hàm tiện ích. O(n log k) với heap kích thước k bên trong — tốt khi
# k << n; nếu k gần bằng n, cứ gọi sorted().
heapq.nlargest(3, a)
heapq.nsmallest(3, a)
# Gộp k-way các iterable đã sắp xếp. Lazy: O(1) bộ nhớ mỗi nguồn,
# tổng O(N log k). Đây là lõi của external sorting và LSM compaction.
merged = list(heapq.merge([1, 4, 7], [2, 5, 8], [3, 6, 9]))
Khác biệt giữa heappushpop và heapreplace hay làm người ta vấp. heappushpop(h, x) trả về min(x, h[0]) — nếu x nhỏ hơn mọi thứ trong heap, nó quay ra ngay và heap không bị đụng tới. heapreplace(h, x) luôn pop minimum hiện tại trước rồi mới chèn x, nên x có thể lọt vào heap ngay cả khi nó nhỏ hơn phần tử vừa bị bỏ ra. Với vòng lặp “giữ k phần tử lớn nhất”, heappushpop mới là thứ bạn cần.
Max-heap trong Python: mẹo đổi dấu
heapq chỉ có min. Không có heapq.heapifymax trong API công khai. Cách xử lý chuẩn là đổi dấu:
import heapq
max_heap = []
for value in [5, 1, 9, 3]:
heapq.heappush(max_heap, -value) # lưu giá trị đối
largest = -heapq.heappop(max_heap) # đổi dấu lại khi lấy ra -> 9
peek_largest = -max_heap[0] # O(1)
Hai lưu ý gây đau trong code thật:
- Chỉ hoạt động với số.
-xkhông xác định với string, tuple, và đa số object. Với những thứ đó, hoặc bọc trong một class đảo ngược phép so sánh, hoặc push một tuple(-priority, payload)trong đó chỉ có priority dạng số bị đổi dấu. - Chú ý sự bất đối xứng của khoảng số nguyên trong các ngôn ngữ có độ rộng cố định. Trong Python, số nguyên có độ chính xác tuỳ ý nên
-xluôn an toàn; trong C/Java,-Integer.MIN_VALUEtràn số và quay lại chính nó. Đáng nhớ khi port code.
Phương án sạch hơn khi payload không so sánh được là push tuple và để phép so sánh tuple theo thứ tự từ điển của Python làm việc:
import heapq
import itertools
counter = itertools.count() # bộ phá hoà đơn điệu tăng
tasks = []
def add_task(priority, task):
# (priority, thứ tự insert, payload)
# Counter đảm bảo một thứ tự toàn phần, nên `task` KHÔNG BAO GIỜ bị so sánh.
# Nếu thiếu nó, hai priority bằng nhau sẽ khiến Python so sánh payload,
# và điều đó ném TypeError với object không định nghĩa thứ tự.
heapq.heappush(tasks, (priority, next(counter), task))
add_task(2, {"name": "send email"})
add_task(1, {"name": "pay invoice"})
add_task(2, {"name": "archive logs"})
priority, _, task = heapq.heappop(tasks) # -> (1, ..., {'name': 'pay invoice'})
Bộ phá hoà itertools.count() không phải trang trí tuỳ chọn — nó là cách sửa chuẩn cho lỗi TypeError: '<' not supported between instances of 'dict' and 'dict' mà một heap (priority, payload) ngây thơ sẽ ném ra ngay lần đầu hai priority trùng nhau. Nó cũng làm cho queue ổn định: các priority bằng nhau ra theo thứ tự insert.
ADT priority queue
Priority queue là kiểu trừu tượng; binary heap là bản cài đặt thông dụng. Giao diện:
| Thao tác | Ý nghĩa | Chi phí trên binary heap |
|---|---|---|
insert(item, priority) | Thêm một phần tử | O(log n) |
extract_min() / extract_max() | Bỏ và trả về phần tử ưu tiên cao nhất | O(log n) |
peek() | Xem phần tử ưu tiên cao nhất | O(1) |
decrease_key(item, p) | Nâng độ ưu tiên của một phần tử | O(log n) nếu biết chỉ số của nó |
merge(other) | Gộp hai queue | O(n) |
Hai thao tác cuối là chỗ binary heap bộc lộ giới hạn, và là chỗ các biến thể heap kỳ lạ chứng minh giá trị:
| Biến thể heap | insert | extract-min | decrease-key | merge | Trong thực tế |
|---|---|---|---|---|---|
| Binary heap | O(log n) | O(log n) | O(log n) | O(n) | Mặc định. Hằng số tốt nhất, cache tốt nhất |
| d-ary heap | O(log_d n) | O(d · log_d n) | O(log_d n) | O(n) | d = 4 thường nhanh nhất thực tế: nông hơn, hợp cache hơn |
| Binomial heap | O(log n) | O(log n) | O(log n) | O(log n) | Khi gộp queue diễn ra thường xuyên |
| Fibonacci heap | O(1) am. | O(log n) am. | O(1) am. | O(1) | Tối ưu lý thuyết cho Dijkstra; hằng số tệ tới mức thường thua |
| Pairing heap | O(1) | O(log n) am. | o(log n) am. | O(1) | Bản thay thế thực dụng của Fibonacci heap |
Dòng Fibonacci heap là một bài học nổi tiếng về khoảng cách giữa tiệm cận và thực tế. Nó biến Dijkstra thành O(E + V log V) thay vì O((V + E) log V), đó là một cải tiến lý thuyết thật sự — vậy mà nó gần như không bao giờ được dùng, vì hằng số nhân và pattern bộ nhớ đuổi theo pointer khiến nó chậm hơn binary heap trên mọi input bạn thực sự gặp. Biết là nó tồn tại; còn dùng thì dùng heapq.
Vấn đề decrease_key trong Python. heapq hoàn toàn không có decrease_key, vì một list phẳng không cho bạn cách nào tìm chỉ số của một phần tử. Cách xử lý chuẩn là lazy deletion: push entry mới (tốt hơn) mà không xoá entry cũ, rồi bỏ qua các entry lỗi thời khi chúng nổi lên.
import heapq
def dijkstra_style_loop(start, neighbours):
"""Pattern lazy deletion mà mọi bản Dijkstra dựa trên heapq đều dùng."""
dist = {start: 0}
pq = [(0, start)]
while pq:
d, u = heapq.heappop(pq)
if d > dist.get(u, float("inf")):
continue # entry lỗi thời: một đường tốt hơn đã được
# tìm thấy sau khi entry này được push. Bỏ qua.
for v, w in neighbours(u):
nd = d + w
if nd < dist.get(v, float("inf")):
dist[v] = nd
heapq.heappush(pq, (nd, v)) # push một bản trùng thay vì
# giảm key của entry cũ
return dist
Heap có thể phình tới O(E) entry thay vì O(V), nhưng mỗi entry lỗi thời bị loại trong O(1) và tổng vẫn là O(E log E) = O(E log V). Đây là cách tiếp cận idiomatic của Python và là thứ bạn nên viết. Xem ./14-shortest-path-algorithms.md.
Heapsort
Heap cho ra một thuật toán sắp xếp gần như miễn phí: heapify mảng, rồi liên tục đổi root với phần tử cuối và thu nhỏ heap đi một.
def heapsort(a):
"""Tại chỗ, O(n log n) worst case, O(1) bộ nhớ phụ, KHÔNG stable."""
n = len(a)
def sift_down(i, size):
# Bản max-heap: đẩy a[i] chìm xuống khi nó còn nhỏ hơn child lớn nhất
while True:
left = 2 * i + 1
if left >= size:
return
child = left
right = left + 1
if right < size and a[right] > a[left]:
child = right
if a[i] >= a[child]:
return
a[i], a[child] = a[child], a[i]
i = child
for i in range(n // 2 - 1, -1, -1): # dựng MAX-heap: O(n)
sift_down(i, n)
for end in range(n - 1, 0, -1): # n-1 lần lấy ra: O(n log n)
a[0], a[end] = a[end], a[0] # phần tử lớn nhất về đúng vị trí cuối
sift_down(0, end) # khôi phục heap trên a[0:end]
return a
O(n log n) worst case và O(1) bộ nhớ của heapsort là những đảm bảo chặt hơn hẳn quicksort. Vậy mà nó vẫn thua quicksort trong thực tế vì pattern truy cập (i → 2i+1) nhảy loạn khắp mảng và phá huỷ cache locality, trong khi phép partition của quicksort là hai lượt quét tuần tự. Vai trò production thật sự của nó là làm phương án dự phòng trong introsort (std::sort của C++): chạy quicksort cho tới khi đệ quy sâu một cách đáng ngờ, rồi chuyển sang heapsort để đảm bảo O(n log n). Xem ./08-sorting-algorithms.md.
Pattern two-heaps: running median
Bài toán: các con số đến lần lượt từ một stream, và sau mỗi số bạn phải báo cáo median. Sắp xếp lại là O(n log n) mỗi phần tử. Chèn vào list đã sắp xếp là O(n) mỗi phần tử. Pattern two-heaps biến nó thành O(log n) cho insert và O(1) cho median.
Ý tưởng: cắt dữ liệu tại median thành hai nửa, và giữ mỗi nửa trong loại heap đặt biên của nửa đó lên đỉnh.
max-heap "low" min-heap "high"
(nửa dưới của dữ liệu) (nửa trên của dữ liệu)
(8) <- lớn nhất của nửa dưới (10) <- nhỏ nhất của nửa trên
/ \ / \
(5) (7) (12) (11)
/ \ /
(2) (3) (15)
Hai root kẹp lấy median. Nếu hai size bằng nhau, median là trung bình của
chúng; nếu `low` nhiều hơn một phần tử, median là root của low.
Các bất biến phải duy trì sau mỗi lần insert:
- Thứ tự: mọi phần tử trong
lowđều≤mọi phần tử tronghigh. - Cân bằng:
len(low) - len(high)bằng 0 hoặc 1.
import heapq
class MedianFinder:
"""Running median trên stream. add() là O(log n), median() là O(1)."""
def __init__(self):
self.low = [] # max-heap của nửa dưới, lưu dạng đổi dấu
self.high = [] # min-heap của nửa trên
def add(self, num):
# Ba dòng, và chúng duy trì cả hai bất biến một cách vô điều kiện:
# 1. Luôn push vào `low` trước (nên `low` là nửa lớn hơn theo mặc định).
heapq.heappush(self.low, -num)
# 2. Chuyển phần tử lớn nhất của low sang `high`. Điều này đảm bảo bất biến
# THỨ TỰ: thứ nằm trên đỉnh `low` giờ chắc chắn <= root của high.
heapq.heappush(self.high, -heapq.heappop(self.low))
# 3. Nếu điều đó làm `high` lớn hơn, chuyển phần tử nhỏ nhất về — khôi phục CÂN BẰNG.
if len(self.high) > len(self.low):
heapq.heappush(self.low, -heapq.heappop(self.high))
def median(self):
if len(self.low) > len(self.high):
return float(-self.low[0]) # số lượng lẻ: root của low
return (-self.low[0] + self.high[0]) / 2 # số lượng chẵn: trung bình hai root
Ba dòng của add đáng để nghiền ngẫm. Bản ngây thơ — “so sánh với hai root, quyết định push vào heap nào, rồi cân bằng lại” — cần bốn hoặc năm nhánh rẽ và rất dễ sai tinh vi ở các ca biên heap rỗng và giá trị bằng nhau. Push vô điều kiện vào low, chuyển ngay đỉnh sang high, rồi cân bằng ngược lại nếu cần, xử lý được mọi trường hợp mà không cần điều kiện nào ngoài phép kiểm tra size cuối cùng.
Pattern này tổng quát hoá ở đâu. Bất cứ khi nào bạn cần duy trì một điểm cắt trên một multiset động — median, một percentile, “phần tử lớn thứ k tính tới giờ”, “cửa sổ hiện tại có cân bằng không” — hai heap quay mặt vào nhau qua biên chính là công cụ. Các biến thể:
- Sliding-window median: cùng cấu trúc cộng thêm lazy deletion (một
dictđếm số phần tử chờ bị xoá, được dọn khỏi đỉnh heap khi chúng nổi lên). - Bài toán IPO / lập lịch task: một max-heap các phần tử đã sẵn sàng theo lợi nhuận và một min-heap các phần tử chưa sẵn sàng theo thời điểm mở khoá.
- Percentile khác 50%: đổi bất biến cân bằng từ
|low| − |high| ∈ {0,1}thành tỷ lệ bạn cần.
Tìm phần tử lớn thứ k
Một bài toán kinh điển với ba lời giải tốt và một lời giải tệ.
import heapq
import random
# ---- Cách 1: sắp xếp. O(n log n) thời gian, O(n) hoặc O(1) bộ nhớ. ----------
def kth_largest_sort(nums, k):
"""Đơn giản nhất và thường nhanh nhất với n nhỏ — Timsort có hằng số rất tốt."""
return sorted(nums, reverse=True)[k - 1]
# ---- Cách 2: min-heap kích thước k. O(n log k) thời gian, O(k) bộ nhớ. ------
def kth_largest_heap(nums, k):
"""
Chỉ giữ k phần tử lớn nhất đã thấy. Root của heap là phần tử NHỎ NHẤT trong
k phần tử đó, đúng bằng phần tử lớn thứ k toàn cục khi stream kết thúc.
"""
heap = nums[:k]
heapq.heapify(heap) # O(k)
for x in nums[k:]:
if x > heap[0]: # loại bỏ O(1) — đa số phần tử rơi vào đây
heapq.heapreplace(heap, x) # O(log k): pop nhỏ nhất, push x
return heap[0]
# Bản một dòng tương đương trong standard library (cùng thuật toán O(n log k)):
# heapq.nlargest(k, nums)[-1]
# ---- Cách 3: quickselect. O(n) trung bình, O(n^2) worst, O(1) bộ nhớ. -------
def partition(a, lo, hi):
"""Lomuto partition với pivot ngẫu nhiên; trả về chỉ số cuối cùng của pivot."""
r = random.randint(lo, hi)
a[r], a[hi] = a[hi], a[r] # ngẫu nhiên hoá vô hiệu hoá input adversarial
pivot = a[hi]
i = lo
for j in range(lo, hi):
if a[j] < pivot:
a[i], a[j] = a[j], a[i]
i += 1
a[i], a[hi] = a[hi], a[i]
return i
def quickselect(a, k):
"""Phần tử NHỎ thứ k (0-indexed). Phá vỡ thứ tự của `a`."""
lo, hi = 0, len(a) - 1
while True:
if lo == hi:
return a[lo]
p = partition(a, lo, hi)
if k == p:
return a[k]
if k < p:
hi = p - 1 # chỉ đệ quy vào MỘT phía — đó là lý do nó O(n)
else:
lo = p + 1
def kth_largest_quickselect(nums, k):
"""Lớn thứ k = nhỏ thứ (n - k), 0-indexed."""
return quickselect(list(nums), len(nums) - k)
So sánh:
| Cách | Thời gian | Bộ nhớ | Streaming? | Khi nào dùng |
|---|---|---|---|---|
| Sắp xếp toàn bộ | O(n log n) | O(n) | Không | n nhỏ, hoặc bạn cần luôn các thống kê thứ tự khác |
Min-heap kích thước k | O(n log k) | O(k) | Có | k ≪ n, hoặc dữ liệu là stream / không vừa bộ nhớ |
| Quickselect | O(n) TB, O(n²) worst | O(1) | Không | Cả mảng nằm trong bộ nhớ và bạn muốn một đáp án nhanh nhất |
| Max-heap + k lần pop | O(n + k log n) | O(n) | Không | Bạn cần top k theo thứ tự và k nhỏ |
Hai điểm đáng lưu ý:
Vì sao quickselect là O(n). Quicksort đệ quy vào cả hai nửa, cho T(n) = 2T(n/2) + O(n) = O(n log n). Quickselect chỉ đệ quy vào nửa có thể chứa đáp án, cho T(n) = T(n/2) + O(n), và cấp số nhân n + n/2 + n/4 + … = 2n cộng lại thành O(n). Pivot ngẫu nhiên là thứ khiến worst case O(n²) (luôn chọn phần tử cực trị) trở nên cực kỳ khó xảy ra thay vì bị kích hoạt bởi dữ liệu đã sắp xếp. Xem ./20-brute-force-greedy-and-randomised-algorithms.md và ./19-recursion-and-divide-and-conquer.md.
Vì sao cách dùng heap thắng trong thực tế nhiều hơn bảng trên gợi ý. Quickselect tốt hơn về tiệm cận nhưng cần cả mảng nằm trong bộ nhớ và làm thay đổi mảng đó. Min-heap kích thước k hoạt động trên một stream không biết trước độ dài, dùng O(k) bộ nhớ bất kể n lớn cỡ nào, và — nhờ điều kiện chặn if x > heap[0] — thực hiện một phép loại bỏ O(1) rẻ tiền cho tuyệt đại đa số phần tử. Với bài toán “top 10 mặt hàng xu hướng trong 100 triệu event”, heap không chỉ dễ hơn, nó là cách duy nhất trong hai cách thực sự chạy được.
Gộp k-way
Ứng dụng thường ngày còn lại của heap: gộp k dãy đã sắp xếp thành một output đã sắp xếp. Lấy đầu nhỏ nhất trong k dãy, xuất nó ra, tiến dãy đó lên một, lặp lại. Một heap kích thước k biến “đầu nhỏ nhất” thành câu hỏi O(log k).
import heapq
def k_way_merge(sorted_lists):
"""
O(N log k) với N là tổng số phần tử. So với việc nối tất cả lại rồi sắp xếp:
O(N log N). Khi k nhỏ và N khổng lồ, khác biệt là thật — và khác với sắp xếp,
cách này chạy được theo kiểu streaming.
"""
heap = []
for list_index, lst in enumerate(sorted_lists):
if lst:
# (giá trị, thuộc list nào, vị trí trong list đó)
heapq.heappush(heap, (lst[0], list_index, 0))
out = []
while heap:
value, list_index, pos = heapq.heappop(heap)
out.append(value)
next_pos = pos + 1
if next_pos < len(sorted_lists[list_index]):
heapq.heappush(heap, (sorted_lists[list_index][next_pos], list_index, next_pos))
return out
# Trong production, hãy dùng standard library — nó lazy, nên không bao giờ
# tạo ra toàn bộ output và chạy được trên file, generator, và DB cursor:
# heapq.merge(*sorted_iterables, key=None, reverse=False)
Đây là bước merge của external sorting (sắp xếp từng chunk vừa RAM, ghi ra đĩa, rồi gộp k-way các file), bước compaction của các storage engine LSM-tree như RocksDB và Cassandra, và cách một query engine phân tán gộp kết quả bộ phận đã sắp xếp từ k shard. Xem ../../data-engineer/vi/10-big-data-and-distributed-computing.md.
Best Practices
- Dùng
heapq, không tự viết heap. Nó cài bằng C, đúng đắn, và dùng tối ưu hoá dịch-lỗ-trống. Hãy tự viết một lần để hiểu sift-up/sift-down, rồi đừng bao giờ viết lại. heapifykhi bạn đã có toàn bộ dữ liệu; chỉheappushtrong vòng lặp khi bạn chưa có.O(n)so vớiO(n log n)cho cùng một kết quả.- Luôn thêm bộ phá hoà vào các entry dạng tuple.
(priority, itertools.count(), payload). Thiếu nó, các priority bằng nhau khiến Python so sánh payload, và điều đó némTypeErrorvới bất cứ thứ gì không định nghĩa thứ tự — và nó sẽ không lỗi cho tới lần trùng priority đầu tiên trong production. - Đổi dấu để có max-heap, nhưng chỉ với số. Với key không phải số, chỉ đổi dấu phần priority dạng số trong một tuple, hoặc bọc phần tử trong một class có
__lt__đảo ngược. - Đừng dùng heap khi bạn cần tìm hoặc xoá phần tử bất kỳ. Trong heap đó là
O(n). Hãy dùng balanced BST,SortedList, hoặc heap kèm một dictionary chỉ số. - Dùng lazy deletion thay cho
decrease_key. Push entry đã cải thiện và bỏ qua entry lỗi thời khi pop (if d > dist[u]: continue). Đây là idiom Dijkstra chuẩn trong Python và nó ổn về mặt tiệm cận. - Với bài toán “top k”, chặn heap ở kích thước k.
heapq.nlargest(k, iterable)hoặc một min-heap kích thướcktự viết choO(n log k)thời gian vàO(k)bộ nhớ — cực kỳ quan trọng khinkhông vừa RAM. - Ưu tiên
heappushpophơnheappush+heappop. Một lần sift-down thay vì hai thao tác heap, và heap không bao giờ phình vượt kích thước mục tiêu. - Nghĩ tới hai heap bất cứ khi nào phải duy trì một điểm cắt động. Running median, percentile, “rẻ nhất đang sẵn sàng so với mở khoá sớm nhất” — cặp heap quay mặt vào biên chính là pattern đó.
- Đừng sắp xếp chỉ để tìm một cực trị.
min(a)làO(n);sorted(a)[0]làO(n log n). Tương tự, đừng sắp xếp chỉ để tìm phần tử thứ k — hãy dùng quickselect hoặc heap có chặn kích thước. - Nhớ vai trò thật sự của heapsort. Là một thuật toán sắp xếp độc lập, nó thua quicksort và Timsort về cache, nhưng
O(n log n)worst case vàO(1)bộ nhớ khiến nó là tấm lưới an toàn đúng đắn bên trong introsort. heapqkhông thread-safe. Hãy bảo vệ bằng lock, hoặc dùngqueue.PriorityQueuevốn đã bọcheapqkèm sẵn phần khoá.
Tài liệu tham khảo
- roadmap.sh — Data Structures & Algorithms
- Heap (data structure) — Wikipedia
- Binary heap — Wikipedia
- Priority queue — Wikipedia
- Heapsort — Wikipedia
- Quickselect — Wikipedia
- Fibonacci heap — Wikipedia
- Python Documentation —
heapq: Heap queue algorithm - Python Documentation —
queue.PriorityQueue - CLRS — Introduction to Algorithms, Chapter 6: Heapsort
- MIT 6.006 — Introduction to Algorithms (OpenCourseWare)
- VisuAlgo — Binary Heap visualization
- Big-O Cheat Sheet
Part of the Data Structures & Algorithms Roadmap knowledge base.
Overview
A sorted array answers “what is the smallest element?” in O(1) but costs O(n) to insert into. A plain list answers insertion in O(1) but costs O(n) to find the minimum. A balanced BST does both in O(log n), but you are paying for a total ordering you did not ask for.
A heap is the structure for the case where you only ever need one end of the ordering: the smallest element, or the largest, repeatedly. It gives you O(1) peek at the extreme, O(log n) insert, O(log n) extract, and — the part that surprises people — O(n) to build one from an unsorted array. It does all of this with no pointers at all, stored in a flat array, which makes it one of the most cache-friendly structures in this whole knowledge base.
The abstract operation a heap implements is the priority queue: an ADT where elements are dequeued in priority order rather than insertion order. That ADT is everywhere in real systems:
- Dijkstra’s algorithm and A* repeatedly extract the nearest unvisited node — see ./14-shortest-path-algorithms.md.
- Prim’s algorithm repeatedly extracts the cheapest edge crossing the cut — see ./15-minimum-spanning-trees.md.
- OS schedulers pick the highest-priority runnable task.
- Event simulation processes the next event by timestamp.
- Huffman coding repeatedly merges the two least-frequent symbols.
- Rate limiters, timers, and job queues fire the soonest-due item.
heapq.merge, external sorting, and LSM-tree compaction mergeksorted runs by repeatedly taking the smallest head.- Streaming top-k and running median keep a bounded window of the interesting elements.
A common source of confusion first: “heap” here has nothing to do with “the heap” in memory management. The dynamic allocation region and this data structure share a name for historical reasons and are otherwise unrelated.
Fundamentals
The heap property
A binary heap is a complete binary tree (see ./10-tree-data-structures.md) satisfying one local invariant:
- Min-heap: every node’s key is
≤both of its children’s keys. The minimum is therefore at the root. - Max-heap: every node’s key is
≥both of its children’s keys. The maximum is at the root.
That is the entire invariant. Note how weak it is compared to a BST:
A valid min-heap: NOT a BST, and that's fine:
(1) - 3 is in 1's left subtree and 8 is in
/ \ 1's right subtree, but 3 > ... nothing.
(3) (2) - There is no ordering between siblings.
/ \ / \ - In-order traversal gives 7 3 5 1 8 2 9,
(7) (5) (8) (9) which is not sorted.
A heap is only partially ordered: it knows relationships along root-to-leaf paths and nothing about left-right relationships. That is precisely why it is cheaper than a BST — it does less work because it promises less. You cannot search a heap for an arbitrary key in better than O(n), and you cannot iterate it in sorted order without destroying it.
The implicit array — a tree with no pointers
Because a heap is a complete binary tree, its nodes can be numbered level by level with no gaps, and stored in a flat array. The parent/child relationships become arithmetic:
index 0
(1)
/ \
index 1 index 2
(3) (2)
/ \ / \
(7) (5) (8) (9)
3 4 5 6
array: [ 1, 3, 2, 7, 5, 8, 9 ]
index: 0 1 2 3 4 5 6
For 0-based indexing:
| Relationship | Formula |
|---|---|
Left child of i | 2i + 1 |
Right child of i | 2i + 2 |
Parent of i | (i − 1) // 2 |
Is i a leaf? | 2i + 1 >= n |
| First leaf index | n // 2 |
| Last non-leaf index | n // 2 − 1 |
(Some textbooks use 1-based indexing, where the formulas are the slightly prettier 2i, 2i+1, and i // 2. Always check which convention a source uses before transcribing.)
This layout is the heap’s second big advantage after its complexity:
- Zero pointer overhead. A BST node costs a value plus two pointers (16 bytes of overhead on a 64-bit machine); a heap element costs the value alone.
- Excellent cache locality. The top few levels of the heap — the ones touched on every single operation — occupy the first few dozen array slots, which stay in L1 cache permanently. A pointer-based tree scatters those same nodes across the address space.
- No allocation per element. Growth is amortized
O(1)array append (see ./04-arrays.md).
The catch is that it only works because the tree is complete. A general binary tree stored this way would need 2^h slots for h levels, which is why this trick is specific to heaps.
Sift-up and sift-down
Every heap operation is one of two repair routines. Both walk a single root-to-leaf path, so both are O(log n).
Sift-up (also called bubble-up, percolate-up, sift-down in CPython’s confusing internal naming) fixes a node that may be too small for its position — used after appending a new element at the end:
push 0 into [1, 3, 2, 7, 5, 8, 9]:
append at index 7: 0 < parent 7 -> swap: 0 < parent 3 -> swap:
(1) (1) (1)
/ \ / \ / \
(3) (2) (3) (2) (0) (2)
/ \ / \ / \ / \ / \ / \
(7) (5) (8) (9) (0) (5) (8) (9) (3) (5) (8) (9)
/ / /
(0) (7) (7)
0 < parent 1 -> swap -> done
Sift-down (bubble-down, percolate-down, heapify) fixes a node that may be too large for its position — used after moving the last element to the root during a pop:
pop from [1, 3, 2, 7, 5, 8, 9]: take 1, move last (9) to the root, sift down.
(9) 9 > min(3,2)=2 (2) 9 > min(8,9)=8 (2)
/ \ -> swap with 2 / \ -> swap with 8 / \
(3) (2) ============> (3) (9) ===========> (3) (8)
/ \ / / \ / / \ /
(7) (5) (8) (7) (5) (8) (7) (5) (9)
The crucial detail in sift-down: you must compare against the smaller of the two children (for a min-heap). Swapping with the larger child would put a value above a smaller value and break the invariant immediately. This is the single most common bug in a hand-written heap.
class MinHeap:
"""A binary min-heap over a flat Python list. No pointers, no nodes."""
def __init__(self):
self.a = []
def __len__(self):
return len(self.a)
def peek(self):
"""The minimum. O(1) — it is just a[0]."""
return self.a[0]
def push(self, item):
"""Append at the end, then sift up. O(log n)."""
self.a.append(item)
self._sift_up(len(self.a) - 1)
def pop(self):
"""Remove and return the minimum. O(log n)."""
last = self.a.pop()
if not self.a:
return last # the heap had exactly one element
top = self.a[0]
self.a[0] = last # move the last leaf to the root...
self._sift_down(0) # ...and let it sink back to its level
return top
def _sift_up(self, i):
"""Move a[i] up while it is smaller than its parent."""
item = self.a[i]
while i > 0:
parent = (i - 1) // 2
if self.a[parent] <= item:
break # heap property restored
self.a[i] = self.a[parent] # shift the parent down; no full swap needed
i = parent
self.a[i] = item # single write at the final resting place
def _sift_down(self, i):
"""Move a[i] down while it is larger than its SMALLEST child."""
n = len(self.a)
item = self.a[i]
while True:
left = 2 * i + 1
if left >= n:
break # a[i] is a leaf
right = left + 1
child = left
if right < n and self.a[right] < self.a[left]:
child = right # pick the smaller child — getting this wrong
# silently breaks the invariant
if item <= self.a[child]:
break
self.a[i] = self.a[child]
i = child
self.a[i] = item
@classmethod
def heapify(cls, items):
"""Build a heap from an arbitrary iterable in O(n) — see below."""
h = cls()
h.a = list(items)
for i in range(len(h.a) // 2 - 1, -1, -1):
h._sift_down(i)
return h
Note the “hole” technique in _sift_up and _sift_down: instead of swapping on every step (three writes each), the code shifts the displaced element into place and writes the moving item exactly once at the end. Same complexity, roughly a third of the memory traffic. CPython’s heapq uses the same trick.
Why heapify is O(n), not O(n log n)
The naive way to build a heap from n items is n pushes, each O(log n), giving O(n log n). The heapify above does better: it calls _sift_down on every non-leaf node, from the last one backwards to the root, and runs in O(n).
The surprise resolves once you notice that _sift_down’s cost depends on the node’s height, not the tree’s, and almost every node is near the bottom.
In a heap of n nodes:
| Level (from the bottom) | Number of nodes | Max sift-down distance | Total work |
|---|---|---|---|
| 0 (leaves) | ~n/2 | 0 | 0 |
| 1 | ~n/4 | 1 | n/4 |
| 2 | ~n/8 | 2 | 2n/8 |
| 3 | ~n/16 | 3 | 3n/16 |
| … | … | … | … |
h (root) | 1 | h | h |
Summing the work:
Total = Σ (n / 2^(k+1)) · k for k = 0 .. h
= n · Σ k / 2^(k+1)
∞
and the series Σ k / 2^(k+1) converges to 1
k=0
Total ≈ n · 1 = O(n)
The intuition without the algebra: half the nodes are leaves and do zero work; a quarter do at most one swap; an eighth do at most two. The nodes that could be expensive are exactly the nodes there are almost none of. The series 1/2 + 2/4 + 3/8 + 4/16 + … converges, so the total is linear.
Contrast with building the heap by sift-up instead (for i in range(n): sift_up(i)). That is O(n log n) and not linear, because sift-up’s cost depends on depth, and half the nodes are at maximum depth. Same tree, opposite direction, different complexity. This is a genuinely instructive pair — the asymmetry is entirely about which end of the tree the crowd is at.
Practically: heapq.heapify(a) is O(n); a loop of heapq.heappush is O(n log n). When you already have all the data, always heapify.
Complexity summary
| Operation | Binary heap | Sorted array | Unsorted array | Balanced BST |
|---|---|---|---|---|
peek min/max | O(1) | O(1) | O(n) | O(log n) |
push | O(log n) | O(n) | O(1) amortized | O(log n) |
pop extreme | O(log n) | O(1) (from the end) | O(n) | O(log n) |
Build from n items | O(n) | O(n log n) | O(n) | O(n log n) |
| Search arbitrary key | O(n) | O(log n) | O(n) | O(log n) |
| Delete arbitrary key | O(n) to find + O(log n) | O(n) | O(n) | O(log n) |
| Decrease-key (index known) | O(log n) | O(n) | O(1) | O(log n) |
| Sorted iteration | O(n log n) (destructive) | O(n) | O(n log n) | O(n) |
| Space overhead | none | none | none | 2 pointers/node |
| Merge two structures | O(n) | O(n) | O(1) | O(n) |
The row that decides most designs is search: a heap cannot find an arbitrary element faster than a linear scan. If your workload needs both “give me the minimum” and “delete this specific element”, a bare heap is the wrong choice — you need a heap plus an index map, or a balanced BST, or the lazy-deletion pattern described below.
Key Concepts
Python’s heapq
Python has no heap class; heapq is a set of functions that operate on an ordinary list interpreted as a min-heap. This is the implementation you should use in production — it is written in C, well tested, and uses exactly the algorithms above.
import heapq
a = [5, 7, 9, 1, 3]
heapq.heapify(a) # O(n) — in-place, a is now a valid min-heap
heapq.heappush(a, 4) # O(log n)
smallest = heapq.heappop(a) # O(log n) -> 1
peek = a[0] # O(1) — the minimum is always at index 0
# Fused operations: cheaper than the two calls, and they keep the size fixed
heapq.heappushpop(a, 6) # push then pop — one sift-down, no growth
heapq.heapreplace(a, 6) # pop then push — the item is NOT compared first
# Convenience helpers. O(n log k) with an internal size-k heap — good when
# k << n; if k is close to n, just call sorted().
heapq.nlargest(3, a)
heapq.nsmallest(3, a)
# k-way merge of already-sorted iterables. Lazy: O(1) memory per source,
# O(N log k) total. This is the core of external sorting and LSM compaction.
merged = list(heapq.merge([1, 4, 7], [2, 5, 8], [3, 6, 9]))
The distinction between heappushpop and heapreplace catches people out. heappushpop(h, x) returns min(x, h[0]) — if x is smaller than everything in the heap, it comes straight back out and the heap is untouched. heapreplace(h, x) always pops the current minimum first, then inserts x, so x can end up in the heap even if it is smaller than the item removed. For a “keep the k largest” loop, heappushpop is what you want.
Max-heaps in Python: the negation trick
heapq is min-only. There is no heapq.heapifymax in the public API. The standard workaround is to negate:
import heapq
max_heap = []
for value in [5, 1, 9, 3]:
heapq.heappush(max_heap, -value) # store the negation
largest = -heapq.heappop(max_heap) # negate again on the way out -> 9
peek_largest = -max_heap[0] # O(1)
Two caveats that bite in real code:
- It only works for numbers.
-xis undefined for strings, tuples, and most objects. For those, either wrap in a comparison-reversing class or push a(-priority, payload)tuple where only the numeric priority is negated. - Watch the asymmetry of the integer range in fixed-width languages. In Python, integers are arbitrary precision, so
-xis always safe; in C/Java,-Integer.MIN_VALUEoverflows back to itself. Worth knowing when porting.
The cleaner alternative when the payload is not comparable is to push tuples and let Python’s lexicographic tuple comparison do the work:
import heapq
import itertools
counter = itertools.count() # a monotonically increasing tie-breaker
tasks = []
def add_task(priority, task):
# (priority, insertion_order, payload)
# The counter guarantees a total order, so `task` is NEVER compared.
# Without it, two equal priorities would make Python compare the payloads,
# which raises TypeError for objects that define no ordering.
heapq.heappush(tasks, (priority, next(counter), task))
add_task(2, {"name": "send email"})
add_task(1, {"name": "pay invoice"})
add_task(2, {"name": "archive logs"})
priority, _, task = heapq.heappop(tasks) # -> (1, ..., {'name': 'pay invoice'})
The itertools.count() tie-breaker is not optional decoration — it is the standard fix for the TypeError: '<' not supported between instances of 'dict' and 'dict' that a naive (priority, payload) heap throws the first time two priorities collide. It also makes the queue stable: equal priorities come out in insertion order.
The priority queue ADT
A priority queue is the abstract type; a binary heap is the usual implementation. The interface:
| Operation | Meaning | Binary heap cost |
|---|---|---|
insert(item, priority) | Add an element | O(log n) |
extract_min() / extract_max() | Remove and return the highest-priority item | O(log n) |
peek() | Look at the highest-priority item | O(1) |
decrease_key(item, p) | Raise an item’s priority | O(log n) if you know its index |
merge(other) | Combine two queues | O(n) |
The last two are where binary heaps show their limits, and where the exotic heap variants earn their keep:
| Heap variant | insert | extract-min | decrease-key | merge | In practice |
|---|---|---|---|---|---|
| Binary heap | O(log n) | O(log n) | O(log n) | O(n) | The default. Best constants, best cache behaviour |
| d-ary heap | O(log_d n) | O(d · log_d n) | O(log_d n) | O(n) | d = 4 is often fastest in practice: shallower, cache-friendlier |
| Binomial heap | O(log n) | O(log n) | O(log n) | O(log n) | When merging queues is frequent |
| Fibonacci heap | O(1) am. | O(log n) am. | O(1) am. | O(1) | Theoretically optimal for Dijkstra; constants so bad it usually loses |
| Pairing heap | O(1) | O(log n) am. | o(log n) am. | O(1) | The practical stand-in for a Fibonacci heap |
The Fibonacci heap row is a famous lesson in the gap between asymptotics and reality. It makes Dijkstra’s algorithm O(E + V log V) instead of O((V + E) log V), which is a real theoretical improvement — and it is almost never used, because its constant factors and pointer-chasing memory pattern make it slower than a binary heap on any input you will actually encounter. Know it exists; reach for heapq.
The decrease_key problem in Python. heapq has no decrease_key at all, because a flat list gives you no way to find an item’s index. The standard workaround is lazy deletion: push the new (better) entry without removing the old one, and discard stale entries when they surface.
import heapq
def dijkstra_style_loop(start, neighbours):
"""The lazy-deletion pattern every heapq-based Dijkstra uses."""
dist = {start: 0}
pq = [(0, start)]
while pq:
d, u = heapq.heappop(pq)
if d > dist.get(u, float("inf")):
continue # stale entry: a better path was found
# after this one was pushed. Skip it.
for v, w in neighbours(u):
nd = d + w
if nd < dist.get(v, float("inf")):
dist[v] = nd
heapq.heappush(pq, (nd, v)) # push a duplicate instead of
# decreasing the old entry's key
return dist
The heap can grow to O(E) entries rather than O(V), but each stale entry is discarded in O(1) and the total is still O(E log E) = O(E log V). This is the idiomatic Python approach and it is what you should write. See ./14-shortest-path-algorithms.md.
Heapsort
A heap gives a sorting algorithm almost for free: heapify the array, then repeatedly swap the root with the last element and shrink the heap by one.
def heapsort(a):
"""In-place, O(n log n) worst case, O(1) extra space, NOT stable."""
n = len(a)
def sift_down(i, size):
# Max-heap version: sink a[i] while it is smaller than its largest child
while True:
left = 2 * i + 1
if left >= size:
return
child = left
right = left + 1
if right < size and a[right] > a[left]:
child = right
if a[i] >= a[child]:
return
a[i], a[child] = a[child], a[i]
i = child
for i in range(n // 2 - 1, -1, -1): # build a MAX-heap: O(n)
sift_down(i, n)
for end in range(n - 1, 0, -1): # n-1 extractions: O(n log n)
a[0], a[end] = a[end], a[0] # largest goes to its final position
sift_down(0, end) # restore the heap over a[0:end]
return a
Heapsort’s O(n log n) worst case and O(1) space are strictly better guarantees than quicksort’s. It nonetheless loses to quicksort in practice because its access pattern (i → 2i+1) jumps around the array and destroys cache locality, while quicksort’s partition is two sequential scans. Its real production role is as the fallback in introsort (C++‘s std::sort): quicksort until the recursion gets suspiciously deep, then switch to heapsort to guarantee O(n log n). See ./08-sorting-algorithms.md.
The two-heaps pattern: running median
The problem: numbers arrive one at a time from a stream, and after each one you must report the median. Re-sorting is O(n log n) per element. Inserting into a sorted list is O(n) per element. The two-heaps pattern makes it O(log n) insert and O(1) median.
The idea: split the data at the median into two halves, and keep each half in the heap that puts its boundary at the top.
max-heap "low" min-heap "high"
(lower half of the data) (upper half of the data)
(8) <- largest of the low half (10) <- smallest of the high half
/ \ / \
(5) (7) (12) (11)
/ \ /
(2) (3) (15)
The two roots straddle the median. If the sizes are equal, the median is
their average; if `low` has one extra, the median is low's root.
The invariants to maintain after every insert:
- Order: every element in
low≤every element inhigh. - Balance:
len(low) - len(high)is 0 or 1.
import heapq
class MedianFinder:
"""Running median over a stream. add() is O(log n), median() is O(1)."""
def __init__(self):
self.low = [] # max-heap of the lower half, stored negated
self.high = [] # min-heap of the upper half
def add(self, num):
# Three lines, and they maintain both invariants unconditionally:
# 1. Always push into `low` first (so `low` is the default larger half).
heapq.heappush(self.low, -num)
# 2. Move low's largest across to `high`. This guarantees the ORDER
# invariant: whatever ends up on top of `low` is now <= high's root.
heapq.heappush(self.high, -heapq.heappop(self.low))
# 3. If that made `high` bigger, move its smallest back — restores BALANCE.
if len(self.high) > len(self.low):
heapq.heappush(self.low, -heapq.heappop(self.high))
def median(self):
if len(self.low) > len(self.high):
return float(-self.low[0]) # odd count: low's root
return (-self.low[0] + self.high[0]) / 2 # even count: average the roots
The three-line add is worth studying. The naive version — “compare against the roots, decide which heap to push into, then rebalance” — needs four or five branches and is easy to get subtly wrong on the empty-heap and equal-value edge cases. Pushing unconditionally into low, immediately shipping the top across to high, and then rebalancing back if needed handles every case with no conditionals except the final size check.
Where the pattern generalizes. Any time you need to maintain a split point over a dynamic multiset — the median, a percentile, “the k-th largest so far”, “is the current window balanced” — two heaps facing each other across the boundary is the tool. Variants:
- Sliding-window median: same structure plus lazy deletion (a
dictof counts of elements pending removal, purged from the heap tops as they surface). - IPO / task-scheduling problems: a max-heap of available items by profit and a min-heap of unavailable items by unlock time.
- Percentiles other than 50%: change the balance invariant from
|low| − |high| ∈ {0,1}to whatever ratio you need.
Finding the k-th largest element
A classic problem with three good answers and one bad one.
import heapq
import random
# ---- Approach 1: sort. O(n log n) time, O(n) or O(1) space. -----------------
def kth_largest_sort(nums, k):
"""Simplest and often fastest for small n — Timsort has excellent constants."""
return sorted(nums, reverse=True)[k - 1]
# ---- Approach 2: a size-k min-heap. O(n log k) time, O(k) space. ------------
def kth_largest_heap(nums, k):
"""
Keep only the k largest seen so far. The heap's root is the SMALLEST of
those k, which is exactly the k-th largest overall once the stream ends.
"""
heap = nums[:k]
heapq.heapify(heap) # O(k)
for x in nums[k:]:
if x > heap[0]: # O(1) reject — most elements land here
heapq.heapreplace(heap, x) # O(log k): pop the smallest, push x
return heap[0]
# Equivalent one-liner in the standard library (same O(n log k) algorithm):
# heapq.nlargest(k, nums)[-1]
# ---- Approach 3: quickselect. O(n) average, O(n^2) worst, O(1) space. -------
def partition(a, lo, hi):
"""Lomuto partition with a random pivot; returns the pivot's final index."""
r = random.randint(lo, hi)
a[r], a[hi] = a[hi], a[r] # randomization defeats adversarial input
pivot = a[hi]
i = lo
for j in range(lo, hi):
if a[j] < pivot:
a[i], a[j] = a[j], a[i]
i += 1
a[i], a[hi] = a[hi], a[i]
return i
def quickselect(a, k):
"""k-th SMALLEST (0-indexed). Destroys the order of `a`."""
lo, hi = 0, len(a) - 1
while True:
if lo == hi:
return a[lo]
p = partition(a, lo, hi)
if k == p:
return a[k]
if k < p:
hi = p - 1 # recurse into ONE side only — this is why it's O(n)
else:
lo = p + 1
def kth_largest_quickselect(nums, k):
"""k-th largest = (n - k)-th smallest, 0-indexed."""
return quickselect(list(nums), len(nums) - k)
The comparison:
| Approach | Time | Space | Streaming? | When to use |
|---|---|---|---|---|
| Full sort | O(n log n) | O(n) | No | Small n, or you need the other order statistics too |
Size-k min-heap | O(n log k) | O(k) | Yes | k ≪ n, or the data is a stream / does not fit in memory |
| Quickselect | O(n) avg, O(n²) worst | O(1) | No | The whole array is in memory and you want the fastest single answer |
| Max-heap + k pops | O(n + k log n) | O(n) | No | You need the top k in order and k is small |
Two things worth noting:
Why quickselect is O(n). Quicksort recurses into both halves, giving T(n) = 2T(n/2) + O(n) = O(n log n). Quickselect recurses into only the half that can contain the answer, giving T(n) = T(n/2) + O(n), and the geometric series n + n/2 + n/4 + … = 2n sums to O(n). The random pivot is what makes the O(n²) worst case (always picking the extreme) astronomically unlikely rather than triggerable by sorted input. See ./20-brute-force-greedy-and-randomised-algorithms.md and ./19-recursion-and-divide-and-conquer.md.
Why the heap approach wins in practice more often than the table suggests. Quickselect is asymptotically better but needs the entire array in memory and mutates it. The size-k heap works on a stream of unknown length, uses O(k) memory regardless of n, and — because of the if x > heap[0] guard — performs a cheap O(1) rejection for the vast majority of elements. For “top 10 trending items out of 100 million events”, the heap is not just easier, it is the only one of the two that runs at all.
k-way merge
The other everyday application of a heap: merging k already-sorted sequences into one sorted output. Take the smallest head across all k sequences, emit it, advance that sequence, repeat. A heap of size k makes “smallest head” an O(log k) question.
import heapq
def k_way_merge(sorted_lists):
"""
O(N log k) where N is the total element count. Compare with concatenating
everything and sorting: O(N log N). When k is small and N is huge, the
difference is real — and unlike sorting, this is streaming.
"""
heap = []
for list_index, lst in enumerate(sorted_lists):
if lst:
# (value, which list, position in that list)
heapq.heappush(heap, (lst[0], list_index, 0))
out = []
while heap:
value, list_index, pos = heapq.heappop(heap)
out.append(value)
next_pos = pos + 1
if next_pos < len(sorted_lists[list_index]):
heapq.heappush(heap, (sorted_lists[list_index][next_pos], list_index, next_pos))
return out
# In production, use the standard library — it is lazy, so it never
# materializes the full output and works on files, generators, and DB cursors:
# heapq.merge(*sorted_iterables, key=None, reverse=False)
This is the merge step of external sorting (sort chunks that fit in RAM, write them out, then k-way merge the files), the compaction step of LSM-tree storage engines like RocksDB and Cassandra, and the way a distributed query engine combines sorted partial results from k shards. See ../../data-engineer/en/10-big-data-and-distributed-computing.md.
Best Practices
- Use
heapq, not a hand-rolled heap. It is C-implemented, correct, and uses the hole-shifting optimization. Write your own once to understand sift-up/sift-down, then never again. heapifywhen you have all the data;heappushin a loop only when you do not.O(n)versusO(n log n)for the identical result.- Always add a tie-breaker to tuple entries.
(priority, itertools.count(), payload). Without it, equal priorities make Python compare the payloads, which raisesTypeErrorfor anything without a defined ordering — and it will not fail until the first priority collision in production. - Negate for a max-heap, but only with numbers. For non-numeric keys, negate just the numeric priority in a tuple, or wrap the item in a class with
__lt__reversed. - Do not use a heap when you need to search or delete arbitrary elements. That is
O(n)in a heap. Use a balanced BST, aSortedList, or a heap paired with an index dictionary. - Use lazy deletion for
decrease_key. Push the improved entry and skip stale ones on pop (if d > dist[u]: continue). This is the standard Python Dijkstra idiom and it is asymptotically fine. - For “top k”, bound the heap at size k.
heapq.nlargest(k, iterable)or a manual size-kmin-heap givesO(n log k)time andO(k)memory — critical whenndoes not fit in RAM. - Prefer
heappushpopoverheappush+heappop. One sift-down instead of two heap operations, and the heap never grows past its target size. - Reach for two heaps whenever you must maintain a dynamic split point. Running median, percentiles, “cheapest available vs. soonest unlocked” — the boundary-facing pair is the pattern.
- Do not sort just to find one extreme.
min(a)isO(n);sorted(a)[0]isO(n log n). Similarly, do not sort just to find the k-th element — use quickselect or a bounded heap. - Remember heapsort’s real job. As a standalone sort it loses to quicksort and Timsort on cache behaviour, but its
O(n log n)worst case andO(1)space make it the right safety net inside introsort. heapqis not thread-safe. Guard it with a lock, or usequeue.PriorityQueue, which wrapsheapqwith the locking already done.
References
- roadmap.sh — Data Structures & Algorithms
- Heap (data structure) — Wikipedia
- Binary heap — Wikipedia
- Priority queue — Wikipedia
- Heapsort — Wikipedia
- Quickselect — Wikipedia
- Fibonacci heap — Wikipedia
- Python Documentation —
heapq: Heap queue algorithm - Python Documentation —
queue.PriorityQueue - CLRS — Introduction to Algorithms, Chapter 6: Heapsort
- MIT 6.006 — Introduction to Algorithms (OpenCourseWare)
- VisuAlgo — Binary Heap visualization
- Big-O Cheat Sheet