← Cấu trúc dữ liệu & Giải thuật← Data Structures & Algorithms
Cấu trúc dữ liệu & Giải thuậtData Structures & Algorithms7 Th8, 2026Aug 7, 202631 phút đọc27 min read

Heap & Priority QueueHeaps & Priority Queues

Mục lục
  1. Tổng quan
  2. Kiến thức nền tảng
  3. Tính chất heap
  4. Mảng ngầm định — một cái cây không có pointer
  5. Sift-up và sift-down
  6. Vì sao heapify là O(n) chứ không phải O(n log n)
  7. Bảng tổng hợp độ phức tạp
  8. Khái niệm chính
  9. heapq của Python
  10. Max-heap trong Python: mẹo đổi dấu
  11. ADT priority queue
  12. Heapsort
  13. Pattern two-heaps: running median
  14. Tìm phần tử lớn thứ k
  15. Gộp k-way
  16. Best Practices
  17. Tài liệu tham khảo
Table of contents
  1. Overview
  2. Fundamentals
  3. The heap property
  4. The implicit array — a tree with no pointers
  5. Sift-up and sift-down
  6. Why heapify is O(n), not O(n log n)
  7. Complexity summary
  8. Key Concepts
  9. Python’s heapq
  10. Max-heaps in Python: the negation trick
  11. The priority queue ADT
  12. Heapsort
  13. The two-heaps pattern: running median
  14. Finding the k-th largest element
  15. k-way merge
  16. Best Practices
  17. 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:

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:

Đó 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 i2i + 1
Right child của i2i + 2
Parent của i(i − 1) // 2
i có phải leaf?2i + 1 >= n
Chỉ số leaf đầu tiênn // 2
Chỉ số non-leaf cuối cùngn // 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:

Đ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_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ố nodeKhoảng sift-down tối đaTổng công
0 (leaf)~n/200
1~n/41n/4
2~n/822n/8
3~n/1633n/16
h (root)1hh

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)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)O(n); một vòng lặp heapq.heappushO(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ácBinary heapMảng đã sắp xếpMảng chưa sắp xếpBalanced BST
peek min/maxO(1)O(1)O(n)O(log n)
pushO(log n)O(n)O(1) amortizedO(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úcO(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 heappushpopheapreplace 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:

  1. Chỉ hoạt động với số. -x khô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.
  2. 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 -x luôn an toàn; trong C/Java, -Integer.MIN_VALUE trà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ĩaChi 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ấtO(log n)
peek()Xem phần tử ưu tiên cao nhấtO(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 queueO(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ể heapinsertextract-mindecrease-keymergeTrong thực tế
Binary heapO(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 heapO(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 heapO(log n)O(log n)O(log n)O(log n)Khi gộp queue diễn ra thường xuyên
Fibonacci heapO(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 heapO(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 caseO(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:

  1. Thứ tự: mọi phần tử trong low đều mọi phần tử trong high.
  2. 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ể:

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áchThời gianBộ nhớStreaming?Khi nào dùng
Sắp xếp toàn bộO(n log n)O(n)Khôngn nhỏ, hoặc bạn cần luôn các thống kê thứ tự khác
Min-heap kích thước kO(n log k)O(k)k ≪ n, hoặc dữ liệu là stream / không vừa bộ nhớ
QuickselectO(n) TB, O(n²) worstO(1)KhôngCả mảng nằm trong bộ nhớ và bạn muốn một đáp án nhanh nhất
Max-heap + k lần popO(n + k log n)O(n)KhôngBạn cần top k theo thứ tự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./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

Tài liệu tham khảo

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:

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:

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:

RelationshipFormula
Left child of i2i + 1
Right child of i2i + 2
Parent of i(i − 1) // 2
Is i a leaf?2i + 1 >= n
First leaf indexn // 2
Last non-leaf indexn // 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:

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 nodesMax sift-down distanceTotal work
0 (leaves)~n/200
1~n/41n/4
2~n/822n/8
3~n/1633n/16
h (root)1hh

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

OperationBinary heapSorted arrayUnsorted arrayBalanced BST
peek min/maxO(1)O(1)O(n)O(log n)
pushO(log n)O(n)O(1) amortizedO(log n)
pop extremeO(log n)O(1) (from the end)O(n)O(log n)
Build from n itemsO(n)O(n log n)O(n)O(n log n)
Search arbitrary keyO(n)O(log n)O(n)O(log n)
Delete arbitrary keyO(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 iterationO(n log n) (destructive)O(n)O(n log n)O(n)
Space overheadnonenonenone2 pointers/node
Merge two structuresO(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:

  1. It only works for numbers. -x is 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.
  2. Watch the asymmetry of the integer range in fixed-width languages. In Python, integers are arbitrary precision, so -x is always safe; in C/Java, -Integer.MIN_VALUE overflows 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:

OperationMeaningBinary heap cost
insert(item, priority)Add an elementO(log n)
extract_min() / extract_max()Remove and return the highest-priority itemO(log n)
peek()Look at the highest-priority itemO(1)
decrease_key(item, p)Raise an item’s priorityO(log n) if you know its index
merge(other)Combine two queuesO(n)

The last two are where binary heaps show their limits, and where the exotic heap variants earn their keep:

Heap variantinsertextract-mindecrease-keymergeIn practice
Binary heapO(log n)O(log n)O(log n)O(n)The default. Best constants, best cache behaviour
d-ary heapO(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 heapO(log n)O(log n)O(log n)O(log n)When merging queues is frequent
Fibonacci heapO(1) am.O(log n) am.O(1) am.O(1)Theoretically optimal for Dijkstra; constants so bad it usually loses
Pairing heapO(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:

  1. Order: every element in low every element in high.
  2. 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:

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:

ApproachTimeSpaceStreaming?When to use
Full sortO(n log n)O(n)NoSmall n, or you need the other order statistics too
Size-k min-heapO(n log k)O(k)Yesk ≪ n, or the data is a stream / does not fit in memory
QuickselectO(n) avg, O(n²) worstO(1)NoThe whole array is in memory and you want the fastest single answer
Max-heap + k popsO(n + k log n)O(n)NoYou 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

References