论文阅读：Declarative Sub-Operators for Universal Data Processing

TL;DR: Sub-Operators是一个介于SQL与机器码之间的中间层，该团队在LingoDB上实现了相关特性，有助于扩展SQL语句和实现SQL执行引擎

VLDB 23的论文，也是TUM（德国慕尼黑工业大学）的论文

这篇论文是对Database Technology for the Masses: Sub-Operators as First-Class Entities的一个实现

原文下载地址：Full text

什么是Sub-Operators？

全网搜了一下，感觉Sub-Operators这个说法，是他们最早提出来的

找到一篇中文博客，感觉基本能把“什么是Sub-Operators”的问题基本讲清楚

论文阅读：Sub-Operators

这个组还出了个PPT在FG DB 2022讲Sub-Operators（但没找到视频），作为补充资料看看还可以

Sub-Operators in Query Execution with a focus on Joining and Filtering

这里的Opertor应当理解为算子

而Sub-Operator本质是IR（抽象层），充当SQL与硬件的中间层

评论：这个图似曾相识，这个位置之前应该是给MLIR的😅

正文

Section1：介绍

Section2：对于Sub-Operator的需求

这部分内容只是他们另外一篇论文的简要概述：Database Technology for the Masses: Sub-Operators as First-Class Entities

但个人认为：P1通过抽象“减少实现复杂度”这点不好说——因为抽象组件的组合本生也会增加复杂度，P2“Expressing Computations beyond SQL“是个不错的提议——是不是意味着这个方案强绑定MLIR？不懂🙄P3”Execution on Modern Hardware”则是LLVM的特性，LingoDB搭上顺风车罢了

Section3：Sub-Operator的概念设计

（1）使用Tuple Stream

什么样的东西能被成为tuple streams？

Tuple streams（元组流）是指一种数据流处理模型，其中数据以一系列的元组（tuples）形式在网络中流动。每个元组通常包含一组相关的数据项，这些数据项可以是数值、字符串或者其他数据类型。Tuple streams 模型广泛应用于实时数据分析、事件驱动系统和流处理框架中。

在 Tuple streams 模型中，数据流经过一系列处理单元（也称为操作符或算子），每个处理单元对流中的元组进行处理，并可能将处理后的结果发送给下一个处理单元。这种模型非常适合处理实时数据流，因为它可以连续地接收、处理和产生输出。

以下是 Tuple streams 的几个关键概念：

元组（Tuple）：

元组是数据的基本单位，通常由一个或多个字段组成。

元组可以携带有关事件的信息，例如时间戳、事件类型、相关数据等。

流（Stream）：

流是由元组组成的连续序列，代表了一段时间内的数据流。

数据流可以是无限的，也可以是有限的，取决于数据源的性质。

算子（Operator）：

算子是对流中的元组执行特定操作的组件，比如过滤、映射、聚合等。

算子可以串联起来形成复杂的处理管道。

拓扑（Topology）：

拓扑是指算子及其连接方式的图形表示，描述了数据如何在各个算子之间流动。

拓扑可以是线性的、分叉的或循环的，取决于数据流的处理逻辑。

窗口（Window）：

窗口是一个时间段或一组元组，用于定义对数据进行聚合或计算的时间范围。

窗口可以是滑动窗口、滚动窗口或会话窗口等。

一些知名的流处理框架，如 Apache Storm 和 Apache Flink，都支持 Tuple streams 模型。这些框架提供了一种灵活的方式来构建和部署实时数据处理应用。在这些框架中，开发者可以通过定义算子和配置拓扑来构建复杂的流处理应用，而不需要关心底层的细节。

（2）显式声明状态

（3）引用

（4）显式控制流（我没看懂）

（5）视图（给State提供额外属性，使其可以处理复杂结构）

给了一个SQL的分解为Sub-Operator的方案：

Table scan -> Scan

selection -> Map

filter(if condition) -> Tuple Stream Filter

left/right outer join -> inserted explicitly into the multimap

复杂的aggregation操作，则需要需要多个map

Section4：复杂算法的实现

举了groupjoin, complex aggregations, window functions三个例子

Section5：自动并行化

只要能实现Map-Reduce操作就行

提出了一种，有别于plan-driven parallelism和morsel-driven parallelism的方案，值得深究😋

Section6：上述内容在LingoDB的实现

在原有框架的基础上，加上SubOp的部分

Section7：与友商对比与&没有加Sub-Operator的老数据库的部分（自动略）

能被顶会收录，数据肯定都不会难看😛这没意思

如果站要说的话，就是要自圆其说：Sub-Operator的加入给LingoDB带来了什么？结果表明：在效率上几乎没差：实验结果就运行快了20ms，但编译时间也要增加20ms

（插个题外话：见到有人吐槽MLIR编译慢的问题-传送门，如果从事相关研究，后面可能要注意情况）

Section8：相关工作

“Spark’s operators seem similar to the sub-operators we proposed.”

不太了解，值得抽空看看

Section9：未来工作

提供前端界面和可供调用的依赖库（期待😍）
向量化运行（DuckDB在用的，没得说）
分布式和异构（虽然去年也这么说，但一年确实解决不了😂项目在这方面确实取得了进步）
自适应Pipeline

Section10：结语

MLIR op实现内容

对应在include/mlir/Dialect/SubOperator/SubOperatorOps.td目录下

以MultiMap为例，include/mlir/Dialect/SubOperator/SubOperatorOps.td中是这样定义的：

def MultiMap : SubOperator_Type<"MultiMap", "multimap",[State,LookupAbleState]> {
    let summary = "multi map: key -> [values]";
    let parameters = (ins "StateMembersAttr":$keyMembers,"StateMembersAttr":$valueMembers );
    let assemblyFormat = "`<` custom<StateMembers>($keyMembers) `,` custom<StateMembers>($valueMembers) `>`";
    let extraClassDeclaration = [{
        StateMembersAttr getMembers();
    }];
}

lib/SubOperator/SubOperatorTypes.cpp中给出了实现（Attribute是LLVM中定义的）

mlir::subop::StateMembersAttr mlir::subop::MultiMapType::getMembers() {
   std::vector<Attribute> names;
   std::vector<Attribute> types;
   names.insert(names.end(), getKeyMembers().getNames().begin(), getKeyMembers().getNames().end());
   names.insert(names.end(), getValueMembers().getNames().begin(), getValueMembers().getNames().end());
   types.insert(types.end(), getKeyMembers().getTypes().begin(), getKeyMembers().getTypes().end());
   types.insert(types.end(), getValueMembers().getTypes().begin(), getValueMembers().getTypes().end());
   return mlir::subop::StateMembersAttr::get(this->getContext(), mlir::ArrayAttr::get(this->getContext(), names), mlir::ArrayAttr::get(this->getContext(), types));
}

SubOperatorOps.td中的ContinuousView：

def ContinuousView : SubOperator_Type<"ContinuousView", "continuous_view", [State]> {
    let summary = "continuous view type";
    let parameters = (ins "State":$based_on);
    let assemblyFormat = "`<` $based_on `>`";
     let extraClassDeclaration = [{
         StateMembersAttr getMembers(){return getBasedOn().getMembers();}
     }];
}

def CreateContinuousView : SubOperator_Op<"create_continuous_view", [SubOperator]> {
    let arguments = (ins AnyType:$source);
    let results = (outs ContinuousView:$result);
    let assemblyFormat = [{ $source `:` type($source)  `->` type($result) attr-dict }];
    let extraClassDeclaration = [{
      std::vector<std::string> getWrittenMembers();
      std::vector<std::string> getReadMembers();
    }];
}

lib/SubOperator/SubOperatorOps.cpp中实现的方法

std::vector<std::string> subop::CreateContinuousView::getWrittenMembers() {
   std::vector<std::string> res;
   auto names = getSource().getType().cast<mlir::subop::State>().getMembers().getNames();
   for (auto name : names) {
      res.push_back(name.cast<mlir::StringAttr>().str());
   }
   return res;
}
std::vector<std::string> subop::CreateContinuousView::getReadMembers() {
   std::vector<std::string> res;
   auto names = getSource().getType().cast<mlir::subop::State>().getMembers().getNames();
   for (auto name : names) {
      res.push_back(name.cast<mlir::StringAttr>().str());
   }
   return res;
}