
![NebulaGraph 源码解读系列｜客户端的通信秘密——fbthrift](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-graph-source-code-reading-07.png)

# 概述

Nebula Clients 给用户提供了多种编程语言的 API 用于和 NebulaGraph 交互，并且对服务端返回的数据结构进行了重新封装，便于用户使用。

目前 Nebula Clients 支持的语言有 C++、Java、Python、Golang 和 Rust。

# 通信框架

Nebula Clients 使用了 `fbthrift` [https://github.com/facebook/fbthrift](https://github.com/facebook/fbthrift) 作为服务端和客户端之间的 RPC 通信框架，实现了跨语言的交互。

fbthrift 提供了三方面的功能：

1. 生成代码：fbthrift 可将不同语言序列化成数据结构
2. 序列化：将生成的数据结构序列化
3. 通信交互：在客户端、服务端之间传输消息，收到不同语言的客户端的请求时，调用相应的服务端函数

# 例子

这里以 Golang 客户端为例，展示 fbthrift 在 NebulaGraph 中的应用。

1. `Vertex` 结构在服务端的定义：
```cpp
struct Vertex {
	Value vid;
	std::vector<Tag> tags;

	Vertex() = default;
};
``` 
2. 首先, 在 `src/interface/common.thrift` 中定义一些数据结构：

```cpp
struct Tag {
		1: binary name,
		// List of <prop_name, prop_value>
		2: map<binary, Value> (cpp.template = "std::unordered_map") props,
} (cpp.type = "nebula::Tag")

struct Vertex {
		1: Value     vid,
		2: list<Tag> tags,
} (cpp.type = "nebula::Vertex")
```

在这里我们定义了一个 Vertex 的结构，其中 `(cpp.type = "nebula::Vertex")` 标注出了这个结构对应了服务端的 `nebula::Vertex`。

3. fbthrift 会自动为我们生成 Golang 的数据结构：
```golang
// Attributes:
//  - Vid
//  - Tags
type Vertex struct {
	Vid *Value `thrift:"vid,1" db:"vid" json:"vid"`
	Tags []*Tag `thrift:"tags,2" db:"tags" json:"tags"`
}

func NewVertex() *Vertex {
	return &Vertex{}
}

...

func (p *Vertex) Read(iprot thrift.Protocol) error { // 反序列化
	...
}

func (p *Vertex) Write(oprot thrift.Protocol) error { // 序列化
	...
}
```
4. 在 `MATCH (v:Person) WHERE id(v) == "ABC" RETURN v` 这条语句中：客户端向服务端请求了一个顶点(`nebula::Vertex`)，服务端找到这个顶点后会进行**序列化**，通过 RPC 通信框架的 `transport` 发送到客户端，在客户端收到这份数据时，会进行**反序列化**，生成对应客户端中定义的数据结构(`type Vertex struct`)。

# 客户端模块

在这个章节会以 nebula-go 为例，介绍客户端的各个模块和其主要接口。

1. 配置类 **Configs**，提供全局的配置选项。
```golang
type PoolConfig struct {
	// 设置超时时间，0 代表不超时，单位 ms。默认是 0
	TimeOut time.Duration
	// 每个连接最大空闲时间，当连接超过该时间没有被使用将会被断开和删除，0 表示永久 idle，连接不会关闭。默认是 0
	IdleTime time.Duration
	// max_connection_pool_size: 设置最大连接池连接数量，默认 10
	MaxConnPoolSize int
	// 最小空闲连接数，默认 0
	MinConnPoolSize int
}
```
2. 客户端会话 **Session**，提供用户直接调用的接口。
```golang
//管理 Session 特有的信息
type Session struct {
	// 用于执行命令的时候的身份校验或者消息重试
	sessionID  int64
	// 当前持有的连接
	connection *connection
	// 当前使用的连接池
	connPool   *ConnectionPool
	// 日志工具
	log        Logger
	// 用于保存当前 Session 所用的时区
	timezoneInfo
}
```
- 接口定义有以下
```golang
	// 执行 nGQL，返回的数据类型为 ResultSet，该接口是非线程安全的。
	func (session *Session) Execute(stmt string) (*ResultSet, error) {...}
	// 重新为当前 Session 从连接池中获取连接
	func (session *Session) reConnect() error {...}
	// 做 signout，释放 Session ID，归还 connection 到 pool
	func (session *Session) Release() {
```
3. 连接池 **ConnectionPool**，管理所有的连接，主要接口有以下
```golang
// 创建新的连接池, 并用输入的服务地址完成初始化
func NewConnectionPool(addresses []HostAddress, conf PoolConfig, log Logger) (*ConnectionPool, error) {...}
// 验证并获取 Session 实例
func (pool *ConnectionPool) GetSession(username, password string) (*Session, error) {...}
```

4.  连接 **Connection**，封装 **thrift** 的网络，提供以下接口
```golang
// 和指定的 ip 和端口的建立连接
func (cn *connection) open(hostAddress HostAddress, timeout time.Duration) error {...}
// 验证用户名和密码
func (cn *connection) authenticate(username, password string) (*graph.AuthResponse, error) {
// 执行 query
func (cn *connection) execute(sessionID int64, stmt string) (*graph.ExecutionResponse, error) {...}
// 通过 SessionId 为 0 发送 "YIELD 1" 来判断连接是否是可用的
func (cn *connection) ping() bool {...}
// 向 graphd 释放 sessionId
func (cn *connection) signOut(sessionID int64) error {...}
// 断开连接
func (cn *connection) close() {...}
```
5. 负载均衡 **LoadBalance**，在连接池里面使用该模块
   - 策略：轮询策略

# 模块交互解析

![模块交互图](https://www-cdn.nebula-graph.com.cn/nebula-blog/client-01.png)

1. 连接池
   - 初始化：
     - 在使用时用户需要先创建并初始化一个**连接池** ConnectionPool，连接池会在初始化时会对用户指定的 Nebula 服务所在地址建立**连接** Connection，如果在用集群部署方式部署了多个 Graph 服务，连接池会采用**轮询的策略**来平衡负载，对每个地址建立近乎等量的连接。
   - 管理连接：
     - 连接池内维护了两个队列，**空闲连接队列** idleConnectionQueue 和**使用中的连接队列** activeConnectionQueue，连接池会定期检测过期空闲的连接并将其关闭。这两个队列在增删元素的时候会通过**读写锁**来确保多线程执行的正确性。
     - 当 Session 向连接池请求连接时，会检查空闲连接队列中是否有可用的连接，如果有则直接返回给 Session 供用户使用；如果没有可用连接并且当前的总连接数没有超过配置中限定的最大连接数，则新建一个连接给 Session；如果已经到达了最大连接数的限制，返回错误。
   - 一般只有在程序退出时才需要关闭连接池, 在关闭时池中所有的连接都会被断开。
2. 客户端会话
   - **客户端会话 Session** 通过连接池生成，用户需要提供用户密码进行校验，在校验成功后用户会获得一个 Session 实例，并通过 Session 中的连接与服务端进行通信。最常用的接口是 `execute()`，如果在执行时发生错误，客户端会检查错误的类型，如果是网络原因则会**自动重连**并尝试再次执行语句。
   - 需要注意，一个 Session **不支持被多个线程同时使用**，正确的方式是用多个线程申请多个 Session，每个线程使用一个 Session。
   - Session 被释放时，其持有的连接会被放回到连接池的**空闲连接队列**中，以便于之后被其他 Session 复用。
3. 连接
   - 每个连接实例都是等价的，可以被任意 Session 持有，**这样设计的目的是这些连接可以被不同的 Session 复用**，减少反复开关 Transport 的开销。
   - 连接会将客户端的请求发送到服务端并将其结果返回给 Session。
4. 用户使用示例
```golang
// Initialize connection pool
pool, err := nebula.NewConnectionPool(hostList, testPoolConfig, log)
if err != nil {
	log.Fatal(fmt.Sprintf("Fail to initialize the connection pool, host: %s, port: %d, %s", address, port, err.Error()))
}
// Close all connections in the pool when program exits
defer pool.Close()

// Create session
session, err := pool.GetSession(username, password)
if err != nil {
	log.Fatal(fmt.Sprintf("Fail to create a new session from connection pool, username: %s, password: %s, %s",
		username, password, err.Error()))
}
// Release session and return connection back to connection pool when program exits
defer session.Release()

// Excute a query
resultSet, err := session.Execute(query)
if err != nil {
	fmt.Print(err.Error())
}
```

# 返回数据结构

客户端对部分复杂的服务端返回的查询结果进行了封装并添加了接口，以便于用户使用。

| 查询结果基本类型 |          封装后的类型          |
| :--------------: | :----------------------------: |
|       Null       |
|       Bool       |
|      Int64       |
|      Double      |
|      String      |
|       Time       |          TimeWrapper           |
|       Date       |
|     DateTime     |        DateTimeWrapper         |
|       List       |
|       Set        |
|       Map        |
|      Vertex      |              Node              |
|       Edge       |          Relationship          |
|       Path       |          PathWrraper           |
|     DateSet      |           ResultSet            |
|        -         | Record(用于ResultSet 的行操作) |

对于 `nebula::Value`，在客户端会被包装成 `ValueWrapper`，并通过接口转换成其他结构。(i.g. `node = ValueWrapper.asNode()`)


## 数据结构的解析

对于语句 `MATCH p= (v:player{name:"Tim Duncan"})-[]->(v2) RETURN p`，返回结果为：
```
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| p                                                                                                                                                                                                                         |
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| <("Tim Duncan" :bachelor{name: "Tim Duncan", speciality: "psychology"} :player{age: 42, name: "Tim Duncan"})<-[:teammate@0 {end_year: 2016, start_year: 2002}]-("Manu Ginobili" :player{age: 41, name: "Manu Ginobili"})> |
+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
Got 1 rows (time spent 11550/12009 us)
```
我们可以看到返回的结果包含了一行，类型是一条路径. 此时如果需要取得路径终点(v2)的属性，可以通过如下操作实现：
```golang
// Excute a query
resultSet, _ := session.Execute("MATCH p= (v:player{name:"\"Tim Duncan"\"})-[]->(v2) RETURN p")

// 获取结果的第一行, 第一行的 index 为0
record, err := resultSet.GetRowValuesByIndex(0)
if err != nil {
	t.Fatalf(err.Error())
}

// 从第一行中取第一列那个 cell 的值
// 此时 valInCol0 的类型为 ValueWrapper 
valInCol0, err := record.GetValueByIndex(0)

// 将 ValueWrapper 转化成 PathWrapper 对象
pathWrap, err = valInCol0.AsPath()

// 通过 PathWrapper 的 GetEndNode() 接口直接得到终点
node, err = pathWrap.GetEndNode()

// 通过 node 的 Properties() 得到所有属性
// props 的类型为 map[string]*ValueWrapper
props, err = node.Properties()
```

# 客户端地址

各语言客户端 GitHub 地址：
- [https://github.com/vesoft-inc/nebula-cpp](https://github.com/vesoft-inc/nebula-cpp)
- [https://github.com/vesoft-inc/nebula-java](https://github.com/vesoft-inc/nebula-java)
- [https://github.com/vesoft-inc/nebula-python](https://github.com/vesoft-inc/nebula-python)
- [https://github.com/vesoft-inc/nebula-go](https://github.com/vesoft-inc/nebula-go)
- [https://github.com/vesoft-inc/nebula-rust](https://github.com/vesoft-inc/nebula-rust)

《开源分布式图数据库NebulaGraph完全指南》，又名：Nebula 小书，里面详细记录了图数据库以及图数据库 NebulaGraph 的知识点以及具体的用法，阅读传送门：[https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf](https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf)

交流图数据库技术？加入 Nebula 交流群请先[填写下你的 Nebula 名片](https://wj.qq.com/s2/8321168/8e2f/)，Nebula 小助手会拉你进群~~


大家都用过 Nebula 好用的客户端，比如 Java 和 Golang 客户端…在本文中你将了解到客户端们背后的神秘力量，借助这股力量你也能做个其他语言的客户端。

NebulaGraph 源码解读系列｜客户端的通信秘密——fbthrift


![NebulaGraph 源码解读系列 ｜ MATCH 中变长 Pattern 的实现](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-graph-source-code-reading-06.png)

# 目录

- 问题分析
    - 定长 Pattern
    - 变长 Pattern 与变长 Pattern 的组合
- 执行计划
    - 拓展一步
    - 拓展多步
    - 保存路径
    - 变长拼接
- 总结

MATCH 作为 openCypher 语言的核心，通过简洁的 Pattern 形式，可以让用户方便地表达图库中的关联关系。变长模式又是 Pattern 中用来描述路径的一种常用形式，对变长模式的支持是 Nebula 兼容 openCypher MATCH 功能的第一步。

由之前的系列文章可以了解到，Nebula 的执行计划是由许多的物理算子组成，每个算子都负责执行特有的计算逻辑，在 MATCH 的实现中也会涉及前述文章中的这些算子，比如 GetNeighbors、GetVertices、Join、Project、Filter、Loop 等等。因为 Nebula 的执行计划不同于关系数据库中的树状结构，在执行的流程上其实是一个有环的图。如何把 MATCH 中的变长 Pattern 变成 Nebula 的物理计划是 Planner 要解决的问题的重点。以下便简单介绍一下在 Nebula 中解决变长 Pattern 问题的思路。

# 问题分析

## 定长 Pattern

在使用 MATCH 语句时，定长 Pattern 也是比较常用的查询形式。如果把定长 Pattern 理解成向外拓展 X 步的变长 Pattern，认为其是后者的一种特例，那么定长和变长 Pattern 的实现便可以统一起来，如下所示：

```shell
// 定长 Pattern MATCH (v)-[e]-(v2)
// 变长 Pattern MATCH (v)-[e*1..1]-(v2)
```

上述示例中的区别就是变量 e 的类型，定长时 e 表示的是一条边，而变长时 e 表示的是长度为 1 的边列表。

## 变长 Pattern 与变长 Pattern 的组合

在 openCypher 的 MATCH 语法里，Pattern 可以灵活的组合以表达复杂路径。如下所示，变长 Pattern 再接变长 Pattern：

```shell
MATCH (v)-[e*1..3]-(v2)-[ee*2..4]-(v3)
```

上述的过程可以是个不断延伸的过程，通过变长定长模式的不同排列，可以组合出非常复杂的路径。所以我们必须找到一种生成 plan 的模式才能方便的递归迭代整个过程。其中需要考虑如下的因素：

1. 后面变长 Pattern 的路径依赖前面所有变长路径；
2. 变长 Pattern 后面的所有的符号（或者变量）表示的结果是“变化”的；
3. 每一步在往外拓展之前需要对起点进行去重；

我们可以注意到，如果可以生成 Pattern 中 `()-[:like*m..n]-` 的部分的执行计划，那么后面继续进行组合迭代就变得有迹可循，如下所示：

```shell
()-[:like*m..n]- ()-[:like*k..l]- ()
 \____________/   \____________/   \_/
    Pattern1         Pattern2       Pattern3
```

# 执行计划

下面便分析模式中 `()-[:like*m..n]-` 的部分，查看其如何转换成 Nebula 的物理执行计划的。上面模式描述的意思是向外拓展 m 到 n 步，在 Nebula 中向外拓展一步是通过 GetNeighbors 算子完成的。如果要向外拓展多步，需要不断在上一步拓展的基础上再调用 GetNeighbors 算子，将每次获取的点边数据首尾连接就会拼接成一个路径（path）。虽然用户最后需要的只是 m 到 n 步的路径，但是在执行的过程中依然需要从第 1 步开始拓展直到第 n 步。并且每步拓展过程中的路径结果都需要保存下来，以便输出或者给下一步使用。最后只要拿出长度在区间 m 到 n 步之间的路径即可。

## 拓展一步

先来看看走一步的计划是什么样子，因为 Nebula 数据存储的方式为起点和出边放置在一起，所以获取起点和出边的数据是不需要跨 partition 的。但是边的终点数据一般是跨 partition 的，需要单独通过 GetVertices 接口来获取点的属性。除此之外，在向外拓展之前，最好要把拓展的起点数据进行去重，避免 storage 重复扫描。所以走一步的执行计划如下图所示：

![拓展一步](https://www-cdn.nebula-graph.com.cn/nebula-blog/pattern-01.png)

## 拓展多步

拓展多步的过程其实就是将上述的过程重复，但是我们会注意到 GetNeighbors 可以获取起点的属性，所以在拓展下一步时，是可以省掉一步 GetVertices 操作。拓展两步的执行计划就变为：

![拓展一步](https://www-cdn.nebula-graph.com.cn/nebula-blog/pattern-two-steps.png)

## 保存路径

由于最后可能需要返回每一步拓展的路径，所以在上述拓展过程中，还需要将所有的路径进行保存。连接两步之间的路径可以通过 join 算子完成。同时因为模式 `()-[e:like*m..n]-` 的返回结果中 e 表示的是一列数据（边的 list），所以上面每步拓展路径需要通过 union 的方式进行结果集的合并。执行计划进一步演变为：

![拓展一步](https://www-cdn.nebula-graph.com.cn/nebula-blog/pattern-three-steps.png)

## 变长拼接

由上面的过程便可以生成模式  `()-[e:like*m..n]-` 的物理计划，当多个类似模式做拼接时，就是再把上述的过程进行迭代。不过在进行模式迭代之前，还需要对上面计划得到的结果进行过滤，因为我们期望是得到 m 到 n 步的结果，上面的数据集中包含了从第 1 步到第 n 步的所有结果，通过对路径的长度做个简单的筛选即可。变长模式拼接之后的计划变为：

![拓展一步](https://www-cdn.nebula-graph.com.cn/nebula-blog/pattern-multi-patterns.png)

通过上述一步步的分解，我们终于得到了最初 MATCH 语句期望的执行计划，可以看到在把一个复杂模式转换成底层的拓展接口时还是颇费功夫。当然上面的计划可以做些优化，比如把多步拓展的过程使用 Loop 算子进行封装，复用一步拓展的 sub-plan，这里不再详细展开。感兴趣的用户可以参考 [nebula 源码实现](https://github.com/vesoft-inc/nebula/blob/master/src/graph/planner/match/Expand.cpp)。

# 总结

上述过程演示了一个变长 Pattern 的 MATCH 语句的执行计划生成过程，相信大家这时会有这样一个疑惑，为什么基本的一些路径拓展在 Nebula 中会生成这么复杂的执行计划？对比 Neo4j 的实现，几个算子即可完成相同的工作，在这里会变成繁琐的 DAG 呢？

这个问题的本质原因是 Nebula 的算子更接近底层的接口，缺少一些更上层的图操作语义上的抽象。算子力度太细，就会导致上层的优化等实现需要考虑太多的细节。后面会对执行算子进一步的梳理，来逐步的完善 MATCH 功能和提升性能。

《开源分布式图数据库NebulaGraph完全指南》，又名：Nebula 小书，里面详细记录了图数据库以及图数据库 NebulaGraph 的知识点以及具体的用法，阅读传送门：[https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf](https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf)

交流图数据库技术？加入 Nebula 交流群请先[填写下你的 Nebula 名片](https://wj.qq.com/s2/8321168/8e2f/)，Nebula 小助手会拉你进群~~


MATCH 作为 Nebula  v2.x 中一个重要的功能，本文主要讲解下 MATCH 中变长 Pattern 的实现。

NebulaGraph 源码解读系列 ｜MATCH 中变长 Pattern 的实现


![NebulaGraph 源码解读系列 ｜ Vol.03 Planner 的实现](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-graph-source-code-reading-05.png)

上篇我们讲述了 Query Engine Optimizer 部分的内容，在本文我们讲解下 Query Engine 剩下的 Scheduler 和 Executor 部分。

# 概述

在执行阶段，执行引擎通过 Scheduler（调度器）将 Planner 生成的物理执行计划转换为一系列 Executor，驱动 Executor 的执行。
Executor，即执行器，物理执行计划中的每个 PlanNode 都会对应一个 Executor。

# 源码定位

调度器的源码在 `src/scheduler` 目录下：
```
src/scheduler
├── AsyncMsgNotifyBasedScheduler.cpp
├── AsyncMsgNotifyBasedScheduler.h
├── CMakeLists.txt
├── Scheduler.cpp
└── Scheduler.h
```
Scheduler 抽象类定义了调度器的公共接口，可以继承该类实现多种调度器。
目前实现了 AsyncMsgNotifyBasedScheduler 调度器，它基于异步消息通信与广度优先搜索避免栈溢出。
执行器的源码在 `src/executor` 目录下：
```
src/executor
├── admin
├── algo
├── CMakeLists.txt
├── ExecutionError.h
├── Executor.cpp
├── Executor.h
├── logic
├── maintain
├── mutate
├── query
├── StorageAccessExecutor.cpp
├── StorageAccessExecutor.h
└── test
```

# 执行过程

首先，调度器从执行计划的根节点开始通过使用广度优先搜索算法遍历整个执行计划并根据节点间的执行依赖关系，构建它们的消息通知机制。
执行时，每个节点收到它的所依赖的节点全部执行完毕的消息后，会被调度执行。一旦自身执行完成，又会发送消息给依赖自己的节点，直至整个计划执行完毕。


```C++
void AsyncMsgNotifyBasedScheduler::runExecutor(
    std::vector<folly::Future<Status>>&& futures,
    Executor* exe,
    folly::Executor* runner,
    std::vector<folly::Promise<Status>>&& promises) const {
    folly::collect(futures).via(runner).thenTry(
        [exe, pros = std::move(promises), this](auto&& t) mutable {
            if (t.hasException()) {
                return notifyError(pros, Status::Error(t.exception().what()));
            }
            auto status = std::move(t).value();
            auto depStatus = checkStatus(std::move(status));
            if (!depStatus.ok()) {
                return notifyError(pros, depStatus);
            }
            // Execute in current thread.
            std::move(execute(exe)).thenTry(
                [pros = std::move(pros), this](auto&& exeTry) mutable {
                    if (exeTry.hasException()) {
                        return notifyError(pros, Status::Error(exeTry.exception().what()));
                    }
                    auto exeStatus = std::move(exeTry).value();
                    if (!exeStatus.ok()) {
                        return notifyError(pros, exeStatus);
                    }
                    return notifyOK(pros);
                });
        });
}
```

每个 Executor 会经历 create-open-execute-close 四个阶段：

## create

根据节点类型生成对应的 Executor。

## open

在 Executor 正式执行前做一些初始化操作，以及慢查询终止和内存水位的判断。
Nebula 支持手动 `kill` 掉某个查询语句的执行，因此每个 Executor 执行前需要检查下当前执行计划状态，若被标记为 `killed`，则终止执行。
每个 Query 类型的 Executor 执行前，还需要检查当前系统所占用内存是否达到内存水位。若达到内存水位，则终止执行，这能在一定程度上避免 OOM。

```C++
Status Executor::open() {
    if (qctx_->isKilled()) {
        VLOG(1) << "Execution is being killed. session: " << qctx()->rctx()->session()->id()
            << "ep: " << qctx()->plan()->id()
            << "query: " << qctx()->rctx()->query();
        return Status::Error("Execution had been killed");
    }
    auto status = MemInfo::make();
    NG_RETURN_IF_ERROR(status);
    auto mem = std::move(status).value();
    if (node_->isQueryNode() && mem->hitsHighWatermark(FLAGS_system_memory_high_watermark_ratio)) {
        return Status::Error(
            "Used memory(%ldKB) hits the high watermark(%lf) of total system memory(%ldKB).",
            mem->usedInKB(),
            FLAGS_system_memory_high_watermark_ratio,
            mem->totalInKB());
    }
    numRows_ = 0;
    execTime_ = 0;
    totalDuration_.reset();
    return Status::OK();
}
```

## execute

Query 类型的 Executor 的输入和输出都是一张表（DataSet）。
Executor 的执行基于迭代器模型：每次计算时，调用输入表的迭代器的 `next()` 方法，获取一行数据，进行计算，直至输入表被遍历完毕。
计算的结果构成一张新表，输出给后续的 Executor 作为输出。

```C++
folly::Future<Status> ProjectExecutor::execute() {
    SCOPED_TIMER(&execTime_);
    auto* project = asNode<Project>(node());
    auto columns = project->columns()->columns();
    auto iter = ectx_->getResult(project->inputVar()).iter();
    DCHECK(!!iter);
    QueryExpressionContext ctx(ectx_);

    VLOG(1) << "input: " << project->inputVar();
    DataSet ds;
    ds.colNames = project->colNames();
    ds.rows.reserve(iter->size());
    for (; iter->valid(); iter->next()) {
        Row row;
        for (auto& col : columns) {
            Value val = col->expr()->eval(ctx(iter.get()));
            row.values.emplace_back(std::move(val));
        }
        ds.rows.emplace_back(std::move(row));
    }
    VLOG(1) << node()->outputVar() << ":" << ds;
    return finish(ResultBuilder().value(Value(std::move(ds))).finish());
}
``` 

如果当前 Executor 的输入表不会被其他 Executor 作为输入时，这些输入表所用的内存会在执行阶段被 drop 掉，减小内存占用。

```C++
void Executor::drop() {
    for (const auto &inputVar : node()->inputVars()) {
        if (inputVar != nullptr) {
            // Make sure use the variable happened-before decrement count
            if (inputVar->userCount.fetch_sub(1, std::memory_order_release) == 1) {
                // Make sure drop happened-after count decrement
                CHECK_EQ(inputVar->userCount.load(std::memory_order_acquire), 0);
                ectx_->dropResult(inputVar->name);
                VLOG(1) << "Drop variable " << node()->outputVar();
            }
        }
    }
}
```

## close

Executor 执行完毕后，将收集到的一些执行信息如执行时间，输出表的行数等添加到 profiling stats 中。
用户可以在 profile 一个语句后显示的执行计划中查看这些统计信息。

```

Execution Plan (optimize time 141 us)

-----+------------------+--------------+-----------------------------------------------------+--------------------------------------
| id | name             | dependencies | profiling data                                      | operator info                       |
-----+------------------+--------------+-----------------------------------------------------+--------------------------------------
|  2 | Project          | 3            | ver: 0, rows: 56, execTime: 147us, totalTime: 160us | outputVar: [                        |
|    |                  |              |                                                     |   {                                 |
|    |                  |              |                                                     |     "colNames": [                   |
|    |                  |              |                                                     |       "VertexID",                   |
|    |                  |              |                                                     |       "player.age"                  |
|    |                  |              |                                                     |     ],                              |
|    |                  |              |                                                     |     "name": "__Project_2",          |
|    |                  |              |                                                     |     "type": "DATASET"               |
|    |                  |              |                                                     |   }                                 |
|    |                  |              |                                                     | ]                                   |
|    |                  |              |                                                     | inputVar: __TagIndexFullScan_1      |
|    |                  |              |                                                     | columns: [                          |
|    |                  |              |                                                     |   "$-.VertexID AS VertexID",        |
|    |                  |              |                                                     |   "player.age"                      |
|    |                  |              |                                                     | ]                                   |
-----+------------------+--------------+-----------------------------------------------------+--------------------------------------
|  3 | TagIndexFullScan | 0            | ver: 0, rows: 56, execTime: 0us, totalTime: 6863us  | outputVar: [                        |
|    |                  |              |                                                     |   {                                 |
|    |                  |              |                                                     |     "colNames": [                   |
|    |                  |              |                                                     |       "VertexID",                   |
|    |                  |              |                                                     |       "player.age"                  |
|    |                  |              |                                                     |     ],                              |
|    |                  |              |                                                     |     "name": "__TagIndexFullScan_1", |
|    |                  |              |                                                     |     "type": "DATASET"               |
|    |                  |              |                                                     |   }                                 |
|    |                  |              |                                                     | ]                                   |
|    |                  |              |                                                     | inputVar:                           |
|    |                  |              |                                                     | space: 318                          |
|    |                  |              |                                                     | dedup: false                        |
|    |                  |              |                                                     | limit: 9223372036854775807          |
|    |                  |              |                                                     | filter:                             |
|    |                  |              |                                                     | orderBy: []                         |
|    |                  |              |                                                     | schemaId: 319                       |
|    |                  |              |                                                     | isEdge: false                       |
|    |                  |              |                                                     | returnCols: [                       |
|    |                  |              |                                                     |   "_vid",                           |
|    |                  |              |                                                     |   "age"                             |
|    |                  |              |                                                     | ]                                   |
|    |                  |              |                                                     | indexCtx: [                         |
|    |                  |              |                                                     |   {                                 |
|    |                  |              |                                                     |     "columnHints": [],              |
|    |                  |              |                                                     |     "index_id": 325,                |
|    |                  |              |                                                     |     "filter": ""                    |
|    |                  |              |                                                     |   }                                 |
|    |                  |              |                                                     | ]                                   |
-----+------------------+--------------+-----------------------------------------------------+--------------------------------------
|  0 | Start            |              | ver: 0, rows: 0, execTime: 1us, totalTime: 19us     | outputVar: [                        |
|    |                  |              |                                                     |   {                                 |
|    |                  |              |                                                     |     "colNames": [],                 |
|    |                  |              |                                                     |     "type": "DATASET",              |
|    |                  |              |                                                     |     "name": "__Start_0"             |
|    |                  |              |                                                     |   }                                 |
|    |                  |              |                                                     | ]                                   |
-----+------------------+--------------+-----------------------------------------------------+--------------------------------------  
```  
以上，源码解析 Query Engine 相关的模块就讲解完毕了，后续将讲解部分特性内容。

如果你在使用 NebulaGraph 过程中遇到任何问题，可以查阅《开源分布式图数据库NebulaGraph完全指南》来排查问题哟~

《开源分布式图数据库NebulaGraph完全指南》，又名：Nebula 小书，里面详细记录了图数据库以及图数据库 NebulaGraph 的知识点以及具体的用法，阅读传送门：[https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf](https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf)

交流图数据库技术？加入 Nebula 交流群请先[填写下你的 Nebula 名片](https://wj.qq.com/s2/8321168/8e2f/)，Nebula 小助手会拉你进群~~


上篇我们讲述了 Query Engine Optimizer 部分的内容，在本文我们讲解下 Query Engine 剩下的 Scheduler 和 Executor 部分。

NebulaGraph 源码解读系列 ｜Scheduler 和 Executor 两兄弟


![NebulaGraph 源码解读系列 ｜ 基于 RBO 的 Optimizer 实现](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-graph-source-code-reading-04.png)

上篇我们讲述了一个执行计划是如何生成的，这次我们来看下这个生成的执行计划是被 Optimizer 优化的。

# 概述

Optimizer，优化器，顾名思义就是一个用来优化执行计划的组件。数据库的优化器通常分为两类，一类是基于规则的优化器 RBO（Rule-basd optimizer），一类是基于代价的优化 CBO（Cost-based optimizer），前者完全基于预设的优化规则进行优化，匹配的条件和优化的结果都比较固定；后者则会根据收集的数据统计信息计算不同执行计划的执行代价，尽量选择代价最小的执行计划。

目前 NebulaGraph 主要实现得是 RBO，所以本文也主要集中讲述 NebulaGraph 中的 RBO 实现。

# 源码定位

优化器的源码实现都在`src/optimizer`目录下面，其中的文件结构如下所示：

```shell
.
├── CMakeLists.txt
├── OptContext.cpp
├── OptContext.h
├── OptGroup.cpp
├── OptGroup.h
├── Optimizer.cpp
├── Optimizer.h
├── OptimizerUtils.cpp
├── OptimizerUtils.h
├── OptRule.cpp
├── OptRule.h
├── rule
│   ├── CombineFilterRule.cpp
│   ├── CombineFilterRule.h
│   ├── EdgeIndexFullScanRule.cpp
│   ├── EdgeIndexFullScanRule.h
|    ....
|
└── test
    ├── CMakeLists.txt
    ├── IndexBoundValueTest.cpp
    └── IndexScanRuleTest.cpp
```

其中`test`目录是测试，`rule`目录则是预设的规则集，其他的源文件则是优化器的具体实现。

而优化器优化查询的入口则在`src/service/QueryInstance.cpp`文件中，如下所示：

```C++
Status QueryInstance::validateAndOptimize() {
    auto *rctx = qctx()->rctx();
    VLOG(1) << "Parsing query: " << rctx->query();
    auto result = GQLParser(qctx()).parse(rctx->query());
    NG_RETURN_IF_ERROR(result);
    sentence_ = std::move(result).value();

    NG_RETURN_IF_ERROR(Validator::validate(sentence_.get(), qctx()));
    NG_RETURN_IF_ERROR(findBestPlan());

    return Status::OK();
}
```

`findBestPlan`函数会调用优化器，优化并返回一个全新的优化过的执行计划。

# 优化过程简述

NebulaGraph 目前设计的执行计划从拓扑角度来讲是一个**有向无环图**，通过每个节点指向它的依赖节点来组织，理论上每个节点可以指定任意节点的结果作为输入，但是使用一个还没执行的节点的结果是没有意义的，所以在生成执行计划的时候会**限制只能使用已经执行过的节点作为输入**。同时，执行计划也执行循环和条件分支这样的特殊节点。如图1 所示，该执行计划由查询语句`GO 2 STEPS FROM 'Tim Duncan' OVER like`产生。

![NebulaGraph 源码解读系列 ｜ 基于 RBO 的 Optimizer 实现](https://www-cdn.nebula-graph.com.cn/nebula-blog/optimizer-plan-generator.png)

图1

优化器目前的主要功能就是根据预设模式在执行计划中进行匹配，如果匹配成功，再调用相应的转换函数将匹配到的部分执行计划按预设的规则进行转换。比如，将 `GetNeighbor` → `Limit` 形式的执行计划转换成 limit 下推的`GetNeighbor`算子，实现算子下推优化。

# 具体实现

首先，优化器不会直接在执行计划上操作，而是先将执行执行计划转换成`OptGroup`、`OptGroupNode`。`OptGroup` 代表的是一个单独的优化组（通常指一个或多个同等的算子），`OptGroupNode` 则是代表一个独立的算子，同时还有指向依赖以及分支的指针，也就是说`OptGroupNode` **保留了执行计划的拓扑结构**。之所以要做这样的结构转换，主要是抽象出执行计划的拓扑结构，屏蔽掉一些不需要执行细节（比如循环和条件分支），以及在新的结构中方便保存一些规则匹配的上下文。

转换过程基本上是一个简单的**先序遍历**，并在遍历的过程中把算子转换成对应的`OptGroup`以及`OptGroupNode`。为了方便描述，这里把`OptGroup`以及`OptGroupNode`组成的结构称为优化计划，和执行计划做区分。

转换完成后就会开始匹配规则以及做相应的优化计划转换。这里会遍历所有预定义的规则，而每个规则都会在在优化计划上做一个 Bottom-Up 的遍历匹配，具体来说就是从最叶子层`OptGroup`开始，一直到根节点的`OptGroup`，在每个`OptGroup`节点上对节点内的`OptGroupNode`做 Top-Down 的遍历来进行规则模式的匹配。

如图2 所示，这里要匹配的模式是`Limit`->`Project`→`GetNeighbors`，按照 Bottom-Up 的顺序，首先在`Start`节点按照 Top-Down 的顺序匹配，`Start`不等于`Limit`匹配失败，然后从`GetNeighbors`开始同样 Top-Down 匹配失败，直到`Limit`开始才匹配成功。匹配成功后，会根据规则定义的`transform`函数，将匹配到的部分优化计划进行转换，比如图2 会将`Limit`和`GetNeighbors`合并。

![NebulaGraph 源码解读系列 ｜ 基于 RBO 的 Optimizer 实现](https://www-cdn.nebula-graph.com.cn/nebula-blog/optimizer-transform.jpeg)

图2

最后，优化器会把已经完成优化的优化计划重新转换成执行计划，这里和第一步相反，不过也是一个**递归遍历转换**的过程。

# 如何添加新规则

在前面的文章中，我们了解整个优化器组件的实现部分，不过对于添加优化规则来说并不需要了解太多优化器的实现细节，只需要了解如何定义新规则即可。这里，我们以`Limit`下推为例讲解一个典型的优化规则的实现。`Limit`下推规则的源码详见`src/optimizer/rule/LimitPushDownRule.cpp`文件：

```C++

std::unique_ptr<OptRule> LimitPushDownRule::kInstance =
    std::unique_ptr<LimitPushDownRule>(new LimitPushDownRule());

LimitPushDownRule::LimitPushDownRule() {
    RuleSet::QueryRules().addRule(this);
}

const Pattern &LimitPushDownRule::pattern() const {
    static Pattern pattern =
        Pattern::create(graph::PlanNode::Kind::kLimit,
                        {Pattern::create(graph::PlanNode::Kind::kProject,
                                         {Pattern::create(graph::PlanNode::Kind::kGetNeighbors)})});
    return pattern;
}
```

```C++
StatusOr<OptRule::TransformResult> LimitPushDownRule::transform(
    OptContext *octx,
    const MatchedResult &matched) const {
    auto limitGroupNode = matched.node;
    auto projGroupNode = matched.dependencies.front().node;
    auto gnGroupNode = matched.dependencies.front().dependencies.front().node;

    const auto limit = static_cast<const Limit *>(limitGroupNode->node());
    const auto proj = static_cast<const Project *>(projGroupNode->node());
    const auto gn = static_cast<const GetNeighbors *>(gnGroupNode->node());

    int64_t limitRows = limit->offset() + limit->count();
    if (gn->limit() >= 0 && limitRows >= gn->limit()) {
        return TransformResult::noTransform();
    }

    auto newLimit = static_cast<Limit *>(limit->clone());
    auto newLimitGroupNode = OptGroupNode::create(octx, newLimit, limitGroupNode->group());

    auto newProj = static_cast<Project *>(proj->clone());
    auto newProjGroup = OptGroup::create(octx);
    auto newProjGroupNode = newProjGroup->makeGroupNode(newProj);

    auto newGn = static_cast<GetNeighbors *>(gn->clone());
    newGn->setLimit(limitRows);
    auto newGnGroup = OptGroup::create(octx);
    auto newGnGroupNode = newGnGroup->makeGroupNode(newGn);

    newLimitGroupNode->dependsOn(newProjGroup);
    newProjGroupNode->dependsOn(newGnGroup);
    for (auto dep : gnGroupNode->dependencies()) {
        newGnGroupNode->dependsOn(dep);
    }

    TransformResult result;
    result.eraseAll = true;
    result.newGroupNodes.emplace_back(newLimitGroupNode);
    return result;
}

std::string LimitPushDownRule::toString() const {
    return "LimitPushDownRule";
}
```

定义一个规则首先先继承`OptRule`类。然后，实现`pattern`接口，这里要求返回需要匹配的模式，模式是算子和算子的依赖组成，比如`Limit`->`Project`->`GetNeighbors`。然后需要实现`transform`接口，`transform`接口会传入一个匹配的优化计划，我们根据预定义的模式来解析匹配到的优化计划，并对优化计划做相应的优化转换，比如把`Limit`算子合并到`GetNeighbors`算子，最后返回优化过的优化计划计划即可。

只需要正确实现这两个接口，我们的新的优化规则就可以正常工作了。

以上，为 NebulaGraph Optimizer 的介绍。

如果你在使用 NebulaGraph 过程中遇到任何问题，可以查阅《开源分布式图数据库NebulaGraph完全指南》来排查问题哟~《开源分布式图数据库NebulaGraph完全指南》，又名：Nebula 小书，里面详细记录了图数据库以及图数据库 NebulaGraph 的知识点以及具体的用法，阅读传送门：[https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf](https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf)

交流图数据库技术？加入 Nebula 交流群请先[填写下你的 Nebula 名片](https://wj.qq.com/s2/8321168/8e2f/)，Nebula 小助手会拉你进群~~


上篇我们讲述了一个执行计划是如何生成的，这次我们来看下这个生成的执行计划是被 Optimizer 优化的。

NebulaGraph 源码解读系列 ｜ 基于 RBO 的 Optimizer 实现

上篇我们讲到 Validator 会将由 Parser 生成的抽象语法树（AST）转化为执行计划，这次，我们来讲下执行计划是如何生成的。

NebulaGraph 源码解读系列 ｜ Planner 的实现


![NebulaGraph 源码解读系列 ｜ Vol.01 详解 Validator](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-graph-source-code-reading-02.png)

# 整体架构  

NebulaGraph Query Engine 主要分为四个模块，分别是 Parser、Validator、Optimizer 和 Executor。  

![](https://www-cdn.nebula-graph.com.cn/nebula-blog/code-reading-validator.png)

Parser 完成对语句的词法语法解析并生成抽象语法树（AST），Validator 会将 AST 转化为执行计划，Optimizer 对执行计划进行优化，而 Executor 负责实际数据的计算。

这篇文章我们主要介绍 Validator 的实现原理。

# 目录结构

Validator 代码实现在 `src/validator` 和 `src/planner` 目录。

`src/validator` 目录主要包括各种子句的 Validator 实现，比如 `OrderByValidator`、`LimitValidator`、`GoValidator` 等等。

``` 
validator/
├── ACLValidator.h
├── AdminJobValidator.h
├── AdminValidator.h
├── AssignmentValidator.h
├── BalanceValidator.h
├── DownloadValidator.h
├── ExplainValidator.h
├── FetchEdgesValidator.h
├── FetchVerticesValidator.h
├── FindPathValidator.h
├── GetSubgraphValidator.h
├── GoValidator.h
├── GroupByValidator.h
├── IngestValidator.h
├── LimitValidator.h
├── LookupValidator.h
├── MaintainValidator.h
├── MatchValidator.h
├── MutateValidator.h
├── OrderByValidator.h
├── PipeValidator.h
├── ReportError.h
├── SequentialValidator.h
├── SetValidator.h
├── TraversalValidator.h
├── UseValidator.h
├── Validator.h
└── YieldValidator.h 
```

src/planner/plan 目录**定义了所有 PlanNode 的数据结构，用于生成最终的执行计划**。比如，当查询语句中含有聚合函数时，执行计划中会生成 Aggregate 节点，Aggregate 类会指定聚合函数计算时所需的全部信息，包括分组列和聚合函数表达式，Aggregate 类定义在 Query.h 中。Nebula 定义了一百多种 PlanNode，`PlanNode::kind` 定义在 PlanNode.h 中，在此不做详细阐述。

``` 
planner/plan/
├── Admin.cpp          
├── Admin.h             // administration related  nodes
├── Algo.cpp
├── Algo.h              // graph algorithm related nodes
├── ExecutionPlan.cpp
├── ExecutionPlan.h     // explain and profile nodes
├── Logic.cpp
├── Logic.h             // nodes introduced by the implementation layer
├── Maintain.cpp
├── Maintain.h          // schema related nodes
├── Mutate.cpp
├── Mutate.h            // DML related nodes
├── PlanNode.cpp
├── PlanNode.h          // plan node base classes
├── Query.cpp
├── Query.h             // DQL related nodes
└── Scan.h              // index related nodes
 ```

src/planner 目录还定义了 nGQL 和 match 语句的 planner 实现，用于生成 nGQL 和 match 语句执行计划。


# 源码解析

validator 入口函数是 `Validator::validate(Sentence*, QueryContext*)`，负责将 parser 生成的抽象语法树转化为执行计划，QueryContext 中会保存最终生成的执行计划 root 节点，函数代码如下：

```cpp 
Status Validator::validate(Sentence* sentence, QueryContext* qctx) {
    DCHECK(sentence != nullptr);
    DCHECK(qctx != nullptr);

    // Check if space chosen from session. if chosen, add it to context.
    auto session = qctx->rctx()->session();
    if (session->space().id > kInvalidSpaceID) {
        auto spaceInfo = session->space();
        qctx->vctx()->switchToSpace(std::move(spaceInfo));
    }

    auto validator = makeValidator(sentence, qctx);
    NG_RETURN_IF_ERROR(validator->validate());

    auto root = validator->root();
    if (!root) {
        return Status::SemanticError("Get null plan from sequential validator");
    }
    qctx->plan()->setRoot(root);
    return Status::OK();
} 
```

该函数首先获取当前 session 的 space 信息并保存在 ValidateContext中，之后调用 `Validator::makeValidator()` 和 `Validator::validate()` 函数。
 
`Validator::makeValidator()` 的功能是生成子句的 validator，该函数会首先生成 SequentialValidator，SequentialValidator 是 validator 的入口，所有语句都会首先生成 SequentialValidator。

`SequentialValidator::validateImpl()` 函数会调用 `Validator::makeValidator()` 生成相应子句的 validator。函数代码如下：

```cpp
Status SequentialValidator::validateImpl() {
    Status status;
    if (sentence_->kind() != Sentence::Kind::kSequential) {
        return Status::SemanticError(
                "Sequential validator validates a SequentialSentences, but %ld is given.",
                static_cast<int64_t>(sentence_->kind()));
    }
    auto seqSentence = static_cast<SequentialSentences*>(sentence_);
    auto sentences = seqSentence->sentences();

    seqAstCtx_->startNode = StartNode::make(seqAstCtx_->qctx);
    for (auto* sentence : sentences) {
        auto validator = makeValidator(sentence, qctx_);
        NG_RETURN_IF_ERROR(validator->validate());
        seqAstCtx_->validators.emplace_back(std::move(validator));
    }

    return Status::OK();
}
 ```

同样地，PipeValidator、AssignmentValidator 和 SetValidator 也会生成相应子句的 validator。

`Validator::validate()` 负责生成执行计划，函数代码如下：

```cpp
Status Validator::validate() {
    auto vidType = space_.spaceDesc.vid_type_ref().value().type_ref().value();
    vidType_ = SchemaUtil::propTypeToValueType(vidType);

    NG_RETURN_IF_ERROR(validateImpl());

    // Check for duplicate reference column names in pipe or var statement
    NG_RETURN_IF_ERROR(checkDuplicateColName());

    // Execute after validateImpl because need field from it
    if (FLAGS_enable_authorize) {
        NG_RETURN_IF_ERROR(checkPermission());
    }

    NG_RETURN_IF_ERROR(toPlan());

    return Status::OK();
}
```

该函数首先检查 space 和用户权限等信息，之后调用函数 `Validator:validateImpl()` 完成子句校验，`validateImpl()` 函数是 Validator 类的纯虚函数，利用多态调用不同子句的 `validatorImpl()` 实现函数。最后调用 `Validator::toPlan()` 函数生成执行计划，`toPlan()` 函数会生成子句的执行计划，子执行计划会被连接形成完整的执行计划，比如 match 语句中通过函数 `MatchPlanner::connectSegments()` 连接子执行计划，而 nGQL 语句则通过 `Validator::appendPlan()` 实现。


# 举例

下面我们以 nGQL 语句为例具体介绍一下以上流程。

语句：

```
GO 3 STEPS FROM "vid" OVER edge 
WHERE $$.tag.prop > 30 
YIELD edge._dst AS dst 
| ORDER BY $-.dst
```
这条 nGQL 语句在 validator 阶段主要经历三个过程：

## 制作子句 validator

首先会调用 `Validator::makeValidator()` 生成 SequentialValidator。在 `SequentialValidator::validateImpl()` 函数中会生成 PipeValidator，PipeValidator 会制作左右子句的 validator，分别是 GoValidator 和 OrderByValidator。
 
## 子句校验

子句校验阶段会分别校验 Go 和 OrderBy 子句。

以 Go 语句为例，会先校验语义错误，比如 aggregate 函数使用不当、表达式类型不匹配等等，然后依次校验内部子句，校验过程中会把校验的中间结果保存在 GoContext 中，作为 GoPlanner 生成执行计划的依据。比如 validateWhere() 会保存过滤条件表达式用于之后生成 Filter 执行计划节点。

```cpp
    NG_RETURN_IF_ERROR(validateStep(goSentence->stepClause(), goCtx_->steps));  // 校验 step 子句
    NG_RETURN_IF_ERROR(validateStarts(goSentence->fromClause(), goCtx_->from)); // 校验 from 子句
    NG_RETURN_IF_ERROR(validateOver(goSentence->overClause(), goCtx_->over));   // 校验 over 子句
    NG_RETURN_IF_ERROR(validateWhere(goSentence->whereClause()));               // 校验 where 子句
    NG_RETURN_IF_ERROR(validateYield(goSentence->yieldClause()));               // 校验 yield 子句
``` 

## plan 生成
  
Go 语句的子执行计划由 GoPlanner::transform(Astcontext*) 函数生成，代码如下：

```cpp
StatusOr<SubPlan> GoPlanner::transform(AstContext* astCtx) {
    goCtx_ = static_cast<GoContext *>(astCtx);
    auto qctx = goCtx_->qctx;
    goCtx_->joinInput = goCtx_->from.fromType != FromType::kInstantExpr;
    goCtx_->joinDst = !goCtx_->exprProps.dstTagProps().empty();

    SubPlan startPlan = QueryUtil::buildStart(qctx, goCtx_->from, goCtx_->vidsVar);

    auto& steps = goCtx_->steps;
    if (steps.isMToN()) {
        return mToNStepsPlan(startPlan);
    }

    if (steps.steps() == 0) {
        auto* pt = PassThroughNode::make(qctx, nullptr);
        pt->setColNames(std::move(goCtx_->colNames));
        SubPlan subPlan;
        subPlan.root = subPlan.tail = pt;
        return subPlan;
    }

    if (steps.steps() == 1) {
        return oneStepPlan(startPlan);
    }
    return nStepsPlan(startPlan);
}
```

该函数首先调用 QueryUtil::buildStart() 构造start 节点，然后根据四种不同 step 的情况采用不同的方式生成计划。本例中语句会采用 nStepPlan 策略。

GoPlanner::nStepsPlan() 函数代码如下：

```cpp
SubPlan GoPlanner::nStepsPlan(SubPlan& startVidPlan) {
    auto qctx = goCtx_->qctx;

    auto* start = StartNode::make(qctx);
    auto* gn = GetNeighbors::make(qctx, start, goCtx_->space.id);
    gn->setSrc(goCtx_->from.src);
    gn->setEdgeProps(buildEdgeProps(true));
    gn->setInputVar(goCtx_->vidsVar);

    auto* getDst = QueryUtil::extractDstFromGN(qctx, gn, goCtx_->vidsVar);

    PlanNode* loopBody = getDst;
    PlanNode* loopDep = nullptr;
    if (goCtx_->joinInput) {
        auto* joinLeft = extractVidFromRuntimeInput(startVidPlan.root);
        auto* joinRight = extractSrcDstFromGN(getDst, gn->outputVar());
        loopBody = trackStartVid(joinLeft, joinRight);
        loopDep = joinLeft;
    }

    auto* condition = loopCondition(goCtx_->steps.steps() - 1, gn->outputVar());
    auto* loop = Loop::make(qctx, loopDep, loopBody, condition);

    auto* root = lastStep(loop, loopBody == getDst ? nullptr : loopBody);
    SubPlan subPlan;
    subPlan.root = root;
    subPlan.tail = startVidPlan.tail == nullptr ? loop : startVidPlan.tail;

    return subPlan;
}
```

Go 语句生成的子执行计划如下：
```
Start -> GetNeighbors -> Project -> Dedup -> Loop -> GetNeighbors -> Project -> GetVertices -> Project -> LeftJoin -> Filter -> Project
```

Go 语句的功能是完成图的拓展，GetNeighbors 是执行计划中最重要的节点，GetNeighbors 算子会在运行期访问存储服务，拿到通过起点和指定边类型一步拓展后终点的 id，多步拓展通过 Loop 节点实现，Start 到 Loop 之间是 Loop 子计划，当满足条件时 Loop 子计划会被循环执行，最后一步拓展节点在 Loop 外实现。Project 节点用来获取当前拓展的终点 id，Dedup 节点对终点 id 进行去重后作为下一步拓展的起点。GetVertices 节点负责取终点 tag 的属性，Filter 做条件过滤，LeftJoin 的作用是合并 GetNeightbors 和 GetVertices 的结果。

OrderBy 语句的功能是对数据进行排序，子执行计划会生成 Sort 节点。

左右子句计划生成之后，PipeValidator::toPlan() 函数会调用 Validator::appendPlan() 连接左右子计划并得到最终的执行计划。完整执行计划如下：
```
Start -> GetNeighbors -> Project -> Dedup -> Loop -> GetNeighbors -> Project -> GetVertices -> Project -> LeftJoin -> Filter -> Project -> Sort -> DataCollect 
```

以上 Validator 部分就介绍完毕。

## 论坛相关问题


> 问：如何找寻 parser/GraphParser.hpp 文件

答：.h 文件是由编译时产生的文件，编译一次就有文件了。

以上为本篇文章的介绍内容。

交流图数据库技术？加入 Nebula 交流群请先[填写下你的 Nebula 名片](https://wj.qq.com/s2/8321168/8e2f/)，Nebula 小助手会拉你进群~~


在 NebulaGraph Query Engine 中 Validator 主要负责将 Parser 生成的 AST 转化为执行计划，这次，我们来详细讲解下 Validator 的实现原理。

NebulaGraph 源码解读系列 ｜ 详解 Validator


![NebulaGraph 源码解读系列 ｜ Vol.01 NebulaGraph Overview](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-graph-source-code-reading-01.png)

上篇序言中我们讲述了源码解读系列的由来，在 NebulaGraph Overview 篇中我们将带你了解下 NebulaGraph 的架构以及代码仓分布、代码结构和模块规划。

# 1. 架构

NebulaGraph 是一个开源的分布式图数据库。Nebula采用存储计算分离的设计，解耦存储与计算。同时在数据库内核之外，我们也提供了很多周边工具，比如数据导入，监控，部署，可视化，图计算等等。  
Nebula 设计请参见[《图数据库综述与 Nebula 在图数据库设计的实践》](https://nebula-graph.com.cn/posts/nebula-graph-design-in-practice/)。  

整体架构设计如下图所示：  

![](https://docs-cdn.nebula-graph.com.cn/docs-2.0/1.introduction/2.nebula-graph-architecture/nebula-graph-architecture-1.png) 
  
查询引擎采用无状态设计，可轻松实现横向扩展，分为语法分析、语义分析、优化器、执行引擎等几个主要部分。  

详细设计参见[《图数据库的查询引擎设计》](https://nebula-graph.com.cn/posts/nebula-graph-query-engine-overview/)，[《初识 NebulaGraph 2.0 Query Engine》](https://nebula-graph.com.cn/posts/nebula-query-engine-introduction/)。  

查询引擎架构设计如下图所示：  

![](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-reading-qe-architecture.png)  

Storage 包含两个部分， 一是 meta 相关的存储， 我们称之为 Meta Service ，另一个是 data 相关的存储， 我们称之为 Storage Service。 

Storage Service 共有三层：最底层是 Store Engine；之上便是我们的 Consensus 层，实现了 Multi Group Raft；最上层，便是我们的 Storage interfaces，这一层定义了一系列和图相关的 API。  

详细设计参见[《图数据库的存储设计》](https://nebula-graph.com.cn/posts/nebula-graph-storage-engine-overview/)  。  

存储引擎架构设计如下图所示：  

![](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-reading-storage-architecture.png)

# 2. 代码仓库概览　　
　　
欢迎来到 [vesoft](https://github.com/vesoft-inc) 代码仓库（vesoft 为图数据库 NebulaGraph 开发商）。

目前 Nebula 产品架构中，包含了图数据库内核，客户端，工具，测试框架，编译，可视化，监控等。

本文的主要目的是简单介绍 NebulaGraph 主要 Repo 的代码结构，并说明各个模块的基本功能。后续会有更多的详细设计说明。希望能够帮助到社区读者更好地理解 NebulaGraph，并能够为 Nebula 社区做出自己的贡献，比如提交 Feature，修复 Bug，提交文档等。  

以下列出 vesoft-inc 仓库中大部分的代码仓库：
- [nebula](https://github.com/vesoft-inc/nebula)：Nebula 1.0 的内核代码
- [nebula graph](https://github.com/vesoft-inc/nebula-graph)：Nebula 2.0 查询计算引擎
- [nebula storage](https://github.com/vesoft-inc/nebula-storage)：Nebula 2.0 存储引擎
- [nebula common](https://github.com/vesoft-inc/nebula-common)：Nebula 2.0 内核工具包
- Nebula Clients
    - [nebula-java](https://github.com/vesoft-inc/nebula-java)：Java 客户端
    - [nebula-cpp](https://github.com/vesoft-inc/nebula-cpp)：CPP 客户端
    - [nebula-go](https://github.com/vesoft-inc/nebula-go)：Go 客户端
    - [nebula-python](https://github.com/vesoft-inc/nebula-python)：Python 客户端
- Nebula Tools
    - [nebula-importer](https://github.com/vesoft-inc/nebula-importer)：基于 Go 客户端实现的高性能数据导入工具
    - [nebula-spark-utils](https://github.com/vesoft-inc/nebula-spark-utils)：收录工具 Spark Connector、Exchange、Algorithm
    - [nebula-br](https://github.com/vesoft-inc/nebula-br)：备份恢复工具
    - [nebula-ansible](https://github.com/vesoft-inc/nebula-ansible)、[nebula-operator](https://github.com/vesoft-inc/nebula-operator)：部署工具
- Nebula Test
    - [nebula-bench](https://github.com/vesoft-inc/nebula-bench)：压力与性能测试工程
    - [nebula-chaos](https://github.com/vesoft-inc/nebula-chaos)：混沌测试工程
- Compiling
    - [nebula-third-party](https://github.com/vesoft-inc/nebula-third-party)：NebulaGraph 图数据库内核依赖的第三方包
    - [nebula-gears](https://github.com/vesoft-inc/nebula-gears)：NebulaGraph 图数据库内核工具链
- [nebula-graph-studio](https://github.com/vesoft-inc/nebula-graph-studio)：NebulaGraph 可视化工具
  
# 3. 代码结构及模块说明

## 3.1 NebulaGraph
```
├── cmake
├── conf
├── LICENSES
├── package
├── resources
├── scripts
├── src
│   ├── context
│   ├── daemons
│   ├── executor
│   ├── optimizer
│   ├── parser
│   ├── planner
│   ├── scheduler
│   ├── service
│   ├── session
│   ├── stats
│   ├── util
│   ├── validator
│   └── visitor
└── tests
    ├── admin
    ├── bench
    ├── common
    ├── data
    ├── job
    ├── maintain
    ├── mutate
    ├── query
    └── tck
```  

- conf/：查询引擎配置文件目录
- package/：graph 打包脚本
- resources/：资源文件
- scripts/：启动脚本
- src/：查询引擎源码目录
    - src/context/：查询的上下文信息，包括 AST（抽象语法树），Execution Plan（执行计划），执行结果以及其他计算相关的资源。
    - src/daemons/：查询引擎主进程
    - src/executor/：执行器，各个算子的实现
    - src/optimizer/：RBO（基于规则的优化）实现，以及优化规则
    - src/parser/：词法解析，语法解析，：AST结构定义
    - src/planner/：算子，以及执行计划生成
    - src/scheduler/：执行计划的调度器
    - src/service/：查询引擎服务层，提供鉴权，执行 Query 的接口
    - src/session/：Session 管理
    - src/stats/：执行统计，比如 P99、慢查询统计等
    - src/util/：工具函数
    - src/validator/：语义分析实现，用于检查语义错误，并进行一些简单的改写优化
    - src/visitor/：表达式访问器，用于提取表达式信息，或者优化
- tests/：基于 BDD 的集成测试框架，测试所有 NebulaGraph 提供的功能


## 3.2 Nebula Storage

```
├── cmake
├── conf
├── docker
├── docs
├── LICENSES
├── package
├── scripts
└── src
    ├── codec
    ├── daemons
    ├── kvstore
    ├── meta
    ├── mock
    ├── storage
    ├── tools
    ├── utils
    └── version
```  

- conf/：存储引擎配置文件目录
- package/：storage 打包脚本
- scripts/：启动脚本
- src/：存储引擎源码目录
    - src/codec/：序列化反序列化工具
    - src/daemons/：存储引擎和元数据引擎主进程
    - src/kvstore/：基于 raft 的分布式 KV 存储实现
    - src/meta/：基于 KVStore 的元数据管理服务实现，用于管理元数据信息，集群管理，长耗时任务管理等
    - src/storage/：基于 KVStore 的图数据存储引擎实现
    - src/tools/：一些小工具实现
    - src/utils/：代码工具函数

## 3.3 Nebula Common

```
├── cmake
│   └── nebula
├── LICENSES
├── src
│   └── common
│       ├── algorithm
│       ├── base
│       ├── charset
│       ├── clients
│       ├── concurrent
│       ├── conf
│       ├── context
│       ├── cpp
│       ├── datatypes
│       ├── encryption
│       ├── expression
│       ├── fs
│       ├── function
│       ├── graph
│       ├── hdfs
│       ├── http
│       ├── interface
│       ├── meta
│       ├── network
│       ├── plugin
│       ├── process
│       ├── session
│       ├── stats
│       ├── test
│       ├── thread
│       ├── thrift
│       ├── time
│       ├── version
│       └── webservice
└── third-party
```  
Nebula Common 仓库代码是 Nebula 内核代码的工具包，提供一些常用工具的高效实现。一些常用工具包相信各位工程师一定也是了然于心。这里只对其中和图数据库密切相关的目录进行说明。  

- src/common/clients/：meta，storage 客户端的 CPP 实现
- src/common/datatypes/：NebulaGraph 中数据类型及计算的定义，比如 string，int，bool，float，Vertex，Edge 等。
- rc/common/expression/：nGQL 中表达式的定义
- src/common/function/：nGQL 中的函数的定义
- src/common/interface/：graph、meta、storage 服务的接口定义

以上为本篇文章的介绍内容。

交流图数据库技术？加入 Nebula 交流群请先[填写下你的 Nebula 名片](https://wj.qq.com/s2/8321168/8e2f/)，Nebula 小助手会拉你进群~~


在 NebulaGraph Overview 篇中我们将带你了解下 NebulaGraph 的架构以及代码仓分布、代码结构和模块规划。

NebulaGraph 源码解读系列 ｜ Vol.01 NebulaGraph Overview


![NebulaGraph 源码解读系列 ｜ Vol.00 序言](https://www-cdn.nebula-graph.com.cn/nebula-blog/nebula-graph-source-code-reading-00.png)

NebulaGraph 是由杭州悦数科技有限公司（官网：[https://www.vesoft.com/cn/](https://www.vesoft.com/cn/)）开源的一款分布式图数据库，它主要用来解决伴随着海量数据产生，在关联数据分析、挖掘方面面临的新挑战。自 2019 年 5 月开源以来，NebulaGraph 受到了广泛的关注，许多企业、技术团队、开发者将 NebulaGraph 应用到业务上构建知识图谱、风控、数据治理、反欺诈、实时推荐等场景。在 Nebula 社区中，越来越多用户从案例分享中掌握 NebulaGraph 的使用方法，与此同时，出现了一种声音，部分用户希望能了解 NebulaGraph 背后的实现思路和原理。

在这样的背景下，我们希望通过 NebulaGraph 源码解读系列，分析设计思路和实现原理，帮助大家深入了解 NebulaGraph，更好地使用 NebulaGraph，同样的，这也将有利于你和 NebulaGraph 社区一块共建更好的 NebulaGraph。

# 内容概述

源码解读系列主要从 **NebulaGraph 系统架构**和**核心模块**展开，此外针对社区用户关心的架构限制带来的性能问题将在最后一个章节分析此类问题。

源码解读系列虽名为源码解读，但并非只是对代码实现的分析和函数讲解，更侧重于从设计角度带你掌握 NebulaGraph 实现原理，透过实现的代码来了解背后的设计思路。（由于 NebulaGraph 目前仍处于快速迭代阶段，持续有新功能进入主分支，故部分不断迭代的核心模块的讲解不会过多地深入代码细节）。

目前源码解读系列章节规划如下：

- NebulaGraph Overview：带你了解 NebulaGraph 架构和代码仓分布、代码结构和模块规划；
- 核心模块讲解：讲述语义分析、优化、调度等系统模块，Java、Python 等各类客户端的运行原理；
- 组件通信：讲述 NebulaGraph 中通信机制；
- 2.0 新特性讲解：从 Variable Length Pattern Match 和索引选择两个点切入讲述 Match 实现的原理；
- 架构限制和解决方案：针对社区的慢查询中止、超大点处理的问题讲述对应的解决方案；

希望大家读完本系列内容之后，对 NebulaGraph 有一定的了解，明白 NebulaGraph 新功能的实现原理，遇到问题时能从实现角度更快定位问题解决问题。以及，在 NebulaGraph 仓库贡献代码时能更好地写出符合 NebulaGraph 设计思路的代码。

# NebulaGraph 完全指南

源码解析系列旨在带你了解 NebulaGraph 代码背后的实现原理，而 NebulaGraph 完全指南（链接：[https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf](https://docs.nebula-graph.com.cn/site/pdf/NebulaGraph-book.pdf)）则是使用大全，从图数据库是什么开始讲起，介绍 nGQL 具体的使用以及对应的参数含义，全方位解决 NebulaGraph 使用问题。

# 背景知识

本系列内容围绕 NebulaGraph 这一分布式图数据库展开，但是仍然需要读者具备下列技术背景：
- 数据库基本原理
- NebulaGraph 架构
- C++ 语言：NebulaGraph 由 C++ 编程语言实现，读者需要有一定的 C++ 编程基础，这将有利于快速了解代码。

# 写在文末

这是第一次我们写系列讲解内容，当中若存在不妥和改进点，希望读者能在论坛、微信群等渠道反馈给我们，好让我们更好地产出内容。

交流图数据库技术？加入 Nebula 交流群请先[填写下你的 Nebula 名片](https://wj.qq.com/s2/8321168/8e2f/)，Nebula 小助手会拉你进群~~


这次，我们深入了解下 NebulaGraph 背后的代码实现逻辑和原理。这只是一个开始…

标签