This article provides the performance test report for NebulaGraph Community Edition v3.5.0.

NebulaGraph Database v3.5.0 Benchmark Report

Memory management in databases is an important module in database kernel design, and measurable and controllable memory is an important guarantee for the stability of the database. Similarly, memory management is also crucial for NebulaGraph Database.
The multi-degree association query feature of graph databases often leads to a huge demand for memory in the execution layer of graph databases. This article mainly introduces the new feature Memory Tracker in NebulaGraph v3.4, aiming to achieve fine-grained control of memory usage by the Memory Tracker module, reduce the risk of system OOM kill for graphd and storaged, and improve the kernel stability of NebulaGraph database.


# Available Memory


Before diving into Memory Tracker, let's introduce some relevant background knowledge: available memory.

## Process Available Memory

Here, we briefly introduce how the system determines available memory under various modes.

**Physical Machine Mode**

The database kernel reads the system directory /proc/meminfo to determine the actual memory and available memory in the current environment. Memory Tracker takes "actual physical memory" as the "maximum memory that the process can use".

**Container/cgroup Mode**

In the `nebula-graphd.conf` file, there is a configuration item `FLAG_containerized` to determine whether the database is running in a container. After setting `FLAG_containerized` (false by default) to `true`, the kernel reads the files under the relevant `cgroup path` to determine how much memory the current process can use. There are two versions of `cgroup` - v1 and v2, and v2 is used as an example here.

| FLAG                               | Default Value                       | Explanation                           |
| :--------------------------------- | :---------------------------- | :----------------------------- |
| FLAG_cgroup_v2_memory_max_path     | /sys/fs/cgroup/memory.max     | Determine the maximum memory usage by reading the path |
| FLAG_cgroup_v2_memory_current_path | /sys/fs/cgroup/memory.current | Determine the current memory usage by reading the path |

For example, to control the memory quota for graphd and storaged on a single machine separately, follow these steps:：

Step 1：Set `FLAG_containerized=true`；

Step 2：Create `/sys/fs/cgroup/graphd/`，`/sys/fs/cgroup/storaged/`，and configure `memory.max` under their respective directories；

Step 3：Add relevant configuration in `etc/nebula-graphd.conf`，and `etc/nebula-storaged.conf`

```
--containerized=true
--cgroup_v2_controllers=/sys/fs/cgroup/graphd/cgroup.controllers
--cgroup_v2_memory_stat_path=/sys/fs/cgroup/graphd/memory.stat
--cgroup_v2_memory_max_path=/sys/fs/cgroup/graphd/memory.max
--cgroup_v2_memory_current_path=/sys/fs/cgroup/graphd/memory.current
```

## Memory Tracker Available Memory

After obtaining the "process available memory", the system needs to convert it into memory that Memory Tracker can track. There is a conversion formula between "process available memory" and "Memory Tracker available memory".

```
memtracker_limit = ( total - FLAGS_memory_tracker_untracked_reserved_memory_mb ) *  FLAGS_memory_tracker_limit_ratio
```
![usable_memory](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/usable_memory.png)

| **FLAG**                                    | **Default Value** | **Explanation**                                                     | **Dynamic Change** |
| :------------------------------------------ | ---------- | ------------------------------------------------------------ | -------------- |
| memory_tracker_untracked_reserved_memory_mb | 50 M        | Memory Tracker manages memory allocated through new/delete, but there may be other ways a process consumes memory besides this method; for example, by calling low-level malloc/free. Memory not tracked by this flag is subtracted during calculations. | Yes            |
| memory_tracker_limit_ratio                  | 0.8        | Memory Tracker allows you to specify the proportion of memory it can use. In some scenarios, we may need to reduce this value to prevent out-of-memory (OOM) issues. | Yes            |

Here we'll elaborate on the usage of `memory_tracker_limit_ratio`:

- In a mixed deployment environment where multiple graphd or storaged instances are deployed together, it needs to be reduced. For example, if graphd only occupies 50% of the memory, you need to manually change it to 0.5 in `nebula-graphd.conf`;
- Value range: In addition to the (0,1] value range for `memory_tracker_limit_ratio`, there are two special values defined:
   - `2`：Dynamically adjusts available memory based on the current system runtime environment as perceived by the database kernel. This method is not real-time and may have some degree of inaccuracy;
   - `3`：The limit will be set to an extremely large value, effectively disabling Memory Tracker;

# How Memory Tracker is Designed and Implemented

Next, let's discuss the design and implementation of Memory Tracker. The overall Memory Tracker design consists of several sub-modules, including **Global new/delete operator**, **MemoryStats**, **system malloc**, and **Limiter**. In this section, we'll focus on the Global new/delete operator and MemoryStats modules.


![memory_tracker](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/memory_tracker.png)

## Global new/delete operator

Memory Tracker overloads the global new/delete operator, taking control of memory allocation and deallocation, allowing memory quota allocation management before actual memory allocation takes place. This process is broken down into two steps:

- Step 1: Report memory allocation through MemoryStats；
- Step 2: Invoke jemalloc for actual memory allocation behavior；

> **jemalloc**：Memory Tracker does not change the underlying malloc mechanism, still using jemalloc for memory allocation and deallocation.

## MemoryStats

Global memory usage statistics are managed through GlobalMemoryStats and ThreadMemoryStats for global memory and thread-internal memory, respectively;

## ThreadMemoryStats

`thread_local`  variable, execution engine threads maintain their MemoryStats in their ThreadMemoryStats, including "memory Reservation information" and "whether to allow exceptions to be thrown on memory exceeded";

- **Reservation**

Each thread reserves a 1 MB memory quota to avoid frequent requests to GlobalMemoryStats for quota. When applying or returning, ThreadMemoryStats will use a larger memory block as the exchange unit with the global scope. 

- **alloc**：

Only when the locally reserved 1 MB memory is used up, will it ask the global scope for the next 1 MB. This approach is used to minimize the frequency of requesting memory quota from the global scope;


- **dealloc**：

The returned memory is first added to the thread's reserved memory, and when the reserve quota exceeds 1 MB, 1 MB is returned, and the rest is kept;

```
 // Memory stats for each thread.
 struct ThreadMemoryStats {
   ThreadMemoryStats();
   ~ThreadMemoryStats();
 
   // reserved bytes size in current thread
   int64_t reserved;
   bool throwOnMemoryExceeded{false};
 };
```

- **throwOnMemoryExceeded**  

Whether to throw an exception when the thread exceeds the memory quota. Only when `throwOnMemoryExceeded` is set to true will `throw std::bad_alloc` occur. See the `Catch std::bac_alloc` section for scenarios where `throw std::bad_alloc` needs to be disabled.

### GlobalMemoryStats

Global memory quota, maintaining limit and used variables.

- **limit**：limit: Calculated from the runtime environment and configuration information, it represents the maximum memory Memory Tracker can manage. Limit serves the same purpose as the Limiter module. For detailed memory conversion, see the "Memory Tracker Available Memory" section above;

- **used**：Atomic variable, summarizing all reported used memory from threads (including reserved portions). If used + try_to_alloc > limit, and when `throw OnMemoryExceeded` is set to true, `std::bad_alloc` exception will be thrown.


## Catch std::bac_alloc

Since Memory Tracker overloads new/delete and affects all threads, including third-party threads, throwing `bad_alloc` may cause unintended behavior in some third-party threads. To prevent such issues, we use an active memory detection approach in the code path, enabling memory detection in operators, RPC modules, etc.

## Memory detection in operators

In graph/storage operators, add `try...catch` (performing calculations/allocating memory in the current thread) and `thenError` (asynchronously submitting calculation tasks via `folly::Executor`) to detect Memory Tracker throwing `std::bad_alloc`. 

The database then returns an error code through Status, causing the query to fail;
When debugging memory, you can enable the `FLAGS_memory_tracker_detail_log` configuration item in the `nebula-graphd.conf` file and reduce the `memory_tracker_detail_log_interval_ms` to observe memory usage before and after the query;

**folly::future asynchronous execution**

```c++
thenValue([this](StorageRpcResponse<GetNeighborsResponse>&& resp) {
    memory::MemoryCheckGuard guard;
    // memory tracker turned on code scope
    return handleResponse(resp);
})
.thenError(folly::tag_t<std::bad_alloc>{},
    [](const std::bad_alloc&) {
    // handle memory exceed
})
```

**Synchronous execution**

```c++
memory::MemoryCheckGuard guard; \
try {
    // ...
} catch (std::bad_alloc & e) { \
    // handle memory exceed
}
```

## Memory detection in RPC

RPC mainly addresses memory quota control issues during the serialization/deserialization of Request/Response objects. Since storaged responses are all encapsulated in the DataSet structure, the problem is transformed into memory detection during DataSet serialization and deserialization.

**Serialization**： DataSet object construction is in the NebulaGraph operator result return logic, and by default, memory detection is enabled;

**Deserialization**：Explicitly enabled through `MemoryCheckGuard`, exceptions can be caught in `StorageClientBase::getResponse's onError`;

## Error Codes

To help identify which module has issues, NebulaGraph has added relevant error codes, indicating memory exceeded exceptions in graphd and storaged respectively:

```
E_GRAPH_MEMORY_EXCEEDED = -2600, // Graph memory exceeded
E_STORAGE_MEMORY_EXCEEDED = -3600, // Storage memory exceeded
```

# How NebulaGraph Works
1. [NebulaGraph Architecture — A Bird’s Eye View](https://www.nebula-graph.io/posts/nebula-graph-architecture-overview)
2. [Benchmark: NebulaGraph vs Dgraph vs JanusGraph](https://www.nebula-graph.io/posts/benchmarking-mainstraim-graph-databases-dgraph-nebula-graph-janusgraph)
3. [Comparison: Neo4j vs NebulaGraph vs JanusGraph](https://www.nebula-graph.io/posts/performance-comparison-neo4j-janusgraph-nebula-graph)
4. [Deploy the Graph Database on Kubernetes](https://www.nebula-graph.io/posts/how-to-deploy-nebula-graph-in-kubernetes)


This article mainly introduces the new feature Memory Tracker in NebulaGraph v3.4, aiming to achieve fine-grained control of memory usage by the Memory Tracker module, reduce the risk of system OOM kill for graphd and storaged, and improve the kernel stability of NebulaGraph graph database.

Memory Tracker:  Memory Management Practice in NebulaGraph Database

With the rapid development of network information technology, data is gradually developing towards multi-source heterogeneity. Inside the multi-source heterogeneous data lies countless inextricable relations. And this kind of relations, together with the the network structure, are sure essential for data analysis. Unfortunately, when it comes to large scale data analysis, the traditional relational databases are poor in association detection and expressions. Therefore, graph data has attracted great attention for its powerful ability in expressions. Graph computing uses a graph model to express and solve the problem. Graphs can integrate with multi-source data types. In addition to displaying the static basic features for data, graph computing also finds its chance to display the graph structure and relationships hidden in the data. Thus graph becomes an important analysis tool in social network, recommendation system, knowledge graph, financial risk control, network security and text retrieval.

# Practical Algorithms

In order to support the business needs of large-scale graph computing, [Nebula Graph](https://github.com/vesoft-inc/nebula) provides [PageRank](https://en.wikipedia.org/wiki/PageRank) and [Louvain](https://en.wikipedia.org/wiki/Louvain_method) community-discovered graph computing algorithms based on [GraphX](https://github.com/apache/spark/tree/master/graphx). Users can execute these algorithm applications by submitting Spark tasks. In addition, users can also write Spark programs by using Spark Connector to call other built-in graph algorithms in GraphX, such as LabelPropagation, ConnectedComponent, etc.

## PageRank

PageRank is an algorithm raised by Google to rank web pages in their search engine results. PageRank is a way of measuring the importance of website pages.

### Introduction to PageRank

PageRank was named after Larry Page and Sergey Brin from Stanford in the United States. PageRank works by counting the number and quality of links to a page to determine a rough estimate of how important the website is. The underlying assumption is that more important websites are likely to receive more links from other websites.

### Applications of PageRank

#### Content recommendation based on similarity

With the help of the PageRank, you can recommend similar content to users based on their browse history and view time when analyzing social applications such as Twitter and Facebook.

#### Analyze the social influence of users

You can analyze the social influence of users according to their PageRank values in social network analysis.

#### Importance research for papers

Judge a paper's quality according to its PageRank value. Actually the PageRank algorithm comes from the idea of judging the quality of the papers. In addition, PageRank also finds its usage in data analysis and mining.

### Implementation PageRank

PageRank in GraphX is implemented based on the Pregel computing model. The algorithm contains three procedures:

1. Set a same initial PageRank value for every vertex (web page) in the graph;
1. The first iteration: Send a message along the edge. Each vertex receives all the vertices information along its related edges and gets a new PageRank value;
1. The second iteration: Put the PageRank values you get in the first iteration to the formulas corresponding to different algorithm models and get the new PageRank values.

## Louvain method

The Louvain method for community detection is a method to extract communities from large networks. The method is an aggregation algorithm for graphs.

### Introduction to Louvain

The inspiration for Louvain is the optimization of modularity as the algorithm progresses. If a node joins a certain community and maximizes the modularity of the community compared to other communities, then the node belongs to the community. If a node doesn't increase the modularity after joining other communities, it should stay in the current community.

#### Modularity

##### Modularity formula

The modularity Q measures the density of links inside communities compared to links between communities. Modularity is defined as:

$$Q = \frac{1}{2m} \times \sum_{i,j}[A_{ij} - \frac{k_ik_j}{2m}]\delta(c_i,c_j)  \\
 conditions: \delta(u,v) = \begin{cases}
                 1 &\text{when u == v} \\
                 0 &\text {}else
                 \end{cases}
$$                 

where

- $A_{ij}$ represents the edge weight between nodes $i$ and $j$.
- $K_i$ reprsents the sum of the weights of the edges attached to nodes $i$.
- $C_i$ is the community of the node $i$.
- $m$ is the sum of all of the edge weights in the graph.

**Deformation for the modularity formula**

In this formula, the formula is only meaningful when node i and node j belong to the same community. Therefore the formula measures the closeness within a certain community. The simplified deformation of this formula is as follows:

$ Q = \frac{1}{2m} \times \sum{i,j}[A{i,j} - \frac{k_ik_j}{2m}]\delta(c_i,c_j) \\
= \frac{1}{2m}[\sum{i,j}A{i,j} - \frac{\sum_ik_i \sum_jk_j}{2m}]\delta(c_i,c_j) \\
= \frac{1}{2m} \sum_c[\sum{in} - \frac{{\sum{tot}}^{2}}{2m}] $

where

- $\Sigma_{in}$ represents the sum of the weights of the edges in community c.
- $\Sigma_{k_n}$ represents the sum of the weighs of the edges connected with the nodes in community c (because $i$ belongs to community c), including the edges inside and outside the community attached to the node $i$.
- $\Sigma_{k_j}$ represents the sum of the weighs of the edges connected with the nodes in community c (because $j$ belongs to community c), including the edges inside and outside the community attached to the node $j$.
- $\Sigma{tot}$ replaces $\Sigma{k_i}$ and $\Sigma_{k_j}$. (Namely the sum of the weights of the edges in community c plus the sum of the weights of the edges in other community attached to community c. )

**Calculate the modularity change**

In the Louvain method, there is no need to calculate the specific modularity for each community. You only need to compare the modularity changes after adding a certain node to the community. That is to say you only need to calculate △Q.

When inserting node i to a certain community, the modularity change is:

$$\Delta{Q} = [\frac{\sum{in} + k{i,in}}{2m} -({\frac{\sum{tot} + k_i}{2m}})^{2}] - [\frac{\sum{in}}{2m} - (\frac{\sum_{tot}}{2m})^{2} - (\frac{k_i}{2m})^{2}] \\
= [\frac{k{i,in}}{2m} - \frac{\sum{tot}k_i}{2m^2}]$$

where

- $k_{i,in}$ is the sum of the weights of the edges between $i$ and other nodes in the community (This is 2 times of the actual weight in the new community, because all the edges attached to node i in the community are calculated twice).
- $K_i$ is the weighted degree of $i$. Thus when implementing the algorithm, you only need to calculate the $[k{i,in} - \sum{tot} \times \frac{k_i}{m}]$.

### Applications of Louvain

- Financial risk control

In financial risk control scenarios, you can use the method to identify fraud gangs based on the users behaviors.

- Social network

Louvain divides the social network based on the breadth and strength of nodes association in a graph. Louvain also analyzes complex networks and the closeness among a group of people.

- Recommendation system

Community detection based on the users’ interests. More accurate and effective customized recommendation can be implemented based on the community and collaborative filtering algorithm.

### Implement Louvain

The Louvain method contains two stages. The procedure is actually the iteration of the two stages.

**Stage 1:** Continuously traverse the nodes in the graph, and compare the modularity changes introduced by nodes in each neighbor community. Then add a single node to the community that can maximize the modularity. (For example, node v is added to communities A, B, and C respectively. The modularity increments of the three communities are -1, 1, 2. Then node v should be added to community C.)

**Stage 2:** Process based on the first stage. The nodes belonging to the same community are merged into a super node to reconstruct the graph. That is,  the community is regarded as a new node in the graph. At this time, the weight of the edges between the two super nodes are the sum of the weight of the edges attached to the original nodes in the two super nodes. That is, the sum of the weight of the edges in the two communities.

Following are details for the two stages.

![Louvain](https://user-images.githubusercontent.com/57335825/99624951-03ebcf00-29e4-11eb-9000-911efdf519d1.png)

In the first stage, the nodes are traversed and then added to the communities they belong to. In this way, we get the middle graph and the four communities.

In the second stage, the nodes in the community are merged into a super node. The community nodes have self-connected edges whose weight is twice the sum of the weights of the edges connected between all nodes in the community. The edge weight between the communities is the sum of the weights of the edges connecting the two communities. For example, the red community and the light green community are connected by (8,11), (10,11), (10,13). So the weight of the edges between the two communities is 3.

> Note: The weight inside the community is twice the weight of the edges between all internal nodes, because the concept of Kin is the sum of the edges of all nodes in the community and node i. When calculating the Kin for a certain community, each edge is actually counted twice.

The Louvain method continuously iterates stage 1 and stage 2 until the algorithm is stable (the modularity for the graph does not change) or the iterations reach the maximum.

# Practice the proceeding algorithms

## Demo environment

- Three virtual machines:
   - Cpu name: Intel(R) Xeon(R) Platinum 8260M CPU @ 2.30GHz
   - Processors: 32
   - CPU Cores: 16
   - Memory Size: 128G
- Software environment:
   - Spark: spark-2.4.6-bin-hadoop2.7 a cluster with three nodes
   - yarn V2.10.0: a cluster with three nodes
   - NebulaGraph V1.1.0: distributed deployed, used the default configurations

## Dataset for testing

1. Create graph space

```shell
CREATE SPACE algoTest(partition_num=100, replica_factor=1);
```

2. Create schema

```shell
CREATE TAG PERSON()
CREATE EDGE FRIEND(likeness double);
```

3. Import data

Use Spark Writer to import data offline into NebulaGraph.

4. Test results

The resource allocation fot the Spark job is `--driver-memory=20G --executor-memory=100G --executor-cores=3`.

- PageRank algorithm execution time on a dataset with hundred million nodes is 21 minutes.
- Louvain algorithm execution time on a dataset with hundred million nodes is 1.3 hours.

## How to use NebulaGraph algorithm

1. Download and pack the nebula-algorithm project to a jar package.

```shell
$ git clone git@github.com:vesoft-inc/nebula-java.git
$ cd nebula-java/tools/nebula-algorithm
$ mvn package -DskipTests
```

2. Configure the file `src/main/resources/application.conf`.

```xml
{
  # Spark relation config
  spark: {
    app: {
        # not required, default name is the algorithm that you are going to execute.
        name: PageRank

        # not required
        partitionNum: 12
    }

    master: local

    # not required
    conf: {
        driver-memory: 8g
        executor-memory: 8g
        executor-cores: 1g
        cores-max:6
    }
  }

  # NebulaGraph relation config
  nebula: {
    # metadata server address
    addresses: "127.0.0.1:45500"
    user: root
    pswd: nebula
    space: algoTest
    # partition specified while creating nebula space, if you didn't specified the partition, then it's 100.
    partitionNumber: 100
    # nebula edge type
    labels: ["FRIEND"]

    hasWeight: true
    # if hasWeight is true，then weightCols is required， and weghtCols' order must be corresponding with labels.
    # Noted: the graph algorithm only supports isomorphic graphs,
    #        so the data type of each col in weightCols must be consistent and all numeric types.
    weightCols: [“likeness”]
  }

  algorithm: {
    # the algorithm that you are going to execute，pick one from [pagerank, louvain]
    executeAlgo: louvain
    # algorithm result path
    path: /tmp

    # pagerank parameter
    pagerank: {
        maxIter: 20
        resetProb: 0.15  # default 0.15

    }

    # louvain parameter
    louvain: {
        maxIter: 20
        internalIter: 10
        tol: 0.5
   }
  }
}
```

3. Make sure that you have installed and started the Spark service on your machine.
4. Submit the nebula algorithm application:

```shell
spark-submit --master xxx --class com.vesoft.nebula.tools.algorithm.Main /your-jar-path/nebula-algorithm-1.0.1.jar -p /your-application.conf-path/application.conf
```

If you are interested in the content, welcome to try nebula-algorithm.

# References

- NebulaGraph: [https://github.com/vesoft-inc/nebula](https://github.com/vesoft-inc/nebula)
- GraphX: [https://github.com/apache/spark/tree/master/graphx](https://github.com/apache/spark/tree/master/graphx)
- nebula-algorithm: [https://github.com/vesoft-inc/nebula-java/tree/v1.0/tools/nebula-algorithm](https://github.com/vesoft-inc/nebula-java/tree/v1.0/tools/nebula-algorithm)

> Hi, I’m Nicole, the software engineer in NebulaGraph. Let me know if you have any thoughts and suggestions on this article. Welcome to talk to me on our [forum](https://discuss.nebula-graph.io/). Thanks!

# You Might Also Like

- [Exploring the S&P 100 Index Stocks Using Graph Machine Learning](https://nebula-graph.io/posts/stock-interrelation-analysis-nebula-graph-machine-learning/)
- [Analyzing Relationships in Game of Thrones With NetworkX, Gephi, and NebulaGraph (Part Two)](https://nebula-graph.io/posts/game-of-thrones-relationship-networkx-gephi-nebula-graph-part-two/)


In order to support the business needs of large-scale graph computing, NebulaGraph provides PageRank and Louvain community-discovered graph computing algorithms based on GraphX. Users can execute these algorithm applications by submitting Spark tasks. In addition, users can also write Spark programs by using Spark Connector to call other built-in graph algorithms in GraphX, such as LabelPropagation, ConnectedComponent, etc.

Practicing Graph Computation with GraphX in NebulaGraph

# Dag Controller Introduction

Dag Controller is a component of NebulaGraph Explorer, which was released after several tests and tunings. It mainly solves the problem of integration of OLTP and OLAP, and the problem of graph algorithms in complex scenarios. To learn more, see [https://docs.nebula-graph.io/master/nebula-explorer/deploy-connect/ex-ug-deploy/#dag_controller](https://docs.nebula-graph.io/master/nebula-explorer/deploy-connect/ex-ug-deploy/#dag_controller).

Below is some information I would like to share about Dag Controller. Feel free to leave a comment and let's discuss it together.

I'm sure you're no stranger to OLTP and OLAP, but let me briefly introduce them again here: OLTP is a fast-response, real-time online data processing method. OLAP is an offline, data algorithm method for complex scenarios. For NebulaGraph, OLTP has a variety of query statements, such as GO, MATCH, etc. OLAP has a variety of graph algorithms, such as Pagerank, Louvain, WCC (Weakly Connected Components), K-Core, and Jaccard.

OLTP and OLAP do not exist independently. For example, we can use the subgraph returned by a MATCH statement as input to the PageRank algorithm, and the result of the PageRank can be written to NebulaGraph to continue executing a MATCH statement or other algorithms. Like building blocks, various [OLTP and OLAP](https://www.nebula-graph.io/posts/OLTP-vs-OLAP-what-are-the-differences) can be assembled to form a data processing method for complex scenarios.

The Dag Controller is a system that handles scenarios where OLTP and OLAP need to be linked and executed. 

![oltp-vs-olap](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/olap-and-oltp/1.png)

## Architecture

![dag-architecture](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/olap-and-oltp/2.png)

The responsibilities of Dag Controller:

1. It provides an HTTP interface for external communication and access.

It is used for job submission, stopping, deleting and other operations, as well as the configuration of the system environment.

2. Execution of DAG.

When Dag Controller executes DAG, the OLAP will call NebulaGraph Analytics, and the OLTP will call graphd to complete the execution of nGQL.

NebulaGraph Analytics is our graph computing system, supporting Pagerank, Louvain, WCC, Jaccard, and other graph algorithms, supporting HDFS and NebulaGraph data sources.

Graphd, metad, and storaged are components in NebulaGraph. Graphd is mainly responsible for nGQL parsing, storaged is responsible for data storage, and metad is responsible for metadata storage.

# Cases

## Case 1

![dag-use-case](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/olap-and-oltp/4.png)

The above figure shows a DAG model that runs the pagerank algorithm on a subgraph. First, we get a subgraph using an nGQL statement, and then we run the pagerank algorithm on this subgraph.

This method can be used when our graph size is particularly large and we only want to run the algorithm on part of the graph data.

## Case 2

![olap-and-oltp-use-case](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/olap-and-oltp/3.png)

The above figure shows a model to compute the shortest paths for two classes of vertices.

First, the vertex IDs of the two classes are obtained separately by using nGQL, and then the vertex IDs of the two classes are passed to the shortestpath algorithm, which computes the paths between the two classes in the full graph.

Each algorithm can be set up to run the algorithm based on the full graph or subgraph.

There are various DAG models, and different DAG models can be built according to different business scenarios.

# Technical implementation

## DAG model

The DAG refers to a directed graph without loops, and an instance of DAG is considered as a job, and a job has multiple task.

The task in Dag Controller can be an nGQL or a graph algorithm such as PageRank, Louvain, SSSP, etc.

When a job is executed, it needs to sort the task first, there is a lot of related code on the internet, so I won't repeat it here.

## Parallel execution

To ensure the execution efficiency of DAG, multiple DAGs need to be executed in parallel. At the same time, within a DAG, the task without upstream and downstream dependencies also needs to be executed in parallel.

How can multiple DAGs and tasks be executed in parallel? In simple terms, the DAGs and tasks are handled by two thread pools respectively.

The details are as follows:

1. When the system starts, allocate the job thread pool and task thread pool to handle the execution of job and task respectively.

2. Unexecuted jobs are periodically obtained from the database and transferred to the job thread pool.

3. When a job is executed, the task is sorted according to its upstream and downstream dependencies, and then it determines whether the execution of all upstreams of each task is completed, and then the task is handed over to the task thread pool for execution when the upstream execution is completed, or waits if the upstream execution is not completed.

4. If the job thread pool is full during job execution, you need to wait to obtain unexecuted jobs at regular intervals. The same is done when the task thread pool is full.

## Type check

Note that the data type verification problem exists between the data input and output of tasks. For example, task2 is the downstream of task1. The input of task2 must be of type int, and the output of task1 must be of type int.

## DAG stop

When you stop a job, you need to stop multiple task that are running in parallel. A task has a preparation phase, a running phase, and in the running phase there are cross-machine, multi-process situations. Stopping a job requires avoiding the orphan process problem.

## Custom algorithm support

We support treating a customer's algorithm as a task for DAG building. First, configure the algorithm-related parameter information in the system. When executing a job, the system is responsible for running the algorithm that corresponds to the task.

# How NebulaGraph Works
1. [NebulaGraph Architecture — A Bird’s Eye View](https://www.nebula-graph.io/posts/nebula-graph-architecture-overview)
2. [Benchmark: NebulaGraph vs Dgraph vs JanusGraph](https://www.nebula-graph.io/posts/benchmarking-mainstraim-graph-databases-dgraph-nebula-graph-janusgraph)
3. [Comparison: Neo4j vs NebulaGraph vs JanusGraph](https://www.nebula-graph.io/posts/performance-comparison-neo4j-janusgraph-nebula-graph)
4. [Deploy the Graph Database on Kubernetes](https://www.nebula-graph.io/posts/how-to-deploy-nebula-graph-in-kubernetes)


Dag controller mainly solves the problem of integration of OLTP and OLAP, and the problem of graph algorithms in complex scenarios. 

Integration  Practice of OLTP and OLAP


# What is geospatial data?

Geospatial data is information related to geospatial entities, such as points, lines, and polygons.

NebulaGraph 2.6 supports geospatial data. You can store, compute, and retrieve geospatial data in NebulaGraph. Geography is a data type supported in NebulaGraph. It is composed of latitude and longitude that represents geospatial data.

## How to use geospatial data in NebulaGraph?

**Create Schema**

The following example shows how to create tags. You can create edge types in the same way.

NebulaGraph currently supports three types of geospatial data: Point, LineString, and Polygon. The following shows how to create geography types and how to insert geospatial data.

```ngql
CREATE TAG any_shape(geo geography);
CREATE TAG only_point(geo geography(point));
CREATE TAG only_linestring(geo geography(linestring));
CREATE TAG only_polygon(geo geography(polygon));
```

When no geography type is specified, it means that you can store data of any type; when a type is specified, it means that only geospatial data of that type can be stored, such as `geography (point)`, which means that you can only store spatial information of points.

**Insert Data**

Insert data in the `geo` column of the `any_shape` tag.

```ngql
INSERT VERTEX any_shape(geo) VALUES "101":(ST_GeogFromText("POINT(120.12 30.16)"));
INSERT VERTEX any_shape(geo) VALUES "102":(ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
INSERT VERTEX any_shape(geo) VALUES "103":(ST_GeogFromText("POLYGON((75.3 45.4, 112.5 53.6, 122.7 25.5, 93.9 28.6, 75.3 45.4))"));
```

Insert data in the `geo` column of the `only_point` tag.

```ngql
INSERT VERTEX only_point(geo) VALUES "201":(ST_Point(120.12，30.16)"));;
```

Insert data in the `geo` column of the `only_linestring` tag.

```ngql
INSERT VERTEX only_linestring(geo) VALUES "302":(ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
```

Insert data in the `geo` column of the `only_polygon` tag.

```ngql
INSERT VERTEX only_polygon(geo) VALUES "403":(ST_GeogFromText("POLYGON((75.3 45.4, 112.5 53.6, 122.7 25.5, 93.9 28.6, 75.3 45.4))"));
```

When the data inserted does not meet the requirements of the specified type, the data insertion fails.

```ngql
(root@nebula) [geo]> INSERT VERTEX only_polygon(geo) VALUES "404":(ST_GeogFromText("POINT((75.3 45.4))"));
[ERROR (-1005)]: Wrong value type: ST_GeogFromText("POINT((75.3 45.4))")
```

We can see that the geospatial data insertion method is rather peculiar, and is very different from the insertion of basic types such as `int`, `string`, and `bool`.

Let's take `ST_GeogFromText("POINT(120.12 30.16)")` as an example, `ST_GeogFromText` is a geographic location information parsing function, which accepts a string type of geographic location data in WKT (Well-Known Text) standard format.

`POINT(120.12 30.16)` represents a geographic point with longitude 120°12′E and latitude 30°16′N; the `ST_GeogFromText` function parses and constructs a geography data object from the WKT parameter, and then the `INSERT` statement stores it in the NebulaGraph in WKB (Well-Known Binary) standard.

## Geospatial functions

The geospatial functions supported by NebulaGraph can be divided into the following main categories:

- Constructing functions

  - `ST_Point(longitude, latitude)`: Constructs a `geography point` object based on a latitude and longitude pair.

- Parsing functions

  - `ST_GeogFromText(wkt_string)`: Parses `geography` objects from the WKT text.

  - `ST_GeogFromWKB(wkb_string)`: Parses `geography` objects from the WKB text. # Not yet supported, because NebulaGraph does not yet support binary strings.

- Format setting functions

  - `ST_AsText(geography)`: Outputs the `geography` object in the WKT text format.

  - `ST_AsBinary(geography)`: Outputs the `geography` object in the WKB text format. # Not yet supported, because NebulaGraph does not yet support binary strings.

- Conversion functions

  - `ST_Centroid(geography)`: Calculates the center of gravity of a `geography` object, which is a `geography point` object.

- The predicate function

  - `ST_Intersects(geography_1, geography_2)`: Determines whether two `geography` objects intersect.

  - `ST_Covers(geography_1, geography_2)`: Determines if the first `geography` object completely covers the second.

  - `ST_CoveredBy(geography_1, geography_2)`: The inverse of ST_Covers.

  - `ST_DWithin(geography_1, geography_2, distance_in_meters)`: Determines if the shortest distance between two `geography` objects is less than the given distance.

- The metric function

  - `ST_Distance(geography_1, geography_2)`: Calculates the distance between two `geography` objects.

These function interfaces follow the OpenGIS Simple Feature Access and ISO SQL/MM standards. For details, see [NebulaGraph doc](https://docs.nebula-graph.io/3.3.0/3.ngql-guide/6.functions-and-expressions/14.geo/).

## Geospatial Index

**What is a geospatial index?**

Geospatial indexes are indexes that can be used to quickly filter data based on the predicate `ST_Intersects` and `ST_Covers` functions. 

NebulaGraph uses the Google S2 library as the geospatial index. 

The S2 library projects the Earth's surface into a tangent square, then recursively quadruples each square surface of the square n times, and uses a space-filling curve, the Hilbert curve, to connect the centers of these small square lattices.

When n is infinitely large, this Hilbert curve almost fills the square.

The S2 library uses a Hilbert curve of order 30.

![](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/Exploring-Geospatial-Data-with-NebulaGraph/1.jpeg)

The following figure shows that the Earth is filled with Hilbert curves.

![](https://www-cdn.nebula-graph.io/nebula-website-5.0/images/blogs/Exploring-Geospatial-Data-with-NebulaGraph/2.jpeg)


It can be seen that the Earth's surface is divided into cells by these Hilbert curves. For any geographic shape on the earth's surface, such as a city, a river, or a person's location, we can use several of these cells to completely cover the geographic shape.

Each cell is identified by a unique int64 CellID. Thus, the spatial index of a geographic object is the set of S2 cells that are constructed to completely cover the geographic shape.

When constructing an index of a geospatial object, a collection of different S2 cells that completely cover the indexed object is constructed. The indexing query based on spatial predicate functions quickly filters out a large number of irrelevant geographic objects by finding the intersection between the set of S2 cells that cover the queried object and the S2 cells that cover the indexed object.


**Create a geography index**

```ngql
CREATE TAG any_shape_geo_index on any_shape(geo)
```

For geospatial data with the type `point`, it can be represented by an S2 cell of order 30, so a point corresponds to one index entry; for geospatial data with the type `inestring` and `polygon`, we use multiple S2 cells of different levels to cover it, so it will correspond to multiple index entries.

Spatial indexing is used to speed up the lookup of all geo predicates, for example:

```ngql
LOOKUP ON any_shape WHERE ST_Intersects(any_shape.geo, ST_GeogFromText("LINESTRING(3 8, 4.7 73.23)"));
```

When there is no spatial index on the `geo` column of `any_shape`, this statement will first read all the data of `any_shape` into memory and then use it to calculate whether it intersects with the point (3.0, 8.0), which is generally more expensive. When the amount of data in `any_shape` is large, the computation overhead will be unacceptable.

When the `geo` column of `any_shape` has a spatial index, the statement will first use the spatial index to filter out most of the data that intersected by the line, but there will still be some that may be intersected when read into memory, so there is still one more calculation to be done. In this way, the spatial index quickly filters out most of the data that is not likely to intersect at a small cost, and a small percentage is filtered, greatly reducing the computational overhead.

You can store, compute, and retrieve geospatial data in NebulaGraph. Geography is a data type supported in NebulaGraph.

Exploring Geospatial Data with NebulaGraph

Based on the actual demand and subsequent scalability, the NebulaGraph database is used for technical selection. The batch import performance in the actual business scenarios of NebulaGraph needs to be first verified.

This article shares the best practice of using NebulaGraph Spark Connector by performing import jobs via Spark on Yarn distributed tasks and with CSV files on HDFS as the data source.

# About NebulaGraph Spark Connector

The following figure shows the use cases and advantages of NebulaGraph Spark Connector. For more details, see [NebulaGraph Spark Connector](https://docs.nebula-graph.io/3.0.1/nebula-spark-connector/).

<pic>


# Environment Information

- Hardware environment 

| **Name**     |  **Value**  | **Recommend** |
| :----------- | :------: | -------: |
| Local Disk (SSD) |   2 T    | At least 2 T |
| CPU          | 16 C * 4 |    128 C |
| Memory         |  128 GB  |    128 GB |

- Software environment

| **Name**               | **Version**  |
| :--------------------- | :---------: |
| NebulaGraph           |    3.0.0    |
| NebulaGraph Spark Connector |    3.0.0    |
| Hadoop                 | 2.7.2U17-10 |
| Spark                  |   2.4.5U5   |

- Data size

| **Name**     | **Value** |
| :----------- | :----: |
| Data size      | 200 GB  |
| Vertex | 930 million |
| Edge   | 970 million |

# Deployment Plan

- Method: Distributed, 3 nodes
- See [Deploy a NebulaGraph cluster with RPM/DEB package on multiple servers](https://docs.nebula-graph.com.cn/3.0.1/4.deployment-and-installation/2.compile-and-install-nebula-graph/deploy-nebula-graph-cluster/)

There are roughly three steps for the deployment.

1. Download the RPM package of NebulaGraph and install it.
2. Modify configuration files in batch.
3. Start the cluster service.

The above-mentioned three steps are performed with the `root` user. For non-root users, use `sudo` to execute the commands.

## Download and Install the NebulaGraph RPM Package

Execute the following commands:

```bash
wget https://os-cdn.nebula-graph.com.cn/package/3.0.0/nebula-graph-3.0.0.el7.x86_64.rpm
wget https://oss-cdn.nebula-graph.com.cn/package/3.0.0/nebula-graph-3.0.0.el7.x86_64.rpm.sha256sum.txt
rpm -ivh nebula-graph-3.0.0.el7.x86_64.rpm
```

Note: The default installation path is `/usr/local/nebula`. Make sure the memory resources are sufficient.

## Modify Configuration Files in Batch

```bash
sed -i 's?--meta_server_addrs=127.0.0.1:9559?--meta_server_addrs=172.1x.x.15:9559,172.1x.x.176:9559,172.1x.1x.149:9559?g' *.conf
sed -i 's?--local_ip=127.0.0.1?--local_ip=172.1x.1x.149?g' *.conf
sed -i 's?--meta_server_addrs=127.0.0.1:9559?--meta_server_addrs=172.1x.x.15:9559,172.1x.x.176:9559,172.1x.1x.149:9559?g' *.conf
sed -i 's?--local_ip=127.0.0.1?--local_ip=172.1x.x.15?g' *.conf
sed -i 's?--meta_server_addrs=127.0.0.1:9559?--meta_server_addrs=172.1x.x.15:9559,172.1x.x.176:9559,172.1x.1x.149:9559?g' *.conf
sed -i 's?--local_ip=127.0.0.1?--local_ip=172.1x.x.176?g' *.conf
```

Note: The IP addresses in the above commands are internal IP addresses, used for communication between nodes. 


## Start the NebulaGraph Cluster Service

```bash
/usr/local/nebula/scripts/nebula.service start all
```

The above command is to start the cluster service, the following is to check the status of the cluster service.

```bash
ps aux|grep nebula
```

The returned result shows 3 processes for each service.

```bash
/usr/local/nebula/bin/nebula-metad --flagfile /usr/local/nebula/etc/nebula-metad.conf
/usr/local/nebula/bin/nebula-graphd --flagfile /usr/local/nebula/etc/nebula-graphd.conf
/usr/local/nebula/bin/nebula-storaged --flagfile /usr/local/nebula/etc/nebula-storaged.conf
```

Note: If the returned result shows less than 3 processes, run `/usr/local/nebula/scripts/nebula.service start all` several times or run `/usr/local/nebula/scripts/nebula.service restart all` to restart the service.


# Visualization Service

The NebulaGraph Studio is used as the visualize the NebulaGraph cluster service. Visit `http://n01v:7001` to log into the NebulaGraph Studio. (This is my network environment which is not accessible from the Internet.)

<pic>

- Login: 10.x.x.1 (The IP of any node):9669
- Username/Password: `root`/`nebula`

Refer to [Use nGQL (CRUD)](https://docs.nebula-graph.com.cn/3.0.1/2.quick-start/4.nebula-graph-crud) for common-used nGQL statements.


## Use NebulaGraph

Add the Storage nodes to the cluster.

```nebula
ADD HOSTS 172.x.x.121:9779, 172.16.11.218:9779,172.16.12.12:9779;
```

List all nodes and check whether the status of the cluster is `ONLINE by running `SHOW HOSTS` or `SHOW HOSTS META`.
Create a graph space, equivalent to a database in relational databases.
```nebula
CREATE SPACE mylove (partition_num = 15, replica_factor = 3, vid_type = FIXED_STRING(256));//The number of partitions is recommended to be 5 times the number of nodes; the number of replicas is an odd number, generally set to 3; if the data type of vids is string, make the length as long as a reasonable value, otherwise, it takes up too much disk space.
```

Create a Tag, equivalent to an entity.: 

```nebula
CREATE TAG entity (name string NULL, version string NULL);
```  

Create an edge, equivalent to a relationship.

```nebula
CREATE EDGE relation (name string NULL);  
```

Add `limit` when querying data, otherwise, the query may take a long time.

```nebula
match (v) return v limit 100;
```

# Read and Import CSV Files with Spark Connector 

References:

- [NebulaSparkWriterExample (Scala Json) from the official](https://github.com/vesoft-inc/nebula-spark-utils/blob/master/example/src/main/scala/com/vesoft/nebula/examples/connector/NebulaSparkWriterExample.scala)
- [NebulaSparkWriterExample (Java Json) from an ace](https://www.jianshu.com/p/930e0343a28c)

Code example of NebulaSparkWriterExample

```
import com.facebook.thrift.protocol.TCompactProtocol
import com.vesoft.nebula.connector.{
  NebulaConnectionConfig,
  WriteMode,
  WriteNebulaEdgeConfig,
  WriteNebulaVertexConfig
}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession
import org.slf4j.LoggerFactory
object NebulaSparkWriter {
  private val LOG = LoggerFactory.getLogger(this.getClass)
  var ip = ""
  def main(args: Array[String]): Unit = {
    val part = args(0)
    ip = args(1)
    val sparkConf = new SparkConf
    sparkConf
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .registerKryoClasses(Array[Class[_]](classOf[TCompactProtocol]))
    val spark = SparkSession
      .builder()
      .master("local")
      .config(sparkConf)
      .getOrCreate()
    spark.sparkContext.setLogLevel("WARN")
    if("1".equalsIgnoreCase(part)) writeVertex(spark)
    if("2".equalsIgnoreCase(part)) writeEdge(spark)
    spark.close()
  }
  def getNebulaConnectionConfig(): NebulaConnectionConfig = {
    val config =
      NebulaConnectionConfig
        .builder()
        .withMetaAddress(ip + ":9559")
        .withGraphAddress(ip + ":9669")
        .withTimeout(Integer.MAX_VALUE)
        .withConenctionRetry(5)
        .build()
    config
  }
  def writeVertex(spark: SparkSession): Unit = {
    LOG.info("start to write nebula vertices: 1 entity")
    val df = spark.read.option("sep", "\t").csv("/home/2022/project/origin_file/csv/tag/entity/").toDF("id", "name", "version")
    val config = getNebulaConnectionConfig()
    val nebulaWriteVertexConfig: WriteNebulaVertexConfig = WriteNebulaVertexConfig
      .builder()
      .withSpace("mywtt")
      .withTag("entity")
      .withVidField("id")
      .withVidAsProp(false)
      .withUser("root")
      .withPasswd("nebula")
      .withBatch(1800)
      .build()
    df.coalesce(1400).write.nebula(config, nebulaWriteVertexConfig).writeVertices()
  }
  def writeEdge(spark: SparkSession): Unit = {
    LOG.info("start to write nebula edges: 2 entityRel")
    val df = spark.read.option("sep", "\t").csv("/home/2022/project/origin_file/csv/out/rel/relation/").toDF("src", "dst", "name")
    val config = getNebulaConnectionConfig()
    val nebulaWriteEdgeConfig: WriteNebulaEdgeConfig = WriteNebulaEdgeConfig
      .builder()
      .withSpace("mywtt")
      .withEdge("relation")
      .withSrcIdField("src")
      .withDstIdField("dst")
      .withSrcAsProperty(false)
      .withDstAsProperty(false)
      .withUser("root")
      .withPasswd("nebula")
      .withBatch(1800)
      .build()
    df.coalesce(1400).write.nebula(config, nebulaWriteEdgeConfig).writeEdges()
  }
}
```
## Details on NebulaSparkWriterExample 

The following are the details of the functions.

- `spark.sparkContext.setLogLevel("WARN")`: Sets the log print level and prevents the INFO log. 


- `withTimeout(Integer.MAX_VALUE)`: Sets the connection timeout. The timeout can be set as large as possible. The default value is 1 minute. When the number of timeouts is greater than that of retries, the Spark task fails.

- `option("sep", "\t")`: Specifies the separator of CSV files, otherwise, it defaults to 1 column.

- `toDF("src", "dst", "name")`: The schema of the dataset, i.e., converting `Dataset<Row>` to `DataFrame`, otherwise, the `VidField` cannot be specified.

- `withVidField("id")`: This function only works when the column name is set, so a schema must be specified.

- `withVidAsProp(false)`: When the ID is set as the VID by default, data will not be written repeatedly as a property that will not take up disk space.

- `withSrcIdField("src")`: Sets the `IdField` of the source vertex.

- `withDstIdField("dst")`: Sets the `IdField` of the destination vertex.

- `withSrcAsProperty(false)`: Saves disk space.

- `withDstAsProperty(false)`: Saves disk space.

- `withBatch(1000)`: The number of rows of data imported in batch. `WriteMode.UPDATE` is <=512 by default. The value of `WriteMode.INSERT` can be larger (Gigabit NIC/Bandwidth 5Gbps /Local SSD = 1500)

- `coalesce(1500)`: Adjustable according to the number of concurrent tasks. Too much data on a single partition can easily cause executor OOM exceptions.

# Submit Tasks to the Spark Cluster

```spark
nohup spark-submit  --master yarn --deploy-mode client --class com.xxx.nebula.connector.NebulaSparkWriter --conf spark.dynamicAllocation.enabled=false --conf spark.executor.memoryOverhead=10g  --conf spark.blacklist.enabled=false --conf spark.default.parallelism=1000 --driver-memory 10G --executor-memory 12G --executor-cores 4 --num-executors 180 ./example-3.0-SNAPSHOT.jar >  run-csv-nebula.log 2>&1 &
```

## Monitoring with `iotop`

The `iotop` command can be used to monitor the disk IO of the Spark cluster.

```bash
Total DISK READ :      26.61 K/s | Total DISK WRITE :     383.77 M/s
Actual DISK READ:      26.61 K/s | Actual DISK WRITE:     431.75 M/s
```
## Monitoring with `top` 

The `top` command can be used to monitor the CPU and memory of the Spark cluster.

```bash
top - 16:03:01 up 8 days, 28 min,  1 user,  load average: 6.16, 6.53, 4.58
Tasks: 205 total,   1 running, 204 sleeping,   0 stopped,   0 zombie
%Cpu(s): 28.3 us, 14.2 sy,  0.0 ni, 56.0 id,  0.6 wa,  0.0 hi,  0.4 si,  0.5 st
KiB Mem : 13186284+total,  1135004 free, 31321240 used, 99406592 buff/cache
KiB Swap:        0 total,        0 free,        0 used. 99641296 avail Mem 
  PID USER      PR  NI    VIRT    RES    SHR S  %CPU %MEM     TIME+ COMMAND                                                                         
27979 root      20   0 39.071g 0.026t   9936 S 564.6 20.8  83:22.03 nebula-storaged                                                                 
27920 root      20   0 2187476 804036   7672 S 128.2  0.6  17:13.75 nebula-graphd                                                                   
27875 root      20   0 6484644 1.990g   8588 S  58.5  1.6  14:14.22 nebula-metad     
```
## Other Resource Monitoring

<pic>

# Service Optimization

## Configuration Optimization of `nebula-storaged.conf`

Here, I modified the value of parameters in the `nebula-storaged.conf` file.

```bash
# Default reserved bytes for a batch operation.
--rocksdb_batch_size=4096
# Default block cache size used in BlockBasedTable.
# The unit is MB. It is usually set to one third of the memory value. The memory is 128 GB.
--rocksdb_block_cache=44024
############## rocksdb Options ##############
--rocksdb_disable_wal=true
# In JSON, the name and value of each option of rocksdb DBOptions is a string. Such as “option_name”:“option_value”. Use a comma to separate each option.
--rocksdb_db_options={"max_subcompactions":"3","max_background_jobs":"3"}
# In JSON, the name and value of each option of rocksdb ColumnFamilyOptions is a string. Such as “option_name”:“option_value”. Use a comma to separate each option.
--rocksdb_column_family_options={"disable_auto_compactions":"false","write_buffer_size":"67108864","max_write_buffer_number":"4","max_bytes_for_level_base":"268435456"}
# In JSON, the name and value of each option of rocksdb BlockBasedTableOptions is a string. Such as “option_name”:“option_value”. Use a comma to separate each option.
--rocksdb_block_based_table_options={"block_size":"8192"}
# The number of processors per request.
--max_handlers_per_req=10
# The interval time between clusters.
--heartbeat_interval_secs=10
--raft_rpc_timeout_ms=5000
--raft_heartbeat_interval_secs=10
--wal_ttl=14400
# The max value for batch import.
--max_batch_size=1800
# Reduces memory usage when the value is true.
--enable_partitioned_index_filter=true
# The data is indirectly filtered at the bottom storage layer to prevent encountering the super node in the production environment.
--max_edge_returned_per_vertex=10000
```
## Optimizations of the Linux System

```bash
ulimit -c unlimited
ulimit -n 130000
sysctl -w net.ipv4.tcp_slow_start_after_idle=0
sysctl -w net.core.somaxconn=2048
sysctl -w net.ipv4.tcp_max_syn_backlog=2048
sysctl -w net.core.netdev_max_backlog=3000
sysctl -w kernel.core_uses_pid=1
```

# Verify Import Results

```nebula
SUBMIT JOB STATS;
SHOW JOB ${ID}
SHOW STATS;
```
- The entity insertion rate is approximately 27,837 items/s (for this import performance calculation only).

- The relationship insertion rate is about 26,276 items/s (for this import performance calculation only).

- The import performance will be better if the server is better configured. In addition, bandwidth, cross-datacenter or not, disk IO, and even network fluctuations are also factors that affect performance.

# Performance Tests

- Query for the specified entities based on properties.

  ```nebula
  MATCH (v:entity) WHERE v.entity.name == 'Lifespan' RETURN v;
  ```

  It takes 0.002558 (s).

- One-hop queries.

  ```nebula
  MATCH (v1:entity)-[e:propertiesRel]->(v2:attribute) WHERE id(v1) == '70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a' RETURN v2 limit 100;  
  ```
  It takes 0.003571 (s)

- Two-hop queries.

  ```nebula
  MATCH p=(v1:entity)-[e:propertiesRel*1..2]->(v2) WHERE id(v1) == '70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a' RETURN p;
  ```
  It takes 0.005143 (s)

- Query for all the properties of an edge.

  ```nebula
  FETCH PROP ON propertiesRel '70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a' -> '0000002d2e88d7ba6659db83893dedf3b8678f3f80de4ffe3f8683694b63a256' YIELD properties(edge); 
  ```  
  It takes 0.001304 (s)

  ```nebula
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*1]->(v2) return p;
  ```
  å
  It takes 0.02986 (s)

  ```nebula
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*2]->(v2) return p; 
  ```

  It takes 0.07937 (s)

  ```nebula
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*3]->(v2) return p; 
  ```

  It takes 0.269 (s)

  ```nebula
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*4]->(v2) return p;
  ```

  It takes 3.524859 (s)

  ```nebula 
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*1..2]->(v2) return p; 
  ```

  It takes 0.072367 (s)

  ```nebula 
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*1..3]->(v2) return p;
  ```

  It takes 0.279011 (s)

  ```nebula 
  match p=(v:entity{name:"张三"})-[e:entityRel|propertiesRel*1..4]->(v2) return p; 
  It takes 3.728018 (s)
  ```

 - Query for the properties of vertices and the edge in the shortest path in bilateral directions between `A_vid` and `B_vid`.

  ```nebula
  FIND SHORTEST PATH WITH PROP FROM "70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a" TO "0000002d2e88d7ba6659db83893dedf3b8678f3f80de4ffe3f8683694b63a256" OVER * BIDIRECT YIELD path AS p; 
  ```
  It takes 0.003096 (s)

  ```nebula
  FIND ALL PATH FROM "70da43c5e46f56c634547c7aded3639aa8a1565975303218e2a92af677a7ee3a" TO "0000002d2e88d7ba6659db83893dedf3b8678f3f80de4ffe3f8683694b63a256" OVER * WHERE propertiesRel.name is not EMPTY or propertiesRel.name >=0 YIELD path AS p;
  ```

  It takes 0.003656 (s)

# Problems Encountered

## Guava version conflict

```
Caused by: java.lang.NoSuchMethodError: com.google.common.base.Stopwatch.createStarted()Lcom/google/common/base/Stopwatch;
```

After troubleshooting, I found that one of the dependent modules used Guava version 22.0, while the Spark cluster came with 14.0, resulting in a conflict, and it did not work properly. For tasks running on Spark clusters, Spark loads Guava packages with higher priority than its packages.

The package we depend on uses methods that are newer in Guava version 22.0 and not available in version 14.0. Without being able to modify each other's code, the following options are available:

1. The Spark cluster package can be upgraded, which is risky and can cause unknown problems.

2. Rename your Guava package using the Maven plugin.

The second way is used here, using the Maven plugin shade (link: https://maven.apache.org/plugins/maven-shade-plugin/) to rename the package to solve the problem.

```java
<plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-shade-plugin</artifactId>
    <version>3.2.4</version>
    <executions>
        <execution>
            <phase>package</phase>
            <goals>
                <goal>shade</goal>
            </goals>
            <configuration>
                <relocations>
                    <relocation>
                        <pattern>com.google.common</pattern>
                        <shadedPattern>my_guava.common</shadedPattern>
                    </relocation>
                </relocations>
                <filters>
                    <filter>
                        <artifact>*:*</artifact>
                        <excludes>
                            <exclude>META-INF/maven/**</exclude>
                            <exclude>META-INF/*.SF</exclude>
                            <exclude>META-INF/*.DSA</exclude>
                            <exclude>META-INF/*.RSA</exclude>
                        </excludes>
                    </filter>
                </filters>
            </configuration>
        </execution>
    </executions>
</plugin>
```
## Spark Blacklist Mechanism
```
Blacklisting behavior can be configured via spark.blacklist.*.
```
The default value of `spark.blacklist.enabled` is `false`. If the value is set to `true`, Spark will no longer schedule tasks to the blacklisted executors. The blacklisting algorithm can be further configured by setting `spark.blacklist`.
  
*This article was written by Enqiu Jia and first published on the WeChat Official Account *只要5分钟不要996*. This article is translated and published with permission.

This is a best practice of using NebulaGraph Spark Connector by performing import jobs via Spark on Yarn distributed tasks and with CSV files on HDFS as the data source.

Offline CSV Import of Billions of Data | Best Practice Using NebulaGraph Spark Connector


![nebula-graph-index-explained-featured](https://user-images.githubusercontent.com/57335825/167098195-bde39149-3b4a-4d52-843d-0b56a2309a97.jpg)

The term Index in NebulaGraph is quite similar to the same term in relational databases, but they are not exactly the same. I noticed that some NebulaGraph users are often confused when getting started with NebulaGraph. Typically, people want to know what exactly NebulaGraph Index is, when should they use it, and how it impacts the performance of NebulaGraph.

Today I'm going to walk you through the Index concept in NebulaGraph and hopefully, this article will answer these questions.

Let's get started!

## What exactly NebulaGraph Index is

To put it short, NebulaGraph Index is only used when the vertextID is not specified, or only when **properties** of vertices or edges are defined in the query conditions. 

Index is only used in a starting entry of a graph query. If a query is in the pattern: (a->b->c, where c is with the condition "foobar"), since the only filter `condition-foobar` is on `c`, this query under the hood will start to look for `c`, and then it walks reversely through the `->` to `b`, and finally to `a`. Thus, the NebulaGraph Index will be used and only be possibly used when locating c.

### Index is used only to seek starting points

We know that in RDBMS, indexing is to create a duplicate of sorted data to enable QUERY with conditional filtering on the sorted data, in order to **accelerate the query in reads** and it also brings additional data writes.

> Note: in RDBMS/Tabular DB, indexing some columns means to create extra data that are sorted on those columns to make a query with those columns' conditions to be scanned faster, rather than scanning from the original table data sorted based on the key only.

In NebulaGraph, the index is to create a duplicate of sorted **Vertex/Edge PROP DATA** to **locate the starting point of a QUERY**.

Not all of queries relied on the index, here are some example queries, where the starting points are only defined using conditions, rather than VertextIDs. Let's call them `pure property condition starting queries`:

```cypher
#### Queries relying on NebulaGraph Index

# query 0 pure-property-condition-start query
LOOKUP ON tag1 WHERE col1 > 1 AND col2 == "foo" \
    YIELD tag1.col1 as col1, tag1.col3 as col3;

# query 1 pure-property-condition-start query
MATCH (v:player { name: 'Tim Duncan' })-->(v2:player) \
        RETURN v2.player.name AS Name;
```

In both `query 0` and `query 1`, the pattern is to "Find VID/EDGE only based on given property conditions". 

On the contrary, the starting point is VertexID based instead in `query 2` and `query 3`:

```cypher
#### Queries not based on NebulaGraph Index

# query 2, walk query starting from given vertex VID: "player100"

GO FROM "player100" OVER follow REVERSELY \
        YIELD src(edge) AS id | \
    GO FROM $-.id OVER serve \
        WHERE properties($^).age > 20 \
        YIELD properties($^).name AS FriendOf, properties($$).name AS Team;

# query 3, walk query starting from given vertex VID: "player101" or "player102"

MATCH (v:player { name: 'Tim Duncan' })--(v2) \
        WHERE id(v2) IN ["player101", "player102"] \
        RETURN v2.player.name AS Name;
```

If we look into `query 1` and `query 3`, which shared the same condition on vertices on `tag:player`, which are both `{ name: 'Tim Duncan' }`, they are differentiated in starting points:

For `query 3`, the index is not required as the query will start from the known vertex ID in `["player101", "player102"]` and thus:
- It'll directly fetch vertex data from `v2`'s vertex IDs
- then to GetNeighbors(): walk through edges of `v2`, GetVertices() for the next hop: `v` and filter based on the property: `name`

For `query 1`, the query has to start from `v` due to no known vertex IDs were provided:
- It'll do IndexScan() first to find all vertices only with the property condition of `{ name: 'Tim Duncan' }`
- Then, GetNeighbors(): walk through edges of `v`, GetVertices() for the next hop: `v2` 

Now, we know the whole point that matters here is on **whether we know the vertexID**. And the above differences could be shown in their execution plans with PROFILE or EXPLAIN like the following:

| `query 1`, requires index(on tag: player), pure property condition query as starting point | `query 3`, no index required, query starting from known vertex IDs |
| ------------------------------------------------------------ | ------------------------------------------------------------ |
| ![query-based-on-index](https://user-images.githubusercontent.com/57335825/167237932-effed472-6390-460c-89b5-c759644dc674.png)         | ![query-requires-no-index](https://user-images.githubusercontent.com/57335825/167237957-14981b0b-61c2-4ea9-ab60-02ee49c4843b.png)   |

### Why NebulaGraph index is an enabler rather than an accelerator

Can't those queries be done without indexes?

It's possible in theory with a full scan, but disabled without an index.

The reason is that NebulaGraph stores data in a distributed and graph-oriented way, the full scan of data was considered too expensive to be allowed.

> Note: from v3.0, it's possible to do TopN Scan without index, where the `LIMIT <n>` is used, this is different from the full scan case(index is a must), which will be explained later.
>
> ```cypher
> MATCH (v:player { name: 'Tim Duncan' })-->(v2:player) \
>         RETURN v2.player.name AS Name LIMIT 3;
> ```

### Why starting point only

Index data is not used in the traversal. It could confuse us to think of index as sorting data based on properties, does it accelerate the traversal with property condition filtering? The answer is no.

In NebulaGraph, the data is structured in a way to enable fast graph-traversal, which is already indexed/sorted on vertex ID(for both vertex and edge) in the raw data, where traversal(underlying in storage, it's calling GetNeighbors interface) of the given vertex is cheap and fast due to the persistent storage.

So in summary:

> NebulaGraph Index is sorted property data to find the starting vertex or edge on given pure property conditions.

## Facts on NebulaGraph Index

To understand more details/limitations/cost of Nebula, let's reveal more about its design. Here are some facts in short:

- Index Data is stored and shared together with Vertex Data

- It's **Left Match** based only: It's RocksDB Prefix Scan under the hood

- Effect on write and read path(to see its cost):

  - Write Path: Extra Data written + Extra Read request introduced
  - Read Path: RBO(Rule-based optimization), Fan Out(to all shards)

- Data Full Scan LIMIT Sample(not full scan) is supported without index
  - `LOOKUP ON t YIELD t.name | LIMIT 1`

  - ```cypher
    MATCH (v:player { name: 'Tim Duncan' })-->(v2:player) \
            RETURN v2.player.name AS Name LIMIT 3;
    ```

The key info can be seen in one of my [sketch notes](https://www.siwei.io/en/sketch-notes/):

![nebula-graph-index-sketch](https://user-images.githubusercontent.com/57335825/167237972-507c1344-636e-48f5-9b3f-fc8ba5dbaa69.png)

> We should notice that only the left match is supported in pure-property-condition-start queries. For queries like wildcard or regular expression, Full-text Index/Search is to be used, where an external elastic search is integrated with Nebula: please check [NebulaGraph Full text index](https://docs.nebula-graph.io/3.1.0/4.deployment-and-installation/6.deploy-text-based-index/2.deploy-es/) for more.

Within this sketch note, more highlights are:

- It's a Local Index Design
  - The index is stored and shared locally together with the graph data.
  - It's sorting based on property value, and the index search is underlying a RocksDB prefix scan, that's why only left match is supported.

- There are costs in the writing path
  - The index enables the RDBMS-like property Condition Based Query with cost in the write path including not only the extra write but also, random read, to ensure the data consistency.
  - Index Data writing is done in a synchronous way

- For the reading path:
  - In pure-property-condition-start queries, in GraphD, the index will be selected with Rule-based-optimization like this example, where, in a rule, the col2 to be sorted first is considered optimal with the condition: col2 equals 'foo'.
  - After the index was chosen, the index-scan request will be fanout to storageD instances, and in the case of filters like LIMIT N, it will be pushed down to the storage side to reduce data payload.
    - Note: it was not shown in the sketch but actually from v3.0, the nebula graph allows LIMIT N Sample Property condition query like this w/o index, which is underlying pushing down the LIMIT filter to storage side.


Takeaways:

- Use index only when we have to, as it's costly in write cases and if the limit N sample is the only needed case and it's fast enough, we can use that instead.
- Index is left to match
  - composite index order matters, and should be created carefully.
  - for full-text search use case, use [full-text index](https://docs.nebula-graph.io/3.1.0/4.deployment-and-installation/6.deploy-text-based-index/2.deploy-es/) instead.

## How to use the index

We should always refer to the [documentation](https://docs.nebula-graph.io/3.1.0/3.ngql-guide/14.native-index-statements/), and I just put some highlights on this here:

- To create an index on a tag or edge type to specify a list of props in the order that we need.

  - `CREATE INDEX`

- If an index was created after existing data was inserted, we need to trigger an index asynchronously to rebuild the job, as the index data will be written in a synchronous way only when the index is created.

  - `REBUILD INDEX`

- We can see the index status after `REBUILD INDEX` is issued.

  - `SHOW INDEX STATUS`

- Queries levering index could be LOOKUP, and with the pipeline, in most cases, we will do follow-up graph-walk queries like:

  ```sql
  LOOKUP ON player \
    WHERE player.name == "Kobe Bryant"\
    YIELD id(vertex) AS VertexID, properties(vertex).name AS name |\
    GO FROM $-.VertexID OVER serve \
    YIELD $-.name, properties(edge).start_year, properties(edge).end_year, properties($$).name;
  ```

- Or in MATCH query like this, under the hood, v will be searched on index and v2 will be walked by default graph data structure without involving index.

  ```cypher
  MATCH (v:player{name:"Tim Duncan"})-->(v2:player) \
    RETURN v2.player.name AS Name;
  ```

## Recap

Finally, Let's Recap

- Index is to sort Property DATA to enable finding starting point on a given Pure Property Condition(no VertexID provided)
- Index is **not** for traversal
- Index is left match, **not** for full-text search
- Index has cost on WRITE
- Remember to REBUILD after `CREATE INDEX` on existing data

Happy Graphing!

## About the author
Wey Gu is the Developer Advocate of NebulaGraph. He is passionate about spreading the graph technology to the developer community and trying his best to make distributed graph database more accessible. Follow him on [Twitter](https://twitter.com/wey_gu) or visit his [blog](https://www.siwei.io/en/) for more fun stuff.


This article explains what NebulaGraph's native index is, and when and how to use index.

Everything you need to know about NebulaGraph Index

In this article, I will try to walk you through four different ways that you can make NebulaGraph and Apache Spark work together: Spark Connector, Nebula Exchange, Nebula Algorithm, and PySpark.

4 different ways to work with NebulaGraph in Apache Spark


![Introducing TOSS: Achieving Eventual Consistency of Edges in NebulaGraph](https://user-images.githubusercontent.com/90186547/141244981-99cea678-95fe-4929-b764-ce61e3b8520c.jpg)

NebulaGraph has just released v.2.6. In this version, TOSS is certainly one of the important features. Here a detailed explanation about TOSS will be given.

# Let's start from a GO statement

As we all know, there are two types of edges, directed and undirected edges. When traversing directed edges, you can traverse forward or reversely. NebulaGraph also supports this semantics. For example:

```golang
go from "101" over known reversely yield known.kdate, id($$);
```

The above statement starts from Vertex 101 to find all the corresponding neighbors reversely. However, when you insert an edge into NebulaGraph, the command will be like:

```golang
insert edge known(degree) VALUES "100" -> "101":(299792458);
```

Seemingly, the above statement only specifies the outgoing edge. This is because NebulaGraph will specify the incoming edge in the background when you insert an edge.

# How to insert an edge into NebulaGraph

Take the `INSERT` statement above as an example, the background execution process contains the following:

* Nebula Console sends the INSERT request to the NebulaGraph server.

* After the NebulaGraph server receives the request, it adds an incoming edge for each outgoing edge and sends `AddEdgeRequest` to their hosts respectively.

* After the host (Nebula Storage server) receives `AddEdgeRequest`, it inserts the edge locally (via the Raft protocol) and returns the result to the NebulaGraph server.

* The NebulaGraph server then returns the results from both hosts to the Nebula Console for querying purpose.

The flow diagram is as follows:

![How to insert an edge into NebulaGraph](https://user-images.githubusercontent.com/90186547/141245971-9d62e48b-7646-4d9c-a929-d8ac7d45c737.png)

If you are familiar with network/distributed programming, you may see the problem now. The graph service uses RPC to call both storage services. When the INSERT operation is executed enough times, one RPC succeeds while the other fails due to timeout. In other words, an `INSERT` operation may succeed on the outgoing edge while fail on the incoming-edge.

If now a user requires consistent property settings for both outgoing edge and incoming edge, the client has to retry the query infinitely. It is inappropriate for NebulaGraph, s a database product, to rely on the client for data atomicity.

A requirement thus comes into being, that is, to ensure the atomicity of outgoing edge and incoming edge. This means that the outgoing edge and incoming edge should be updated either successfully or they should fail at the same time. And TOSS (Transaction On Storage Side) is the feature to ensure the eventual consistency of edges upon `INSERT`, `UPDATE`, and `UPSERT` operations.

# How to use TOSS

With the release of NebulaGraph v2.6.0, the TOSS function has also been launched. The feature is set to `Disabled` by default due to performance and stability considerations. You can find the `enable_experimental_feature` option in the NebulaGraph server configuration file and set it to `True`. Then you need to restart the graphd service for the feature to take effect. The command is as follows:

```
--enable_experimental_feature=true
```

Then the operations `CREATE SPACE` / `CREATE EDGE` / `INSERT` / `UPDATE` will achieve eventual consistency of edges in NebulaGraph. (Just execute the operations as before)

Note: The TOSS feature will be only applied to incremental data after it is enabled. 


More information about v2.6, please refer to [NebulaGraph Database Manual](https://docs.nebula-graph.io/3.0.1/pdf/NebulaGraph-EN.pdf). It records in detail the knowledge points and specific usage of the graph database and the graph database NebulaGraph.

Welcome to join our [Slack channel](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) if you want to discuss with the rest of the NebulaGraph community!

NebulaGraph has just released v2.6. In this version, TOSS is certainly one of the important features. Here a detailed explanation about TOSS will be given.

Introducing TOSS: Achieving Eventual Consistency of Edges in NebulaGraph


![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

NebulaGraph 2.0 supports full-text indexing by using an external full-text search engine. To understand this new feature, let's review the architecture and storage model of NebulaGraph 2.0.

Full-Text Indexing in NebulaGraph 2.0


![storage-format-in-nebula-graph-2.0](https://user-images.githubusercontent.com/57335825/118632761-5119f380-b803-11eb-8ab4-ca4e8a7e7bf6.png)

NebulaGraph 2.0 has changed a lot over its releases. In the storage architecture design, the encoding format has been changed, which has the most significant impact on its users. In NebulaGraph, data is stored as KV-pairs in RocksDB. This article covers several issues such as the differences between the old and new encoding formats and why the format must be changed.

# Encoding Format in NebulaGraph 1.0

Let's start with a brief review of the encoding format in NebulaGraph 1.0. For those who are not familiar with it, I recommend that they read this post: [An Introduction to NebulaGraph's Storage Engine](https://nebula-graph.io/posts/nebula-graph-storage-engine-overview/). In NebulaGraph 1.0, the vertex IDs can only be represented by values of the `int` type, so all VertexIDs are stored in int64.

* Vertex key format in NebulaGraph 1.0

![Vertex key format in NebulaGraph 1.0](https://user-images.githubusercontent.com/57335825/118632935-8292bf00-b803-11eb-8f97-08f1207971fe.png)

* Edge key format in NebulaGraph 1.0

![Edge key format in NebulaGraph 1.0](https://user-images.githubusercontent.com/57335825/118633021-9a6a4300-b803-11eb-9d33-3f9743938b84.png)

For a VertexID, its corresponding PartID is obtained after hash, so a vertex and all its incoming and outgoing edges, of which only the IDs of the sourcing vertices are  hash processed, are mapped in the same partition. **It should be noted that in NebulaGraph 1.0, the first bytes of a vertex and an edge are of the same Type**. That is, for a vertex, all its tags are not stored physically contiguously. They may be stored as shown in the following figure. For the vertex named `src`, its three tags (tag1, tag2, and  tag3) may actually be separated by edges.

![rocksdb key format](https://user-images.githubusercontent.com/57335825/118633166-c38ad380-b803-11eb-8fb6-bc8d7f83a95c.png)

Such a key format can meet most requirements for operations on data in NebulaGraph 1.0. For example, when the prefix is specified, data can be retrieved by the `FETCH` and the `GO` manipulations.

# Encoding Format in NebulaGraph 2.0

In the releases prior to the 2.0 GA release, the encoding format in the underlying storage layer was actually the same as that in NebulaGraph 1.0. For an integer VertexID, it is stored in exactly the same format as in 1.0. However, for a string VertexID, it is changed from int64, which occupies 8 bytes, to `FIXED_STRING` of a fixed length which needs to be specified by users when they run `CREATE SPACE`. If a VertexID is shorter than the specified length, the system will automatically pad it with `\0`. For a VertexID that exceeds the specified length, the system will report an error directly.

In the 2.0 GA release, several changes were made to the underlying storage encoding format. Therefore, to upgrade NebulaGraph to this release, a specific tool is necessary. This tool converts the data from the old format to the new format. The following figures show the key formats implemented in the 2.0 GA release.

## Key Formats in NebulaGraph 2.0

* Vertex key format

![Vertex key format](https://user-images.githubusercontent.com/57335825/118633434-0fd61380-b804-11eb-98fb-93e80c0fbc5c.png)

* Edge key format

![Edge key format](https://user-images.githubusercontent.com/57335825/118633579-35631d00-b804-11eb-835e-32928e22e179.png)

## Key Format Comparison between NebulaGraph 1.0 and 2.0

![Key Format Comparison - Vertex](https://user-images.githubusercontent.com/57335825/118633663-47dd5680-b804-11eb-944a-83ecb909c154.png)

![Key Format Comparison - Edge](https://user-images.githubusercontent.com/57335825/118633757-63486180-b804-11eb-9b9f-2534d76be91f.png)

As shown in the preceding figures, some major changes are made:

1. As mentioned earlier, the length of VertexID is changed from 8 bytes to n bytes. For an integer VertexID, n takes the value of 8. For a string VertexID, n takes the value of the specified length.
2. In the vertex key format, the timestamp byte is removed, and in the edge key format, the timestamp byte is replaced with a one-byte placeholder.
3. For the key format of a vertex and an edge, **the Type byte differs from each other**, which separates vertices and edges physically. 

Why these changes are made? Here are the main reasons: 

1. The changes made to VertexID are mainly to make it support string IDs while still being compatible with int IDs in NebulaGraph 1.0. In the Storage layer, VertexIDs are encoded as bytes. However, VertexIDs of the corresponding type, which is determined by the configuration of the graph space, are returned in the results. 
    > Why FIXED_STRING is applied to the string IDs? If a fixed length is not specified, scanning by prefixes is not possible. By padding the length with \0, the prefixes of all vertices and edges are of the same length, so that the corresponding prefix query can be performed.
2. Removing the timestamp byte is mainly to improve the performance, because saving data as multiple versions lowers the performance and currently the MVCC implementation in NebulaGraph is of a low priority.
    > A one-byte long placeholder is reserved in the edge key format, which is purposely designed for TOSS (Transaction on Storage Side). It is mainly used to identify the outgoing edge and the incoming edge of an edge. We will write another article to introduce the TOSS feature in detail. Please stay tuned for our updates.
3. The main benefit of separating vertices and edges physically is that all tags of a vertex can be easily and quickly retrieved, which is extensively used in Cypher's `MATCH` statements. Before this change, data is scanned by a prefix composed of the same `Type + VertexID`, which may greatly reduce the performance because vertices and edges may be mixed together. After different Types are implemented, scanning by the VertexType + VertexID prefix can retrieve all tags quickly.
    > In NebulaGraph 1.0, vertices and edges can be stored with the same prefix because there is no need to fetch all tags for a vertex. At the code level, however, such an implementation has great effect on the performance. For example, the FETCH operation in NebulaGraph 1.0 works by scanning by the VertexID prefix, but for a super vertex, retrieving all its tags with such an operation performs poorly. Besides, if the SCAN operation provided by the Storage layer is used to retrieve all the vertices of an entire graph, the entire RocksDB is actually scanned.

In addition to the changes made to the key format of vertices and edges, the format of the indexes has actually changed. 

Why these changes? The first reason is that with `NULL` support in NebulaGraph 2.0, the index semantics must be changed accordingly. The second one is that in NebulaGraph 1.0, the string fields in indexes are actually treated as strings of variable length. Therefore, whenever an index on a string field is used in a `LOOKUP` statement, only an equivalent query is used. In NebulaGraph 2.0, however, `FIXED_STRING` is applied to the string fields for indexes and the data VertexID, and for the `LOOKUP` statements, range scanning is supported for the string type index. For example, `LOOKUP ON index1 WHERE col > "aaa"`. We will introduce this design in details in the future articles.

Interested in this topic? Join the NebulaGraph [Slack channel](https://join.slack.com/t/nebulagraph/shared_invite/zt-7ybejuqa-NCZBroh~PCh66d9kOQj45g) and talk to the community.


This article covers several issues such as the differences between the old and new encoding formats and why the format must be changed.