Проталкивание вычислений#

Trino может проталкивать обработку запросов или частей запросов в подключенный источник данных. Это означает, что конкретный предикат, функция агрегации или другая операция передается в нижележащую базу данных или систему хранения для выполнения.

Результаты такого pushdown могут давать следующие преимущества:

Улучшение общей производительности запросов
Снижение сетевого трафика между Trino и источником данных
Снижение нагрузки на удаленный источник данных

Эти преимущества часто приводят к существенному снижению затрат.

Поддержка pushdown зависит от конкретного коннектора и соответствующей нижележащей базы данных или системы хранения.

Проталкивание предикатов#

Predicate pushdown оптимизирует фильтрацию строк. Он использует выведенный фильтр, обычно получаемый из условия в предложении WHERE, чтобы исключить лишние строки. Обработка проталкивается в источник данных через коннектор, а затем выполняется самим источником данных.

Если predicate pushdown для конкретного выражения успешен, план EXPLAIN для запроса не включает операцию ScanFilterProject для этого выражения.

Проталкивание проекций#

Projection pushdown оптимизирует фильтрацию по столбцам. Он использует столбцы, указанные в предложении SELECT, и других частях запроса, чтобы ограничить доступ только к этим столбцам. Обработка проталкивается в источник данных через коннектор, и затем источник данных читает и возвращает только необходимые столбцы.

Если projection pushdown успешен, план EXPLAIN для запроса обращается только к релевантным столбцам в Layout операции TableScan.

Проталкивание разыменования#

Projection pushdown и dereference pushdown ограничивают доступ к релевантным столбцам, но dereference pushdown более избирателен. Он ограничивает доступ так, чтобы читать только указанные поля внутри верхнеуровневого или вложенного типа данных ROW.

Например, рассмотрим таблицу в коннекторе Hive, содержащую столбец типа ROW с несколькими полями. Если запрос обращается только к одному полю, dereference pushdown позволяет файловому reader читать только это одно поле внутри строки. То же применимо к полям строки, вложенной в верхнеуровневую строку. Это может давать значительную экономию объема данных, считываемых из системы хранения.

Проталкивание агрегаций#

Aggregation pushdown может выполняться при соблюдении следующих условий:

Если aggregation pushdown в целом поддерживается коннектором.
Если pushdown конкретной функции или функций поддерживается коннектором.
Если структура запроса позволяет выполнить pushdown.

Проверить, выполняется ли pushdown для конкретного запроса, можно через EXPLAIN plan запроса. Если агрегатная функция успешно протолкнута в коннектор, план explain не показывает оператор Aggregate. План explain показывает только операции, выполняемые Trino.

В качестве примера: мы загрузили набор данных TPC-H в базу PostgreSQL, а затем выполнили запрос через коннектор PostgreSQL:

SELECT regionkey, count(*)
FROM nation
GROUP BY regionkey;

Получить explain plan можно, добавив перед этим запросом EXPLAIN:

EXPLAIN
SELECT regionkey, count(*)
FROM nation
GROUP BY regionkey;

Explain plan для этого запроса не показывает оператор Aggregate с функцией count, так как эта операция теперь выполняется коннектором. Функцию count(*) можно увидеть как часть оператора PostgreSQL TableScan. Это показывает, что pushdown выполнен успешно.

Fragment 0 [SINGLE]
    Output layout: [regionkey_0, _generated_1]
    Output partitioning: SINGLE []
    Output[regionkey, _col1]
    │   Layout: [regionkey_0:bigint, _generated_1:bigint]
    │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?}
    │   regionkey := regionkey_0
    │   _col1 := _generated_1
    └─ RemoteSource[1]
            Layout: [regionkey_0:bigint, _generated_1:bigint]

Fragment 1 [SOURCE]
    Output layout: [regionkey_0, _generated_1]
    Output partitioning: SINGLE []
    TableScan[postgresql:tpch.nation tpch.nation columns=[regionkey:bigint:int8, count(*):_generated_1:bigint:bigint] groupingSets=[[regionkey:bigint:int8]], gro
        Layout: [regionkey_0:bigint, _generated_1:bigint]
        Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
        _generated_1 := count(*):_generated_1:bigint:bigint
        regionkey_0 := regionkey:bigint:int8

Push-down может не выполниться по ряду причин:

добавление условия в запрос
использование другой агрегатной функции, которую нельзя протолкнуть в коннектор
использование коннектора без поддержки pushdown для конкретной функции

В результате план explain показывает операцию Aggregate, выполняемую Trino. Это явный признак того, что pushdown в удаленный источник данных не выполняется, и вместо этого агрегацию выполняет Trino.

Fragment 0 [SINGLE]
    Output layout: [regionkey, count]
    Output partitioning: SINGLE []
    Output[regionkey, _col1]
    │   Layout: [regionkey:bigint, count:bigint]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
    │   _col1 := count
    └─ RemoteSource[1]
           Layout: [regionkey:bigint, count:bigint]

Fragment 1 [HASH]
    Output layout: [regionkey, count]
    Output partitioning: SINGLE []
    Aggregate(FINAL)[regionkey]
    │   Layout: [regionkey:bigint, count:bigint]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
    │   count := count("count_0")
    └─ LocalExchange[HASH][$hashvalue] ("regionkey")
       │   Layout: [regionkey:bigint, count_0:bigint, $hashvalue:bigint]
       │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
       └─ RemoteSource[2]
              Layout: [regionkey:bigint, count_0:bigint, $hashvalue_1:bigint]

Fragment 2 [SOURCE]
    Output layout: [regionkey, count_0, $hashvalue_2]
    Output partitioning: HASH [regionkey][$hashvalue_2]
    Project[]
    │   Layout: [regionkey:bigint, count_0:bigint, $hashvalue_2:bigint]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
    │   $hashvalue_2 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("regionkey"), 0))
    └─ Aggregate(PARTIAL)[regionkey]
       │   Layout: [regionkey:bigint, count_0:bigint]
       │   count_0 := count(*)
       └─ TableScan[tpch:nation:sf0.01, grouped = false]
              Layout: [regionkey:bigint]
              Estimates: {rows: 25 (225B), cpu: 225, memory: 0B, network: 0B}
              regionkey := tpch:regionkey

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

Aggregation pushdown не поддерживает ряд более сложных выражений:

сложные операции группировки, такие как ROLLUP, CUBE или GROUPING SETS
выражения внутри вызова агрегатной функции: sum(a * b)
приведения типов: sum(integer_column)
aggregations with ordering
aggregations with filter

Проталкивание join#

Join pushdown позволяет коннектору делегировать операцию соединения таблиц нижележащему источнику данных. Это может приводить к росту производительности и позволяет Trino выполнять оставшуюся обработку запроса на меньшем объеме данных.

Особенности поддерживаемого pushdown для join различаются для каждого источника данных и, соответственно, для каждого коннектора.

Однако есть общие условия, которые должны выполняться, чтобы join можно было протолкнуть:

все предикаты, являющиеся частью join, должны поддерживать pushdown
таблицы в join должны быть из одного каталога

Проверить, выполняется ли pushdown для конкретного join, можно по плану EXPLAIN запроса. План explain не показывает оператор Join, если join протолкнут в источник данных через коннектор:

EXPLAIN SELECT c.custkey, o.orderkey
FROM orders o JOIN customer c ON c.custkey = o.custkey;

Следующий план получен при запросе данных TPC-H из базы PostgreSQL через коннектор PostgreSQL. В нем нет оператора Join, что является результатом успешного pushdown для join.

Fragment 0 [SINGLE]
    Output layout: [custkey, orderkey]
    Output partitioning: SINGLE []
    Output[custkey, orderkey]
    │   Layout: [custkey:bigint, orderkey:bigint]
    │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?}
    └─ RemoteSource[1]
           Layout: [orderkey:bigint, custkey:bigint]

Fragment 1 [SOURCE]
    Output layout: [orderkey, custkey]
    Output partitioning: SINGLE []
    TableScan[postgres:Query[SELECT l."orderkey" AS "orderkey_0", l."custkey" AS "custkey_1", r."custkey" AS "custkey_2" FROM (SELECT "orderkey", "custkey" FROM "tpch"."orders") l INNER JOIN (SELECT "custkey" FROM "tpch"."customer") r O
        Layout: [orderkey:bigint, custkey:bigint]
        Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
        orderkey := orderkey_0:bigint:int8
        custkey := custkey_1:bigint:int8

Обычно pushdown для join полезен. Но при pushdown join количество строк также может увеличиться относительно размера входных данных join. Это может повлиять на производительность.

Проталкивание limit#

Предложение LIMIT или FETCH FIRST уменьшает количество записей, возвращаемых оператором. Проталкивание limit позволяет коннектору проталкивать обработку таких запросов с неотсортированными записями в нижележащий источник данных.

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

Запросы включают части вида LIMIT N или FETCH FIRST N ROWS.

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

Проталкивание Top-N#

Комбинация Предложение LIMIT или FETCH FIRST и Предложение ORDER BY формирует небольшой набор записей для возврата из большого отсортированного набора данных. Она опирается на порядок, чтобы определить, какие записи нужно вернуть, и поэтому оптимизируется иначе, чем Проталкивание limit.

Проталкивание для такого запроса называется Top-N pushdown, поскольку операция возвращает верхние N строк. Он позволяет коннектору проталкивать обработку таких запросов в нижележащий источник данных и тем самым существенно снижает объем данных, передаваемых и обрабатываемых в Trino.

Запросы включают части вида ORDER BY ... LIMIT N или ORDER BY ... FETCH FIRST N ROWS.

Реализация и поддержка зависят от коннектора, поскольку разные источники данных поддерживают разный синтаксис SQL и обработку.

Например, в следующем разделе приведены два запроса, по которым можно понять, как определить поведение Top-N pushdown.

Сначала конкретный пример запроса Top-N поверх базы PostgreSQL:

SELECT id, name
FROM postgresql.public.company
ORDER BY id
LIMIT 5;

Получить explain plan можно, добавив перед этим запросом EXPLAIN:

EXPLAIN SELECT id, name
FROM postgresql.public.company
ORDER BY id
LIMIT 5;

Fragment 0 [SINGLE]
    Output layout: [id, name]
    Output partitioning: SINGLE []
    Stage Execution Strategy: UNGROUPED_EXECUTION
    Output[id, name]
    │   Layout: [id:integer, name:varchar]
    │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: ?}
    └─ RemoteSource[1]
           Layout: [id:integer, name:varchar]

Fragment 1 [SOURCE]
    Output layout: [id, name]
    Output partitioning: SINGLE []
    Stage Execution Strategy: UNGROUPED_EXECUTION
    TableScan[postgresql:public.company public.company sortOrder=[id:integer:int4 ASC NULLS LAST] limit=5, grouped = false]
        Layout: [id:integer, name:varchar]
        Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
        name := name:varchar:text
        id := id:integer:int4

Во втором примере показан запрос Top-N на коннекторе tpch, который не поддерживает функциональность Top-N pushdown:

SELECT custkey, name
FROM tpch.sf1.customer
ORDER BY custkey
LIMIT 5;

Соответствующий план запроса:

Fragment 0 [SINGLE]
    Output layout: [custkey, name]
    Output partitioning: SINGLE []
    Stage Execution Strategy: UNGROUPED_EXECUTION
    Output[custkey, name]
    │   Layout: [custkey:bigint, name:varchar(25)]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
    └─ TopN[5 by (custkey ASC NULLS LAST)]
       │   Layout: [custkey:bigint, name:varchar(25)]
       └─ LocalExchange[SINGLE] ()
          │   Layout: [custkey:bigint, name:varchar(25)]
          │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: ?}
          └─ RemoteSource[1]
                 Layout: [custkey:bigint, name:varchar(25)]

Fragment 1 [SOURCE]
    Output layout: [custkey, name]
    Output partitioning: SINGLE []
    Stage Execution Strategy: UNGROUPED_EXECUTION
    TopNPartial[5 by (custkey ASC NULLS LAST)]
    │   Layout: [custkey:bigint, name:varchar(25)]
    └─ TableScan[tpch:customer:sf1.0, grouped = false]
           Layout: [custkey:bigint, name:varchar(25)]
           Estimates: {rows: 150000 (4.58MB), cpu: 4.58M, memory: 0B, network: 0B}
           custkey := tpch:custkey
           name := tpch:name

В предыдущем плане запроса операция Top-N TopN[5 by (custkey ASC NULLS LAST)] выполняется в Fragment 0 движком Trino, а не исходной базой данных.

Обратите внимание: по сравнению с запросом, выполненным через коннектор tpch, в explain plan запроса через коннектор postgresql отсутствует ссылка на операцию TopN[5 by (id ASC NULLS LAST)] в Fragment 0. Отсутствие оператора Trino TopN в Fragment 0 плана запроса демонстрирует, что запрос получает выгоду от оптимизации Top-N pushdown.