diff --git a/docs/research/2026-08-22-distributed-join.md b/docs/research/2026-08-22-distributed-join.md new file mode 100644 index 0000000..f0fc560 --- /dev/null +++ b/docs/research/2026-08-22-distributed-join.md @@ -0,0 +1,235 @@ +# Соединение двух 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-таблиц — запрещён + +```sql +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) отвечают: + +```text +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 — правая часть собирается один раз и рассылается + +```sql +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`): + +```text +SELECT order_id FROM ods.order_rep -- шард отдаёт свой кусок правой части +SELECT count() FROM ods.order_rep +GLOBAL ALL INNER JOIN `_data_14241958685358625318_…` ON … -- шард соединяет локальную левую с разосланной копией +``` + +Инициатор один раз прочёл правую часть со всех шардов, собрал временную +таблицу в памяти, разослал её и слил частичные результаты. Цена GLOBAL — +память инициатора и рассылка по сети, поэтому он для малой правой части; +заказы (2851 строка) малы. + +### Подзапрос слева выключает раздачу по шардам + +```sql +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 на кластере + +Строка правой части, не нашедшая пары, добавляется в ответ каждым шардом: + +```sql +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 (одна фиктивная строка вернулась дважды — по разу с каждого шарда) +``` + +Так ведут себя обе версии планировщика. Механика прямая: каждый шард получил +копию правой части, у каждого она не нашла пары в его локальном куске, и +каждый честно отдал её как непарную. Стоит обернуть левую часть в подзапрос — +соединение выполняется один раз на инициаторе, и ответ становится верным: + +```sql +… 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 на кластере в документации не описано — +здесь источник только эмпирика.