Параллельный hash join

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

Hash join: две фазы, build и probe

Сначала — как hash join работает в принципе, без параллелизма. Соединение двух таблиц по равенству ключей делается в две фазы.

Фаза build — построение. Берётся одна из двух таблиц, обычно меньшая (её называют build side, строящая сторона). По ней строится хеш-таблица: каждая строка кладётся в хеш-таблицу под ключом соединения. Хеш-таблица — структура, которая по значению ключа мгновенно отдаёт строки с этим ключом.

Фаза probe — зондирование. Берётся вторая таблица (probe side, зондирующая сторона). Каждая её строка «пробивается» по построенной хеш-таблице: по ключу соединения ищутся совпадающие строки build-стороны. Нашлись — выдаётся соединённая пара.

Связь с pipelines из первого урока модуля прямая: фаза build — это pipeline breaker. Probe не может начаться, пока хеш-таблица не построена целиком — иначе часть совпадений потеряется. Поэтому build всегда полностью завершается до probe, и эти две фазы лежат в разных pipelines.

Почему наивная параллелизация build буксует

Параллелим build. Очевидная мысль: пусть все потоки сканируют build-сторону (это мы уже умеем — параллельный скан) и пусть каждый кладёт прочитанные строки в одну общую хеш-таблицу.

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

А блокировки убивают параллелизм. Если каждая вставка в хеш-таблицу требует взять блокировку, то потоки выстраиваются в очередь к этой блокировке. Build-сторона может быть большой, вставок миллионы — и на каждой потоки толкаются за общий замок. Восемь потоков вместо параллельной работы значительную часть времени ждут друг друга у блокировки. Параллельный по замыслу build вырождается в почти последовательный. Это та же беда, что и перекос нагрузки из урока про morsel-driven, только источник другой — не неравные куски, а борьба за общий ресурс. Общая структура под параллельной записью — это узкое место.

Решение: локальные хеш-таблицы, потом слияние

DuckDB обходит конкуренцию приёмом «сначала локально, потом объединить» — тем же, что лежит в основе многих параллельных алгоритмов.

Идея такая. Не заставляем потоки писать в одну общую хеш-таблицу. Вместо этого каждый поток строит свою собственную, локальную хеш-таблицу. Поток сканирует свои morsel-ы build-стороны и складывает прочитанные строки в свою личную хеш-таблицу — ту, к которой больше никто не обращается.

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

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

Ключевой момент: синхронизация теперь происходит один раз — на слиянии в конце, — а не на каждой из миллионов вставок. Дорогая часть фазы build (чтение build-стороны и заполнение хеш-таблиц) идёт без всякой конкуренции, параллельно. Координация осталась только на дешёвом финальном слиянии. Стоимость синхронизации вынесена из горячего цикла.

Параллельная фаза probe

Фаза build завершилась — есть одна глобальная хеш-таблица. Теперь probe, и она параллелится легко.

Probe устроена так: каждая строка probe-стороны пробивается по хеш-таблице независимо от всех остальных строк. Чтобы пробить одну строку, поток лишь читает глобальную хеш-таблицу — ищет в ней совпадения по ключу. А чтение — операция, которую много потоков делают параллельно без всякой конкуренции: глобальная хеш-таблица на фазе probe уже не меняется, она достроена и заморожена, в неё только смотрят.

Поэтому probe раскладывается на потоки точно как параллельное сканирование. Probe-сторона нарезается на morsel-ы, диспетчер раздаёт их потокам, каждый поток берёт morsel строк probe-стороны и пробивает их по общей (но теперь read-only) хеш-таблице. Никаких блокировок: писать в хеш-таблицу никто не пытается, все только читают. Probe масштабируется по ядрам почти линейно.

Складывается общая картина параллельного hash join. Фаза build: дорогая работа (чтение build-стороны, заполнение хеш-таблиц) идёт параллельно через локальные хеш-таблицы, синхронизация — один раз на слиянии. Фаза probe: полностью параллельна, потоки лишь читают замороженную глобальную хеш-таблицу. Конкуренция за общий ресурс сведена к единственной точке — слиянию локальных таблиц — и убрана из всех горячих циклов.

Какую сторону DuckDB ставит в build

Ещё одна деталь, влияющая на скорость join. У соединения две таблицы, и одна станет build-стороной, другая probe-стороной. Выбор не безразличен.

Build-сторона целиком превращается в хеш-таблицу, и эта хеш-таблица должна уместиться в памяти (про случай, когда не умещается, — отдельный модуль про larger-than-memory). Probe-сторона в хеш-таблицу не складывается — её строки лишь протекают сквозь probe потоком. Значит build-стороной выгодно делать меньшую из двух таблиц: меньше build-сторона — меньше и дешевле строить хеш-таблицу, меньше памяти под неё. Большую таблицу логично пустить probe-стороной — она просто протечёт сквозь фазу probe.

DuckDB выбирает build-сторону сам. Оптимизатор оценивает размеры входов по статистике (мы касались этого в модуле про storage-формат — статистика, zonemap) и обычно ставит в build меньшую сторону. Увидеть его решение можно в плане: в выводе EXPLAIN у оператора HASH_JOIN видно, какая сторона пошла на build, а какая на probe.

EXPLAIN SELECT o.id, c.name
FROM orders o
JOIN customers c ON o.customer_id = c.id;

Фрагмент вывода:

PHYSICAL_PLAN
  HASH_JOIN
    Join Type: INNER
    Conditions: customer_id = id
    Build Side: customers
    Probe Side: orders

Строка Build Side: customers подтверждает: оптимизатор поставил в build меньшую таблицу customers (тысяча строк), а большую orders (миллион) пустил probe-стороной. По хеш-таблице из тысячи строк миллион строк orders пробьётся быстро, и на фазе probe это идеально параллелится.

Попробуй сам

Понаблюдайте за параллельным hash join.

1. Создайте большую и маленькую таблицы: CREATE TABLE orders AS SELECT range AS id, range % 5000 AS customer_id, (random()*100)::INTEGER AS amount FROM range(20000000); и CREATE TABLE customers AS SELECT range AS id, ('cust_' || range) AS name FROM range(5000);.
2. Выполните EXPLAIN SELECT count(*) FROM orders o JOIN customers c ON o.customer_id = c.id;. Найдите HASH_JOIN и строки Build Side / Probe Side. Какую таблицу движок поставил в build и почему именно её?
3. Включите таймер (.timer on) и выполните этот join с агрегацией на всех ядрах. Запишите время.
4. Ограничьте до одного потока: SET threads = 1;, повторите. Во сколько раз медленнее? Параллельный hash join должен дать заметное ускорение и на build, и на probe.
5. Выполните EXPLAIN ANALYZE для join-запроса на всех ядрах. Найдите оператор HASH_JOIN, посмотрите его время. Прикиньте, какую долю времени занял join по сравнению со сканированием.
6. Поразмышляйте: что было бы, если бы build-стороной случайно стала большая таблица orders? Почему хеш-таблица из 20 млн строк хуже, чем из 5 тысяч — по памяти и по скорости построения?

Этот эксперимент показывает hash join в параллельном исполнении: build и probe оба масштабируются по ядрам, а выбор меньшей таблицы в build делает соединение дешёвым.