Dynamic Partition Pruning Deep Dive

Partition Pruning이란 무엇인가!

Pruning은 가지치기 라는 사전적 의미를 가지고 있습니다. 즉 Partition Pruning은 불필요한 파티션을 가지치기 하고 필요한 파티션 내의 데이터만 읽는다는 의미로 이해할 수 있습니다.

Filter push down

Dynamic Partition Pruning의 경우 Filter push down을 이용하는 최적화 기법입니다.
SELECT * FROM table WHERE year = 2025

위와 같이 특정 테이블에서 year 파티션이 2025년인 값을 선택하는 쿼리가 있다고 가정해보겠습니다.

기본적으로는 Basic data-flow와 같은 흐름으로 데이터를 읽게 되는데, 스캔을 먼저 진행한 후 WHERE 절의 필터를 적용하는 방식입니다. 하지만 순서를 바꿔 필터를 먼저 적용한 뒤 스캔을 하게 되면 읽어야 할 데이터의 양을 줄일 수 있습니다. 필터를 스캔보다 먼저 적용하는 것이 Filter push down의 개념입니다.

그럼 다시 Partition Pruning으로 돌아가봅시다.

이 그림과 같이 파티션이 여러개 있는 상황에서 필터를 적용한 뒤 스캔을 하는 것을 Partition Pruning이라고 이해할 수 있습니다.

SELCT * FROM Sales JOIN Date WHERE Date.day_of_week = 'mon'
과 같은 쿼리를 수행했다고 가정해봅시다. 아래 그림처럼 Date 테이블은 크기가 작기 때문에 필터가 적용되어도 성능 향상을 기대하기 어려울 수 있습니다.

이런 경우 양쪽 테이블을 모두 스캔한 뒤 조인하고 그 후에 필터를 적용하는 방안도 있지만, 이 경우 역시 좋은 대안이 될 수 없습니다.
그래서 아래처럼 Dynamic Pruning을 적용하여 최적화를 할 수 있습니다. 작은 테이블의 Filter 결과를 큰 테이블인 Sales 테이블에 적용하여 풀 스캔을 방지합니다.

Spark에서는 쿼리를 수행하게 되면 아래와 같이 Logical plan 단계, Physical Plan 단계를 거쳐 실제 작업이 수행되게 됩니다.

이때 Logical optimization 단계와 Physical Planning 단계에서 어떻게 Pruning이 적용되고 최적화를 할 수 있는지 살펴보겠습니다.

(1) Logical Optimization 단계

왼쪽은 파티션이 된 Fact 테이블, 오른쪽은 파티셔닝이 되지 않은 Dimension 테이블입니다. 파티셔닝된 Fact 테이블의 풀 스캐닝을 막기 위해 Dimension 테이블에서의 필터를 동일하게 Fact 테이블에도 적용한 뒤 조인을 수행합니다.

(2) Physical optimization 단계에서 일어날 수 있는 최적화

(1)에서의 한가지 최적화 포인트는 바로 Dimension 테이블을 스캔한 뒤 필터링하는 과정이 중복된다는 것 입니다. 즉 Dimension 테이블 스캔 후 필터링 1회, Fact 테이블의 풀 스캐닝을 막기 위해 Dimension 테이블에 또 다시 쿼리를 하게 되는 횟수 1회까지 동일 내용의 쿼리가 중복된다는 의미입니다.

이 부분을 이해하기 위해 Broadcast hash join 기법을 이해해야합니다.

Broadcast hash join의 경우 Driver를 통해 전체 노드에 Dataframe을 올린 뒤 Shuffle이 일어나지 않도록 하여 조인을 수행합니다.

Broadcast hash join 아이디어에 착안해서, Dimension 테이블에서의 필터 결과 값인 hash table을 왼쪽 테이블의 필터 값으로 적용합니다.

코드를 살펴보면 reuseEnabled, hasBenefit의 조건을 통해 DPP를 적용할 지 결정합니다. Pruning이 적용될 경우 DynamicPruningSubquery 필터가 생성됩니다.

  private def insertPredicate(
      pruningKey: Expression,
      pruningPlan: LogicalPlan,
      filteringKey: Expression,
      filteringPlan: LogicalPlan,
      joinKeys: Seq[Expression],
      partScan: LogicalPlan): LogicalPlan = {
    val reuseEnabled = conf.exchangeReuseEnabled
    val index = joinKeys.indexOf(filteringKey)
    lazy val hasBenefit = pruningHasBenefit(pruningKey, partScan, filteringKey, filteringPlan)
    if (reuseEnabled || hasBenefit) {
      // insert a DynamicPruning wrapper to identify the subquery during query planning
      Filter(
        DynamicPruningSubquery(
          pruningKey,
          filteringPlan,
          joinKeys,
          index,
          conf.dynamicPartitionPruningReuseBroadcastOnly || !hasBenefit),
        pruningPlan)
    } else {
      // abort dynamic partition pruning
      pruningPlan
    }
  }

hasBenefit의 경우 Helper 함수인 pruningHasBenefit을 통해 결정되며 예상된 pruning의 크기가 overhead보다 클 경우 True를 반환합니다.