【OpenGauss源码学习 —— （SortGroup算子）】

声明：本文的部分内容参考了他人的文章。在编写过程中，我们尊重他人的知识产权和学术成果，力求遵循合理使用原则，并在适用的情况下注明引用来源。
本文主要参考了 OpenGauss6.0.0 的开源代码和《OpenGauss数据库源码解析》一书

概述

  今天有幸看到了一篇专利：基于OpenGauss的数据分组聚集方法。该专利提出了一种基于OpenGauss的高效数据分组聚集方法，旨在解决现有技术在处理海量数据时因无效排序和分组导致的性能问题，下面一起来学习一下专利所提出的方案。

专利内容概述

技术背景

• 传统的 SQL 聚集操作（如 COUNT、MAX、MIN、SUM、AVG）在进行 GROUP BY 和 ORDER BY 处理时，可能会导致大量无用数据的计算，从而降低系统性能。
• 现有方案通常是：
  • 先分组再排序，但所有数据都需要先分组计算，即使最终只需要 Top-N 组，仍会处理所有数据，造成不必要的计算开销。
  • 先排序再分组，但所有数据都要排序，最终只取 Top-N 组，依然涉及大量不必要的排序计算。

本发明的改进

• 核心思想：在 GROUP BY 过程中仅维护 Top-N 组的数据，丢弃其余数据，减少计算开销。
• 该方法结合 OpenGauss 数据库，通过限制分组数据的数量，减少排序和分组操作的资源消耗，提高查询性能。

具体方法

包含两个主要阶段：

1. 数据扫描与分组

• 扫描数据，检查该数据属于哪一组：
  • 若数据所在的分组属于 Top-N 目标组，直接写入该分组。
  • 若数据的分组不在 Top-N 目标组，则直接丢弃该数据。
  • 若数据的分组尚不存在但应属于 Top-N 目标组，则创建新分组，并在插入时淘汰 Top-N 组中排名最末的分组。

2. 聚集计算

• 仅对 Top-N 组 进行聚集运算（COUNT、SUM、MAX、MIN、AVG）。
• 由于只处理 Top-N 组，大幅减少存储和计算资源，提高查询效率。

系统架构

该方法被设计为一个完整的系统，包括以下模块：

• 数据扫描模块：扫描数据集合。
• 数据分组模块：按规则选择合适的分组。
• 数据写入模块：将数据写入对应的分组。
• 分组创建模块：当需要新分组时，创建并插入合适位置。
• 分组维护模块：插入新分组的同时淘汰最末位的分组，保持 Top-N 组数量稳定。
• 聚集运算模块：执行 COUNT、SUM、MAX 等聚集运算。

实验结果

• 相比现有方法，该方法在执行 SQL GROUP BY + ORDER BY + LIMIT N 查询时，计算效率显著提高：
• 对比结果：
  • ​HashAgg（先分组）​：内存114MB，耗时14秒。
  • ​​GroupAgg（先排序）​：内存2.3GB，耗时43秒。
  • ​​本方案：内存32KB，耗时1.6秒。
    ​ - 结论：显著优化资源消耗与执行效率，尤其适合海量数据Top-N分析场景。

适用场景

• 适用于大规模数据分析，特别是 OpenGauss 数据库下的 Top-N 分组聚集查询，如日志分析、排名计算等。

源码分析

ExecInitSortGroup 函数

  ExecInitSortGroup 函数负责初始化 SortGroupState 结构，为 SortGroup 节点的执行做好准备。它创建并配置 SortGroupState，绑定执行计划 (Plan)、执行环境 (EState)，并设定 ExecSortGroup 作为执行函数。该函数还初始化外部子计划 (outerPlan)，分配元组存储槽 (TupleSlot)，继承元组描述符 (TupleDesc)，并确保 SortGroup 节点在查询执行过程中正确处理 GROUP BY + ORDER BY 逻辑。通过该初始化，ExecSortGroup 能够高效执行分组排序，并支持 Top-N 结果优化，提高查询性能。函数源码如下所示：（openGauss-server-v6.0.0-RC1\src\gausskernel\runtime\executor\nodeSortGroup.cpp）

/* ----------------------------------------------------------------
 *        ExecInitSortGroup
 *
 *        Creates the run-time state information for the SortGroupState node
 *        produced by the planner and initializes its outer subtree.
 *
 *        创建 `SortGroupState` 运行时状态，并初始化其外部子计划（outer subtree）。
 * ----------------------------------------------------------------
 */
SortGroupState *ExecInitSortGroup(SortGroup *node, EState *estate, int eflags)
{
   
    SortGroupState *sortGroupState;  // 用于存储分组排序的执行状态

    /*
     * 创建 `SortGroupState` 结构体
     */
    sortGroupState = makeNode(SortGroupState); // 分配并初始化 SortGroupState 结构
    sortGroupState->ss.ps.plan = (Plan *)node; // 绑定执行计划
    sortGroupState->ss.ps.state = estate;      // 绑定执行状态
    sortGroupState->ss.ps.ExecProcNode = ExecSortGroup; // 设定执行函数 `ExecSortGroup`

    // 设置默认的 bound，LONG_MAX 表示无界（不限制返回行数）
    sortGroupState->bound = LONG_MAX; 

    // 初始时 `sort_Done` 置为 false，表示排序尚未完成
    sortGroupState->sort_Done = false;

    /*
     * 检查 eflags 是否包含不支持的标志位：
     * - EXEC_FLAG_MARK：表示支持标记（mark）和恢复（restore），但 `SortGroup` 不支持。
     * - EXEC_FLAG_BACKWARD：表示支持反向扫描，`SortGroup` 也不支持。
     */
    Assert(!(eflags & (EXEC_FLAG_MARK | EXEC_FLAG_BACKWARD)));

    /*
     * 初始化 `SortGroupState` 的子计划（outerPlan）。
     * `ExecInitNode` 递归初始化外部子计划（outer subtree）。
     */
    outerPlanState(sortGroupState) = ExecInitNode(outerPlan(node), estate, eflags);

    /*
     * 杂项初始化
     *
     * `SortGroupState` 节点不会调用 `ExecQual`（条件过滤）或 `ExecProject`（投影计算），
     * 因此不需要初始化 `ExprContext`。
     */

    /*
     * 初始化返回结果的 Tuple Slot。
     * `ExecInitResultTupleSlot` 分配一个 TupleTableSlot（用于存储执行结果）。
     */
    ExecInitResultTupleSlot(estate, &sortGroupState->ss.ps);

    // 初始化扫描用的 Tuple Slot（用于存储从外部计划读取的元组）
    ExecInitScanTupleSlot(estate, &sortGroupState->ss);

    // 由于 SortGroupState 不需要进行投影操作，因此 `ps_ProjInfo` 设为空
    sortGroupState->ss.ps.ps_ProjInfo = NULL;

    /*
     * 从外部计划中获取元组描述信息，并赋值给 `SortGroupState`
     */
    ExecAssignScanTypeFromOuterPlan(&sortGroupState->ss);

    /*
     * 从目标列表（TargetList）中获取结果的 Tuple Descriptor，
     * 并将其赋值给 `SortGroupState` 的 `ps_ResultTupleSlot`。
     */
    ExecAssignResultTypeFromTL(&sortGroupState->ss.ps,
                               sortGroupState->ss.ss_ScanTupleSlot->tts_tupleDescriptor->td_tam_ops);

    /*
     * 断言确保 `ps_ResultTupleSlot` 的元组描述符已正确初始化
     */
    Assert(sortGroupState->ss.ps.ps_ResultTupleSlot->tts_tupleDescriptor->td_tam_ops);

    // 返回初始化完成的 `SortGroupState`
    return sortGroupState;
}

SortGroupState 结构体

  SortGroupState 用于管理 SortGroup 节点执行状态的结构体，它存储了排序分组的相关信息，控制查询执行过程中 GROUP BY + ORDER BY 的逻辑。结构体定义如下：

/* ----------------
 *	 SortGroupState information
 *
 *	 存储 `SortGroup` 执行状态的信息
 * ----------------
 */
typedef struct SortGroupState {
   
    ScanState ss;                     /* 继承自 `ScanState`，其第一个字段是 `NodeTag`，用于标识节点类型 */
    int64 bound;                      /* 若有限制，表示需要返回多少个分组（Top-N）；默认 `LONG_MAX` 表示无限制 */
    struct SortGroupStatePriv *state; /* 指向 `SortGroupStatePriv`，用于存储 `nodeSortGroup.c` 中的私有排序状态 */
    bool sort_Done;                   /* 排序是否已经完成？`true` 表示排序已完成 */
    bool *new_group_trigger;          /* 指示哪些元组属于新的分组，帮助 `ExecSortGroup` 返回新的分组数据 */
    const char *spaceType;            /* 记录 `SortGroup` 使用的存储空间类型（如“memory”或“disk”），用于 `EXPLAIN` 解析 */
    int64 spaceUsed;                  /* 记录 `SortGroup` 使用的存储空间大小（字节数），用于 `EXPLAIN ANALYZE` 统计 */
} SortGroupState;

结构体作用总结
  SortGroupState 主要用于管理 SortGroup 逻辑节点的执行状态，支持 GROUP BY + ORDER BY 操作。它的核心功能包括：

1. 存储分组排序的执行状态（sort_Done）。
2. 控制 Top-N 查询优化（bound）。
3. 指示新分组边界（new_group_trigger）。
4. 管理排序存储的使用情况（spaceType & spaceUsed）。
5. 通过 SortGroupStatePriv 维护具体排序实现（state）。

SortGroupStatePriv 结构体

  SortGroupStatePriv 是 SortGroupState 的 私有状态结构体，用于管理分组排序的底层数据存储和排序操作。它负责控制跳表（Skiplist）和磁盘存储（LogicalTapeSet） 的使用，以优化 GROUP BY + ORDER BY 查询。结构体定义如下：

/*
 * Private state of a SortGroupState operation.
 *
 * SortGroupState 的私有状态，管理分组排序的底层数据存储与排序逻辑。
 */
typedef struct SortGroupStatePriv {
   
    MemoryContext maincontext;  /* 用于存储分组元数据的内存上下文 */
    MemoryContext tuplecontext; /* 用于存储内存中的元组数据 */

    int nKeys;                  /* 排序键的列数（GROUP BY 的列数） */

    bool holdTupleInMem;        /* 是否将元组存储在内存中？否则存储在 `LogicalTapeSet`（临时文件） */

    int64 allowedMem;           /* 允许的最大内存大小（字节），超出则转为磁盘存储 */
    int64 max_groups;           /* Skiplist 可存储的最大分组数（Top-N 限制） */

    bool *new_group_trigger;    /* 标记新分组出现的位置，避免重复处理 */

    /*
     * tupDesc 仅用于 MinimalTuple（最小元组格式）和 CLUSTER 操作。
     */
    TupleDesc tupDesc;          /* 元组描述符，定义数据格式 */
    SortSupport sortKeys;       /* 排序键数组，长度为 nKeys */

    Skiplist skiplist;          /* 使用跳表（Skiplist）存储 Top-N 组，提高查询效率 */

    LogicalTapeSet *tapeset;    /* 逻辑磁带集（LogicalTapeSet），用于存储磁盘上的临时排序数据 */
    int *freeTape;              /* 记录 LogicalTapeSet 中空闲磁带编号的数组 */
    int freeTapeSize;           /* 当前空闲磁带的数量 */
    int freeTapeCap;            /* freeTape 数组的最大容量 */

    TupleIter iter;             /* 元组迭代器，保存当前读取的位置，支持流式查询 */</