Join distribution: BROADCAST против PARTITIONED

Прошлый урок решал вопрос «в каком порядке джойнить таблицы». Этот — другой, не менее важный: «как именно распределить один джойн по кластеру». Когда Trino соединяет две таблицы, данные физически разбросаны по воркерам, а для join по равенству строки с совпадающим ключом обязаны встретиться на одной ноде. Свести их можно двумя принципиально разными способами: разослать одну таблицу целиком всем (BROADCAST) или перераспределить обе по хэшу ключа (PARTITIONED).

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

Build-сторона и probe-сторона

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

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

BROADCAST join: разослать build-сторону всем

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

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

Что здесь цена. Build-сторона рассылается на каждую ноду и на каждой целиком держится в памяти как хэш-таблица. Если build-таблица — 50 МБ, а воркеров 20, по сети уходит 50 МБ × 20 = 1 ГБ суммарной рассылки, и каждый из 20 воркеров тратит 50 МБ памяти на свою копию. Пока build-сторона мала, это копейки. Но если build-сторона велика — её копия не влезет в память воркера, и join упадёт по памяти. BROADCAST работает, только когда build-сторона достаточно мала.

PARTITIONED join: перераспределить обе таблицы по хэшу

При PARTITIONED join (его ещё зовут hash join по способу распределения) обе таблицы перераспределяются по сети по хэшу join-ключа. Хэш-функция от ключа определяет, на какой воркер уедет строка, — и применяется к строкам обеих таблиц одинаково. Поэтому строки с одинаковым ключом из обеих таблиц гарантированно попадают на один воркер; там они и соединяются.

Что здесь хорошо. Каждый воркер держит только свою долю build-стороны — ту часть, чьи ключи отхэшировались на него. Build-таблица не копируется целиком никуда: она разрезана по воркерам, и суммарная память кластера складывается. Это позволяет джойнить большие таблицы с большими — то, что BROADCAST не может в принципе.

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

Компромисс и свойство join-distribution-type

Сведём выбор в таблицу — это и есть суть компромисса.

Управляет выбором свойство — конфигурационное join-distribution-type, сессионное join_distribution_type. Три значения:

• AUTOMATIC — по умолчанию. CBO сам решает по статистике: оценивает размер build-стороны и, если она достаточно мала, выбирает BROADCAST, иначе PARTITIONED.
• BROADCAST — принудительно broadcast.
• PARTITIONED — принудительно partitioned.

Решение AUTOMATIC опирается на порог: build-сторона уходит в BROADCAST, только если её оценочный размер не превышает лимит. Лимит задаётся свойством join-max-broadcast-table-size (сессионно join_max_broadcast_table_size) и по умолчанию составляет 100 МБ. Build-сторона мельче порога — BROADCAST; крупнее — PARTITIONED.

И снова всё упирается в статистику. Чтобы оценить размер build-стороны, CBO нужны row count и data size этой таблицы. Нет статистики — нет надёжной оценки размера — AUTOMATIC выбирает вслепую и может промахнуться: назначить BROADCAST таблице, которая на деле велика.

Как увидеть тип распределения

Тип распределения каждого джойна виден в EXPLAIN — у join-узла указан distribution type. Полезно сравнить план до и после ANALYZE или при разных значениях join_distribution_type:

EXPLAIN
SELECT o.orderkey, c.name
FROM tpch.sf100.orders o
JOIN tpch.sf1.customer c ON o.custkey = c.custkey;

В плане у join-узла будет помечен тип. Здесь orders из sf100 — крупная (probe-сторона), customer из sf1 — небольшая. CBO, видя по статистике, что customer мала, скорее всего выберет для неё REPLICATED (так в плане обозначается build-сторона broadcast join’а): маленький customer рассылается на все ноды, большой orders остаётся на месте. Если бы обе таблицы были крупными, в плане стоял бы PARTITIONED с шаффлом обеих.

Попробуй сам

На песочнице курса (Trino 481):

1. Возьмите EXPLAIN для join tpch.sf100.orders с tpch.sf1.customer по custkey. Найдите join-узел и определите тип распределения. Объясните, какая таблица стала build-стороной и почему CBO выбрал именно этот тип.

2. Выполните тот же EXPLAIN дважды: сначала с SET SESSION join_distribution_type = 'BROADCAST';, затем с 'PARTITIONED'. Сравните планы. Опишите, что в каждом случае происходит с probe-стороной — двигается она по сети или нет. Верните AUTOMATIC.

3. Рассуждение в двух абзацах. Первый: почему BROADCAST дешевле по сети, но падает по памяти на большой build-стороне — распишите, что именно копируется и куда. Второй: команда жалуется, что join валится с out of memory, хотя кластер по объёму данных должен справляться; назовите вероятную причину и два шага лечения.