Getting Started with CockroachDB

Copyright ©2022 Packt Publishing

All rights reserved. No part of this book may be reproduced, stored in a retrieval system, or transmitted in any form or by any means, without the prior written permission of the publisher, except in the case of brief quotations embedded in critical articles or reviews.

Every effort has been made in the preparation of this book to ensure the accuracy of the information presented. However, the information contained in this book is sold without warranty, either express or implied. Neither the author, nor Packt Publishing, and its dealers and distributors will be held liable for any damages caused or alleged to be caused directly or indirectly by this book.

Packt Publishing has endeavored to provide trademark information about all of the companies and products mentioned in this book by the appropriate use of capitals. However, Packt Publishing cannot guarantee the accuracy of this information.

Early Access Publication: Getting Started with CockroachDB

Early Access Production Reference: B16571

Published by Packt Publishing Ltd.

Livery Place

35 Livery Street

Birmingham

B3 2PB, UK

ISBN: 978-1-80056-065-9

www.packt.com

Table of Contents

1. Getting Started with CockroachDB

2. 1 CockroachDB – A Brief Introduction

1. The history and evolution of databases

1. SQL

2. Object-oriented databases

3. NoSQL

4. NewSQL

2. Database concepts

1. Cardinality

2. Overview of database models

3. Processing models

4. Embedded and mobile databases

5. Database storage engines

3. CAP theorem

1. Consistency and partition tolerance (CP)

2. Availability and partition tolerance (AP)

3. Consistency and availability (CA)

4. CockroachDB

1. Why yet another database?

2. Inspiration

3. Key terms and concepts

4. High-level overview

5. Summary

3. 2 How Does CockroachDBWork Internally?

1. Technical requirements

2. Installing a single-node CockroachDB cluster using Docker

3. Execution of a SQL query

1. SQL query execution

2. Parsing

3. Logical planning

4. Physical planning

5. Query execution

4. Managing a transactional key-value store

5. Data distribution across multiple nodes

1. MSKVS

2. Meta ranges

3. Table data

6. Data replication for resilience and availability

1. What is consensus?

2. Raft distributed consensus protocol

7. Interactions with the disk for data storage

1. Storage engine

8. Summary

4. 3 Atomicity, Consistency, Isolation, and Durability (ACID)

1. An overview of ACID properties

1. Atomicity

2. Consistency

3. Isolation

4. Durability

2. ACID from CockroachDB’s perspective

1. Atomicity

2. Consistency

3. Isolation

4. Durability

3. Summary

5. 4 Geo-Partitioning

1. Technical requirements

2. Introduction to geo-partitioning

3. Cloud regions and zones

1. Region

2. Zone

3. Regions and zones on various cloud providers

4. Geo-partitioning in CockroachDB

1. Single region

2. Multi-region

5. Summary

6. 6 How Indexes Work in CockroachDB

1. Introduction to indexes

2. Different types of indexes

1. Primary indexes

2. Secondary indexes

3. Hash-sharded indexes

4. Duplicate indexes

5. Inverted indexes

6. Partial indexes

7. Spatial indexes

8. Table joins and indexes

3. Best practices while using indexes

4. Summary

1. Cover

2. Table of contents

Getting Started with CockroachDB

Welcome to Packt Early Access. We’re giving you an exclusive preview of this book before it goes on sale. It can take many months to write a book, but our authors have cutting-edge information to share with you today. Early Access gives you an insight into the latest developments by making chapter drafts available. The chapters may be a little rough around the edges right now, but our authors will update them over time. You’ll be notified when a new version is ready.

This title is in development, with more chapters still to be written, which means you have the opportunity to have your say about the content. We want to publish books that provide useful information to you and other customers, so we’ll send questionnaires out to you regularly. All feedback is helpful, so please be open about your thoughts and opinions. Our editors will work their magic on the text of the book, so we’d like your input on the technical elements and your experience as a reader. We’ll also provide frequent updates on how our authors have changed their chapters based on your feedback.

You can dip in and out of this book or follow along from start to finish; Early Access is designed to be flexible. We hope you enjoy getting to know more about the process of writing a Packt book. Join the exploration of new topics by contributing your ideas and see them come to life in print.

Getting Started with CockroachDB

1. CockroachDB - A Brief Introduction

2. How Does CockroachDB Work Internally?

3. Atomicity, Consistency, Isolation, and Durability (ACID)

4. Geo-Partitioning

5. Fault Tolerance and Auto-Rebalancing

6. How Indexes Work in CockroachDB

7. Schema Creation and Management

8. Exploring Admin User Interface

9. Security Aspects

10. Troubleshooting Issues

1 CockroachDB – A Brief Introduction

In this chapter, we will go over the history of databases, where we will learn about the evolution of SQL, NoSQL, and NewSQL databases, various relational models, different categories for classifying databases, and timelines. Later, we will discuss the CAP theorem. Finally, we will briefly discuss the motivation for creating a new database and learn about the basic architecture of CockroachDB.

The following topics will be covered in this chapter:

- History and evolution of databases

- Database concepts

- CAP theorem

- CockroachDB

The history and evolution of databases

A database is a collection of data that can be organized, managed, modified, and retrieved using a computer. The system that helps with managing data in a database is called a database management system (DBMS).

In the 1950s and 1960s, several advancements were made in terms of processors, storage, memory, and networks. We also had our first programming languages, COBOL and FORTRAN. The development of hard disk drives for data storage further spurred the development of databases. Around the same time, the first notion of a modern-day computer with a mouse and graphical user interface came into existence, making it easy for the general public to consume it. In this section, we will discuss how various types of databases evolved.

SQL

The first database was designed by Charles William Bachman III, an American computer scientist. In 1963, he developed the Integrated Data Store (IDS), which gave rise to the concept of the navigational database. In navigational databases, we can find records by chasing references from other objects. For example, let’s say that in a school database, you want to find all the students from a specific grade in a specific school. In a navigational database, first, you have to go to the group of students that belong to a particular school and then to the group that belongs to a particular grade. So, records can be accessed by hierarchical navigation. Based on IDS, Bachman later developed the CODASYL database model in 1969. CODASYL stands for Conference/Committee on Data Systems Languages, which was a consortium to guide the development of programming languages. Around the same time Edgar F. Codd, an IBM employee, developed the IBM Information Management System (IMS), which was based on the hierarchical database model. A hierarchical database model is a data model in which the data is designed in a tree-like structure. In 1970, Donald D. Chamberlin and Raymond F. Boyce developed Structured Query Language (SQL) based on what they’d learned about IMS. They initially called it Structured English Query Language (SEQUEL), which System R was later developed with by a group at the IBM San Jose research laboratory. In 1976, QUEL, which is a relational database query language designed by Michael Ralph Stonebraker, was developed as part of the Interactive Graphics Retrieval System (INGRES) database management system at the University of California, Berkeley.

Based on QUEL and SQL, several databases were implemented. Some of the most prominent ones include Post Ingres (Postgres), Sybase, Microsoft SQL, IBM DB2, Oracle, MariaDB, and MySQL.

Object-oriented databases

In the 1980s, object-oriented database systems (OODBMSes) grew in popularity. In OODBMSes, information is represented as objects compared to tables in relational databases. Some of the important ones include  Gemstone/S ,  Objectivity/DB ,  InterSystems Cache ,  Perst ,  ZODB ,  Wakanda ,  ObjectDB ,  ODABA , and  Realm .

NoSQL

The concept of non-SQL or non-relational databases has existed since the 1960s, but the term NoSQL became has much more popular in the last decade. NoSQL databases focus on performance and scaling and mostly rely on a non-relational data model such as a document, key-value, wide-column, or graph to organize the data. Some of the most popular ones in this category include Cassandra, MongoDB, Couchbase, Dynamo, FoundationDB, Neo4j, and Hbase.

NewSQL

With the introduction of the on-demand availability of compute, storage, and network resources and the pay-as-you-go model, which is collectively known as cloud computing, the amount of data that we collect, process, manage, and analyze has been growing exponentially. Although it was relatively easier for some of the NoSQL databases to adapt to the cloud, it is still much harder for traditional SQL databases to do so. Many of them are better suited for vertical scaling and do not consider geographically distributed data, the shared-nothing architecture, and enormous scale as part of their initial design. This created a void. We needed SQL databases that are cloud-native, scale well with data growth, and are easy to manage. Many companies developed in-house solutions on top of existing SQL databases:

- Facebook developed TAO, a NoSQL graph API built on top of sharded MySQL.

- YouTube developed Vitess to easily scale and manage MySQL clusters.

- Dropbox developed Edgestore, a metadata store to power their services and products, which again was built on top of MySQL.

- GreenPlum developed a massively parallel data platform by the same name for analytics, machine learning, and AI on top of Postgres.

However, it was still relatively harder and painful to manage the data as the underlying database was not built to scale.

In 2012, Google published a seminal paper on Google Spanner: a globally distributed database service and storage solution. Spanner essentially combined the important features of SQL databases such as ACID transactions, strongly consistent reads, and the SQL interface with some of the features that were only available with NoSQL databases, such as scaling across geographical locations, multi-site replication, and failover. It created a new category of databases called NewSQL, which is meant to indicate a combination of SQL features at NoSQL scale. YugabyteDB and CockroachDB were developed later, both of which got their inspiration from Google Spanner.

Database concepts

In this section, we will learn about some of the core database concepts, including cardinality, database models, and various processing models.

Cardinality

Before we discuss database models, it is important to know about cardinality. Cardinality refers to the relationship between two entities or tables. The most popular ones include one-to-many, many-to-one, and many-to-many.

One-to-one relationship

In the case of a one-to-one relationship, a row or entry in one entity or table can be related to only one row in another entity or table. For example, in a department of motor vehicles database, let’s say there are two tables called  License Info  and  Driver Info , as shown in the following diagram:

Figure 1.1 – An example of a one-to-one relationship

Here,  Driver ID  can only be assigned to one driver as it has to uniquely identify a driver. Also, a driver can only be assigned one  Driver ID . So, here, any row in the  License Info  table will be associated with a specific row in the  Driver Info  table.

One-to-many relationship

In a one-to-many relationship, a single row from one entity or table can be associated with multiple rows in another entity or table.

For example, let’s consider the  Driver Info  and  City Info  tables shown here:

Figure 1.2 – An example of a one-to-many relationship

Here, for every row in  City Info , there will be multiple rows in  Driver Info , as there can be many drivers that live in a particular city.

Many-to-many relationship

In a many-to-many relationship, a single row in one entity or table can be associated with multiple rows in another entity or table and vice versa.

For example, let’s consider two tables:  Vehicle Ownership History , where we are maintaining the history of ownership of a given vehicle, and  Driver Ownership History , where we are maintaining the history of vehicles owned by a given driver:

Figure 1.3 – An example of a many-to-many relationship

Here, a driver can own multiple vehicles and a vehicle can have multiple owners over time. So, a given row in the  Vehicle Ownership History  table can be associated with multiple rows in the  Driver Ownership History  table. Similarly, a given row in the  Driver Ownership History  table can be associated with multiple rows in the  Vehicle Ownership History  table.

Now, let’s take a look at some of the most important database models.

Overview of database models

A database model determines how the data is stored, organized, and modified. Databases are typically implemented based on a specific data model. It is also possible to borrow concepts from multiple database models when you are designing a new database. The relational database model happens to be the most widely known as has been popularized by databases such as Oracle, IBM DB2, and MySQL.

Hierarchical database model

In the hierarchical database model, the data is organized in the form of a tree. There is a root at the first level and multiple children at the subsequent levels. Since a single parent can have multiple children, one-to-many relationships can easily be represented here. A child cannot have multiple parents, so this results in the advantage of not being able to model many-to-many relationships.

IBM’s Information Management System (IMS) was the first database that implemented this data model.

The following diagram shows an example of a hierarchical database model:

Figure 1.4 – An example of a hierarchical database model

Typically, the tree starts with a single root and the data is organized into this tree. Any node except the leaves can have multiple children, but a child can have only one parent.

Network model

The network model was developed as an enhancement of the hierarchical database model to accommodate many-to-many relationships. The network model relies on a graph structure to organize its data. So, there is no concept of a single root, and a child can have multiple parents and a parent can have multiple children. Integrated Data Store (IDS), IDMS (Integrated Database Management Systems), and Raima Database Manager (RDM) are some of the popular databases that use the network model.

As shown in the following diagram, there is no single root and a given child (for example, Object 2 can have multiple parents; that is, Object 1 and Object 3):

Figure 1.5 – An example of a network model

Relational model

Although the network model was an improvement over the hierarchical model, it was still a little restrictive when it came to representing data. In the relational model, any record can have a relationship with any other with the help of a common field. This drastically reduced the design’s complexity and made it easier to independently add, update, and access records, without having to walk down the tree or traverse the graph. SQL was combined with the relational database model to provide a simple query interface to add and retrieve data.

All the popular traditional databases such as Oracle database, IBM DB2, MySQL, MariaDB, and Microsoft SQL Server implement relational data models.

Let’s look at two tables called  Employee  and  Employee Info :

Figure 1.6 – Employee tables showing the column names

Here,  Employee ID  is the common field or column between the  Employee  and  Employee Info  tables. The  Employee  table is responsible for ensuring that a given  Employee ID  is unique, while  Employee Info  is responsible for more detailed information about a given employee.

Object-relational model

The object-relational model, as the name suggests, combines the best of the relational and object data models. The concept of objects, classes, and inheritance are directly supported as first-class citizens as part of the database and in queries. SQL:1999, the fourth revision of SQL, introduced several features for embedding object concepts into the relational database. One of the main features was to create structured user-defined types with  CREATE TYPE  to define an object’s structure.

Over time, relational databases have added more support for objects. There is a varying degree of support for object concepts in Oracle database, IBM DB2, PostgreSQL, and Microsoft SQL Server.

Given the scope of this book, we will not discuss the Entity-relational model, Object model, Document model, Star Schema, Snowflake Schema, and many other less well-known models.

Now, let’s look at how databases can be classified based on what kind of workloads they can be used for.

Processing models

Based on how you want to consume and process data, databases can be categorized into four different processing systems. Let’s take a look.

Online transaction processing (OLTP)

OLTP systems support the concept of transactions. A transaction refers to the ability to atomically apply changes (insert, update, delete, and read) to a given system. One popular example is a bank, where withdrawing or depositing money to a given bank account must be done atomically to ensure data is not lost or incorrect. So, the main purpose here is to maintain data integrity and consistency. Also, these systems are generally suited for fast-running queries.

Online analytical processing (OLAP)

OLAP focuses mostly on running queries to analyze multi-dimensional data and to extract some intelligence or patterns from it. Typically, such systems support generating some sort of report that can be used for marketing, sales, financing, budgeting, management, and many more. Data mining and data analytics applications would typically have to have an OLAP system in some form. OLAP doesn’t deal with transactions, and emphasis is more on analyzing large amounts of data from different sources to extract business intelligence. Some databases also provide built-in support for MapReduce to run queries across a large set of data.

A data warehouse is a piece of software that’s used for reporting and data analysis. Warehouses are typically developed for OLAP. It is also very common to retrieve the data from OLTP in batches or bulk, run it through an Extract, Load, and Transform (ELT) or Extract, Transform, and Load (ETL) data transformation pipeline, and store it in an OLAP system.

Online event processing (OLEP)

OLEP guarantees strong consistency without the traditional atomic commit protocols or distributed locking. OLEP also focuses on high performance, larger scales, and fault tolerance.

Hybrid transaction/analytical processing (HTAP)

As the name suggests, this system tries to provide the best of both transactions and analytical processing. Most of the NoSQL and NewSQL databases provide support for managing both transactional and analytical workloads.  Vitess  is a database clustering system that can be used to scale and shard MySQL instances.  Vitess  provides HTAP features on top of MySQL by allowing a given MySQL instance to be configured as master or read-only, where read-only can be used for analytical queries and  MapReduce . It is possible to use CockroachDB as HTAP by propagating changes with the help of change data capture (CDC) in the OLTP cluster or primary cluster to a separate cluster, which is solely used for analytical processing.

Now, let’s learn a bit about embedded and mobile databases, including why they exist and some of the most popular ones in this space.

Embedded and mobile databases

Embedded databases usually refer to databases that can be tightly integrated into an application, without needing separate hardware to support them. Also, they don’t have to be managed separately. Some of the most popular embedded databases include SQLite, Berkeley DB from Oracle Corporation, and SQL Server Compact from Microsoft Corporation. Embedded databases are also very useful for testing purposes as they can be started within test suites.

Mobile database refers to the class of databases that work with very limited memory footprint and compute and can be deployed within a mobile device. They are typically used for storing user data for apps running on mobile devices. SQLite, SQL Server Compact, Oracle database Lite, Couchbase Lite, SQL Anywhere, SQL Server Express, and DB2 Everyplace belong to this category,

Database storage engines

A database storage engine is a component within a database management system that is responsible for Create, Read, Update, Delete (CRUD) operations and transferring data between disk and memory, without compromising data integrity. Some of the most popular ones include Apache Derby, HSQLDB, InfinityDB, LevelDB, RocksDB, and SQLite. CockroachDB initially started with RocksDB as its database engine, but from release 20.2 onward, Pebble will be the database engine by default. Pebble, as per Cockroach Labs, is a RocksDB inspired and RocksDB compatible key-value store focused on the needs of CockroachDB. RocksDB was implemented in C++, whereas Pebble has been implemented in Golang. This makes it easier to manage and maintain as CockroachDB itself has been written in Golang. This means that we only have to deal with one language now.

CAP theorem

Eric A. Brewer gave a keynote talk in 2000 titled Towards Robust Distributed System at a symposium on Principles of Distributed Computing, summarizing his years of learning about distributed systems. Brewer talked about key aspects of a distributed system: consistency, availability, and tolerance toward network partition. Consistency refers to the fact that every read should see the data from the most recent write; otherwise, it should error out. Availability means every requested read or write should receive a non-error response. Partition tolerance indicates that the system should continue to serve, irrespective of delays and communication failures between nodes in the system. CAP theorem claims that, at most, you can only have two of these three properties in a distributed system.

Consistency and partition tolerance (CP)

A CP database provides consistency and partition tolerance but cannot provide availability. This is also called a CAP-consistent system. Let’s understand this by looking at an example:

Figure 1.7 – Consistency and partition tolerance (CP) system

Let’s consider the system shown in the preceding diagram, where two servers are serving read and write traffic. For this example, let’s say writes only land on Server 1 and reads only land on Server 2. So long as Server 1 can talk to Server 2, all the writes that come to Server 1 can be propagated synchronously to Server 2. This ensures that any reads that come to Server 2 are always consistent, which means they see the latest data written by the latest write in Server 1:

Figure 1.8 – Consistency and partition tolerance (CP) system

Now, let’s say that, as shown in the preceding diagram, the communication between Server 1 and Server 2 has broken down and now Server 1 is no longer able to propagate the writes synchronously. This results in partitioning. Since the data cannot be propagated between the two servers, read or write traffic cannot be served until we resolve the partition issue as we have to ensure data consistency.

Some of the most popular databases that have CP characteristics are HBase, Couchbase, and MongoDB. CockroachDB also falls into this category.

Availability and partition tolerance (AP)

In this case, a database is guaranteed to always be available and it can tolerate partitioning, but at the cost of consistency. This is also known as a CAP-available system. Here, the application is expected to deal with data consistency:

Figure 1.9 – Availability and partition tolerance (AP) system

Similar to the previous example, if the communication between Server 1 and Server 2 breaks down, Server 1 and Server 2 continue to serve the traffic but reads to Server 1 and Server 2 might return different versions of the data, based on when the communication has failed and whether there was any change to that data, after the communication failure. Cassandra, Riak, and CouchDB are popular examples of AP databases.

Consistency and availability (CA)

In the case of a CA database, the system cannot tolerate partitioning but can guarantee consistency and availability. Traditional databases with single-server deployments with no replication or slaves can be classified as CA. Now, many traditional RDMBS databases can be configured in various ways to have CA, CP, or AP as desired.

CockroachDB

The name CockroachDB was inspired by the insect that goes by the same name. Just like how cockroaches have been surviving for millions of years and colonizing the entire planet and thriving, CockroachDB instances are supposed to replicate and repair data, spread naturally across multiple availability zones, and survive total regional failures. Also, once CockroachDB becomes part of a given software ecosystem, it’s impossible to get rid of or replace it, just like cockroaches. Here, we will discuss why there is a need for yet another database, known as Inspiration, and provide a high-level overview of CockroachDB.

Why yet another database?

As more companies shift from on-premises to the cloud, they are looking for SQL datastores on various cloud platforms to manage their transactional data. Most of the traditional databases such as MySQL, Postgres, and Oracle are not built for the cloud. This necessitates the need for a cloud-native, consistent, distributed SQL that can scale with the growth of data. CockroachDB fills this gap.

Inspiration

As we previously discussed in the NewSQL section, in 2012, Google published a seminal paper on Google Spanner: a globally distributed database service and storage solution. Although Google Spanner combined the best of both SQL and NoSQL and was very useful for a lot of applications, it was not available for public usage. Also, Google Spanner was and still is not an open source project and has only been available on Google Cloud Platform since 2017. So, this created a necessity for an open source Spanner-like database that can be used in different cloud providers and on-premises. Around 2012, Spencer Kimball, Peter Mattis, and Ben Darnell were working at Google on the Google File System and Google Reader projects. They also got acquainted with both Bigtable and Spanner during their tenure at Google. They decided to build something very similar to Spanner to make it available for everyone and started an open source project on GitHub in 2014. After a year, they decided to leave Google and founded Cockroach Labs in 2015 before officially working on CockroachDB in June 2015.

Key terms and concepts

Before we look at the various functional layers, let’s look at some of the key concepts and terms. A CockroachDB cluster refers to a group of nodes that act as a single logical unit. A node is a single machine that runs an instance of CockroachDB. CockroachDB stores all the data as sorted key-value pairs. These keys are divided into ranges. CockroachDB replicates each range and stores each replica on a different node. For each range, there will be a leaseholder, which acts as a primary owner of a given range and receives and coordinates all the traffic for that range. For each range, one of the replicas acts as a leader for write requests and ensures that the majority of the replicas are in consensus, before committing a given write. For each range, there will be a time-ordered log of writes, called a raft log, for which the majority of replicas agreed upon.

High-level overview

CockroachDB is a cloud-native, consistent, highly scalable relational database. Some of the primary goals of CockroachDB are to provide strong consistency, geo-distribution of data, high availability, SQL support, easy deployment, and less maintenance. Since we will be dealing with CockroachDB internals in detail in subsequent chapters, we will just provide a high-level overview here:

Figure 1.10 – Consistency and partition tolerance (CP) system

CockroachDB exposes a SQL interface, using which clients can interact with the database. Client requests can land on any node within a given cluster and work just fine since all the nodes are symmetrical.

CockroachDB can be divided into five functional layers:

- SQL

- Transactional

- Distribution

- Replication

- Storage

The SQL layer is responsible for receiving SQL queries and converting them into key-value operations. The transactional layer ensures that all CRUD operations that happen on multiple key-value pairs are transactional. The distribution layer is responsible for ensuring ranges are evenly distributed among all the available nodes in a cluster. The replication layer ensures that ranges are replicated synchronously, whenever there is a change. Finally, the storage layer is responsible for managing key-value data on the disk.

Summary

In this chapter, we learned about the evolution of databases, how databases can be categorized based on various criteria, CAP theorem, and a brief introduction to CockroachDB. By now, you should also be familiar with database and processing models, what the CP, CA, and AP systems in CAP theorem offer, and the functional layers of CockroachDB.

In the next chapter, we will take a deep dive into CockroachDB’s architecture and design concepts.

2 How Does CockroachDBWork Internally?

In the previous chapter, we have learned about the evolution of databases and the high-level architecture of CockroachDB. In this chapter, we will go a bit deeper into each of the layers of CockroachDB and explore how CockroachDB works internally. We will also discuss some of the core design aspects that form the basic pillars of CockroachDB.

CockroachDB can be broadly divided into five main layers, as outlined here:

- Structured Query Language (SQL)

- Transactional

- Distribution

- Replication

- Storage

Each of these layers will be explained as the main topics of this chapter, in the following order:

- Installing a single-node CockroachDB cluster using Docker

- Execution of a SQL query

- Managing a transactional key-value store

- Data distribution across multiple nodes

- Data replication for resilience and availability

- Interactions with the disk for data storage

Since we will be trying out some commands in this chapter, it’s important to have a working environment for the same. So, we will start with the technical requirements, where we will go over how to set up a single-node CockroachDB cluster.

Technical requirements

To try out some of the commands in this chapter, you will need a single-node CockroachDB cluster. There are several ways of installing CockroachDB on a computer, as outlined here:

- Use a package manager such as  Homebrew  to install it, but this option only works on a Mac

- Download binaries, extract it, and set it in the  PATH  variable

- Use Kubernetes to orchestrate CockroachDB pods

- Build from source and install

- Download a Docker image and run it

We will cover various ways of installing and running CockroachDB on your computer in Chapter 9, Various Deployment Strategies, where we will discuss various deployment strategies at length. In the current chapter, we will just go over how to run CockroachDB using Docker since the steps are common, irrespective of the operating system that you are using.

For using CockroachDB with Docker, you need a computer with the following:

- At least 4 gigabytes (GB) of random-access memory (RAM)

- 250 GB of disk space

- Docker installed

In the next section, we will learn about installing CockroachDB.

Installing a single-node CockroachDB cluster using Docker

Let’s take a look at how to install a single-node CockroachDB cluster, which will be required to try out some of the commands that will be introduced in this chapter. Here are the steps you need to follow:

1. Ensure the Docker daemon is running with the following command:

docker version

Pull the most recent stable version with  tag v<xx.y.z>  from https://hub.docker.com/r/cockroachdb/cockroach/. Take a look at the following example:

docker pull cockroachdb/cockroach:v20.2.4

2. Make sure this image is available and the version is correct with the help of the following command:

docker images | grep cockroach

cockroachdb/cockroach   v20.2.4             d47481b0b677        2 days ago          329MB

If running docker on windows, then replace grep with findstr

docker images | findstr cockroach

3. Create a bridge network. A bridge network allows multiple containers to communicate with each other. Here’s the code you’ll need to create one:

$ docker network create -d bridge crdb_net

4. Create a volume. Volumes are used for persisting data generated and used by Docker containers. Since CockroachDB is a database, it needs a place to store the data, and hence we should attach a volume to the container. You can do this by running the following command:

$ docker volume create crdb_vol1

5. Start a CockroachDB node using the following command:

$ docker run -d \

--name=crdb1 \

--hostname=crdb1 \

--net=crdb_net \

-p 26257:26257 -p 8080:8080  \

-v "crdb_vol1:/cockroach/cockroach-data"  \

cockroachdb/cockroach:v20.2.4 start \

--insecure \

--join=crdb_vol1

Here is an explanation of what each of these options means:

- docker run : This starts a Docker container.

- --name : Name of the container.

- --hostname : Hostname of the container. This will be useful if you are running multiple containers and want to join them to form a cluster.

- --net : Bridge network. This will be useful if you have more than one container that wants to communicate with other containers.

- -p 26257:26257 : Port mapping for inter-node or SQL client for node communication.

- -p 8080:8080 : Port mapping used for HyperText Transfer Protocol (HTTP) requests to the CockroachDB console.

- -v "crdb_vol1:/cockroach/cockroach-data"  :  Mounts the host directory as a data volume.

- cockroachdb/cockroach:v20.2.4 start : Command to start the CockroachDB node.

- --insecure : Option to start the node in insecure mode.

- --join=crdb_vol1 : Here, you can specify multiple hostnames of CockroachDB nodes that will form a cluster. For the current chapter, we just need a single-node cluster.

6. Initialize the cluster with the following command:

$ docker exec -it crdb1 ./cockroach init --insecure

Cluster successfully initialized

7. To ensure that the CockroachDB node is functional, we can create a database and a table, insert some data, and run a query.The following command is used for starting the SQL shell:

docker exec -it crdb1 ./cockroach sql –insecure

The following command is used for creating a database:

root@:26257/defaultdb> CREATE DATABASE testdb;

The following command is used for creating a table:

root@:26257/defaultdb> CREATE TABLE testdb.testtable (id INT PRIMARY KEY, string name);

8. Verify the columns by running the following code:

root@:26257/defaultdb> SHOW COLUMNS from testdb.testtable;

column_name | data_type | is_nullable | column_default | generation_expression |  indices  | is_hidden

--------------+-----------+-------------+----------------+-----------------------+-----------+------------

id          | INT8      |    false    | NULL           |                       | {primary} |   false

string      | NAME      |    true     | NULL           |                       | {}        |   false

(2 rows)

Time: 134ms total (execution 133ms / network 1ms)

9. Insert some data, as follows:

root@:26257/defaultdb> INSERT INTO testdb.testtable VALUES (1,'Spencer Kimball'), (2,'Ben Darnell'), (3,'Peter Mattis');

Run a query to fetch the contents of the table

root@:26257/defaultdb> SELECT * FROM testdb.testtable;

id |     string

-----+------------------

1 | Spencer Kimball

2 | Ben Darnell

3 | Peter Mattis

(3 rows)

Time: 5ms total (execution 3ms / network 2ms)

Next, we will learn about each of the layers, starting with the SQL layer.

Execution of a SQL query

Any application that talks to CockroachDB can use Postgres-compatible drivers to talk to CockroachDB. If you prefer object-relational mappers (ORMs), Golang ORM (GORM),  go-pg , and SQLBoiler can be used to interact with CockroachDB. Irrespective of where the actual data resides, you can issue a query to any of the nodes in the cluster. Whichever node the query lands in, that node acts as a gateway.

Requests from SQL clients are received as SQL statements. The SQL layer is responsible for converting these SQL statements into a plan of key-value operations and passing it to the transaction layer. CockroachDB, as of version 21.2, supports both native drivers and the Postgres wire protocol. A wire protocol defines how two applications can communicate over a network.

CockroachDB has support for most American National Standards Institute (ANSI) SQL standards to change table structures and data. Next, we will look at the various stages of query execution.

SQL query execution

As with any other database, there are standard steps involved in processing incoming SQL requests and serving the data.

Before going through the different steps of a SQL layer, it’s important to get ourselves familiar with some terms related to SQL, as follows

- An SQL parser is software that scans a given SQL statement in its string form and tries to make sense of it. This involves lexical analysis, which is extracting the tokens or the keywords, and syntactic analysis, where you make sure the entire query is valid and can be represented in a form that makes it easier to execute the query. Lex is a popular lexical analyzer and Yacc (which stands for Yet Another Compiler-Compiler) is a software that processes grammar and generates a parser.

- An abstract syntax tree (AST) or a syntax tree is a tree representation of a given domain-specific language. Nodes in the tree represent constructs in that language. ASTs are essential for deciding on how to execute a given query.

Some of the main steps in a SQL layer include query parsing, logical planning, physical planning, and query execution. Vectorized query execution is enabled by default, which gives much better performance.

Parsing

CockroachDB uses Goyacc, Golang’s equivalent of the famous C Yacc to generate a SQL parser from a grammar file located at  pkg/sql/parser/sql.y . This parser then converts the input into an AST comprising of tree nodes, where the node types are defined under the  pkg/sql/sem/tree  package.

Logical planning

Once we have an AST, it then has to be transformed into a logical plan. A logical plan defines how various clauses can be logically ordered. As part of this transformation, semantic analysis is initially done for type checking, name resolution, and to ensure the query is valid. Later, this logic plan is simplified without changing the overall semantics and is optimized based on the cost. Here, cost refers to the total time taken to return the results of a query. Cost optimization involves picking the right indexes, query optimization, and selecting the best strategies for sorting and joining. We can view the logical plan using  EXPLAIN <SQL Statement> .

The following example shows the output of a sample  EXPLAIN  query:

root@:26257/defaultdb> EXPLAIN SELECT * FROM testdb.testtable;

tree |        field        |    description

-------+---------------------+--------------------

| distribution        | full

| vectorized          | false

scan |                     |

| estimated row count | 1

| table               | testtable@primary

| spans               | FULL SCAN

(6 rows)

Time: 11ms total (execution 10ms / network 1ms)

Physical planning

In this phase, all the participating nodes are determined based on where the data resides and also who is the primary owner for a given range, which is also known as a range’s leaseholder. Typically, partial result sets are gathered from various nodes that are then sent to the coordinator or the gateway node, which aggregates all the results and sends a single response back to the SQL client. Let’s look at an example of what this looks like for a sample query for the table we created previously in the Technical requirements section.

Let’s assume we have six names that are distributed across four nodes in a cluster. The following table shows information about the leaseholder nodes, key ranges, and user data:

Figure 2.1 – Table showing information about key ranges

Now, let’s assume that a SQL client sends a query to fetch all the names from the  testdb.testtable  table. In this example, this request lands in node 1, so it acts as a gateway node that is responsible for collecting the data required to serve this request after coordinating with all relevant leaseholder nodes, getting a partial result, aggregating the data, and sending it back to the SQL client. This process is illustrated in the following screenshot:

Figure 2.2 – How a given query is served from the gateway node

At Gateway Node, the main challenge is to identify whether a given computation should be pushed down to nodes where the data resides or to do the same computation within the coordinator node on an aggregated result. The physical plan aims to make the best use of parallel computing and at the same time reduce the overall data that gets transferred between data and coordinator nodes. Also, table joins bring in an added layer of complexity. We can view the physical plan using  EXPLAIN(DISTSQL) <SQL statement> .  DISTSQL  generates a Uniform Resource Locator (URL) for a physical query plan, which includes high-level information about how the query will be executed.

The following example shows the output of a sample  EXPLAIN(DISTSQL)  query:

root@:26257/defaultdb> EXPLAIN(DISTSQL) SELECT * FROM testdb.testtable where id > 1;

automatic |                                               url

------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

false   | https://cockroachdb.github.io/distsqlplan/decode.html#eJyMj0FLxDAQhe_-ivBOKtFt9ZaTohULdXdtCwraQ7YZlkK3qZkUlNL_Lm1A8SDsKbz3Mu-bGcEfLRSS1212m67F6X1alMVzdiaKJEvuSnEuHvLNk_DE3uwu58frXUvi5THJE9EY8T5E0TWJGBKdNbTWB2KoN8SoJHpna2K2brbG5UNqPqEiiabrBz_blURtHUGN8I1vCQrljMhJG3KrCBKGvG7apfZng5veNQftviBR9LpjJVZXF6gmCTv432r2ek9Q8SSPx-fEve2Y_pD_a46mSoLMnsKJbAdX09bZesEEuVnmFsMQ-5DGQaRdiKZqOvkOAAD__6E0gSw=

(1 row)

Time: 21ms total (execution 16ms / network 4ms)

As you can see, the output is a downloadable URL for the actual physical plan. If you click on that URL, it will take you to the physical plan, as shown in the following screenshot:

Figure 2.3 – Visual representation of a physical plan in a single-node cluster

The  EXPLAIN ANALYZE  statement (https://www.cockroachlabs.com/docs/v21.1/sql-statements) executes a SQL query and generates a statement plan with execution statistics, as illustrated in the following code snippet:

root@:26257/testdb> EXPLAIN ANALYZE SELECT * FROM testdb.testtable;

automatic |                                               url

------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

true    | https://cockroachdb.github.io/distsqlplan/decode.html#eJyMUE1L60AU3b9fMZzVe4-xNhZdzMqqEQKxrU0XfpDFNHMpgSQT596ipeS_SxJUXAiuhvMx5xzuEfxSwSB-WKXzZKHmi3n6-BSrvzdJtsnu038qi9P4eqP-q9v18k4JsbjtpH_EbiuCRuMdLWxNDPOMCLlGG3xBzD701HEwJO4NZqpRNu1eejrXKHwgmCOklIpgsOkD12QdhdMpNByJLash9rPvsg1lbcMBGllrGzbqBBrBv7IKZJ1RM2iw2KpSUtZkVDSZXcxqhsb2IPThis7O1RXyTsPv5WsRi90RTNTp369eE7e-Yfo2-KfkaZdrkNvReBn2-1DQKvhiqBnhcvg3EI5YRjUaQdKMUpd3f94DAAD__yWSjvw=

(1 row)

Time: 4ms total (execution 3ms / network 1ms)

The output is shown here:

Figure 2.4 – Statement plan with execution statistics

EXPLAIN ANALYZE (DEBUG)  executes a query and generates a link to a ZIP file that contains the physical statement plan (https://www.cockroachlabs.com/docs/stable/explain-analyze.html#distsql-plan-viewer), execution statistics, statement tracing, and other information about the query. The code is illustrated in the following snippet:

root@:26257/testdb> EXPLAIN ANALYZE (DEBUG) SELECT * FROM testdb.testtable;

text

--------------------------------------------------------------------------------

Statement diagnostics bundle generated. Download from the Admin UI (Advanced

Debug -> Statement Diagnostics History), via the direct link below, or using

the command line.

Admin UI: http://crdb1:8080

Direct link: http://crdb1:8080/_admin/v1/stmtbundle/676836398188756993

Command line: cockroach statement-diag list / download

(6 rows)

Time: 92ms total (execution 91ms / network 1ms)

Query execution

During query execution, the physical plan is pushed down to all the data nodes that would be involved in serving a given query. CockroachDB uses work units called logical processors, which will be responsible for executing relevant computations. Logical processors across data nodes also communicate with each other so that data can be sent back to the coordinator or the gateway node.

There are two types of query execution, as outlined here:

- Non-vectorized or row-oriented query execution

- Vectorized or column-oriented query execution

Vectorized query execution (column-oriented query execution) is more suited to analytical workloads, whereas non-vectorized query execution (row-oriented query execution) is preferred for transactional workloads. By default, vectorized query execution is enabled on CockroachDB.

Now, let’s take a look at how CockroachDB provides a transactional key-value store.

Managing a transactional key-value store

A transactional layer involves implementing a concurrency control protocol, as multiple transactions can try to update the same data at the same time, which can result in a conflict. In concurrency control, there are two different ways of dealing with conflicts, as outlined here:

- Avoid conflicts altogether with pessimistic locking—for example, a read/write lock.

- Let the conflict happen but detect it with optimistic locking and resolve it—for example, multi-version concurrency control (MVCC).

CockroachDB uses MVCC. In MVCC, there can be multiple versions of the same record, but you resolve the conflicts before committing the changes.

In CockroachDB, a given transaction is executed in three phases, outlined next:

1. A transaction is started with a target range that will participate in the transaction. A new transaction record is created to track the status of the transaction. It will have the initial state as  PENDING . At the same time, a write intent is created. In CockroachDB, instead of directly writing the data to the storage layer, data is written to a provisional state called  write intent . Here, the  intent  flag indicates that the value will be committed once the transaction is committed. CockroachDB uses MVCC for concurrency control. The transaction identifier (ID) is used to resolve conflicts for write intents. Each node involved in the transaction returns the timestamp used for the write, and the coordinator node selects the highest timestamp among all write timestamps and uses it in the final commit timestamp.

2. The transaction is marked as committed by updating its transaction record. The  commit  value also contains the candidate timestamp. The candidate timestamp is a temporary timestamp to denote when the transaction is committed and is selected as the actual node coordinating the transaction. Once the transaction is completed, control is returned to the client.

3. After the transaction is committed, all written intents are updated in parallel by removing the  intent  flag. The transaction coordinator does not wait for this step to be completed before returning the control to the client.

We will learn more about conflict resolution, atomicity, consistency, isolation, and durability (ACID), logical clocks, and transaction management in the next chapter. Next, we will go over the distributed layer.

Data distribution across multiple nodes

A table in CockroachDB can be partitioned, and this is discussed in Chapter 4, Partitioning Data Across Multiple Geo-Locations, where we talk about geo-partitioning. CockroachDB stores the data in a monolithic sorted key-value store (MSKVS). Key-space is all the data you have in a given cluster, including information about its location. Key-space is divided into contiguous batches, called ranges. The MSKVS makes it easy to access any data from any node, which makes it possible for any node in the cluster to act as a gateway node, coordinating one or more data nodes while serving client requests.

MSKVS

MSKVS contains two categories of data, as outlined here:

1. System data, which contains meta ranges, where the data of each range can be found within the cluster.

2. User data, which is the actual table data.

Meta ranges

The location of ranges is maintained in two-level indexes, known as meta ranges. The first level (a.k.a.  meta1 ) points to the second level (a.k.a.  meta2 ) and the second-level indexes point to the actual data. This is shown in the following screenshot:

Figure 2.5 – Meta-range management in two levels

Every node in the cluster has complete information of  meta1  and that range is never split.  meta2  data is cached on nodes. These are invalidated whenever ranges change, and the cache gets updated with the latest value.

Let’s look at an example.

Here, we will understand what  meta1  and  meta2  data looks like for an alphabetically sorted column. When we write ranges, square brackets ‘ [ ‘ and ‘ ] ’ indicate that the number is included in the range, and parentheses ‘ ( ‘ and ‘ ) ’ mean that the number is excluded.

Let’s look at some examples here:

- [1,10] —range starts from  1  and ends at  10  as both numbers are included

- [1,5) —range starts from  1  and ends at  4 , as  5  is excluded

- (1, 8) —range starts from  2 , as  1  is excluded and it ends at  7 , as  8  is excluded

Let’s now understand what  meta1  and  meta2  look like for an alphabetically sorted column, as follows:

1. meta1  contains addresses of nodes that contain  meta2  replicas. Let’s assume there are two  meta1  entries for simplicity. The first  meta1  entry points to the  meta2  range for keys  [A-M) , and the second  meta1  entry points to the  meta2  range for keys  [M-Z] . Here,  maxKey  indicates the rest of the range till the maximum available key. Since we are talking about an alphabetically sorted column, that would start at  M , as the previous range excluded  M  and ends at  Z , which is the last letter of the alphabet. The code is illustrated here:

meta1/M -> node1:26257, node2:26257, node3:26257

meta1/maxKey -> node4:26257, node5:26257, node6:26257

2. meta2  contains addresses of the nodes containing actual user data for that alphabetically sorted column. The first entry in the value always refers to the leaseholder, which is the primary owner of a given range. The code is illustrated here:

meta2 entry for the range [A-G)

meta2/G -> node1:26257, node2:26257, node3:26257

meta2 entry for the range [G-M)

meta2/M -> node1:26257, node2:26257, node3:26257

meta2 entry for the range [M-Z)

meta2/Z -> node4:26257, node5:26257, node6:26257

meta2 entry for the range [Z-maxKey)

meta2/maxKey->  node4:26257, node5:26257, node6:26257

Table data

When a new table is created, the table and its secondary indexes all point to a single range. Once the range size exceeds 512 megabytes (MB), the range is split into two. This continues as the data grows. Table ranges are replicated to multiple nodes for survivability so that even if some nodes in the cluster shut down or crash, there will not be any data loss.

Next, we will learn about replication and Raft, a distributed consensus algorithm.

Data replication for resilience and availability

This layer is responsible for ensuring that the table data is replicated to more than one node and also keeps the data consistent between replicas.

The replication factor indicates how many replicas of a specific table’s data should be kept—for example, if the replication factor is 3, CockroachDB keeps three copies of all the table data. The number of node failures that can be tolerated without data loss = (replication factor – 1) / 2; for example, if the replication factor is 3, then (3 – 1) / 2 = 1 node failure can be tolerated. Whenever a node goes down, CockroachDB automatically detects it and works toward making sure the data in the node that went down is replicated to other nodes, in order to honor the replication factor and also to increase survivability.

CockroachDB uses the Raft distributed consensus algorithm, which ensures a quorum of replicas agree on changes to ranges before those changes are committed.

What is consensus?

Consensus is a concept in distributed systems that is used for fault-tolerance and reliability when some nodes either go down or will not be reachable because of network issues. Consensus involves multiple nodes agreeing to changes before they are committed. If all the nodes involved in the consensus are not available, then an agreement can still be made, as long as a majority of the nodes are available–for example, in a cluster of nine nodes, we need at least five nodes that are able to communicate with each other in order to reach an agreement.  Paxos ,  Multi-Paxos ,  Raft , and  Blockchain  are some of the popular consensus algorithms.

Raft distributed consensus protocol

The word Raft is supposed to be the combination of R (which stands for reliable, replicated, and redundant), A (which stands for and), and FT (which stands for fault tolerance). Although it’s not an acronym, the word Raft is supposed to be a system that provides reliability, replication, redundancy, and fault tolerance.

In Raft, all nodes that have a replica of a given range will be part of a Raft group. Each node can be in one of the following states:

- Leader—Acts as a leader of the Raft group. Responsible for managing data mutations and ensuring that data is consistent between the leader and its followers using log replication.

- Follower—Follows a leader and works with the leader in order to keep the data consistent.

- Candidate—In the absence of a clear leader, any participating node can try to become a leader. A node that is trying to become a leader is called a candidate.

There are mainly two types of remote procedure call (RPC) requests, as listed here:

- RequestVote —Used for requesting votes by candidate node to other participating nodes

- AppendEntries —Used for log replication and heartbeat

Let’s now understand how leader election happens within Raft.

Leader election

Initially, all nodes of a Raft group start as followers. If they don’t hear from a leader, they become a candidate. The candidate votes for itself and requests votes by sending a  RequestVote  message to other participating nodes. Any candidate with a majority of votes becomes the leader. This process is called leader election. If two nodes end up with the same number of votes, then there will be re-election. Also, election timeout is randomized among the nodes of a Raft group, which ensures each participating node becomes a candidate at different points of time. This reduces the chance of a split vote.

After the election, the leader keeps sending heartbeats through an  AppendEntries  message to all its followers, and the followers keep responding. This ensures that the election term is maintained.

There will be re-election when one or more nodes don’t hear from the leader for a certain period of time. This is called election timeout.

Log replication

Once a leader is elected, all changes to the data go through the leader. Every change is first recorded in the leader node’s log. The actual node’s value is not updated till the change is committed. In order to commit the change, the leader first broadcasts it via an  AppendEntries  message to all its followers. After this, the leader waits for a majority of the nodes to reply back. The followers respond back once they make an entry in their own logs. Once the leader receives a response from a majority of the nodes, it then commits the change and modifies the node value. After this, the leader notifies all the followers that the change is committed.

Let’s take a look at an example of how changes go through the leader and are replicated to all the nodes in a Raft group. In this example, there are three nodes in a Raft group. Node 2 is the leader, and Node 1 and Node 3 are followers, as shown in the following screenshot:

Figure 2.6 – Raft group with Node 2 as the leader and node value set to 10

Let’s look at the changes through the following steps:

1. Now, a client makes a request to the Raft leader to change the value from 10 to 30, as illustrated in the following screenshot:

Figure 2.7 – Client requesting the node leader to change the value from 10 to 30

2. When the leader receives this request, it appends this new entry of changing the value from 10 to 30 in its replication log, as illustrated in the following screenshot:

Figure 2.8 – Node 2 appends the entry to change the value from 10 to 30 in its replication log

3. After making an entry in its replication log, the same is broadcasted to all its followers, as illustrated in the following screenshot:

Figure 2.9 – Node 2, the leader, broadcasts this entry of changing the value from 10 to 30 to all its followers

4. All the followers now make an entry in their replication log and acknowledge to the leader that they have done so, as illustrated in the following screenshot:

Figure 2.10 – Followers acknowledge back after they append the new entry in their own replication logs

5. Once the leader receives an acknowledgment from a majority of the followers, it commits this entry locally and changes the actual node value from 10 to 30, and later broadcasts to all its followers that the entry is committed. This is illustrated in the following screenshot:

Figure 2.11 – Once a majority of the followers acknowledge, Node 2, the leader, commits it locally by changing the actual node value from 10 to 30 and later sends a notification to all its followers that the entry is committed

6. In the last step, all the followers also commit the entry by changing their node values from 10 to 30, as illustrated in the following screenshot:

Figure 2.12 – All the followers acknowledge back after they append the new entry in their own replication logs

This completes the process of log replication for a single change. This process is replayed for all the changes.

If the leader crashes during this negotiation, the replication log can be in an inconsistent state. The new leader fixes this inconsistency by forcing its followers to replicate its own log. Specific details on how this is done are beyond the scope of this book.

Next, we will understand how CockroachDB interacts with the disk.

Interactions with the disk for data storage

The storage layer is responsible for reading from and writing to the disk. Each node in a CockroachDB cluster should have at least one storage attachment. Data is stored as key-value pairs on the disk using a storage engine.

Storage engine

A database management system (DBMS) uses a storage engine to perform CRUD (which stands for create, read, update, and delete) operations on the disk. Usually, the storage engine acts as a black box, so you get more options to choose from based on your own requirements, and also, storage engines evolve independently of the DBMSs that are using them.

Storage engines use a variety of data structures to store the data. The popular ones are listed here:

- Hash table

- B+ tree

- Heap

- Log-structured merge-tree (LSM-tree)

Storage engines also work with a wide range of storage types, including

- Solid-state drive (SSD)

- Flash storage

- Hard disk

- Remote storage

CockroachDB primarily supports Pebble as the storage engine, as of version 21.1. Previously, it also supported RocksDB.

Let’s look at Pebble.

Pebble

Pebble is primarily a key-value store that provides atomic write batches and snapshots. Starting from version 20.2, CockroachDB uses the Pebble storage engine by default. Pebble was developed to address two concerns, which are outlined here:

1. Focusing the storage engine’s features to primarily address the requirements of CockroachDB.

2. Improving the performance by bringing in certain optimizations that are not part of RocksDB.

Also, since it’s developed by Cockroach Labs engineers, it’s easy to maintain and control its roadmap. This has also increased the overall productivity as Pebble is written in Golang, just as is CockroachDB itself.

Summary

In this chapter, we learned about all the layers of CockroachDB and how a given request is processed through these layers. We also went through how queries are handled; how a transactional key-value store works; Raft, a distributed consensus algorithm; and a bit about storage engines.

In the next chapter, we will understand ACID and how it’s implemented in CockroachDB.

3 Atomicity, Consistency, Isolation, and Durability (ACID)

In the previous chapter, we learned about the different layers of CockroachDB. In this chapter, we will learn about what ACID is, its importance, and what the ACID guarantees that CockroachDB provides are.

ACID guarantees the following things:

- Atomicity: This is achieved through the notion of a transaction, in which all the statements within a transaction are executed as a single unit. So, either all of them succeed or fail together.

- Consistency: The database state should be consistent before and after a transaction is executed and should ensure that the database constraints are never violated.

- Isolation: Multiple transactions can get executed independently at the same time, without running into each other.

- Durability: Changes, once committed, always remain intact, irrespective of any system or network failures.

The following topics will be covered in this chapter:

- An overview of ACID properties.

- ACID from CockroachDB’s perspective.

An overview of ACID properties

In this section, we will discuss each of the ACID properties and understand their importance in avoiding data loss and corruption. First, we will take a look at atomicity.

Atomicity

Atomicity refers to the integrity of a given transaction, which means if a transaction comprises multiple statements, atomicity ensures that either all of them succeed or none of them succeed. Atomicity is important to make sure that there is no data inconsistency because of a transaction getting partially executed.

Let’s try to understand this with an example:

BEGIN TRANSACTION

Read Foo’s Account

Debit $100 from foo’s Account

Read Bar’s Account

Credit $100 to bar’s Account

COMMIT

Here, you have a transaction in which you are debiting the money from  Foo’s Account  and crediting it to  bar’s Account . Here, it’s important that these two activities happen as a single unit of work. Otherwise, it can result in data inconsistency. Consider a case in which you are able to debit the money from  Foo’s Account , but not able to credit it to  bar’s Account  – you will lose track of the money that was deducted from  Foo’s Account . So, either all of them should succeed, or none of them should.

Next, we will learn about consistency.

Consistency

Consistency in ACID is an overloaded term and can mean several things, including the following:

- Ensuring that the transactions in the future see the effects of transactions that are already committed

- Ensuring database constraints are not violated once a given transaction is committed

- Ensuring that all the operations in a transaction are executed correctly

Basically, consistency is responsible for ensuring the database always moves from one valid state to another, which doesn’t result in any data inconsistency or corruption.

In the context of a CAP theorem (Consistency, Availability, and Partition Tolerance), consistency indicates that in a distributed system, all the reads receive the most recent write, or it will error out. As per the CAP theorem, you can only have two of consistency, partition tolerance, and availability. Since consistency will be important to most of the applications, you have to choose between availability and partition tolerance.

Next, we will go through various isolation levels and try to understand the implication of each on a transaction.

Isolation

Isolation deals with the guarantees a database provides when multiple clients are interacting with the same set of data.

Some of the popular isolation levels are as follows:

- Serializable: This is the highest isolation level which requires acquiring a lock on the data you are operating. Transactions in CockroachDB implement the highest isolation level which is serializable. This means that transactions will never result in inconsistent or corrupt data. In the case of CockroachDB, this is provided by using range-level locks called write intents.

- Snapshot: This is a non-lock-based concurrency control, so no locks are not used, however, if a conflict is detected between concurrent transactions, then only one of them is allowed to commit.

- Read uncommitted: This is the lowest isolation level which allows dirty reads, which means changes made by live transactions that are not yet committed.

- Repeatable read: Repeatable read guarantees that you only read a committed value and also already read data cannot be changed by some other transaction. However, it does allow a phantom read. A phantom read happens when, between two reads of the same data, some other parallel transaction adds new data and they show up in subsequent reads, once they are committed.

- Read committed: This isolation level guarantees that any read you do is already committed.

With an example, let’s try to understand the difference between read committed, repeatable read, and serializable:

Figure 3.1 – Two transactions, T1 and T2, happening in parallel

In the previous example, there are two parallel transactions,  T1  and  T2 .

For the discussion, let’s assume that the following figure is the sequence of events:

Figure 3.2 – Ordering of operations for transactions T1 and T2

Here, serializable guarantees that the  T1  transaction sees the exact same value in Steps 3 and 8, although there are new rows added and some rows got deleted by the  T2  transaction.

With read committed, Step 9 will see new rows added by the  T2  transaction and it will not see the rows deleted by the  T2  transaction, since all the changes by  T2  are already committed.

In the case of a repeatable read, Step 9 will only see the new data committed by  T1 .

Durability

Durability guarantees that any changes that are committed are permanent, irrespective of failures related to nodes, memory, storage, or the network. Databases achieve durability by flushing out the transactional log to non-volatile storage like solid-state drives and magnetic storage devices.

Many of the Database Management Systems (DBMS) use the concept of a transaction log to ensure they can recover from system crashes. The transaction log is also called a binary log, database log, or audit trail. The transaction log is usually stored on an external storage device.

Let’s say a database node crashes while executing a bunch of transactions. Now, whenever it comes back, it goes through the transaction log to determine which transactions were committed and which were uncommitted at the time of the crash. If a transaction is committed, all the changes made during that transaction are replayed. If a transaction was uncommitted, all the changes made by the transaction are rolled back.

Next, we will learn about how ACID properties are supported in CockroachDB.

ACID from CockroachDB’s perspective

In this section, we will go over how each of the ACID properties is implemented in CockroachDB and what guarantees they provides. Like in the previous section, we will start with atomicity.

Atomicity

As we learned in the first section, atomicity ensures that all the statements in a given transaction are executed as a single unit – that is, either all of them succeed or all of them fail. This condition should be guaranteed irrespective of machine, network, and memory failures. This is essential to make sure multiple queries don’t run into each other.

CockroachDB allows you to have ACID transactions that can span the entire cluster, touching multiple nodes and geographical locations. CockroachDB supports this using an atomic protocol called parallel commits.

In the previous chapter, we learned about transaction records and write intent. A transaction record keeps track of the current state of the transaction and is maintained in the range where the first write occurs. Whenever we are changing a value, they are not directly written to the storage layer. Instead, a value is written to an intermediate state known as  write intent . Write intent acts an Multiversion concurrency control (MVCC) record, with a link to the transaction record. Write intent acts as a replicated lock, which houses a replicated provisional value.

Write intent has been shown in the following figure with a sample transaction with two writes:

Figure 3.3 – Transaction timeline without parallel commit

Any transaction that comes across a write intent should also go through corresponding transaction records and, based on its state, decide how to treat the write intent. A commit flips the transaction record state to  committed . Once the transaction is committed, write intents are cleaned up asynchronously.

Now, let’s take a look at how traditional atomic transactions worked in CockroachDB without parallel commits and later with parallel commits.

Atomic transactions without parallel commits

Prior to version 19.2, there was no parallel commit concept and transactions worked similar to a two-phase commit protocol. Let’s explore this with an example.

Let’s say there is a transaction with three writes, as shown in the following example:

BEGIN

write Tesla

write GameStop

write Amazon

COMMIT

This entire flow has been depicted in Figure 3.4:

Figure 3.4 – Transaction timeline without parallel commit

Consider the following:

1. The transaction record is created in the first range where the write happens. In this case, that happens to be the  Tesla  value. So, along with the provisional  Tesla  value, a write intent is created with the transactional record in a  PENDING  state.

2. As and when the rest of the writes are received, they create their own write intents. All these write intents are said to be pipelined because CockroachDB doesn’t wait until they succeed before receiving the next statement within the same transaction from the SQL client.

3. Once the  COMMIT  is issued in the transaction, CockroachDB waits for all the write intents to succeed with replication.

4. Once all of them succeed, the transaction record state is changed to  COMMITTED . Again, this is replicated for durability and the transaction is considered committed once the replication is completed.

5. At this time, CockroachDB sends back an acknowledgment that the transaction is committed.

6. Later, the rest of the write intents are resolved and are eventually cleaned up.

This mechanism is similar to a two-phase commit protocol, in which write intents can be compared to the  prepare  phase and marking the transaction record as committed is similar to the  commit  phase. The main problem here is that it is blocking, since CockroachDB waits for all the write intents to succeed before marking the transaction record as committed. If the coordinator node crashes, then it’s impossible to recover from that.

CockroachDB overcomes this problem by implementing a two-phase commit on top of a consensus protocol, RAFT, which we discussed in the previous chapter. This ensures that the transaction records themselves are replicated and highly available, and can recover from a coordinator crash.

Having a two-phase commit protocol on top of RAFT introduces more latency. This is because CockroachDB first waits for all the write intents to succeed before changing the transaction status to committed. Later, it has to wait until changing the transaction status itself has succeeded, as that involves one more round of consensus.

Now, let’s see how an atomic transaction with parallel commit avoids this added transactional latency.

Atomic transactions with parallel commits

Parallel commit was introduced to reduce the transaction latency observed in the previously discussed two-phase commit such as protocol.

In the previous protocol, the transaction record has to wait until all the write intents have succeeded to change the status to committed. In parallel commit, there is a new status called  STAGING . The transaction record also includes the list of all the keys for which there are write operations in the current transaction. A transaction can be implicitly assumed to be committed if all the writes that are listed in the transaction record have succeeded and reached consensus.

Let’s go over the same transaction with three writes.

BEGIN

write Tesla

write GameStop

write Amazon

COMMIT

This entire flow has been depicted in Figure 3.5:

Figure 3.5 – Transaction timeline with parallel commit

Here, the key difference is that the logic of a transaction being committed depends on the status of all the writes involved in that transaction. Also, if we know the status of all the writes, we don’t have to wait until the transaction record status is explicitly updated to  committed  after reaching consensus. Because of this, transaction’s coordinator node can acknowledge that a given transaction has been committed successfully to the SQL client once the coordinator observes that all the writes in that transaction have succeeded. The other important change in this protocol is that the initiation of pipelining of the write to the transaction record with the  STAGING  status is done after a  COMMIT  for a given transaction is received from the SQL client. Pipelining the write to the transaction record is done in parallel with pipelining the write intents, in order to speed up the entire process.

Now, let’s take a look at how a transaction status is recovered whenever a transaction coordinator crashes in the middle of a transaction.

Transaction status recovery

Now, let’s see what happens if the coordinator crashes before it can update the transaction record to either  COMMITTED  or  ABORTED . In this case, whenever there is a transaction with a conflicting write intent, it looks up that write intent’s transaction record. Once it sees that the status is  STAGING , it cannot decide whether that transaction was  COMMITTED  or  ABORTED . So, now it starts the status recovery process.

During transaction status recovery, each write intent involved in that transaction is consulted to see if it succeeded. If all the write intents have succeeded, the transaction is assumed to be  COMMITTED , and if not, to be  ABORTED . After this, the appropriate status is updated so that any other conflicting transaction in the future doesn’t have to go through the status recovery process again.

Status recovery can be very expensive, especially if it involves multiple writes with ranges that do not share leaseholders. If multiple leaseholders are involved in status recovery, there will be multiple roundtrips to several nodes, before we can recover the status. To avoid this, CockroachDB does two things.:

1. The transaction coordinator node marks the transaction record as  COMMITTED  or  ABORTED  as soon as it can.

2. Transaction coordinators periodically send heartbeats to their transaction records. This helps the conflicting transactions to determine if a transaction is still alive or not.

In the next section, we will learn about how consistency is ensured within CockroachDB.

Consistency

As discussed earlier, consistency deals with two things:

1. Ensuring no database rules are violated

2. Making sure that transactions that are getting executed in parallel on the same set of data do not conflict with each other, which is necessary to avoid data consistency issues.

For the first one, it boils down to making sure that the database doesn’t have any bugs and does whatever it claims. Jepsen is an effort to improve the safety of distributed databases, queues, and consensus systems. During Jepsen testing, a given system is verified for whether it lives up to its documentation’s claims. CockroachDB passed Jepsen testing in 2017.

In CAP theorem, which we discussed in the first chapter, consistency means every read sees the latest write or errors out. CockroachDB is a CP (consistent and partition tolerant) system, which means its highly consistent and, whenever there are partitions, the system becomes unavailable rather than ending up with inconsistent data.

Let’s now learn about isolation and what kind of isolation CockroachDB provides.

Isolation

CockroachDB uses something called a serializable snapshot, which is an optimistic, multi-version, timestamp-ordered concurrency control system.

It’s a distributed, lockless, recoverable, and serializable protocol. Distributed, as multiple nodes can be involved. Lockless, as operations are performed without locks and correctness is ensured by aborting transactions that violate serializability. Recoverable, since aborted transactions don’t have any effect on the state of the database, which is ensured by the atomic commit protocol. Serializable, since CockroachDB guarantees a consistent database state by ensuring serial execution of composite transactions is correct.

Next, we will learn about durability in CockroachDB.

Durability

Durability guarantees that any changes that are committed are permanent. CockroachDB uses the RAFT consensus algorithm to ensure that all writes for a transaction record and write intents are durable. We have already discussed RAFT at length in Chapter 2, How Does CockroachDB Work Internally?

CockroachDB replicates each range three times by default and ensures that each replica is stored on different nodes. If a minority of the nodes fail, CockroachDB continues to operate and does not result in inconsistency or loss of data.

Let’s take a look at how durability works in a three-node cluster:

Figure 3.6 – Transaction timeline with parallel commit

As you can see, node 1 is the lease holder for data1 and node 2 has the replica. Similarly, node 3 is the lease holder for data3, and node 1 has the replica. Next, node 2 is the lease holder for data2, and node 3 has the replica. Here, node 1 is also acting as a gateway node, where the initial request from the SQL client lands, and it also coordinates other nodes in the cluster to serve the request:

Figure 3.7 – Transaction timeline with parallel commit

Now, let’s say node 2 is not available due to a system failure. Since node 2 was the lease holder of data2, now the coordinator is unable to get the data for data2 from node 2, since it’s not available. Now, the RAFT group for data2 will hold an election and the lease holder will be reassigned. In this case, it has to be node 3, as that’s the only other node that has the replica of data2.

Figure 3.8 – Transaction timeline with parallel commit

As you can see in Figure 3.8, node 3 acts as the lease holder for both data2 and data3 and the cluster is still fully functional, in spite of a node failure.

Now, if node 3 also goes down for some reason, since we don’t have any lease holder for data2, if a request involves serving the data for data2, it cannot be completed, affecting the availability of data2 on a given CockroachDB cluster.

As you can see, CockroachDB can tolerate certain node failures, as long as there are lease holders for all the data. Otherwise, availability takes a hit. In either case, there won’t be any data loss.

Users can also configure replication zones for databases, tables, rows, indexes, and system data. We will discuss these configurations in Chapter 5, Fault Tolerance and Auto-Recovery, where we will discuss fault tolerance and auto-recovery.

Summary

In this chapter, we learned about the four basic pillars of any database: Atomicity, Consistency, Isolation, and Durability. As a recap, atomicity ensures that a transaction is executed as a single unit of work. Consistency involves making sure that any database operation doesn’t violate any of the database constraints. In the context of a CAP theorem, consistency refers to the fact you will never read stale or uncommitted data. CockroachDB provides both serializable and snapshot isolation levels. CockroachDB uses the RAFT protocol for transaction records and write intents to guarantee that all committed data is durable and permanent, irrespective of node failures.

In the next chapter, we will go over fault tolerance and auto-recovery, and what some of the configurations are in CockroachDB.

4 Geo-Partitioning

In the previous chapter, we learned about what ACID is, why we need it, and how it’s supported in CockroachDB. Here, we will learn all about geo-partitioning. Geo-partitioning is one of the most important reasons why you will want to use a distributed SQL such as CockroachDB.

In this chapter, you will get a basic understanding of what geo-partitioning is and why this feature is useful for you. We will also go over some cloud jargon and some of the options provided by various cloud providers to distribute your data geographically for better resiliency, performance, and availability. At the end of the chapter, we will go over different ways of geo-partitioning your data in CockroachDB.

The following topics will be covered in this chapter:

- Introduction to geo-partitioning

- Cloud regions and zones

- Geo-partitioning in CockroachDB

Technical requirements

All the queries in this chapter are available at https://github.com/kishendas/cockroachdb-tutorial.

Introduction to geo-partitioning

As the word “geo-partition” suggests, the data is partitioned based on geographical locations. Geo-partitioning refers to the mechanism of storing the data in various geographical locations, based on where the data is getting consumed.

For example, let’s say you are an airlines company that has custome rs from every continent. Since you have a global presence, it would be beneficial to keep the users’ data close to where they live. This will help in serving the data locally and quickly.

Figure 4.1 shows an example of a table whose rows are partitioned based on geolocation across three different continents. Rows are stored in specific databases based on their locality:

Figure 4.1 – An example of a geo-partitioned table

1. Some of the main advantages of geo-partitioning are as follows:

- It helps serve data quickly, as we reduce the number of network hops.

- Failures in a given geolocation only affect a small set of users.

- It provides data compliance as per local rules and standards, for example, the California Consumer Privacy Act of 2018 (CCPA) and the European Union’s General Data Protection Regulation (GDPR).

- It helps provide better disaster recovery and resiliency if an entire geolocation experiences a natural or human-triggered calamity.

Let’s now learn about cloud regions and availability zones.

Cloud regions and zones

In this section, we will learn about some jargon related to the cloud. We will also learn about regions and zones, and how they have been realized by various cloud providers. These concepts are important to understand to decide how you want to distribute your data and what sort of guarantees you want to provide:

- Cloud: A cloud is nothing but a bunch of servers on multiple data centers that are positioned in strategic locations across the globe. These data centers provide resources such as storage, network, and compute on demand, and they belong to a specific cloud provider.

- Cloud provider: A cloud provider is an organization that provides various services on its public or private cloud platform.

- Public cloud: In a public cloud, resources that you consume are hosted on the cloud provider’s data center. The cloud provider is responsible for maintaining, upgrading, and operating cloud resources. Since you are consuming resources maintained by a third party, there are additional security risks here.

- Private cloud: In the case of the private cloud, the resources are usually hosted on a company’s own data center, but they can also be hosted by a cloud provider. In a private cloud, all the resources are dedicated to a single organization and isolated from other organizations; hence it’s more controlled and secured.

- Multi-cloud: A given platform is called multi-cloud, where you consume resources from multiple public cloud providers.

- Hybrid cloud: In a hybrid cloud, you will be combining resources from a public cloud along with resources from a private cloud and/or on-premises.

Region

A region refers to an actual physical location, where your cloud resources are housed. Each cloud provider has different notions of a region.

It’s very important to understand how regions are implemented by different cloud providers, as it determines the following things:

- Cloud cost: Resources in some regions are cheaper than in others.

- Multi-cloud and hybrid cloud strategy: This includes disaster recovery, high availability, data replication, data migration, data sharing, failover, and so on.

- Latency: The whole idea of geo-partitioning is about reducing the latency by keeping the data close to the customer. So, it becomes apparent to select a region in strategic locations, which reduces overall latency.

- Data compliance: Depending on where the region is located, you might have different data compliance requirements. Also, some countries might insist that the data of their citizens cannot leave the country, in which case you will be forced to pick some regions in that country.

- Services and features: Not all the services and features are available in all the regions. So, this sometimes reduces the choice of regions.

Zone

A region consists of multiple zones. A zone refers to a more specific location within a given region.

Availability zone

An availability zone is an isolated data center that doesn’t share any resource with other zones within the same region. All the communication between availability zones happens through a high-speed network. A region is supposed to have at least two availability zones that help in implementing redundancy, failover, and high availability.

Now, let’s understand the definition of region and zone by some of the top cloud providers.

Regions and zones on various cloud providers

In this section, we will briefly go over what region and zone mean on the top four cloud providers.

Amazon Web Services

Region: A Region is a physical location that consists of multiple data centers.

Availability Zone: A group of discrete data centers that provide redundancy to cloud resources is called an Availability Zone. Availability Zones help in implementing features such as high availability, fault tolerance, reliability, and scale.

AWS Local Zone: Local Zones provide resources that are located close to your end users. This will be useful in services such as gaming and streaming, which require low latency, high throughput, and elastic services.

Google Cloud Platform

Region: A region is a collection of zones.

Zones: A zone is a deployment within a region. You should use multiple zones to provide high availability and fault tolerance.

Microsoft Azure

Region: A region is a set of data centers connected within a perimeter determined by the latency and connected through a fast network.

Geography: An area of the world containing at least one Azure region. Geography spans multiple regions and is fault tolerant, even in the event of a complete regional failure.

Availability zones: Unique physical locations within a region. Each availability zone comprises one or more data centers with resources that are not shared with other zones.

Oracle Cloud

Region: A region is a localized area and is made of several availability domains.

Availability domains: Availability domains are made of one or more data centers, and they do not share any resources amongst them and are connected through a fast network.

Each availability domain has three fault domains. Fault domains ensure your resources are from different availability domains, which offers improved resiliency.

Next, we will learn about how to achieve geo-partitioning with CockroachDB.

Geo-partitioning in CockroachDB

CockroachDB provides two topology patterns, which provide two levels of data resiliency, latency and availability.

Single region

Here, the entire data is in a single region.

CockroachDB defines two variations of single-region topology, development and production, as follows:

- Development: This pattern is very straightforward, where you just have a single node in an availability zone, with multiple clients talking to it. This pattern is useful for testing purposes. This topology can also be used on your laptop or desktop. As part of your Continuous Integration/Continuous Deployment (CI/CD) pipeline, you can have a dedicated stage in which you provision a single-node cluster and later can run a bunch of system tests that interact with a real database. Since the clients will be local to the data, reads and writes will be much faster, although there is no resiliency. The following is an example of a single-region deployment:

Figure 4.2 – Single-node deployment in the US-West (northern California) region, where all the clients are also deployed in the same region

- Basic production: Here, you can have nodes deployed in more than one availability zone within the same region. It is ideal to have at least three nodes in three different availability zones within the same region for consensus purposes. This pattern takes advantage of many CockroachDB features, such as replication, rebalancing, and resiliency. This topology can withstand up to a single-node failure. If two nodes fail, then some ranges might not have any leaseholders due to lack of consensus and will become unavailable. You would also need a load balancer to spread the traffic from clients across three nodes evenly. The following is a single-region deployment with three nodes:

Figure 4.3 – Single-region three-node deployment in the US-West (northern California) region, where all the clients are also deployed in the same region

In basic production topology, client requests hit a load balancer first and later they are forwarded to one of the nodes, which acts as a gateway node. The gateway node then coordinates with relevant leaseholders, gathers all the data, and serves the data back to the client.

Multi-region

Here, the data is spread across multiple regions. You can use row-level control to distribute the rows geographically. The following figure shows a multi-region deployment, which spans three different regions in the US:

Figure 4.4 – Multi-region nine-node deployment covering US-East, US-Central, and US-West regions

You can choose the following topologies based on your requirement:

- Geo-partitioned replicas

- Geo-partitioned leaseholders

- Duplicate indexes

- Follower reads

- Follow-the-workload

Geo-partitioning options are only available with the enterprise license of CockroachDB. If you try to use enterprise features without the enterprise license, you will see the following message:

ERROR: use of partitions requires an enterprise license. see https://cockroachlabs.com/pricing?cluster=65244c3a-2d63-432c-a8b4-c70a53459ca1 for details on how to enable enterprise features

SQLSTATE: XXC02

You can visit the URL mentioned in the error message and get an enterprise license for a specific trial period.

Applying an enterprise license to your cluster would involve the following steps:

root@localhost:26258/defaultdb> SET CLUSTER SETTING cluster.organization = 'Self';

SET CLUSTER SETTING

Time: 166ms total (execution 165ms / network 0ms)

root@localhost:26258/defaultdb> SET CLUSTER SETTING enterprise.license = 'crl-0-EKD0mYsGGAIiBFNlbGY';

SET CLUSTER SETTING

Time: 191ms total (execution 191ms / network 0ms)

Now, let’s take a deeper look at each one of these topologies. All these topologies would need a multi-region deployment. The easiest way to create a multi-region cluster for experiment purposes is to create an account at https://cockroachlabs.cloud and request for a three-region, nine-node cluster.

Geo-partitioned replicas

In the case of geo-partitioned replicas, you have to have a column for the geolocation. It then has to be combined with the table’s unique identifier to form a compound primary key. For example, let’s say you have an ID, which is a Universally Unique Identifier (UUID) – you can combine that with the city or the state to form a compound primary key. Then, you have to partition the table and all the secondary indexes based on that column, and each partition will have its own replicas. Once you have this, you can ask CockroachDB to place data belonging to each partition in some specific region.

For example, let’s say you have users from California and Ohio. All the rows belonging to users from California can be stored in the US-West region and users in Ohio can be stored in the US-East region. Here, the assumption is that data will be consumed locally, so both read and write latencies will always be fewer.

For example, if you operate a food delivery service, your users will place orders from the city they live in, and also the food will be delivered within the same city. So, in this case, it makes total sense to use a geo-partitioned replica for your users. Since all the data and its replica are housed in the same region, if an entire region goes down, the data will not be available.

Let’s see an example in which we are going to create a geo-partitioned replicas topology. Basically, the replicas will be pinned to a particular region so that local reads and writes are faster:

1. Create a table called  users , where you have  city  as one of the columns. This will help in partitioning this table by city:

CREATE TABLE users (

id UUID NOT NULL DEFAULT gen_random_uuid(),

first_name STRING NOT NULL,

last_name STRING NOT NULL,

city STRING NOT NULL,

PRIMARY KEY (city ASC, id ASC) );

2. Now, create a secondary index as follows:

CREATE INDEX users_first_name_last_name_index ON users (city, first_name, last_name);

3. Create partitions for the table based on  city . Let’s consider three different cities that are in the west, east, and central part of the USA:

ALTER TABLE users PARTITION BY LIST (city) (

PARTITION sfo VALUES IN ('san francisco'),

PARTITION aus VALUES IN ('austin'),

PARTITION ny VALUES IN ('new york')

);

4. Create partitions for the secondary index based on  city :

ALTER INDEX users_first_name_last_name_index PARTITION BY LIST (city) (s

PARTITION sfo VALUES IN ('san francisco'),

PARTITION aus VALUES IN ('austin'),

PARTITION ny VALUES IN ('new york')

);

5. For the table and its secondary index, create a replication zone, which will pin the replica of a given partition to its relevant region. The  <table>@*  syntax lets you create zone configurations for all identically named partitions of a table, saving you multiple steps:

ALTER PARTITION sfo OF INDEX users@* CONFIGURE ZONE USING constraints = '{"+region=us-west":1}', num_replicas=3;

ALTER PARTITION aus OF INDEX users@* CONFIGURE ZONE USING constraints = '{"+region=us-central":1}', num_replicas=3;

ALTER PARTITION ny OF INDEX users@* CONFIGURE ZONE USING constraints = '{"+region=us-east":1}', num_replicas=3;

6. Now, let’s execute  SHOW CREATE TABLE  to see how the partitions are created for the table and the secondary index:

SHOW CREATE TABLE users;

The sample output is as follows:

SHOW CREATE TABLE users;

table_name |                                                create_statement

-------------+------------------------------------------------------------------------------------------------------------------

users      | CREATE TABLE public.users (

|     id UUID NOT NULL DEFAULT gen_random_uuid(),

|     first_name STRING NOT NULL,

|     last_name STRING NOT NULL,

|     city STRING NOT NULL,

|     CONSTRAINT "primary" PRIMARY KEY (city ASC, id ASC),

|     INDEX users_first_name_last_name_index (city ASC, first_name ASC, last_name ASC) PARTITION BY LIST (city) (

|         PARTITION sfo VALUES IN (('san francisco')),

|         PARTITION aus VALUES IN (('austin')),

|         PARTITION ny VALUES IN (('new york'))

|     ),

|     FAMILY "primary" (id, first_name, last_name, city)

| ) PARTITION BY LIST (city) (

|     PARTITION sfo VALUES IN (('san francisco')),

|     PARTITION aus VALUES IN (('austin')),

|     PARTITION ny VALUES IN (('new york'))

| );

| ALTER TABLE defaultdb.public.users CONFIGURE ZONE USING

|     num_replicas = 3;

| ALTER PARTITION sfo OF INDEX defaultdb.public.users@primary CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-west: 1}';

| ALTER PARTITION sfo OF INDEX defaultdb.public.users@users_first_name_last_name_index CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-west: 1}';

| ALTER PARTITION aus OF INDEX defaultdb.public.users@primary CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-central: 1}';

| ALTER PARTITION aus OF INDEX defaultdb.public.users@users_first_name_last_name_index CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-central: 1}';

| ALTER PARTITION ny OF INDEX defaultdb.public.users@primary CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-east: 1}';

| ALTER PARTITION ny OF INDEX defaultdb.public.users@users_first_name_last_name_index CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-east: 1}'

(1 row)

Time: 278ms total (execution 278ms / network 0ms)

Here, you can see that replicas are constrained to relevant regions. So, if an entire region goes down, a partition in that region becomes unavailable.

Next, we will go through the geo-partitioned leaseholders topology.

Geo-partitioned leaseholders

Like in the case of a geo-partitioned replica, you still need a column that has geolocation. You will also need a compound primary key, which is a combination of a unique ID and geolocation.

Here, the main difference is that you only pin the leaseholder to a specific location, but the replicas can be stored in different regions. Since we are only pinning the leaseholder, reads will always be faster but writes take more time, since data is replicated across regions, which takes more time as replication also involves consensus.

Let’s see an example, in which we are going to create a geo-partitioned leaseholders topology. Basically, the leaseholder will be pinned to a particular region, so that local reads are faster:

1. Create a table called  users , where you have  city  as one of the columns. This will help in partitioning this table by city:

CREATE TABLE users (

id UUID NOT NULL DEFAULT gen_random_uuid(),

first_name STRING NOT NULL,

last_name STRING NOT NULL,

city STRING NOT NULL,

PRIMARY KEY (city ASC, id ASC) );

2. Now, create a secondary index as follows:

CREATE INDEX users_first_name_last_name_index ON users (city, first_name, last_name);

3. Create partitions for the table based on  city . Let’s consider three different cities that are in the west, east, and central part of the USA:

ALTER TABLE users PARTITION BY LIST (city) (

PARTITION sfo VALUES IN ('san francisco'),

PARTITION aus VALUES IN ('austin'),

PARTITION ny VALUES IN ('new york')

);

4. Create partitions for the secondary index based on  city :

ALTER INDEX users_first_name_last_name_index PARTITION BY LIST (city) (

PARTITION sfo VALUES IN ('san francisco'),

PARTITION aus VALUES IN ('austin'),

PARTITION ny VALUES IN ('new york')

);

5. For the table and its secondary index, create a replication zone, which will pin the leaseholder of a given partition to its relevant region:

ALTER PARTITION sfo OF INDEX users@*

CONFIGURE ZONE USING

num_replicas = 3,

constraints = '{"+region=us-west":1}',

lease_preferences = '[[+region=us-west]]';

ALTER PARTITION aus OF INDEX users@*

CONFIGURE ZONE USING

num_replicas = 3,

constraints = '{"+region=us-central":1}',

lease_preferences = '[[+region=us-central]]';

ALTER PARTITION ny OF INDEX users@*

CONFIGURE ZONE USING

num_replicas = 3,

constraints = '{"+region=us-east":1}',

lease_preferences = '[[+region=us-east]]';

6. Now, let’s execute  SHOW CREATE TABLE  to see how the partitions are created for the table and the secondary index:

SHOW CREATE TABLE users;

The sample output is as follows:

SHOW CREATE TABLE users;

table_name |                                                create_statement

-------------+------------------------------------------------------------------------------------------------------------------

users      | CREATE TABLE public.users (

|     id UUID NOT NULL DEFAULT gen_random_uuid(),

|     first_name STRING NOT NULL,

|     last_name STRING NOT NULL,

|     city STRING NOT NULL,

|     CONSTRAINT "primary" PRIMARY KEY (city ASC, id ASC),

|     INDEX users_first_name_last_name_index (city ASC, first_name ASC, last_name ASC) PARTITION BY LIST (city) (

|         PARTITION sfo VALUES IN (('san francisco')),

|         PARTITION aus VALUES IN (('austin')),

|         PARTITION ny VALUES IN (('new york'))

|     ),

|     FAMILY "primary" (id, first_name, last_name, city)

| ) PARTITION BY LIST (city) (

|     PARTITION sfo VALUES IN (('san francisco')),

|     PARTITION aus VALUES IN (('austin')),

|     PARTITION ny VALUES IN (('new york'))

| );

| ALTER TABLE defaultdb.public.users CONFIGURE ZONE USING

|     num_replicas = 3;

| ALTER PARTITION sfo OF INDEX defaultdb.public.users@primary CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-west: 1}',

|     lease_preferences = '[[+region=us-west]]';

| ALTER PARTITION sfo OF INDEX defaultdb.public.users@users_first_name_last_name_index CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-west: 1}',

|     lease_preferences = '[[+region=us-west]]';

| ALTER PARTITION aus OF INDEX defaultdb.public.users@primary CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-central: 1}',

|     lease_preferences = '[[+region=us-central]]';

| ALTER PARTITION aus OF INDEX defaultdb.public.users@users_first_name_last_name_index CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-central: 1}',

|     lease_preferences = '[[+region=us-central]]';

| ALTER PARTITION ny OF INDEX defaultdb.public.users@primary CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-east: 1}',

|     lease_preferences = '[[+region=us-east]]';

| ALTER PARTITION ny OF INDEX defaultdb.public.users@users_first_name_last_name_index CONFIGURE ZONE USING

|     num_replicas = 3,

|     constraints = '{+region=us-east: 1}',

|     lease_preferences = '[[+region=us-east]]'

(1 row)

Time: 37ms total (execution 37ms / network 0ms)

Here, you can see that the lease preference is restricted to relevant regions, which will ensure that the leaseholders are always pinned to a specific region.

Next, we will go through the duplicate indexes topology.

Duplicate indexes

The duplicate indexes topology is useful in cases where you write once and read it from various locations. For example, let’s say you are managing the credit cards of folks who keep traveling throughout the US very often. If you pin the data to a single region, whenever the user moves out of that region, it will slow down the reads. So, duplicate indexes come in handy to solve this issue.

Just like the previous two cases, you will have a compound primary key with a combination of an ID and geolocation. Here, you can create a partition based on that column, but you only pin the leaseholder to a specific region.

In our example, it can be the primary address of the user. Here, the credit card information can be replicated in different regions to cover the entire US. Since only the leaseholder is responsible for writes and reads, your reads will always be routed to the pinned region of the leaseholder. This again introduces latency.

Now, you can create secondary indexes for the credit card. For example, assuming that your leaseholder is pinned to the west coast, you can create secondary indexes, such as  id_creditcard_east  and  id_creditcard_central , which can be constrained to US-East and US-Central regions respectively. This will also guarantee that there are local leaseholders for secondary indexes in all the regions, which drastically reduces the read latency, as they were served locally always.

Since we already have multiple copies of the original data and we are creating secondary indexes that are also replicated, now there are a lot of copies of the same data in multiple regions. So, this increases the write latencies, as all these copies have to be updated and a multi-region consensus has to be reached:

1. Let’s say you are maintaining local attractions of the USA, which can be accessed by users throughout the USA:

CREATE TABLE local_attractions (

id UUID NOT NULL DEFAULT gen_random_uuid(),

name STRING NOT NULL,

address STRING NOT NULL,

city STRING NOT NULL,

PRIMARY KEY (id ASC)

);

2. Create a replication zone and pin the leaseholder to a specific region:

ALTER TABLE local_attractions

CONFIGURE ZONE USING

num_replicas = 3,

constraints = '{"+region=us-central":1}',

lease_preferences = '[[+region=us-central]]';

3. Create secondary indexes for the other two regions. Here, storing a column improves the performance of queries that retrieve its values, but you cannot use these stored columns in the filtering logic:

CREATE INDEX idx_west ON local_attractions (city)

STORING (name);

CREATE INDEX idx_east ON local_attractions (city)

STORING (name);

4. For these secondary indexes, define the replication zone, once again pinning the leaseholder to the relevant region:

ALTER INDEX local_attractions@idx_west

CONFIGURE ZONE USING

num_replicas = 3,

constraints = '{"+region=us-west":1}',

lease_preferences = '[[+region=us-west]]';

ALTER INDEX local_attractions@idx_east

CONFIGURE ZONE USING

num_replicas = 3,

constraints = '{"+region=us-east":1}',

lease_preferences = '[[+region=us-east]]';

5. Now, let’s execute  SHOW CREATE TABLE  to see how the partitions are created for the secondary indexes:

SHOW CREATE TABLE local_attractions;

The sample output is as follows:

SHOW CREATE TABLE local_attractions;

table_name     |                create_statement

--------------------+--------------------------------------------------

local_attractions | CREATE TABLE public.local_attractions (

|     id UUID NOT NULL DEFAULT gen_random_uuid(),

|     name STRING NOT NULL,

|     address STRING NOT NULL,

|     city STRING NOT NULL,

|     CONSTRAINT "primary" PRIMARY KEY (id ASC),

|     INDEX idx_west (city ASC) STORING (name),

|     INDEX idx_east (city ASC) STORING (name),

|     FAMILY "primary" (id, name, address, city)

| )

(1 row)

Time: 28ms total (execution 28ms / network 0ms)

Here, you can see that there are multiple identical indexes for multiple regions. So, whenever there are queries involving city and name, they can be served locally, hence reducing the latency of reads. Since we are maintaining identical indexes in multiple regions, the writes are much slower. This topology is useful where the data doesn’t change much but is accessed frequently in all the regions.

Next, we will go through the follower reads topology.

Follower reads

If you want low-read latency but don’t care about slightly older data, you can use this topology. In this case, you add the  AS OF SYSTEM TIME  clause in your reads, which then avoids the round trip to the leaseholder, and data is served locally. Writes would still need a multi-region consensus. You should not use this topology if you need strong consistency. Please refer to the CAP theorem section in Chapter 1, CockroachDB - a Brief Introduction if you want to understand what consistency here means.

Let’s create a sample table to understand how this works:

CREATE TABLE local_attractions (

id UUID NOT NULL DEFAULT gen_random_uuid(),

name STRING NOT NULL,

address STRING NOT NULL,

city STRING NOT NULL,

PRIMARY KEY (id ASC)

);

Here, in the  SELECT  query, you should use the  AS OF SYSTEM TIME follower_read_timestamp() . The  follower_read_timestamp()  function returns the  TIMESTAMP  data type with the  statement_timestamp()  -  4.8s  value:

SELECT city FROM local_attractions

AS OF SYSTEM TIME follower_read_timestamp()

WHERE city = 'san francisco';

Since the data is always retrieved locally, without involving the leaseholder, you might get stale or older data. Once again, do not use this topology if you need strong consistency.

Next, we will look at follow-the-workload topology.

Follow-the-workload

This is the default topology, if you don’t use any of the previous ones. This topology works well if a given table is active in a single region, which means clients are doing reads and writes that are in the same region. Here, the read latency will be low in the active region and it will be more in non-active regions, as the leaseholder will be in the active region. Writes still need a multi-region consensus and can be slower.

The following is a table that should help you to decide which topology might be relevant for your database workload:

Pattern

Latency

Resiliency

Configuration

Geo-partitioned replicas

Fast regional reads and writes

Can withstand one availability zone failure per partition

Geo-partitioned table

Partition replicas pinned to regions

Geo-partitioned leaseholders

Fast regional reads

Slower cross-region writes

Can withstand one regional failure

Geo-partitioned table

Partition replicas spread across regions

Partition leaseholders pinned to regions

Duplicate indexes

Fast regional reads

Very slow cross-region writes, as even indexes must be replicated

Can withstand one regional failure

Multiple identical indexes

Index replicas spread across regions

Index leaseholders pinned to regions

Follower reads

Fast regional reads for stale data

Slower cross-region writes

Can withstand one regional failure

Applications configured to use follower reads

Follow-the-workload

Fast regional reads, in the active region

Slower cross-region reads, in non-active regions.

Slower cross-region writes

Can withstand one regional failure

No configuration required as this is the default

Figure 4.5 – Topology cheat sheet

In this section, we learned about various geo-partitioning topologies and how to configure them in CockroachDB. Based on latency, data consistency, and resiliency requirements, we should select the appropriate topology.

Summary

In this chapter, we learned what geo-partitioning is and why it is useful to geo-partition your data. Then, we covered all the important jargon in the cloud world, especially how each major cloud provider has defined regions and availability zones. We later discussed how to configure various multi-region topologies based on your application requirements.

In the next chapter, we will go over fault tolerance and auto-recovery with CockroachDB.

6 How Indexes Work in CockroachDB

In the previous chapter, we learned about fault-tolerance and auto-recovery strategies in CockroachDB. In this chapter, we will learn everything about indexes, what they are, and how they improve query times.

The following topics will be covered in this chapter:

- Introduction to indexes

- Different types of indexes

- Best practices while using indexes

Introduction to indexes

A database index helps with returning the query results quickly, by avoiding full table scans. An index can be created for a specific table and can include one or more keys. Keys refer to the columns in the table. However, there will be extra space used to keep a separate sorted copy of indexed columns.

Let’s take a simple example and understand how an index works.

Consider a population table with the following columns and some sample values.

id

country

continent

population_in_millions

1

India

Asia

1378

2

USA

North America

331

3

South Africa

Africa

60

4

China

Asia

1400

5

Switzerland

Europe

8.5

Figure 6.1 – Population table

Now, let’s say you just want to retrieve the list of populations for specific continents, for example:

SELECT population_in_millions, country FROM population WHERE continent = “Asia”;

Here, in order to find rows 1 and 4, which have countries in Asia, you would have to iterate through each of the rows in the table, which is called a full table scan.

Now, if you want to avoid a full table scan, you can create an index on the continent as follows:

CREATE INDEX ON population (continent);

Internally, CockroachDB keeps track of all the continents and keeps a mapping from a given continent to all its relevant rows, as shown here:

Africa -> (3)

Asia    -> ( 1, 4 )

Europe -> ( 5 )

North America -> (2 )

Now, if you run the same query,  SELECT population_in_millions, country FROM population WHERE continent = “Asia” , once again, CockroachDB will identify that the filtering condition has the column  continent  and there is an index already available for that. So, in this case, based on the value  Asia , it will directly retrieve rows (1 , 4) from the continent index and get relevant column values and return it. In this case, it has avoided a full table scan. Although this example only has five rows, the same concept is applicable, even when a table contains millions of rows. So, in such cases, avoiding a full table scan can significantly improve the query performance. At the same time, writes tend to get a little slower as, with each write, even the index must be updated.

When you create an index on a column or a set of columns, CockroachDB internally makes a copy of the values of that column or columns and sorts them. So, whenever you execute a query that involves filters on indexed columns, a subset of rows is selected from the index first, rather than scanning the entire table. This improves the overall query performance.

Next, we are going to discuss various types of indexes that are available in CockroachDB.

Different types of indexes

Based on the query pattern and columns in the table, you should decide what kind of index is going to help with the performance.

Following are the types of indexes available in CockroachDB:

- Primary index

- Secondary index

- Hash-sharded index

- Duplicate indexes

- Inverted indexes

- Partial indexes

- Spatial indexes

- Table joins and indexes

- Best practices while using indexes

In the next set of subsections, we are going to discuss each type of index and when to use them, starting with the primary index.

Primary indexes

A primary key uniquely identifies a given row in a table. This means that the primary key is unique for a given row and duplicate values or NULLs are not allowed.

Whenever you create a table in CockroachDB, it’s recommended to have an explicit primary key, so that CockroachDB automatically creates an index for it, which can be used to filter the rows for better performance. Even if you don’t create a primary key during table creation, CockroachDB by default creates a primary key called  rowid , which will have a unique value for each row, but its performance will not be as good as that of the primary key.

Let’s understand how indexes work with an example, where we are going to create a database and a table with a primary key:

1. Create a database called  test :

root@localhost:26257/defaultdb> CREATE DATABASE IF NOT EXISTS test;

CREATE DATABASE

Time: 279ms total (execution 279ms / network 0ms)

Switch to the database ‘test’.

root@localhost:26257/defaultdb> use test;

SET

Time: 116ms total (execution 115ms / network 0ms)

2. Create a table called  accounts  with  id  being the primary key:

root@localhost:26257/test> CREATE TABLE accounts (

id UUID PRIMARY KEY,

name string,

balance INT8

);

CREATE TABLE

Time: 195ms total (execution 195ms / network 0ms)

3. Now, we will look at the indexes created for the  accounts  table using the  SHOW INDEXES  command:

SHOW INDEXES FROM accounts;

table_name | index_name | non_unique | seq_in_index | column_name | direction | storing | implicit

-------------+------------+------------+--------------+-------------+-----------+---------+-----------

accounts     | primary         |     false         |                             1 | id                        | ASC               |    false    |    false

(1 row)

Time: 7ms total (execution 7ms / network 0ms)

4. In order to understand how this primary key index helps with query performance, we can use  EXPLAIN  to look at the statement plans.

5. If you are retrieving all the accounts without any filters, obviously a full scan is required as we must return all the rows:

root@localhost:26257/test> explain select * from accounts;

tree    |         field         |     description

------------+-----------------------+-----------------

| distribution          | full

| vectorized            | false

scan    |                       |

| estimated row count   | 1

| table                 | accounts@primary

| spans                 | FULL SCAN

(6 rows)

6. Now if you want to retrieve just one row based on the ID, you can avoid a full table scan, since CockroachDB has already indexed the  id  column.

7. As you can see in the following  explain  statement, within the spans, now we no longer do a full table scan:

root@localhost:26257/test> explain select * FROM accounts where id = '123e4567-e89b-12d3-a456-426655440000';

tree  |         field         |    description

----------+-----------------------+-------------------

| distribution          | local

| vectorized            | false

scan  |                       |

| estimated row count   | 1

| table                 | accounts@primary

| spans                 | [/'123e4567-e89b-12d3-a456-426655440000' - /'123e4567-e89b-12d3-a456-426655440000']

(6 rows)

If multiple columns are used in queries, you should also consider creating a composite primary key that includes all the columns that are often used together.

In the next section, we will learn about secondary indexes.

Secondary indexes

A secondary index is an index that you create on non-primary columns. If your query involves retrieving a column that’s not a primary key and you want to improve the query performance, you can create secondary indexes. Any index that you create on a non-primary key is called a secondary index and duplicate values are allowed for secondary indexes. For the  test.accounts  table, if the query contains a non-primary column such as  name , then we would still need a full table scan. Let’s try this with an example, where we will just use a non-primary column in the query:

root@localhost:26257/test> explain select name FROM accounts where name = 'crdb' ;

tree     |         field         |     description

--------------+-----------------------+-------------------

| distribution          | full

| vectorized            | false

filter    |                       |

│        | filter                | name = 'crdb'

└── scan |                       |

| estimated row count   | 1

| table                 | accounts@primary

| spans                 | FULL SCAN

(8 rows)

Since we are now filtering on a non-primary column, CockroachDB must inspect each row and apply a filtering condition, and the index on the primary key  id  doesn’t help here. So, let’s create one more index on the column  name :

root@localhost:26257/test> CREATE INDEX on accounts ( name );

CREATE INDEX

Time: 2.053s total (execution 0.256s / network 1.797s)

Whenever you create a secondary index, CockroachDB automatically creates a composite index including the primary key. Also, the index on the column  name  is called a secondary index:

root@localhost:26257/test> show indexes from accounts;

table_name |         index_name          | non_unique | seq_in_index | column_name | direction | storing | implicit

-------------+-------------------+------------+--------------+-------------+-----------+---------+-----------

accounts     | primary                         |     false         |                             1 | id                        | ASC               |    false    |    false

accounts     | accounts_name_idx |         true         |                             1 | name                   | ASC               |    false    |    false

accounts     | accounts_name_idx |         true         |                             2 | id                        | ASC               |    false    |     true

(3 rows)

Now if you run the previous query, you should see that the full table scan is avoided because of the new index that we have created:

root@localhost:26257/test> explain select name from accounts where name = 'crdb' ;

tree   |       field        |    description

----------+--------------------+--------------------

| distribution       | local

| vectorized         | false

Scan |                    |

| estimated row count| 1

| table              | accounts@accounts_name_idx

| spans              | [/'crdb' - /'crdb']

(6 rows)

Time: 1ms total (execution 1ms / network 0ms)

Hash-sharded indexes

Hash-sharded indexes are useful to improve query performance when you must create an index on a column that’s a sequence. Hash-sharded indexes are helpful in evenly spreading the traffic to a sequential range across multiple ranges to avoid hotspots for any given range. Since this is a new experimental feature, the implementation and overall performance might change over time.

1. Within the client session, you have to first enable this feature as shown in the following code block:

root@localhost:26257/test> set experimental_enable_hash_sharded_indexes = ON;

SET

Time: 1ms total (execution 0ms / network 0ms)

2. Let’s create a table called  customers  with integer and string data types. Here, the  id  column is supposed to be a sequence:

root@localhost:26257/test> create TABLE customers ( id int PRIMARY KEY, name string);

CREATE TABLE

Time: 160ms total (execution 160ms / network 0ms)

3. Now, let’s create the hash-sharded index for this primary key, as shown in the following code:

root@localhost:26257/test> ALTER TABLE customers ALTER PRIMARY KEY USING COLUMNS (id) USING HASH WITH BUCKET_COUNT = 10;

NOTICE: primary key changes are finalized asynchronously; further schema changes on this table may be restricted until the job completes

ALTER TABLE

Time: 4.551s total (execution 0.297s / network 4.253s)

When you create a hash-sharded index, CockroachDB creates  n_buckets  computed columns, shards the primary index ID into  n_buckets  number of shards, and then stores each index shard in the underlying key-value store with one of the computed column's hash as its prefix.

4. Let’s look at how indexes on the  customers  table look now:

root@localhost:26257/test> show indexes from customers;

table_name |         index_name         | non_unique | seq_in_index |                    column_name                    | direction | storing | implicit

-------------+------------------+------------+--------------+-----------------------------+-----------+---------+-----------

customers    | primary                        |     false         |                             1 | crdb_internal_id_shard_5000 | ASC               |    false    |    false

customers    | primary                        |     false         |                             2 | id                                                                | ASC               |    false    |    false

customers    | customers_id_key |     false         |                             1 | id                                                                | ASC               |    false    |    false

customers    | customers_id_key |     false         |                             2 | crdb_internal_id_shard_5000 | ASC               |    false    |     true

(4 rows)

Time: 34ms total (execution 26ms / network 8ms)

You can create a hash-sharded secondary index as well.

Duplicate indexes

Duplicate indexes are useful in improving the read performance. Please refer to Chapter 4, Geo-Partitioning, where we have already discussed duplicate indexes and how they work internally.

Next, we will learn about inverted Indexes.

Inverted indexes

Inverted indexes store the mapping of values within JSONB, arrays, and spatial data to the row that holds that value. For example, if you have a column where you are storing a JSON document, and let’s say that JSON document contains a key called  country , then you can add a  WHERE  clause in your query, where you can say get me all the rows that have  country:USA  and  country:Canada .

Inverted indexes filter on components of tokenizable data. The JSONB data type is built on two structures that can be tokenized:

- Objects - Collections of key and value pairs where each key-value pair is a token.

- Arrays – Lists of values where every value in the array is a token.

Let’s look at the following JSON document:

"student": [

{

"id":"01",

"firstname": “Steve”,

"lastname": “Jobs”

},

{

"id":"02",

"firstname": “Steve”,

"lastname": “Wozniak”

}

]

Now, the inverted index for the preceding JSON will have an entry for each component, which maps to the original document as follows:

"student" :    “id” : “01”

"student" :    “firstname” : “Steve”

"student" :    “lastname” : “Jobs”

"student" :    “id” : “02”

"student" :    "lastname": “Wozniak”

Now you can search the JSON document based on student ID, student first name, student last name, and so on.

Partial indexes

A partial index is typically created based on a Boolean expression. CockroachDB internally indexes the columns and rows that evaluate to true for a given expression.

Let’s understand a partial index with an example:

1. First, we will create the table  books  with a few columns:

root@localhost:26257/test> create table books ( id int, title string, author string, price float );

CREATE TABLE

Time: 270ms total (execution 269ms / network 1ms)

2. Let’s create the partial index based on the price of the book. Here, we are creating an index for all the books that are priced more than $50:

root@localhost:26257/test> CREATE INDEX ON books (id, title, author) WHERE price > 50.00;

CREATE INDEX

Time: 2.596s total (execution 0.314s / network 2.282s)

3. Now, whenever you use a filtering condition that matches with the one in the partial index, a partial index will be used to retrieve a subset of rows:

root@localhost:26257/test> explain select id, name, author from books where price > 50.0;

tree |                   field                   |                                            description

-------+---------------------+-------------------------------------------------

| distribution                   | full

| vectorized                        | false

scan |                                                  |

| estimated row count | 1

| table                                   | books@books_id_name_author_idx (partial index)

| spans                                   | FULL SCAN

(6 rows)

Time: 2ms total (execution 1ms / network 1ms)

Partial indexes improve the query performance in the following ways:

- They contain fewer rows than full indexes. During read queries, only rows in the partial index are scanned, if there is a match in the filtering condition. Since partial indexes contain fewer rows compared to regular indexes, we will be scanning fewer rows, so it performs better than a regular index.

- Write queries on tables with a partial index only perform an index write when the rows inserted satisfy the partial index predicate, unlike regular indexes, which are updated during every write.

- In the next section, we will learn about spatial indexes.

Spatial indexes

Spatial indexes were introduced in the v20.2.16 version,  are used to store information about spatial objects, and mostly work with two-dimensional data types such as GEOMETRY and GEOGRAPHY. A spatial object holds information about a geographical location in the form of an object. Here, an object can be a point, a line, a polygon, or an area.

A spatial index is a special type of inverted index. A spatial index maps from a cell in a quadtree to one or more shapes whose coverings include that cell. Each cell can be part of multiple shapes, where a given cell represents a location.

Spatial indexes are useful in the following situations:

- We are filtering based on spatial predicate functions, for example,  ST_COVERS(*) ,  ST_CONTAINS ,  ST_Equals ,  ST_Overlaps , and so on.

- Joins that involve columns that store spatial objects.

CockroachDB uses the S2 geometry library (https://s2geometry.io/) to divide the space being indexed and stores the information in a quadtree data structure.

A quadtree is a tree data structure in which each internal node has exactly four children. Each cell in a quadtree has information about four child cells in the next level. In the following diagram, you can see an example of how an image can be represented using a quadtree data structure:

Figure: 6.2 - Image representation using a Quadtree data structure

Following are some examples of creating spatial indexes:

First, let’s create a table with  GEOGRAPHY  and  GEOMETRY  columns:

root@localhost:26258/test> CREATE TABLE geo_table (

id UUID PRIMARY KEY,

geog GEOGRAPHY(GEOMETRY,4326) NULL,

geom GEOMETRY(GEOMETRY,3857) NULL

);

CREATE TABLE

Time: 151ms total (execution 149ms / network 2ms)

Following is an example of creating a spatial index on a  GEOMETRY  object with default settings:

root@localhost:26258/test> CREATE INDEX geom_idx_1 ON geo_table USING GIST(geom);

CREATE INDEX

Time: 1.647s total (execution 0.137s / network 1.511s)

Following is an example of creating a spatial index on a  GEOGRAPHY  object with default settings:

root@localhost:26258/test> CREATE INDEX geog_idx_1 ON geo_table USING GIST(geog);

CREATE INDEX

Time: 1.709s total (execution 0.144s / network 1.564s)

Fine-tuning spatial indexes is beyond the scope of this book and will be covered in subsequent editions.

Next, we will learn how to improve the performance of queries that involve table joins.

Table joins and indexes

Indexes are useful even when you are joining tables. You can inspect the fields that are used in filtering conditions and create appropriate indexes to avoid a full scan of other indexes.

For example, let’s look at two tables:  customers  and  purchase_orders .

The  customers  table stores the information about the customers as follows:

root@localhost:26258/test> CREATE TABLE customers (

id UUID PRIMARY KEY,

name STRING,

email STRING,

phone STRING

);

CREATE TABLE

Time: 209ms total (execution 209ms / network 0ms)

The  purchase_orders  table stores information about the purchase orders made by the customers. Here, the  customer_id  column references the  id  column of the  customers  table:

root@localhost:26258/test> CREATE TABLE purchase_orders (

id UUID PRIMARY KEY,

customer_id UUID NOT NULL REFERENCES customers ( id ),

n_of_items INT,

total_price DECIMAL(10,2)

);

CREATE TABLE

Time: 843ms total (execution 264ms / network 579ms)

Now, let’s say you want to know the name of all the customers who have purchased more than five items. Following is the  explain  plan for this:

root@localhost:26258/test> explain select n_of_items, name from purchase_orders    INNER JOIN customers ON purchase_orders.customer_id = customers.id and n_of_items > 5;

tree              |                   field                   |               description

-----------------+---------------------+--------------------------

| distribution                   | full

| vectorized                        | false

hash join              |                                                  |

│                              | equality                             | (customer_id) = (id)

│                              | right cols are key    |

├── filter         |                                                  |

│         │                   | filter                                  | n_of_items > 5

│         └── scan |                                                  |

│                              | estimated row count | 1

│                              | table                                   | purchase_orders@primary

│                              | spans                                   | FULL SCAN

└── scan              |                                                  |

| estimated row count | 1

| table                                   | customers@primary

| spans                                   | FULL SCAN

As you can see, we are making use of primary indexes on both the  customers  and  purchase_orders  tables. But we also have a filtering condition for which we are using the column  n_of_items . We can further improve the query performance by adding one more index for the column  n_of_items :

CREATE INDEX ON purchase_orders (n_of_items) STORING (customer_id);

Now, let’s once again look at the  explain  plan for the previous query:

root@localhost:26258/test> explain select n_of_items, name from purchase_orders    INNER JOIN customers ON purchase_orders.customer_id = customers.id and n_of_items > 5;

tree         |                   field                   |                                            description

------------+---------------------+-------------------------------------------------

| distribution                   | full

| vectorized                        | false

hash join |                                                  |

│                   | equality                             | (customer_id) = (id)

│                   | right cols are key    |

├── scan |                                                  |

│                   | estimated row count | 1

│                   | table                                   | purchase_orders@purchase_orders_n_of_items_idx

│                   | spans                                   | [/6 - ]

└── scan |                                                  |

| estimated row count | 1

| table                                   | customers@primary

| spans                                   | FULL SCAN

As you can see, now we no longer do a full scan of the  purchase_orders@primary  index. So, based on the columns used in filtering conditions during table joins, you can create appropriate indexes.

Next, we will go over some of the best practices to consider related to indexes that can help in improving the query performance.

Best practices while using indexes

Whenever you are using indexes, you can follow certain guidelines to make sure you get the best performance for your queries. Following are some of the key points to remember:

- Avoid creating an index on a sequence. Due to the nature of how the columns are sharded, sometimes we can have range hotspots, where most of the requests are coming for the same range, which can slow down the query. It would be best to use UUIDs or randomly generated keys. If you must create an index on a column that is sequential in nature, you should use hash-sharded indexes, as discussed in the Hash-sharded indexes section.

- If you are using multiple columns in your  WHERE  clause or in your  ORDER BY  clause, you should consider creating an index for all these columns.

- In your  WHERE  clause, make sure to have filters that are more restrictive before the ones that are a bit more generic. For example,  =  and  IN  should come before  LIKE ,   > ,  != , and so on.

- You should drop the indexes that are not getting used. This will improve the write performance, as fewer indexes will have to be updated. Right now, there is no easy way to know the unused indexes. This requires manually going through the logical plans and identifying the indexes that are not getting used. You can use  DROP INDEX  to drop a specific index.

For example, let’s drop an index created previously, in the Partial indexes section:

root@localhost:26258/test> DROP INDEX books@books_id_name_author_idx;

NOTICE: the data for dropped indexes is reclaimed asynchronously

HINT: The reclamation delay can be customized in the zone configuration for the table.

DROP INDEX

Time: 1.659s total (execution 0.178s / network 1.482s)

After this, if you execute  SHOW INDEXES , you should not see the  books@books_id_name_author_idx  index:

root@localhost:26258/test> SHOW INDEXES FROM books;

table_name | index_name | non_unique | seq_in_index | column_name | direction | storing | implicit

-------------+------------+------------+--------------+-------------+-----------+---------+-----------

books              | primary         |     false         |                             1 | rowid               | ASC               |    false    |    false

(1 row)

Time: 17ms total (execution 17ms / network 0ms)

We have discussed various types of indexes other than primary and secondary. Make sure you understand these specialized indexes and use them appropriately.

You can also select a specific index in the query if you think that’s going to improve the performance.

Summary

In this chapter, we learned about indexes, several special types of indexes, how they work internally, and the best practices for maximum query performance. It is important to understand the columns in your tables and query patterns and pick relevant indexes for maximum query performance.

In the next chapter, we will learn about high availability and how to deploy CockroachDB in order to achieve zero downtime and to make it highly available.