Velox表达式计算原理调研

velox背景

velox是Meta的统一的计算引擎，主要使用在Presto、Spark等，velox是由C++实现的向量化计算引擎，其执行引擎包含Task、Driver、Operator等概念；执行引擎有内到外执行，Driver与运行线程对应。Operator执行时使用火山模型-拉的模式依次执行。

velox将Plan转换为由PlanNode组成的一棵树，然后将PlanNode转换为Operator，Operator作为基础的算子，其基类主要定义了addInput、IsBlocked、getOutput等接口来满足数据的处理和流动。

velox表达式

以FilterProject的Operator为例，在Operator中会使用有一个 std::unique_ptr<ExprSet> exprs_的变量，用来执行过滤和投影的计算。ExprSet是FilterProject计算的核心，本文主要研究下ExprSet如何执行计算。

ExprSet是对Expr的封装，Expr表示velox中可执行的表达式。

本文以 cast(a as bigint) > 1 表达式为例，来介绍如何实现Velox表达式的执行，其中包含一些源码引用。

计算目标：RowVector

velox是向量化计算引擎，velox表达式的计算目标是向量，向量在velox用Vector来表示，出于性能和内存占用的考虑，velox有多种编码的Vector来适配不同的场景，例如FlatVector、SimpleVector、DictionaryVector等。

velox中还有一种表示多列向量的结构RowVector；RowVector逻辑看做是列式表模型；在存储上，它是包含列向量Vector的数组；childrens的size对应列的个数。每个列向量的类型可以是FlatVector，也可以是DictionaryVector等。

下面是一个RowVector格式示例，包含三列，类型分别为INTERGER、VARCHAR、VARCHAR。

计算过程

接下来以一个三列的RowVector作为示例，RowVector的逻辑值如下所示，

本文会使用给定表达式：cast(a as bigint) > 1，来调研velox的内部实现。

接下来先抛出几个问题，通过源码的方式来逐步回答如下问题

• 表达式是怎么表示的，又是如何执行？
  
• 是逐行执行，还是列批量执行？
  
• 输入有a、b、c三列，在计算的过程中是否会用到b、c两列？也就是b、c会占用多余的内存？
  
• 如果列a是Dictionary编码，表达式计算会将a物化后计算吗？针对不同的编码有没有优化
  
• 如果列a中转换失败，表达式计算会崩溃吗？抛出exception还是结果为null
  
• 在执行的过程中，velox还有哪些优化措施？
  

表达式数据结构及执行流程

与其他语言的表达式一样，表达式往往使用一棵树来描述，表达式树的静态节点继承自core::ITypedExpr，包含五种节点类型

对于cast(a as bigint) > 1表达式，其对应的表达式树（编译前）如下：

velox对于表达式执行，主要包括表达式编译和执行两部分；表达式编译的过程类似PlanNode转换为Operator的过程，即把执行计划中静态的表达式转换为可执行的表达式实例。

编译

其中表达式编译前是core::ITypedExpr，编译后使用exec::Expr类型表示。

执行

执行过程：使用深度遍历执行即可，因为父节点依赖子节点的执行结果；

Expr：type_表示返回的类型，inputs_表示其孩子节点，如果当前表达式是函数，vectorFunction_表示对应函数的指针。

class Expr {
...
private:
  const TypePtr type_;
  const std::vector<std::shared_ptr<Expr>> inputs_;
  const std::string name_;
  const std::shared_ptr<VectorFunction> vectorFunction_;
  const bool specialForm_;
  const bool supportsFlatNoNullsFastPath_;
  std::vector<VectorPtr> inputValues_;
}

执行实现

执行主要是使用Expr::eval方法进行，函数前面如下：

• 其中rows表示那些行需要参与计算
  
• context包含输入的RowVector和内存池相关的上下文
  
• result表示表达式执行后的结果，类型为VectorPtr
  

class Expr {
...
public:
  void eval(
      const SelectivityVector& rows,
      EvalCtx& context,
      VectorPtr& result,
      const ExprSet* FOLLY_NULLABLE parentExprSet = nullptr);
...
}

EvalCtx的结构这里简单列下其主要成员：

• 其中row_表示表达式的输入；
  
• peeledFields_和peeledEncoding_与剥离逻辑有关，后面在做介绍。
  

class EvalCtx {
  const RowVector* FOLLY_NULLABLE row_;
  bool inputFlatNoNulls_;
  // Corresponds 1:1 to children of 'row_'. Set to an inner vector
  // after removing dictionary/sequence wrappers.
  std::vector<VectorPtr> peeledFields_;
  // Set if peeling was successful, that is, common encodings from inputs were
  // peeled off.
  std::shared_ptr<PeeledEncoding> peeledEncoding_;
}

回到Expr::eval方法，其主要调用栈如下：

• eval
  
• evalEncodings
  
• evalWithNulls
  
• evalAllImpl

• if (isSpecialForm())
  

• evalSpecialFormWithStats(rows, context, result);

• return;
  

• evalArgsDefaultNulls
  
• for (int32_t i = 0; i < inputs_.size(); ++i)
  
• inputs_[i]->eval(remainingRows.rows(), context, inputValues_[i]);
  
• applyFunction
  

从调用的顺序可以看出，velox的表达式执行总体是一个后序遍历的框架，先执行孩子节点的表达式计算，再执行当前节点的applyFunction。

按理说，一个后序遍历执行下每个表达式逻辑不会很复杂，直接在将每个孩子节点的表达式结果放在inputValues_递归调用就可以，为什么中间还有evalEncoding、evalWithNulls、这些中间过程呢？事实上出于性能的考虑、velox对于特定的场景进行了极致的优化。接下来会将前文提到的问题与这些优化进行结合来描述，揭开表达式执行的面纱。

表达式执行细节

evalEncodings实现

在evalEncodings的实现中，首先介绍下DictionaryVector，然后介绍对于DictionaryVector编码的Vector如何进行编码剥离和如何进行剥离。

DictionaryVector简介

velox中大量用到一种Vector类型：DictionaryVector，DictionaryVector是一种字典编码。其背后实现是包含一个dictionaryValues_成员作为内部Vector，indices_记录每一行数据对应内部Vector的字段索引，在有重复值的场景下较为有用。

• 好处：存储使占用内存空间小，计算时可以只对dictionaryValues_操作，减少重复计算。
  
• 坏处：对外层Vector取值时需要decode出来，decode的过程也是通过indices_查找内部Vector的值的过程；同时，Dictionary支持多层嵌套，这种情况下想要获取某一行的值，需要一层一层拨开最内层的vector，其性能可想而知。
  

为了便于对DictionaryVector取值，velox提供了DecodedVector类，支持将DictionaryVector“物化”，其实现正是一层层剥离出来DictionaryVector的最内层Vector。

为什么要剥离?

在表达式计算中假如a列是Dict(Flat)的类型，对于只有一次，假如a的最内层Vector的长度是3，值为["2"、"3"、"5"]；a的长度为1000，值为["2", "3", "3", "3", "5"...]，其取值范围仅限于"2","3","5"；

在执行cast(a as bigint)时，直观的逻辑是遍历a，循环1000次，执行cast(a as bigint)；但是这样不是最高效的；

事实上只需要对内层Vector执行计算，只需要循环3次即可，不需要对1000个物化后的值进行计算，这也是evalEncodings存在的意义，在多层的情况下，比如Dict(Dict(Dict(Flat)))，先物化后计算会更加浪费计算资源；除了DictionaryVector，还有ConstantVector编码也有类似的问题；这里以DictionaryVector的剥离为例：

evalEncodings主要做的事情，是将特殊编码的Vecctor如DictionaryVector背后的值拿出来，而不是直接对外层的逻辑值进行计算（以避免可能的物化代价），其过程具体来说：

• 对每个特定字段，判断是否为Flat类型，如果不是Flat，对encoding进行剥离，得到剥离后的vector和encoding。
  
• 对于剥离后的最内层vector进行计算，得到结果。
  
• 然后将上述结果使用第一步中的encoding进行重新封装。
  

剥离实现

剥离的过程主要使用了PeeledEncoding::peel方法，最后得到VectorPtr的数组，包含的是内层的Vector。

std::vector<VectorPtr> peeledVectors;
  auto peeledEncoding = PeeledEncoding::peel(
      vectorsToPeel, rowsToPeel, localDecoded, propagatesNulls_, peeledVectors);

实现过程是一个do while循环，通过逐个字段（5行）、逐层（20行）剥离，直到最内层不为DICTIONARY编码（13行），完整实现还会有Const类型的处理，这里隐去细节，关注主要逻辑。

do {
    peeled = true;
    BufferPtr firstIndices;
    maybePeeled.resize(numFields);
    for (int fieldIndex = 0; fieldIndex < numFields; fieldIndex++) {
      auto leaf = peeledVectors.empty() ? vectorsToPeel[fieldIndex]
                                        : peeledVectors[fieldIndex];
      if (leaf == nullptr) {
        continue;
      }
      ...
      auto encoding = leaf->encoding();
      if (encoding == VectorEncoding::Simple::DICTIONARY) {
      ...
        setPeeled(leaf->valueVector(), fieldIndex, maybePeeled);
      } else {
        ...
      }
    }
    if (peeled) {
      ++numLevels;
      peeledVectors = std::move(maybePeeled);
    }
  } while (peeled && nonConstant);

最终得到的peeledVectors数组，元素按照字段的序号，最终会放在ExprCtx的peeledFields_中。

ExprCtx怎么用这个剥离后的vector呢？注意到ExprCtx有一个getField方法，是用来获取特定列的vector用于计算；接下来是找到调用getField的地方。

const VectorPtr& EvalCtx::getField(int32_t index) const {
  const VectorPtr* field;
  if (!peeledFields_.empty()) {
    field = &peeledFields_[index];
  } else {
    field = &row_->childAt(index);
  }
  ...
  return *field;
}

回到最开始的表达式执行流程，在执行evalAllImpl时，前面有一句

if (isSpecialForm()) {
    evalSpecialFormWithStats(rows, context, result);
    return;
  }

在我们的cast(a as bigint) > 1 表达式中，其中a对应的执行表达式是FieldReference，其满足isSpecailForm()

class FieldReference : public SpecialForm

所以在执行到FieldReference时（FieldReference是叶子节点，后序遍历会先执行），会调用evalSpecailForm，其实现中会调用到context.getField(index_)（12行）。

通过以上可以看出在获取RowVector的字段取值时，会使用剥离后的内层Vector进行计算。

if (inputs_.empty()) {
    row = context.row();
  } else {
  // ...
  }
  if (index_ == -1) {
    auto rowType = dynamic_cast<const RowType*>(row->type().get());
    VELOX_CHECK(rowType);
    index_ = rowType->getChildIdx(field_);
  }
  VectorPtr child =
      inputs_.empty() ? context.getField(index_) : row->childAt(index_);
  // ...

回顾下整个过程：

• 在eval的最开始先使用了evalEncodings来完成剥离，剥离后的结果放在了context中；
  
• 然后调用evalAllImpl中遍历每一个叶子节点，FieldReference作为叶子节点被执行时，使用的已经是剥离后的结果。
  
• 同时从12行也解决了我们一个疑问, cast(a as bigint) > 1会不会用到b/c字段、答案是不会，只会取index_对应的值；在剥离的过程中会不会用到呢？答案也是不会，因为distinct_fields是根据表达式来计算，而不是输入内容，表达式里面只有a，所以distinct_fields只会剥离a。
  
• 在计算完剥离的数据后，velox还会将原来的encoding在wrap到计算结果中，例如cast(a as bigint)真正执行了3次，真正外部需要的是1000个结果，需要要用wrap encoding。
  

evalWithNulls实现

evalWithNulls顾名思义，是要对null值进行处理，为什么要处理null？总所周知，在SQL中用到的大部分函数对于输入为null的数据，结果也是确定的null，例如 1+null的结果是null；

这种情况下只需要判断表达式的某一行输入是否为null，没必要真正的执行表达式计算。接下来看下velox的evalWithNulls的具体实现流程：

• 判断每一列是否有null值（6行）
  
• 如果有null值（12行），将为null的行去除（14行）后交给evalAll处理，evalAll只对非null的行进行处理（17行）
  
• 处理完以后，在将null值补充到结果中（20行）
  

if (propagatesNulls_ && !skipFieldDependentOptimizations()) {
    bool mayHaveNulls = false;
    for (auto* field : distinctFields_) {
      const auto& vector = context.getField(field->index(context));
      //...
      if (vector->mayHaveNulls()) {
        mayHaveNulls = true;
        break;
      }
    }
    if (mayHaveNulls) {
      LocalSelectivityVector nonNullHolder(context);
      if (removeSureNulls(rows, context, nonNullHolder)) {
        ScopedVarSetter noMoreNulls(context.mutableNullsPruned(), true);
        if (nonNullHolder.get()->hasSelections()) {
          evalAll(*nonNullHolder.get(), context, result);
        }
        auto rawNonNulls = nonNullHolder.get()->asRange().bits();
        addNulls(rows, rawNonNulls, context, result);
        return;
      }
    }
  }

可以看出，velox是简单的将null的行去除，以达到不计算null行的效果。

applyFunction实现

在表达式的所有子节点执行完，会执行applyFunction，说明当前表达式节点是一个函数调用，接下来看下其核心实现：

• 其中包括对ascii字符的优化处理，如果输入全都是ascii，输出也是ascii，则使用函数的callAscii进行更高效的处理。
  
• 然后是核心（18行）调用vectorFunction_->apply来对结果进行处理
  
• 输入是inputValues_数组，该数组长度与函数的表达式孩子节点数相等，作为函数的参数（在上述执行流程中，遍历子节点执行时，结果放在了inputValues_）。
  
• result为输出，结果为VectorPtr
  
• 从这里可以看出vectorFunction_的输入参数是列向量，而非一行行数据传进去。
  

void Expr::applyFunction(
    const SelectivityVector& rows,
    EvalCtx& context,
    VectorPtr& result) {
  stats_.numProcessedVectors += 1;
  stats_.numProcessedRows += rows.countSelected();
  auto timer = cpuWallTimer();
  std::optional<bool> isAscii = std::nullopt;
  if (FLAGS_enable_expr_ascii_optimization) {
    computeIsAsciiForInputs(vectorFunction_.get(), inputValues_, rows);
    isAscii = type()->isVarchar()
        ? computeIsAsciiForResult(vectorFunction_.get(), inputValues_, rows)
        : std::nullopt;
  }
  try {
    vectorFunction_->apply(rows, inputValues_, type(), context, result);
  } catch (const VeloxException& ve) {
    throw;
  } catch (const std::exception& e) {
    VELOX_USER_FAIL(e.what());
  }
  // ...
}

VectorFunction是什么？

从VectorFunction的定义可以看出，apply的输入参数是列向量的列表，在实现VectorFunction时只要实现对VectorFunction的继承即可。

class VectorFunction {
// ...
  virtual void apply(
      const SelectivityVector& rows,
      std::vector<VectorPtr>& args, // Not using const ref so we can reuse args
      const TypePtr& outputType,
      EvalCtx& context,
      VectorPtr& result) const = 0;
}

但是是否所有的velox函数都是通过继承VectorFunction来实现呢？答案是否定的，每个函数在实现时参数都要处理列向量，还是比较复杂的，大部分的函数只需要实现单行的处理逻辑就可以了，其他行遍历执行即可，这种函数在velox中称为SimpleFunction。在一些列向量作为输入优势明显的场景下：比如聚合求值、列向量为Const编码、列向量为Dictionary编码等，可以将函数实现为VectorFunction。

velox的大部分函数是SimpleFunction，实现单行处理的逻辑，最简单的场景下只需要实现call函数即可，

template <typename T>
struct CeilFunction {
  template <typename TOutput, typename TInput = TOutput>
  FOLLY_ALWAYS_INLINE void call(TOutput& result, const TInput& a) {
    if constexpr (std::is_integral_v<TInput>) {
      result = a;
    } else {
      result = ceil(a);
    }
  }
};

以上是SimpleFunction的最简单形式，SimpleFunction虽然是行处理，但是velox依然支持很多函数实现方面的优化：

• Null处理，大部分函数支持null进null出，如果函数希望对于null返回其他值，可以重写callNullable方法，同时还有callNullFree的语法糖。
  
• 确定性：一个函数的输入固定后，输出是确定的，如果希望是不确定性行为，可以设置is_deterministic，比如返回随机数等。
  
• static constexpr bool is_deterministic = false;
  
• Ascii字符快速处理：支持实现callAscii方法，来高效处理输入进是ascii编码的情况。
  
• 字符串零拷贝：通过设置reuse_strings_from_arg，支持重用输入字符串。

• static constexpr int32_t reuse_strings_from_arg = 0;
  

最后一个问题：SimpleFunction是怎么转化为VectorFunction的，毕竟expr中使用的都是VectorFunction，velox是通过一个simpleFunctionAdapter来实现，在注册SimpleFunction函数时，会用到SimpleFunctionAdapterFactoryImpl

// This function should be called once and alone.
template <typename UDFHolder>
void registerSimpleFunction(const std::vector<std::string>& names) {
  mutableSimpleFunctions()
      .registerFunction<SimpleFunctionAdapterFactoryImpl<UDFHolder>>(names);
}

来看SimpleFunctionAdapterFactoryImpl的实现：

template <typename UDFHolder>
class SimpleFunctionAdapterFactoryImpl : public SimpleFunctionAdapterFactory {
 public:
  // Exposed for use in FunctionRegistry
  using Metadata = typename UDFHolder::Metadata;
  explicit SimpleFunctionAdapterFactoryImpl() {}
  std::unique_ptr<VectorFunction> createVectorFunction(
      const core::QueryConfig& config,
      const std::vector<VectorPtr>& constantInputs) const override {
    return std::make_unique<SimpleFunctionAdapter<UDFHolder>>(
        config, constantInputs);
  }
};

可以看出来在createVectorFunction中实现了SimpleFunction到VectorFunction的转化；

这个转化是在哪里产生呢？是在Expr的构造过程中，在ExprPtr compileExpression的编译过程中，simpleFunction会被变换成VectorFunction，放入Expr中

auto simpleFunctionEntry =
            simpleFunctions().resolveFunction(call->name(), inputTypes)) {
      VELOX_USER_CHECK(
          resultType->equivalent(*simpleFunctionEntry->type().get()),
          "Found incompatible return types for '{}' ({} vs. {}) "
          "for input types ({}).",
          call->name(),
          simpleFunctionEntry->type(),
          resultType,
          folly::join(", ", inputTypes));
      auto func = simpleFunctionEntry->createFunction()->createVectorFunction(
          config, getConstantInputs(compiledInputs));
      result = std::make_shared<Expr>(
          resultType,
          std::move(compiledInputs),
          std::move(func),
          call->name(),
          trackCpuUsage);

失败处理

还有一个问题，如果列a中转换失败，表达式计算会崩溃吗？抛出exception还是结果为null

在表达式中，如果cast(a as bigint)，如果a是字符串，转换失败会发生什么？直接来看源码，Cast对应的表达式是CastExpr

class CastExpr : public SpecialForm

接下来看下其evalSpecialForm实现，在CastExpr的转换中多次调用了context.applyToSelectedNoThrow，看函数名字应该是不会抛出exception。

context.applyToSelectedNoThrow(rows, [&](int row) {
// ...
}

事实是这样吗？接下来看其实现，确实handle了exception；

template <typename Callable>
  void applyToSelectedNoThrow(const SelectivityVector& rows, Callable func) {
    rows.template applyToSelected([&](auto row) INLINE_LAMBDA {
      try {
        func(row);
      } catch (const std::exception& e) {
        setError(row, std::current_exception());
      }
    });
  }

看setError的实现：第5行其实抛出了exception，这里是根据EvalCtx的throwOnError_字段进行判断，如果throwOnError_ = true，会抛出exception；否则在addError中会设置错误信息。

void EvalCtx::setError(
    vector_size_t index,
    const std::exception_ptr& exceptionPtr) {
  if (throwOnError_) {
    throwError(exceptionPtr);
  }
  addError(index, toVeloxException(exceptionPtr), errors_);
}

在Expr.h中  bool throwOnError_{true}; 可以看到默认值是true，所以在cast失败时，会抛出exception；如果希望不抛出exception，可以通过ScopedVarSetter设置，在TryExpr.cpp的实现中，我们看到也有类似的调用，设置后，被try包裹的表达式不会抛出exception。

ScopedVarSetter throwOnError(context.mutableThrowOnError(), false);

TryExpr会通过context.errors() 获取表达式的错误，在处理错误的过程中，会将结果设置为null。

一个Expr就是全部吗？

事实上，在velox中，还有一个ExprSet的类，存储了Expr的列表，ExprSet也有一个eval方法，会依次调用Expr列表中eval方法，优点是，多个Expr处理时，可以对公共子表达式只处理一次，这里不在赘述。

在实际的Operator的应用中，ExprSet使用的比较多，而不是直接使用Expr，比如FilterProject这个Operator，使用ExprSet同时存储了Filter的一个Expr，和Project对应的多个Expr。

在velox整个表达式的实现过程中，velox对于不同的场景做了特定的优化，值得学习。表达式执行的过程中“见招拆招”，对于不同的输入，选择更高效的执行路径。velox源码中还有很多细节的处理，限于篇幅和水平，还有一些相关的概念没有涉及到，比如公共子表达式探测、And/OR表达式拍平、常量表达式折叠、SIMD等。

参考

https://github.com/facebookincubator/velox

https://facebookincubator.github.io/velox/develop/expression-evaluation.html

https://facebookincubator.github.io/velox/develop/scalar-functions.html