그룹핑 알고리즘

그룹핑 알고리즘 (Grouping Algorithm)

Pinot의 그룹핑 알고리즘(GROUP BY 쿼리 처리 시 사용)이 서버의 메모리 부족을 막기 위해 결과를 트리밍하는 휴리스틱에 대해 배우는 가이드예요.

출처: Grouping Algorithm

본문

SSE (단일 스테이지 엔진)

SSE 쿼리 실행의 다양한 단계에서의 group by 결과 근사화

세그먼트 내부

세그먼트 내에서 행을 그룹핑할 때 Pinot는 세그먼트당 최대 numGroupsLimit개의 그룹을 유지합니다. 이 값은 기본적으로 100,000이며 pinot.server.query.executor.num.groups.limit 속성으로 구성할 수 있습니다.

세그먼트의 그룹 수가 이 값에 도달하면 추가 그룹은 무시되고 반환되는 결과가 완전히 정확하지 않을 수 있습니다. 값에 도달하면 numGroupsLimitReached 속성이 쿼리 응답에서 true로 설정됩니다. 멀티 스테이지 쿼리에서는 errorOnNumGroupsLimit=true를 설정해 한도에 도달했을 때 부분 결과를 반환하는 대신 쿼리를 실패시킬 수 있습니다.

꼬리 그룹 트리밍

내부 세그먼트 그룹이 계산된 뒤 Pinot 쿼리 엔진은 선택적으로 꼬리 그룹을 트리밍합니다. 꼬리 그룹은 쿼리에 사용된 ORDER BY 절을 기준으로 순위가 낮은 그룹입니다.

세그먼트 그룹 트리밍이 활성화되면 쿼리 엔진은 꼬리 그룹을 트리밍하고 max(minSegmentGroupTrimSize, 5 * LIMIT)만 유지합니다. 여기서 LIMIT는 쿼리가 반환하는 최대 레코드 수입니다(보통 LIMIT 절로 설정). Pinot는 꼬리 그룹을 트리밍할 때 결과의 정확성을 보장하기 위해 최소 5 * LIMIT개 그룹을 유지합니다. 트리밍은 정렬과 limit이 지정된 경우에만 수행됩니다.

이 값은 다음 옵션을 전달해 쿼리별로 덮어쓸 수 있습니다:

SELECT * 
FROM ...
OPTION(minSegmentGroupTrimSize=value)

교차 세그먼트

세그먼트 내 그룹핑이 끝나면 Pinot는 세그먼트 결과를 병합하고 꼬리 그룹을 트리밍해, 그룹이 더 많아지면 max(minServerGroupTrimSize, 5 * LIMIT)개 그룹을 유지합니다.

minServerGroupTrimSize는 기본적으로 5,000이며 pinot.server.query.executor.min.server.group.trim.size 속성을 구성해 조정할 수 있습니다. 이 속성을 -1로 설정하면 교차 세그먼트 트리밍을 비활성화할 수 있습니다.

교차 세그먼트 트리밍이 활성화되면 서버는 결과를 브로커로 보내기 전에 꼬리 그룹을 트리밍합니다. 세그먼트별 결과 병합 시 메모리 사용량을 줄이기 위해 그룹 수가 trimThreshold에 도달할 때도 꼬리 그룹을 트리밍합니다.

trimThreshold는 서버가 메모리 부족에 빠지는 것을 보호하기 위해 쿼리당 서버에 허용되는 그룹 수의 상한입니다. 트리밍이 너무 잦아지지 않도록 실제 트리밍 크기는 trimThreshold / 2로 제한됩니다. 위 식과 결합하면 쿼리의 실제 트리밍 크기는 min(max(minServerGroupTrimSize, 5 * LIMIT), trimThreshold / 2)로 계산됩니다.

이 설정은 기본적으로 1,000,000이며 pinot.server.query.executor.groupby.trim.threshold 속성으로 조정할 수 있습니다.

임계값이 높을수록 트리밍량은 줄지만 힙 메모리를 더 많이 소비합니다. 임계값이 1,000,000,000을 초과하면 서버는 결과를 브로커에 반환하기 전에 그룹을 한 번만 트리밍합니다.

이 값은 다음 옵션을 전달해 쿼리별로 덮어쓸 수 있습니다:

SELECT * 
FROM ...
OPTION(groupTrimThreshold=value)

브로커에서

브로커가 여러 서버가 반환한 그룹의 최종 병합을 수행할 때 또 다른 트리밍 단계가 있습니다. 꼬리 그룹이 트리밍되고 max(minBrokerGroupTrimSize, 5 * LIMIT)개 그룹이 유지됩니다.

minBrokerGroupTrimSize의 기본값은 5000입니다. 이는 pinot.broker.min.group.trim.size 속성으로 조정할 수 있습니다.

GROUP BY 동작

Pinot는 정의하지 않으면 기본 LIMIT 10을 설정하며 이는 SSE GROUP BY 쿼리에도 적용됩니다. 따라서 SSE에서 limit을 지정하지 않으면 Pinot는 10개 그룹을 반환합니다. MSE는 그런 암묵적 10행 결과 제한을 적용하지 않습니다. 그럼에도 결과 크기가 중요하면 LIMIT를 명시적으로 설정하세요. Querying Pinot도 참고하세요.

Pinot는 메모리 사용량을 줄이고 쿼리 성능을 높이기 위해 ORDER BY 절을 기준으로 꼬리 그룹을 트리밍합니다. 대부분의 경우 결과가 충분히 좋은 근사치가 되도록 최소 5 * LIMIT개 그룹을 유지합니다. 구성 가능한 최소 트리밍 크기로 유지되는 그룹을 늘려 정확성을 높일 수 있지만 메모리 오버헤드가 더 커집니다.

ORDER BY가 없으면 SSE group-by는 순위를 정의하지 않습니다. 기본적으로 결과 테이블은 요청된 결과 크기에 도달하면 새 그룹 키 수용을 멈추므로 처리 순서에 따라 어떤 키가 살아남는지가 달라질 수 있습니다. GROUP BY ... LIMIT N을 top-N 쿼리로 취급하지 마세요.

결정적 부분 집합이 필요하면 accurateGroupByWithoutOrderBy=true를 설정하세요: Pinot는 서버·브로커 집계 중 사전순으로 가장 작은 그룹 키를 유지합니다. 이것은 집계 값으로 그룹에 순위를 매기지 않으며 numGroupsLimit이 버린 키를 복구하지 못합니다. top-N 결과에는 ORDER BY를 사용하세요. Query options를 참고하세요.

HAVING 동작

쿼리에 HAVING 절이 있으면 이미 꼬리 그룹이 트리밍된 병합 GROUP BY 결과에 적용됩니다. HAVING 절이 ORDER BY 순서와 반대라면 조건에 맞는 그룹이 이미 트리밍되어 반환되지 않을 수 있습니다. 예:

SELECT SUM(colA) 
FROM myTable 
GROUP BY colB 
HAVING SUM(colA) < 100 
ORDER BY SUM(colA) DESC 
LIMIT 10

이런 경우에는 최소 트리밍 크기를 늘려 더 많은 그룹을 유지하세요.

예제

단순한 키 집계 쿼리:

SELECT i, j, count(*) AS cnt
FROM tab
GROUP BY i, j
ORDER BY i ASC, j ASC
LIMIT 3;

간소화된 실행 계획(트리밍이 어디서 일어나는지 보여줌):

BROKER_REDUCE(sort:[i, j],limit:10) <- sort and trim groups to minBrokerGroupTrimSize
  COMBINE_GROUP_BY <- sort and trim groups to minServerGroupTrimSize
    PLAN_START
      GROUP_BY <- limit to numGroupsLimit, then sort and trim to minSegmentGroupTrimSize
        PROJECT(i, j)
          DOC_ID_SET
            FILTER_MATCH_ENTIRE_SEGMENT

간결함을 위해 위 계획은 실제 남는 그룹 수가 min(trim_value, 5*limit)인 것을 언급하지 않았습니다.

MSE (멀티 스테이지 엔진)

SSE와 비교해 MSE는 유사한 알고리즘을 쓰지만 주목할 차이점이 있습니다:

  • MSE는 쿼리 결과 수를 (10으로) 암묵적으로 제한하지 않습니다.
  • MSE는 교차 세그먼트 데이터 집계 시 그룹 수를 제한하지 않습니다.
  • MSE는 Pinot가 LIMIT + OFFSET을 리프/최종 집계로 안전하게 push할 수 있을 때, 집계가 없는 DISTINCT 및 집계가 없는 GROUP BY 쿼리를 LIMIT으로 기본 트리밍합니다. 이 기본 활성 경로는 여전히 페이지네이션을 존중하며 /*+ aggOptions(is_enable_group_trim='false') */로 쿼리별 비활성화할 수 있습니다.
  • 집계 함수를 계산하는 집계 쿼리는 is_enable_group_trim으로 옵트인하지 않는 한 기본적으로 그룹 트리밍을 비활성으로 유지합니다.
  • MSE는 브로커에서 결과를 집계하지 않고 최종 집계 처리를 서버로 push합니다.

기본 MSE 알고리즘은 다음 다이어그램에 나와 있습니다:

기본 MSE group by 결과 근사화

세그먼트 수준에서 그룹 수를 제한하는 것 말고도 유사한 제한이 intermediate 단계에도 적용됩니다. 멀티 스테이지 엔진(MSE)은 서브쿼리를 허용하므로 실행 계획에서 리프(맨 아래)와 맨 위 단계 사이에 intermediate 집계를 수행하는 단계가 임의 개수 있을 수 있고, 각 단계는 여러 AggregateOperator 인스턴스로 구현될 수 있습니다. 연산자는 기본적으로 고유 그룹 수를 100,000으로 제한하며, numGroupsLimit 옵션이나 num_groups_limit 집계 힌트로 덮어쓸 수 있습니다. 이 제한은 단일 연산자 인스턴스에 적용되므로 다음 단계가 받을 수 있는 총량은 num_instances * num_groups_limit입니다.

집계 함수가 있는 집계 쿼리는 다음으로 다른 단계에서 그룹 제한·트리밍을 활성화할 수 있습니다:

  • is_enable_group_trim 힌트 — 모든 SSE/MSE 수준의 트리밍과 교차 세그먼트 수준의 그룹 제한을 활성화합니다. minSegmentGroupTrimSize 값은 따로 설정해야 합니다. Pinot는 이미 안전한 집계 없는 DISTINCT 및 집계 없는 GROUP BY 트리밍 경로를 기본 활성화하므로, 이 힌트는 집계 함수가 있거나 false로 옵트아웃할 때 주로 중요합니다. 기본값: 집계 쿼리에 대해 false; 안전한 집계 없는 DISTINCT/GROUP BY 리프 트리밍은 기본적으로 켜져 있음.
  • mse_min_group_trim_size 힌트 — intermediate 단계에서 group by 결과의 정렬·트리밍을 트리거합니다. is_enable_group_trim 힌트가 필요합니다. 기본값: 5000.

위 힌트를 사용하면 쿼리 처리는 다음과 같습니다:

리프 단계에서 SSE를 활용하는 MSE 쿼리 실행의 다양한 단계에서의 group by 결과 트리밍

실제 처리는 쿼리에 따라 달라지며, SSE 리프 단계 집계 컴포넌트가 없고 모든 수준에서 AggregateOperator에 의존할 수 있습니다. 또한 트리밍은 정렬·limit 전파에 의존하므로, order by 컬럼을 사용할 수 없는 서브쿼리에서는 트리밍이 일어나지 않을 수 있습니다.

예제

  • SSE 예제의 쿼리에 힌트를 적용하면, 즉:

    SELECT /*+ aggOptions(is_enable_group_trim='true', mse_min_group_trim_size='10') */        
i, j, count(*) as cnt
 FROM myTable
 GROUP BY i, j
 ORDER BY i ASC, j ASC
 LIMIT 3

    실행 계획은 다음과 같아야 합니다:

    LogicalSort
  PinotLogicalSortExchange(distribution=[hash])
    LogicalSort
      PinotLogicalAggregate <- aggregate up to num_groups_limit groups, then sort and trim output to group_trim_size
        PinotLogicalExchange(distribution=[hash[0, 1]])
          LeafStageCombineOperator(table=[mytable])
            StreamingInstanceResponse
              CombineGroupBy <- aggregate up to minSegmentGroupTrimSize groups
                GroupBy <- aggregate up to numGroupsLimit groups, optionally sort and trim to minSegmenGroupTrimSize
                  Project
                    DocIdSet
                      FilterMatchEntireSegment

    위 계획에서 트리밍은 GroupBy, CombineGroupBy, AggregateOperator(즉 PinotLogicalAggregate의 물리 구현) 세 연산자에서 일어납니다.

  • 조인 결과를 집계하는 경우, 예:

    select /*+  aggOptions(is_enable_group_trim='true', mse_min_group_trim_size='3') */ 
       t1.i, t1.j, count(*) as cnt
from tab t1
join tab t2 on 1=1
group by t1.i, t1.j
order by t1.i asc, t1.j asc
limit 5

    다음 실행 계획을 만들어야 합니다:

    LogicalSort
  PinotLogicalSortExchange(distribution=[hash])
    LogicalSort
      PinotLogicalAggregate(aggType=[FINAL]) <- aggregate up to num_groups_limit groups, then sort and trim output to group_trim_size
        PinotLogicalExchange(distribution=[hash[0, 1]])
          PinotLogicalAggregate(aggType=[LEAF]) <- aggregate up to num_groups_limit groups, then sort and trim output to group_trim_size
            LogicalJoin(condition=[true])
              PinotLogicalExchange(distribution=[random])
                LeafStageCombineOperator(table=[mytable])
                  ...
                    FilterMatchEntireSegment
              PinotLogicalExchange(distribution=[broadcast])
                LeafStageCombineOperator(table=[mytable])
                  ...
                    FilterMatchEntireSegment

    여기에는 리프 단계 SSE 연산자가 없고 모든 집계 단계가 MSE 연산자 PinotLogicalAggregate로 구현됩니다.

구성 파라미터

파라미터 기본값 쿼리 오버라이드 설명
pinot.server.query.executor.max.execution.threads -1 (모든 실행 스레드 사용) SET maxExecutionThreads = value; 쿼리당 사용되는 최대 실행 스레드 수(세그먼트 처리 병렬성).
pinot.server.query.executor.num.groups.limit 100,000 SET numGroupsLimit = value; 세그먼트당 허용되는 최대 그룹 수.
pinot.server.query.executor.num.groups.warn.limit 150,000 SET numGroupsWarningLimit = value; 쿼리 연산자가 유지하는 그룹 수에 대한 경고 임계값. 도달하면 Pinot가 응답 메타데이터에 numGroupsWarningLimitReached=true를 설정하고 경고를 로그하지만 쿼리 실행은 계속합니다.
pinot.server.query.executor.min.segment.group.trim.size -1 (비활성) SET minSegmentGroupTrimSize = value; 세그먼트 수준에서 그룹을 트리밍할 때 유지할 최소 그룹 수.
pinot.server.query.executor.min.server.group.trim.size 5,000 SET minServerGroupTrimSize = value; 서버 수준에서 그룹을 트리밍할 때 유지할 최소 그룹 수.
pinot.server.query.executor.groupby.trim.threshold 1,000,000 SET groupTrimThreshold = value; 서버 수준 트리밍을 트리거하는 그룹 수.
pinot.broker.min.group.trim.size 5000 SET minBrokerGroupTrimSize = value; 브로커에서 그룹을 트리밍할 때 유지할 최소 그룹 수. SSQ(*)에만 적용.
pinot.broker.mse.enable.group.trim false (집계 쿼리에는 비활성) /*+ aggOptions(is_enable_group_trim='value') */ Pinot가 기본적으로 트리밍하지 않을 때 MSQ 집계 쿼리에 대한 그룹 트리밍을 활성화. Pinot는 이미 안전한 집계 없는 DISTINCT/GROUP BY 트리밍 경로를 기본 활성화하므로, 해당 경로를 쿼리에서 옵트아웃하려면 힌트를 false로 설정하세요.
pinot.server.query.executor.mse.min.group.trim.size 5000 /*+ aggOptions(mse_min_group_trim_size='value') */ 또는 SET mseMinGroupTrimSize = value; intermediate 단계에서 그룹을 트리밍할 때 유지할 그룹 수. MSQ(**)에만 적용.

(*) SSQ - Single-Stage Query (단일 스테이지 쿼리)

(**) MSQ - Multi-Stage Query (멀티 스테이지 쿼리)

더 알아보기 (Learn more)