JOIN в CHYT

В статье представлено высокоуровневое описание устройства JOIN в CHYT.

Сортировка vs шардирование

Схема хранения таблиц в YTsaurus принципиально отличается от таковой в ClickHouse. В YTsaurus базовый примитив хранения — статическая таблица, строки которой лежат по чанкам, произвольным образом раскиданным по кластеру. Статические таблицы крайне неэффективны в задачах, которые требуют точечного чтения данных (так как эти точечные чтения, как правило, приводят к Random IO на жестком диске). Для быстрых точечных чтений в YTsaurus есть более эффективный и сложный примитив, который называется динамические таблицы.

Однако многие batch-приложения, обрабатывающие потоком большой объем данных, требуют так или иначе работать в модели, когда часть колонок являются ключевыми. В YTsaurus поддерживается понятие сортированной таблицы — в схеме таблицы может быть отражен тот факт, что ее строки отсортированы по некоторому префиксу колонок. Такие колонки называются ключевыми. Подобная метаинформация позволяет эффективно реализовать, например, операцию Sorted Reduce, отсутствующую в оригинальной парадигме Map-Reduce. В YTsaurus семейство Reduce очень богатое: в частности, операции типа Reduce поддерживают foreign-таблицы с LEFT INNER JOIN семантикой.

В ClickHouse основой для распределенного хранения данных является не сортировка, как в YTsaurus, а шардирование. По умолчанию данные размазываются по шардам согласно некоторому ключу шардирования на основании остатка от деления выражения шардирования на суммарный вес всех шардов (подробнее можно прочитать в документации ClickHouse. Также зачастую пользователи сами управляют логикой шардирования, самостоятельно вставляя данные на конкретные хосты.

Обе схемы достигают одной цели разными способами:

В YTsaurus сортировка обеспечивает локальность строк с одним значением ключа в одном чанке (либо в наборе подряд идущих).
В ClickHouse шардирование обеспечивает локальность строк с одним значением выражения шардирования на одной машине.

Как работает запрос в CHYT?

В разделе Анатомия запроса детально описывается устройство CHYT. Любой SELECT-запрос разбивается координатором на порции по основной таблице, после чего каждая порция обрабатывается независимо на своем инстансе.

Подобная схема исполнения очень близка к таковой в Distributed-движке ClickHouse за тем исключением, что в ClickHouse чтение данных происходит буквально "из-под ног" у инстанса, а в YTsaurus чанки разбросаны произвольным образом по машинам кластера, что компенсируется очень толстой сетью.

Такой метод хорошо пригоден для потоковых запросов из одного источника, не содержащих JOIN. Однако как только возникает необходимость джойнить, приходится придумывать, как совмещать относящиеся друг к другу строки из разных источников, не порождая при этом необходимость делать случайные чтения.

Виды JOIN в ClickHouse

Как вообще может работать координация (то есть распределение нагрузки между инстансами) JOIN в распределенном окружении? В ClickHouse возможны следующие стратегии исполнения конструкции lhs JOIN rhs USING/ON ...:

Distributed local JOIN: если таблицы шардированы одинаковым образом, то можно исполнять JOIN независимо на каждом инстансе, так как верно, что пара ключей, соединенная JOIN, не может оказаться на разных машинах. Таким образом, lhs и rhs на каждом инстансе интерпретируются как соответствующие им локальные таблицы.
GLOBAL JOIN: если использовать ключевое слово GLOBAL рядом с JOIN, то можно форсировать систему поступить следующим образом. На координаторе запроса полностью исполняется и материализуется правый аргумент rhs и его сериализованное представление рассылается вместе с запросом по инстансам, а инстансам предлагается пользоваться этим представлением для получения правой части в своей памяти. Данный метод хорош, когда rhs сравнительно небольшого размера, а инстансов сравнительно немного: легко видеть, что при невыполнение одного из этих условий можно упереться в раздачу таблицы с координатора по подзапросом по сети (либо вообще по памяти на координаторе). Этот метод не требует никаких дополнительных условий на согласованность схемы хранения/шардирования на таблицах.
JOIN via subqueries. ClickHouse позволяет окружить lhs и/или rhs скобками, и это существенно влияет на план исполнения:
— Если окружить lhs в скобки, то ClickHouse теряет какую-либо информацию про устройство lhs, в частности, пропадает знание про распределённую натуру левой части. В такой ситуации на координаторе независимо исполняется левая часть, независимо исполняется правая часть, правая часть поднимается в оперативную память в хеш-таблицу и дальше происходит полное исполнение JOIN только на координаторе.
— Если окружить rhs скобками, то ClickHouse произведет распределенный запрос, как если бы запрос выглядел просто как SELECT lhs. Затем отправит на инстансы свои запросы, оставив JOIN (rhs) как есть. Далее каждый инстанс будет исполнять rhs независимо, что может привести к кратно большей нагрузке, так как каждый инстанс будет материализовывать правую часть независимо. Против последней проблемы в ClickHouse есть защитный механизм, который по умолчанию запрещает такое поведение и приводит к ошибке Double-distributed IN/JOIN subqueries is denied.

Подробнее об этом читайте в документации ClickHouse, релевантные ссылки:

Как же выглядит аналогичная классификация в CHYT? Пункты 2 и 3 работают точно так же: стратегия 2 хороша, если правая часть таблицы маленькая. Первый вариант стратегии 3 подходит, если правая таблица очень большая, но есть возможность ждать долго. Хочется заметить, что CHYT предназначен для быстрых аналитических запросов, а если что-то требует джойнить большие разнородные таблицы, то лучше использовать YQL и Map-Reduce. Поэтому данный способ не рекомендуется.

В стратегии 1 в CHYT присутствует существенное отличие, о чем можно прочитать ниже.

Sorted JOIN

Вместо логики Distributed JOIN, использующей идентичность схемы шардирования аргументов, в CHYT естественным образом возникает логика Sorted JOIN, использующая идентичность схемы сортировки аргументов. Для хорошо понимающих, как устроена операция Sorted Reduce, Sorted JOIN работает ровно таким же образом.

Для того, чтобы воспользоваться стратегией Sorted JOIN, необходимо использовать обычную конструкцию lhs JOIN rhs USING/ON ..., но на lhs и rhs налагаются следующие дополнительные ограничения:

lhs и rhs должны быть сортированными таблицами. Пусть lhs отсортирована по колонкам l1, l2, ..., ln, а rhs – по колонкам r1, r2, ..., rm.
Условие JOIN должно выглядеть как набор равенств l1 = r1, ..., lk = rk для некоторого k (сами равенства могут идти в произвольном порядке). Это может быть выражено как набором равенств в ON-клаузе, так и набором общих ключевых колонок в USING-клаузе, но не условием в WHERE-клаузе.

При соблюдении этих условий можно переиспользовать логику координации из операции Sorted reduce, формируя пары соответствующих диапазонов из lhs и rhs и распределяя их по инстансам в подзапросах. Если же это условие не выполнено, то возникнет ошибка и придется пользоваться либо стратегией 2 (использовать GLOBAL JOIN), либо вторым вариантом стратегии 3 (заключать правую часть в подзапрос).