External hash join: спилл обеих сторон и обработка по партициям

Hash join — основной алгоритм соединения таблиц в DuckDB. Он быстр, но требователен к памяти: классический hash join держит в RAM целую hash-таблицу по одной из сторон соединения. Если эта сторона велика, hash-таблица не помещается в память. External hash join — механизм, позволяющий соединению отработать в этом случае, спиллив части обеих таблиц на диск.

Этот урок разбирает hash join изнутри: фазы build и probe, что именно упирается в память, и как партиционирование обеих сторон превращает большое соединение в серию мелких. Логика родственна внешней агрегации из прошлого урока, но с важным отличием — здесь участвуют две таблицы, и спиллить нужно обе согласованно.

Hash join в памяти: build и probe

Сначала — как hash join работает, когда памяти хватает. Соединение A JOIN B ON A.k = B.k выполняется в две фазы.

Фаза build. Движок выбирает одну сторону — обычно меньшую, назовём её build-сторона — и строит по ней hash-таблицу: ключ соединения отображается на строки с этим ключом. Build-сторона целиком прочитывается и материализуется в hash-таблице в памяти.

Фаза probe. Движок сканирует вторую сторону — probe-сторону. Для каждой её строки вычисляется хеш ключа, и по hash-таблице ищутся совпадающие строки build-стороны. Найденные пары — это результат join.

Память hash join определяется размером build-стороны: вся она лежит в hash-таблице. Probe-сторона память почти не потребляет — она течёт потоком. Поэтому DuckDB старается сделать build-стороной меньшую таблицу (это решает оптимизатор по статистике кардинальности). Но даже меньшая таблица бывает слишком большой для RAM — тогда нужен external hash join.

Что ломается при нехватке памяти

Проблема та же, что у агрегации: hash-таблица build-стороны должна быть в RAM целиком, потому что probe-строка с любым ключом может совпасть с любой build-строкой. Нельзя «обработать кусок build-стороны и забыть» — probe-строки, которым нужен этот кусок, ещё не пришли.

Если build-сторона не влезает в memory_limit, наивный hash join падает с OOM. Решение — снова партиционирование, но теперь синхронное по обеим таблицам.

Партиционирование обеих сторон

Ключевая идея external hash join: разбить обе таблицы на партиции по хешу ключа соединения, согласованно, и соединять партицию с партицией.

Наблюдение, которое делает это корректным: чтобы две строки соединились, у них должен быть равный ключ A.k = B.k. Равные ключи дают равный хеш. Значит, если применить одну и ту же хеш-функцию и одно и то же разбиение к обеим таблицам, то строка из A и совпадающая с ней строка из B попадут в партиции с одинаковым номером. Строка из партиции 3 таблицы A может совпасть только со строками партиции 3 таблицы B — в других партициях её пар нет.

Это превращает одно большое соединение в N независимых мелких: партиция A[0] join B[0], партиция A[1] join B[1], и так далее. Каждую пару партиций можно обработать отдельно, и build-сторона одной пары — это всего лишь 1/N от полной build-стороны.

Почему спиллить нужно обе стороны

Это главное отличие external join от external aggregation. В агрегации был один поток данных — строки таблицы; партиционировалась и спиллилась одна структура. В join данных два потока — две таблицы, и партиционировать нужно обе, согласованно.

Build-сторону спиллить очевидно нужно — это она не влезает в hash-таблицу. Но и probe-сторону тоже: probe-строки партиции i нужны ровно тогда, когда обрабатывается пара партиций i. Пока DuckDB обрабатывает пару 0, probe-строки партиции 5 нужно где-то держать — и если их много, они отправляются в spill-файл, чтобы быть поднятыми, когда дойдёт очередь до пары 5.

Так обе таблицы оказываются разложены по партициям, частично на диске. Затем DuckDB обрабатывает пары партиций по очереди: поднимает с диска A[i] и B[i], строит мелкую hash-таблицу по build-партиции, прогоняет probe-партицию, выдаёт результат, освобождает память, переходит к паре i+1.

Видим внешний join в работе

Воспроизведём соединение, где build-сторона заведомо не влезает в маленький лимит.

-- две большие таблицы со связью по ключу
CREATE TABLE customers AS
SELECT range AS id, ('name_' || range) AS name, random() AS score
FROM range(8000000);

CREATE TABLE orders AS
SELECT range AS order_id,
       range % 8000000 AS customer_id,
       random() * 1000 AS amount
FROM range(40000000);

SET memory_limit = '500MB';
SET temp_directory = '/tmp/join-spill';

-- соединение 40 млн orders с 8 млн customers при лимите 500 МБ
SELECT c.name, count(*) AS order_count, sum(o.amount) AS total
FROM orders o
JOIN customers c ON o.customer_id = c.id
GROUP BY c.name
LIMIT 5;
-- запрос ОТРАБОТАЕТ: hash-таблица по customers не влезла в 500 МБ,
-- но обе таблицы партиционировались и спиллились

SET memory_limit = '500MB';
EXPLAIN ANALYZE
SELECT count(*) FROM orders o JOIN customers c ON o.customer_id = c.id;

В выводе оператор HASH_JOIN будет помечен как работающий в external-режиме. Во время выполнения в /tmp/join-spill появятся spill-файлы партиций обеих таблиц, исчезающие по завершении. Подними memory_limit так, чтобы hash-таблица по customers поместилась в RAM, повтори — метка external пропадёт.

Практические следствия

Build-сторона определяет память. Соединение «огромная факт-таблица join небольшой справочник» дёшево по памяти даже при гигантской факт-таблице: build-сторона — маленький справочник, probe-сторона течёт потоком. А вот «большая join большая» по большому ключу — кандидат на спилл. Понимание, какая сторона build, объясняет память запроса.

Фильтруй до join. Чем меньше build-сторона, тем меньше hash-таблица и тем вероятнее, что спилл не понадобится. Filter pushdown, урезающий build-сторону до join, напрямую снижает риск out-of-core. Это конкретная причина, почему фильтры из модуля про оптимизатор важны для памяти.

External join медленнее in-memory, но завершается. Как и вся out-of-core машинерия DuckDB: спилл стоит времени (запись и чтение партиций с диска), но запрос отрабатывает, а не падает. Цена — время; выгода — соединение любого размера выполнимо.

Попробуй сам

1. Создай dim (5 млн строк: id, label) и fact (30 млн строк: id, dim_id со ссылкой на dim, value).
2. Задай SET temp_directory = '/tmp/join-demo'; и SET memory_limit = '450MB'; Выполни SELECT d.label, sum(f.value) FROM fact f JOIN dim d ON f.dim_id = d.id GROUP BY d.label LIMIT 5; — запрос должен отработать.
3. Во время выполнения посмотри /tmp/join-demo из другого терминала — есть ли spill-файлы? По завершении они должны исчезнуть.
4. Сделай EXPLAIN ANALYZE того же запроса при 450MB и при большом лимите (например, 6GB). Сравни метку режима у HASH_JOIN.
5. Подумай: какая из таблиц (dim или fact) станет build-стороной и почему? Если бы dim была не 5 млн, а 500 строк — изменился бы риск спилла при лимите 450MB? Сформулируй, от чего зависит память hash join.