In this chapter we will study the key elements of the selected systems one by one and highlight their differences and similarities compared to RethinkDB. To achieve that, we will make a distinction into to two other system categories. The first one includes state of the art distributed replication systems, such as SMART, Niobe and Petal, and the second category includes widely used database replication systems. We made this separation in order to achieve a clear distinction and due to the systems' common features.

4.1 RethinkDB vs Niobe vs Petal vs SMART

Quorum. Due to the fact that SMART, Niobe and Petal use Paxos as their GSM, a majority of servers who are able to communiciate with each other, is required in order to ensure progress. The same applies to our system, with the main difference that RethinkDB uses Raft as its GSM.

Regarding the heartbeat process SMART, Niobe and RethinkDB use the classic heartbeat technique where the leader exchanges messages periodically with other replicas. Petal uses a similar technique called liveness module.

Reads. All four systems implement different techniques regarding data reading. RethinkDB offers three possible read modes. The default way is the "single" value, which returns values that are in memory (but not necessarily written to disk) on the primary replica. There is also the "majority" mode which only returns values that are safely committed on disk on a majority of replicas. This requires sending a message to every replica on each read, so it is the slowest but most consistent(ensures safety even in the case of a system partition). Finally, the "outdated" mode will return values that are in memory on an arbitrarily-selected replica. This is the fastest but least consistent.

In Niobe, if a replica receives a Read request from a front-end, it first acquires a lock for reading, guaranteeing that no Update requests are being processed(see section 4.2 in John MacCormick et al). Then, the replica checks that it believes itself to be the primary, and if not, replies with an appropriate hint. Otherwise, it checks that it holds a read lease from every secondary it believes is alive. The leasing mechanism employs an extra parameter, which is the maximum amount of network or other delay that is permitted before we consider a fault has occurred.

Petal uses three of its modules(liveness, data access, virtual to physical translator) to service client reads. This process is separated into three main steps: (1) The virtual disk directory translates the client-supplied virtual disk identifier into a global map identifier. (2) The specified global map determines the server responsible for translating the given offset. (3) The physical map at the specified server translates the global map identifier and the offset to a physical disk and an offset within that disk.

Finally, in SMART when the client sends a read request,and after some communication delay the leader receives it and sends a proposal. The leader logs this proposal immediately, while the other replicas must wait until they receive it. Once the leader receives LOGGED messages from a quorum, it begins executing the request.

Recovery mechanisms. RethinkDB, SMART and Petal use different variations of the snapshooting technique to ensure data safety. On the other hand, Niobe uses a recovery mechanism based on a dirty-region log, and the GSM ensures data is consistent before clients are allowed access to the affected data item.

Reconfiguration. RethinkDB uses a slight variation of Raft's reconfiguration mechanism, which will be explained in a following section. Petal and SMART use a similar approach to reconfiguration based on the multi-paxos protocol. However, Petal's GSM also enforces two additional restrictions. First, requests are not pipelined(α = 1). Second, only two types of migration are allowed: adding one machine, or removing one machine. This ensures quorums from consecutive configurations always overlap. On the contrary, SMART does not require any overlap at all between consecutive configurations. Finally, Niobe uses a centralized GSM to store and update the system configuration, but does not specify the implementation of the GSM.

4.2 RethinkDB vs Cassandra vs HBase

In this section we will compare three commercially used distributed database systems.

HBase is wide-column store based on Apache Hadoop and on concepts of BigTable. Apache HBase is a NoSQL key/value store which runs on top of HDFS (Hadoop Distributed File System) and its operations run in real-time on its database rather than MapReduce jobs. HBase is partitioned to tables, and tables are further split into column families.

Cassandra is Wide-column store based on ideas of BigTable and DynamoDB. Apache Cassandra is a leading NoSQL, distributed database management system which offers continuous availability, high scalability and performance, strong security, and operational simplicity.

Indexing. In RethinkDB when the user creates a table, they have the option of specifying the attribute that will serve as the primary key. When the user inserts a document into the table, if the document contains the primary key attribute, its value is used to index the document. Otherwise, a random unique ID is generated for the index automatically. The primary key of each document is used by RethinkDB to place the document into an appropriate shard, and index it within that shard using a B-Tree data structure. Querying documents by primary key is efficient, because the query can immediately be routed to the right shard and the document can be looked up in the B-Tree. RethinkDB also supports both secondary and compound indexes, as well as indexes that compute arbitrary expressions.

In HBase, data is stored in tables, which have rows and columns. A row in HBase consists of a row key and one or more columns with values associated with them. HBase uses a single index that is lexicographically sorted on the primary row key. Access to records, in any way other than through the primary row, requires scanning over potentially all the rows in the table to test them against your filter. Hbase does not natively support secondary indexes, but one use-case of Triggers is that a trigger on a "put" can automatically keep a secondary index up-to-date, and therefore not to burden the application.

Cassandra is a partitioned row store, where rows are organized into tables with a required primary key. The first component of a table's primary key is the partition key; within a partition, rows are clustered by the remaining columns of the PK. Other columns may be indexed independent of the PK. Cassandra also provides a native indexing mechanism in secondary indexes. Secondary indexes work off of the columns values. The user has to declare a secondary index on a Column Family.

Cap theorem. In theoretical computer science, the CAP theorem also known as Brewer's theorem, states that it is impossible for a distributed computer system to simultaneously provide all three of the following guarantees: (1) Consistency (all nodes see the same data at the same time). (2) Availability (a guarantee that every request receives a response about whether it succeeded or failed). (3) Partition tolerance (the system continues to operate despite arbitrary partitioning due to network failures)

Both RethinkDB and HBase sacrifice availability in favor of consistency and partition tolerance. On the other hand, Cassandra aims to optimize performance first and foremost, therefore goes for availability and partition tolerance instead of consistency.

ACID. (Atomicity, Consistency, Isolation, Durability) is a set of properties that guarantee that database transactions are processed reliably. In the context of databases, a single logical operation on the data is called a transaction.

• Atomicity: In HBase all mutations are atomic within a row. That mutate several rows will not be atomic across the multiple rows. Any write will either wholly succeed or wholly fail. However, operations that mutate several rows will not be atomic across the multiple rows. In Cassandra, a write is atomic at the partition-level, meaning inserting or updating columns in a row is treated as one write operation. Cassandra does not support transactions in the sense of bundling multiple row updates into one all-or-nothing operation. In RethinkDB, write atomicity is supported on a per-document basis - updates to a single JSON document are guaranteed to be atomic. However, RethinkDB operations are never atomic across multiple keys.
• Consistency: HBase seeks to maintain consistency. In contrast to "eventually consistent" Cassandra, which allows to tune between availability and consistency, HBase does not offer various consistency level settings. Thus, the price of HBase's strong consistency is that writes can be slower. On the other hand, in RethinkDB data always remains immediately consistent and conflict-free, and a read that follows a write is always guaranteed to see the write.
• Isolation: Cassandra and HBase offer row-level isolation, which means that writes to a row are isolated to the client performing the write and are not visible to any other user until they are complete. By default, RethinkDB can return data from concurrent writes that have not been committed to disk yet. The read_mode option to table allows control of the isolation level.
• Durability: All three systems are strictly durable in a way that data that have been committed to disk will survive permanently.

Below is a table that sums up the various properties of the three database systems.

Like all other systems, RethinkDB is not without its flaws. In this section we will describe the the most important limitations we came across while examining the system. To achieve a clear representation of the limitations we categorized them into their respective categories.

System requirements. When it comes to system requirements, there are no strict limitations, except that RethinkDB servers must have at least 2GB of RAM. In addition, although RethinkDB runs on several systems (Linux 32bit and 64bit, OS X 10.7) and has experimental ARM support, it does not run on Windows.

Table and Sharding limitations. RethinkDB does not limit the number of databases that can be created, but it limits the number of shards to 32. It also doesn't limit the number of tables that can be created per database or cluster. Each table requires a minimum of approximately 10MB disk space on each server in a cluster, and adds an overhead of 8MB RAM to the server it's replicated on. While there is no hard limit on the size of a single document, there is a recommended limit of 16MB. The maximum size of a JSON query is 64M. Finally, primary keys are limited to 127 characters.

Failover. When a server fails, it may be because of a network availability issue or something more serious, such as system failure. In a multi-server configuration, where tables have multiple replicas distributed among multiple physical machines, RethinkDB will be able to maintain availability automatically in many cases. To perform automatic failover for a table, the following requirements must be met: (1) The cluster must have three or more servers, (2) The table must be configured to have three or more replicas, (3) A majority (greater than half) of replicas for the table must be available.

In Rethinkdb if the primary replica for a table fails, as long as more than half of the table's voting replicas and more than half of the voting replicas for each shard remain available, one of those voting replicas will be arbitrarily selected as the new primary. There will be a brief period of unavailability, but no data will be lost. If the primary replica specified in a table's configuration comes back online after a failure, it will return to being the primary.

Other considerations. As mentioned beforehand, RethinkDB does not have full ACID support. Furthermore, advanced computations are generally slow compared to a column-oriented system. Finally, RethinkDB trades off write availability in favor of data consistency.

In this section we will go into detail with the technical aspects of the system's implementation.

6.1 Overview

Basic primitives. Each RethinkDB process has a thread pool with a fixed number of threads. Each of these threads is running an event loop that processes IO notifications from the OS and messages from other threads.

Each RethinkDB server in a cluster maintains a TCP connection to every other server in the cluster. The messages sent across these TCP connections fit into two basic paradigms which are called "mailboxes" and "directory". A mailbox is an object that is created dynamically on a specific server. It has an address, which is serializable; if one server has the address of a mailbox on another server, it can send messages to that mailbox. The directory consists of a set of values that each server broadcasts to every other server it's connected to. A server sets up a mailbox to process a certain type of request and then sends that mailbox's address over the directory so other servers can discover it. These mailbox addresses in the directory are called "business cards".

Executing queries. During start-up one of the constructed objects is an instance, which listens for incoming TCP connections from client drivers. When it receives a message, it converts the message from the client into a tree of ReQL terms. An operation on a table is represented as a read or write object; the result of the operation is represented as a read or write response object. Every RethinkDB table is represented as a B-tree. The blocks of the B-tree are stored in RethinkDB's custom page cache. When the store command finishes executing the read or write against the B-tree, it returns a read or write response, which is passed back via the same way it arrived.

Table configuration. The table's configuration describes which servers should be performing which role for which part of the table. When the user changes the table's configuration, objects on various servers must be created and destroyed to implement the user's new configuration, while making sure that the tables' contents are correctly copied from server to server. There is one multi-table manager per server, created upon initiation. It keeps a record for each table that is known to exist, whether that table has a replica on that server or not. If the server is supposed to be a replica for a given table, the multi-table manager will construct a table manager for that table.

RethinkDB uses the Raft consensus algorithm to manage table meta-data, but not to manage user queries. RethinkDB uses a custom implementation of Raft that's designed to work with the RethinkDB event loop and networking primitives. There is one Raft cluster per RethinkDB table. Each table manager instantiates an object for each raft member. Among other things, the state contains the table's current configuration and a collection of contract objects which describe what each server is supposed to be doing with respect to each region of the table.

6.2 RethinkDB's Raft Implementation

As mentioned above RethinkDB uses a custom implementation for the Raft algorithm, which has several differences with the one presented in the Raft paper. These differences are listed below.

• Server initiation. In RethinkDB's implementation, when a new peer joins the cluster it is initialized by copying the log, snapshot, etc. from an existing peer, instead of starting with a blank state.
• Snapshot. RethinkDB's Raft implementation deviates from the Raft paper in the order of snapshots. The Raft paper proposes that the first step should be to save the snapshot, but because RethinkDB only stores one snapshot at a time and requires that snapshot to be right before the start of the log, it saves the snapshot if and when the log is updated instead.
• Committing index. The implementation deviates from the Raft paper in that they persist the commit index to disk whenever it changes.
• Heartbeat process. Their implementation deviates significantly from the Raft paper in that they don't actually send a continuous stream of heartbeats from the leader to the followers. Instead, they send a "virtual heartbeat stream"; They send a single "start virtual heartbeats" message when a server becomes leader and another "stop virtual heartbeats" message when it ceases to be leader, and they rely on the underlying networking layer to detect if the connection has failed.
• Log entries. RethinkDB's implementation slightly differs from the Raft paper in that when the leader commits a log entry, it immediately sends an append-entries message so that clients can commit the log entry too. This is necessary because RethinkDB uses "virtual heartbeats" in place of regular heartbeats, and virtual heartbeats don't update the commit index.
• Leader election. In RethinkDB a candidate sends out request-vote messages and waits to get enough votes. It blocks until it is elected or a "cancel signal" message is received. The candidate is responsible for detecting the case where another leader is elected and also for detecting the case where the election times out, and pulsing the "cancel signal" message.
• Watchdog. In RethinkDB there is no way for the leader to reply to a virtual heartbeat. So the leader needs some other way to find out that another peer's term is higher than its term. In order to make this work, RethinkDB's implementation varies from the Raft paper in how leaders handle RequestVote messages; Using an additional "watchdog" mechanism RethinkDB detects whether a peer has heard from a leader by using a fixed timer which checks periodically.
• Resending Messages. In the Raft paper, servers retry messages if they do not receive a response in a timely manner. RethinkDB implementation deviates from the Raft paper slightly in that it does not retry the exact same message necessarily. If the snapshot has been updated, it sends a newer snapshot on the next try.

Before we conclude our examination of RethinkDB, we will conduct some experiments regarding availability and performance. Specifically we will test the system's performance and availability in the following categories:

• Read Performance
• Write Performance
• Ram Usage Statistics
• Availability

All tests were executed on typical personal computer with Intel(R) Core(TM) i7-4510U CPU (dual core with hyper-threading) running at 2.00GHz, 4GB RAM running at 1600MHz and a solid state disk with reading speed up to 540 MB/sec and writing speed up to 500 MB/sec. For all experiments we use threes servers to achieve the majority constraint and separated the data into three shards with three replicas each.

7.1 Read Performance

The goal of this experiment is to measure reading rates under various circumstances such as increasing traffic in varying table size. We considered three different table sizes, small, medium and large which are explained in numbers below.

Observations. We observed that read rates show fluctuations possibly because of the way the system distributes the machine's resources. As we can see in figures 7.1 and 7.2 read performance increases with a steady rate for up to four clients and then stabilizes. This can be explained due to our machine's specifications (4 threads). For large tables as seen in figure 7.3, the results are inconclusive because the machine runs out of available resources which leads to total breakdown.

7.2 Write Performance

To measure the writing performance of the system we simply continuously increase the number of client requests and evaluate the system's response.

Observations. As explained earlier, write speeds exhibit a steady increase up to four clients peaking at approx. 4KBps. It is worth mentioning that writing performance in general follows a spiky behavior as shown in figure 7.5.

7.3 RAM Usage Statistics

In this experiment we will measure the RAM usage while increasing the amount of client requests. Similarly to section 7.1 we used varying table sizes.

Observations. We observe that both for small and medium table sizes, the increase in RAM usage seems to be linear. As expected, medium size tables require more RAM than the small ones. Finally, in the case of large tables the system drastically consumes all available memory leading the machine to a failure state.

7.4 Availability

In the final experiment, we tested the system's availability under certain circumstances such as server failure and online reconfiguration. In the following experiments the recorded delay during reconfiguration is the delay that is experienced from a client during constant read requests. The delay figures reported below do not consider network delay due to the servers running locally. Also, the numbers are measured in seconds.

Observations. As we can see in table 7.2 the number do not seem to follow any pattern in the small and medium table size categories. In the large size category there is a big delay when performing a state transfer into a smaller number shards and replicas and the system starts to behave abnormally as can be seen figure 7.9. Before the start of the reconfiguration (indicated by the red spike in writes) the system functions more or less normally. However, after reconfiguration the system becomes only periodically available to clients. Worse still, in the case of transfer into a bigger number of shards and replicas the system becomes unresponsive and unable to finish the configuration process. This is further proof that the system requires a lot of resources in order to function properly during reconfiguration.

Finally, delay in case of a server crash is the same for small and medium table sizes. The experiment was impossible to carry out in large tables due to the system running out of resources, as mentioned previously.

In this project we thoroughly examined every core aspect of RethinkDB and the Raft consensus algorithm which is the underlying foundation for RethinkDB. Regarding Raft we tried to break it down into separate parts to obtain a better understanding of each component. We then proceed to compare it to the benchmark consensus algorithm of distributed systems, Paxos, and highlight their main differences and similarities. After this in depth study, we reached the conclusion that Raft's simplicity and role separation are key elements that will contribute to the future establishment as building and study foundation.

After studying the underlying mechanism we went on to an upper level of the system. To gain a deeper insight of the system's functionalities, we examined the provided features, the practical considerations and the system's caveats. We then compared the system to two basic system categories; widely used database replication and state-of-the-art distributed replication. Considering all the above, we form the conjecture that RethinkDB belongs to the top shelf of distributed database systems. However, it is not always the optimal choice depending on the user's requirements.

Finally, to test RethinkDB's availability and performance we conducted a series of benchmark-like simulations. Although some experiments were inconclusive, the general picture we obtained is that the system operates normally under relatively regular sized data. On the other hand, for big amount of data our personal computer was unable to handle the system's resources requirements. This leads us to the assumption that a typical personal computer is not suitable to manage large scale data. It would be interesting to furtherly test the system on a high-end computer or on a cluster of personal computers.

• RethinkDB Runner
• Killer Script
• Ram Calculator for Rethinkdb
• Python Client 1
• Python Client 2
• Python Client 3