LLM WikiAccess-protected knowledge portal
← 스터디 홈
139편 · 약 13분

Apache DataFusion 55: MERGE INTO·Sort Pushdown·Range Partitioning으로 Rust 쿼리 엔진이 Lakehouse 표준에 가까워진 방법

# Apache DataFusion 55: MERGE INTO·Sort Pushdown·Range Partitioning으로 Rust 쿼리 엔진이 Lakehouse 표준에 가까워진 방법

요약

Apache DataFusion은 Rust로 작성된 내장형 분석 쿼리 엔진이다. Ballista(분산 실행), Apache DataFusion Comet(Spark 가속), Delta Lake, Iceberg 구현체 등 여러 프로젝트의 쿼리 실행 레이어로 사용된다. DataFusion 55.0.0(2026년 7~8월 릴리스)은 세 가지 주요 기능으로 Lakehouse 워크로드 적합성을 높였다:

  1. MERGE INTO: Delta Lake·Iceberg 스타일의 upsert·delete를 DataFusion SQL로 직접 표현
  2. Sort Pushdown: ORDER BY·TopK 쿼리에서 Parquet 정렬 메타데이터를 활용해 SortExec을 제거하고 I/O를 줄임
  3. Range Partitioning: 정렬 기준 파티션 경계를 안정적으로 계산하는 분산 파티션 전략 완성

이전 릴리스인 DataFusion 54(2026년 6월)에서 LATERAL 조인·스칼라 서브쿼리 물리 실행이 추가됐고, 55에서는 그 위에 Lakehouse 표준에 더 가까운 DML 기능이 올라왔다.

MERGE INTO: Lakehouse Upsert의 핵심

왜 MERGE INTO가 필요한가

Lakehouse 아키텍처(Delta Lake, Iceberg, Hudi)에서 가장 흔한 패턴은 SCD(Slowly Changing Dimension) Type 1 upsert다:

  • 새 레코드면 INSERT
  • 기존 레코드면 UPDATE
  • 특정 조건이면 DELETE

이를 표현하는 SQL 문법이 ANSI SQL:2003에서 정의한 MERGE INTO다. DataFusion 54 이전까지는 이 문법을 지원하지 않아, MERGE 패턴을 구현하려면 INSERT와 DELETE를 별도 트랜잭션으로 나눠서 실행하거나 델타 엔진 레이어에 의존해야 했다.

DataFusion 55의 MERGE INTO 구현 계층

DataFusion 55는 세 계층에서 MERGE INTO를 구현했다.

1. SQL 파서 / 논리 계획

datafusion-expr 크레이트에 MergeInto 논리 노드가 추가됐다. WHEN MATCHED·WHEN NOT MATCHED 조건 브랜치가 각각 UPDATE·DELETE·INSERT 연산으로 변환된다.

MERGE INTO target AS t
USING source AS s ON t.id = s.id
WHEN MATCHED AND s.active = FALSE THEN DELETE
WHEN MATCHED THEN UPDATE SET t.name = s.name, t.updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT (id, name, active, updated_at)
  VALUES (s.id, s.name, s.active, s.updated_at);

2. 물리 플래너

SQL 플래너는 ON 절의 join condition을 파싱해 match key를 추출하고, 소스와 타깃 테이블의 full outer join 계획을 생성한다.

3. TableProvider 훅

TableProvider 트레이트에 merge_into 메서드가 추가됐다. 엔진은 이 훅을 통해 merge 로직을 델타 엔진이나 Iceberg writer에 위임할 수 있다. DataFusion 위에서 동작하는 delta-rs나 Iceberg-rs가 네이티브 MERGE INTO를 지원할 수 있는 확장점이다.

주의: DataFusion 55의 MERGE INTO는 SQL 파서·플래너 계층이 완성된 것이다. 원자적 upsert는 merge_into 훅을 구현한 델타 엔진(delta-rs, Iceberg-rs)이 담당한다. MemTable이나 일반 Parquet 소스에서는 아직 전체 의미론이 완성되지 않았다.

Sort Pushdown: ORDER BY / TopK에서 I/O 절약

문제: 불필요한 전체 정렬

Parquet 파일이 이미 특정 컬럼으로 정렬되어 있어도, DataFusion이 이를 알지 못하면 모든 데이터를 읽어 메모리에서 다시 정렬한다:

SELECT * FROM events ORDER BY created_at ASC LIMIT 100;

events가 created_at 순으로 정렬된 Parquet 파일 세 개로 구성된 경우, 이론적으로는 첫 파일의 앞부분만 읽으면 된다. 그러나 정렬 메타데이터가 쿼리 플랜에 반영되지 않으면 세 파일 전체를 읽고 정렬한 뒤 100개를 추린다.

DataFusion 55의 접근법

DataFusion 55는 Sort Pushdown 에픽의 핵심 단계를 완성했다.

1. 선언적 파일 정렬 정보 활용

WITH ORDER (col ASC) 힌트 또는 Parquet sorting_columns 메타데이터에서 파일별 정렬 정보를 추출해 TableScan 논리 노드에 preferred_ordering 속성으로 저장한다:

CREATE EXTERNAL TABLE events
STORED AS PARQUET
WITH ORDER (created_at ASC)
LOCATION 's3://bucket/events/';

2. SortExec 제거

쿼리 플래너가 ORDER BY created_at LIMIT 100 요청을 받을 때, TableScan의 preferred_ordering이 요청 정렬과 일치하면 SortExec를 물리 계획에서 제거한다. 대신 파일을 순서대로 읽어 DynamicFilter 임계값을 유지한다.

3. DynamicFilter를 통한 파일 스킵

TopK 쿼리에서 이미 수집한 N개의 최솟값을 기준으로, 그 기준보다 작은 max를 가진 파일을 스킵한다. 정렬된 파일 집합에서 후반부 파일을 전혀 읽지 않을 수 있다.

Sort Pushdown 동작 다이어그램

최적화 전 file_1 ts: 1–100 file_2 ts: 101–200 file_3 ts: 201–300 전체 읽기 → SortExec (메모리 정렬) LIMIT 100 300행 읽기 → 메모리 정렬 → 100행 Sort Pushdown 적용 후 file_1 ✓ ts: 1–100 읽기 file_2 ✗ ts: 101–200 스킵 file_3 ✗ ts: 201–300 스킵 순차 읽기 + DynamicFilter LIMIT 100 100행만 읽기 → SortExec 없음 → 즉시 반환 ORDER BY ts ASC LIMIT 100 — ts 기준 정렬된 Parquet 3개 파일, 각 100행 파일 선언: CREATE EXTERNAL TABLE … WITH ORDER (ts ASC)
Sort Pushdown: ORDER BY LIMIT 쿼리 최적화 전·후 비교

Range Partitioning 완성

배경: 글로벌 정렬과 Shuffle 비용

분산 쿼리 엔진에서 글로벌 ORDER BY는 전체 데이터를 정렬 기준으로 재분배(shuffle)해야 한다. DataFusion은 해시 파티셔닝(RepartitionExec)을 오래전부터 지원했지만, 정렬 범위를 기준으로 파티션 경계를 나누는 Range Partitioning은 완전히 구현되지 않은 상태였다.

Range Partitioning이 중요한 이유:

  • ORDER BY ... LIMIT 글로벌 정렬 쿼리에서 Shuffle 최소화
  • Parquet 파일 쓰기 시 각 파티션이 정렬된 상태로 출력되어, 이후 Sort Pushdown의 전제 조건(WITH ORDER) 충족
  • Sort-Merge Join에서 파티션 간 데이터 중복 방지

DataFusion 55는 RangePartitionExec을 안정화하여 물리 계획에서 정식으로 사용할 수 있게 됐다. 경계 계산은 샘플링 기반 통계를 사용한다.

DataFusion 55가 기반 프로젝트에 미치는 영향

DataFusion을 쿼리 엔진으로 사용하는 프로젝트들은 이번 릴리스의 이점을 자동으로 활용할 수 있다:

프로젝트기대 효과
delta-rsRust 네이티브 MERGE INTO — Python 우회 없이 직접 호출 가능
DataFusion CometSpark MERGE INTO 플랜의 DataFusion 실행 경로 확보
Ballista분산 MERGE INTO·Range Partitioning 글로벌 정렬 쿼리
LanceDB / Influx IOxORDER BY 포인트 쿼리에서 Parquet I/O 절감
Iceberg-rsTableProvider 훅을 통한 네이티브 MERGE 지원 경로

Sort Pushdown 효과를 최대화하는 테이블 설계

Sort Pushdown은 파일이 실제로 정렬되어 있을 때만 작동한다. 데이터 쓰기 파이프라인에서 정렬을 유지하는 방법:

Delta Lake

-- 정기 OPTIMIZE로 파일 정렬 유지
OPTIMIZE events WHERE created_at >= '2026-08-01';

DataFusion 직접 쓰기

-- 쓰기 시 정렬 보장
INSERT INTO events
SELECT * FROM staging ORDER BY created_at;

Iceberg

테이블 속성에 SORT ORDER (created_at ASC) 설정 후 Iceberg writer가 자동으로 정렬된 파일을 생성한다.

파일 정렬이 유지되지 않으면 WITH ORDER 선언이 있어도 Sort Pushdown이 잘못된 결과를 낼 수 있다. DataFusion은 선언을 신뢰하기 때문에, 잘못된 선언은 쿼리 정확성 문제로 이어진다.

Open Questions

  • Sort Pushdown에서 DynamicFilter 임계값이 여러 파티션에서 동시에 업데이트될 때 경합 조건 처리 방식이 아직 공식 문서에 명시되지 않았다.
  • Range Partitioning의 파티션 경계 샘플링 전략(샘플 크기, 재샘플링 조건)이 릴리스 노트에 상세히 기술되지 않았다.
  • MERGE INTO에서 소스와 타깃 간 스키마 불일치 처리가 SQL:2003 표준을 얼마나 엄격히 따르는지 확인이 필요하다.

References