LabHub

Blog

CockroachDB Distributed SQL and NewSQL Operations Guide

한국어English日本語

CockroachDB

Introduction

Traditional RDBMSs rely on the vertical scaling (Scale-Up) of a single node, and they run into limits on the horizontal scalability and geographic distribution that global services demand. NoSQL solved the scalability problem but had to give up ACID transactions and SQL compatibility. NewSQL is an attempt to combine the strengths of both worlds, and CockroachDB is the flagship open-source distributed SQL database, born out of inspiration from the Google Spanner paper.

CockroachDB is compatible with the PostgreSQL wire protocol while providing automatic sharding, multi-region replication, and distributed transactions at the Serializable isolation level out of the box. True to its name, its design philosophy is to survive any failure the way a cockroach does. In this article, we analyze CockroachDB's internal architecture in depth and systematically cover cluster construction, schema design, performance tuning, and failure recovery procedures in real production environments.

1. Understanding NewSQL and Distributed SQL

What Is NewSQL

NewSQL is a term first used in 2011 by Matthew Aslett. It refers to the category of databases that keep the ACID guarantees and the SQL interface of traditional RDBMSs while providing NoSQL-level horizontal scalability. Its core characteristics are as follows.

CharacteristicTraditional RDBMSNoSQLNewSQL
SQL supportFullLimited / unsupportedFull
ACID transactionsFullPartial (eventual consistency)Full
Horizontal scalingImpossible / limitedNativeNative
Automatic shardingUnsupportedSupportedSupported
Distributed transactionsUnsupportedUnsupportedSupported
Geographic distributionManual replication onlyPossibleNative

How Distributed SQL Relates to NewSQL

Distributed SQL is one way of implementing NewSQL. Every node can process SQL queries, data is automatically spread across multiple nodes, and distributed transactions guarantee consistency through a consensus protocol. CockroachDB, YugabyteDB, and TiDB fall into this category, and Google Cloud Spanner is the pioneer of the field.

2. A Deep Dive into the CockroachDB Architecture

The Layered Architecture

CockroachDB is made up of four major layers. Each layer operates independently while exposing an abstracted interface to the layer above it.

┌─────────────────────────────────────────────────────────┐
SQL Layer  (SQL parse, optimize, plan, PostgreSQL compatibility)├─────────────────────────────────────────────────────────┤
Transaction Layer  (Distributed ACID transactions, MVCC, timestamp oracle)├─────────────────────────────────────────────────────────┤
Distribution Layer  (Range splitting, Leaseholder election, meta ranges)├─────────────────────────────────────────────────────────┤
Replication Layer  (Raft consensus, MultiRaft, snapshot transfer)├─────────────────────────────────────────────────────────┤
Storage Layer  (Pebble - LSM Tree based key-value storage engine)└─────────────────────────────────────────────────────────┘

SQL Layer: It implements the PostgreSQL wire protocol, so existing PostgreSQL client drivers can be used as they are. The SQL parser turns a query into an AST, and then the cost-based optimizer (CBO) produces the optimal execution plan.

Transaction Layer: It uses multi-version concurrency control (MVCC) so that reads and writes do not block each other. It tracks causality in a distributed environment with an HLC (Hybrid Logical Clock), and its default isolation level is Serializable.

Distribution Layer: It splits the entire key space into Ranges of 512MiB each and distributes them across the whole cluster. Every Range has a Leaseholder, which serves read requests directly.

Replication Layer: It keeps 3 replicas of each Range by default and guarantees consistency between replicas with the Raft consensus algorithm.

Storage Layer: It used to use RocksDB, but has since moved to Pebble, an engine the CockroachDB team built themselves in Go. Being based on an LSM (Log-Structured Merge-Tree), it is optimized for write performance.

The Raft Consensus Algorithm and MultiRaft

CockroachDB guarantees data consistency through the Raft consensus protocol. Every write operation is routed to the Raft leader of the Range, and it is committed only once a majority (Quorum) of the replicas has acknowledged that write. With 3 replicas, 2 acknowledgements are needed; with 5 replicas, 3 acknowledgements are needed.

A single CockroachDB cluster can hold hundreds of thousands of Ranges. Running an independent Raft group for every Range makes the heartbeat and message-processing overhead explode. To solve this, CockroachDB implemented MultiRaft. It batches the Raft messages of the several Ranges that sit on the same node into one transmission and consolidates heartbeats at the node level, cutting network overhead dramatically.

MultiRaft message flow:

Node 1                    Node 2                    Node 3
┌──────────────┐          ┌──────────────┐          ┌──────────────┐
Range 1 (L)  │──┐       │ Range 1 (F)  │          │ Range 1 (F)Range 5 (F)  │  │ batch │ Range 5 (L)  │          │ Range 5 (F)Range 9 (F)  │  ├──────▶│ Range 9 (F)  │          │ Range 9 (L)Range 12 (L) │──┘       │ Range 12 (F) │          │ Range 12 (F)└──────────────┘          └──────────────┘          └──────────────┘
(L)=Leader, (F)=Follower

The Lifecycle of a Distributed Transaction

Let us walk step by step through how a distributed transaction is processed in CockroachDB.

  1. The gateway node receives the request: The client sends the SQL statement to an arbitrary node.
  2. SQL parsing and execution planning: The SQL Layer parses the query and produces an optimized execution plan.
  3. Transaction record creation: The Transaction Layer creates a transaction record in the PENDING state on the Range where the transaction's first write happens.
  4. Writing Intents: An Intent (a provisional write) is recorded on each target key. An Intent tells other transactions that the key is being modified by a transaction that is still in flight.
  5. Consensus through Raft: Each Intent write must obtain majority consensus through the Raft leader of its Range.
  6. Parallel Commit: The transaction record's state is changed to STAGING while consensus for all Intents proceeds in parallel. Once every Intent has reached consensus, the transaction is implicitly COMMITTED.
  7. Intent resolution: Each Intent is asynchronously converted into a regular MVCC value and the transaction record is updated to COMMITTED.

3. Cluster Installation and Production Configuration

The Minimum Production Topology

In a production environment you must place at least 3 nodes in different availability zones (AZs). That way the Raft quorum survives even a single-AZ failure, so the service can keep running.

# Start node 1 (AZ-a)
cockroach start \
  --insecure \
  --advertise-addr=node1.example.com:26257 \
  --join=node1.example.com:26257,node2.example.com:26257,node3.example.com:26257 \
  --locality=region=ap-northeast-2,zone=az-a \
  --store=path=/data/cockroach,attrs=ssd \
  --max-offset=500ms \
  --cache=.25 \
  --max-sql-memory=.25 \
  --background

# Start node 2 (AZ-b)
cockroach start \
  --insecure \
  --advertise-addr=node2.example.com:26257 \
  --join=node1.example.com:26257,node2.example.com:26257,node3.example.com:26257 \
  --locality=region=ap-northeast-2,zone=az-b \
  --store=path=/data/cockroach,attrs=ssd \
  --max-offset=500ms \
  --cache=.25 \
  --max-sql-memory=.25 \
  --background

# Start node 3 (AZ-c)
cockroach start \
  --insecure \
  --advertise-addr=node3.example.com:26257 \
  --join=node1.example.com:26257,node2.example.com:26257,node3.example.com:26257 \
  --locality=region=ap-northeast-2,zone=az-c \
  --store=path=/data/cockroach,attrs=ssd \
  --max-offset=500ms \
  --cache=.25 \
  --max-sql-memory=.25 \
  --background

# Initialize the cluster (once only)
cockroach init --insecure --host=node1.example.com:26257

# Check cluster status
cockroach node status --insecure --host=node1.example.com:26257

The main options:

Kubernetes-Based Deployment

When you deploy CockroachDB on Kubernetes in production, use a StatefulSet, because each Pod has to keep a stable network identity and a persistent volume.

apiVersion: apps/v1
kind: StatefulSet
metadata:
  name: cockroachdb
  namespace: crdb
spec:
  serviceName: cockroachdb
  replicas: 3
  selector:
    matchLabels:
      app: cockroachdb
  template:
    metadata:
      labels:
        app: cockroachdb
    spec:
      affinity:
        podAntiAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            - labelSelector:
                matchExpressions:
                  - key: app
                    operator: In
                    values:
                      - cockroachdb
              topologyKey: topology.kubernetes.io/zone
      containers:
        - name: cockroachdb
          image: cockroachdb/cockroach:v24.3.4
          ports:
            - containerPort: 26257
              name: grpc
            - containerPort: 8080
              name: http
          env:
            - name: COCKROACH_CHANNEL
              value: kubernetes-helm
          command:
            - /cockroach/cockroach
            - start
            - --advertise-addr=$(POD_NAME).cockroachdb.crdb.svc.cluster.local
            - --join=cockroachdb-0.cockroachdb.crdb.svc.cluster.local:26257,cockroachdb-1.cockroachdb.crdb.svc.cluster.local:26257,cockroachdb-2.cockroachdb.crdb.svc.cluster.local:26257
            - --locality=region=ap-northeast-2,zone=$(NODE_ZONE)
            - --cache=2GiB
            - --max-sql-memory=2GiB
            - --logtostderr=WARNING
          resources:
            requests:
              cpu: '2'
              memory: '8Gi'
            limits:
              cpu: '4'
              memory: '8Gi'
          volumeMounts:
            - name: datadir
              mountPath: /cockroach/cockroach-data
  volumeClaimTemplates:
    - metadata:
        name: datadir
      spec:
        accessModes: ['ReadWriteOnce']
        storageClassName: gp3-encrypted
        resources:
          requests:
            storage: 100Gi

Key configuration points:

4. Designing a Multi-Region Topology

Survival Goals and Table Locality

CockroachDB's multi-region capability rests on two core concepts.

Survival Goal:

Table Locality:

-- Configure regions on the database
ALTER DATABASE myapp PRIMARY REGION "ap-northeast-2";
ALTER DATABASE myapp ADD REGION "us-east-1";
ALTER DATABASE myapp ADD REGION "eu-west-1";

-- Configure it to survive a region failure
ALTER DATABASE myapp SURVIVE REGION FAILURE;

-- Users table: distributed across regions row by row
CREATE TABLE users (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email STRING NOT NULL UNIQUE,
    name STRING NOT NULL,
    region crdb_internal_region NOT NULL DEFAULT 'ap-northeast-2',
    created_at TIMESTAMPTZ DEFAULT now()
) LOCALITY REGIONAL BY ROW AS region;

-- Config table: fast reads from every region (rarely updated)
CREATE TABLE app_config (
    key STRING PRIMARY KEY,
    value JSONB NOT NULL,
    updated_at TIMESTAMPTZ DEFAULT now()
) LOCALITY GLOBAL;

-- Orders table: Leaseholder pinned to the Korean region
CREATE TABLE orders (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    user_id UUID NOT NULL REFERENCES users(id),
    total_amount DECIMAL(12,2) NOT NULL,
    status STRING NOT NULL DEFAULT 'pending',
    created_at TIMESTAMPTZ DEFAULT now()
) LOCALITY REGIONAL BY TABLE IN "ap-northeast-2";

Optimizing Multi-Region Write Latency

In a multi-region environment, write latency inevitably grows, because reaching a Raft quorum requires a network round trip between regions. If the round-trip latency between Seoul and Virginia is about 180ms, a write commit takes at least 180ms. The strategies for mitigating this are as follows.

  1. Make aggressive use of REGIONAL BY ROW: When a user's data sits in the region that user connects from, most writes complete inside that region.
  2. Asynchronous write patterns: Put requests that need an immediate response into a local queue and handle the actual distributed write in the background.
  3. Use Follower Reads: For reads that can tolerate slightly stale data, use a Follower Read so that you do not have to travel to the Leaseholder in a remote region.
-- Follower Read: read data from up to 4.8 seconds ago off a local replica
SELECT * FROM users
AS OF SYSTEM TIME follower_read_timestamp()
WHERE region = 'us-east-1';

-- Bounded Staleness Read: guarantees data no more than 10 seconds stale
SELECT * FROM app_config
AS OF SYSTEM TIME with_max_staleness('10s');

5. Online Schema Changes and CDC

How Online DDL Works

CockroachDB performs online schema changes without locking the table. Internally it uses a bridging strategy that transitions schema versions incrementally, so data consistency holds even when the nodes of the cluster do not all switch to the new schema at the same time.

-- Add a column to a large table (online, no downtime)
ALTER TABLE orders ADD COLUMN shipping_address STRING;

-- Create an index (built incrementally in the background)
CREATE INDEX CONCURRENTLY idx_orders_status_created
ON orders (status, created_at DESC);

-- Monitor schema change progress
SELECT job_id, job_type, description, status, fraction_completed
FROM [SHOW JOBS]
WHERE job_type = 'SCHEMA CHANGE'
ORDER BY created DESC
LIMIT 5;

Cautions when changing a schema:

Change Data Capture (CDC) Configuration

CockroachDB's Changefeed streams table changes in real time to external systems (Kafka, Cloud Storage, and so on). It is used as a core building block of event-driven architectures.

-- Stream change events to Kafka
CREATE CHANGEFEED FOR TABLE orders, users
INTO 'kafka://kafka-broker1:9092?topic_prefix=crdb_'
WITH
  format = 'json',
  updated,
  resolved = '10s',
  min_checkpoint_frequency = '30s',
  schema_change_policy = 'backfill',
  kafka_sink_config = '{"Flush": {"MaxMessages": 1000, "Frequency": "1s"}}';

-- Check Changefeed status
SELECT job_id, description, status, error, running_status
FROM [SHOW JOBS]
WHERE job_type = 'CHANGEFEED'
ORDER BY created DESC;

-- Pause and resume a Changefeed when a problem occurs
PAUSE JOB (SELECT job_id FROM [SHOW JOBS] WHERE job_type = 'CHANGEFEED' AND status = 'running' LIMIT 1);
RESUME JOB <job_id>;

Key recommendations for operating Changefeeds:

6. Comparing NewSQL Databases

CockroachDB vs TiDB vs YugabyteDB vs Spanner

Comparison itemCockroachDBTiDBYugabyteDBGoogle Cloud Spanner
Foundational paperGoogle SpannerGoogle Spanner + F1Google SpannerDeveloped in-house by Google
SQL compatibilityPostgreSQLMySQLPostgreSQL + Cassandra QLProprietary SQL (GoogleSQL)
Consensus algorithmRaft (MultiRaft)Raft (TiKV level)Raft (DocDB level)Paxos
Storage enginePebble (Go, LSM)RocksDB (C++, LSM)DocDB (C++, LSM)Colossus (in-house)
Default isolation levelSerializableSnapshot IsolationSnapshot IsolationExternal Consistency
Clock synchronizationHLC + NTPTSO server (centralized)HLC + NTPTrueTime (atomic clocks)
HTAP supportLimitedTiFlash integration (strong)LimitedLimited
LicenseBSL (Apache after 3 years)Apache 2.0Apache 2.0Managed offering only
Managed serviceCockroachDB CloudTiDB CloudYugabyteDB ManagedGCP only
Multi-regionNativeTiDB DashboardNativeNative
Free tierServerless (free)Cloud free planTrial90 days free

Where Each Database Fits Best

When CockroachDB is a good fit:

When TiDB is a good fit:

When YugabyteDB is a good fit:

When Google Cloud Spanner is a good fit:

7. Performance Tuning in Practice

Query Performance Analysis

CockroachDB offers an EXPLAIN ANALYZE feature similar to PostgreSQL's, but it has the distinctive property of a distributed execution plan.

-- Inspect the distributed execution plan
EXPLAIN ANALYZE
SELECT o.id, o.total_amount, u.name
FROM orders o
JOIN users u ON o.user_id = u.id
WHERE o.status = 'pending'
  AND o.created_at > now() - INTERVAL '24 hours'
ORDER BY o.created_at DESC
LIMIT 50;

-- Example output:
-- planning time: 2ms
-- execution time: 45ms
-- distribution: full (executed in parallel on 3 nodes)
-- vectorized: true
--
-- • limit (count: 50)
-- │ estimated row count: 50
-- │
-- └── • sort (order: created_at DESC)
--     │
--     └── • lookup join
--         │ table: users@primary
--         │ equality: (user_id) = (id)
--         │
--         └── • filter
--             │ filter: (status = 'pending') AND (created_at > ...)
--             │
--             └── • scan
--                   table: orders@idx_orders_status_created
--                   spans: [/'pending' - /'pending']

Key Performance Tuning Parameters

-- Cluster-level settings
SET CLUSTER SETTING kv.rangefeed.enabled = true;
SET CLUSTER SETTING kv.range_merge.queue_enabled = true;
SET CLUSTER SETTING sql.defaults.distsql = 'auto';

-- Adjust the automatic statistics collection cadence (large tables)
SET CLUSTER SETTING sql.stats.automatic_collection.min_stale_rows = 500;

-- Session-level optimization
SET vectorize = 'on';
SET distsql = 'auto';
SET reorder_joins_limit = 6;

-- Minimize range locking during large bulk operations
SET CLUSTER SETTING kv.bulk_io_write.max_rate = '256MiB';

Index Strategy

In CockroachDB, index design has to take the characteristics of a distributed environment into account.

-- Composite index: order the columns to match the query pattern
CREATE INDEX idx_orders_user_status ON orders (user_id, status)
STORING (total_amount, created_at);

-- Partial index: index only active orders to save storage
CREATE INDEX idx_active_orders ON orders (user_id, created_at DESC)
WHERE status IN ('pending', 'processing');

-- Descending index: for patterns that query the newest data often
CREATE INDEX idx_orders_recent ON orders (created_at DESC)
STORING (status, total_amount);

-- Check index usage
SELECT ti.index_name,
       ts.range_count,
       ts.approximate_disk_bytes / 1024 / 1024 AS size_mb,
       s.statistics->>'query_count' AS query_count
FROM crdb_internal.table_indexes ti
JOIN crdb_internal.index_usage_statistics s
  ON ti.index_id = s.index_id AND ti.descriptor_id = s.table_id
JOIN crdb_internal.table_span_stats ts
  ON ti.descriptor_id = ts.table_id
WHERE ti.descriptor_name = 'orders'
ORDER BY query_count DESC;

8. Failure Scenarios and Recovery Procedures

Scenario 1: Single Node Failure

This is the most common kind of failure. In a 3-replica configuration, even if 1 node goes down the remaining 2 form a quorum, so the service is not interrupted.

Detection: Check for the DEAD or SUSPECT state under Node Status in CockroachDB's built-in UI.

# Check node status
cockroach node status --insecure --host=node1.example.com:26257

# Check the logs of a specific node
cockroach debug zip /tmp/debug.zip --insecure --host=node1.example.com:26257

# Safely decommission a failed node (when it cannot be recovered)
cockroach node decommission <node_id> --insecure --host=node1.example.com:26257

Recovery procedure:

  1. Identify the cause of the failure (disk failure, OOM, network, and so on)
  2. Restart the node once the cause is resolved - CockroachDB automatically applies the Raft log to resynchronize it
  3. If it cannot be recovered, remove it safely with node decommission and then add a new node
  4. Confirm that Ranges short of replicas automatically create new replicas on other nodes

Scenario 2: Full Region Failure

A database configured with SURVIVE REGION FAILURE keeps serving even if one region disappears completely. That said, Ranges whose Leaseholder was in the failed region see temporary latency while a new Leaseholder is elected.

Recovery procedure:

  1. Wait until the nodes in the failed region are automatically judged DEAD (5 minutes by default)
  2. Confirm that Raft leader re-election proceeds automatically in the remaining regions
  3. Once the failed region is restored, restarting the nodes rejoins them to the cluster automatically
  4. Monitor latency until Leaseholder redistribution finishes

Scenario 3: Clock Synchronization Failure

In CockroachDB, when the clock offset between nodes exceeds 80% of the --max-offset value, that node shuts itself down voluntarily. An NTP service failure can trigger this situation.

# Check the clock offset between nodes
cockroach debug timeutil --insecure --host=node1.example.com:26257

# Check NTP synchronization status
chronyc tracking
chronyc sources -v

# Force NTP synchronization
sudo chronyc -a makestep

Preventive measures:

Scenario 4: Range Under-Replication

When Ranges end up with fewer replicas than the target (3 by default), data safety is at risk.

-- Find under-replicated ranges
SELECT range_id, start_key, end_key, replicas, lease_holder
FROM crdb_internal.ranges_no_leases
WHERE array_length(replicas, 1) < 3;

-- Check replication queue status
SELECT store_id, queue_name, process_count, process_failure_count
FROM crdb_internal.kv_store_status
WHERE queue_name = 'replicate';

9. Jepsen Testing and Consistency Guarantees

CockroachDB continuously runs the Jepsen tests that verify the consistency of distributed systems. Jepsen is a framework developed by Kyle Kingsbury (Aphyr) that injects failures such as network partitions, process crashes, and clock skew while verifying a database's consistency guarantees.

CockroachDB's consistency properties:

The CockroachDB team automatically runs Jepsen tests every night on a 5-node cluster, combining 7 workloads with a variety of failure scenarios, and it has fixed every bug found over more than 2 years of continuous testing (inconsistencies related to the timestamp cache, duplicate application caused by internal retries, and so on).

10. Operational Monitoring and Alerting

Key Monitoring Metrics

CockroachDB's built-in UI (port 8080 by default) offers rich dashboards. Using Prometheus and Grafana alongside it makes long-term trend analysis and custom alerts possible.

# Prometheus scrape configuration
scrape_configs:
  - job_name: 'cockroachdb'
    metrics_path: '/_status/vars'
    scheme: 'http'
    static_configs:
      - targets:
          - 'node1.example.com:8080'
          - 'node2.example.com:8080'
          - 'node3.example.com:8080'
    relabel_configs:
      - source_labels: [__address__]
        target_label: instance
        regex: '(.+):8080'
        replacement: '${1}'

Key alerting rules:

MetricThresholdMeaning
liveness_livenodesLower than expectedA node has failed
ranges_underreplicated> 0 (sustained 5 min or more)Too few replicas, data at risk
ranges_unavailable> 0Quorum lost, service impacted
sql_service_latency_p99> 500msQuery performance degraded
admission_io_overload> 0.8Disk I/O bottleneck
clock_offset_mean> max_offset * 0.6Clock synchronization at risk
capacity_used_percent> 70%Approaching disk capacity exhaustion

11. Production Operations Checklist

Before Deploying the Cluster

Schema Design

In Production

Preparing for Failures

Conclusion

CockroachDB abstracts away a large part of the complexity of a distributed SQL database, letting a developer who is comfortable with PostgreSQL build a globally distributed system at a relatively low learning cost. Strong consistency guarantees through Raft consensus, native multi-region support, online schema changes, and Serializable isolation out of the box make CockroachDB an attractive choice for services where data consistency matters, such as finance, e-commerce, and SaaS.

It is not a silver bullet, however. Because it is HLC-based, it does not provide external consistency as strong as Google Spanner's TrueTime, and it still lacks powerful HTAP features like TiDB's TiFlash. Multi-region write latency is a constraint of physics, so it has to be accounted for at the architecture design stage. The BSL licensing policy is another part that calls for legal review when you evaluate adoption.

The key point is to choose the database that fits your own workload and requirements. I hope the architectural understanding, cluster configuration, multi-region topology design, performance tuning, and failure recovery procedures covered in this article serve as a practical guide for anyone evaluating CockroachDB or already running it.

References

Comments

No comments yet.

Sign in to leave a comment