
![Overall Architecture - NebulaGraph](https://user-images.githubusercontent.com/57335825/87115631-a8852580-c228-11ea-877a-b48ba92fbb8b.png)

 NebulaGraph is an [open-source](https://en.wikipedia.org/wiki/Open-source_software) distributed graph database solution. As an open-source project, we would like those who are interested in graph databases to know as much as possible about it. This includes how NebulaGraph is designed and why it is a highly performant database.

As a result, we have developed a series of graph database architecture articles. In them, we are going to cover three areas. First, we will provide a high-level overview of the architecture of NebulaGraph – the focus of this article. Next, we will cover how NebulaGraph’s storage engine works. And finally, we will illustrate how NebulaGraph’s query engine works, including how we have designed nGQL, our graph query language.

# NebulaGraph Database Architecture Diagram

![NebulaGraph Architecture Diagram](https://user-images.githubusercontent.com/38887077/76277232-1ba27d80-62c3-11ea-9389-cc1aca329c7b.png)

In the diagram above, there is a fully deployed NebulaGraph cluster that contains three services. They include a Meta Service, Query Service, and Storage Service. Each service has its own executable binary. These [binaries](https://github.com/vesoft-inc/nebula/releases) can be deployed either on the same set of hosts or on different hosts.

## Meta Service

In the below diagram, on the right side of the architecture diagram, it shows a cluster of Meta Service. The Meta Service follows a leader/follower architecture. The leader is elected by all Meta Service hosts in the cluster and serves all traffic. Followers stand by and take the updates replicated from the leader. Once the leader dies, one of the followers will be re-elected as the new leader.

![Meta Service](https://user-images.githubusercontent.com/38887077/76277238-1e9d6e00-62c3-11ea-9d10-8391c26b0b20.png)

The Meta Service does not only store and provide meta information about the graph, such as schema and graph partition information. It also takes the role of the coordinator to orchestrate the data movement and to force the leader shift.

## Separation of Query and Storage Layers

![Separation of Query and Storage Layers](https://user-images.githubusercontent.com/38887077/76277242-2230f500-62c3-11ea-8566-4b6be458adee.png)

In the above diagram, to the left of the Meta Service, is NebulaGraph’s main services. NebulaGraph divides the query execution and the graph storage into two separate services. Above the dotted line is the Query Service. Below the dotted line is the Storage Service.
This separation has several benefits. The most straightforward benefit is that the query layer and storage layer can be expanded or shrunk independently. The separation also makes the linear horizontal expansion possible. Last but not the least, the separation provides an opportunity to make it possible for the Storage Service to serve multiple computation layers.
The current Query Service can be considered a high-priority computation layer. Meanwhile an iterative computation framework could be another one.

## Stateless Query Layer

![Stateless Query Layer](https://user-images.githubusercontent.com/38887077/76277245-25c47c00-62c3-11ea-912b-5de3ce4e18fe.png)

As to the Query Service, each individual Query Service host runs a stateless query engine. The Query Service hosts will never communicate with each other. They only read the meta information from the Meta Service and interact with the Storage Service.

This design makes sure the Query Service cluster can be easily managed by Kubernetes (aka. k8s) or be deployed on the cloud. A cloud deployment is already available for NebulaGraph.
See [how to deploy NebulaGraph on Kubernetes](https://nebula-graph.io/posts/how-to-deploy-nebula-graph-in-kubernetes/).

There are two ways to balance the Query Service load. The most common way is to place a [load balancer](https://nebula-graph.io/posts/nebula-graph-storage-banlancing-data-migration/) in front of the Query Service. The second way is to configure the client library with all Query Service hosts’ IP addresses. The client will randomly pick one to connect to.

Each query engine takes the request from the client, parses the statement, and generates an Abstract Syntax Tree (AST). Then the AST will be handed over to the execution planner and the optimizer. The final AST will be passed to the executors.

Below is a flow chart of how NebulaGraph’s query engine works. For a detailed explanation, please refer to [an introduction to the query engine](https://nebula-graph.io/posts/nebula-graph-query-engine-overview/).

![image](https://user-images.githubusercontent.com/38887077/78201029-28318480-74c3-11ea-90ed-03e9c2ca25d4.png)

## Shared-nothing Distributed Storage Layer

![Shared-nothing Distributed Storage Layer](https://user-images.githubusercontent.com/38887077/78201059-3a132780-74c3-11ea-95a0-61249c1566a4.png)

The Storage Service is designed to be a [shared-nothing distributed architecture](https://en.wikipedia.org/wiki/Shared-nothing_architecture). Each storage host has multiple local key/value stores as the physical data storage.

A quorum consensus protocol (we chose [RAFT](https://raft.github.io/) over paxos due to its simplicity) is built on top of the key/value stores. Together, they provide a distributed key/value store. On top of this distributed key/value store, a graph semantic layer is provided. It translates the graph operations into key/value operations. The graph data (vertices and edges) are hashed into multiple partitions by the rule (vertex_id % the_number_of_partitions).
A partition is a virtual set of data in NebulaGraph. These partitions are allocated to all storage hosts. The allocation information is stored in the Meta Service and can be accessed by storage hosts and query hosts.

# Conclusion

This bird’s eye view into NebulaGraph covers the architecture to provide a basic understanding of how the distributed graph database works. For a more detailed introduction to the major components of the system, please refer to:

- [An introduction to NebulaGraph's Storage Engine](https://nebula-graph.io/posts/nebula-graph-storage-engine-overview/)
- [An Introduction to NebulaGraph's Query Engine](https://nebula-graph.io/posts/nebula-graph-query-engine-overview/)

If you have anything to share about distributed systems or graph databases, please feel free to leave us a comment below. We look forward to hear from you.

# You might also like

1. [The Review on Graph Database](https://nebula-graph.io/posts/review-on-graph-databases/)
1. [Graph Query Language Comparison Series - Gremlin vs Cypher vs nGQL](https://nebula-graph.io/posts/graph-query-language-comparison-cypher-gremlin-ngql/)


NebulaGraph is an open-source, distributed, highly performant graph database. This article explains its technical architecture in detail.

NebulaGraph Database Architecture — A Bird’s Eye View


![Understanding Subgraph in NebulaGraph 2.0](https://user-images.githubusercontent.com/57335825/124857891-aabac500-dfdf-11eb-8461-f77ffe6f6136.png)

# Introduction

In An Introduction to [NebulaGraph 2.0 Query Engine](https://nebula-graph.io/posts/nebula-query-engine-introduction/), I introduced how the query engine differs between V2.0 and V1.0 of NebulaGraph.

![NebulaGraph Architecture Change](https://user-images.githubusercontent.com/57335825/124858026-e3f33500-dfdf-11eb-8f0f-0821af5d84b7.png)

As shown in the preceding figure, you can see that when a query statement is sent from the client, how the query engine parses the statement, generates an AST, validates the AST, and then generates an execution plan. In this article, I will introduce more about the query engine through the new subgraph feature in NebulaGraph 2.0 and focus on the execution plan generation process, to help you understand the source code better.

# Definition of Subgraph

A subgraph of a graph is composed of a subset of the graph's vertices and a subset of its edges. When a starting vertex is given, a traversal starts from it and expands following the specified edge type until all the vertices at the specified number of hops are reached. All the reachable vertices and edges form a subgraph. 

# Syntax of Subgraph

```shell
GET SUBGRAPH [<step_count> STEPS] FROM {<vid>, <vid>...} [IN <edge_type>, <edge_type>...]
[OUT <edge_type>, <edge_type>...] [BOTH <edge_type>, <edge_type>...]
```

- step_count: Specifies the number of hops from the starting vertex to obtain a subgraph consisting of vertices and edges that are reached at 0 to `step_count` hops. It must be a non-negative integer. The default value is 1.
- vid: Specifies the ID of the starting vertex.
- edge_type: Specifies the edge type. `IN`, `OUT`, and `BOTH` can be added to specify the direction of the edges from the starting vertex. By default, `BOTH` is used.

# Implementation of Subgraph

When the query engine receives a `GET SUBGRAPH` statement, the Parser, composed of Flex and Bison, will extract the necessary content from the query statement according to the specified rule, that is, the `get_subgraph_sentence` rule defined in [parser.yy](https://github.com/vesoft-inc/nebula-graph/blob/master/src/parser/parser.yy), and then generate an AST, as shown in the following figure.

![NebulaGraph Subgraph Implementation](https://user-images.githubusercontent.com/57335825/124858306-5e23b980-dfe0-11eb-8f66-bd15790a0eef.png)

And then in the Validate stage, the generated AST is validated, aiming to make sure that all the inputs from users conform to the syntax rule. For more information, see [An Introduction to NebulaGraph 2.0 Query Engine](https://nebula-graph.io/posts/nebula-query-engine-introduction/). When the validation is successful, the content of the AST will be extracted and an execution plan will be generated.

How is the execution plan generated? Database solutions may differ in the form of the execution plan, but their implementations are the same, that is, depending on which operators are used and how the query layer interacts with the storage layer, because data needs to be taken from the storage layer for each query statement. In NebulaGraph, the query engine interacts with the storage layer through RPC (fbthrift), of which the interfaces are defined in the `interface` directory of the `common` repository. `getNeighbors` and `getProps` are the key interfaces.

In fbthrift, `getNeighbors` is defined as follows:

```shell
struct GetNeighborsRequest {    
    1: common.GraphSpaceID                      space_id,
    2: list<binary>                             column_names,
    3: map<common.PartitionID, list<common.Row>>
        (cpp.template = "std::unordered_map")   parts,
    4: TraverseSpec                             traverse_spec
}
```

To understand the variables in the preceding structure, see the comments in the source code: [https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift](https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift).

It mainly aims to enable this process:

1. The query engine passes the starting vertex and the specified edge types according to the defined structure.
2. In the storage layer, the starting vertex is located.
3. The properties of the vertex and its outgoing edges are assembled into a table and the table is returned to the query engine. The table format is defined in `GetNeighborsResponse`. For more information, see [https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift](https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift).
4. The query engine extracts the necessary content from the table.

For example, for the NBA dataset, when `Tim Duncan` and `Manu Ginobili` are given as the starting vertices and `like` of both directions are given as the edge type, we can get the information of `$^.player.name`, `like._dst`, `$$.player.name`, and `like.likeness` as follows:

![NebulaGraph Subgraph](https://user-images.githubusercontent.com/57335825/124858735-32ed9a00-dfe1-11eb-8bb1-057bed6ac316.png)

Because the traversal is done in both directions, `+ like` in the fourth column of the preceding table represents outgoing edges, and `- like` in the fifth column represents incoming edges.

In NebulaGraph, edges are stored with their source vertices in the storage layer, so all the properties of a vertex and its outgoing edges can be obtained through the `GetNeighbor` interface. However, if you want to obtain the property information of the destination vertices of these edges, the `getProps` interface is needed. A `FETCH` statement also needs this interface to obtain the property information of a vertex or an edge. For more information about `GetPropRequest`, see [https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift](https://github.com/vesoft-inc/nebula-common/blob/master/src/common/interface/storage.thrift).

## Execution Plan

With the preceding interfaces, an execution plan can be generated. The operators, `start`, `getNeighbor`, `subgraph`, `loop`, and `dataCollect`, are necessary.

- start: A leaf node of an execution plan. It does nothing but informs the scheduler that no more operator can be relied upon. In other words, this is the termination condition in the recursive algorithm.
- loop: Equivalent to the `WHILE` loop in C language. It is composed of three parts: `depend`, `condition`, and `loopBody`. `depend` is used in multi-statements and the PIPE and no more details about it is involved in this article. `condition` is equivalent to the termination condition. `loopBody` is equivalent to the loop body in a `WHILE` loop.
- subgraph: Responsible for extracting the properties of `_dst`, the destination vertex, from the `getNeighbor` result, filtering out the destination vertices that have been traversed to avoid repeatedly fetching data from the storage layer, and then using them as the input for the next hop executed by `getNeighbor`.
- datacollect: Responsible for assembling the properties of the reachable vertices and their edges into the vertices and edge types.

For more information about these operators, see the source code: [https://github.com/vesoft-inc/nebula-graph/tree/master/src/executor](https://github.com/vesoft-inc/nebula-graph/tree/master/src/executor).

Let's take the graph in the following figure as an example to introduce how a subgraph is generated.

![Build a Subgraph in NebulaGraph](https://user-images.githubusercontent.com/57335825/124859204-0a19d480-dfe2-11eb-8a1d-c83af10038fe.png)

**Single-hop traversal**

It is easy to start from Vertex A and traverse along the `like` edges at one hop for the vertices and edges. In this scenario, only getNeighbor and dataCollect are needed. The execution plan is as follows.

![Single-hop traversal](https://user-images.githubusercontent.com/57335825/124859285-37668280-dfe2-11eb-9021-ab538e2ac90b.png)

**Multi-hop traversal**

The single-hop traversal is a special case of the multi-hop traversal, so it can be treated as part of a multi-hop traversal. In a three-hop traversal starting from Vertex A along the `like` edges at three hops, the destination vertices can be extracted after `getNeighbor` is executed and these vertices can be the starting vertices to call the `getNeighbor` interface. Two loops are sufficient for such a scenario. The number of hops can be set as the termination condition in `loop`. Therefore, the execution plan is as follows.

![Multi-hop traversal](https://user-images.githubusercontent.com/57335825/124859410-77c60080-dfe2-11eb-836e-3450dfa005bd.png)

**Input and output**

Generally, the input of each operator is the output of the one that it depends on, so according to the dependencies represented in the execution plan, you can see the input and output of each operator. However, in some cases, for example, in a multi-hop traversal, the input of each `getNeighbor` is the destination vertex that is reached following an edge, that is, the output of the `subgraph` operator, so the output of `subgraph` is the input of `getNeighbor`. Such a dependency is different from that shown in the preceding figure. Therefore, the input and output of each operator must be set separately. From [An Introduction to NebulaGraph 2.0 Query Engine](https://nebula-graph.io/posts/nebula-query-engine-introduction/), you know that the input and output of each operator are stored in a hash table, of which values are of the `vector` type. The following ResultMap table is an example.

![ResultMap](https://user-images.githubusercontent.com/57335825/124859705-f4f17580-dfe2-11eb-8e9e-5b7bbadc5686.png)

- The starting vertex is stored in ResultMap["StartVid"].
- The input of getNeighbor is ResultMap["StartVid"] and its output is stored in ResultMap["GN_1"].
- The input of getNeighbor is ResultMap["StartVid"] and its output is stored in ResultMap["GN_1"].
- No data is generated by loop. It is used as a logic loop, so neither input nor output is necessary.
- The input of dataCollect is ResultMap["GN_1"] and its output is stored in ResultMap["DATACOLLECT_2"].

At this time, the getNeighbor operator puts the result of each loop at the end of the vector in ResultMap["GN_1"], and the subgraph operator takes the value from the end of the vector in ResultMap["GN_1"] and then stores the starting vertex for the next hop in ResultMap["StartVid"].

After the first hop, the result of ResultMap is as follows.
​
![ResultMap](https://user-images.githubusercontent.com/57335825/124859885-3eda5b80-dfe3-11eb-8eff-e0705c0404e9.png)

To make the result clearer, only the properties of `_dst` are shown as the result of `getNeighbor`. In the actual result, the properties of all the edges and the starting vertex are listed, as show in Table 1.

`subgraph` receives "GN_1" as its input, extracts the properties of `_dst`, and then stores the result in "StartVid".

After the second hop traversal, the ResultMap is as follows.

![ResultMap](https://user-images.githubusercontent.com/57335825/124860033-882aab00-dfe3-11eb-8237-0987c712006a.png)

After the third hop traversal, the ResultMap is as follows.

![ResultMap](https://user-images.githubusercontent.com/57335825/124860085-a2fd1f80-dfe3-11eb-8f14-54aaaa0950ca.png)

Finally, `dataCollect` takes all the vertex sets and edge sets traversed in the process, assembles them into the final result, and returns it for users.

# Example

In this section, I will use an example to show the execution plan structure of a subgraph in NebulaGraph 2.0:

1. Open Nebula Console.
2. Switch to the `nba` graph space. 
3. Run `EXPLAIN format="dot" GET SUBGRAPH 2 STEPS FROM 'Tim Duncan' IN like, serve`. Data of the DOT format is generated on Nebula Console.
4. Open the [Graphviz Online](https://dreampuf.github.io/GraphvizOnline/) website and copy and paste the DOT data on it.

The execution plan is shown as follows.

![Result in DOT format](https://user-images.githubusercontent.com/57335825/124860296-fc654e80-dfe3-11eb-86fc-589ed9a3babe.png)

In this plan, `depend` of `loop` depends on `Start_0`. However, because there is no multi-statement or PIPE, it is not necessary.
​
Welcome to join the NebulaGraph [Slack](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) channel to learn more about NebulaGraph.


This article introduces the query engine, the execution plan generation process in particular, through the new subgraph feature in NebulaGraph 2.0 to help you understand the source code better.

Understanding Subgraph in NebulaGraph 2.0


![An Introduction to NebulaGraph 2.0 Query Engine](https://user-images.githubusercontent.com/57335825/111279545-c20f2480-8675-11eb-8699-a8429c237dec.png)

# Introduction

Compared with V1.0, NebulaGraph, a distributed graph database, has been significantly changed in V2.0. One of the most obvious changes is that in NebulaGraph 1.0, the code of the Query, Storage, and Meta modules are placed in one repository, while from NebulaGraph 2.0 onwards, these modules are placed in three repositories:

* `nebula-graph`: Mainly contains the code of the Query module.
* `nebula-common`: Mainly contains expression definitions, function definitions, and some public interfaces.
* `nebula-storage`: Mainly contains the code of the Storage and Meta modules.

This article introduces the overall structure of the Query layer and uses an nGQL statement to describe how it is processed in the four main modules of the Query layer.

# Architecture

This figure shows the architecture of the Query layer. It has four submodules:

* Parser: Performs lexical analysis and syntax analysis.
* Validator: Validates the statements.
* Planner: Generates and optimizes the execution plans.
* Executor: Executes the operators.

![The architecture of the Query layer](https://user-images.githubusercontent.com/57335825/111279912-2af69c80-8676-11eb-8377-492a0955c3cd.png)

# Source Code Hierarchy

Now, let's look at the source code hierarchy under the`nebula-graph`repository.

```plain
|--src
    |--context     // contexts for validation and execution
    |--daemons 
    |--executor    // execution operators
    |--mock
    |--optimizer   // optimization rules
    |--parser      // lexical analysis and syntax analysis
    |--planner     // structure of the execution plans
    |--scheduler   // scheduler
    |--service
    |--util        // basic components
    |--validator   // validation of the statements
    |--visitor
```

#  How a Query is Executed

Starting from the release ofNebulaGraph 2.0-alpha, nGQL gets to support vertex IDs of the`String`type. It is working to support vertex IDs of the`Int`type in NebulaGraph 2.0 to get compatible with NebulaGraph 1.0.

To introduce how the Query layer works, let's take an nGQL statement like`GO FROM "Tim" OVER like WHERE like.likeness > 8.0 YIELD like._dst`as an example. Its data flow through the Query layer is shown in the following figure.

![How a Query is Executed](https://user-images.githubusercontent.com/57335825/111281267-b886bc00-8677-11eb-91ba-fcbf3371ce06.png)

When the statement is input, it is processed as follows.

## Stage 1: Generating an AST

In the first stage, the statement is parsed by the Parser (composed of Flex and Bison) and its corresponding AST is generated. The structure of the AST is shown in the following figure.

![Generating an AST](https://user-images.githubusercontent.com/57335825/111280086-5ed1c200-8676-11eb-98fa-86ba7029a7a9.png)

In this stage, the Parser intercepts the statements that do not conform to the syntax rules. For example, a statement like`GO "Tim" FROM OVER like YIELD like._dst`will be directly intercepted in this stage because of its invalid syntax.

## Stage 2: Validation

In the second stage, the Validator performs a series of validations on the AST. It mainly works on these tasks:

* Validating metadata

When parsing the `OVER` , `WHERE` , and `YIELD` clauses, the Validator looks up the Schema and verifies whether the edge and tag data exist or not. For an `INSERT` statement, the Validator verifies whether the types of the inserted data are the same as the ones defined in the Schema.

* Validating contextual reference

For multiple statements, like `$var = GO FROM "Tim" OVER like YIELD like._dst AS ID; GO FROM $var.ID OVER serve YIELD serve._dst`, the Validator verifies `$var.ID`: first to see if `var` was defined, and then to check if the `ID` property is attached to the `var` variable. If `$var.ID` is replaced with `$var1.ID` or `$var.IID` , the validation fails.

* Validating type inference

The Validator infers what type the result of an expression is and verifies the type against the specified clause. For example, the `WHERE` clause requires the result to be a Boolean value, a NULL value, or empty.

* Replacing `*`

Take a statement like `GO FROM "Tim" OVER * YIELD like._dst, like.likeness, serve._dst` as an example. When verifying the `OVER` clause, the Validator needs to query the Schema and replace `*` with all the edges defined in the Schema. For example, if only `like` and `serve` edges are defined, the statement is as follows.

```plain
GO FROM "Tim" OVER serve, like YIELD like._dst, like.likeness, serve._dst
```

* Validating input and output

For an nGQL statement with the PIPE operator, such as`GO FROM "Tim" OVER like YIELD like._dst AS ID | GO FROM $-.ID OVER serve YIELD serve._dst`, the Validator verifies `$-.ID` . In the example statement, the `ID` property in the second clause was already defined in the previous clause, so the second clause is verified legal. If `$-.ID` is changed to `$-.a` , where `a` was not defined, the second clause is illegal.

## Stage 3: Generating an Execution Plan

When the validation succeeds, an execution plan is generated in Stage 3. Its data structure is stored in the `src/planner` directory, with the following logical structure.

![Generating an Execution Plan](https://user-images.githubusercontent.com/57335825/111280198-86c12580-8676-11eb-8226-d4e7f9a8c474.png)

### Query Execution Flow

The generated execution plan is a directed acyclic graph where the dependencies between nodes are determined in the [toPlan()](https://github.com/vesoft-inc/nebula-graph/blob/master/src/validator/PipeValidator.cpp) function of each module in the Validator. As shown in the preceding figure, the Project node depends on the Filter node, the Filter node depends on the GetNeighbor node, and so on, up to the Start node.

During the execution stage, the Executor generates an operator for each node and starts scheduling from the root node (e.g., the Project node in this example). If the root node is found to be dependent on other nodes, the Executor **recursively calls** the nodes that the root node depends on until it finds a node that is not dependent on any node (e.g., the Start node in this example), and then starts execution. After the execution, the Executor continues executing the node that depends on the executed node (e.g., the GetNeighbor node in this example), and so on, until the root node is reached.

### Query Data Flow

The input and output of each node are also determined in [toPlan()](https://github.com/vesoft-inc/nebula-graph/blob/master/src/validator/PipeValidator.cpp). Although the execution is performed in the order defined in the execution plan, the input of each node can be customized and does not completely depend on the previous node, because the input and output of all nodes are actually stored in a hash table, where the keys are the names customized during the creation of each node. For example, if the hash table is named as `ResultMap`, when creating the Filter node, you can determine that the node takes data from `ResultMap["GN1"]`, then puts the result into `ResultMap["Filter2"]`, and so on. All these work as the input and output of each node. The hash table is defined in `src/context/ExecutionContext.cpp` under the `nebula-graph` repository. The execution plan is not really executed, so the value of each key in the associated hash table is empty (except for the starting node, where the input variables hold the starting data), and they will be computed and filled in the Executor stage.

This is a simple example. A more complex one will be shown at the end for you to better understand how an execution plan works.

## Stage 4: Optimizing the Execution Plan

An execution plan generated in the preceding stage can be optimized optionally. In the `etc/nebula-graphd.conf` file, when `enable_optimizer` is set to `true`, the execution plans will be optimized. In this example, when optimization is enabled, the data flow is as follows.

![Optimizing the Execution Plan](https://user-images.githubusercontent.com/57335825/111280908-53cb6180-8677-11eb-83c5-01cb632309dd.png)

As shown in the preceding figure, the Filter node is integrated into the GetNeighbor node. Therefore, when the GetNeighbor operator calls interfaces of the Storage layer to get the neighboring edges of a vertex during the execution stage, the Storage layer will directly filter out the unqualified edges internally. Such optimization greatly reduces the amount of data transfer,which is commonly known as filter push down.

In the execution plan, each node directly depends on other nodes. To explore equivalent transformations and to reuse the same parts of an execution plan, such a direct dependency between nodes is converted into an OptGroupNode to OptGroup dependency. Each OptGroup contains an equivalent set of OptGroupNodes, and each OptGroupNode contains one node in the execution plan. An OptGroupNode depends on an OptGroup but not on other OptGroupNodes. Therefore, from one OptGroupNode, many equivalent execution plans can be explored because the OptGroup it depends on contains different OptGroupNodes, which saves storage space.

All the optimization rules we have implemented so far are considered as RBO (Rule-Based Optimization), which means the plans optimized based on the rules must be better than their original versions. The CBO (Cost-Based Optimization) feature is under development. The optimization process is a "bottom-up" exploration process: For each rule, the root node of the execution plan (in this case, the Project node) is the entry point, along the node dependencies, the bottom node is found, and then from the bottom node, the OptGroupNode in each OptGroup is explored to see if it matches the rule, until the entire execution plan can no longer apply the rule. Then, the next rule is explored.

This figure shows the optimization process of this example.

![Execution plan optimization process](https://user-images.githubusercontent.com/57335825/111280406-bf60ff00-8676-11eb-9f87-71390f706bac.png)

As shown in the preceding figure, when the Filter node is explored, it is found that its children node is GetNeighbors, which matches successfully with the pre-defined pattern in the rule, so a transformation is initiated to integrate the Filter node into the GetNeighbors node, the Filter node is removed, and then the process continues to the next rule.

The optimized code is in the `src/optimizer/` directory under the `nebula-graph` repository.

## Stage 5: Execution

In the fifth stage, the Scheduler generates the corresponding execution operators against the execution plan, starting from the leaf nodes and ending at the root node. The structure is as follows.

![Execution plan execution](https://user-images.githubusercontent.com/57335825/111280526-ea4b5300-8676-11eb-84fa-25353a0b86ed.png)

Each node of the execution plan has one execution operator node, whose input and output have been determined in the execution plan. Each operator only needs to get the values for the input variables, compute them, and finally put the results into the corresponding output variables. Therefore, it is only necessary to execute step by step from the starting node, and the result of the last operator is returned to the user as the final result.

# An Example Query

Now we will get acquainted with the structure of an execution plan by executing a `FIND SHORTEST PATH` statement.

Open nebula-console and enter the following statement.

```plain
FIND SHORTEST PATH FROM "YAO MING" TO "Tim Duncan" OVER like, serve UPTO 5 STEPS
```

To get the details of the execution plan of this statement, precede it with the`EXPLAIN`.

![Detailed execution plan](https://user-images.githubusercontent.com/57335825/111280632-07802180-8677-11eb-8c9d-e016bfee3130.png)

The preceding figure shows, from left to right, the unique ID of each node in the execution plan, their names, the IDs of their dependent nodes, the profiling data (information about the execution of the `PROFILE` command), and the details of each node (including the names of the input and output variables, the column names of the output results, and the parameters of the node).

To visualize the information, you can follow these steps:

1. Use `EXPLAIN format="dot"` instead of `EXPLAIN`. nebula-console can generate the data in the DOT format.

2. Open the [Graphviz Online](https://dreampuf.github.io/GraphvizOnline/) website and paste the generated DOT data. You can see the structure as shown in the following figure. This structure corresponds to the execution flow of the operators in the execution stage.

![Visualized execution plan](https://user-images.githubusercontent.com/57335825/111280728-1ff03c00-8677-11eb-84c8-5d67b6aa97ce.png)

The shortest path algorithm uses the two-way breadth-first search algorithm, expanding from both "YAO MING" and "Tim Duncan", so GetNeighbors, BFSShortest, Project, and Dedup have two operators, respectively, the input is connected by the PassThrough operator, and the path is stitched by the ConjunctPath operator. The LOOP operator then controls the number of steps to be extended outward, and it can be seen that the input of the DataCollect operator is actually taken from the output variable of the ConjunctPath operator.

The information of each operator is in the `src/executor` directory under the `nebula-graph` directory. Feel free to share your ideas with us on GitHub: [https://github.com/vesoft-inc/nebula](https://github.com/vesoft-inc/nebula) or the : [https://github.com/vesoft-inc/nebula/discussions](https://github.com/vesoft-inc/nebula/discussions)

>Hi, I’m Ming Quan, engineer at NebulaGraph. I’m interested in database storage and would like to share my experiences in this regard. Hope my post is of help to you. Please let me know if you have any ideas about this. Thanks.

This article introduces the overall structure of the Query layer and uses an nGQL statement to describe how it is processed in the four main modules of the Query layer. 

An Introduction to NebulaGraph 2.0 Query Engine


![NebulaGraph Query Engine Architecture](https://user-images.githubusercontent.com/57335825/83608282-84e80480-a531-11ea-97d1-7c6fbb66c29f.png)

The query engine is used to process the NebulaGraph Query Language (nGQL) statements. So, an architectural overview of the NebulaGraph query engine is important.

In the below illustration is an overview chart of the query engine. If you are familiar with the SQL execution engine, this should be familiar to you. The NebulaGraph query engine is very similar to the modern SQL execution engine except for the query language parser and the real actions in the execution plan.

![overview chart of the query engine](https://user-images.githubusercontent.com/56643819/72578735-e6f0f600-3911-11ea-8726-0357034b4fad.png)

# Session Manager

NebulaGraph employs the Role Based Access Control. So, when the client first connects to the Query Engine, it needs to authenticate itself. When it succeeds, the query engine creates a new session and returns the session ID to the client. All sessions are managed by the Session Manager. The session will remember the current graph space and access rights to the space. The session will also keep some session-wide configurations and be used as a temporary storage to store information across multiple requests in the same session.

The session will be dropped when the client connection is closed or is idle for a period of time. The length of the idle time is configurable. When the client sends a request to the query engine, it needs to attach the current session ID. Otherwise, the query engine will reject the request. When the query engine accesses the storage engines, it will attach the session object to every request. This is so the storage engine does not have to manage sessions.

# Parser

The first thing that the query engine will do when receiving a request is to parse the statements in the request. This is done by the parser. Most of the parser code is generated by the famous flex/bison tool set. The lexicon and syntax files for the nGQL can be found in the `src/parser` folder in the source code tree. The nGQL is designed in a way that is close enough to SQL. The idea is to lessen the learning curve as much as possible.

Graph databases currently do not have a unified international query language standard. As soon as the ISO’s GQL committee releases their first draft, NebulaGraph plans to make nGQL compatible with the proposed GQL.

The output of the parser is an Abstract Syntax Tree (AST), which will be passed on to the next module, the Execution Planner.

# Execution Planner

The Execution Planner will convert the AST from the parser into a list of actions (execution plan). An action is the smallest unit that can be executed. A typical action could be fetching all neighbors for a given vertex, getting properties for an edge, or filtering vertices or edges based on a given condition.

When converting AST to the execution plan, all IDs will be extracted. This is so the execution plan can be reused. The extracted IDs will be placed in the context for the current request. The context will also be used to store the variables and intermediate results.

# Optimization

The newly generated execution plan will then be passed to the Optimization Framework. There are multiple optimizers registered in the framework. The execution plan will be passed through all optimizers sequentially. Each optimizer can modify (optimize) the plan.

In the end, the final plan can look dramatically different from the original plan. However, the execution result should be exactly the same as the original plan.

# Execution

The last step in the query engine is to execute the optimized execution plan. It is done by the Execution Framework. Each executor will process one execution plan at a time. Actions in the plan will be executed one by one.

The executor will also do limited local optimization, such as deciding whether to run in parallel. Depending on the action, the executor will communicate with the Meta Service or the Storage Engine via their clients.

# You might also like

1. [NebulaGraph Architecture — A Bird’s View](https://nebula-graph.io/posts/nebula-graph-architecture-overview/)
1. [An Introduction to NebulaGraph's Storage Engine](https://nebula-graph.io/posts/nebula-graph-storage-engine-overview/)
1. [The Review on Graph Database](https://nebula-graph.io/posts/review-on-graph-databases/)


The query engine is used to process the NebulaGraph Query Language (nGQL) statements.

An Introduction to NebulaGraph’s Query Engine


![An Introduction to NebulaGraph's Storage Engine](https://user-images.githubusercontent.com/57335825/87172263-cf734400-c288-11ea-96d0-1bb800700d16.png)

The Storage Service of NebulaGraph is composed of two parts. One is the `Meta Service` that stores the meta data. The other is `Storage Service` that stores the data. The two services are in two independent processes. The data and deployment are separated but their architectures are almost the same.

# Architecture

---
![The Architecture of Storage Service](https://user-images.githubusercontent.com/42762957/71571213-2cd94c00-2b14-11ea-9b7f-a65067c61a63.png)

Fig.1 The Architecture of Storage Service

---

As shown in figure 1, there are three layers in the Storage Service. The bottom layer is the local storage engine, providing `get`, `put`, `scan` and `delete` operations on local data. The related interfaces are in `KVStore/KVEngine.h` and users can develop their own local store plugins based on their needs. Currently, NebulaGraph provides a store engine based on RocksDB.

Above the local storage engine is the consensus layer that implements multi group raft. Each partition corresponding to a raft group is for the [data sharding](https://www.nebula-graph.io/posts/what-is-database-sharding). Currently, NebulaGraph uses `hash` to shard data. When creating a space, users need to specify the partition number. Once set, a partition number cannot be changed. Generally, the partition number must meet the need to scale-out in the future.

Above the consensus layer is the storage interface that defines a set of APIs that are related to the graph. These API requirements are translated into a set of kv operations targeting the corresponding partition. It is this layer that makes the storage service a real graph storage. Otherwise it is just a kv storage. NebulaGraph does not separate kv as an independent service. This is because a graph query involves a lot of calculation that uses schema, which does not exist in a kv layer. Such architecture makes pushing down easier.

# Schema & Partition

NebulaGraph stores the vertices, edges, and properties. Efficient property filtering is critical for a Graph Database. NebulaGraph uses tags to indicate vertices types. One vertexID can relate with multiple tags, and each tag has its own properties. In kv storage, we use `vertexID + TagID` to represent a key and the related properties are placed behind a value after encoding. The format is shown in figure 2:

---
![Vertex Key Format](https://user-images.githubusercontent.com/42762957/71571002-1252a300-2b13-11ea-915f-b3fd16920b95.png)

Fig.2 Vertex Key Format

---

- `Type`: one byte, used to indicate key type, current types are data, index, system, etc.
- `Part ID`: three bytes, used to indicate the sharding partition. This field can be used to scan the partition data based on the prefix when re-balancing the partition
- `Vertex ID`: eight bytes, used to indicate vertex ID
- `Tag ID`: four bytes, used to relate tag
- `Timestamp`: eight bytes, not available to users, used in MVCC in the future

Each edge in NebulaGraph is modeled as two independent key-values. One is stored in the same partition as the source vertex, namely out-edge. The other one is stored in the same partition as the destination vertex, namely in-edge. Generally, out-key and in-key are stored in different partitions.

There may be multiple edge types or multiple edges with the same edge type between two vertices. For example, if we define an edge type ‘transfer’, user A may transfer money to user B multiple times. Thus a field `rank` is added to distinguish the transfer records between the two. The edge key format is shown in figure three:

---
![Edge Key Format](https://user-images.githubusercontent.com/38887077/75969929-9c9bf680-5f0a-11ea-83bc-ff26e472095a.png)

Fig.3 Edge Key Format

---

- `Type`: one byte, used to indicate key type, current types are data, index, system, etc.
- `Part ID`: three bytes, used to indicate the sharding partition. This field can be used to **scan the partition data based on the prefix when re-balancing the partition**
- `Vertex ID`: eight bytes, used to indicate source vertex ID in out-edge, dest vertex ID in in-edge
- `Edge Type`: four bytes, used to indicate edge type, greater than zero means out-edge, less than zero means in-edge
- `Rank`: eight bytes, used to indicate multiple edges in one edge type. Users can set the field based on needs and store weight such as _transaction time_, _transaction number_ and _weight of a certain order_
- `Vertex ID` : eight bytes, used to indicate dest vertex ID in out-edge, source vertex ID in in-edge
- `Timestamp`: eight bytes, not available to users, used in MVCC in the future

If the value of an edge type is greater than zero, the corresponding edge key format is shown in figure four. Otherwise if the value is less than zero, the corresponding edge key format is shown in figure five.

---
![Out-key format](https://user-images.githubusercontent.com/38887077/75969938-9f96e700-5f0a-11ea-8469-027fe7ff08c3.png)

Fig.4 Out-key format

---

![In-key format](https://user-images.githubusercontent.com/38887077/75969944-a160aa80-5f0a-11ea-99ed-f4a8807b1828.png)

Fig.5 In-key format

---

The properties of vertices and edges are stored in the values of the corresponding kv pairs after being encoded. As a strong typed database, NebulaGraph needs to get the schema information in the Meta Service before decoding. In addition, when properties are encoded, the schema version is added to support online schema change.

NebulaGraph shards data through a `modulo operation` on vertex ID. All the out-keys, in-keys and tag data are placed in the same partition through the `modulo operation` on vertex ID. In this way, query efficiency is increased dramatically.

For online queries, the most common operation is to do a Breadth-First-Search (BFS) expansion from a vertex. Thus it is the basic operation to get a vertex’s in-edge or out-edge and such operation’s performance determines the traverse’s performance.

In BFS, there are cases for filtering some properties. NebulaGraph guarantees the operation efficiency by putting properties and vertices and edges together. At present, most graph databases verify their efficiency with Graph 500 or the Twitter data set, which are of no eloquence because properties are not included. While most cases use property graphs and BFS needs a lot of filtering as well.

# KVStore

NebulaGraph develops its own kv store to meet performance needs:

- **High performance** pure kv
- **Provided as a library**, as a strong typed database, the performance of pushing down is the key of the performance of **NebulaGraph**
- **Strong data consistency**, determined by the distribution system
- **Written in C++**  

For users who are not sensitive to performance or unwilling to migrate data, NebulaGraph provides plugins for the whole kv store and builds the storage service on the third party kv store. Currently, NebulaGraph provides the HBase plugin.
With RocksDB as the local storage engine, NebulaGraph supports self-management of multiple hard disks to make full use of the parallel property. All users need to do is configure multiple data directories. NebulaGraph manages the distributed kv store in a unified scheduling way with meta service. All the partition distribution data and current machine status can be found in the meta service. Users can input commands in the console to add or remove machines to generate and execute a balance plan in meta service.
NebulaGraph provides its own WAL mode so one can customize the WAL. Each partition owns its WAL to catch up data without WAL splitting to guarantee efficiency. In addition, NebulaGraph defines a Command Log class to conduct special operations. These logs have no real data and are only used to inform all replicas to execute certain operations with raft. NebulaGraph also provides a log class to conduct an atomic operation on partitions, such as with the CAS and the read-modify-write which fully uses raft serialization.
One NebulaGraph kv store cluster supports multiple spaces, and each space is set with its own partition number and replica number. Different spaces are isolated physically. Spaces in the same cluster support different storage engines and sharding strategies.

# The Raft Implementation in NebulaGraph

## Multi Raft Group

As the log IDs must be serial, almost all the implementations adapt a Multi Raft Group to solve this problem. Therefore, the partition number almost determines the performance of the Raft Group. However, it does not mean the more partitions, the better the performance.

Each Raft group stores a set of state information and has its own WAL file. Thus, more partitions require more costs. Also, when there are too many partitions, the batch operation is meaningless if the load is not high enough. For example, with a single online system having 10,000 tps, if its partition number is greater than ten thousand, the tps of each partition per second is likely only 1. So, the batch operation loses its meaning and increases CPU cost.

There are two key points to achieve a Multi Raft Group. First is to share the transport layer because each Raft Group needs to send messages to the corresponding peers. If the transport layer cannot be shared, the connection costs are huge. Second is the threading model. With it, a Multi Raft Group must share a set of thread pools. Otherwise it will end with too many threads in the system, resulting in a large amount of context switch cost.

## Batch

For each Partition, it is necessary to do a batch to improve throughput when writing WAL serially. In general, there is nothing special about a batch, but NebulaGraph designs a special type of WAL based on each part serialization. This adds some challenges.

For example, NebulaGraph uses WAL to implement lock-free CAS operations and each CAS operation will execute after all the previous WALs are committed. So, for a batch, if there are several WALs in CAS type, we need to divide this batch into several smaller serial groups.

## Learner

New machines need to “catch up” data for quite a long time, and there may be accidents during the process. If a new machine joins the raft group as a follower, it will reduce the HA capacity of the entire cluster. NebulaGraph uses the command WAL mentioned above to implement its learner. If the leader encounters the `add learner command` when writing WAL, it will add the learner to its peers and mark it as a learner, so the logs will send to them as normal. But the learners will not be involved in the leader’s election.

## Transfer Leadership

Transfer leadership is extremely important for balance. When moving a partition from one machine to another, NebulaGraph first checks if the source is a leader. If so, it should be moved to another peer. After data migration is completed, it is important to balance the leader again, so that the load on each machine can be balanced.

When transferring leadership, note that the leader abandons its leadership and the follower starts a leader election. When a transfer leadership command is committed, the original leader abandons its leadership. And as for followers, when receiving the command, it starts a leader election. This uses the same execution process with a common raft leader election. Otherwise, some corner cases will occur.

## Membership Change

To avoid split-brain, when members in a Raft Group change, an intermediate state is required. In such a state, the majority of the old group and new group always have an overlap. Thus it prevents the old or new group from making decisions unilaterally. This is the `joint consensus` mentioned in the thesis. To make it even simpler, in his doctoral thesis Diego Ongaro suggests adding or removing a peer once to ensure the overlap between the majority of the new group and the old group . NebulaGraph‘s implementation also uses this approach, except that the way to add or remove a member is different. For details, please refer to `addPeer/removePeer` in the Raft Part class.

## Snapshot

When the leader changes during sending snapshot, the follower may only receive half of the snapshot data. So, a partition data cleanup is required. Because multiple partitions share a single storage, how to clean up the data is a cumbersome problem. In addition, during the snapshot process, a large amount of IO is generated. Considering performance, you do not want this process to share an IO threadPool with the normal Raft. The process requires a large amount of memory. Thus, how you optimize the memory usage is critical for performance.

# Storage Service

- `getNeighbors`: query the in-edge or out-edge of a set of vertices, return the edges and the corresponding properties, support conditional filters
- `Insert vertex/edge`: insert a vertex or edge and its properties
- `getProps`: get the properties of a vertex or an edge

Graph semantics interfaces are translated into kv operations in this layer. To improve the performance of traversal, concurrent operations are also required.

# Meta Service

On the kv store interface, NebulaGraph encapsulates a set of meta-related interfaces. Meta service supports CRUD on schema, cluster management, and user management. Meta service also supports single-machine deployment and multiple replicas to ensure data security.

# You might also like

1. [NebulaGraph Architecture — A Bird’s View](https://nebula-graph.io/posts/nebula-graph-architecture-overview/)
1. [An Introduction to NebulaGraph's Query Engine](https://nebula-graph.io/posts/nebula-graph-query-engine-overview/)
1. [The Review on Graph Database](https://nebula-graph.io/posts/review-on-graph-databases/)

This article introduces to you the storage design of NebulaGraph, an open-source distributed graph database.