docs(research): изучено соединение Distributed-таблиц на кластере
Зачем: тикет #129 (карта #127) требовал проверить, где выполняется JOIN двух Distributed-таблиц, что даёт GLOBAL и как влияет distributed_product_mode, — от этого зависят сборка dds.identity_map и витрина сверки purchase_vs_orders_v этапа 4. Что: заметка docs/research/2026-08-22-distributed-join.md. Ключевое: GLOBAL рассылает правую часть с инициатора на шарды; режим local молча даёт неверный ответ без настоящей ко-локации; подзапрос слева переносит соединение на инициатор; FULL OUTER с голой левой таблицей задваивает непарные строки правой части. Попутная находка: ко-локация заказов по cityHash64(order_id) сломана заводским parallel_distributed_insert_select=2 — починка отдельным тикетом. Проверка: читающие запросы на живом стенде 26.3.17.56 (log_comment r129-* в system.query_log обеих нод) и документация ClickHouse через MCP Context7; стенд не менялся, main не тронут. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
ddadminandClaude Fable 5 committed
1 parent
ea39982bdd
commit
f218795472
1 file changed
+235
@@ -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 на кластере в документации не описано —
|
||||
здесь источник только эмпирика.
|
||||
Reference in new issue
Block a user