푸시다운

푸시다운 (Pushdown)

Trino는 쿼리 또는 쿼리의 일부 처리를 연결된 데이터 소스로 푸시다운(push down)할 수 있어요. 즉 특정 프레디킷, 집계 함수 또는 다른 연산을 기반 데이터베이스나 스토리지 시스템으로 넘겨 그쪽에서 처리하게 해요.

출처: 문서

본문

Trino는 쿼리 또는 쿼리의 일부 처리를 연결된 데이터 소스로 푸시다운할 수 있어요. 특정 프레디킷, 집계 함수, 또는 다른 연산을 기반 데이터베이스나 스토리지 시스템에 넘겨 처리하게 해요. 이 푸시다운의 결과로 다음 이점이 있을 수 있어요.

• 전반적인 쿼리 성능 향상
• Trino와 데이터 소스 사이의 네트워크 트래픽 감소
• 원격 데이터 소스의 부하 감소

이러한 이점은 종종 상당한 비용 절감으로 이어져요. 푸시다운 지원은 각 커넥터와 관련 기반 데이터베이스 또는 스토리지 시스템에 특화돼 있어요.

프레디킷 푸시다운 (Predicate pushdown)

프레디킷 푸시다운은 행 기반 필터링을 최적화해요. 보통 WHERE 절의 조건에서 비롯된 추론된 필터를 사용해 불필요한 행을 생략해요. 처리는 커넥터가 데이터 소스로 푸시다운한 뒤 데이터 소스에서 처리돼요.

특정 절에 대한 프레디킷 푸시다운이 성공하면, 해당 쿼리의 EXPLAIN 계획에 그 절에 대한 ScanFilterProject 연산이 포함되지 않아요.

프로젝션 푸시다운 (Projection pushdown)

프로젝션 푸시다운은 컬럼 기반 필터링을 최적화해요. SELECT 절과 쿼리의 다른 부분에 지정된 컬럼을 사용해 이 컬럼들에 대한 접근을 제한해요. 처리는 커넥터가 데이터 소스로 푸시다운하고, 데이터 소스는 필요한 컬럼만 읽어 반환해요.

프로젝션 푸시다운이 성공하면, 해당 쿼리의 EXPLAIN 계획은 TableScan 연산의 Layout에서 관련 컬럼에만 접근해요.

역참조 푸시다운 (Dereference pushdown)

프로젝션 푸시다운과 역참조 푸시다운은 관련 컬럼 접근을 제한하지만, 역참조 푸시다운은 더 선택적이에요. 최상위 또는 중첩 ROW 데이터 타입 안의 지정된 필드만 읽도록 접근을 제한해요.

예를 들어 여러 필드를 가진 ROW 타입 컬럼이 있는 Hive 커넥터의 테이블을 생각해 봐요. 쿼리가 하나의 필드에만 접근하면, 역참조 푸시다운은 파일 리더가 행 안의 그 단일 필드만 읽게 해요. 최상위 행 안에 중첩된 행의 필드에도 마찬가지가 적용돼요. 이는 스토리지 시스템에서 읽는 데이터 양을 크게 줄일 수 있어요.

집계 푸시다운 (Aggregation pushdown)

집계 푸시다운은 다음 조건이 충족되면 일어날 수 있어요.

• 커넥터가 집계 푸시다운을 일반적으로 지원하는 경우
• 커넥터가 특정 함수 또는 함수들의 푸시다운을 지원하는 경우
• 쿼리 구조가 푸시다운을 허용하는 경우

특정 쿼리에 대해 푸시다운이 수행되는지 쿼리의 EXPLAIN 계획을 보고 확인할 수 있어요. 집계 함수가 커넥터로 성공적으로 푸시다운되면, explain 계획에 그 Aggregate 연산자가 표시되지 않아요. explain 계획은 Trino가 수행하는 연산만 보여 줘요.

예를 들어 TPC-H 데이터셋을 PostgreSQL 데이터베이스에 로드하고 PostgreSQL 커넥터로 쿼리했다고 해요.

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

앞선 쿼리 앞에 EXPLAIN을 붙이면 explain 계획을 얻을 수 있어요.

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

이 쿼리의 explain 계획에는 count 함수가 있는 Aggregate 연산자가 표시되지 않아요. 이 연산은 이제 커넥터가 수행하기 때문이에요. count(*) 함수가 PostgreSQL TableScan 연산자의 일부로 보이는 걸 확인할 수 있어요. 이는 푸시다운이 성공했다는 신호예요.

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

푸시다운을 막을 수 있는 여러 요인이 있어요.

• 쿼리에 조건 추가
• 커넥터로 푸시다운할 수 없는 다른 집계 함수 사용
• 특정 함수에 대한 푸시다운을 지원하지 않는 커넥터 사용

그 결과 explain 계획에 Trino가 수행하는 Aggregate 연산이 표시돼요. 이것은 원격 데이터 소스로의 푸시다운이 수행되지 않고 대신 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

제한 사항 (Limitations)

집계 푸시다운은 여러 더 복잡한 문장을 지원하지 않아요.

• ROLLUP, CUBE, GROUPING SETS 같은 복잡한 그룹화 연산
• 집계 함수 호출 안의 표현식: sum(a * b)
• 강제 변환(coercion): sum(integer_column)
• 정렬이 있는 집계(aggregations with ordering)
• 필터가 있는 집계

조인 푸시다운 (Join pushdown)

조인 푸시다운은 커넥터가 테이블 조인 연산을 기반 데이터 소스에 위임하게 해줘요. 이로 인해 성능상 이점이 있을 수 있고, Trino가 더 적은 양의 데이터에 대해 나머지 쿼리 처리를 수행하게 해줘요.

지원되는 테이블 조인 푸시다운의 세부 사항은 각 데이터 소스, 따라서 각 커넥터에 따라 달라져요.

그러나 조인이 푸시다운되기 위한 일반적인 조건이 몇 가지 있어요.

• 조인의 일부인 모든 프레디킷이 푸시다운 가능해야 함
• 조인의 테이블이 같은 카탈로그에서 와야 함

특정 조인에 대해 푸시다운이 수행되는지 쿼리의 EXPLAIN 계획으로 확인할 수 있어요. 조인이 커넥터에 의해 데이터 소스로 푸시다운되면 explain 계획에 Join 연산자가 표시되지 않아요.

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

다음 계획은 PostgreSQL 데이터베이스의 TPC-H 데이터를 쿼리한 PostgreSQL 커넥터의 결과예요. 성공적인 조인 푸시다운의 결과로 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

보통 조인을 푸시다운하는 것이 유리해요. 조인을 푸시다운하면 조인 입력의 크기 대비 행 수가 늘어날 수도 있어요. 이는 성능에 영향을 줄 수 있어요.

Limit 푸시다운 (Limit pushdown)

LIMIT 또는 FETCH FIRST 절은 문장에 대해 반환되는 레코드 수를 줄여요. Limit 푸시다운은 커넥터가 정렬되지 않은 레코드에 대한 이러한 쿼리의 처리를 기반 데이터 소스로 푸시하게 해줘요.

이 절의 푸시다운은 쿼리 성능을 향상시키고 데이터 소스에서 Trino로 전송되는 데이터 양을 크게 줄일 수 있어요.

쿼리에는 LIMIT N 또는 FETCH FIRST N ROWS 같은 섹션이 포함돼요. 구현과 지원은 서로 다른 데이터 소스가 다양한 기능을 가지므로 커넥터별로 특화돼 있어요.

Top-N 푸시다운 (Top-N pushdown)

LIMIT 또는 FETCH FIRST 절과 ORDER BY 절의 결합은 큰 정렬된 데이터셋에서 반환할 작은 레코드 집합을 만든다. 어떤 레코드를 반환할지 결정하기 위해 정렬 순서에 의존하므로, Limit 푸시다운과 비교해 최적화하는 방식이 꽤 달라요.

이런 쿼리에 대한 푸시다운은 상위 N개 행을 반환하는 연산이므로 Top-N 푸시다운이라고 불러요. 커넥터가 이러한 쿼리의 처리를 기반 데이터 소스로 푸시하게 해주고, 따라서 Trino로 전송되고 Trino에서 처리되는 데이터 양을 크게 줄여요.

쿼리에는 ORDER BY ... LIMIT N 또는 ORDER BY ... FETCH FIRST N ROWS 같은 섹션이 포함돼요. 구현과 지원은 서로 다른 데이터 소스가 서로 다른 SQL 문법과 처리를 지원하므로 커넥터별로 특화돼 있어요.

예를 들어 다음 섹션에서 Top-N 푸시다운 동작을 식별하는 방법을 배울 수 있는 두 쿼리를 찾을 수 있어요.

첫째, PostgreSQL 데이터베이스 위의 Top-N 푸시다운 쿼리의 구체적인 예:

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

앞선 쿼리 앞에 EXPLAIN을 붙이면 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 쿼리 예:

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)]이 소스 데이터베이스가 아니라 Trino에 의해 Fragment 0에서 적용되고 있어요.

참고로, tpch 커넥터 위에서 실행된 쿼리와 비교해 postgresql 커넥터 위에서 실행된 쿼리의 explain 계획은 Fragment 0에 TopN[5 by (id ASC NULLS LAST)] 연산에 대한 참조가 없어요. 쿼리 계획의 Fragment 0에서 Trino Top-N 연산자가 없다는 것은 그 쿼리가 Top-N 푸시다운 최적화의 이점을 누린다는 것을 보여 줘요.

더 알아보기 (Learn more)

푸시다운으로 데이터 소스에 처리를 위임해 성능을 높이는 방법을 배웠어요. 이어서 적응형 계획 최적화(Adaptive plan optimizations)를 살펴보면 좋아요.