Pipeline-модель исполнения

В прошлых модулях мы разобрали, как DuckDB хранит и сжимает данные. Теперь — как он их исполняет, и почему делает это на всех ядрах процессора сразу. Параллелизм в DuckDB не прикручен сбоку — он встроен в саму модель исполнения. Но чтобы понять параллелизм, нужно сперва понять структуру, на которую он ложится: разбиение плана запроса на pipelines. Этот вводный урок модуля про pipeline-модель — про source, операторы и sink, и про pipeline breaker, который заставляет исполнение делать паузу.

От дерева операторов к исполнению

Когда вы отправляете SQL-запрос, DuckDB проходит через парсер, биндер, оптимизатор и строит физический план — дерево операторов. Каждый оператор делает одну операцию: сканирование таблицы, фильтрацию, hash join, агрегацию, сортировку. Дерево задаёт, какой оператор куда подаёт данные.

Но дерево — это структура, описание «что с чем связано». Само по себе оно не говорит, как именно гонять данные через эти операторы. Здесь возможны два подхода.

Классический подход старых СУБД — pull-модель, она же модель Volcano. Корневой оператор «тянет» строку: запрашивает её у нижестоящего, тот у своего нижестоящего, и так до сканирования. Данные движутся снизу вверх по запросу сверху, по одной строке за раз. Это просто, но медленно: вызов на каждую строку, плохая работа с кэшем, никакой векторизации.

DuckDB использует другой подход — push-модель.

Trino: tasks и drivers — pipeline-исполнение в кластере Данные «проталкиваются» снизу вверх: сканирование читает порцию данных и проталкивает её в вышестоящий оператор, тот обрабатывает и проталкивает дальше. И толкается не строка, а целый батч — DataChunk из примерно 2048 значений (это связано с векторизованным движком, разобранным в отдельном модуле). Push-модель и векторизация — естественная пара, и именно push-модель удобно ложится на pipelines.

Что такое pipeline

DuckDB не исполняет дерево операторов как единое целое. Он разрезает его на pipelines — цепочки операторов, через которые данные текут непрерывным потоком без остановки.

У каждого pipeline три части. Source — источник, начало pipeline: оператор, который порождает данные. Чаще всего это сканирование таблицы, но source может быть и чем-то другим. Операторы в середине — фильтры, проекции, преобразования: они принимают батч, обрабатывают его и сразу передают дальше, не накапливая. Sink — приёмник, конец pipeline: оператор, который данные поглощает — записывает результат, строит хеш-таблицу, накапливает состояние агрегации.

Ключевое свойство pipeline — данные текут через него потоком. Source выдал батч; батч прошёл через цепочку операторов середины, каждый что-то с ним сделал и тут же протолкнул дальше; батч дошёл до sink, который его поглотил. Никто в середине не ждёт, пока придут все данные, — каждый батч проходит цепочку насквозь и независимо от других батчей. Pipeline — это конвейер: данные не лежат, а движутся.

Pipeline breaker: где конвейер вынужден остановиться

Если бы любой запрос укладывался в один pipeline, всё было бы просто. Но не укладывается, и причина — некоторые операторы не могут работать в режиме «батч пришёл — батч ушёл». Им нужно увидеть все входные данные целиком, прежде чем выдать хоть один результат. Такие операторы называются pipeline breakers — они «ломают» конвейер, разрезая его на части.

Самый ясный пример — сортировка. ORDER BY не может выдать первую строку результата, пока не увидел все входные строки: вдруг самая последняя строка входа окажется наименьшей и должна идти первой. Сортировка обязана сначала поглотить весь вход целиком, и лишь потом начать выдавать отсортированный результат. Поток прерывается: накопление всего входа — это пауза.

Второй классический pipeline breaker — построение хеш-таблицы в hash join. Перед тем как соединять, join должен полностью построить хеш-таблицу по одной из сторон (build side). Нельзя пробовать соединять, пока хеш-таблица не достроена, — иначе часть совпадений будет потеряна. Построение хеш-таблицы целиком — пауза.

Группирующая агрегация — GROUP BY — устроена похоже: окончательные значения групп не готовы, пока не обработана последняя входная строка, ведь она может попасть в любую группу и изменить её итог.

Pipeline breaker естественно становится границей между pipelines. С одной стороны от него — pipeline, который наполняет breaker (его sink — это и есть breaker, поглощающий вход). С другой стороны — pipeline, который начинается от breaker (его source — это breaker, теперь уже отдающий накопленный результат). Один и тот же оператор-breaker служит sink для одного pipeline и source для следующего.

Как запрос распадается на pipelines

Соберём картину на конкретном запросе. Пусть нужно соединить две таблицы и отсортировать результат:

SELECT o.id, c.name
FROM orders o
JOIN customers c ON o.customer_id = c.id
ORDER BY o.id;

Здесь два pipeline breaker: построение хеш-таблицы для join и сортировка. Они разрезают исполнение на три pipeline.

Pipeline 1 — наполнение хеш-таблицы join. Source — сканирование customers (меньшей таблицы, её удобно положить в хеш-таблицу). Sink — построение хеш-таблицы. Этот pipeline целиком прогоняет customers и строит по нему хеш-таблицу. Пока он не завершится, соединять нельзя.

Pipeline 2 — само соединение и наполнение сортировки. Source — сканирование orders. В середине — оператор probe: каждый батч orders проверяется против уже готовой хеш-таблицы, находятся совпадения. Sink — накопление строк для сортировки. Этот pipeline соединяет данные и складывает результат в сортировку.

Pipeline 3 — выдача отсортированного результата. Source — сортировка, которая теперь, накопив весь вход, отдаёт строки в порядке ORDER BY. Sink — возврат результата клиенту.

Обратите внимание на порядок. Pipeline 1 должен полностью завершиться раньше, чем начнётся pipeline 2: probe в pipeline 2 нуждается в готовой хеш-таблице. Pipeline 2 должен полностью завершиться раньше pipeline 3: сортировка не отдаст ничего, пока не накопила весь вход. Pipeline breaker задаёт не только границу, но и зависимость — порядок, в котором pipelines выполняются.

Почему pipelines важны для параллелизма

Зачем вообще резать запрос на pipelines — ведь можно было бы исполнять дерево как есть? Главная причина — параллелизм, и весь оставшийся модуль будет про это.

Внутри одного pipeline данные текут потоком, и каждый батч проходит цепочку операторов независимо от других батчей. А раз батчи независимы, их можно обрабатывать одновременно: пока поток A гонит через pipeline один батч, поток B гонит через тот же pipeline другой батч. Pipeline — это естественная единица, внутри которой разворачивается параллелизм: несколько потоков прокачивают через один и тот же pipeline разные порции данных.

Pipeline breakers при этом служат точками синхронизации. Внутри pipeline потоки работают независимо и почти не пересекаются. Но на границе — на pipeline breaker — нужна координация: следующий pipeline не может стартовать, пока предыдущий не завершён всеми потоками. Breaker — это барьер, на котором потоки сходятся, прежде чем двинуться к следующему pipeline.

Так возникает общая картина исполнения в DuckDB. Запрос — это последовательность pipelines, разделённых breakers. Внутри каждого pipeline — массовый параллелизм, потоки независимо прокачивают батчи. На границах pipelines — синхронизация, потоки сходятся. Как именно данные внутри pipeline раздаются потокам — это механизм morsel-driven parallelism, и ему посвящён следующий урок. Здесь важно усвоить каркас: pipeline-модель — это та структура, на которую параллелизм DuckDB и ложится.

Попробуй сам

Научитесь видеть pipelines в планах запросов.

1. Создайте две таблицы: CREATE TABLE orders AS SELECT range AS id, range % 1000 AS customer_id FROM range(1000000); и CREATE TABLE customers AS SELECT range AS id, ('cust_' || range) AS name FROM range(1000);.
2. Выполните EXPLAIN SELECT count(*) FROM orders WHERE customer_id > 500;. Это простой запрос — фильтр и агрегация. Сколько здесь pipeline breakers? Агрегация count(*) — это breaker; найдите её в плане.
3. Выполните EXPLAIN SELECT o.id, c.name FROM orders o JOIN customers c ON o.customer_id = c.id;. Найдите в плане оператор HASH_JOIN. Build-сторона join — это pipeline breaker; какую таблицу движок выбрал для построения хеш-таблицы?
4. Выполните EXPLAIN SELECT o.id, c.name FROM orders o JOIN customers c ON o.customer_id = c.id ORDER BY o.id;. Теперь breakers два — join и сортировка. Найдите оператор ORDER_BY в плане.
5. Запустите EXPLAIN ANALYZE для запроса с join и сортировкой. В выводе у операторов есть время и число строк. Прикиньте по дереву, на сколько pipeline разрезается этот запрос и в каком порядке они должны выполняться.

Этот эксперимент учит читать план как набор pipelines: каждый pipeline breaker — это граница, и число breakers определяет, на сколько частей разбито исполнение.