3. Query Execution: Chunks, Lanes, and Sort Order

2장에서 읽을 데이터를 줄이는 방법을 봤다. 이번 장에서는 그렇게 남은 데이터를 어떻게 빨리 처리하는지 다룬다. ClickHouse는 세 수준에서 병렬화한다. 연산자 안에서는 SIMD로 여러 값을 한 번에 계산하고, 노드 안에서는 여러 thread가 연산자를 동시에 실행하고, 노드 사이에서는 shard마다 따로 scan한다.

3.1 Operators Pass Chunks, Not Rows

전통적인 Volcano 방식에서는 연산자가 next()를 부를 때마다 행 하나를 넘긴다. 행마다 가상 함수를 호출하므로, 처리하는 데이터 양에 비해 호출 비용이 크다.

ClickHouse는 MonetDB/X100과 같은 vectorized 모델을 따른다. 연산자는 행 하나가 아니라 chunk를 주고받는다. chunk는 수천 행 분량의 컬럼 배열이다. 그러면 함수 호출이 chunk마다 한 번으로 줄고, 연산자 안쪽의 반복문은 같은 타입의 값이 연속으로 놓인 배열을 돌게 된다. 이런 반복문은 SIMD 명령으로 바꾸기 좋다.

SIMD 코드는 컴파일러의 자동 vectorization으로 만들기도 하고, intrinsic으로 직접 쓰기도 한다. ClickHouse는 같은 반복문을 일반 버전, AVX2 버전, AVX-512 버전처럼 여러 벌로 빌드해 두고, 실행할 때 cpuid 명령으로 CPU가 지원하는 기능을 확인해 가장 빠른 버전을 고른다. 그래서 SSE 4.2만 지원하는 15년 된 서버에서도 돌아가고, 최신 CPU에서는 그 CPU가 지원하는 명령을 모두 활용한다.

3.2 Lanes Across Cores

SQL 쿼리는 물리 연산자들의 그래프로 바뀐다. ClickHouse는 이 그래프를 쿼리 컴파일 시점에 lane 여러 개로 펼친다. lane 수는 worker thread 최대 개수(기본값은 코어 수)와 원본 table 크기에 따라 정해진다. 각 lane은 원본 table의 서로 겹치지 않는 범위를 맡는다. 병렬로 처리할 기회를 최대한 늘리려고 lane은 가능한 한 늦게 합친다.

페이지별 평균 지연을 구하고 정렬하는 쿼리라면 계획은 대략 이렇게 펼쳐진다.

flowchart LR
  r1["범위 1"] --> f1["Filter"] --> rp["Repartition"]
  r2["범위 2"] --> f2["Filter"] --> rp
  r3["범위 3"] --> f3["Filter"] --> rp
  rp --> a1["Aggregate<br/>부분 상태"] --> gm["GroupStateMerge"]
  rp --> a2["Aggregate<br/>부분 상태"] --> gm
  rp --> a3["Aggregate<br/>부분 상태"] --> gm
  gm --> sort["Sort"]

첫 단계에서는 세 lane이 각자 맡은 범위를 동시에 거른다. 그런데 범위마다 필터를 통과하는 비율이 다르면 일부 lane에 일이 몰린다. Repartition 연산자는 결과 chunk를 다음 단계로 넘길 때 lane 사이에 고르게 다시 나눠 이 불균형을 줄인다.

두 번째 단계에서는 각 lane의 Aggregate 연산자가 자기 몫의 행을 page별로 묶어 부분 상태를 만든다. 평균이라면 page마다 합계와 개수다. 이 부분 상태들은 GroupStateMerge 연산자가 하나로 합친다. GroupStateMerge는 pipeline breaker다. 모든 부분 상태가 다 모여야 결과가 확정되므로, 다음 단계는 이 연산자가 끝난 뒤에야 시작할 수 있다. 마지막 단계에서 결과를 다시 나눠 각 조각을 정렬하고 합친다.

연산자는 상태 기계로 구현되어 있다. 각 연산자는 입력을 기다리는 상태, 처리할 준비가 된 상태, 출력을 내놓은 상태를 오간다. 중앙 스케줄러는 없다. worker thread들이 계획 그래프를 계속 돌면서 상태를 진행시킬 수 있는 연산자를 스스로 찾아 실행한다. CPU 캐시에 데이터가 남아 있도록, 같은 lane에서 이어지는 연산자는 같은 thread가 처리하도록 힌트를 준다. 다른 lane의 일을 가져가는 work stealing도 한다.

이 방식은 morsel-driven 병렬화와 비슷하지만 차이가 있다. morsel-driven 방식은 원본을 10만 행 정도의 작은 조각으로 잘게 나누고 실행 중에 병렬도를 정한다. ClickHouse는 최대 병렬도를 계획에 미리 넣고 원본도 훨씬 큰 범위로 나눈다. 대신 Repartition 같은 교환 연산자를 넉넉히 써서, 단계를 거치는 동안 불균형이 쌓이지 않게 한다.

table이 여러 노드에 shard로 나뉘어 있으면, 쿼리를 받은 노드는 가능한 한 많은 일을 다른 노드에 넘긴다. 다른 노드는 원본 컬럼을 그대로 보낼 수도 있고, 거른 행만 보낼 수도 있고, 거르고 묶은 부분 상태를 보낼 수도 있고, 쿼리 전체를 실행한 결과를 보낼 수도 있다. 위 예에서는 각 노드가 page별 부분 상태를 보내고, 받은 노드의 GroupStateMerge가 로컬 결과와 함께 합친다.

3.3 Sort Order Shapes the Plan

2장에서 본 정렬 순서는 데이터를 거를 때만 쓰이지 않는다. 실행 계획을 줄이는 데도 쓰인다.

GROUP BY 컬럼이 primary key의 앞부분이면 같은 key 값을 가진 행이 연달아 나온다. 이때는 hash table을 만들 필요 없이 연달아 나오는 묶음을 차례로 집계하면 된다. 이 sort aggregation은 hash aggregation보다 메모리를 훨씬 적게 쓰고, 한 묶음이 끝나는 즉시 결과를 다음 연산자로 넘길 수 있다.

ORDER BY 컬럼이 primary key의 앞부분이면 디스크에 저장된 순서대로 읽기만 하면 정렬된 결과가 나온다. 이때는 정렬 연산자를 계획에서 아예 뺀다.

3.4 Other Optimizations

몇 가지를 더 짧게 정리한다.

필터 순서. 여러 컬럼에 조건이 걸려 있으면 선택도가 높을 것으로 보이는 조건부터 차례로 평가한다. 앞 조건을 통과한 행이 하나라도 있는 chunk만 다음 조건으로 넘기므로, 조건을 거칠수록 읽고 계산할 양이 줄어든다. 다만 아주 선택적인 조건이 하나라도 있을 때만 이렇게 한다. 그렇지 않으면 모든 조건을 병렬로 평가하는 편이 빠르다.

쿼리 컴파일. a * b + c + 1 같은 식은 연산자 세 개 대신 LLVM으로 컴파일한 연산자 하나로 합친다. GROUP BY의 여러 집계 함수나 여러 key의 정렬도 컴파일한다. 그러면 가상 함수 호출이 줄고, 데이터가 레지스터와 캐시에 머문다. 같은 식이 여러 쿼리에서 정해진 횟수 이상 나타날 때만 컴파일하고, 결과는 캐시해서 다시 쓴다.

hash table. 2024년 3월 기준으로 30가지가 넘는 hash table을 쓴다. 하나의 템플릿에서 hash 함수, allocator, cell 타입, resize 정책을 바꿔 가며 만들고, 연산자마다 key 타입과 예상 크기에 맞는 것을 고른다. key가 아주 적으면 hash 없이 key를 바로 bucket 번호로 쓰는 lookup table을 쓰고, 문자열 key는 길이에 따라 다른 hash 함수를 쓴다.

join. 예전에는 join 지원이 약해서 비정규화 table을 쓰는 경우가 많았다. 지금은 SQL의 join 종류를 모두 지원하고, hash join은 hash table을 여러 조각으로 나눠 thread 사이의 잠금 경합을 줄이는 병렬 알고리즘을 쓴다.

자원 격리. 동시에 많은 쿼리가 돌 때 thread 수를 코어 수에 맞춰 조절하고, 서버, 사용자, 쿼리 수준에서 메모리 한도를 둔다. 집계, 정렬, join이 메모리 한도를 넘으면 쿼리를 취소하지 않고 디스크를 쓰는 외부 알고리즘으로 바꾼다.

이 장의 내용을 한 문장으로 줄이면 이렇다. ClickHouse는 저장할 때 정한 정렬 순서를 읽기와 실행 양쪽에서 끝까지 활용한다.