{'fraction_non_alphanumeric': 0.0372407641418271, 'fraction_numerical': 0.011057013305582397, 'mean_word_length': 4.6981152054214315, 'pattern_counts': {'":': 0, '<': 0, '<?xml version=': 0, '>': 1, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
G-Tran: Making Distributed Graph Transactions Fast

Graph transaction processing raises many unique challenges such as random data access due to the irregularity of graph structures, low throughput and high abort rate due to the relatively large read/write sets in graph transactions. To address these challenges, we present G-Tran -- an RDMA-enabled distributed in-memory graph database with serializable and snapshot isolation support. First, we propose a graph-native data store to achieve good data locality and fast data access for transactional updates and queries. Second, G-Tran adopts a fully decentralized architecture that leverages RDMA to process distributed transactions with the MPP model, which can achieve high performance by utilizing all computing resources. In addition, we propose a new MV-OCC implementation with two optimizations to address the issue of large read/write sets in graph transactions. Extensive experiments show that G-Tran achieves competitive performance compared with other popular graph databases on benchmark workloads.


INTRODUCTION
Graph data are abundant today in both industrial applications and academic research. In order to support efficient graph data storage and management, graph databases have become an essential infrastructure. However, most existing graph databases [1-4, 6, 10, 23,29,33,41] have shortcomings in their design and functionality that lead to performance bottlenecks in processing complex graph transactions. For example, JanusGraph [2], OrientDB [6] and ArangoDB [4] use BigTable [13] or Multi-Model storage to represent graphs, leading to significant read/write amplifications during graph queries and updates. Neo4j [3] and TigerGraph [23] call themselves native graph stores, as to distinguish from the nonnative stores (e.g., relational, columnar, key-value stores), and represent graphs as adjacency-lists or compressed formats (e.g., CSR, CSC). These data layouts are more efficient for graph queries but suffer from poor data locality when updates are frequent, thus incurring extra overheads on data retrieval and impairing both latency and throughput ( §3).
In addition to handling graph-specific read/write, transaction isolation is another important functionality in DBMSs, which controls how transaction integrity is visible to concurrent users in order to maintain the correctness of transaction processing. Transaction isolation (ideally strict serializability) is challenging when dealing with large-scale graph data for the following reasons. Read-only graph transactions tend to have large read sets since a large number of vertices and edges can be easily involved after just two to three hops of traversal starting from a vertex, as most real-world graphs exhibit a power-law degree distribution. Similarly, read-write transactions may also have large write sets (see Figure 6). These large read/write sets lead to high contention in concurrent transaction processing. As graph transactions have relatively long processing time, consequently the contention becomes more serious, which in turn leads to high abort rate and low throughput.
These unique challenges in graph transaction processing motivate us to design a new distributed graph database system, called G-Tran, for high-performance transaction processing on property graphs [7]. To the best of our knowledge, G-Tran is the first RDMA-enabled graph database that provides strong consistency, i.e., serializability (SR) and snapshot isolation (SI ), low latency and high throughput for graph transaction processing. We highlight some unique designs of G-Tran as follows: • We design a graph-native data store with efficient data and memory layouts, which offers good data locality and fast data access for read/write graph transactions under frequent updates. • We propose a fully decentralized system architecture by leveraging the benefits of RDMA to avoid the bottleneck from centralized transaction coordinating, and each worker executes distributed transactions under the MPP (i.e., massive parallel processing) model. • G-Tran presents a multi-version-based optimistic concurrency control (MV-OCC) protocol, which is specifically designed to reduce the abort rate and CPU overheads in concurrent graph transaction processing. We demonstrate the effectiveness of our system designs and the overall performance by comparing G-Tran with the state-of-the-art graph databases [2-4, 23] using benchmark workloads [30,42]. The results show that G-Tran can achieve up to orders of magnitude improvements over the existing graph databases and obtain high throughput at both SR and SI isolation levels.

BACKGROUND
Property Graph. G-Tran adopts the property graph (PG) model to represent graph data because of the generality and expressiveness of PG. In a PG, vertices represent entities in an application and (directed) edges model the relationships between two entities. Both entities and relationships may have a set of properties to describe their attributes (e.g., names, gender, etc.) in the form of key-value pairs. Many existing graph databases, such as Titan [1], JanusGraph [2], OrientDB [6] and Neo4J [3], adopt PG to model their graph data, while using Gremlin [5] as the query language. Currently, G-Tran also uses Gremlin (the latest 3.0 standard) as its query language.
Concurrency Control Protocols. Concurrency control ensures atomicity and isolation for database transactions. G-Tran supports both serializability (SR) and snapshot isolation (SI). SR has the strictest constraint that all concurrent transactions should execute their operations logically as if they are executed in sequence. SI relaxes the constraint to require that only all reads in a transaction should see a consistent snapshot of the database. There are three kinds of concurrency control protocols that are widely used in databases to implement different isolation levels: two-phase locking (2PL) [32], optimistic concurrency control (OCC) [40], and multiple version concurrency control (MVCC) [49]. 2PL is the most common and simplest protocol, which uses locks to avoid conflicts among concurrent transactions. OCC does not use lock, but it avoids conflicts by validation after a transaction completes its execution. Generally, OCC handles transactions in three steps: process, validate, and commit/abort. In comparison, MVCC provides point-in-time consistent views for multiple transactions at the same time by maintaining multi-versions of each object with timestamps, which incurs a higher storage overhead.
InfiniBand and RDMA. InfiniBand has become quite commonly in use recently and led to the development of many new distributed systems [31,36,39,65]. InfiniBand offers two network communication stacks: IP over InfiniBand (IPoIB) and remote direct memory access (RDMA). IPoIB implements a TCP/IP stack over InfiniBand to allow current socket-based applications to be executable without modification. In contrast, RDMA provides a verbs API, which enables zero-copy data transmission through the RNIC without involving the OS. RDMA has two verb operations: two-sided send/recv verbs and one-sided read/write verbs. Two-sided verbs provide a socket-like message exchange mechanism, which still incurs CPU overheads on remote machines. One-sided verbs can directly bypass both the kernel and CPU of a remote machine to achieve low latency.

CHALLENGES AND DESIGN CHOICES
We have briefly discussed the characteristics of graph transaction processing and the limitations of existing graph databases in §1. In this section, we analyze the important factors that should be considered in the design of a high-performance distributed graph database. We summarize the challenges of distributed graph transaction processing as follows.
(C1) Graph data can easily result in poor locality for reads and writes in databases after continuous updates due to its irregularity [53].
(C2) Due to the power-law distribution on vertex degree and the small world phenomenon of most real-world graphs, the cost of traversal-based queries can be very high after multi-hops (e.g., ≥ 2) fan-out [51].
(C3) Graph traversal query can involve relatively larger read/write set and longer execution time compared with those queries on other storage models (e.g., KVS) due to the connectedness of graph data, which may lead to higher contention and lower throughput in transactions.
(C4) Latency and scalability is another critical issue for distributed graph transactions. The network bandwidth, contention likelihood, and CPU overheads are the main bottlenecks [64]. Traditional centralized system architecture (i.e., assigning a master node as the global coordinator) may limit the scalability for OLTP.
We conduct experiments in §6 to verify the above issues. Here, we discuss the design choices to address the aforementioned challenges.
We first consider (C4). A significant overhead for distributed transaction processing is the large number of roundtrip messages that are needed to ensure ACID. In recent years, many systems [9,20,27,37,46,52,57,64] have been proposed to improve the performance of distributed transaction processing by leveraging RDMA as it can remove the overheads on network and CPU, although these works mainly focus on RDBMS or KVS instead of graph databases. Using RDMA for graph transaction processing can also improve the performance in terms of throughput, latency and scalability. However, the design of an RDMA-based distributed inmemory database is non-trivial. Specifically we move from a pure shared-nothing or shared memory architecture to a hybrid memory layout (described as below) via RDMA one-sided read/write and two-sided send/recv. These new features require us to have an overall system redesign from the storage layer to the transaction processing layer in order to tightly integrate each system component.  To address (C1) and (C2), we propose a new storage layout for property graphs to achieve both good data locality and efficient data access. It stores graph data under a graph-native schema, but organizes the (arbitrary-length) adjacency-lists of vertices into fixed-size rows and separately maintains vertex/edge properties into a key-value store. The storage layer of G-Tran splits the memory space of each machine into two parts. One part follows the shared-nothing architecture to store one piece of graph partition locally, and the other part follows the shared-memory architecture to compose an RDMA-based distributed memory space for remote property data access. Such design is based on the facts that: (1) to implement efficient RDMA-based data structures (e.g., map, tree) are non-trivial, as the address space of each object should be recorded explicitly and this requires more memory footprint [69]; (2) not all data are necessary to be shared and to manage different regions of data in different memory spaces compactly can further improve the data access efficiency for local search and scan ( §4.2).
For (C3), 2PL would incur high overhead due to locking on large portions of data. OCC is a good choice for readintensive workloads, but we can further reduce the overhead of isolation maintenance in concurrent transactions by integrating OCC with multi-versioning. We adopt a multi-version OCC (MVOCC) protocol to coordinate concurrent transaction processing. MV-OCC has the following advantages. First, a multi-versioning mechanism allows read-only transactions to access old versions of objects without imposing any consistency overhead on read-write transactions. Second, OCC applies validation to check if there is any conflicting updates without locks. We propose our own optimistic optimizations in the MVOCC protocol to reduce the abort rate ( §4.3).
Moreover, G-Tran adopts an MPP-enabled decentralized architecture to process transaction execution ( §4.4). Existing distributed databases [27,29] follow a centralized approach to process tasks such as transactional metadata management and timestamp ordering on a master node. The master is likely a bottleneck of the entire cluster and affects the overall performance [64]. Instead, in our decentralized architecture, all servers share the load of each transaction according to the locality of its read/write sets and each server can handle multiple sub-queries simultaneously with multi-threads (i.e., MPP) for speed-up. In addition, RDMA can significantly reduce the costs of essential operations such as clock synchronization, timestamp ordering, transaction status synchronization in the decentralized architecture, which also helps improve the scalability. Figure 1 shows the architecture of G-Tran. G-Tran supports multiple client connections to server nodes through a regular Ethernet network. Each client provides a user console and a light communication lib for submitting transactions and receiving query results. A cluster monitor monitors the load of each server by a heartbeat mechanism and conducts the assignment of incoming transactions from clients for load balancing among servers. G-Tran server nodes are connected with each other via InfiniBand for RDMA communication. Each server node is composed of two layers, i.e., the storage layer and the OLTP layer.

SYSTEM DESIGN 4.1 Overview
The storage layer, i.e., the data store, keeps a property graph in two parts: topology data and property data. Topology data refer to the graph topology, i.e., vertices and edges in the format of adjacency lists. Property data are the keys and values of the vertex/edge properties. We partition a graph into shards over server nodes. G-Tran constructs a consistent global address space over all the nodes in a cluster, so that the location of any object in the data store can be retrieved by its ID. The data store has a multi-version schema to support consistent view for concurrent transactions ( §4.2).  The OLTP layer of each server node sets a group of worker threads to process incoming transactions. Worker threads interact with the data store to process data reads/writes. To allow remote data access and fast communication via RDMA, each server node registers a chunk of memory at NIC during the initialization stage, which divides the memory space of a server node into two regions, RDMA allocated memory and local memory, as depicted in Figure 1. We place different system components in different memory regions according to their functionalities, in order to enjoy the benefits of both local memory management (i.e., efficient maintenance and access of data structures such as tree, map, lock, etc.) and RDMA (i.e., low CPU overhead, fast remote data access and atomicity guarantee) for concurrent transaction processing.
G-Tran adopts MPP model [23] in the OLTP layer ( §4.4). That is, a cross-server transaction is executed simultaneously in the places where its data (i.e., the read/write sets) are located and each transactional operation can be processed in parallel based on its load. The MPP model leads to better data locality and enables parallel execution of a single transaction. As a trade-off, parallel execution over multiple servers requires more message passing for consistency control, but we keep the network communication overhead low by using RDMA. We construct an RDMA-enabled mailbox (i.e., the green box in Figure 1) to process point-to-point message sending/receiving at thread-level, including one-sided RDMA read/write and two-sided RDMA send/recv. Besides the mailbox, in the RDMA allocated memory, G-Tran also maintains some system components (e.g., timestamp bufs, transaction status table) for global transactional metadata access, which is needed in our MV-OCC protocol ( §4.3). However, the transactions' private metadata (e.g., begin time, commit time, read/write set) and other structures (e.g., indexes) that do not need to be shared with other servers, are stored in the local memory.

Data Store
As shown in Figure 2, the data store in each server is composed of three components: Vertex Table (in red color), Edge Table (in blue color), and Multi-Version Key-Value Store (MV-KVS, in orange).
All vertex objects are stored in the Vertex Table. Every vertex object has a fixed-size vertex header to record the information (e.g., ID and label) of a vertex. The begin time and end time in the vertex header indicate the visible time period of the vertex, i.e., the period that the vertex is accessible to all active transactions. Usually, the begin/end time of a vertex is exactly the commit time of the transaction that creates/deletes this vertex. In addition, the vertex header also links to two row-lists, which record the connected edges (i.e., Edge RowList *) and properties (i.e., VP RowList *) of the vertex, respectively. To store the adjacency list of a vertex, we use an Edge Cell (in green) to represent each adjacent vertex, and then arrange all the Edge Cells in an adjacency list into rows in ascending order of the vertex IDs. Note that each row has a fixed number of cells. If one row is filled up, a new row will be allocated from the memory pool until the entire adjacency list is stored.
Each edge cell is mapped to one edge object stored in the Edge Table. Since an edge =( 1 , 2 ) connects two vertices, there are two edge cells (of 1 and 2 ) pointing to the same edge object of . If is directed, then the edge cell of 1 / 2 also keeps a direction sign to indicate that is an out/in-edge of 1 / 2 . Each edge object also has an edge header, which records its ID, label, begin time, end time, and a link to the row-list, i.e., EP RowList *, that stores the properties of the edge. Both VP RowList and EP RowList have the same layout as that of Edge RowList. Each VP/EP Cell in the VP/EP RowList records the ID of a property object and a pointer that links to its multi-version property values in the MV-KVS.
The MV-KVS is divided into two regions as shown in Figure 2: MV-tuple region and value region. The MV-tuple region stores a set of pre-allocated, fixed-size MV-tuples, where each MV-tuple records the begin and end time of a version of the corresponding property. All MV-tuples (i.e., the different versions) of a property object are ordered by their begin/end time for version searching during data access. Each MV-tuple keeps a pointer that links to where the value of this version is stored in the value region. We will discuss  Figure 3: The MV-OCC protocol of G-Tran how to operate on the multi-versions of a property object in §4.3.
Design Principle. The key design of our data store is that all components have fixed sizes (except property values with variable lengths) and are aligned compactly in memory space wherever possible. This provides good data locality and is critical for efficient read and write (e.g., insert a new edge). As dynamic memory allocation is a costly operation, G-Tran uses memory pooling to pre-allocate memory buffers for all type of components (i.e., headers, cells, row-lists, tuples). This data layout enables that graph traversals will be executed as a combination of row-based scanning and object-based filtering (on MV-KVS) via zero-CPU-overhead one-sided RDMA read/write. In addition, it also benefits the multi-threaded execution of concurrent transactions ( §4.4).

The MV-OCC Protocol
To guarantee a transaction is serializable, we should hold two features in a multi-version storage.
(1) Read stability: the visible version of a record for should keep unchanged during the processing of . Traditionally, this can be implemented by having a read lock on the record or by validating if there is no newly committed version of the record before commits.
(2) Phantom avoidance: the read set of should keep unchanged during the processing of . Usually, this can be implemented by having an index/row-level lock on the entire read set or by re-scanning the read set in a validation phase.
However, to ensure SR by locks or re-scanning has a high cost, especially in a distributed setting. To avoid locks and re-scanning, we propose our own MV-OCC design with specific optimizations to maintain isolation. Our MV-OCC follows a general procedure of OCC but combines with our multi-version-based commit/abort rules. Before we discuss the details, we first define the basic components and some necessary concepts. Table (TST) is a distributed table maintained in the RDMA-allocated memory in each server node. TST records the real-time status (i.e., execution, validation, commit, abort) of each active transaction . Once starts the processing of a new phase, its worker thread will update 's status in TST through RDMA atomic write. When another worker thread wants to check 's status for its own transaction processing, it can apply one-sided RDMA read to fetch the value with microseconds latency. Table (RCT) is another distributed table but stored in the local memory of server nodes. RCT records the metadata of all active read-write transactions such as their IDs, commit time and write sets. RCT is indexed by a B-Tree using the commit time as the key. RCT will be used in the validation phase of each read-write transaction to avoid the high cost of read set re-scanning.

Recent Committed
Multi-Versioning Read/Write Rules. Figure 3 shows the workflow of a transaction in G-Tran. When a transaction reads a property object, it finds the visible version of the property value, i.e., the MV-tuple whose begin time (BT ) and end time (ET ) overlap with the BT of . If wants to update the property object, it creates a new version, i.e., a new MVtuple, and inserts its transaction ID (i.e., ) in the ET field of the current version and the BT field of the new version, as shown by the two MV-tuples in the Execution Phase in Figure 3. This indicates that the current property object is in the process of being updated, where the current version has been "locked" and the new version has not been committed yet. Then, if commits successfully later on, these two fields will be replaced by the commit time (CT ) of , i.e., ( ) in the Commit/Abort Phase in Figure 3. This indicates that the current version has ended at ( ) and a new version beginning at ( ) is created. But if aborts, then the BT field of the new version will be set as -Inf to indicate that this version has become invisible forever. All the old versions whose ET is before the earliest BT of all active transactions, as well as the versions with -Inf BT, will be cleaned and recycled back to the memory pool by garbage collection ( §5).
We now describe the details of the specific three phases of the transaction workflow in Figure 3.
In the execution phase, a transaction first obtains its BT, i.e., the timestamp when its processing starts, for version visibility checking. Then, is executed and its read/write set is constructed by accessing the Vertex/Edge Tables and the MV-KVS, following the multi-versioning read/write rules. We propose an optimistic pre-read mechanism in the execution phase to reduce the abort rate. We illustrate the idea by the example given in the Execution Phase in Figure 3, where a transaction is doing the read-scanning on MV-tuples. Assume that 's BT > 10. Then, will read the version (let it be V1) whose BT is 10 and ET is (meaning that another transaction is in the process of updating the corresponding property object). In this case, instead of aborting directly, optimistic pre-read assumes that will commit successfully later and executes according to the status of as follows. There are four possible cases: • 's status is execution: the version (let it be V2) next to V1 is a new version created by and V2 has not been committed yet. In this case, reads V1 and validates the read-set consistency (i.e., a new version is indeed not committed) in its validation phase.
• 's status is validation: we optimistically assume that will commit, and thus directly pre-reads V2 now but validates the commit dependency (i.e., if is indeed committed) in 's validation phase. Note that reading V1 causes read instability and leads to abort.
• 's status is commit: directly reads V2 as has committed and the CT of is definitely earlier than the CT of .
• 's status is abort: ignores the new version V2, and reads the current version V1. In the case of is a read-write transaction, if sees in the visible MV-tuples when is inserting a new version, it should abort itself immediately (except = ), since this case belongs to a write-write conflict.
In the validation phase, a read-write transaction first obtains its CT, which is the timestamp when the validation begins (i.e., when the transaction logically commits). Then, checks read stability and phantom avoidance through conflict checking. Specifically, based on the BT and CT of , i.e., ( ) and ( ), we obtain from RCT all the read-write transactions, W-Trxs, whose commit time falls in the period [ ( ), ( )] and their write sets may change the read set of . If there is no overlapping element between the read set of and the write sets of W-Trxs, then can commit successfully. Otherwise, should abort. The above conflict checking is executed as two set comparison on all server nodes simultaneously in a MPP manner ( §4.4). We also propose an optimistic validation strategy to improve the success rate of commit. During the validation of , if we find that is in conflict with another transaction , where is in the validation phase too, we do not abort immediately. Instead, we optimistically assume that will abort later and continue the validation process of after recording such a dependency between and (i.e., should commit only if does abort). At the end of the validation phase of , we perform status checking for all the dependent transactions (if any). can commit itself only if all its dependent transactions due to optimistic pre-read have committed and all its dependent transactions due to optimistic validation have aborted. Note that if is a read-only transaction, its validation phase only needs to check those dependent transactions due to optimistic pre-read because has no write set.
In the commit/abort phase, a read/write transaction physically effects its write set if commits, or discards if aborts, where the corresponding MV-tuples are updated according to the multi-versioning read/write rules.
Design Principle. Following MV-OCC, there is no coarsegrained read/write locks or re-scanning during the entire transaction processing, which improves both the throughput and latency of transaction processing. The rationality of adopting the two optimistic optimizations in the execution and validation phases is that graph transaction workloads are mostly read-heavy as mentioned in §3. Moreover, to achieve snapshot isolation, MV-OCC can skip optimistic pre-read in the execution phase and the entire validation phase, and performs the regular commit (or abort when write-write conflict happens).

Distributed Transaction Processing with MPP
We now describe how G-Tran leverages RDMA to process distributed transactions with the MV-OCC protocol.
With the use of RDMA, the design goal of G-Tran's execution model is to effectively parallelize the transaction processing across servers while removing the computation bottlenecks (e.g., centralized coordinator, stragglers, locks). Existing RDMA-enabled databases [20,57,64] usually apply RDMA atomic verbs, compare-and-swap (CAS) and fetchand-add (FAA), to lock and fetch records from remote machines to a local machine, and then perform the local transactional updates before writing them back to remote machines. However, such an approach is not efficient for graph transactions due to their large read/write sets. It is expensive to apply CAS & FAA and locks on large amounts of data at a time, as this incurs high CPU overheads and impairs transaction throughput. Instead of fetching data remotely as [20,57] did, G-Tran processes transactions on the servers where the data locates at and scales out the transaction processing across the servers through message passing via one-sided RDMA write. This approach avoids extra CPU overhead incurred by remote locking and RDMA-based CAS & FAA by shifting the overhead to network, which is small by using Infiniband. G-Tran's decentralized architecture is shown as Figure 4, where all servers have the same layout and play the same roles in the cluster. Without loss of generality, we only discuss the processing of one single transaction in one server. When a server receives (assigned by the cluster monitor), this server becomes the unique coordinator of and handles the transactional metadata/result management of , i.e., to aggregate intermediate/final results of if needed (e.g., for operators COUNT, MAX, etc.) and to update 's status in the local shard of TST. Both TST and RCT are read/write-enabled, shared data structures via RDMA. In order to maintain the atomicity and consistency of TST/RCT read/write, we adopt two different strategies according to their specific workloads.
Each update in TST is executed by a local atomic write on the coordinator server in order to avoid unsafe one-sided RDMA read from other server nodes. Applying atomic operation here is acceptable as a transaction status occupies only 2 bits. However, inserting an entire write set into RCT is a much bigger operation and implementing it on the RDMA allocated memory via RDMA-based CAS & FAA for read/write consistency incurs high CPU overhead and thread contention. Thus, we maintain RCT in the local memory of the servers, while using two-sided RDMA send/recv to query and update the RCT entries in order to ensure its read/write consistency. In addition, we maintain a timestamp buf in the RDMA allocated memory to synchronize the earliest BT among the active transactions on all servers through one-sided RDMA read, so as to obtain the global earliest BT, which is used by RCT to garbage-collect the expired entries (i.e., transactions and their write sets).
The entire processing of a transaction is split into shards cross all servers and executed in parallel. First, the coordinator of sends initial commands to all servers via one-sided RDMA write to trigger the start of 's processing. But not every server can be activated for , as it depends on the actual data locality of the read set of . If a server indeed has one shard of , then it follows the MV-OCC protocol to process . According to the protocol, when needs to update its status, for example, a server has to abort due to a local conflict, it will synchronize 's updated status to every shard of TST on the servers via one-sided RDMA write. Thus, at the beginning of each operation in 's processing, the first step G-Tran has to do is status checking. If 's status has been changed, then we need to switch the processing of to the corresponding phase. In above example, will abort and terminate on all servers once they observe 's new status. Thus, TST works as a global flag for each active transaction, and all servers can fast access/update it through RDMA. In contrast, the read/write sets of constructed in the execution phase of MV-OCC are recorded in each server locally. And in the validation phase of , it first fetches all the potential conflicting transactions through RCT, and then does conflict checking on their read/write sets as described in §4.3. Finally, if can commit successfully, besides following the procedures of MV-OCC in §4.3, the final result will be aggregated from each server to 's coordinator and then sent to the client.
The MPP execution of each operator can speed up the query latency and improve the utilization of CPU resource. If one operator has a relatively higher load than the others, G-Tran will use multiple threads to process this operation in parallel in order to achieve a shorter completion time. We follow the state-of-the-art MPP implementation from Grasper [15] while also take into consideration the sideeffects of NUMA architecture [28,47]. As shown in Figure 4, each thread in G-Tran's worker thread pool is bound with one CPU core and further divided into several logical thread regions. This design can achieve better cache locality and memory locality on CPUs in cross-NUMA nodes. We omit the details here as this optimization is not the focus of our paper.

SYSTEM IMPLEMENTATION
G-Tran was implemented in C++, currently with 46K+ lines of code. To support RDMA communication, we developed a mailbox to achieve efficient RDMA one-sided read/write and two-sided send/recv based on RDMA ibverbs library. We also provide a general TCP-based version of the system, i.e., G-Tran without RDMA, which uses ZeroMQ TCP sockets to achieve point-to-point communication and MPI to coordinate the inter-process communication over Ethernet. Here we briefly discuss some of the system implementation details, while all code and details will be released on GitHub later.
Timestamp Ordering. Unified timestamp (together with MV-OCC) allows concurrent transactions to read a consistent snapshot of the database, while the timestamp order should be matched with the real time order. We follow the solution proposed by FaRMv2 [52] for global time synchronization using Marzullo's algorithm [45], where any server in the cluster can play the role of clock master and other servers periodically synchronize their clocks via RDMA writes.
Garbage Collection (GC). MVCC-based protocol usually has a high overhead on the storage. Thus, as an in-memory graph database, GC is critical for G-Tran to avoid the growth of memory consumption when servers run continuously. G-Tran's GC was implemented through one scanning thread and two GC threads as default, but users may configure the exact number of GC threads based on their workloads. However, recycling the memory slots occupied by obsolete objects requires write locks to guarantee data consistency for active read/write transactions, which leads to a degradation in transaction throughput and latency. In order to reduce the impact of GC, we separate the scanning process and execute GC jobs batch-by-batch. The scanning process collects the obsolete objects, which include all old versions whose visible time periods have expired, all invalid versions that are generated due to the aborted transactions, and all empty Edge/VP/EP rows that have been deleted. We pack different types of obsolete objects into different types of GC tasks based on their costs, where each type of tasks handles only one type of obsolete objects and a specific threshold is set to control the batch size. The scanning thread periodically scans and collects obsolete objects. Once a batch has been collected, the GC threads will be activated to garbage-collect these objects. We show the importance of GC by experiments in §6.3.
Indexes. Indexes are critical to achieve efficient query operations such as HAS and WHERE. G-Tran supports standard indexes (e.g., hashtables, B+ trees) on text and numerical values for fast look-up or range search on vertex/edge labels and property values. Users can specify the type of indexes to be constructed and the specific target keys (e.g., a certain property) for indexing via a client console. Then, G-Tran servers coordinate with each other to build a distributed index map in memory. When read-write transactions commit, the indexes will also be updated in real-time to maintain the consistency. and TigerGraph Developer Edition [23]. The experiments were run on a cluster of 10 machines, each equipped with two 8-core Intel Xeon E5-2620v4 2.1GHz processors, 128GB memory and Mellanox ConnectX-3 40Gbps Infiniband HCA, running on CentOS 6.9 with OFED 3.2 Infiniband driver. For fair comparison, we used the same number of computing threads in each machine for all the systems, and tuned the configurations of each system to give its best performance as we could. All query latency reported are the average of five runs and all throughput values are averaged over 300 seconds. Datasets. We used four datasets: one small and one large synthetic property graphs created by LDBC-SNB 1 data generator, and two real-world property graphs crawled from DBpedia 2 (including two parts, citation data and citation links) and Amazon Product 3 (including product reviews and metadata). The small/large datasets were used for the system evaluation on the single-machine/distributed environment respectively. Table 1  In addition, we also selected 8 queries with heavier workloads, i.e., IC1-IC4 and IS1-IS4, in the LDBC SNB benchmark [30] for single-query latency evaluation. Note that we did not use the LDBC benchmark for the comprehensive evaluation as it does not support UPDATE/DELETE workloads and the queries are only applicable on its own synthetic datasets. We list the templates of all queries in both LBV and LDBC benchmarks in Appendix A.

Evaluation of System Designs
We first evaluate the effectiveness of the various system designs, including the data store, the decentralized architecture, the optimizations in MV-OCC (i.e., optimistic pre-read and optimistic validation), and the speed-up due to RDMArelated designs.

Evaluation of Individual System Designs.
To examine the effect of each individual design on the system performance, we created three variants of G-Tran: G-Tran without RDMA but using IPoIB (denoted as IPoIB), G-Tran without the two optimizations in MV-OCC (denoted as No-Opt), and a centralized version (denoted as Cent) with a global master (i.e., transaction coordinator). We tested them on two large graphs, LDBC-L and Amazon, using 8 machines with 20 computing threads per machine. We used a mixed workload formed by READ and WRITE queries in the LBV benchmark.  Half of the queries are WRITE, which create a relatively high-contention scenario. Figure 5 reports the transaction throughput and the abort rate of G-Tran and its three variants, at both SI and SR isolation levels. Compared with G-Tran, the throughput of its IPoIB variant is reduced around 30% -50%, which is due to the higher latency of normal network connection and the extra CPU overhead between NIC and the OS kernel. However, the IPoIB variant still significantly outperforms existing systems as reported in §6.2, which shows that other system designs also play important roles in G-Tran's high performance.
Disabling optimistic pre-read and optimistic validation in the MV-OCC protocol also leads to a degradation of the throughput, especially at SR isolation. As shown in Figure 5(a), on the LDBC-L dataset, No-Opt's throughput (4.49K tps) at SR drops 25.9% compared with G-Tran's throughput (7.55K tps). No-Opt's throughput at SI is only slightly dropped as our multi-version-based solution can already eliminate much of the contention at SI isolation. However, the abort rate of the No-Opt variant is significantly higher. The increase in the abort rate is more obvious on the Amazon dataset as it is a real-world graph with a power-law distribution on vertex degree, leading to higher contention and abort rate. The result shows that the optimizations in the MV-OCC protocol can effectively improve the processing of concurrent graph transactions which have large read/write-sets. We report the distribution of the sizes of the read/writesets of the transactions in Figure 6, showing that although more transactions have relatively smaller read/write-sets, there are also a large number of transactions having large read/write-sets. This also explains the relatively high abort rate of G-Tran at SR compared with that at SI.  1  59  117  175  233  291  349  407  465  523  581  639  697  755  813  871  929  987  1045  1103  1161  1219  1277  1335  1393  1451  1509  1567  1625  1683  1741  1799  1857  1915  1973  2031  To demonstrate the advantage of the decentralized architecture over RDMA for distributed transaction processing, we further compare G-Tran with its Cent version. At SR isolation, Cent has the lowest throughput for both datasets (i.e, 4K tps and 7.66K tps). Because in Cent's setting, the master plays the role of a global coordinator which handles the coordinating tasks of all concurrent transactions. These tasks create significant CPU and network overheads on a single node, which becomes the bottleneck and limits the general processing power of the entire distributed system. We also observe an increase in the abort rate of the Cent version, because the averagely higher latency per transaction leads to higher contention and in turn increases the overall abort rate.
6.1.2 Evaluation of Data Store. Next, we evaluate how the design of the data store effects G-Tran's performance.
Since we cannot disable the data store in G-Tran individually as we did for the other features, we conducted the experiments on a single-machine setting to exclude the influence of G-Tran's RDMA-aware components and decentralized architecture. In addition, before we started to run the workloads, we first warmed up each system by running the mixed LBV workload in §6.1.1 for an extended period of time, to simulate the real scenario that graph data locality has been broken after continuous updates. It leads to more random access on the entities of the graph, which can thus be used to indicate the effectiveness of the data layout.
We used 8 complex queries (i.e., IC1-IC4 and IS1-IS4) in the LDBC benchmark [30] to evaluate the performance. Consider these benchmark queries can only function on the LDBC synthetic dataset, we also involved a typical multi-hop query template into evaluation with the format:
We compared G-Tran with JanusGraph, ArangoDB, Neo4J and TigerGraph, where all the systems ran on a single machine using 20 threads. Table 2 and Table 3 report the query latency of the LDBC queries and the -hop traversal queries,  IC1  IC2  IC3 IC4  IS1  IS2  IS3  IS4  G-Tran  7,085   respectively 4 . G-Tran achieves the shortest latency on all LDBC queries. In particular, for the complex queries, i.e., IC1-IC4, G-Tran's latency is two orders of magnitude smaller than that of JanusGraph and ArangoDB. The gap between G-Tran and Neo4J is smaller but also 2-3 times in most cases. TigerGraph achieves competitive performance in its "run" stage, but every query needs an "install" stage before running and this "install" stage is costly. For the -hop traversal queries, Table 3 shows that latency on all systems increases exponentially as increases, because the size of the read-set grows exponentially for each hop of traversal on the graph. But after traversing more than 2 hops of neighbors, G-Tran starts to show orders of magnitude advantage over the other systems. We explain the results by analyzing the storage design of each system as follows. Neo4J stores all edges of the entire graph into a global sequential table. Consequently, a graph traversal will suffer from jump addressing in physical storage space on Neo4J, because each newly inserted edge is appended directly to the tail of the sequential table without locality guarantee. As for TigerGraph, although it is not open source, considering that it requires users to indicate the edge type (e.g., friendship) during a traversal, we conjecture TigerGraph stores edges separately based on their types. Then, this layout will require an extra aggregation to merge those separate edge sets when a traversal involves multi-type edges, which explains why TigerGraph has poor performance on the -hop traversal queries. Both JanusGraph and ArangoDB store edges as individual cells or collections. Specifically, JanusGraph stores edge cells of one vertex together with its property cells in one row, which incurs extra overhead for locating edges from properties. ArangoDB stores edges as collections of documents. Thus, the execution of multi-hop traversal needs to locate the connected edges of each vertex by searching on edge collections. In comparison, G-Tran divides arbitrary-length adjacencylists into fixed-size rows, which provides more efficient edge insertion and deletion through allocating new rows or compacting under filled rows. Note that we allocate a new row only after filling up all blanks in the existing row and the sparse row compaction is processed in periodical. As a result, traversals in G-Tran can access the edges of a vertex sequentially and the data locality will not be broken even after frequent updates.

Throughput Analysis
Now, we compared G-Tran with existing systems in terms of throughput in both distributed and single-machine settings. We applied three workloads for evaluation: (1) Read-Only (RO), formed by READ queries in the LBV benchmark, (2) Read-Intensive (RI ), which consists of 80% READ and 20% WRITE queries, (3) Write-Intensive (WI ), which consists of 20% READ and 80% WRITE queries. The specific query values (e.g., VPIDs, EPIDs, etc.) in the query templates were randomly generated to form the query-sets for throughput evaluation.
6.2.1 Distributed Processing. We first compared G-Tran and its IPoIB variant (G-Tran-IPoIB) with JanusGraph (using HBase as the storage backend) and ArangoDB. For Neo4J and TigerGraph, we only can download their Developer Edition which does not support distributed processing. We set isolation at SI because both ArangoDB and JanusGraph do not support SR. Figures 7-9 report the throughput scalability of each system on the three workloads respectively over the LDBC-L dataset. Specifically, for scale-up evaluation, we depolyed all systems on 8 machines but varying the number of threads per machine from 4 to 20. For scale-out evaluation, we ran them on 4 to 10 machines with 20 threads/machine. Similar performance pattern is also observed on Amazon dataset, we skip the detailed reports here due to the limited space.
G-Tran achieves significantly higher throughput than other systems over all three workloads. Although the performance of G-Tran-IPoIB is degraded when RDMA features are disabled, it still outperforms both JanusGraph and ArangoDB in all cases. This indicates that the use of RDMA in G-Tran is not the only reason for its high performance, but other system components are also important as discussed in §6.1. As we increase the number of machines from 4 to 10, the results show that using RDMA indeed brings an advantage to distributed processing because G-Tran has higher rate of increase in throughput than G-Tran-IPoIB. Overall, when more machines are used, G-Tran's throughput increases more on RI and WI workloads. This phenomenon can also be observed in scale-up performance. In contrast, the throughput of both JanusGraph and ArangoDB are relatively low and scales poorly. Besides caused by their data storage as we discussed in §6.1.2, this is also related to how the execution engine of each system processes the concurrent transactions. G-Tran's MPP model enables higher parallelism to process each transaction when more resources are available. Thus, these transactions can commit earlier and there is less contention when accessing data. However, both JanusGraph and ArangoDB follow the one-thread-one-transaction mechanism. When more threads are available, more transactions will join in the system and being processed simultaneously. This increases the contention among all concurrent transactions and incurs more overheads. Moreover, when more  machines are involved, the graph will be partitioned into more shards, which breaks the locality of the graph and leads to extra overhead on communication.
6.2.2 Single-Machine Processing. We compared G-Tran with Neo4J, TigerGraph, JanusGraph and ArangoDB for singlemachine transaction throughput at both SI and SR. We used BerkeleyDB as the backend of JanusGraph for its evaluation at SR, as only BerkeleyDB supports SR as a single-machine system. Neo4J and TigerGraph do not support SI, while ArangoDB does not support SR. TigerGraph and JanusGraph failed to load DBPedia as explained in §6.1.2. We used 20 threads for all the systems. As reported in Figure 10 and Figure 11, G-Tran achieves significantly higher throughput than other systems on all workloads at both SR and SI. Note that in single-machine setting, the performance advantages of G-Tran do not come from RDMA and its decentralized architecture, but mainly from its data store design and MV-OCC protocol. JanusGraph has worse performance than others because G-Tran, Tiger-Graph and Neo4J have native graph stores with tailored designs for transaction processing, while JanusGraph is built upon a general NoSQL-based store. G-Tran's good performance comes mainly from (1) its multi-version based storage, which enables lock-free snapshot reads; (2) the optimizations in MV-OCC protocol, which reduces contention among the concurrent transactions with less abort and retry; (3) its MPP execution engine, which enables parallel processing inside each transaction. By contrast, Neo4J's transaction engine does not support SR natively but requires explicit locks on query language level to achieve SR. TigerGraph's low throughput is related to its costly "install" stage, which is used to pre-translate and optimize queries before their real execution. G-Tran requires neither explicit lock nor "install" stage to process transactions. In addition, ArangoDB can achieve higher throughput than JanusGraph due to its MVCC-based storage. But it has inefficient data layout on graph store, thus ArangoDB's throughput is still lower than that of G-Tran.

Performance of Garbage Collection
We also evaluated the effectiveness of G-Tran's Garbage Collection (GC) mechanism by comparing the performance difference between enabling and disabling GC during regular transaction processing. We conducted the experiment using 8 machines with 20 computing threads per machine on the LDBC-L dataset and executed the same workload at SR isolation as we did in §6.1.1. Figure 12 and Figure 13 report the real-time throughput and memory utilization of G-Tran for a period of 600 seconds from the beginning, i.e., as soon as G-Tran finishes the data loading. The throughput is higher at the beginning as there is not many GC jobs to do, and the system becomes stabilized at around 200 seconds. The result shows that, when GC is disabled, the memory consumption of G-Tran increases linearly with time. But when GC is enabled, the memory consumption remains relatively stable. We observe some obvious drops at 200s, 295s, 380s and 500s in Figure 13, which are as a result of the periodical garbage collection that releases the occupied memory and returns it back to the memory pool. During the whole process, the negative side-effects of GC execution on transaction throughput is minor. We can compare the two curves in Figure 12, the throughput of G-Tran in the GC-enabled case follows very closely to the case when GC is disabled, showing that our GC mechanism has low overhead. Actually, at the later period, in the GC-enabled case, G-Tran's throughput even has a tiny increase. This is because at this time, the system storage has been accumulated by many invalid/expired versions of various objects, to clean them up timely can de-fragment those sparse rows in the vertex/edge tables and accordingly improve the entire memory locality, which helps improve the efficiency of data scan and search.
As the GC mechanism categorizes and packages different GC objects into different types of tasks, the result demonstrates the effectiveness of G-Tran's GC mechanism for recycling various types of obsolete objects in a timely fashion (as otherwise the memory footprint would increase linearly over time). We can also observe a significant drop on the throughput in the GC-disabled curve at round 140s in Figure 12. While at the same time point, the GC-disabled curve for memory utilization in Figure 13 shows a non-continuous increase. We guess that at this moment, there were a large number of write transactions coming into the system, and system itself suffered from their processing and commits/aborts in short-term.

RELATED WORK
Graph Databases. There exist many graph databases like Neo4J [3], Titan [1], JanusGraph [2], ArangoDB [4] and OrientDB [6], etc. However, they have no MPP-based distributed transaction processing and most of them support only low isolation level (e.g., snapshot, read committed). Both Grasper [15,18] and TigerGraph [23] target at massive graph queries with native graph store, but Grasper focuses only on OLAP workload instead of OLTP and TigerGraph is not designed for high performance. A1 [12] is an in-memory graph database built upon FaRM [26,27], while it has no graph native design on memory management and transactional optimizations. Compared with these existing systems, G-Tran has distinguished designs in its graph-native data store, RDMA-featured decentralized architecture and MV-OCC, which contribute to its high performance as reported in §6. Graph Processing Systems. Many graph processing systems have been proposed [8,19,34,35,50,58,61,63,67,68] based on Pregel model [43] or other computation models [55,62,66]. But they focus on offline graph workloads such as PageRank, Connected Component and SSSP. There are also other systems that aim at complex graph analytics and mining [16,17,59,60], or streaming graph processing [14,24,44], which can not be used for graph OLTP.
Distributed Transaction Processing. Many systems have been proposed in recent years for distributed transaction processing, including Google's Spanner [21], Granola [22], FaSST [38] and others [25,48,54,56]. Some of them leverage new hardwares such as RDMA, HTM and NVM to achieve high performance (e.g., FaRM [26,27] and DrTM [20,57]). FaRM proposed an RDMA-friendly protocol to enable strict serializiable transactions with high throughput, low latency, and high availability. DrTM proposed an OCC protocol combining both HTM [11] and RDMA to ensure the strong consistency and atomicity. However, these systems are not specially designed for graph, which has its own unique challenges ( §1 and §3).