Operators: scan, filter, join и другие

Driver — это цепочка операторов. Мы много раз упоминали операторы — scan, filter, join — но не разбирали их по отдельности. Этот урок закрывает пробел: что такое оператор как исполняющая единица, какие бывают основные операторы, как они связаны с PlanNode логического плана и почему один JoinNode превращается в пару операторов.

Operator — самый мелкий уровень таксономии: одна конкретная операция над данными.

Оператор как звено цепочки

Operator (оператор) — компонент, который потребляет, трансформирует и производит данные. Это атомарная операция: один оператор делает одну вещь — читает, или фильтрует, или соединяет, или агрегирует.

Операторы не работают поодиночке. Они выстроены в цепочку внутри driver: выход одного оператора — вход следующего. Driver прогоняет через эту цепочку Page за Page. Каждый оператор берёт входящий Page, делает с ним свою операцию, выдаёт исходящий Page следующему оператору.

У оператора есть общий контракт: ему дают входной Page, у него забирают выходной Page. За этим единым контрактом — разная начинка: scan читает из коннектора, join держит хэш-таблицу. Но снаружи для driver все операторы одинаковы — это и позволяет собирать из них произвольные цепочки.

Операторы и PlanNode: исполняющие двойники

Откуда операторы берутся. Вспомните логический план из модуля про жизненный цикл — дерево PlanNode: TableScan, FilterNode, JoinNode, AggregationNode. Операторы — это исполняющие двойники этих узлов плана.

PlanNode описывает операцию абстрактно: «здесь фильтр по такому-то предикату». Оператор — это та же операция, воплощённая в исполняемый код, работающий с Page внутри driver. При переходе от плана к исполнению узлы плана разворачиваются в операторы.

Связь не всегда один-к-одному. Часть PlanNode превращается в один оператор (FilterNode -> filter-оператор), а часть — в несколько операторов (JoinNode, как увидим, — в пару). Но общий принцип таков: логический план описывает операции, исполнение их выполняет операторами.

Основные операторы

Пройдём главные операторы — это словарь, который нужен для чтения EXPLAIN ANALYZE.

TableScan (scan) — первый оператор source-цепочки. Читает данные из коннектора через PageSource, выдаёт Page вниз по цепочке. У него нет входного оператора — данные он берёт из источника, по своему split. Это лист цепочки исполнения.

Filter — применяет предикат к каждой строке входного Page, оставляет только прошедшие. Выходной Page содержит подмножество строк входного. Соответствует FilterNode и условию WHERE.

Project (проекция) — вычисляет выражения и формирует выходной набор столбцов: выбирает нужные столбцы, считает производные (price * quantity), применяет функции. Соответствует ProjectNode и списку SELECT.

Aggregation — вычисляет агрегаты: count, sum, avg, min, max, при наличии GROUP BY — по группам. Соответствует AggregationNode. Важная деталь — про неё ниже.

Join — соединяет два потока данных по условию. Соответствует JoinNode. Тоже с особенностью — про неё ниже.

Exchange — обменивается данными с другими стадиями: принимает данные, пришедшие по сети от задач другой стадии, или отправляет свои. Это оператор-граница между фрагментами; ему посвящён отдельный урок про exchange-модель.

Join: два оператора из одного узла

JoinNode логического плана при исполнении превращается не в один оператор, а в пару. Причина — в самом алгоритме хэш-join, которым Trino соединяет данные.

Хэш-join работает в две фазы. Сначала build-фаза: одна из таблиц (build-side, обычно меньшая) читается целиком, и из неё строится хэш-таблица в памяти — индекс по join-ключу. Затем probe-фаза: вторая таблица (probe-side, обычно большая) читается потоково, и для каждой её строки в хэш-таблице ищется совпадение по ключу.

Эти две фазы исполняются двумя разными операторами:

• HashBuilder (оператор build-стороны) — потребляет build-side и строит из неё хэш-таблицу.
• HashJoin / Lookup (оператор probe-стороны) — потребляет probe-side и для каждой строки делает поиск в готовой хэш-таблице.

Между фазами есть зависимость: probe не может начать поиск, пока build не достроил хэш-таблицу — искать-то не в чем. Поэтому build-сторона — это блокирующая часть join: пока меньшая таблица не прочитана целиком и хэш-таблица не готова, probe ждёт. Это объясняет, почему так важно, какая таблица назначена build-стороной: build-side целиком держится в памяти как хэш-таблица. Если build-стороной по ошибке окажется большая таблица — память кончится. Выбор build-стороны и тип распределения join (broadcast или partitioned) — решение оптимизатора, и оно разбирается в модуле про cost-based optimizer.

Partial и final: два оператора агрегации

С агрегацией — похожая история, и мы её уже встречали в модуле про жизненный цикл. Один AggregationNode в распределённом плане часто превращается в два оператора агрегации: partial и final.

• Aggregation(PARTIAL) — частичная агрегация. Выполняется в source-стадии, прямо у чтения данных: каждая задача сворачивает свои строки в частичные агрегаты по группам, локально.
• Aggregation(FINAL) — финальная агрегация. Выполняется после сетевого обмена (HASH-распределение по ключу группировки): частичные агрегаты со всех задач суммируются в окончательный результат по каждой группе.

Смысл — резко сократить объём данных, идущих по сети: между стадиями летят уже свёрнутые частичные агрегаты, а не исходные строки. Это та же двухфазная агрегация, что и в EXPLAIN (TYPE DISTRIBUTED), но теперь видно, что каждая фаза — отдельный оператор в своей цепочке driver.

Операторы в общей картине

Сведём операторный уровень с остальной таксономией. Distributed planning разрезал план на стадии. Каждая стадия разворачивается в задачи на воркерах. Задача разворачивает фрагмент в pipeline и запускает драйверы. Driver — это цепочка операторов, и каждый оператор — исполняющий двойник какого-то PlanNode. Данные текут по цепочке операторов Page за Page.

Операторы — это, в буквальном смысле, то, что делает работу: scan читает байты из источника, filter проверяет предикаты, hash builder строит хэш-таблицу, hash join ищет совпадения. Всё остальное в таксономии — stage, task, driver — это структура, организующая операторы в распределённое параллельное исполнение. А сама обработка данных происходит в операторах.

Попробуй сам

Операторы видны в EXPLAIN ANALYZE — каждый показан отдельной строкой со своей статистикой:

1. Выполните EXPLAIN ANALYZE SELECT custkey FROM tpch.sf1.customer WHERE acctbal > 5000. Найдите в выводе операторы scan, filter, project. Посмотрите статистику каждого.
2. Выполните EXPLAIN ANALYZE для join tpch.sf1.nation и tpch.sf1.customer. Найдите два оператора join — постройку хэш-таблицы и probe. Определите, какая таблица стала build-стороной.
3. Выполните EXPLAIN ANALYZE для запроса с GROUP BY. Найдите оба оператора агрегации — PARTIAL и FINAL — и убедитесь, что они в разных фрагментах.
4. Для каждого найденного оператора назовите, какому PlanNode логического плана он соответствует.
5. Сформулируйте письменно, почему один JoinNode даёт два оператора и почему build-сторона блокирующая.