CTE 优化

1. 背景

公用表表达式（CTE）可以被视为在单个DML语句执行范围内定义的临时结果集。CTE类似于派生表，它并不作为对象永久存储，只在查询执行期间存在。在开发过程中使用CTE可以提升SQL语句的可读性，便于管理和维护复杂的查询逻辑。

WITH Sales_CTE (SalesPersonID, TotalSales) AS
(
    SELECT SalesPersonID, SUM(TotalSaleAmount)
    FROM Sales
    GROUP BY SalesPersonID
)
SELECT p.FirstName, p.LastName, s.TotalSales
FROM SalesPerson AS p
JOIN Sales_CTE AS s ON p.SalesPersonID = s.SalesPersonID
WHERE s.TotalSales > 50000
​
UNION ALL
​
SELECT p.FirstName, p.LastName, s.TotalSales
FROM Manager AS p
JOIN Sales_CTE AS s ON p.ManagerID = s.SalesPersonID
WHERE s.TotalSales > 50000
​

在这个例子中，Sales_CTE是一个包含销售人员ID和他们总销售额的CTE。这个CTE首先被用于从销售人员表SalesPerson中检索总销售额超过50,000的销售人员。然后，相同的CTE再次被用于从经理表Manager中检索同样条件的销售经理，使用UNION ALL联合两个查询结果。

CTE复用实际上就是让上游生成的数据可以被多个下游多次消费。为了实现这个功能，我们需要考虑下面两个问题：

1. 框架如何支持一个生产者对应多个消费者的拓扑?
2. 如何在算子级别实现生产者和消费者的有效管理, 避免死锁?

2. 执行引擎

特征	流系统	MPP引擎	批引擎
数据Spill	流式处理架构侧重于内存操作，通过异步流动的方式处理状态持久化以确保精确一次性（Exactly-once）的数据处理。	MPP 引擎通常避免数据Spill，尽量吧数据流pipeline 化，以优化性能。	批处理引擎长周期、大数据量的计算任务而设计，需要支持容错和断点续传机制，通常会将中间结果持久化到磁盘上，以便在出现故障时能够从中断点恢复处理。
拓扑依赖	查询计划确定 DAG 之后拓扑基本不变，后续调整分片和并发。	静态拓扑，查询计划确定后依赖关系一般不变。执行过程中要处理数据分布和节点间通信。	静态拓扑，任务一旦开始执行，依赖关系通常不会改变。可以预先规划任务间的数据流。
依赖管理和容错	必须实时管理依赖状态，容错机制需立即响应，如保存状态快照以便失败恢复。	依赖关系较为固定，容错通过重新执行失败的查询	通常通过重试失败的Task来管理容错，可能涉及到重新处理大量数据。
示例系统	Apache Flink	Greenplum, Presto	Apache Hadoop, Apache Spark

本文主要讲的是CTE在MPP引擎中的相关实现。

从数据库内核的角度来看上述这两个问题，会从优化器和计算引擎两个方面来考虑这个问题，接下去我们就从这两个模块说明具体的算法和问题。

• 优化器：如何发现CTE以及决策复用多少个CTE，如何描述一个生产者多消费者的DAG模型；
• 计算引擎： 主要解决如何具体实现消费者生产者的管理，如何做进度控制和内存效率以及调度框架如何调度这种DAG模型**。

3 优化器

3.1 如何识别和表示CTE？

CTE复用的第一步是在QueryPlan中发现SQL里CTE并表示其数据结构。在上图类似的数据库架构中，FE/Optimizer 基础功能就是用meta 描述SQL对应的执行计划，主要有几个方面：

1. 算子输出列顺序，类型和各层如何对齐
2. 各个算子对应的物理实现算法
3. 整个执行计划的拓扑结构（数据怎么传输，拓扑网络该怎么连边）。

识别和表示CTE，一般都是通过 Parser 和 Analyze 链路中完成，也就是在SQL做词法分析和语法分析的阶段完成。

1. 对于 With 子查询，可以在 Parser 过程中，给这部分SQL转化成独立的PlanNode或者打标,同样按照这种思路对于系统里面定义的View也可以做复用。

2. 除了用户显示定义的CTE，我们也可以主动的找到SQL里面的相同子树来做复用,通常是通过在完整的QueryPlan中做子图匹配，但是选择多大的子树来做匹配会影响算法复杂度。目前看到的类似优化实现是Flink用calcite digest这套机制来做匹配。

  # 系统中Parser会直接把CTE子查询展开inline掉的情况
  PlanNode {
    List<PlanNode> children;
    int node_id;
    int origin_producer_id; // Parser阶段标记好
  }

3.2 需要判断 CTE-inline 和 CTE-Reuse 哪种方案更优 ？

在处理完识别CTE之后，我们就需要决定这个Plan是不是应该做CTE复用。简单的想CTE是能显著减少重复数据计算和IO，肯定会比不复用要好，但是事实比这个要复杂很多，如下图：

1. Inline 可以给每个CTE部分做不同的优化，比如PredicatePushDown（8-b）
2. CTE-Reuse的收益就是可以减少重复数据计算和IO （8-a）
3. 部分CTE复用

为了解决哪种方案更优，这个Rule就很好决策应该放到Cascades 框架中，通过计算这几种可能性的Cost来比较哪种更优，对应的规则的伪代码如下：

  GroupExpr {
    List<Group> children;
    ExprNode curExpr;//like JOIN/AGG
  }
  
  Group {
    List<GroupExpr> exprs;
    Map<Prop, GroupExpr> bestExprs; // lowest cost
  }
  
  CTEReuseRule {
    // Step #1 Collect CTE -> Map<node_id, CTEProducers>
  Map<int, PlanNode> cte_producers = collectCTEProducers();
  // Step #2 Replace subTree to CTE
  foreach (node in root.children) {
    if (cte_producers.get(node.origin_producer_id) != null) {
      replaceChildToCTEProducer();
    } else {
      ReplaceSubTreeToCTE(node, cte_producers);
    }
  }
  // Step #3 ReplaceEachCTE
  forecah (cte in cte_producers) {
      ReplaceSubTreeToCTE(cte, cte_producers);
  }
  }
  

3.3 规避CTE-Reuse 死锁问题？

MPP引擎出现死锁的问题，本质上是执行顺序拓扑上出现了环行依赖，如下图： 在该执行计划中（5.a），嵌套循环连接（NLJoin）节点会首先执行其外部（左侧）子节点，这会触发左侧的公共表表达式消费者（CTECConsumer）的执行。然而，这个CTECConsumer将会被阻塞，因为它尝试读取的数据还未被生成。由于NLJoin的外部子节点被阻塞，内部（右侧）子节点也将永远无法得到执行，而内部子节点本应负责执行公共表表达式生产者（CTEProducer）。

那么怎么去解决或者规避这个问题呢？

• RuleBase检测是否会发生死锁？

  算法：按照执行顺序遍历整棵树，检查是否存在两个CTE-Producer的节点被同时依赖，然后出现死锁。

    ````
      PlanNode {
        List<PlanNode> children;
        int nodeId;
        int originProducer_id; // Parser阶段标记好
        bool needSpill;
      }
      
      // return <depend all ctes,  fully consumed CTE-producers>
      Pair<List<PlanNode>, List<<PlanNode>> CollectDependCTEProducers(PlanNode root) {
        switch(root.type) {
          case HashJoin:
            // probe-side
            Pair<List<PlanNode>, List<<PlanNode>> left = CollectDependCTEProducers(root.children(0));
            // build-side
            Pair<List<PlanNode>, List<<PlanNode>> right = CollectDependCTEProducers(root.children(1));
            // check all-cte from left intersect on right full-consumed ctes
            markCycleCTE();
            return Pair<left.first+right.first, right.second>;
            ...
          default:
            // collect all CTE in 
            return Pair<left.first+right.first, left.first+ right.second>;
        }
      }
      
    ````
  

• Cascades框架检测该节点是否会发生死锁？

目前没看到现实实践，这里分享下个人思考，这里难点在于如何通过GroupExpr来传递和记录这些属性。

1. 把MemCTEProducer和SpillCTEProducer的物理算子都通过implement引入。

2. 增加物理属性重复读和不可重复读的物理属性；给GroupExpr Derive出来依赖的CTE算子集合（这里比较复杂，需要在计算bestExpr的时候收集完整）。

3. 对HashJoin这种算子向下require两种属性，但是对于不可重复读的需要检查是否会发生CTE依赖问题，如果出现的情况，MemCTEProduce的Expr需要被剔除。

• 发现死锁以后怎么解决?
  1. 把死锁的Plan剔除（如果发现这种情况就自动退化）
  2. 标记该死锁节点需要 Spill，考虑是否划算。数据落盘之后，这部分数据相当于可以无限写入，也就不会出现依赖的问题，比如上述例子中 嵌套循环连接（NLJoin）节点会首先执行其外部（左侧）子节点，这会触发左侧的公共表达式消费者（CTEConsumer）的执行，这个时候就可以触发CTEConsumer的数据拉取和落地。

4. 计算引擎侧

解决完优化器的问题之后，接下去我们来处理计算的问题：

1. 算子上实现一个生产者和多消费者的有效管理，算子本身其实可以看作是Shuffle（RPC）节点，本质没有特殊逻辑。

   • Buffer版本：一种特殊的RingBuffer算法可以解决，核心是限制内存大小，size要改成动态计算。
   • Spill 版本：可以直接CTE-Producer把数据落盘，下游算子再来消费。

2. 任务调度上如何支持一个生产者对应多个消费者的拓扑以及做进度控制？ 这部分不同的引擎下差异很大，就不过多描述。把遇到的问题简单列举以供参考：如果都是Volcano模型下的Pull模式，在划分Stage的时候需要考虑好是否支持多Root？节点内并行并发访问？内存是否会溢出？如何Spill？

5. Recursive CTE

递归CTE 是一种特殊类型的CTE，它能够以递归的方式引用自身来执行重复的查询操作。 递归CTE的基本结构由两部分组成：一个是初始查询（称为锚点查询），定义了递归的起点；另一个是递归成员，通过引用CTE自身来实现递归逻辑。这两部分通过UNION ALL操作符连接起来。

下面是一个表示递归CTE的SQL例子，它用来展示一个员工和他们的上级经理的层级结构（即员工的报告链）：

WITH RECURSIVE EmployeeHierarchy AS (
  -- 锚点查询：选择最顶层的员工（没有经理的员工）
  SELECT EmployeeID, FirstName, LastName, ManagerID
  FROM Employees
  WHERE ManagerID IS NULL
  UNION ALL
  -- 递归成员：通过关联自身来选择下一级的员工
  SELECT e.EmployeeID, e.FirstName, e.LastName, e.ManagerID
  FROM Employees e
  INNER JOIN EmployeeHierarchy eh ON e.ManagerID = eh.EmployeeID
)
SELECT * FROM EmployeeHierarchy;

这种CTE 主要靠计算引擎提供对应的递归能力，比如RecursiveUnion（PG中是如此）。对于优化器来说，它和普通的CTE复用也就没有特殊区别。