Структуры данных и двоичная куча

В первой лекции мы оценивали отдельный алгоритм. Теперь рассмотрим задачу, в которой действия повторяются: в симулятор поступают события, и каждый раз нужно обработать ближайшее по времени. Одного поиска минимума недостаточно — после удаления события и добавления новых данных его приходится искать снова.

Можно хранить все события в массиве. Но прежде чем выбирать представление, выпишем, что именно программа должна уметь делать и сколько раз потребуется каждое действие. Так мы переходим от алгоритма над готовым входом к структуре данных, которая поддерживает целую последовательность операций.

Структура данных (Data structure) — Представление данных вместе с набором поддерживаемых операций и оценками их стоимости.

Для нашего примера нужны три операции: добавить событие, посмотреть ближайшее и извлечь его. Их объединяет абстракция очереди с приоритетами: первым обрабатывается элемент с наименьшим приоритетом, независимо от времени его добавления.

Очередь с приоритетами (Priority queue) — Абстрактная структура, которая хранит элементы с приоритетами и позволяет добавлять элемент и извлекать наименьший (или наибольший).

Будем обозначать операции как insert(x), getMin() и removeMin(). Повторы разрешены: удаление убирает одно вхождение минимума. Если одинаковые приоритеты должны обслуживаться в порядке прихода, можно хранить пары «приоритет, номер добавления».

В примере ниже используем готовую очередь из стандартной библиотеки. Пока важно увидеть её контракт; в следующих разделах самостоятельно построим двоичную кучу, которая этот контракт реализует.

import heapq

def ordered(values: list[int]) -> list[int]:
    heap = values.copy()
    heapq.heapify(heap)
    result = []
    while heap:
        result.append(heapq.heappop(heap))
    return result

print(ordered([5, 1, 4, 2]))

use std::cmp::Reverse;
use std::collections::BinaryHeap;

fn ordered(values: &[i64]) -> Vec<i64> {
    let mut heap = BinaryHeap::from(
        values.iter().copied().map(Reverse).collect::<Vec<_>>()
    );
    let mut result = Vec::new();
    while let Some(Reverse(value)) = heap.pop() {
        result.push(value);
    }
    result
}

fn main() {
    println!("{:?}", ordered(&[5, 1, 4, 2]));
}

[1, 2, 4, 5]

Итак, мы определили поведение структуры, но ещё не выбрали способ хранения. Начнём с двух вариантов на массиве: их ограничения покажут, какую работу должна ускорить куча.

Добавим 7, затем 2. Без порядка получим [7,2]; по убыванию — тоже [7,2]. А вставка 5 уже различит варианты: [7,2,5] против [7,5,2]. Предскажите, где будет легче извлечь минимум.

Первый вариант — неупорядоченный массив. Новый элемент дописываем в конец за O(1), если свободная ёмкость уже есть. Зато минимум приходится искать среди всех n элементов: порядок хранения ничего о нём не сообщает.

После поиска удалять элемент со сдвигом всего хвоста необязательно. Порядок нам не нужен, поэтому заменим найденный минимум последним элементом и сократим используемую часть массива. Поиск остаётся линейным, само удаление после поиска — постоянным.

i = argmin(A[0..n−1])
swap(A[i], A[n−1])
n = n − 1

Второй вариант — поддерживать массив отсортированным по убыванию. Тогда минимум всегда в конце, и его чтение и удаление стоят O(1). Цена переносится во вставку: чтобы сохранить порядок, иногда приходится сдвинуть почти весь массив.

Сохранённый отдельно минимум ускоряет только чтение: после его удаления новый минимум придётся искать заново. У отсортированного по убыванию массива дорогой становится вставка, зато извлечение с конца не требует поиска. Сравните на одной последовательности два представления ниже.

Чтобы выбрать вариант, оцените всю нагрузку. Если сначала идут n добавлений, а затем одно извлечение, неупорядоченного массива достаточно. Если вставки и извлечения чередуются примерно поровну, линейная операция выполняется снова и снова. В модели ниже можно сравнить именно такие последовательности.

Мы увидели компромисс двух массивов. В основном маршруте перейдём к коротким путям в двоичном дереве. Разбиение на блоки можно исследовать отдельно: оно не требуется для понимания кучи.

Для иерархии возьмём двоичное дерево: у каждого узла не больше двух детей. Важно ограничить его высоту, иначе цепочка из n узлов снова приведёт к линейной работе. Поэтому заранее зафиксируем форму дерева.

Сначала разместим 6 узлов: один в корне, два на следующем уровне, затем три на третьем слева направо. Массив [0,1,2,3,4,5] хранит эти позиции подряд. Если оставить дыру, следующая ячейка уже не соответствовала бы следующему месту дерева.

Полное бинарное дерево (Complete binary tree) — Все уровни, кроме последнего, заполнены; последний заполняется слева направо.

Если заполнять уровни слева направо без пропусков, на глубине h помещается до 2ʰ узлов, а высота непустого дерева равна ⌊log₂ n⌋. Новое место всегда известно: это следующая свободная позиция на последнем уровне.

Расположим узлы в массиве по уровням. Тогда рёбра не нужно хранить явно: дети и родитель вычисляются по индексу. Например, у узла 1 дети находятся на позициях 3 и 4, а его родитель — на позиции 0.

Модель показывает два вида одних данных: ячейки массива и узлы дерева. Пока проследите, как позиции соответствуют друг другу; правило размещения значений разберём в следующем разделе.

Форма дерева дала короткие пути и удобное хранение. Теперь зададим порядок значений на рёбрах, чтобы минимум всегда находился в заранее известном месте.

Полностью сортировать массив было дорого из-за требования упорядочить каждую пару соседей. Для быстрого доступа к минимуму достаточно более слабого условия: родитель должен быть не больше своих детей.

Инвариант min-кучи (Heap property) — Для каждого узла значение его родителя не превосходит значения любого из его детей.

Возьмём любой узел и пройдём от него к корню. При каждом шаге вверх значение не увеличивается, значит значение корня не больше выбранного элемента. Так локальное правило для каждого ребра гарантирует глобальный минимум в корне, и getMin() становится O(1).

Например, массив [1, 4, 2, 8, 5, 3] — корректная min-куча: каждый родитель не больше детей. При этом 4 стоит раньше 2, а 5 раньше 3. Порядок между разными ветвями не нужен; именно эта свобода позволит быстро менять структуру.

Куча поддерживает два условия: форму полного дерева и порядок на каждом ребре. При вставке и удалении будем сначала сохранять форму, а затем восстанавливать только те рёбра, где мог нарушиться порядок.

Чтобы вставить x и сохранить форму дерева, допишем его в конец массива. Все старые рёбра остаются прежними. Нарушиться может только новое ребро: x иногда меньше своего родителя.

Если это произошло, поменяем x с родителем. Теперь x поднялся на один уровень, и проверять нужно только его нового родителя. Повторяем, пока x не окажется в корне или пока родитель не станет не больше него. Это просеивание вверх.

Вставьте 2 в [3,5,4,9,8] и сначала предскажите число обменов. Дописывание даёт [3,5,4,9,8,2]; индекс 5 имеет родителя 2 со значением 4. После первого обмена [3,5,2,9,8,4]. Родитель индекса 2 — корень со значением 3; второй обмен даёт [2,5,3,9,8,4]. Теперь i = 0 и подъём заканчивается.

A.push(x)
i = length(A) − 1
while i > 0 and A[i] < A[(i−1)//2]:
    swap(A[i], A[(i−1)//2])
    i = (i−1)//2

Например, вставим 2 в кучу [3, 5, 4, 9, 8]. Элемент попадёт на позицию 5, обменяется с 4, затем с 3. Получится [2, 5, 3, 9, 8, 4]. Мы исправили один путь к корню, а не пересортировали все шесть элементов.

В модели добавьте значение меньше текущего минимума и проследите этот путь. Затем попробуйте большое значение: оно останется в конце. Оценка O(log n) ограничивает худший случай, хотя конкретная вставка может завершиться сразу.

Код ниже применяет тот же механизм к потоку данных: хранит только k наибольших значений. В min-куче из k кандидатов корень — самый слабый из выбранных, поэтому по нему удобно решать, достоин ли новый элемент попасть в набор.

import heapq

def kth_largest(values: list[int], k: int) -> int | None:
    if k <= 0:
        return None
    heap = []
    for value in values:
        heapq.heappush(heap, value)
        if len(heap) > k:
            heapq.heappop(heap)
    return heap[0] if len(heap) == k else None

print(kth_largest([3, 2, 1, 5, 6, 4], 2))

use std::cmp::Reverse;
use std::collections::BinaryHeap;

fn kth_largest(values: &[i64], k: usize) -> Option<i64> {
    if k == 0 { return None; }
    let mut heap = BinaryHeap::new();
    for &value in values {
        heap.push(Reverse(value));
        if heap.len() > k { heap.pop(); }
    }
    if heap.len() == k { heap.peek().map(|x| x.0) } else { None }
}

fn main() {
    println!("{}", kth_largest(&[3, 2, 1, 5, 6, 4], 2).unwrap());
}

5

При вставке потенциальная ошибка двигалась от листа к корню. Для удаления минимума потребуется обратное направление: место освобождается в корне, и исправлять порядок будем сверху вниз.

def sift_up(a):
    # Последний элемент уже добавлен; восстановите один путь к корню.
    pass

Минимум находится в корне, но просто вырезать первую ячейку нельзя: сдвиг изменит все связи между родителями и детьми. Сохраним ответ, перенесём в корень последний элемент и уменьшим размер массива. Форма полного дерева сохранится.

У нового корня могут оказаться меньшие дети. Выберем меньшего из существующих детей и обменяемся с ним, если он меньше текущего элемента. Меньший ребёнок займёт верхнюю позицию корректно, а возможное нарушение спустится на один уровень. Повторяем до восстановления порядка.

Четыре отдельных действия: 1) сохранить A[0] как ответ; 2) снять последний элемент и, если куча не пуста, перенести его в корень; 3) выбрать меньшего из существующих детей; 4) обменять только при строгом нарушении и повторить ниже. При равных детях можно выбрать любого; модель выбирает левого. При равенстве родителя и меньшего ребёнка останавливаемся.

if length(A) == 0: error("empty")
if length(A) == 1: return A.pop()
answer = A[0]
A[0] = A.pop()
i = 0
while 2*i+1 < length(A):
    j = 2*i+1
    if j+1 < length(A) and A[j+1] < A[j]: j += 1
    if A[i] <= A[j]: break
    swap(A[i], A[j]); i = j
return answer

Для кучи из одного элемента после удаления ничего просеивать не нужно. Для остальных случаев каждый обмен опускается ровно на уровень, поэтому извлечение, как и вставка, занимает не больше O(log n). Чтение минимума обходится без изменений структуры.

Теперь контракт очереди с приоритетами реализован полностью. Проверим, что получится, если применить извлечение минимума ко всем элементам подряд: так структура данных превратится в алгоритм сортировки.

Положим все n элементов в min-кучу и будем извлекать минимум, пока куча не опустеет. Каждый следующий ответ не меньше предыдущего, а каждый исходный элемент извлекается ровно один раз. Значит, полученная последовательность отсортирована.

Для min-кучи [1,3,2] последовательные removeMin() возвращают 1, затем 2, затем 3; выходной массив растёт [] → [1] → [1,2] → [1,2,3], а куча уменьшается. Вариант ниже с max-кучей использует освободившийся конец исходного массива вместо отдельного выхода.

n вставок и n удалений дают O(n log n) времени. В таком варианте куча и результат используют дополнительную память порядка n. Чтобы отсортировать в исходном массиве, слегка изменим организацию алгоритма: будем хранить max-кучу.

Максимум меняем с последним элементом текущей кучи. В конце массива постепенно растёт отсортированный суффикс, а в оставшемся префиксе после каждого обмена восстанавливается max-куча. Суффикс больше не трогаем.

buildMaxHeap(A)
for end = n−1 down to 1:
    swap(A[0], A[end])
    siftDownMax(A, root=0, size=end)

Разделение на префикс-кучу и готовый суффикс — инвариант сортировки. Просеивание можно написать циклом, поэтому этому варианту достаточно O(1) дополнительной памяти. Однако обычный heapsort неустойчив: дальние обмены могут поменять порядок равных ключей.

В оценке мы пока разрешили O(n log n) на начальное построение. Когда весь массив уже дан, вставлять элементы по одному необязательно. Разберём построение снизу вверх и получим более точную оценку этой фазы.

Один и тот же готовый массив можно собрать n вставками сверху, до O(n log n), или исправлять поддеревья снизу. Во втором варианте листья уже готовы; их примерно половина, и просеивать их вообще не нужно.

Операция heapify превращает готовый массив в кучу. Все листья уже являются корректными кучами из одного элемента. Начнём с последнего внутреннего узла и будем двигаться к корню, просеивая каждый узел вниз.

К моменту обработки узла оба его поддерева уже удовлетворяют свойству кучи. Поэтому обычного просеивания достаточно, чтобы получить корректную кучу и во всём этом поддереве. Порядок обхода здесь является частью доказательства.

for i = floor(n/2)−1 down to 0:
    siftDown(A, i, n)

Грубая оценка «n узлов по log n» завышает работу: большинство узлов находится почти у листьев и далеко спуститься не может. Примерно n/2 листьев не требуют просеивания, около n/4 узлов спускаются не более чем на уровень, около n/8 — не более чем на два. Суммируем стоимость по высотам, а не умножаем все узлы на высоту корня.

В реализации ниже отдельно показаны построение и извлечение. Обратите внимание: обе операции используют одно и то же просеивание вниз, но запускают его для разных узлов и с разной текущей границей кучи.

Сравнивать структуры удобнее в два шага: сначала понять, какое правило они сохраняют, затем посмотреть стоимость действий. Инвариант — правило, которое остаётся верным после каждой операции. Например, в min-куче родитель не больше своих детей: поэтому минимум всегда находится в корне, хотя весь массив не отсортирован.

Теперь уточним действия. Добавить x — принять новый элемент и сохранить инвариант; нужное место выбирает сама структура. Вставить в позицию i — поместить элемент в указанную ячейку, сдвинув следующие вправо. В отсортированном массиве, блоках и куче произвольная позиция может нарушить инвариант, поэтому такой операции в их интерфейсе нет.

Удалить — убрать элемент из уже известной позиции, без затрат на его поиск; для стека удаляем вершину, для очереди — голову. Найти x — найти позицию заданного значения. Мин и Макс — прочитать наименьшее и наибольшее значения, не удаляя их. Так мы не смешиваем чтение минимума с его извлечением.

Здесь n — число элементов, b — размер блока, k = ⌈n/b⌉ — число блоков. Для поиска значения в блоках выполняем двоичный поиск в каждом блоке. Прочерк означает, что операция не предусмотрена этим интерфейсом. Знак † обозначает дополнительный проход по содержимому: поиск, минимум и максимум не входят в основные операции стека и FIFO-очереди.

Оценки даны для худшего случая; звёздочкой отмечена амортизированная стоимость с учётом расширения динамического массива. Стек хранится в массиве, FIFO-очередь — в кольцевом буфере, все варианты используют O(n) памяти. Размер b выбираем заранее под ожидаемый масштаб данных; его изменение требует пересборки. Чтобы удалить заданное значение, сначала прибавьте стоимость его поиска. Например, удаление произвольного x из кучи занимает O(n), а извлечение известного минимума из корня — O(log n).

def sift_down(a: list[int], i: int) -> None:
    while 2 * i + 1 < len(a):
        j = 2 * i + 1
        if j + 1 < len(a) and a[j + 1] < a[j]:
            j += 1
        if a[i] <= a[j]:
            break
        a[i], a[j] = a[j], a[i]
        i = j

def heapify(a: list[int]) -> None:
    for i in range(len(a) // 2 - 1, -1, -1):
        sift_down(a, i)

def remove_min(a: list[int]) -> int | None:
    if not a:
        return None
    result = a[0]
    last = a.pop()
    if a:
        a[0] = last
        sift_down(a, 0)
    return result

data = [7, 2, 9, 4, 1, 8]
heapify(data)
result = []
while data:
    result.append(remove_min(data))
print(result)

fn sift_down(a: &mut [i64], mut i: usize) {
    while 2 * i + 1 < a.len() {
        let mut j = 2 * i + 1;
        if j + 1 < a.len() && a[j + 1] < a[j] { j += 1; }
        if a[i] <= a[j] { break; }
        a.swap(i, j);
        i = j;
    }
}

fn heapify(a: &mut [i64]) {
    for i in (0..a.len() / 2).rev() { sift_down(a, i); }
}

fn remove_min(a: &mut Vec<i64>) -> Option<i64> {
    let result = *a.first()?;
    let last = a.pop()?;
    if !a.is_empty() {
        a[0] = last;
        sift_down(a, 0);
    }
    Some(result)
}

fn main() {
    let mut data = vec![7, 2, 9, 4, 1, 8];
    heapify(&mut data);
    let mut result = Vec::new();
    while let Some(value) = remove_min(&mut data) { result.push(value); }
    println!("{:?}", result);
}

[1, 2, 4, 7, 8, 9]

Теперь вы умеете проверять min-кучу по рёбрам, выполнять вставку и извлечение, выбирать кучу под последовательность запросов. Для одного нужного ранга иногда можно обойтись без хранения всей очереди: эту идею продолжит QuickSelect в лекции 03.

def is_min_heap(a):
	# Напишите решение здесь
	pass

Справочник собирает определения из разобранных разделов. Используйте поиск, если нужно восстановить термин; затем переходите к оценке ограничений и задачам для самостоятельного решения.

Общий справочник всех лекций ↗

Найдено: 14 из 14

Мы сравнили представления по стоимости отдельных операций. Для задачи теперь сложим эти стоимости: параметры здесь — не только число элементов n, но и число операций q, размер ответа k и максимальная текущая длина очереди. Выпишите последовательность запросов: стоимость одной структуры может отличаться на порядки для разных нагрузок.

Подставляем максимумы в суммарную стоимость

• Для q = 200 000 извлечений со сканированием массива длиной порядка q возможны около q²/2 ≈ 2 · 10¹⁰ просмотров. Куча уменьшает оценку до O(q log q).
• Если нужно k = 100 лучших из n = 10⁶ результатов, куча размера k требует O(n log(k + 1)) времени и O(k) памяти; не храните миллион элементов без необходимости.
• Если весь стартовый набор уже дан, постройте кучу за O(n). Последовательная вставка всех значений допускает O(n log n); последовательное извлечение всех элементов после heapify также занимает O(n log n).

Память, идентификаторы и повторы

Массив из 200 000 значений i64 занимает 1,6 МБ полезных данных; пары с идентификаторами и Python-объекты требуют больше. Если нужны удаление произвольной записи или изменение ключа по id, одной стандартной кучи недостаточно: потребуется индекс или стратегия отложенного удаления с учётом её памяти.

Сумма qᵢ log(nᵢ + 1) по тестам важнее стоимости одного максимального теста. Проверьте пустую очередь, один элемент, много одинаковых приоритетов, чередующиеся вставки и удаления. При вычисляемом приоритете, например x² + y², выбирайте тип, вмещающий промежуточный результат.

def is_min_heap(a):
	# Напишите решение здесь
	pass

def sift_down(a, i):
    # Выберите меньшего существующего ребёнка и исправьте один путь.
    pass

def sift_up(a):
    # Последний элемент уже добавлен; восстановите один путь к корню.
    pass

def top_k(a, k):
	# Напишите решение здесь
	pass

Примените разобранные приёмы самостоятельно. Для каждой задачи сначала сформулируйте требования к ответу и предложите простое решение. Затем подставьте максимальные ограничения, оцените время и память и решите, какое улучшение требуется. Подсказки с идеей и оценкой раскрываются отдельно.

Уровни задают порядок практики в курсе, а не единую шкалу платформ. Если уровень отличается от отметки LeetCode, это указано в карточке. Для Codeforces используется оценка курса.