Engineering Atlas › System Design Fundamentals

Wide-column stores for huge write loads

12 min read · last reviewed 2026-10-17

Open in the interactive app: practice, quiz, mock interview →

In one minute

How it works

Data model

Think of each table as a two-level map. The outer level is the partition key, hashed (Cassandra) or range-sorted (Bigtable, HBase) to place it on nodes. The inner level is a sorted map from clustering key to a set of cells. A query that names a partition and a clustering range reads one contiguous slice from one replica set, which is the fast path. A query that does not name the partition must ask every node.

Every cell carries a write timestamp. Concurrent writes to the same cell resolve by last write wins on that timestamp (Cassandra docs), and deletes write a tombstone with a timestamp that shadows older cells.

Two architectures

Bigtable and HBase. The key space is split into contiguous ranges called tablets or regions, each served by exactly one server at a time. A master assigns tablets and splits them as they grow. Data and logs live in a distributed file system (Colossus or HDFS), so a server failure means another server reopens the tablet from shared storage; in Bigtable only metadata moves to the replacement node, which is why recovery is fast (Google Cloud docs). Bigtable now also accepts read-only GoogleSQL queries, but they run on the same nodes and should still filter on the row key (Google Cloud docs). Consistency per row is strong because one server owns each row.

Cassandra and ScyllaDB. Nodes sit on a consistent-hash ring with virtual nodes. Each partition is replicated to N nodes (usually 3), and any node can coordinate a request. The client chooses consistency per query: QUORUM means a majority (n/2 + 1) of replicas, so QUORUM writes plus QUORUM reads overlap on at least one replica (Cassandra docs), so reads see the latest acknowledged write; ONE is faster and weaker. Hinted handoff, read repair, and periodic anti-entropy repair push replicas back into agreement.

Write and read path

A write goes to a commit log and a memtable on each replica, then flushes to SSTables, exactly as in any LSM tree. A read gathers cells for the requested partition from the memtable and the SSTables that might hold it (Bloom filters prune the rest), merges by timestamp, and drops shadowed cells.

Modelling rules

Down the stack

Layer 1The contract

Cassandra promises, per partition, a sorted set of rows by clustering key and per-cell last-write-wins by timestamp. A write at consistency level QUORUM returns success once a majority of the partition's replicas have appended it to their commit log and memtable; a QUORUM read then sees it, provided clocks that generate timestamps are sane. There is no isolation across partitions. A logged batch guarantees that all its mutations will eventually apply, not that readers never see a partial batch. Compare-and-set (IF NOT EXISTS) runs Paxos, costing several round trips. Bigtable and HBase instead promise strongly consistent single-row reads and writes, because exactly one server owns each row at a time.

Layer 2The mechanism

A Cassandra mutation is serialised and appended to the active commit log segment (32 MiB by default), then applied to the memtable for its table. The default commitlog_sync: periodic fsyncs the commit log every 10 seconds rather than per write (Cassandra docs), so a single node's power loss can drop recent acknowledged writes. Cassandra accepts this because the write also exists on other replicas; batch and group modes fsync before acknowledging at a latency cost.

When a memtable flushes, it becomes an SSTable made of several component files: Data.db holds partitions in token order, compressed with LZ4 in chunks (16 KiB by default in recent versions, Cassandra docs); CompressionInfo.db maps uncompressed offsets to chunk positions; Filter.db is the partition Bloom filter; Statistics.db holds min and max timestamps and tombstone counts; and a partition index (Index.db plus Summary.db in the classic format, or a trie-based Partitions.db and Rows.db in the BTI format added in 5.0, which you opt into with sstable.selected_format: bti) maps partition keys to offsets in Data.db (Cassandra docs). Compaction strategies decide what merges with what: size-tiered (still the table default), leveled, time-window for time series, and the unified strategy added in 5.0 (Cassandra 5.0 features). Tombstones survive at least gc_grace_seconds (864,000 seconds, 10 days, by default, Cassandra docs) so that repair can propagate the delete before it is purged; otherwise a replica that missed the delete would resurrect the data.

Layer 3The machine

The write path is one sequential append plus an in-memory insert, which is why Cassandra sustains high write rates on modest hardware. The read path pays the costs. For each SSTable that might hold the partition, the node checks the in-memory Bloom filter, looks up the partition index, then reads and decompresses at least one whole compression chunk, even to return a 100-byte row. Small chunks reduce that read amplification; large chunks compress better. Partitions are placed by a 64-bit Murmur3 token, so related keys scatter across the ring by design.

The JVM is the other machine-level cost. Memtables, caches, and per-request objects generate garbage, and garbage collection pauses show up directly in p99 latency; Cassandra moved Bloom filters, compression metadata, and optionally memtables off-heap to reduce this. ScyllaDB rewrote the engine in C++ on Seastar, pinning one shard per CPU core with no shared memory between cores, and does its own caching with direct I/O instead of relying on the OS page cache.

Why it's built this way

  1. Writes must be cheap at very high rates, so each one is a sequential log append plus a memory insert, and on-disk files are immutable.
  2. Immutable files mean updates and deletes are new versions, so every cell carries a timestamp and deletes are tombstones.
  3. Any replica may miss a write, so tombstones must outlive repair windows or deleted data comes back.
  4. Reads must avoid scanning every node and file, so the partition key routes a query to a few replicas and Bloom filters prune SSTables.
  5. Per-write fsync on every node is expensive and replication already provides redundancy, so the default leans on replicas for durability.

Diagram

flowchart LR
    C[Client] --> CO["Coordinator node"]
    CO -->|"hash(partition key)"| RING{"Token ring"}
    RING --> R1["Replica A"]
    RING --> R2["Replica B"]
    RING --> R3["Replica C"]
    R1 --> P1["Partition: user 42"]
    P1 --> ROWS["Rows sorted by clustering key"]
    ROWS --> CELLS["Cells: value + timestamp"]

Worked example

A messaging app stores 2 billion messages per day, each about 300 bytes, and serves "last 50 messages in this conversation" at 150,000 reads per second.

The table is keyed PRIMARY KEY ((conversation_id, month), sent_at DESC, message_id). A read names the partition and takes the first 50 clustering rows, so it touches one partition on 2 of 3 replicas at LOCAL_QUORUM. Monthly bucketing caps a busy group chat at a few hundred thousand rows. Without it, a five-year-old company channel would grow into a multi-gigabyte partition that is slow to read, repair, and compact.

Raw volume is 600 GB per day; with replication factor 3 and LSM overhead, roughly 2.2 TB per day of disk. At 4 TB usable per node and one year of retention, the cluster needs around 200 nodes before compression, closer to 100 with it. Writes at 23,000 per second average and 100,000 at peak are well within what 100 nodes absorb, since each Cassandra node handles tens of thousands of writes per second. The design constraint is disk and compaction, not CPU.

In the wild

Trade-offs

OptionStrengthsWeaknessesChoose when
Cassandra or ScyllaDBMulti-region active-active, no single leader, linear write scaleQuery-driven modelling, no joins, LWW conflictsHigh write volume, known access patterns, multi-region
Bigtable or HBaseStrong per-row consistency, range scans, autosplitSingle owner per tablet, region failover pauseLarge sorted scans, analytics-adjacent serving
Relational databaseJoins, transactions, flexible queriesHarder to scale writes horizontallyData fits a few nodes or needs ad hoc queries
Document databaseFlexible nested documents, secondary indexesWeaker at huge sorted time rangesEntity-centric access with varied shapes

Pitfalls in production

Leadership lens

Check yourself

  1. Why does a query without the partition key perform badly in Cassandra?

    Show answer

    The coordinator cannot tell which nodes hold matching rows, so it must ask all of them and merge the results.

  2. How do QUORUM reads and writes give read-your-writes with replication factor 3?

    Show answer

    Each touches at least 2 of 3 replicas, so any read set overlaps any write set on at least one replica holding the newest timestamp.

  3. Why add a month to a conversation's partition key?

    Show answer

    It caps partition size, keeping reads, compaction, and repair cheap even for very active conversations.

  4. What makes deleted data reappear in a leaderless store?

    Show answer

    A replica missed the tombstone, the tombstone was purged after the grace period, and repair later copied the old value back from that replica.

Sources and further study

Primary

Systems docs

Books

Other curricula

Watch

Free resources

Open in the interactive app: practice, quiz, mock interview →