02.09.2025 09:30
MirrorShip 0 140 Источник

Добро пожаловать в серию статей «Техническая внутренняя кухня StarRocks», где мы всесторонне раскрываем принципы и практические детали, лежащие в основе StarRocks, чтобы помочь вам шаг за шагом освоить этот популярный продукт с открытым исходным кодом.

Эта статья основана на офлайн-выступлении автора на StarRocks MeetUp и посвящена опыту и исследованиям StarRocks в планировании запросов с JOIN. Материал разделён на четыре части: предпосылки JOIN, логическая оптимизация JOIN, Join Reorder (перестановка порядка соединений) и планирование распределённых JOIN.

#01 Предпосылки JOIN

1. Типы JOIN

Ниже перечислены распространённые типы JOIN:

  • CROSS JOIN: декартово произведение левой и правой таблиц.

  • FULL / LEFT / RIGHT OUTER JOIN: внешние соединения, при которых согласно семантике строки без совпадений в обеих/левой/правой таблицах дополняются значениями NULL.

  • ANTI JOIN: выводит строки, не имеющие совпадений по отношению соединения; обычно возникает при планировании подзапросов NOT IN или NOT EXISTS.

  • SEMI JOIN: в отличие от ANTI JOIN, выводит только строки, имеющие совпадения по отношению соединения.

  • INNER JOIN: возвращает пересечение левой и правой таблиц; по условию соединения может порождать результат «один-ко-многим».

2. Трудности оптимизации JOIN

Оптимизация эффективности выполнения JOIN ведётся в двух плоскостях:

  • повысить эффективность самого оператора JOIN на одном узле;

  • спланировать рациональный план JOIN, максимально сокращая вход/стоимость выполнения.

Далее — о ключевых трудностях второй части.

  • Проблема 1: множество способов реализации JOIN.

    • В разных сценариях выигрывают разные реализации: сортировочно-сливное соединение (Sort-Merge Join) при соединении упорядоченных данных может быть значительно эффективнее хеш‑соединения (Hash Join), тогда как в распределённых СУБД с хеш‑распределением данных Hash Join часто существенно обгоняет Sort-Merge Join. СУБД должна выбирать реализацию по контексту.

  • Проблема 2: порядок выполнения многотабличных JOIN.

    • В многотабличных запросах выгодно сначала выполнять соединения с высокой селективностью, но вычислить оптимальный порядок сложно. В модели Left-Deep число перестановок для N таблиц достигает 2^(n−1), а в кустистой (Bushy) — 2^(n−1) * C(n−1). Поиск оптимума экспоненциально дорог.

  • Проблема 3: сложность оценки эффекта JOIN.

    • До выполнения SQL систему сложно точно оценить селективность соединений. На практике встречаются многочисленные «один-ко-многим», а агрегации и фильтры меняют распределение данных — входы JOIN оцениваются грубо.

  • Проблема 4: лучший однопроцессорный план не равен лучшему распределённому.

    • В распределённой системе участвуют пересылка (shuffle), широковещание и иные сетевые операции. План, оптимальный для одной машины (без учёта распределения и сетевой стоимости), в распределённой БД часто неоптимален. Нужно учитывать распределение данных и сетевую стоимость.

3. Процесс оптимизации SQL

Оптимизация SQL в StarRocks выполняется оптимизатором, в основном на этапах Rewrite и Optimize. Подробности: «Обзор кода оптимизатора StarRocks» — https://forum.mirrorship.cn/t/topic/18561

4. Принципы оптимизации JOIN

В StarRocks основным алгоритмом JOIN является хеш‑соединение (Hash Join) с построением хеш-таблицы по правой таблице по умолчанию. На этой основе выделим пять направлений оптимизации:

  1. Операторы разных типов JOIN имеют разную производительность; используйте по возможности более быстрые. Примерная иерархия по объёму вывода: SEMI/ANTI JOIN > INNER JOIN > OUTER JOIN > FULL OUTER JOIN > CROSS JOIN.

  2. В Hash Join использовать маленькую таблицу для построения хеш-таблицы значительно эффективнее, чем большую.

  3. При многотабличном JOIN сначала выполняйте соединения с высокой селективностью — это резко сокращает дальнейшие издержки.

  4. Максимально уменьшайте объём данных, участвующих в JOIN.

  5. Максимально сокращайте сетевые издержки, возникающие при распределённом JOIN.

#02 Логическая оптимизация JOIN

Ниже — набор эвристических правил.

1. Преобразование типов соединения

Задача — преобразовывать менее эффективные типы JOIN в более эффективные.

  • Правило 1: CROSS JOIN → INNER JOIN Допустимо, если на соединении есть хотя бы один предикат, задающий отношение соединения.

-- До
SELECT * FROM t1, t2 WHERE t1.v1 = t2.v1;

-- После: t1.v1 = t2.v1 — предикат соединения
SELECT * FROM t1 INNER JOIN t2 ON t1.v1 = t2.v1;

  • Правило 2: OUTER JOIN → INNER JOIN Условия:

    1. В LEFT/RIGHT OUTER JOIN существует предикат по RIGHT/LEFT таблице;

    2. Этот предикат — NULL‑отвергающий (строгий).

-- До
SELECT * 
FROM t1 LEFT OUTER JOIN t2 ON t1.v1 = t2.v1 
WHERE t2.v1 > 0;

-- После: t2.v1 > 0 — строгий предикат по t2
SELECT * 
FROM t1 INNER JOIN t2 ON t1.v1 = t2.v1 
WHERE t2.v1 > 0;

Важно: в OUTER JOIN дополнение NULL выполняется по предикатам секции ON, а не фильтрация, поэтому преобразование неприменимо к предикатам в ON:

SELECT *  
FROM t1 LEFT OUTER JOIN t2 
     ON t1.v1 = t2.v1 AND t2.v1 > 1;

-- Не эквивалентно:
SELECT *  
FROM t1 INNER JOIN t2 
     ON t1.v1 = t2.v1 AND t2.v1 > 1;

О «строгих» предикатах. StarRocks называет строгим (NULL‑отвергающим) предикат, отбрасывающий значения NULL (например, a > 0); нестрогие — те, что не отбрасывают NULL (например, a IS NULL, а также некоторые IF, CASE WHEN и функциональные предикаты). Простая проверка: заменить все проверяемые столбцы на NULL и упростить выражение; если результат True — предикат нестрогий, если False или NULL — строгий.

  • Правило 3: FULL OUTER JOIN → LEFT/RIGHT OUTER JOIN Допустимо, если существует строгий предикат, привязанный к левой/правой таблице.

-- До
SELECT * 
FROM t1 FULL OUTER JOIN t2 ON t1.v1 = t2.v1 
WHERE t1.v1 > 0;

-- После: t1.v1 > 0 — строгий предикат левой таблицы
SELECT * 
FROM t1 LEFT OUTER JOIN t2 ON t1.v1 = t2.v1 
WHERE t1.v1 > 0;

2. Проталкивание предикатов (predicate pushdown)

Цель — заранее фильтровать вход JOIN для повышения производительности.

Для секции WHERE проталкивание возможно при:

  • любом типе JOIN;

  • предикат WHERE привязан к одному из входов.

SELECT *  
FROM t1 LEFT OUTER JOIN t2 ON t1.v1 = t2.v1
        LEFT OUTER JOIN t3 ON t2.v2 = t3.v2
WHERE t1.v1 = 1 AND t2.v1 = 2 AND t3.v2 = 3;

Процесс:

  • Шаг 1: проталкиваем (t1.v1 = 1 AND t2.v1 = 2) и (t3.v2 = 3). Тип меняется: (t1 LEFT OUTER JOIN t2) LEFT OUTER JOIN t3 → (t1 LEFT OUTER JOIN t2) INNER JOIN t3.

  • Шаг 2: далее проталкиваем (t1.v1 = 1) и (t2.v1 = 2); (t1 LEFT OUTER JOIN t2) → (t1 INNER JOIN t2).

Правила для предикатов секции ON отличаются.

  • Для INNER JOIN проталкивание предикатов из ON аналогично WHERE.

  • Для OUTER / SEMI / ANTI JOIN:

    • должен быть LEFT/RIGHT OUTER/SEMI/ANTI JOIN;

    • предикаты соединения можно проталкивать только если они привязаны к правому/левому входу (противоположной стороне).

SELECT *  
FROM t1 LEFT OUTER JOIN t2 
     ON t1.v1 = t2.v1 AND t1.v1 = 1 AND t2.v1 = 2
     LEFT OUTER JOIN t3 
     ON t2.v2 = t3.v2 AND t3.v2 = 3;

  • Шаг 1: проталкиваем (t3.v2 = 3), привязанный к правой таблице в (t2 ⋈ t3). Тип JOIN не меняем (нельзя сделать INNER).

  • Шаг 2: проталкиваем (t2.v1 = 2), привязанный к правой таблице в (t1 ⋈ t2). Предикат t1.v1 = 1 привязан к левой таблице; его pushdown отфильтровал бы данные t1, что противоречит семантике LEFT OUTER JOIN, поэтому его нельзя проталкивать.

3. Извлечение предикатов (по значениям столбцов)

Обычный pushdown работает для конъюнктивных предикатов (AND), но не для дизъюнкций (OR). На практике OR встречается часто, поэтому применяется извлечение предикатов: из дизъюнкций выделяются конъюнктивные ограничения по диапазонам столбцов, которые можно протолкнуть.

-- До извлечения:
SELECT *  
FROM t1 JOIN t2 ON t1.v1 = t2.v1
WHERE (t2.v1 = 2 AND t1.v2 = 3) OR (t2.v1 > 5 AND t1.v2 = 4);

-- После извлечения диапазонов: t2.v1 >= 2 и t1.v2 IN (3, 4)
SELECT *  
FROM t1 JOIN t2 ON t1.v1 = t2.v1
WHERE (t2.v1 = 2 AND t1.v2 = 3) OR (t2.v1 > 5 AND t1.v2 = 4)
  AND t2.v1 >= 2 AND t1.v2 IN (3, 4);

Важно: извлечённые предикаты могут задавать надмножество исходных диапазонов, их нельзя просто «заменить» исходные предикаты.

4. Вывод эквивалентностей

Используем связь соединения для переноса ограничений между таблицами.

-- Исходный SQL
SELECT *  
FROM t1 JOIN t2 ON t1.v1 = t2.v1
WHERE (t2.v1 = 2 AND t1.v2 = 3) OR (t2.v1 > 5 AND t1.v2 = 4);

-- После извлечения диапазонов:
SELECT *  
FROM t1 JOIN t2 ON t1.v1 = t2.v1
WHERE (t2.v1 = 2 AND t1.v2 = 3) OR (t2.v1 > 5 AND t1.v2 = 4)
  AND t2.v1 >= 2 AND t1.v2 IN (3, 4);

-- По эквивалентности (t1.v1 = t2.v1) и (t2.v1 >= 2) → добавляем (t1.v1 >= 2)
SELECT *  
FROM t1 JOIN t2 ON t1.v1 = t2.v1
WHERE (t2.v1 = 2 AND t1.v2 = 3) OR (t2.v1 > 5 AND t1.v2 = 4)
  AND t2.v1 >= 2 AND t1.v2 IN (3, 4) AND t1.v1 >= 2;

Ограничения:

  • WHERE: практически без ограничений — можно «переносить» в обе стороны.

  • ON:

    • для INNER JOIN — как в WHERE;

    • для остальных — поддерживаются только SEMI JOIN и OUTER JOIN, и только однонаправленный перенос, противоположный направлению JOIN (например, для LEFT OUTER JOIN — с левой таблицы на правую). Это связано с тем, что pushdown для OUTER JOIN допускает проталкивание только предикатов правой стороны; перенос предиката на левую сторону не даст ранней фильтрации и лишь добавит стоимость.

Реализация в StarRocks: поддерживаются два отображения (map):

  • Column ↔ Column для эквивалентностей столбцов;

  • Column ↔ значение/выражение для равенств столбцов со значениями. Взаимный поиск по этим map и даёт вывод эквивалентностей.

5. Проталкивание LIMIT

Помимо предикатов, в JOIN поддерживается проталкивание LIMIT. Если в запросе OUTER JOIN или CROSS JOIN, LIMIT можно протолкнуть в дочерний узел с «стабильным» числом выходных строк.

  • Для LEFT OUTER JOIN число выходных строк не меньше числа строк левого входа ⇒ LIMIT можно протолкнуть в левую таблицу (для RIGHT OUTER JOIN — наоборот).

-- До
SELECT * 
FROM t1 LEFT OUTER JOIN t2 ON t1.v1 = t2.v1
LIMIT 100;

-- После
SELECT * 
FROM (SELECT * FROM t1 LIMIT 100) t
LEFT OUTER JOIN t2 ON t.v1 = t2.v1
LIMIT 100;

  • Для CROSS JOIN и FULL OUTER JOIN:

    • CROSS JOIN выдаёт декартово произведение (строки левой × строки правой);

    • FULL OUTER JOIN — как минимум (строки левой + строки правой). Поэтому LIMIT можно проталкивать в обе таблицы:

-- До
SELECT * 
FROM t1 JOIN t2
LIMIT 100;

-- После
SELECT * 
FROM (SELECT * FROM t1 LIMIT 100) x1
JOIN (SELECT * FROM t2 LIMIT 100)
LIMIT 100;

#03 Join Reorder (перестановка порядка соединений)

Join Reorder определяет порядок выполнения многотабличных JOIN, чтобы сначала выполнить наиболее селективные соединения и уменьшить вход последующих операторов.

В StarRocks Join Reorder работает над непрерывной группой INNER JOIN или CROSS JOIN. Такая последовательность называется узел multi-join (Multi Join Node), и именно он — единица перестановки. Если есть несколько Multi Join Node, перестановка выполняется для каждого отдельно.

Распространённые подходы к Join Reorder (по моделям и алгоритмам):

  • Heuristic: эвристики (например, MemSQL), порядок вокруг «звезды» (факт + измерения).

  • Left-Deep: лево-глубокая модель, малое пространство поиска, не гарантирует оптимум.

  • Bushy: «кустистая» модель, большое пространство, содержит оптимум. Типичные алгоритмы:

    • Exhaustive (коммутативность + ассоциативность)

    • Greedy

    • Simulated annealing

    • DP (DPsize, DPsub, DPccp …)

    • Genetic (например, Greenplum)

    • …

В StarRocks реализованы Left-Deep, Exhaustive, Greedy, DPsub. Кратко о двух из них.

1. Exhaustive

Перебор опирается на два правила, покрывающих практически все перестановки:

  • Правило 1: коммутативность JOIN.
    A JOIN B → B JOIN A (учитывая, что тип может измениться: LEFT OUTER после обмена — RIGHT OUTER).

  • Правило 2: ассоциативность JOIN.
    (A JOIN B) JOIN C → A JOIN (B JOIN C).
    В StarRocks различают случаи для INNER/CROSS JOIN и для SEMI JOIN.

2. Greedy

В жадном алгоритме StarRocks использует multi-sequence greedy и улучшение: на каждом «слое» сохраняются десять лучших (локально) планов, которые развиваются дальше; в конце получаем десять лучших жадных планов. Это повышает вероятность попасть в глобальный оптимум, но не гарантирует его.

3. Модель стоимости (Cost Model)

После генерации N планов выбирается лучший по стоимости:

  • Формула стоимости JOIN:

    Join Cost: CPU (Row(L) + Row(R)) + Memory Row(R)

    где Row(L), Row(R) — числа выходных строк левого/правого входов. Учитываются CPU-затраты и память под хеш-таблицу правой таблицы в Hash Join. В StarRocks применяется детальная схема расчёта числа выходных строк.

С учётом различной сложности и времени работы алгоритмов в StarRocks приняты следующие ограничения:

  • До 4 таблиц — полный перебор (Exhaustive).

  • 4–10 таблиц — используются Left-Deep, Greedy и динамическое программирование (DP), которые дают 1, 10 и 1 план соответственно; затем по коммутативности исследуются дополнительные планы.

  • Более 10 таблиц — базируемся на 11 планах (10 Greedy + 1 Left-Deep).

  • При отсутствии статистики cost-ориентированные Greedy и DP работают плохо; используется только 1 план Left-Deep как база для перестановки.

#04 Планирование распределённых JOIN

Перейдём к особенностям распределённого выполнения в StarRocks.

1. Параллельное выполнение MPP

Архитектура StarRocks — MPP. Для простого запроса A JOIN B:

  1. По информации о распределении таблиц A и B данные читаются на разных машинах.

  2. По предикатам соединения данные A и B перераспределяются (shuffle) на одну и ту же группу машин.

  3. На узлах выполняется однопроцессорный JOIN и формируется результат.

В вычислениях участвует не одна машина (чтение A, чтение B, узлы JOIN различны), есть сетевая передача и обмены. Критично снижать сетевые издержки и разумно декомпозировать/распределять план выполнения, максимально используя параллелизм.

2. Оптимизация распределённых JOIN

Для запроса:

SELECT * FROM A JOIN B ON A.a = B.b;

StarRocks может сгенерировать пять базовых планов:

  • Shuffle Join: A и B перераспределяются (shuffle) по ключу соединения на одну и ту же группу машин, затем выполняется JOIN.

  • Broadcast Join: данные B целиком рассылаются (broadcast) на машины A, JOIN выполняется на них. Экономится shuffle A, но B передаётся целиком — подходит, когда B маленькая.

  • Bucket Shuffle Join: оптимизация Broadcast — B перераспределяется по бакетам в соответствии с распределением A и пересылается на машины A; глобально данные B передаются в единственном экземпляре — трафика существенно меньше, чем при Broadcast. Ограничение: ключ соединения должен соответствовать распределению A.

  • Colocate Join: если при создании таблиц A и B они помещены в одну colocate‑группу (Colocate Group), т.е. распределены идентично, и ключ соединения этому соответствует — JOIN выполняется локально без shuffle.

  • Replicate Join: экспериментальная возможность — если на каждой машине, где обрабатывается A, есть полная копия B, JOIN выполняется локально. Требования строгие (число реплик B ≈ числу машин кластера), практическая полезность ограничена.

StarRocks пытается строить все пять планов для каждого JOIN, но из-за семантики отдельных типов возможны не все. Например, для CROSS JOIN — только Broadcast Join.

3. Вывод распределённых JOIN

Распределённые планы строятся через выведение свойств распределения (Distribution Property). Для Shuffle Join оператор JOIN сверху вниз требует от A и B свойства Shuffle по ключу.

Если оператор Scan не может удовлетворить требование, через принудительное обеспечение (enforce) вставляется оператор Shuffle. При генерации физического плана этот оператор «переводится» в оператор Exchange (Exchange) — он осуществляет сетевую пересылку и обмен данными.

Другие типы распределённых JOIN строятся аналогично — JOIN сверху задаёт требуемые свойства, которые проталкиваются вниз.

SELECT * FROM A JOIN B ON A.a = B.b;

4. Сложные распределённые JOIN

В реальных запросах обычно соединяются 3–4 таблицы и более. StarRocks генерирует более сложные комбинации распределённых планов на базе перечисленных типов, например:

SELECT * 
FROM A JOIN B ON A.a = B.b 
         JOIN C ON A.a = C.c;

Комбинируя Shuffle Join и Broadcast Join, получаем ряд вариантов; добавляя Colocate Join и Bucket Shuffle Join — ещё больше комбинаций. Принципы вывода те же: Distribution Property передаются вниз по дереву плана, что и даёт различные комбинации.

Подробнее см. «Обзор кода оптимизатора StarRocks» — https://forum.mirrorship.cn/t/topic/18561

5. Global Runtime Filter

Помимо построения распределённых планов, StarRocks использует особенности выполнения Hash Join для конструирования Global Runtime Filter (глобальный runtime‑фильтр). Процесс Hash Join:

  1. StarRocks сначала извлекает полный набор данных правой таблицы.

  2. По правой таблице строится хеш-таблица.

  3. Затем извлекаются данные левой таблицы.

  4. По хеш-таблице выполняется сопоставление по ключу соединения.

  5. Формируется результат JOIN.

Global Runtime Filter работает между шагами 2 и 3: после получения данных правой таблицы на их основе строится «рантаймовый» фильтр, который отправляется на операторы Scan левой таблицы до их чтения. Это помогает отфильтровать данные заранее и сократить вход JOIN. Поддерживаются: Min/Max, IN‑предикат и Bloom Filter.

#05 Итоги

Мы рассмотрели практики и исследования StarRocks по оптимизации JOIN. Все оптимизации соответствуют заявленным принципам. При самостоятельной оптимизации SQL ориентируйтесь на следующее:

  1. Разные типы JOIN имеют разную производительность; по возможности используйте более быстрые. Примерный порядок: SEMI/ANTI JOIN > INNER JOIN > OUTER JOIN > FULL OUTER JOIN > CROSS JOIN.

  2. В Hash Join лучше строить хеш-таблицу по маленькой таблице — это существенно эффективнее.

  3. В многотабличном JOIN сначала выполняйте соединения с высокой селективностью, чтобы резко снизить последующие издержки.

  4. Максимально сокращайте объём данных, участвующих в JOIN.

  5. Максимально уменьшайте сетевые издержки распределённых JOIN.

Дальнейшие направления развития StarRocks:

  • Поддержка большего числа реализаций JOIN и более «умный» выбор оператора по контексту.

  • С учётом особенностей StarRocks — поддержка дополнительных специализированных алгоритмов Join Reorder.

  • Улучшение оценки стоимости (cost): внедрение новых алгоритмов и структур данных для повышения точности.

  • Поддержка дополнительных вариантов диспетчеризации/планирования с целью оптимизации сетевых издержек.

Комментарии (0)