cascade optimizer如何确定table spool

TableSpool带来的问题

目前IMCI的优化器在处理TableSpool时，由于IMCI优化器需要结合延迟物化执行器和传统的非延迟物化执行器，因此对于每个TableSpool的PrimarySpool(producer)与RefSpool(Consumer)，优化器都会生成两个TableSpool: RIDTableSpool与DataTableSpool，用以下这个Plan为例，我们可以看到在引入Spool之后可能存在的问题：

flowchart TD
    subgraph Primary
        direction RL
        a1(PrimaryRidSpool)
        a2(PrimaryDataSpool)
    end
    subgraph Ref
        direction RL
        a3(RefRidSpool)
        a4(RefDataSpool)
    end
    A(Project) --> B(HashMatch)
    B --> Ref
    
    B --> F(Groupby)
    F --> Primary
    Primary --> E(...)

volcano optimizer需要选出代价最小的table spool来执行这条Query，但是并不是任意选择TableSpool，查询计划都是合法的，因为PrimarySpool与RefSpool含有依赖关系，他们的输出结果必须一样，也就是说，RIDSpool与DataSpool必须成对选择，而在cascade optimizer的框架中，优化每个plan的分支时，对另一个分支的信息都所知甚少，因此我们需要改造优化器的流程，来保证优化器总是选择到合法的Plan

思路

目前解决这个问题的大致思路是：牺牲全局最优，换得plan的合法性，这个做法有点类似PostgreSQL的做法，即：先从RIDTableSpool和DataSpool中选择一个固定下来，随后Primary和Ref都选取同一个TableSpool，以此避免选到非法的plan，为此目前对cascade optimizer的流程做了如下改动：

应用ImplementRule的步骤

在cascade优化器中，这个步骤对应的是OptimizerGroupExprTask，这个Task中，我们保证了Spool的应用顺序，即：PrimarySpool所在的Group总是更先RefSpool所在的Group生成PhysicalPlan，这一步因为我们需要先确定PrimarySpool，如果这一步没有这个保证，那么在JoinReorder等规则等应用下，可能导致RefSpool先被优化，后续需要确定PrimarySpool的时候发现PrimarySpool还未生成PhysicalTableSpool的问题，代码示例如下

void OptimizeGroupExprTask::Execute() {
  RelationBase *rel = group_expr_->GetRelation();
  TableSpool *spool = nullptr;
  if (IsLogicalTableSpool(rel)) {
    spool = static_cast<TableSpool *>(rel);
  }
  
  if (spool && !spool->IsPrimary() && !has_explore_primary_spool_) {
    /* If it is ref table spool, optimize primary spool first */
    has_explore_primary_spool_ = true;
    PushOptTask<OptimizeGroupExprTask>(ctx_, *this);

    Memo *memo = ctx_->GetMemo();
    uint16_t primary_rel_id = spool->GetPrimaryRelID();
    GroupExpr *primary_ge = memo->GetGroupExprByRelId(primary_rel_id);
    ASSERT(primary_ge, "GroupExpr for primary table spool not generated");
    PushOptTask<OptimizeGroupExprTask>(ctx_, primary_ge,
                                       /*explore_only*/ false);
    /**
     * Return now to optimize primary spool first.
     * We will come back here because a duplicated task has been added.
     */
    return;
  }
  DoExecute();
}

计算Physcial Plan的代价时的步骤

在cascade优化器，这个步骤对应的OptimizeInputsTask，该Task中，当我们优化RefSpool时，会先在PrimarySpool对应的group中确定一个可用的Spool，这里根据一个简单的规则来确定RIDSpool和DataSpool：

• 如果能使用rid spool，就选择RIDSpool
• 否则，选择DataSpool，例如：Spool用来缓存Groupby的结果(只能使用DataFmt)
  这个规则很接近最优策略，原因在于：
• RID Spool的优势在于缓存/落盘的数据更少，劣势则需要反复的访问磁盘去取对应的数据，这里实际上是落盘数据量与访问磁盘次数多tradeoff，优劣对比的关键在于用过RID访问此盘数据的代价，在LRU cache足够大的情况下，通过RID访问磁盘的代价变得很低，因此RID是更优的选择。
• 目前优化器的代价模型实际上倾向于使用延迟物化技术(假设lru cache很大，因此选择传递RID)，在这个情况下，选用rid spool是符合优化器的倾向的
• 对于LRU cache不够大的情况，目前考虑到线上小内存实例更多，因此关闭了延迟物化功能，在该情况下，无法出现TableSpool缓存乱序RID的情形，即：缓存顺序RID，或者缓存Data，也符合我们关闭延迟物化的诉求。

随后判断当前优化的RefSpool是否与我们留下的Spool一样，如果当前ref spool能匹配primary group中对应的spool，则计算代价，并返回一个合法的cost，否则不计算cost表示该spool是不合法的。

这个流程带来的问题

删除Spool导致上方算子的requirement无法被满足导致plan生成失败

假设本文开始的例子中PrimaryGroup中含有三个GroupExpr：

由于上文提到的，PrimarySpool总是先被优化以及计算cost，因此经过cost计算之后，可以发现DATA_SPOOL输出的列能够满足RID_SPOOL + COMPUTE_SCALAR，同时cost更低，因此上文中的cost表会被更新

之后如果我们按照OptimizeInputsTask中的步骤所说，删除了DATA_SPOOL，对应的表中DATA_SPOOL相关的cost也需要被删除，因此变成了这样：

本文开头时的图例中，PrimarySpool的parent expr是Groupby，由于Groupby要求输入必须是DATA，但是Group中相关的条目已经被删除，当FindBestPlan通过这个cost table寻找时，由于DATA相关的expr都被删掉了，因此查询会因此找不到可用的plan而报错

HOW TO FIX

上文流程分析中可以看到，主要问题在于，在这个 生成cost table -> 更新cost table -> 删除部分条目的过程中，删除的条目不应该直接从cost table中消除，而是应该用更新前，但是合法的expr与cost补上，也就是说，删除之后，上文的cost table应当是

因此，针对OptimizeInputsTask中对Spool的处理，在删除Spool时增加了以下步骤

• 在删除一个SPOOL时，仍然删除cost table中对应的条目，但是记录下删除条目对应的output property
• 将所有cost table中对应条目的expr包含enforcer全部删除，类似地，也记录下删除条目对应的output property
  • 这是因为DATA_SPOOL被删除之后，enforcer相关的plan可能需要重新计算cost，例如DATA_SPOOL + EXCHANGE可能需要变成RID_SPOOL + CS + EXCHANGE
• 针对所有property，对所有enforcer再下发一个OptimizeInputsTask，目的是重新计算被删除的Property对应的plan的cost
  • 如果是DATA_SPOOL，是不是也要对DATA_SPOOL也要重新下发任务用来更新代价呢？

代码示意如下：

// ....
  bool pick_rid_spool = ShouldPickRidSpool();
  std::vector<PropSet *> erased_psets;
  for (auto it = begin(lowest_cost_exprs); it != end(lowest_cost_exprs);) {
    if (it->second.expr->IsEnforcer()) {
      it = lowest_cost_exprs.erase(it);
	  erased_psets.push_back(it->first);
    } else {
      // set cost erased spool invalid
      if (pick_rid_spool && it->second.expr->GetOutFmt() == OutFmt::DATA) {
        it->second.cost = kCostMaximum;
	    erased_psets.push_back(it->first);
      } else if (!pick_rid_spool &&
                 it->second.expr->GetOutFmt() == OutFmt::RID) {
        it->second.cost = kCostMaximum;
	    erased_psets.push_back(it->first);
      }
      it++;
    }
  }

  PushOptTask<OptimizeInputsTask>(ctx_, *this);
  // reoptimize all enforcer
  for (auto prop_set : erased_psets) {
    for (auto expr : rimary_group->GetPhysicalExprs()) {
      if (!expr->IsEnforcer()) continue;
      expr->ClearGeForReOptimize();
      auto new_ctx = std::make_shared<VolcanoCtx>(*ctx_);
      new_ctx->SetRequiredProperties(prop_set);
      PushOptTask<OptimizeInputsTask>(new_ctx, enforcer);
    }
  }
  return;