EXPLAIN ANALYZE VERBOSE: чтение статистики операторов

Web UI говорит, какая стадия горячая. EXPLAIN ANALYZE говорит почему. Это самый точный инструмент диагностики Trino: он реально исполняет запрос и возвращает распределённый план, где у каждого оператора проставлены фактические числа — сколько строк прошло, сколько CPU потрачено, сколько wall-времени, сколько памяти. Вы держите в руках не оценку оптимизатора, а измеренную реальность исполнения.

Этот урок учит читать вывод EXPLAIN ANALYZE сверху вниз, отделять estimate от actual, находить bottleneck-оператор и распознавать характерные промахи оптимизатора. Без этого навыка тюнинг Trino — гадание; с ним — инженерная работа.

Семейство EXPLAIN

В Trino несколько форм EXPLAIN, и они отвечают на разные вопросы.

EXPLAIN без исполнения хорош, чтобы проверить решения оптимизатора заранее — порядок join, тип распределения, сработал ли pushdown. EXPLAIN ANALYZE — чтобы понять, что произошло на самом деле. VERBOSE добавляет распределения метрик по задачам (min/max/avg) и метрики отдельных операторов, например статистику reader’а или dynamic filtering.

Как устроен вывод

EXPLAIN ANALYZE печатает план как дерево фрагментов. Каждый Fragment — это стадия распределённого плана; внутри фрагмента — дерево операторов. Читать дерево нужно сверху вниз по тексту, но понимать снизу вверх по данным: листовые операторы (TableScan) читают данные, корневой оператор фрагмента отдаёт результат выше.

Рассмотрим реальный сокращённый вывод для join fact- и dimension-таблиц:

Trino version: 481
Fragment 1 [HASH]
    CPU: 4.21m, Scheduled: 9.84m, Blocked 12.30m (Input: 6.10m, Output: 0.00ns), Input: 59986052 rows (1.61GB)
    Output layout: [returnflag, count, sum]
    Output partitioning: SINGLE []
    Aggregate[type = FINAL, keys = [returnflag]]
    │   CPU: 1.18m (28.04%), Output: 3 rows (54B)
    │   Input avg.: 1875376.62 rows, Input std.dev.: 4.21%
    └─ LocalExchange[partitioning = HASH, arguments = [returnflag]]
       │   CPU: 0.92m (21.85%), Output: 96 rows
       └─ RemoteSource[sourceFragmentIds = [2]]
              CPU: 2.11m (50.11%), Output: 59986052 rows (1.61GB)
              Input avg.: 1874564.12 rows, Input std.dev.: 3.98%

Fragment 2 [SOURCE]
    CPU: 47.81m, Scheduled: 71.20m, Blocked 0.00ns, Input: 59986052 rows (1.61GB)
    Output layout: [returnflag, extendedprice]
    Output partitioning: HASH [returnflag]
    Aggregate[type = PARTIAL, keys = [returnflag]]
    │   CPU: 8.04m (16.81%), Output: 96 rows
    └─ InnerJoin[criteria = ("orderkey" = "orderkey_0"), distribution = REPLICATED]
       │   CPU: 21.93m (45.86%), Output: 59986052 rows (1.61GB)
       │   Distribution: REPLICATED
       │   dynamicFilterAssignments = {orderkey_0 -> #df_440}
       ├─ ScanFilterProject[table = tpch:lineitem:sf10, filterPredicate = true, dynamicFilters = {"orderkey" = #df_440}]
       │      CPU: 16.32m (34.13%), Output: 59986052 rows (1.61GB)
       │      Input: 59986052 rows (0B), Filtered: 0.00%
       │      orderkey := tpch:orderkey
       │      extendedprice := tpch:extendedprice
       └─ LocalExchange[partitioning = SINGLE]
          └─ RemoteSource[sourceFragmentIds = [3]]
                 CPU: 1.52m (3.18%), Output: 2278602 rows (39.1MB)

Fragment 3 [SOURCE]
    CPU: 6.40m, Scheduled: 9.90m, Input: 15000000 rows (404MB)
    Output layout: [orderkey_0]
    Output partitioning: BROADCAST []
    ScanFilterProject[table = tpch:orders:sf10, filterPredicate = ("orderdate" >= DATE '1995-01-01')]
           CPU: 6.40m (13.38%), Output: 2278602 rows (39.1MB)
           Input: 15000000 rows (0B), Filtered: 84.81%
           orderkey_0 := tpch:orderkey
           orderdate := tpch:orderdate

Разберём ключевые поля.

На уровне фрагмента строка CPU / Scheduled / Blocked — суммы по всем задачам фрагмента. CPU — чистое процессорное время. Scheduled (wall) — сколько реального времени задачи были на исполнении. Blocked — сколько они ждали (данных на вход, места в выходных буферах). [HASH], [SOURCE], [SINGLE], [BROADCAST] — тип распределения фрагмента.

На уровне оператора проценты CPU: 21.93m (45.86%) показывают долю оператора в CPU своего фрагмента — это и есть карта горячих точек. Output: N rows — фактический выход. У ScanFilterProject есть Input и Filtered: NN% — сколько строк прочитано и какая доля отсеяна предикатом.

Поля cardinality Input avg. и Input std.dev. (видны полнее в VERBOSE) — распределение нагрузки по задачам или драйверам. Низкий std.dev. (единицы процентов) — нагрузка равномерная. Высокий (десятки процентов) — data skew.

Поиск bottleneck: алгоритм

Чтение EXPLAIN ANALYZE ради тюнинга — это процедура из четырёх шагов.

В нашем примере горячий фрагмент — Fragment 2 (CPU 47.81m против 6.40m и 4.21m). Внутри него горячий оператор — InnerJoin (45.86% CPU). Это ожидаемо: join 60 млн строк lineitem с отфильтрованными orders. Важный позитивный сигнал — dynamicFilters = {"orderkey" = #df_440} в ScanFilterProject: dynamic filtering работает, build-сторона (orders) собрала рантайм-предикат по orderkey. Distribution REPLICATED (то же, что BROADCAST) означает: orders отфильтровались до 2.3 млн строк (39 MB), это влезло под порог broadcast, и Trino разослал их на все воркеры вместо дорогого repartition. Здесь оптимизатор отработал правильно.

Сравнение estimate и actual — самый ценный шаг. Trino печатает рядом с операторами оценку (Estimates: {rows: ...}); если оценка говорит «10 тысяч строк», а actual показывает 60 миллионов, оптимизатор строил план на ложных данных и почти наверняка ошибся с порядком join и типом распределения.

Почему фрагменты пронумерованы именно так

Стоит отдельно понять нумерацию фрагментов, иначе вывод читается неверно. EXPLAIN ANALYZE печатает фрагменты сверху вниз, и Fragment 0 — это всегда корень: финальный шаг, отдающий результат клиенту. Фрагменты с большими номерами — ближе к источникам данных. То есть по тексту вы читаете сверху вниз, а данные текут снизу вверх: листовые фрагменты ([SOURCE]) читают из коннекторов, корневой ([SINGLE]) собирает итог.

Связь фрагментов — через операторы RemoteSource и Output partitioning. Когда во фрагменте встречается RemoteSource[sourceFragmentIds = [2]], это значит: данные сюда приходят из Fragment 2 по сети. А строка Output partitioning у фрагмента говорит, как он раздаёт свой результат наверх: SINGLE — всё в одну задачу, HASH [key] — репартиционирование по хэшу ключа, BROADCAST — копирование на все ноды. Прослеживая RemoteSource и Output partitioning, вы реконструируете, как стадии соединены в дерево и где между ними происходят дорогие сетевые обмены.

Это объясняет и тип фрагмента в квадратных скобках. [SOURCE] исполняется там, где доступны входные сплиты, — это data locality. [HASH] исполняется на фиксированном числе нод, и данные распределены по хэшу. [SINGLE] — на одной ноде. Тип фрагмента — не косметика: он определяет, сколько данных и как именно гоняется по сети, а значит прямо влияет на то, какие фрагменты окажутся горячими.

Характерные промахи оптимизатора

Несколько паттернов, которые EXPLAIN ANALYZE выдаёт сразу.

Перекос (skew). В VERBOSE-выводе у горячего оператора Input std.dev. в десятки процентов. Значит одна задача обрабатывает кратно больше строк, чем соседи: стадия не закончится, пока не доедет самая нагруженная задача. Причина — неравномерное распределение значений ключа партиционирования (например, NULL или одно доминирующее значение в ключе join).

Не сработавший pushdown. ScanFilterProject показывает Input: 50000000 rows, Filtered: 0.00%, а отдельным узлом выше стоит FilterProject, который режет 95% строк. Значит фильтр не протолкнулся в коннектор: 50 млн строк прочитаны с диска зря, фильтрация идёт в движке. Сравните с EXPLAIN (TYPE IO), чтобы увидеть, что планировалось прочитать.

Промах cardinality и плохой join order. Estimate под join сильно расходится с actual. Из-за этого Trino выбрал PARTITIONED, где хватило бы BROADCAST, или поставил большую таблицу на build-сторону. Лечится через ANALYZE — сбор статистики.

Дорогой обмен. У фрагмента строка Blocked (особенно Input) сопоставима с Scheduled или больше — стадия половину времени простаивает в ожидании данных по сети. Сигнал, что между стадиями гоняется слишком много данных: репартиционируется большая таблица, которую можно было бы транслировать, или фильтр применяется слишком поздно.

Попробуй сам

На встроенном коннекторе tpch воспроизведите цикл диагностики.

1. Снимите план до сбора статистики:

EXPLAIN ANALYZE
SELECT n.name, sum(l.extendedprice) AS revenue
FROM tpch.sf10.lineitem l
JOIN tpch.sf10.orders o ON l.orderkey = o.orderkey
JOIN tpch.sf10.customer c ON o.custkey = c.custkey
JOIN tpch.sf10.nation n ON c.nationkey = n.nationkey
GROUP BY n.name
ORDER BY revenue DESC;

2. Найдите горячий фрагмент по строке CPU, внутри него — горячий оператор по проценту CPU.
3. Найдите в выводе строки dynamicFilters на сканах. По каким ключам Trino собрал рантайм-фильтры?
4. Перезапустите с EXPLAIN ANALYZE VERBOSE и найдите оператор с наибольшим Input std.dev. — есть ли перекос?
5. Сравните EXPLAIN (TYPE DISTRIBUTED) того же запроса (без ANALYZE) с фактическим выводом: совпали ли estimates оптимизатора с реальными числами строк?

Цель — пройти все четыре шага алгоритма на живом запросе и научиться отличать здоровый план от плана с промахом.