clickstream-data-platform/docs/research/2026-08-22-distributed-join.md

Соединение двух Distributed-таблиц на стенде

Дата исследования: 2026-08-22. Тикет #129, хвост раздела 11 мастер-спеки.

Учебный результат: менти видит на своём кластере, где именно выполняется JOIN двух Distributed-таблиц, чем GLOBAL отличается от обычного соединения и почему «правильный» ключ шардирования в DDL ещё не гарантирует ко-локацию данных.

Проверено двумя дорогами: читающими запросами на живом стенде (ClickHouse 26.3.17.56, 2 шарда × 1 реплика; запросы помечены log_comment = 'r129-*', разбор — по system.query_log обеих нод) и по документации ClickHouse через MCP Context7 (страницы «JOIN Clause → Distributed JOIN», «Distributed table engine», «IN operators → distributed subqueries», настройки distributed_product_mode, prefer_global_in_and_join, parallel_distributed_insert_select).

Короткий вывод

• Обычный JOIN и IN с Distributed-таблицей справа режутся настройкой distributed_product_mode = 'deny' — это заводское значение, стенд его не менял. Ошибка: DISTRIBUTED_IN_JOIN_SUBQUERY_DENIED.
• GLOBAL JOIN собирает правую часть один раз на ноде-инициаторе, кладёт во временную таблицу _data_… и рассылает её на все шарды; каждый шард соединяет свой локальный кусок левой таблицы с этой копией. Это видно в журнале запросов второй ноды дословно.
• Режим local переписывает правую Distributed-таблицу на локальную. Ответ верен только при настоящей ко-локации по ключу соединения; иначе он молча меньше правильного — ошибки не будет.
• Если левая часть — подзапрос, а не голая таблица, ClickHouse 26.3 вообще не раздаёт соединение по шардам: обе стороны стягиваются на инициатор и JOIN выполняется там один раз. GLOBAL и distributed_product_mode при этом ни на что не влияют.
• Ловушка FULL OUTER: при голой левой Distributed-таблице непарная строка правой части попадает в ответ по разу с каждого шарда — на нашем кластере дважды. С подзапросами с обеих сторон ответ верный.
• Находка стенда: ко-локация заказов по cityHash64(order_id) сломана — 1189 из 1662 заказов лежат на обоих шардах одинаковыми строками. Виновата настройка parallel_distributed_insert_select = 2 (заводское значение в 26.3): INSERT … SELECT из Distributed в Distributed выполнился локально на каждом шарде, и строки заказов унаследовали раскладку сырья — по хешу сырой строки. События целы: все 401 185 строк лежат по своему хешу cityHash64(ClientID).

Как выполняется соединение: три наблюдаемых плана

Обычный JOIN голых Distributed-таблиц — запрещён

SELECT count() FROM ods.order_dist AS a
INNER JOIN ods.order_dist AS b ON a.order_id = b.order_id

Обе версии планировщика (enable_analyzer 0 и 1) отвечают:

Code: 288. DB::Exception: Double-distributed IN/JOIN subqueries is denied
(distributed_product_mode = 'deny'). You may rewrite query to use local
tables in subqueries, or use GLOBAL keyword, or set distributed_product_mode
to suitable value. (DISTRIBUTED_IN_JOIN_SUBQUERY_DENIED)

Причина по документации: запрос переписывается для каждого шарда, и правая Distributed-таблица внутри такой переписки означала бы «каждый шард ещё раз спрашивает весь кластер» — N шардов породили бы N×N подзапросов. Режим deny запрещает это по умолчанию. Тот же запрет действует на IN (SELECT … FROM …_dist) — поэтому в исследовательских запросах ниже стоит GLOBAL IN.

GLOBAL JOIN — правая часть собирается один раз и рассылается

SELECT count() FROM ods.order_dist AS a
GLOBAL INNER JOIN ods.order_dist AS b ON a.order_id = b.order_id

Журнал запросов второй ноды показывает оба шага (log_comment r129-bare-global):

SELECT order_id FROM ods.order_rep                -- шард отдаёт свой кусок правой части
SELECT count() FROM ods.order_rep
GLOBAL ALL INNER JOIN `_data_14241958685358625318_…` ON …  -- шард соединяет локальную левую с разосланной копией

Инициатор один раз прочёл правую часть со всех шардов, собрал временную таблицу в памяти, разослал её и слил частичные результаты. Цена GLOBAL — память инициатора и рассылка по сети, поэтому он для малой правой части; заказы (2851 строка) малы.

Подзапрос слева выключает раздачу по шардам

SELECT count()
FROM (SELECT arrayJoin(purchaseID) AS pid FROM ods.event_dist
      WHERE notEmpty(purchaseID)) AS e
INNER JOIN ods.order_dist AS o ON e.pid = o.order_id

Этот запрос прошёл без ошибки даже при deny, и с GLOBAL, и во всех режимах distributed_product_mode — план один и тот же: вторая нода получила два независимых чтения (SELECT … FROM ods.event_rep, SELECT order_id FROM ods.order_rep), а соединение выполнил инициатор, прочитав целиком обе стороны (404 036 строк). Раздачи соединения по шардам здесь нет вовсе, поэтому нет и «двойной распределённости», которую режим deny запрещает.

distributed_product_mode на живых данных

Самосоединение заказов, счёт пар строк. Правильные ответы посчитаны отдельно как суммы квадратов кратностей: по order_id — 5229, по user_id — 6087.

Ключ	GLOBAL	local	allow
order_id (заявленный ключ шардирования)	5229	2851	—
user_id (ко-локации нет и не заявлено)	6087	3281	6087

local переписывает правую часть на локальную таблицу — в журнале второй ноды виден дословный … FROM ods.order_rep INNER JOIN ods.order_rep …. Ответ совпадает с суммой пошардовых соединений и верен только тогда, когда обе строки каждой пары лежат на одном шарде. По user_id это ожидаемо не так. По order_id ответ тоже неверен — потому что настоящая раскладка заказов ключу не соответствует (см. ниже). allow разрешает наивный план «каждый шард спрашивает весь кластер»: ответ верный, чтений — квадрат от числа шардов.

Документация даёт четыре режима (deny, local, global, allow) и отдельную настройку prefer_global_in_and_join, которая молча превращает обычные IN/JOIN в GLOBAL. Стенд ничего из этого не меняет: явный GLOBAL в тексте запроса честнее незаметной настройки — читатель видит цену в самом запросе.

Ловушка FULL OUTER на кластере

Строка правой части, не нашедшая пары, добавляется в ответ каждым шардом:

SELECT countIf(b_flag = 1) FROM ods.order_dist AS a
GLOBAL FULL OUTER JOIN
  (SELECT 'zzz-no-such-order' AS oid, 1 AS b_flag) AS b ON a.order_id = b.oid
-- ответ: 2  (одна фиктивная строка вернулась дважды — по разу с каждого шарда)

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

… FROM (SELECT order_id FROM ods.order_dist) AS a GLOBAL FULL OUTER JOIN … 
-- ответ: 1

Находка: ко-локация заказов сломана

DDL обещает раскладку по cityHash64(order_id), но на живых данных половина строк лежит не на своём шарде:

Шард	на своём шарде	не на своём
1	696	725
2	750	680

1189 из 1662 заказов присутствуют на обоих шардах одинаковыми строками — одна и та же версия, один и тот же _load_id. Пути реплик в keeper у шардов разные, репликация ни при чём.

Причина — parallel_distributed_insert_select = 2, заводское значение в 26.3. По документации: если выполняется INSERT INTO distributed_a SELECT … FROM distributed_b и обе таблицы на одном кластере, запрос выполняется локально на каждом шарде. Ровно это делает разбор заказов (sql/ods/order_parse.sql: INSERT INTO ods.order_dist … FROM stg.orders_raw_dist): каждый шард разобрал свой кусок сырья и записал его в свою локальную таблицу, минуя ключ шардирования цели. Строки заказов унаследовали раскладку STG — хеш сырой строки, то есть случайную относительно order_id. Заказ, попавший в несколько дневных слепков, приезжает несколькими разными сырыми строками и оседает на обоих шардах; ReplacingMergeTree схлопывает дубли только внутри шарда.

Следствия уже наблюдаемы: count() по ods.order_dist даёт 2851 при 1662 заказах, а точное чтение работает только через GROUP BY order_id / argMax — пошардовый FINAL через дубль на соседнем шарде не переступит. События этой болезни не имеют: их пишет матвью приёма от Kafka-таблицы, не из Distributed, и раскладка по cityHash64(ClientID) точна на всех 401 185 строках.

Починка — отдельный тикет, не этого исследования: разбору заказов нужно явное parallel_distributed_insert_select = 0 (строки поедут через ключ шардирования цели), плюс пересборка уже разложенного. Тот же капкан ждёт любую будущую сборку DDS вида INSERT INTO …_dist SELECT … FROM …_dist, у которой ключ цели отличается от ключа источника.

Выводы для identity_map (спека §5)

Сборка карты — соединение покупок из событий с заказами по purchaseID = order_id. Ключ соединения не ключ ко-локации событий, поэтому по политике раздела 6 спеки соединение обязано быть GLOBAL — и эмпирика это подтверждает: обычный JOIN голых таблиц запрещён, local дал бы молча неверный ответ. GLOBAL здесь дёшев: правая часть — заказы, их тысячи.

На практике левая часть сборки — подзапрос (события фильтруются по notEmpty(purchaseID) и разворачиваются arrayJoin), а значит на 26.3 соединение выполнится один раз на инициаторе. GLOBAL в тексте при этом не лишний: он выражает политику спеки, ничего не стоит и защищает запрос от смены плана при другой форме запроса или версии сервера.

Запись результата в dds.identity_map (шардирование cityHash64(client_id)) пойдёт тем же INSERT … SELECT. Источник — результат соединения, а не голая Distributed-таблица, поэтому капкан parallel_distributed_insert_select на эту вставку не распространяется; но проверку раскладки после первой сборки стоит включить в приёмку — теперь ясно, что заявленный ключ и настоящая раскладка умеют расходиться.

Выводы для purchase_vs_orders_v (спека §7)

FULL OUTER GLOBAL JOIN для сверки корректен при одном условии: обе стороны — подзапросы, и соединение выполняется один раз на инициаторе. Витрина это условие выполняет естественно: покупки нужно развернуть из массива и агрегировать (класс duplicate_event иначе не посчитать), заказы — привести к последней версии. С голой Distributed-таблицей слева витрина удваивала бы каждый заказ без события — весь класс lost_event — по числу шардов.

Прогон сверочной формы на живых данных (обе стороны — подзапросы с GROUP BY): 1662 совпадения, 255 покупок без заказа (ждут слепка, awaiting_order), 0 заказов без покупки, без задвоений. Обе величины согласуются с прямыми счётами — форма пригодна как каркас витрины.

Чем проверялось

Все запросы — читающие, от служебного пользователя внутри контейнеров (docker compose exec clickhouse-01 … clickhouse-client), следы — в system.query_log обеих нод по log_comment с префиксом r129-. Документные утверждения (механика GLOBAL, режимы distributed_product_mode, поведение parallel_distributed_insert_select) сверены с документацией ClickHouse через MCP Context7; поведение FULL OUTER на кластере в документации не описано — здесь источник только эмпирика.