前言

Github：https://github.com/HealerJean

博客：http://blog.healerjean.com

一、连接（Join）

在关系型数据库中，数据被分布在多个表中，这些表之间通过特定关系相互关联。SQL JOIN 操作允许我们根据这些关联条件将不同的表合并成一个更完整的结果集。

1、Doris 支持的 JOIN 类型

• INNER JOIN（内连接）：对左表每一行和右表所有行进行 JOIN 条件比较，返回两个表中满足 JOIN 条件的匹配行。详细信息请参考 [SELECT]中有关于联接查询的语法定义
• LEFT JOIN（左连接）：在 INNER JOIN 的结果集基础上。如果左表的行在右表中没有匹配，则返回左表的所有行，同时右表对应的列显示为 NULL。
• RIGHT JOIN（右连接）：与 LEFT JOIN 相反，如果右表的行在左表中没有匹配，则返回右表的所有行，同时左表对应的列显示为 NULL。
• FULL JOIN（全连接）：在INNER JOIN 的结果集基础上。返回两个表中所有的行，如果某行在另一侧表中没有匹配，则另一侧表的相应列显示为 NULL。
• CROSS JOIN（交叉连接）：没有 JOIN 条件，返回两个表的笛卡尔积，即左表的每一行与右表的每一行都进行组合。
• LEFT SEMI JOIN（左半连接）：对左表每一行和右表所有行进行 JOIN 条件比较，如果存在匹配，就返回左表的对应行。
• RIGHT SEMI JOIN（右半连接）：与 LEFT SEMI JOIN 相反，对右表每一行和左表所有行进行 JOIN 条件比较，如果存在匹配，就返回右表的对应行。
• LEFT ANTI JOIN（左反半连接）：对左表每一行和右表所有行进行 JOIN 条件比较，如果没有匹配，则返回左表的对应行。
• RIGHT ANTI JOIN（右反半连接）：与 LEFT ANTI JOIN 相反，对右表每一行和左表所有行进行 JOIN 条件比较，如果没有匹配，则返回这些行。
• NULL AWARE LEFT ANTI JOIN (对 NULL 值特殊处理的左反半连接)：与 LEFT ANTI JOIN 类似，但忽略左表中匹配列为 NULL 的行。

2、Doris 中的 JOIN 物理实现

Doris 支持两种 JOIN 的物理实现方式：Hash Join 和 Nest Loop Join。

• Hash Join： 在右表上根据等值 JOIN 列构建一个哈希表，左表的数据以流式方式通过该哈希表进行 JOIN 计算。这种方法的局限性在于它仅适用于等值 JOIN 条件的情况。
• Nest Loop Join： 通过两层循环，以左表驱动，对左表的每一行逐一遍历右表的每一行，进行 join 条件判断。适用于所有 JOIN 场景，包括处理 Hash Join 无法胜任的情况，比如涉及大于或小于比较条件的查询，或是需要执行笛卡尔积运算的场景。但相比 Hash Join，Nest Loop Join 在性能上可能会有所不及。

3、Doris Hash Join 的实现方式

作为分布式 MPP 数据库，Apache Doris 在 Hash Join 过程中需要进行数据的 Shuffle，进行拆分调度，以确保 JOIN 结果的正确性。以下是几种数据 Shuffle 方式：

说明：Hash Join 的实现方式选择是由 Doris 底层自动决定的，而非调用方直接指定。 Doris 在生成分布式执行计划时，会根据表的数据分布、大小以及查询条件等因素，自动选择最优的 Join 方式，包括 Hash Join、Nest Loop Join、Broadcast Join、Shuffle Join、Bucket Shuffle Join 和 Colocate Join 等。

1）Broadcast Join

Broadcast Join（广播连接）是一种优化的大表与小表连接策略，在分布式计算框架（如Spark、Flink）中广泛应用。其核心思想是将小表的数据广播到所有节点，使每个节点都能在本地完成连接操作，从而避免数据混洗（Shuffle）。

a、核心原理：

1. 广播小表：将小表的数据从驱动节点发送到所有工作节点。
2. 本地连接：每个工作节点使用本地的大表分区数据与广播的小表数据进行连接。
3. 结果合并：各节点完成连接后，将结果合并或直接输出。

b、举例说明

• 大表 S（例如用户行为日志）：

  +----+-------+
  | id | value |
  +----+-------+
  | 1  | A     |
  | 2  | B     |
  | 3  | C     |
  +----+-------+
  

• 小表 R（例如用户信息，尺寸较小）：

  +----+-------+
  | id | info  |
  +----+-------+
  | 2  | X     |
  | 3  | Y     |
  +----+-------+
  

执行过程：

1. 广播小表 R

   • 驱动节点将小表 R 的数据广播到所有工作节点。

2. 本地连接

   • 每个工作节点使用本地的大表分区数据与广播的 R 表进行连接。
   • 例如，节点1处理 S 的分区数据 (1, A)、(2, B)，与广播的 R 表连接，输出 (2, B, X)。
   • 节点 2 处理 (3, C)，与广播的 R 表连接，输出 (3, C, Y)。

3. 结果合并

   • 各节点的连接结果合并为最终输出：

     +----+-------+-----+
     | id | value | info|
     +----+-------+-----+
     | 2  | B     | X   |
     | 3  | C     | Y   |
     +----+-------+-----+
     

c、 使用场景

1. 小表 Join 大表：当其中一个表（通常是右表）尺寸较小，可以高效地广播到所有节点。
2. 减少 Shuffle 开销：避免因数据混洗带来的网络传输和 I/O 开销。
3. 实时或低延迟需求：广播操作通常比 Shuffle 更快，适合对延迟敏感的场景。

d、 限制

• 小表尺寸限制：小表必须足够小，能够高效地广播到所有节点。通常，框架对广播表的大小有限制（如Spark中通过 spark.sql.autoBroadcastJoinThreshold 配置，默认约 10MB ）。

• 内存压力：广播表需要在每个节点的内存中保留一份副本，如果节点数量多或广播表较大，可能导致内存不足。

• 不适用于大表Join大表：当两个表都很大时，广播其中一个表会导致极高的内存和网络开销，此时应选择其他策略（如Shuffle Join）。

e、网络开销分析

• 驱动节点将小表数据发送到所有工作节点，网络开销为 小表大小 × (节点数 - 1)。
• 例如，小表大小为 10MB，集群有 99 个 JOIN 节点，则总开销为 10MB × 99 ≈ 990MB。

2）Partition Shuffle Join

是一种分布式 Join 操作，通过 Shuffle 阶段将左右表中相同连接键（如 id）的数据发送到同一节点，然后在节点内完成 Join。其核心原理如下：

a、核心原理

1. 分区（Partitioning）
   • 左表（S）和右表（R）根据连接键（如 id）计算 Hash 值，并分配到不同分区。
   • 例如：Hash(id) % 分区数 决定数据去向。
2. Shuffle（数据重分布）
   • 系统将左右表中相同 id 的数据通过网络发送到同一节点。
   • 例如：S.id=2 和 R.id=2 会被路由到同一分区（如分区2）。
3. 本地 Join（Local Join）
   • 在每个分区内，对左右表的数据进行本地 Join（如 Hash Join 或 Sort Merge Join）

b、举例说明

假设：

• 左表 S

  +----+-------+
  | id | value |
  +----+-------+
  | 1  | A     |
  | 2  | B     |
  | 3  | C     |
  +----+-------+
  

• 右表 R：

  +----+-------+
  | id | info  |
  +----+-------+
  | 2  | X     |
  | 3  | Y     |
  | 4  | Z     |
  +----+-------+
  

• 分区数：2（分区0和分区1）。

执行过程：

1. 分区计算 Hash(id) % 2
   • S.id=1 → 1 % 2 = 1（分区1）。
   • S.id=2 → 2 % 2 = 0（分区0）。
   • S.id=3 → 3 % 2 = 1（分区1）。
   • R.id=2 → 2 % 2 = 0（分区0）。
   • R.id=3 → 3 % 2 = 1（分区1）。
   • R.id=4 → 4 % 2 = 0（分区0）。
2. Shuffle
   • 分区 0：S.id=2 和 R.id=2、R.id=4。
   • 分区 1 ：S.id=1、S.id=3 和 R.id=3。
3. 本地 Join
   • 分区 0 ：
     • S.id=2 和 R.id=2 匹配，输出 (2, B, X)。
     • R.id=4 无匹配，丢弃。
   • 分区1：
     • S.id=3 和 R.id=3 匹配，输出 (3, C, Y)。
     • S.id=1 无匹配，丢弃。

最终结果：

+----+-------+-----+
| id | value | info|
+----+-------+-----+
| 2  | B     | X   |
| 3  | C     | Y   |
+----+-------+-----+

c、使用场景

• 大表 Join 大表：当左右表数据量都很大时，无法通过 Broadcast Join（广播小表）优化，必须通过 Shuffle 将数据重分布。
• 连接键分布均匀：如果连接键（如 id）的分布均匀，Shuffle 后数据可以均匀分配到各节点，避免数据倾斜。
• 无预分区或 Bucket 优化：如果表未预先按连接键分桶（Bucket ），或分桶规则不一致，必须通过 Shuffle Join 完成。

d、限制

1. 高网络开销：Shuffle 阶段需要将大量数据通过网络传输，可能导致网络瓶颈。
2. 数据倾斜风险：如果连接键分布不均匀（如某些 id 出现次数过多），会导致部分分区数据量过大，影响性能。
3. 内存压力：Shuffle 后的数据需要在内存中完成 Join，如果数据量超过内存限制，可能触发磁盘溢出（Spill），显著降低性能。

e、网络开销

• 假设左右表各有 N 条记录，且每条记录大小为 S 字节。
• 如果连接键分布均匀，每个分区接收的数据量约为 (N * S) / 分区数。
• 但如果数据倾斜，某些分区可能接收 O(N) 的数据。

3）Bucket Shuffle Join

Bucket Shuffle Join（分桶 Shuffle 连接）是一种优化的分布式连接策略，结合了 分桶（Bucket） 和 Shuffle 的思想。其核心在于预先将数据按连接键分桶存储，使得连接时相同连接键的数据位于同一或相关分桶中，从而减少 Shuffle 的数据量或完全避免 Shuffle。

a、核心原理：

1. 数据分桶（Bucket）：
   • 在数据写入时（如存储为表），根据连接键的哈希值将数据分配到固定数量的桶（Bucket）中。
   • 例如，使用 id % 4 将数据分配到 4 个桶中。
2. 连接时匹配桶：
   • 连接操作时，左右表相同桶号的数据会被发送到同一节点（或直接在本地连接，如果桶分布一致）。
   • 由于数据已按连接键分桶，Shuffle 的数据量大大减少，甚至可以避免。

b、示例说明

假设：

• 左表 S（用户行为表）和 右表 R（用户信息表）均按 id 分4个桶。
• 分桶规则：bucket_id = id % 4。

数据分布：

• 表 S：

  Bucket 0: (id=4, value=D), (id=8, value=H)
  Bucket 1: (id=1, value=A), (id=5, value=E)
  Bucket 2: (id=2, value=B), (id=6, value=F)
  Bucket 3: (id=3, value=C), (id=7, value=G)
  

• 表 R：

  Bucket 0: (id=4, info=W), (id=8, info=Z)
  Bucket 1: (id=1, info=X)
  Bucket 2: (id=2, info=Y), (id=6, info=V)
  Bucket 3: (id=3, info=U)
  

执行过程：

1. 分桶存储：

   • 表 S 和 R 在存储时已按 id % 4 分桶，数据物理上存储在对应的桶文件中。

2. 连接时桶匹配：

   • 连接操作时，相同桶号的数据会被处理：
     • Bucket 0：S 中的 (4,D)、(8,H) 与 R 中的 (4,W)、(8,Z) 连接。
     • Bucket 1：S 中的 (1,A)、(5,E) 与 R 中的 (1,X) 连接（id=5 无匹配，丢弃）。
     • Bucket 2：S 中的 (2,B)、(6,F) 与 R 中的 (2,Y)、(6,V) 连接。
     • Bucket 3：S 中的 (3,C)、(7,G) 与 R 中的 (3,U) 连接（id=7 无匹配，丢弃）。

3. 结果合并：各桶的连接结果合并为最终输出：

   +----+-------+-----+
   | id | value | info|
   +----+-------+-----+
   | 2  | B     | Y   |
   | 3  | C     | U   |
   | 4  | D     | W   |
   | 6  | F     | V   |
   | 8  | H     | Z   |
   +----+-------+-----+
   （假设id=1无匹配info，id=5和id=7无匹配，不输出）
   

c、 使用场景

1. 频繁连接的表：当两个表需要频繁进行连接操作时，预先分桶可以显著提升连接性能。
2. 数据分布稳定：连接键的分布稳定，不会频繁变化，确保分桶的有效性。
3. 减少 Shuffle 开销：适用于大表连接，通过分桶减少 Shuffle 的数据量，甚至实现共分区（Co-partition）避免Shuffle。

d、限制

1. 分桶键选择：分桶键必须与连接键一致，否则无法实现桶匹配。
2. 分桶数一致：左右表的分桶数必须相同，否则可能导致数据分布不均。
3. 分区规则不一定要求相同（但建议一致）：分区规则不一致确实很可能导致即使分桶规则相同，数据也无法保证在同一个节点上，从而影响 Bucket Shuffle Join 的有效性
4. 数据倾斜风险：如果连接键分布不均匀，某些桶的数据量可能过大，影响性能。
5. 维护成本：预先分桶需要额外的存储和管理成本，不适合临时查询或动态数据。

e、 网络开销分析

• 分桶存储阶段数据在写入时已按桶分布，无额外网络开销。
• 连接阶段：
  • 如果左右表的桶分布一致（如使用相同的分桶规则），则相同桶号的数据已在同一节点，无需 Shuffle。
  • 如果桶分布不一致，可能需要少量 Shuffle 调整数据位置。

4）Colocate Join

Colocate Join（同分布连接）是一种优化的分布式连接策略，其核心思想是确保左右表中具有相同连接键的数据存储在同一个节点上，从而在连接时无需数据移动（即避免 Shuffle），直接在本地完成连接操作。

a、核心原理：

• 数据分布一致性：在数据写入阶段，通过特定的分布策略（如哈希分布或范围分布），确保左右表中相同连接键的数据被路由到同一个节点。
• 本地连接：由于数据已经同分布，连接操作可以在每个节点上独立进行，无需跨节点传输数据。
• 结果合并：各节点完成本地连接后，将结果合并或直接输出。

为了使得 Table 能够有相同的数据分布，同一 CG Colocation Group（共址分组） 内的 Table 必须保证以下属性相同：

1. 分桶列和分桶数

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

2. 副本数

   同一个 CG 内所有表的所有分区（Partition）的副本数必须一致。如果不一致，可能出现某一个 Tablet 的某一个副本，在同一个 BE 上没有其他的表分片的副本对应。

b、示例说明

假设：

• 左表 S（订单表）和 右表 R（用户信息表），连接键为 user_id。
• 分布策略：使用哈希分布，hash(user_id) % 节点数，确保相同 user_id 的数据存储在同一个节点。

执行过程：

1. 数据写入
   • 左表 S 和右表 R 在写入时，根据 user_id 的哈希值被分配到相同的节点。
   • 例如，user_id=1001 的数据无论来自左表还是右表，都会被存储在节点2上。
2. 连接操作
   • 每个节点使用本地的 S 和 R 表数据，根据 user_id 进行连接。
   • 节点2处理 user_id=1001 的数据，输出连接结果。
3. 结果合并：各节点的连接结果合并为最终输出。

c、使用场景

1. 频繁连接的表：当两个表需要频繁进行连接操作时，预先确保数据同分布可以显著提升连接性能。
2. 实时或低延迟需求：由于避免了 Shuffle ，连接操作可以快速完成，适合对延迟敏感的场景。
3. 大规模数据集：对于无法通过广播或简单分桶优化的大表连接，Colocate Join 提供了一种有效的解决方案。

d、限制

1. 数据倾斜风险：如果连接键分布不均匀，某些节点可能承担过多的数据，导致性能瓶颈。
2. 维护成本：确保数据同分布需要额外的管理和维护工作，如重新分布数据以应对节点增减。
3. 灵活性受限：一旦数据分布策略确定，调整起来可能较为复杂，不适合动态变化的场景。

e、网络开销分析

• 数据写入阶段：数据根据分布策略被路由到相应节点，可能产生一定的网络开销。
• 连接阶段：由于数据已经同分布，连接操作无需额外网络传输，网络开销极低。

二、聚合多维分析

ROLLUP、CUBE 和 GROUPING SETS 是 SQL 中用于多维数据汇总的强大工具，它们能够简化复杂的聚合查询，适用于数据分析、报表生成等场景。以下是详细介绍和示例：

• GROUPING：适用于需要单独判断某一列是否被汇总的场景。例如，在报表中添加标识列，明确哪些维度被汇总。
• GROUPING_ID：适用于需要快速识别汇总行类型的场景。例如，在结果集中添加一个汇总标识列，方便后续处理（如排序、过滤等）。

1）聚合分析

a、ROLLUP

• 定义：ROLLUP 用于生成层次化的汇总数据，按照指定的列顺序从最细粒度逐步汇总到最高层次。

• 适用场景：需要逐级汇总的场景，例如按地区、时间、产品类别等维度进行分层统计。

• 语法：

SELECT col1, col2, ..., aggregate_function(col)
FROM table
GROUP BY ROLLUP(col1, col2, ...);

示例：假设有一个销售数据表 sales，包含 region（地区）、year（年份）、sales_amount（销售额）。使用 ROLLUP 按地区和年份汇总：

SELECT 
    region, 
    year, 
    SUM(sales_amount) AS total_sales
FROM sales
GROUP BY ROLLUP(region, year);


region	year	total_sales
East	2023	3000.00
East	2024	3400.00
East	NULL	6400.00
West	2023	4000.00
West	2024	4600.00
West	NULL	8600.00
NULL	NULL	15000.00

结果：

• 每个地区每年的销售额。
• 每个地区的总销售额（year 为 NULL）。
• 整体总销售额（region 和 year 均为 NULL）。

b、CUBE

定义：CUBE 生成所有可能的汇总组合，适用于多维分析。

适用场景：需要全面分析所有维度组合的场景，例如同时按地区、年份、产品类别统计。

语法：

SELECT col1, col2, ..., aggregate_function(col)
FROM table
GROUP BY CUBE(col1, col2, ...);

示例：按地区、年份和产品类别统计销售额：

SELECT 
    region, 
    year, 
    product_category,
    SUM(sales_amount) AS total_sales
FROM sales
GROUP BY CUBE(region, year, product_category);

结果：

• 每个地区、年份、产品类别的销售额。
• 每个地区、年份的汇总（product_category 为 NULL）。
• 每个地区、产品类别的汇总（year 为 NULL）。
• 每个年份、产品类别的汇总（region 为 NULL）。
• 整体总销售额（所有列为 NULL）。

c、GROUPING SETS

定义：GROUPING SETS 允许用户指定一组列的组合进行独立聚合，提供灵活的定制化汇总。

适用场景：需要特定维度组合的场景，例如只统计某些维度的汇总，而忽略其他组合。

语法：

SELECT col1, col2, ..., aggregate_function(col)
FROM table
GROUP BY GROUPING SETS (
    (col1),
    (col2),
    (col1, col2),
    ...
);

示例：只统计以下组合：

1. 每个地区的总销售额。
2. 每年的总销售额。
3. 每个地区每年的销售额。

SELECT 
    region, 
    year,
    SUM(sales_amount) AS total_sales
FROM sales
GROUP BY GROUPING SETS (
    (region),
    (year),
    (region, year)
);

结果：

• 每个地区的总销售额（year 为 NULL）。
• 每年的总销售额（region 为 NULL）。
• 每个地区每年的销售额。

2）聚合汇总结果

a、GROUPING

定义：GROUPING 函数用于判断某一列是否在汇总行中被省略（即显示为 NULL）。如果是，则返回 1；否则返回 0。

适用场景：

• 区分汇总行和普通数据行。
• 在结果集中添加标识列，明确哪些维度被汇总。

语法：

GROUPING(column_name)

示例：假设我们有一个销售数据表 sales，使用 ROLLUP 按地区和年份汇总：

SELECT 
    region, 
    year, 
    SUM(sales_amount) AS total_sales,
    GROUPING(region) AS is_region_grouped,
    GROUPING(year) AS is_year_grouped
FROM sales
GROUP BY ROLLUP(region, year);

结果：

解释：

• 当 is_region_grouped = 1 时，表示 region 列被汇总（显示为 NULL）。
• 当 is_year_grouped = 1 时，表示 year 列被汇总（显示为 NULL）。

b、GROUPING_ID

定义：GROUPING_ID 函数返回一个整数，用于唯一标识当前行的汇总级别。它是基于 GROUPING 函数的结果计算的二进制编码。

适用场景：

• 快速识别汇总行的类型。
• 在结果集中添加一个汇总标识列，方便后续处理。

语法：

GROUPING_ID(column1, column2, ...)

示例：继续使用 ROLLUP 按地区和年份汇总：

SELECT 
    region, 
    year, 
    SUM(sales_amount) AS total_sales,
    GROUPING_ID(region, year) AS grouping_level
FROM sales
GROUP BY ROLLUP(region, year);

结果：

解释：grouping_level 的值是 GROUPING函数结果的二进制编码的十进制表示。

• (region, year) 普通行：GROUPING(region)=0，GROUPING(year)=0 → 00（二进制）→ 0（十进制）。
• (region, NULL) 年份汇总行：GROUPING(region)=0，GROUPING(year)=1 → 01（二进制）→ 1（十进制）。
• (NULL, NULL) 地区和年份汇总行：GROUPING(region)=1，GROUPING(year)=1 → 11（二进制）→ 3（十进制）。

三、分析函数（窗口函数）

分析函数，也称为窗口函数，是一种在 SQL 查询中对数据集中的行进行复杂计算的函数。窗口函数的特点在于，它们不会减少查询结果的行数，而是为每一行增加一个新的计算结果。窗口函数适用于多种分析场景，如计算滚动合计、排名以及移动平均等。

1、基本概念介绍

1）处理顺序

使用分析函数的查询处理可以分为三个阶段。

1. 执行所有的连接、WHERE、GROUP BY 和 HAVING 子句。
   1. 执行所有的连接操作
   2. 应用 WHERE 子句进行过滤
   3. 执行 GROUP BY 分组
   4. 应用 HAVING 子句对分组结果过滤
2. 将结果集提供给分析函数，并进行所有必要的计算。
   1. 将基础查询的结果集提供给分析函数
   2. 根据 PARTITION BY 子句创建分区
   3. 在每个分区内应用窗口定义
   4. 执行所有必要的分析计算
3. 如果查询的末尾包含 ORDER BY 子句，则处理该子句以实现精确的输出排序。

2）术语解释

a、结果集分区

• 定义：通过 PARTITION BY 子句将结果集划分为多个行组（分区）
• 特点：
  • 与表分区功能无关，是查询处理中的逻辑分区
  • 分析函数在每个分区内独立计算
  • 示例：PARTITION BY department 将按部门划分数

b、窗口

• 定义：为分区内每行定义的滑动数据范围，用于计算
• 窗口类型：
  • 累积窗口：起始行固定（如分区首行），结束行滑动。示例：累积和函数
  • 滑动窗口：起始和结束行都滑动，示例：移动平均值

• 窗口大小：

  • 可包含分区内所有行
  • 也可定义为单行滑动窗口

• 边界处理：靠近分区边界时，窗口自动缩减

• 窗口定义语法：

  ROWS BETWEEN [m] PRECEDING AND [n] FOLLOWING
  -- 或简写
  ROWS [m] PRECEDING  -- 等价于 BETWEEN m PRECEDING AND CURRENT ROW
  

• 当前行处理：

  • 当前行总是包含在计算中
  • 计算 n 个项目时，窗口应指定为( n - 1 )。示例：5天平均值应使用 ROWS BETWEEN 4 PRECEDING AND CURRENT ROW

c、当前行

• 作用：作为窗口开始和结束的参考点

• 应用示例：

  • 中心移动平均：包含当前行、前 6 行和后 6 行（共 13 行窗口）

  • 示例语法：

  • AVG(value) OVER (
      PARTITION BY category
      ORDER BY date
      ROWS BETWEEN 6 PRECEDING AND 6 FOLLOWING
    )