Inside Cassandra: ring, gossip, replicas, quorums; HBase

Where a key lives: the hash ring

Take the paper of 12 Sep 2025: its farmers' sales, "large-scale, continuously increasing", go to Cassandra, to be queried "by market location or product name". Partition them by market location: every Milano sale lands in the partition 'Milano'. Eight machines hold them as equals: "NO MASTER", "NO SLAVE", "PEER TO PEER" (p. 440). So who answers p. 439, "How do you decide which server(s) a key-value resides on?"

Every node, alone, with the same arithmetic. The partitioner turns the key into a number, its token: "system will use MD5(key) to distribute data across nodes" (p. 457). Tokens lie on a circle; each node owns one token, and a key belongs to the first node clockwise from its own token. The replication factor N counts the copies, and the Simple Strategy places the other N − 1 by "Clockwise placement to the next node(s)" (p. 442). A client may send a request to any node, which becomes that request's coordinator (p. 443).

Why a circle, and not the hash modulo the number of nodes? With modulo 8, a ninth machine changes the remainder of almost every key: nearly all the data moves. On the ring a new node takes one arc from one neighbour. That is consistent hashing (background: the slides draw the ring without naming it).

Two partitioners

The "partitioner smack-down" (p. 457) sets the two side by side (Table 1). Keeping keys in order lets the order-preserving one scan a range of keys, at the price of the "'scrabble' distribution": real keys are as uneven as Scrabble letters, and the node owning the M's gets far more rows than the one owning the X's. The hash is the default (background): Cassandra gives up range scans over partition keys and keeps order inside a partition instead, with the clustering key of lesson 21.

Where the copies go

"data replication" (p. 456) needs a "configurable replication factor" and a "replica placement strategy". The slide strikes out the old, more telling names: rack unaware → Simple Strategy; rack aware → Old Network Topology Strategy (since removed, background); data center shard → Network Topology Strategy. The Simple Strategy serves a "Single datacenter" (p. 442) and ignores racks. The Network Topology Strategy serves "Multiple datacenters", "Supporting local queries": a replication factor per datacenter, and inside each one the walk skips nodes of a rack that already holds a copy (Fig. 1).

A Cassandra script starts by creating the keyspace, as in the exercise session (p. 1553):

-- three copies of every row
CREATE KEYSPACE population
  WITH replication = {
    'class': 'SimpleStrategy',
    'replication_factor': 3};

Per datacenter, in the syntax of the Cassandra documentation (not on the slides), for Fig. 1:

-- two copies in each datacenter
CREATE KEYSPACE population
  WITH replication = {
    'class':
      'NetworkTopologyStrategy',
    'DC1': 2, 'DC2': 2};

The hash picks the first node clockwise; the strategy picks the other N − 1: the next nodes, or the next racks of each datacenter.

Nobody in charge: gossip and failure detection

Node B3 reboots at night. With no master, nobody keeps the list of who is alive: each node keeps its own and corrects it by talking to the others. Page 440 draws a rumour — "I'm online!" — travelling from node to node as "gossip".

The "Gossip protocol" (p. 441) runs in rounds. "Each node picks its discussants (up to 3)", however large the cluster, and each conversation is a three-message handshake (Fig. 2). Afterwards both know the union of what they knew, so views "converge": the slide's anti-entropy, entropy being the disorder of nodes that disagree. Few partners per round mean "Always a constant amount of network traffic (except for gossip storms)", when many nodes change state at once.

The membership list

"Cluster Membership" (p. 451): each node keeps one row per node — Address, Heartbeat Counter, Time (local) — and raises only its own counter. "Nodes periodically gossip their membership list"; "On receipt, the local membership list is updated": per address the larger counter wins, stamped with the receiver's own clock, since the clocks are "asynchronous". Fig. 3 replays the slide.

From silence to suspicion

A silent node may be dead or only slow, and a fixed timeout errs one way or the other. The accrual failure detector of p. 452 "outputs a value (PHI) representing suspicion", "Apps set an appropriate threshold", and "PHI = 5 => 10-15 sec detection time". PHI, φ, grows with the silence, measured against the "Inter-arrival times for gossip messages":

1. Let \(\Delta t = t_{now} - t_{last}\) be the silence and F the distribution of past gaps. A live node stays silent this long with the tail probability

   \[ P_{\text{later}}(\Delta t) = 1 - F(\Delta t) \]

2. φ = 1 means a 10 % chance that a live node is this late; 2 means 1 %, 5 means 0.001 %:

   \[ \varphi = -\log_{10} P_{\text{later}}(\Delta t) \]

3. Background: Cassandra takes the gaps as exponential with mean μ, so \(P_{\text{later}} = e^{-\Delta t / \mu}\) and

   \[ \varphi = \frac{\Delta t}{\mu \ln 10} \]

4. With gossip about once a second (μ ≈ 1 s, background), φ = 5 after 5 ln 10 ≈ 11.5 s — inside the slide's 10–15 s. A slower network has a larger μ and a later alarm: the slide's "sensitive to actual inter-arrival time variations".

A write is an append; a read is a merge

A sale arrives for 'Milano'. "Writes" (p. 445): the client "sends write to one front-end node in Cassandra cluster (Coordinator)", "Which (via Partitioning function) sends it to all replica nodes responsible for key".

"Always writable: Hinted Handoff": "If any replica is down, the coordinator writes to all other replicas, and keeps the write until down replica comes back up" — the kept write is a hint — and "When all replicas are down, the Coordinator (front end) buffers writes (for up to an hour)". The slide adds "Atomicity for a given key (i.e., within ColumnFamily)": one write to one partition is applied whole, with no multi-key transactions. And "One ring per datacenter": the coordinator "can also send write to one replica per remote datacenter".

At the replica: commit log, memtable, SSTable

Writes "Need to be lock-free and fast (no reads or disk seeks)". At the replica (p. 446): "1. log it in disk commit log" — an append, so no seek, and the write survives a crash; "2. Make changes to appropriate memtables" — a memtable is an "In-memory representation of multiple key-value pairs", sorted by key. Then the replica acknowledges (Fig. 4).

"Later, when memtable is full or old, flush to disk": once, in key order, as an SSTable ("Sorted String Table – list of key value pairs, sorted by key"), with an index of "(key, position in data sstable) pairs" "And a Bloom filter". SSTables are never modified: newer versions of 'Milano' land in newer files, and compaction merges SSTables, keeping each key's newest version ("Merge key updates, etc.").

The Bloom filter: a cheap "no"

To skip the SSTables that cannot hold a key, each has a Bloom filter (p. 447), a "Compact way of representing a set of items": a bit array and k hash functions. "On insert, set all hashed bits." "On check-if-present, return true if all hashed bits set." One 0 proves absence; all 1s prove nothing — other keys may have set them. Hence "Some probability of false positives" but "Never false negatives".

The slide's example — "k=4 hash functions", "100 items", "3200 bits", "FP rate = 0.02%" — checks out with the standard formula (background): a bit stays 0 with probability about \(e^{-kn/m}\), so

32 bits per key, whatever the key: a "no" costs k bit tests in memory, not a disk read.

Deletes write a tombstone

Immutable files cannot erase a row. "Delete: don't delete item right away" — "add a tombstone to the log"; "Compaction will remove tombstone and delete item" (p. 448). A tombstone is a timestamped delete marker written like any write; it hides older values from reads (Fig. 5). Until compaction a delete adds data (background: a tombstone outlives a grace period, ten days by default, so that replicas that were down hear of it).

Reads merge

A read merges 'Milano' from the memtable and the SSTables whose filter says "maybe", keeping the newest timestamp: reads "May be slower than writes" (p. 446).

Page 455 adds the "Staged Event-Driven Architecture": "A general-purpose framework for high concurrency & load conditioning" built of "stages separated by queues". The slide leaves the link implicit (background): in a Cassandra node writes, reads, gossip and compaction are such stages, so overload shows as a growing queue, and a slow compaction does not stop gossip.

Back to the first puzzle

So the 0.12 ms against 15 ms of p. 459: a write is one log append and one memory insert, with no read, seek or lock; a read merges several files, often on several replicas. Cassandra gave up cheap reads and in-place updates, and pays later instead: flushes, compactions, tombstones. The slide names no hardware or workload: read a ratio, not a benchmark.

How many copies must answer: quorums

Three copies of 'Milano': how many must answer a write, and how many a read? The client chooses (p. 449): "R = read replica count, W = write replica count", "N = total number of replicas for this key".

Reads "Wait for R replicas (R specified by clients)" and check the "remaining N-R replicas" in the background. The coordinator "can contact closest replica (e.g., in same rack)", "also fetches from multiple replicas", and starts a read repair "if any two values are different" (p. 448). Writes "come in two flavors": "Block until quorum is reached", or "Async: Write to any node". A quorum is a majority of the N copies.

Why W + R > N is enough

The slide: "If W+R > N and W > N/2, you have consistency", and "Allowed (W=1, R=N) or (W=N, R=1) or (W=Q, R=Q)". The first half does the work:

1. The write was acknowledged by a set A of W replicas; a later read hears from a set B of R replicas, out of the same N:

   \[ |A \cup B| \le N \]

2. So the two sets share at least

   \[ |A \cap B| = W + R - |A \cup B| \] \[ |A \cap B| \ge W + R - N \]

3. If W + R > N, a replica answering the read holds the write, and the coordinator returns the most recent timestamp: "Return most recent TS" (p. 453).

4. The slide's corners pass: 1 + N, N + 1 and 2⌊N/2⌋ + 2 all exceed N.

The corners pay differently: (1, N) has fast writes, but a read needs every replica up; (N, 1) has fast reads, but one replica down stops the writes; (Q, Q) balances both and survives N − Q failures — one of three, two of five.

Consistency levels

"In reality, a client can choose one of these levels for a read/write operation" (p. 450; Table 2). The older tables of p. 453 use old names: ZERO ("Good luck with that") is gone, and "DCQ for RackAware" is today's LOCAL_QUORUM. In cqlsh the level is set for the statements that follow (documentation syntax, not on the slides):

-- next statements: wait for a majority
CONSISTENCY QUORUM

With three copies in each of two datacenters (computed): QUORUM waits for 4 of the 6, wherever they are; LOCAL_QUORUM for 2 in the coordinator's datacenter, with no trip to the other — what "Supporting local queries" buys; EACH_QUORUM for 2 in each, the strictest and slowest.

The trade, and when it pays

"Cassandra Summary" (p. 460): BASE, "Basically Available Soft-state Eventual Consistency"; Cassandra "Prefers Availability over Consistency", HBase "Prefers (strong) Consistency over Availability" — the CAP corners of lesson 5. Yet the choice is per operation: ONE and ANY keep accepting writes while replicas are down, QUORUM and ALL refuse rather than answer stale — "Eventual consistency, tunable per query" (p. 1550). Table 3 says when it pays.

A read is sure to see the latest acknowledged write when W + R > N; a quorum is ⌊N/2⌋ + 1, chosen per operation.

HBase: the same engine, one owner per row

"HBase" (p. 463) brings Google's Bigtable design into Hadoop: "Major Apache project today, part of Hadoop"; "Facebook uses HBase internally". Its API (p. 464): Get(row), Put(row), Scan(row range, filter) ("range queries") and MultiPut(rows). The scan is the first difference: rows stay sorted by row key, as with the order-preserving partitioner, so a range of rows is one sequential read. (Data model: lesson 32.)

The architecture of p. 465 (Fig. 6) has two floors. Upstairs, HBase: a client asks ZooKeeper which HRegionServer serves its rows, then talks to it directly; the HMaster assigns regions and watches the servers. Downstairs, Hadoop: every byte lives in HDFS DataNodes, which keep the copies.

The "HBase Storage hierarchy" (p. 466), outside in: a table splits into regions, ranges of row keys; a region has "One Store per ColumnFamily (subset of columns with similar query patterns)" — columns read together share a family, stored and read as one Store; a Store has a "Memstore" ("in-memory updates to Store; flushed to disk when full") and "StoreFiles … where the data lives", in blocks. A StoreFile is an HFile, the "SSTable from Google's BigTable": data blocks, indexes and a trailer, every key-value record carrying its whole key (p. 467, Table 4).

Write-ahead log and replay

"Strong Consistency: HBase Write-Ahead Log" (p. 468): a client's put(), delete() and incr() go "to HLog before writing to MemStore", so the server "Can recover from failure". The HLog is HBase's commit log, forced to disk by a Log Flusher (sync()) and cut into files by a Log Roller (rollWriter()). "Log Replay" (p. 469): "After recovery from failure, or upon bootup", "Replay any stale logs (use timestamps to find out where the database is w.r.t. the logs)" — "add edits to the MemStore". One HLog per HRegionServer, not per region, "Avoids many concurrent writes, which on the local file system may involve many disk seeks".

Across datacenters

"Cross-data center replication" (p. 470) reuses the log: each region server of the master cluster ships its HLog edits, in batches, to slave clusters ("Synchronous Call"). ZooKeeper, "actually a file system for control information", keeps /hbase/replication/state, /hbase/replication/peers/<peer cluster number> and /hbase/replication/rs/<hlog>: per region server, the HLogs to ship and the offset reached.

Back to the second puzzle

Same engine, different owner. Cassandra lets any of N equal replicas take a write, so writes never wait for one machine, and replicas may disagree for a while. HBase gives each row one server, so a read sees every acknowledged write of the row, with no quorum arithmetic; but while a crashed server's regions are reassigned and its HLog replayed, those rows cannot be served. Facebook chose that simpler consistency for Messages.

The other columnar: storing each column apart

The deck closes with "Other Columnar DBs" (p. 471) — another meaning of "columnar". Cassandra and HBase keep a row's columns together (HBase within each column family); MonetDB and LucidDB store each column apart, the column storage of lesson 20, for queries that read a few columns of many rows.

"MonetDB" (p. 472) runs an SQL front-end (relational model), an ODMG front-end (object-oriented model), an OLAP tool and a data-mining tool on one "Monet back-end" through the MIL Language, with "full vertical fragmentation": on p. 473 (Fig. 7) Order(id, day, discount) becomes order_id, order_day and order_discount, each pairing an object id, the oid, with one value.

"LucidDB" (pp. 474–476) keeps "column store tables" in one file, db.dat. A table "consists of set of clusters", by default one per column ("Each column maps to single cluster"), and a cluster page "stores the values for a specific set of rowIDs". A B-tree per cluster "maps rid values to pageIds" by each page's first rid: page 49 starts at rid 984 and page 50 at 1248, so page 49 holds rids 984–1247. Inside a page (p. 477) LucidDB stores "just the unique column values", each with "a bit-encoded vector" (Table 6).

Hence the "Conclusion" (p. 478): "Column architecture doesn't read unnecessary columns"; "Avoids decompression costs and performs operations faster" — discount = 0.000 is answered by the vector 0 0 0 1 1 0, decompressing nothing; "Use of compression schemes allow us to lower our disk space requirements".