Mastering RethinkDB

RethinkDB's architecture consists of various components such as cluster, query execution engine, filesystem storage, push changes (real-time feed), and of course RethinkDB client drivers.

Refer to the following diagram to understand the block-level components of RethinkDB:

Pushing changes to a RethinkDB client

This is a revolutionary concept introduced by RethinkDB. Consider this scenario: you are developing an application for the stock market where there are too many changes in a given amount of time. Obviously, we are storing every entry in the database and making sure that other connected nodes or clients know about these changes.

In order to do so, the conventional way is to keep looking (polling) for the data in the particular collection or table in order to find some changes. This improves the latency and turnaround time of packets, and we all know that a network call in a wide area network (WAN) is really costly. An HTTP call in a WAN is really costly.

Then came something called socket. In this, we do the polling operation but from the socket layer, not the HTTP layer. Here, the size of network requests may get reduced, but still we do the polling.

Note

Socket.io is one of the popular projects available for real-time web development.

RethinkDB proposes a reverse approach of this: what about the database itself tells you:

Hey, there are some changes happen in stock value and here are the new and old value.

This is exactly what RethinkDB push changes (change feed in technical terms) does. Once you subscribe to a particular table to look for its changes, RethinkDB just keeps pushing the old and new values of changes to the connected client. By "connected client," I meant a RethinkDB client and not a web application client. The difference between polling and push changes is shown here:

So you will get the changes in the data in one of the RethinkDB clients, say Node.js. And then you can simply broadcast it over the network, using socket probably.

But why are we using socket when RethinkDB can provide us the changes in the data? Because RethinkDB provides it to the middle layer and not the client layer, having a client layer directly talk to the client can be risky. Hence it has not been allowed yet.

But the RethinkDB team is working on another project called Horizon, which solves the issue mentioned previously, to allow clients to communicate to the database using secure layer of the middle tier. We will look at Horizon in detail in Chapter 10, Using RethinkDB and Horizon.

RethinkDB query engine is a very critical and important part of RethinkDB. RethinkDB performs various computations and internal logic operations to maintain high performance along with good throughput of the system.

Refer to the following diagram to understand query execution:

RethinkDB, upon arrival of a query, divides it into various stacks. Each stack contains various methods and internal logic to perform its operation. Each stack consists of various methods, but there are three core methods that play key roles:

To speed up the process, these stacks are transported to every related server and each server begins to evaluate it in parallel to other servers. This process runs recursively in order to merge the data to stream to the client.

The stack in the node grabs the data from the stack after it and performs its own method of execution and transformation. The data from each server is then combined into a single result set and streamed to the client.

In order to speed up the process and maintain high performance, every query is completely parallelized across various relevant clusters. Thus, every cluster then performs the query execution and the data is again merged together to make a single result set.

RethinkDB query engine maintains efficiency in the process too; for example, if a client only requests a certain result that is not in a shared or replicated server, it will not execute the parallel operation and just return the result set. This process is also referred to as lazy execution.

To maintain concurrency and high performance of query execution, RethinkDB uses block-level Multiversion Concurrency Control (MVCC). If one user is reading some data while other users are writing on it, there is a high chance of inconsistent data, and to avoid that we use a concurrency control algorithm. One of the simplest and commonly used methods method by SQL databases is to lock the transaction, that is, make the user wait if a write operation is being performed on the data. This slows down the system, and since big data promises fast reading time, this simply won't work.

Multiversion concurrency control takes a different approach. Here each user will see the snapshot of the data (that is, child copies of master data), and if there are some changes going on in the master copy, then the child copies or snapshot will not get updated until the change has been committed:

RethinkDB does use block-level MVCC and this is how it works. Whenever there is any update or write operation being performed during the read operation, RethinkDB takes a snapshot of each shard and maintains a different version of a block to make sure every read and write operation works in parallel. RethinkDB does use exclusive locks on block level in case of multiple updates happening on the same document. These locks are very short in duration because they all are cached; hence it always seems to be lock-free.

RethinkDB provides atomicity of data as per the JSON document. This is different from other NoSQL systems; most NoSQL systems provide atomicity to each small operation done on the document before the actual commit. RethinkDB does the opposite, it provides atomicity to a document no matter what combination of operations is being performed.

For example, a user may want to read some data (say, the first name from one document), change it to uppercase, append the last name coming from another JSON document, and then update the JSON document. All of these operations will be performed automatically in one update operation.

RethinkDB limits this atomicity to a few operations. For example, results coming from JavaScript code cannot be performed atomically. The result of a subquery is also not atomic. Replace cannot be performed atomically.

RethinkDB supports major used filesystems such as NTFS, EXT and so on. RethinkDB also supports direct I/O filesystems for efficiency and performance, but it is not enabled by default.

About direct I/O

File is stored on disk and when it's been requested by any program, the operating system first puts it into the main memory for faster reads. The operating system can read directly from disk too, but that would slow down the response time because of heavy-cost I/O operation. Hence, the operating system first puts it into the main memory for operation. This is called buffer cache.

Databases generally manage data caching at the application and they do not need the operating system to cache it for them. In such cases, the process of buffering at two places (main memory and application cache) becomes an overhead since data is first moved to the main memory and then the application cache.

This double buffering of data results in more CPU consumption and load on the memory too.

Direct I/O is a filesystem for those applications that want to avoid the buffering at the main memory and directly read files from disk. When direct I/O is used, data is transferred directly to the application buffer instead of the memory buffer, as shown in the following diagram:

Direct I/O can be used in two ways:

• Mounting the filesystem using direct I/O (options vary from OS to OS)

• Opening the file using the O_DIRECT option specified in the open() system call

Direct I/O provides great efficiency and performance by reducing CPU consumption and the overhead of managing two buffers.

Sharding is partitioning where the database is split across multiple smaller databases to improve performance and reading time. In replication, we basically copy the database across multiple databases to provide a quicker look and less response time. Content delivery networks are the best examples of this.

RethinkDB, just like other NoSQL databases, also uses sharding and replication to provide fast response and greater availability. Let's look at it in detail bit by bit.

Sharding in RethinkDB

RethinkDB makes use of a range sharding algorithm to provide the sharding feature. It performs sharding on the table's primary key to partition the data. RethinkDB uses the table's primary key to perform all sharding operations and it cannot use any other keys to do so. In RethinkDB, the shard key and primary key are the same.

Upon a request to create a new shard for a particular table, RethinkDB examines the table and tries to find out the optimal breakpoint to create an even number of shards.

For example, say you have a table with 1,000 rows, the primary key ranging from 0 to 999, and you've asked RethinkDB to create two shards for you.

RethinkDB will likely find primary key 500 as the breaking point. It will store every entry ranging from 0 to 499 in shard 1, while data with primary keys 500 to 999 will be stored in shard 2. The shards will be distributed across clusters automatically.

You can specify the sharding and replication settings at the time of creation of the table or alter it later. You cannot specify the split point manually; that is RethinkDB's job to do internally. You cannot have less server than you shard.

You can always visit the RethinkDB administrative screen to increase the number of shards or replicas:

We will look at this in more detail with practical use cases in Chapter 5, Administration and Troubleshooting Tasks in RethinkDB totally focused on RethinkDB administration.

Let's see in more detail how range-based sharding works. Sharding can be basically done in two ways, using vertical partitioning or horizontal partitioning:

• In vertical partitioning, we store data in different tables having different documents in different databases.

• In horizontal partitioning, we store documents of the same table in separate databases. The range shard algorithm is a dynamic algorithm that determines the breaking point of the table and stores data in different shards based on the calculation.

RethinkDB provides automatic failover handling in a multi-server configuration where multiple replicas of a table are present. In case of node failure due to any reason, RethinkDB finds out the other node to divert the request and maintain the availability. However, there are some requirements that must be met before considering automatic failover handling:

Every table, by default, has a primary replica created by RethinkDB. You can always change that using the reconfigure() command. In case of failure of the primary replica of the table, as long as more than half of the replicas with voting option are available, one of them will be internally selected as the primary replica. There will be a slight offline scenario while the selection is going on in RethinkDB, but that will be very minor and no data will be lost.

As soon as the primary replica comes online, RethinkDB automatically syncs it with the latest documents and switches control of the primary replica to it automatically.

By default, every replica in RethinkDB is created as a voting replica. That means those replicas will take part in the failover process to perform the selection of the next primary replica. You can also change this option using the reconfigure() command.

Automatic failover requires at least three server clusters with three replicas for table. Two server clusters will not be covered under the automatic failover process and the system may go down during the failure of any RethinkDB instance.

In such cases-where RethinkDB cannot perform failover-you need to do it manually using the reconfigure() command, by passing the emergency repair mode key.

Upon running emergency repair mode, each of the shards is first examined and classified into three categories:If more than half of the total shards are available, it will return a healthy status

For each and every shard that can be repaired, RethinkDB will first change all the offline replicas into non-voting replicas. If there is no voting replica available, RethinkDB will choose one non-voting replica and forcefully convert it into a voting replica.

You can specify two options along with the emergency repair option:

Here is the command reference:

r.table(users).reconfigure(
{emergencyRepair: "unsafe_rollback_or_erase"}
).run(conn, callback);

Please note that emergency failover handling is very critical and should be done by a skilled person, or else you may end up losing all your data.

The RethinkDB data model consists of two main components:

RethinkDB data types are further classified into basic data types, dates, binary objects, and geospatial queries.

Refer to the following diagram for more details:

Let's go over to each component in detail.

var hash = md5(email); 
gravatarUrl = 'http://gravatar.com/avatar/' + hash + '?d=retro'; 
r.table('users').get(userId).update({ 
gravatar: r.http(gravatarUrl, {resultFormat: 'binary'}) 
}).run(conn, callback) 

r.table('demo').insert([ 
 { 
  name : "Mumbai", 
  location : r.point(19.0760,72.8777)  
 } 
]) 

RethinkDB model relationships

RethinkDB, besides being a NoSQL database, provides one of the most requested features by SQL developers, that is, JOINS. RethinkDB allows you to model and structure your data in such a way that allows you to perform JOINS over it. There are two ways to model relationships in RethinkDB:

• By using embedded arrays in the document

• By linking documents stored in multiple tables

Let's see how both of them work, along with their merits and demerits.

Embedded arrays

Embedded arrays basically means using an array of objects in the table in each document. Here is a simple example:

{ 
  "id": "7644aaf2-9928-4231-aa68-4e65e31bf219", 
  "name": "Shahid", 
  "chapters": [ 
    {"title": "Chapter 01", "pages": "30+"}, 
    {"title": "Chapter 02", "pages": "30+"} 
  ] 
} 

Here, the "chapters" key contains the array of objects, and each object contains the name of the chapter and the page count. In order to perform the query, here is how we do it in RethinkDB.

To display all chapters from table:

r.db("book").table("chapters").run()

To get all the chapters written by a particular author; we can do that by passing the author ID:

r.db("book").table("chapters").get('7644aaf2-9928-4231-aa68-4e65e31bf219').run()

Here, we are passing the ID of the document, which is a system-generated primary key ID by RethinkDB. Upon passing it, we can access a single document from the table. This is a very basic way of modelling and is not recommended for production use.

Merits of embedded arrays

The following merits are as follows:

• Easy to design and query

• Updates on a single document will automatically update chapters and authors both. Hence, both pieces of data will be atomic

Demerits of embedded arrays

As soon as the size of an embedded array exceeds a certain limit, it costs too much computing power to load the entire array every time we have to perform an operation.

Document linking in multiple tables

This is similar to the foreign key relationships we make in traditional SQL databases. Here, we link two or more tables using the index. We can also query the table using functions provided by RethinkDB to perform various joins (inner, outer, and so on):

As shown in the preceding diagram, we have two tables with multiple documents in them. The Authors table will just contain documents related to authors, say name, location, and (obviously) a primary key.

The Chapters table will contain documents of chapters; each chapter is a separated document instead of an array. The only difference is that each document will contain an index with the same value as the primary key from the Authors table. This way, we link both of them together.

As you can see in the diagram, there is a key in every document in the Chapters table named author_id with the same value as id in the Authors table.

Here is how we query the tables to perform JOINS:

r.db("books").table("chapters").   filter({"author_id": "abc768332"}). run() 

This will return every chapter for the author having ID as abc768332.

You can also use RethinkDB JOINS functions such as eq_join(); we will look at the in the next section.

Merits of document linking

The following merits are:

• Data distribution and very neat design

• No need to load all of the data of chapters at once in order to perform any operation

• No matter what the size of documents in chapters table is, the performance won't be affected as in embedded arrays

Demerits of document linking

The following demerits are:

• There are no foreign key constraints, so linking of data will be complicated

• In the case of an update in one table, it won't automatically update the data in another table

Here is a list of JOIN commands provided by RethinkDB:

• eq_Join

• innerJoin

• outerJoin

• Zip

Before looking at these, let's populate some data in two tables. We have authors and chapters in the books database.

We will add author details in the authors table:

r.db('books').table('authors').insert({name : "Shahid", location : "Mumbai"}) 

We will be adding chapter details in the chapters table:

r.db('books').table('chapters').insert({ 
   author_id : "521b92b1-0d83-483d-a374-b94a400cf699",  
   chapterName : "Chapter 1" 
}) 
 
r.db('books').table('chapters').insert({ 
   author_id : "521b92b1-0d83-483d-a374-b94a400cf699",  
   chapterName : "Chapter 2" 
}) 

Performing eq_Join

Here we are joining the chapters table with the authors table and mentioning which field we have mapped to the ID, that is, author_id:

r.db('books').table('chapters').eqJoin("author_id",r.db('books').table('authors')) 
 

This will return the following:

[ 
   { 
      "left":{ 
         "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
         "chapterName":"Chapter 1", 
         "id":"f0b5b2f7-1f82-41ef-a945-f5fa8259dd53" 
      }, 
      "right":{ 
         "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
         "location":"Mumbai", 
         "name":"Shahid" 
      } 
   }, 
   { 
      "left":{ 
         "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
         "chapterName":"Chapter 2", 
         "id":"f58826d4-e259-4ae4-91e4-d2e3db2d9ad3" 
      }, 
      "right":{ 
         "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
         "location":"Mumbai", 
         "name":"Shahid" 
      } 
   } 
] 

The left and right keys supposedly represent the tables on the left side and the right side of the query. If you map the keys and values with the ReQL query, you can easily understand.

However, this is not how we really want our data. We want processed, result-oriented data, and to do that, we need to use the zip() command. It basically removes all metadata information from the result and gives you only the documents of tables:

r.db('books').table('chapters').eqJoin("author_id",r.db('books').table('authors')).zip() 

This returns the following:

[ 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 1", 
      "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "location":"Mumbai", 
      "name":"Shahid" 
   }, 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 2", 
      "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "location":"Mumbai", 
      "name":"Shahid" 
   } 
] 

Performing inner joins

An inner join, as we all know, equates two tables by comparing each row. It is very similar to working with EQJOIN, except that it compares each document in the table against the target table, while in EQJOIN, we specify which key to compare for.

Here is a simple query to perform an inner join:

r.db('books').table('chapters').innerJoin(r.db('books').table('authors'),function(chapters,authors) { 
  return chapters 
}).zip() 

This function takes the target table as a parameter and the callback function, which contains data of both the tables in the callback argument. If you notice, for understanding, I've named the callback parameters the same as the table name. You can perform a lot of other operations such as comparison, or filtering inside the callback function and then returning the result. Since it's a JOIN, in both the variables, data will be similar-except with different table ID's.

Here is the result for the same:

[ 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 1", 
      "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "location":"Mumbai", 
      "name":"Shahid" 
   }, 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 2", 
      "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "location":"Mumbai", 
      "name":"Shahid" 
   } 
] 

If you observe, the result set is pretty much the same as EQJOIN (the eq_Join() function), except that it provides you the result of each document under the callback function. This makes it really slow, and the RethinkDB team does not recommend it for use in production.

Performing outer joins

Outer join union the result of left join and right join and returns it to the client. So, basically, the result test from both the tables will be combined and returned. Here is a sample query.

r.db('books').table('chapters').outerJoin(r.db('books').table('authors'),function(chapters,authors) { 
  return authors 
}).zip() 

This will combine each document of chapters with each document of authors and return the result. Again, we can access each document of the query here under the callback.

It shall return the following:

[ 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 1", 
      "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "location":"Mumbai", 
      "name":"Shahid" 
   }, 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 2", 
      "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "location":"Mumbai", 
      "name":"Shahid" 
   } 
] 

In order to check it is working, let's just create one more author in the authors table and not create any chapter document entry for it:

r.db('books').table('authors').insert({name : "RandomGuy", location : "California"}) 

Upon running the outer join query again, here is the result:

[ 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 1", 
      "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "location":"Mumbai", 
      "name":"Shahid" 
   }, 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 2", 
      "id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "location":"Mumbai", 
      "name":"Shahid" 
   }, 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 1", 
      "id":"78acabb5-a5b8-434b-acb1-52507b71831d", 
      "location":"California", 
      "name":"RandomGuy" 
   }, 
   { 
      "author_id":"521b92b1-0d83-483d-a374-b94a400cf699", 
      "chapterName":"Chapter 2", 
      "id":"78acabb5-a5b8-434b-acb1-52507b71831d", 
      "location":"California", 
      "name":"RandomGuy" 
   } 
] 

Hence we get all of the result-the union of each document of the table present in the left table (that is, authors) with each document present in the right table (that is, chapters).

Zip

This function performs the merging of left fields with right fields with a JOIN operation into a single dataset.

We have covered various architectural features and the data model of RethinkDB. Let's look over some of the constraints of RethinkDB that you need to take into account while architecting your data store.

RethinkDB divides the limitation into hard and soft limitations. The hard limitations are as follows:

RethinkDB also has some memory limitation, as follows:

RethinkDB, in order to keep performance high, stores some data in the RAM. There are basically three sources of usage of RAM by RethinkDB:

RethinkDB stores metadata of tables in the main memory in order to ensure fast read access. Every table consumes around 8 MB per server for the metadata. RethinkDB organizes the data into blocks, with size ranging from 512 bytes to 4 KB. Out of these blocks, approximately 10 to 26 bytes per block are kept in memory.

Page cache is a very important aspect of performance. It is basically used to store very frequently accessed data in the RAM rather than reading it from disk (except in the case of direct I/O, where the page cache is in the application buffer than RAM). RethinkDB uses this formula to calculate the size of the cache:

Cache size = available memory - 1024 MB / 2

If the cache size is less than 1224 MB, then RethinkDB set the size of page cache to 100 MB. This is why it is recommended to have at least 2 GB of RAM allocated for RethinkDB processes.

You can also change the size of the page cache when you start the server or later, using configuration files.

Every database uses some memory to store the results of ongoing running queries. Since queries differ, in general, there is no exact estimate about memory usage by running queries; however, a rough estimate is between 1 MB and 20 MB, including background processes such as transferring data between nodes, voting processes, and so on.

RethinkDB is indeed a next-generation database with some amazing features and high performance. Due to these architectural advantages, it has been trusted by top-level organizations such as NASA. We covered the architecture of RethinkDB in detail in order to understand how it works and why it is great for real-time databases. We also covered disk storage, clustering, and failover handling of RethinkDB.

Along with the architecture, we also looked over data modeling in RethinkDB. We also looked over one of the most used features of SQL-JOINS working in a NoSQL database, that is, RethinkDB. There is no doubt that RethinkDB is the next big thing.

In next chapter, we are going to learn about the query language of RethinkDB, called ReQL. We will go over it with examples. We will also cover changefeeds, which is personally my favorite feature of RethinkDB.