第2.5章：StarRocks表设计--Colocation Join

一、StarRocks数据划分

1.1 分区

1.2 分桶

二、Colocation Join实现原理

2.1 Colocate Join概述

2.2 Colocate Join实现原理

三、应用案例

注：本篇文章阐述的是StarRocks-3.2版本的Colocation Join

官网文章地址：

Colocate Join | StarRocks

一、StarRocks数据划分

在介绍Colocation Join之前，再回顾下StarRocks的数据划分及tablet多副本机制。StarRocks的数据表按照分区分桶规则，切分成若干个数据分片（tablet）存储在不同的be节点上，每个tablet都有多个副本（默认是3副本）。

在 StarRocks 存储引擎中，用户数据被水平划分为若干个数据分片（Tablet，也称作数据分桶 Bucket）。每个 Tablet 包含若干数据行。各个 Tablet 之间的数据没有交集，并且在物理上是独立存储的。

一个 Tablet 只属于一个数据分区（Partition）。而一个 Partition 包含若干个 Tablet。因为 Tablet 在物理上是独立存储的，所以可以视为 Partition 在物理上也是独立的。Tablet 是数据移动、复制等操作的最小物理存储单元。

若干个 Partition组成一个Table。Partition 可以视为是逻辑上最小的管理单元。数据的导入与删除，仅能针对一个Partition 进行。

下图说明 Table、Partition、Bucket（Tablet）的关系：

Table 按照 Range 的方式按照 date 字段进行分区，得到了 N 个 Partition
每个 Partition 通过相同的 Hash 方式将其中的数据划分为 M 个 Bucket（Tablet）
从逻辑上来说，Bucket 1 可以包含 N 个 Partition 中划分得到的数据，比如下图中的 Tablet 11、Tablet 21、Tablet N1

1.1 分区

逻辑概念，分区用于将数据划分成不同的区间，主要作用是将一张表按照分区键拆分成不同的管理单元。

1.2 分桶

物理概念，StarRocks一般采用Hash算法作为分桶算法。在同一分区内，分桶键哈希值相同的数据会划分到同一个tablet（数据分片），tablet以多副本冗余的形式存储，是数据均衡和恢复的最⼩单位，数据导入和查询最终都下沉到所涉及的 tablet副本上。

二、Colocation Join实现原理

2.1 Colocate Join概述

在数据分布满足一定条件的前提下，计算节点只需做本地 Join，减少跨节点的数据移动和网络传输开销，提高查询性能。Colocate Join 十分适合几张大表按照相同字段分桶的场景，这样可以将数据预先存储到相同的分桶中，实现本地计算。

要理解这个算法，需要先了解以下几个概念：

Colocation Group（CG）：同一 CG 内的表需遵循相同的 Colocation Group Schema（CGS），即表对应的分桶副本具有一致的分桶键、副本数量和副本放置方式。如此可以保证同一 CG 内，所有表的数据分布在相同一组 BE 节点上。
Colocation Group Schema（CGS）：用于描述一个 CG 中的Table，和Colocation相关的通用 Schema 信息。包括分桶列类型，分桶数以及副本数等。
分桶编号Bucket Seq：一个表的数据，根据分桶列 Hash、对桶数取模后落在某一个分桶内。假设一个 Table 的分桶数为 8，则共有 [0, 1, 2, 3, 4, 5, 6, 7] 8 个分桶（Bucket）。因此【分桶列 Hash %桶数】一致的数据会划分到同一个桶中。

2.2 Colocate Join实现原理

Colocation Join 功能，是将一组拥有相同CGS 的 Table 组成一个 CG。并保证这些 Table 对应的数据分片会落在同一个 BE 节点上。使得当 CG 内的表进行分桶列上的 Join 操作时，可以通过直接进行本地数据 Join，减少数据在节点之间的传输耗时。

因此核心问题直接转变成【如果保证这些table对应的数据分片会落在同一个be节点上？】

同一 CG 内的Table必须保证以下属性：

（1）分桶列和分桶数

同一 CG内表的分桶键的类型、数量和顺序完全一致，并且桶数一致，从而保证多张表的数据分片能够一一对应地进行分布控制。

分桶列，即在建表语句中distributed by hash(col1, col2, ...) 中指定的列。分桶列决定了一张表的数据通过哪些列的值进行Hash划分到不同的Tablet 中。同一 CG内的 Table 必须保证分桶列的类型和数量完全一致，并且桶数一致，才能保证多张表的数据分片能够一一对应的进行分布控制。

（2）副本数

同一个 CG内所有表的所有分区（Partition）的副本数必须一致。如果不一致，可能出现某一个 Tablet 的某一个副本，在同一个 BE 上没有其他的表分片的副本对应。不过，同一个 CG 内的表，分区的个数、范围以及分区列的类型不要求一致。

ps：同一个 CG 内所有表的分区键，分区数量可以不同。因为Partition只是一个逻辑上的分区，真正影响数据分布在哪一个BE节点的是由Bucket决定的。

综上，在固定了分桶列和分桶数后，同一个CG内的表会拥有相同的Buckets Seq。而副本数决定了每个分桶内的 Tablet 的多个副本分别存放在哪些 BE 上。假设Buckets Seq为 [0, 1, 2, 3, 4, 5, 6, 7]，BE 节点有 [A, B, C, D] 4个。则一个可能的数据分布如下：

CG 内表的一致的数据分布定义和tablet副本映射，能够保证分桶列值相同的数据都在同一个 BE 节点上，可以进行本地数据 Join。其核心思想是「两次映射」，保证相同的 Distributed Key 的数据会被映射到相同的 Bucket Seq，再保证 Bucket Seq对应的 Bucket 映射到相同的 BE 节点：