Colocated Join 전략

Colocated Join 전략

출처: 문서

Colocated join은 다음 모든 조건이 충족될 때 서버 간 데이터 셔플 없이 조인을 실행하기 위해 Pinot가 사용할 수 있는 전략이에요.

1. 두 테이블 모두 파티셔닝되어 있어요.
2. 파티션 함수가 동일해요.
3. 파티션 수가 동일해요.
4. 조인 조건이 파티션 컬럼 간의 등가예요.
5. 서버에 대한 파티션 할당이 동일해요.
6. 각 파티션에 대해 두 테이블의 해당 파티션의 모든 세그먼트를 가진 서버가 있어요.

예를 들어 다음 쿼리를 실행한다고 가정해요:

SELECT A.col1, B.col2
FROM A
JOIN B
ON A.partitionKeyA = B.partitionKeyB

테이블 A와 B가 같은 함수로 정확히 두 개의 파티션으로 파티셔닝되고 다음과 같이 분산된 시나리오에서요:

1. A의 1번째 파티션은 세그먼트 A0-1과 A0-2로 구성되며 서버 1과 3에 저장돼요.
2. A의 2번째 파티션은 세그먼트 A1-1과 A1-2로 구성되며 서버 2에 저장돼요.
3. B의 1번째 파티션은 세그먼트 B0-1로 구성되며 서버 3에 저장돼요.
4. B의 2번째 파티션은 세그먼트 B1-1로 구성되며 서버 2에 저장돼요.

이 경우 Pinot는 다음과 같은 방식으로 쿼리를 실행하려고 해요:

점선 화살표는 셔플을, 실선 화살표는 서버 내 전송을 의미해요.

부작용으로 이 전략은 다른 기법만큼 많은 서버를 사용하지 않을 수 있어요. 예를 들어 query time partition을 사용하는 같은 쿼리는 3개 서버를 사용할 수 있지만, 이 경우 Pinot는 서버 3과 서버 2만 사용할 수 있어요. 서버 1은 테이블 B의 파티션 2의 모든 세그먼트를 가지지 않기 때문에 사용할 수 없어요.

빈 파티션과 파티션 메타데이터

참여하는 테이블 중 하나가 다른 참여 테이블이 사용하는 파티션에 세그먼트가 없어도 colocated join은 실행될 수 있어요. Pinot는 직접 교환으로 연결된 스테이지에 대해 하나의 정렬된 파티션 클래스 집합을 유지해요. 그룹이 사용하는 클래스에 대해 데이터가 없는 테이블은 데이터가 있는 작업자와 colocated된 빈 작업자를 얻으므로 작업자 ID와 파티션 정렬이 조인 전체에서 일관되게 유지돼요.

Pinot는 colocated 그룹의 어떤 테이블도 그 파티션에 대한 데이터가 없을 때만 파티션 클래스를 제거해요. 예를 들어 테이블이 8개 파티션을 선언하지만 3개만 채울 수 있어요. 조인 그룹이 그 3개 클래스만 사용한다면 잎 스테이지는 8개 대신 3개 작업자를 실행해요. 이는 나중에 파티션을 잘못된 작업자 ID로 옮기지 않고 fan-out을 줄여요.

Pinot는 이 할당을 만드는 데 사용되는 메타데이터를 검증해요:

• 세그먼트 파티션 메타데이터가 테이블 구성과 일치하지 않는 하이브리드 테이블은 계획에 실패해요. 파티션 구성을 변경한 후에는 오래된 세그먼트를 다시 빌드하거나 교체해야 해요. Pinot는 더 이상 일치하지 않는 세그먼트를 조용히 제외하지 않아요.
• 여러 Pinot 클러스터에 분산된 테이블은 이 최적화를 위해 부분 파티션 메타데이터를 노출하지 않아요. 한 클러스터가 다른 클러스터가 겉보기에 빈 파티션을 서비스하는지 여부를 결정할 수 없기 때문이에요.
• 세그먼트를 포함하지만 온라인 레플리카가 없는 파티션은 여전히 계획에 실패해요. 어떤 서버도 전체 파티션을 읽을 수 없기 때문이에요.

이 빈-파티션 처리는 colocated 직접-교환 그룹의 스테이지에 특정돼요. colocated 조인 밖의 파티셔닝된 스캔은 기존 빈-파티션 동작을 유지해요.

필터링된 colocated 조인에 대한 broker 프루닝

기본 논리-플래너 경로에서 broker 세그먼트 프루닝은 필터링 후 행을 포함할 수 있는 파티션 클래스로 colocated 조인을 줄일 수도 있어요. colocated 그룹의 모든 테이블이 파티션 클래스를 프루닝하면 Pinot는 남은 클래스의 1:1 작업자 정렬을 유지하면서 그 클래스를 그룹에서 제거해요. 이는 필터링된 colocated 조인이 사용하는 작업자, 서버, 세그먼트 수를 줄일 수 있어요.

이 최적화는 broker 프루닝이 활성화되어 있고 각 적용 테이블이 segmentPrunerTypes: ["partition"]로 파티션 세그먼트 프루너를 구성해야 해요. 파티션 키에 대한 필터는 그룹의 각 면에 도달해야 해요. 양쪽에 작성되었거나 Calcite가 조인 등가를 통해 필터를 전파하기 때문이에요. 시간 필터는 세그먼트만 프루닝하고 그 자체로는 파티션 클래스를 제거하지 않아요.

그룹의 어떤 구성원이라도 여전히 행을 생성할 수 있으면 Pinot는 파티션 클래스를 유지해요. 프루닝이 모든 클래스가 비어 있음을 증명하면 모든 채워진 클래스도 유지해요. 0-작업자 colocated 그룹은 배선될 수 없기 때문이에요. colocated semi-join의 probe 잎은 적격하지 않아요. 조인 위의 비-colocated 스테이지가 줄어든 서버 집합을 유지해야 한다면 useLeafServerForIntermediateStage=true를 설정해요.

useBrokerPruning 쿼리 옵션 또는 pinot.broker.multistage.logical.planner.use.broker.pruning broker 설정을 사용해 이 동작을 제어해요.

Colocated 조인 활성화 방법

Colocated 조인 최적화는 Pinot 1.3.0에서 기본적으로 비활성화되어 있어요.

broker에서 다음 구성을 설정해 클러스터 전체에서 활성화할 수 있어요:

pinot.broker.multistage.infer.partition.hint=true

또한 다음 쿼리 옵션을 설정해 쿼리별로 활성화/비활성화할 수 있어요:

SET inferPartitionHint=true;
SELECT ...

inferPartitionHint=true일 때 Pinot는 적격한 테이블 스캔에 대해 누락된 파티션 메타데이터를 채워요. 부분 tableOptions 힌트를 추가하면 명시적인 partition_key 또는 partition_function 값이 추론된 값을 재정의하고, Pinot는 결합된 힌트를 테이블 구성에 대해 검증해요. 불일치는 무시되는 대신 계획에 실패해요. 이미 tableOptions(is_replicated='true')로 힌트된 스캔은 이 추론에서 제외되고 그 명시적 복제 힌트를 변경하지 않고 유지해요.

SELECT A.col1, B.col2
FROM A /*+ tableOptions(partition_function='hashcode', partition_key='partitionKeyA', partition_size='4') */
JOIN B /*+ tableOptions(partition_function='hashcode', partition_key='partitionKeyB', partition_size='4') */
ON A.partitionKeyA = B.partitionKeyB

Colocated 조인 강제

colocated join이 활성화되어 있어도 Pinot는 문서 초반에 나열된 조건이 충족됨을 보장할 수 있을 때만 사용해요. 예를 들어 두 테이블 사이의 조인에서 Pinot는 테이블 구성으로 같은 키가 두 테이블을 같은 방식으로(파티션 수 등) 파티셔닝함을 보장할 수 없으면 colocated join을 적용하지 않아요.

일부 복잡한 배포에서는 사용자가 colocation을 사용할 수 있다는 것을 알고 있어도 Pinot가 이러한 제약 조건을 보장하지 못할 수 있어요. 이러한 상황에서 사용자는 /*+ joinOptions(is_colocated_by_join_keys='true') */ 힌트를 추가해 Pinot가 맹목적으로 colocated join을 사용하도록 강제할 수 있어요.

경고: is_colocated_by_join_keys는 파티션 컬럼이 아닌 컬럼으로 테이블을 조인할 때만 권장해요. 파티션 컬럼으로 colocated 방식으로 조인할 때는 위에서 설명한 inferPartitionHint 또는 tableOptions 힌트를 사용하세요.

Colocated 조인이 사용될 수 있도록 보장하는 방법

위에서 언급했듯이 colocated join을 사용하려면 두 테이블의 서버에 대한 파티션 할당이 동일해야 해요. 테이블을 만들 때 파티션을 서버에 수동으로 할당할 수 있지만, 리밸런스의 결과로 언제든지 서버 간에 이동할 수 있어요.

Colocated join이 사용될 수 있도록 보장하려면 두 테이블의 각 파티션에 대해 같은 인스턴스를 할당하도록 Pinot에 지시하는 것이 좋아요. 테이블을 파티셔닝하는 방법에 대해 더 읽으려면 Instance Assignment와 Routing을 참고해요.

Colocated 조인이 사용되고 있는지 검증하는 방법

설명했듯이 이 최적화가 활성화되면 주요 이점은 조인을 실행하기 위해 데이터를 셔플할 필요가 없다는 것이에요. 이는 이 스테이지의 mailbox send 연산자에 대한 rawMessages와 inMemoryMessages 통계로 검증할 수 있어요. 모든 메시지가 메모리에 있어야 하고 rawMessages는 0이어야 해요(또는 전혀 나열되지 않아야 해요).

이 최적화가 적용되고 있는지 검증하는 또 다른 방법은 EXPLAIN IMPLEMENTATION PLAN 명령을 사용하는 것이에요. EXPLAIN IMPLEMENTATION PLAN 명령을 사용해야 해요. 거기서 MAIL_SEND 연산자가 [PARTITIONED]로 장식되고 각 MAIL_SEND가 같은 서버의 다른 작업자에게 데이터를 보낼 것임을 볼 수 있어요.

경고: 이 최적화는 일반 EXPLAIN PLAN 명령에서는 볼 수 없다는 점에 유의하세요.

더 알아보기 (Learn more)

• Join 전략 (Join Strategies)
• Lookup Join 전략
• Apache Pinot 문서 인덱스