동적 필터링

동적 필터링 (Dynamic filtering)

동적 필터링 최적화는 선택적인(selective) 조인에서 조인 조건에 걸러질 데이터를 읽지 않도록 함으로써 쿼리 성능을 크게 개선해 주는 기능이에요.

출처: 문서

본문

동적 필터링은 조인 조건에 의해 필터링될 데이터를 읽지 않도록 하여, 선택적인 조인(selective join)을 포함한 쿼리의 성능을 크게 향상시켜요. 팩트 테이블(fact table) store_sales와 필터링된 디멘전 테이블(dimension table) date_dim을 조인하는 흔한 패턴을 생각해 볼게요:

SELECT count(*)
FROM store_sales
JOIN date_dim ON store_sales.ss_sold_date_sk = date_dim.d_date_sk
WHERE d_following_holiday='Y' AND d_year = 2000;

동적 필터링이 없으면 Trino는 디멘전 테이블에 대한 조건(predicate)을 date_dim 테이블 스캔에 내려보내고, store_sales에는 쿼리에 필터가 없으므로 팩트 테이블의 모든 데이터를 스캔해요. 조인 기준이 매우 선택적이기 때문에 조인 연산자는 프로브(probe) 쪽 행 대부분을 결국 버리게 되죠.

동적 필터링을 활성화하면 Trino는 조인 오른쪽에 있는 처리된 디멘전 테이블에서 조인 조건의 후보 값을 수집해요. 브로드캐스트 조인(broadcast join)의 경우, 이렇게 수집해 만든 런타임 조건은 같은 워커에서 실행되는 조인 왼쪽의 로컬 테이블 스캔으로 내려보내져요.

추가로 이 런타임 조건은 네트워크를 통해 코디네이터에도 전달되어, 코디네이터가 테이블 스캔 스플릿(split)을 열거할 때도 동적 필터링을 수행할 수 있게 해요.

예를 들어 Hive 커넥터의 경우 동적 필터를 사용해 조인 기준에 맞지 않는 파티션 로딩을 건너뜁니다. 이것을 동적 파티션 프루닝(dynamic partition pruning) 이라고 불러요.

동적 필터 수집을 마치면 코디네이터는 파티셔닝 조인(partitioned join)을 위해 이 필터를 네트워크를 통해 워커 노드에도 배포합니다. 이렇게 되면 파티셔닝 조인의 동적 필터를 해당 조인 왼쪽의 테이블 스캔으로 내려보낼 수 있어요.

동적 필터링 최적화로 얻는 이점은 다음과 같아요:

• 전체 쿼리 성능 개선
• Trino와 데이터 소스 사이의 네트워크 트래픽 감소
• 원격 데이터 소스의 부하 감소

동적 필터링은 기본적으로 활성화되어 있어요. enable-dynamic-filtering 설정 속성이나 enable_dynamic_filtering 세션 속성을 false로 설정하면 비활성화할 수 있어요.

동적 필터의 푸시다운 지원은 커넥터마다, 그리고 관련된 기본 데이터베이스/스토리지 시스템에 따라 달라요. 동적 필터링을 지원하는 커넥터 문서에 자세한 내용이 있는데, 예를 들어 Hive 커넥터나 Memory 커넥터 문서를 참고하세요.

분석과 확인 (Analysis and confirmation)

동적 필터링은 여러 요인에 따라 움직여요:

• Trino에서 특정 조인 연산에 대한 플래너(planner)의 동적 필터링 지원. 현재는 =, <, <=, >, >= 또는 IS NOT DISTINCT FROM 조인 조건을 쓰는 내부 조인(inner join)·오른쪽 조인(right join), 그리고 IN 조건을 쓰는 세미 조인(semi-join)을 지원해요.
• 커넥터가 런타임에 테이블 스캔으로 내려간 동적 필터를 활용하는 지원. 예를 들어 Hive 커넥터는 동적 필터를 ORC·Parquet 리더로 내려보내 스트라이프(stripe)·로우그룹(row-group) 프루닝을 수행할 수 있어요.
• 커넥터가 스플릿 열거 단계에서 동적 필터를 활용하는 지원.
• 조인의 오른쪽(빌드, build) 크기.

플래너가 특정 쿼리 계획에 동적 필터를 추가하는지 분석하려면 쿼리의 EXPLAIN 계획을 자세히 살펴보면 돼요. 예를 들어 위 쿼리의 explain 계획은 다음 문으로 얻을 수 있어요:

EXPLAIN
SELECT count(*)
FROM store_sales
JOIN date_dim ON store_sales.ss_sold_date_sk = date_dim.d_date_sk
WHERE d_following_holiday='Y' AND d_year = 2000;

...
Fragment 1 [SOURCE]
    Output layout: [count_3]
    Output partitioning: SINGLE []
    Aggregate(PARTIAL)
    │   Layout: [count_3:bigint]
    │   count_3 := count(*)
    └─ InnerJoin[("ss_sold_date_sk" = "d_date_sk")][$hashvalue, $hashvalue_4]
       │   Layout: []
       │   Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
       │   Distribution: REPLICATED
       │   dynamicFilterAssignments = {d_date_sk -> #df_370}
       ├─ ScanFilterProject[table = hive:default:store_sales, grouped = false, filterPredicate = true, dynamicFilters = {"ss_sold_date_sk" = #df_370}]
       │      Layout: [ss_sold_date_sk:bigint, $hashvalue:bigint]
       │      Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}/{rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}/{rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
       │      $hashvalue := combine_hash(bigint '0', COALESCE("$operator$hash_code"("ss_sold_date_sk"), 0))
       │      ss_sold_date_sk := ss_sold_date_sk:bigint:REGULAR
       └─ LocalExchange[HASH][$hashvalue_4] ("d_date_sk")
          │   Layout: [d_date_sk:bigint, $hashvalue_4:bigint]
          │   Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
          └─ RemoteSource[2]
                 Layout: [d_date_sk:bigint, $hashvalue_5:bigint]

Fragment 2 [SOURCE]
    Output layout: [d_date_sk, $hashvalue_6]
    Output partitioning: BROADCAST []
    ScanFilterProject[table = hive:default:date_dim, grouped = false, filterPredicate = (("d_following_holiday" = CAST('Y' AS char(1))) AND ("d_year" = 2000))]
        Layout: [d_date_sk:bigint, $hashvalue_6:bigint]
        Estimates: {rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}/{rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}/{rows: 0 (0B), cpu: 0, memory: 0B, network: 0B}
        $hashvalue_6 := combine_hash(bigint '0', COALESCE("$operator$hash_code"("d_date_sk"), 0))
        d_following_holiday := d_following_holiday:char(1):REGULAR
        d_date_sk := d_date_sk:bigint:REGULAR
        d_year := d_year:int:REGULAR

계획에서 dynamicFilterAssignments = {d_date_sk -> #df_370}와 스캔의 dynamicFilters = {"ss_sold_date_sk" = #df_370}가 보이면 이 조인에 동적 필터가 걸린 것을 확인할 수 있어요.

실행 후에는 쿼리 통계에서 동적 필터 관련 지표를 확인할 수 있습니다:

"dynamicFiltersStats" : {
      "dynamicFilterDomainStats" : [ {
        "dynamicFilterId" : "df_370",
        "simplifiedDomain" : "[ SortedRangeSet[type=bigint, ranges=3, {[2451546], ..., [2451905]}] ]",
        "collectionDuration" : "2.34s"
      } ],
      "lazyDynamicFilters" : 1,
      "replicatedDynamicFilters" : 1,
      "totalDynamicFilters" : 1,
      "dynamicFiltersCompleted" : 1
}

스캔 연산자 통계에서도 실제로 스플릿이 얼마나 걸러졌는지 볼 수 있어요:

"operatorType" : "ScanFilterAndProjectOperator",
"totalDrivers" : 1,
"addInputCalls" : 762,
"addInputWall" : "0.00ns",
"addInputCpu" : "0.00ns",
"physicalInputDataSize" : "0B",
"physicalInputPositions" : 28800991,
"inputPositions" : 28800991,
"dynamicFilterSplitsProcessed" : 1,

...
 └─ InnerJoin[("ss_sold_date_sk" = "d_date_sk")][$hashvalue, $hashvalue_4]
    │   Layout: []
    │   Estimates: {rows: 11859 (0B), cpu: 8.84M, memory: 3.19kB, network: 3.19kB}
    │   CPU: 78.00ms (30.00%), Scheduled: 295.00ms (47.05%), Output: 296 rows (0B)
    │   Left (probe) Input avg.: 120527.00 rows, Input std.dev.: 0.00%
    │   Right (build) Input avg.: 0.19 rows, Input std.dev.: 208.17%
    │   Distribution: REPLICATED
    │   dynamicFilterAssignments = {d_date_sk -> #df_370}
    ├─ ScanFilterProject[table = hive:default:store_sales, grouped = false, filterPredicate = true, dynamicFilters = {"ss_sold_date_sk" = #df_370}]
    │      Layout: [ss_sold_date_sk:bigint, $hashvalue:bigint]
    │      Estimates: {rows: 120527 (2.03MB), cpu: 1017.64k, memory: 0B, network: 0B}/{rows: 120527 (2.03MB), cpu: 1.99M, memory: 0B, network: 0B}/{rows: 120527 (2.03MB), cpu: 4.02M, memory: 0B, network: 0B}
    │      CPU: 49.00ms (18.85%), Scheduled: 123.00ms (19.62%), Output: 120527 rows (2.07MB)
    │      Input avg.: 120527.00 rows, Input std.dev.: 0.00%
    │      $hashvalue := combine_hash(bigint '0', COALESCE("$operator$hash_code"("ss_sold_date_sk"), 0))
    │      ss_sold_date_sk := ss_sold_date_sk:bigint:REGULAR
    │      Input: 120527 rows (1.03MB), Filtered: 0.00%
    │      Dynamic filters:
    │          - df_370, [ SortedRangeSet[type=bigint, ranges=3, {[2451546], ..., [2451905]}] ], collection time=2.34s
    |
...

이렇게 동적 필터가 조인 오른쪽(빌드) 쪽에서 수집되어 왼쪽(프로브) 스캔에 적용됨으로써 입력이 크게 줄어드는 것을 볼 수 있어요.

더 알아보기 (Learn more)

커넥터별 동적 필터링 세부 내용이 궁금하다면 Hive 커넥터 문서를 이어서 읽어 보세요.