GLOBAL JOIN для распределённых запросов

В распределённых кластерах ClickHouse каждый шард выполняет свою часть запроса. Но что происходит с JOIN, когда правая сторона — подзапрос? Без специальных мер каждый шард запускает этот подзапрос самостоятельно. На кластере из 10 шардов это означает 10 независимых выполнений одного и того же подзапроса. GLOBAL JOIN решает эту проблему.

Проблема N*N: JOIN без GLOBAL

Когда вы выполняете JOIN с подзапросом на Distributed-таблице, координатор (initiator node) отправляет запрос на каждый шард. Каждый шард должен вычислить правую сторону JOIN. Если правая сторона — подзапрос к другой Distributed-таблице, каждый шард снова обращается ко всем шардам:

-- Без GLOBAL: каждый из 10 шардов выполняет подзапрос
SELECT d.user_id, d.event_type, u.name
FROM distributed_events d
JOIN (
    SELECT user_id, name
    FROM distributed_users
    WHERE active = 1
) u ON d.user_id = u.user_id
WHERE d.event_date = '2024-01-15';

На кластере из 10 шардов:

Подзапрос SELECT user_id, name FROM distributed_users WHERE active = 1 выполняется 10 раз — по одному на каждом шарде. При этом сам подзапрос обращается ко всем 10 шардам. Результат: 100 сетевых обращений вместо 10.

GLOBAL JOIN: broadcast вместо повторений

GLOBAL JOIN меняет стратегию: координатор выполняет подзапрос один раз, собирает результат во временную таблицу и отправляет (broadcast) её на все шарды:

-- С GLOBAL: подзапрос выполняется один раз на координаторе
SELECT d.user_id, d.event_type, u.name
FROM distributed_events d
GLOBAL JOIN (
    SELECT user_id, name
    FROM distributed_users
    WHERE active = 1
) u ON d.user_id = u.user_id
WHERE d.event_date = '2024-01-15';

На кластере из 10 шардов:

Экономия: 100 -> 20 обращений (5x меньше на 10 шардах). На 100 шардах разница ещё больше: 10 000 -> 200.

Механизм GLOBAL JOIN

Когда использовать GLOBAL JOIN

GLOBAL JOIN полезен в конкретных сценариях:

Когда НЕ использовать GLOBAL JOIN

Практический пример: dimension enrichment

Типичный use case для GLOBAL JOIN — обогащение фактов из dimension-таблицы:

-- events распределены по 10 шардам
-- countries -- небольшая справочная таблица (250 строк)
-- Без GLOBAL: каждый шард читает countries 10 раз = 100 чтений
-- С GLOBAL: координатор читает countries 1 раз, broadcast 250 строк

SELECT
    e.event_date,
    e.user_id,
    c.country_name,
    c.region
FROM distributed_events e
GLOBAL JOIN (
    SELECT country_code, country_name, region
    FROM distributed_countries
) c ON e.country_code = c.country_code
WHERE e.event_date = '2024-01-15';

Для таблицы в 250 строк broadcast стоит минимально. Но для таблицы в 10 миллионов строк broadcast потребует значительной пропускной способности сети.

GLOBAL IN: аналогичный механизм

GLOBAL также работает с IN (подзапрос):

-- Без GLOBAL: каждый шард выполняет подзапрос
SELECT * FROM distributed_events
WHERE user_id IN (
    SELECT user_id FROM distributed_users WHERE active = 1
);

-- С GLOBAL: подзапрос один раз, результат broadcast
SELECT * FROM distributed_events
WHERE user_id GLOBAL IN (
    SELECT user_id FROM distributed_users WHERE active = 1
);

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

Ключевые выводы

1. Проблема N*N: без GLOBAL каждый из N шардов выполняет подзапрос, обращаясь ко всем N шардам. Итого N*N сетевых обращений.
2. GLOBAL JOIN: координатор выполняет подзапрос один раз, собирает результат во временную таблицу и broadcast на все шарды. Итого 2*N обращений.
3. Используйте GLOBAL JOIN при: JOIN с подзапросом к Distributed-таблице, небольшая правая сторона, большой кластер.
4. НЕ используйте GLOBAL JOIN при: одиночный сервер, огромная правая таблица (миллионы строк), локальные (не Distributed) таблицы.
5. Для частого dimension enrichment на production-кластерах используйте CREATE DICTIONARY + dictGet() — это эффективнее GLOBAL JOIN.