NoSQL ยท 01

NoSQL Fundamentals

Before Redis, DynamoDB, MongoDB or Elasticsearch, there is a shared set of ideas: why these databases exist, what the CAP theorem really constrains, how replicas stay in step, how data is cut across machines, and why you model from queries. Learn these once and every product becomes a set of settings. The page ends with a plain-Java quorum simulation that shows stale and consistent reads with real output.

13
Chapters
5
Diagrams
6
Consistency models
1
Plain-Java simulation
Map ยท Fundamentals ยท Redis ยท DynamoDB ยท MongoDB ยท Elasticsearch ยท Caching ยท Polyglot
01
Chapter One - Why

Why NoSQL Appeared, and What It Costs

"NoSQL" is not one technology. It is a label stuck on a group of databases that, since the late 2000s, made different choices from the relational model. The honest way to learn them is to ask what pressure produced each choice and what was given up in return. Every feature in this section is a trade, not a free improvement.

Three pressures
PressureWhat happenedThe design response
Scale-outOne server could no longer hold the data or take the writes. Buying a bigger machine (scaling up) gets expensive and hits a ceilingSpread data over many cheap machines: partitioning and replication built into the database, not bolted on
Latency at volumeWeb applications needed single-digit-millisecond reads for a known key, millions of times per secondSimple access paths (look up by key), data laid out so one request touches one machine, no joins across machines
Schema flexibilityProduct catalogues, user profiles and event payloads vary per record and change weekly. Altering a large table was slow and riskyStore self-describing records (JSON documents, wide rows) and let the application interpret them
AvailabilityA shop that cannot take an order loses money; a slightly stale product page does notKeep accepting reads and writes during failures and reconcile later (eventual consistency)

The best-known origin stories are Google Bigtable and Amazon Dynamo, both described in papers (Dynamo in 2007), which inspired open-source systems such as HBase, Cassandra and Riak. Document stores such as MongoDB and key-value stores such as Redis and DynamoDB (the managed AWS service, which borrows ideas from the Dynamo paper but is a different system) came from the same era. Today most families also offer features once considered "relational only", so the line has blurred.

What is traded away
You give up (to different degrees)Why it is hard to keep at scaleWhat you do instead
Ad-hoc queries and joinsA join across partitions means moving data between machines on every queryDesign the data for known queries; copy (denormalise) data so one read answers one question
Multi-row ACID transactionsCoordinating a commit across machines costs latency and availabilityAtomic single items, conditional writes, sagas and idempotent retries (chapter 9)
Strong consistency by defaultWaiting for every replica makes every write as slow as the slowest machineTunable consistency per request; design the application to tolerate stale reads
Referential integrityA foreign-key check is a cross-partition read before every writeApplication-level checks, background reconciliation
A mature, shared query languageEach product has its own API and query modelLearn each model; accept less portability
โš  "NoSQL is faster"
  • Not as a rule. A relational database with the right index answers a point lookup in microseconds to a few milliseconds, and PostgreSQL or MySQL on one good server handles far more than most applications will ever need. NoSQL databases are faster for the workload they were shaped for (huge key-value traffic, global scale, flexible documents) and slower or impossible for others. Choose them for a reason you can name, not for a feeling that relational is old.
Mental model: A relational database is a well-organised filing room with an expert clerk: ask anything, the clerk cross-references folders for you. A NoSQL store is a set of labelled lockers in many buildings: you must know which locker to open, and in return you can serve a million people at once. Both are right, for different jobs.
02
Chapter Two - CAP and PACELC

What the CAP Theorem Really Says

CAP is the most quoted and most misquoted idea in this field. Stated precisely (Gilbert and Lynch proved the formal version in 2002, after Eric Brewer's conjecture): in a distributed system that stores data on several machines, when a network partition occurs, you must choose between consistency and availability.

The three letters, defined carefully
LetterPrecise meaning in CAPCommon misunderstanding
C - ConsistencyLinearizability: every read returns the most recent completed write, as if there were one copy of the dataIt is not the "C" of ACID, which means "the database obeys its constraints"
A - AvailabilityEvery request that reaches a working node receives a non-error response, with no promise that it is the latest dataIt is not "99.99% uptime"; it is about whether a live node answers during a partition
P - Partition toleranceThe system keeps operating when messages between nodes are lost or delayedIt is not optional: networks do fail, so every distributed system must tolerate partitions
A network partition: the replicas cannot talk, and you must choose
Data centre A Data centre B Replica A x = 2 Client 1: write x = 2 Replica B x = 1 Client 2: read x X link is down A cannot tell B about x = 2 CP: Replica B refuses the read (error or timeout until the link heals) Consistent, but not available AP: Replica B answers x = 1 (stale, but it answers) Available, but not consistent You cannot have both answers at once. The partition forces the choice.

Because partitions are a fact of life rather than a feature you can decline, the "pick two of three" slogan misleads. The real statement is: partitions will happen; during one, choose C or A. When the network is healthy you can have both. A single-node database is not distributed, so CAP does not apply to it at all.

PACELC: the part that matters every day

Partitions are rare, but latency is constant. Daniel Abadi's PACELC extension says: if there is a Partition, choose Availability or Consistency; Else, when running normally, choose Latency or Consistency. Even with a healthy network, keeping replicas in step costs round trips. A system that waits for every replica before acknowledging a write is consistent and slow; one that acknowledges after one replica is fast and may serve stale reads.

StyleDuring a partition (PAC)Normally (ELC)Typical behaviour
PC/ECConsistencyConsistencyAlways consistent, pays latency: single-leader systems with synchronous replication, consensus-based stores
PA/ELAvailabilityLatencyFast and always answering, may be stale: leaderless systems configured for low consistency levels
PC/ELConsistencyLatencyFast normally, refuses during a partition: a leader with asynchronous replicas that fences off the minority side
PA/ECAvailabilityConsistencyRare: consistent when healthy, falls back to answering when partitioned
โš  Labelling a product "CP" or "AP" once and for all
  • Most modern databases are tunable: Cassandra and DynamoDB let each request choose how many replicas to involve, MongoDB lets you set read and write concerns. The same product behaves like CP or AP depending on configuration, and CAP also ignores latency, which is usually what users feel. Treat the labels as shorthand for a default, then read what the specific setting guarantees.
How to answer in an interview

"CAP says that during a network partition a replicated system must give up either linearizable consistency or availability. Partition tolerance is not a choice. In normal operation the real trade is latency versus consistency, which PACELC captures."

03
Chapter Three - ACID vs BASE

Two Philosophies of Correctness

Relational databases popularised ACID. Many NoSQL systems describe themselves with the playful opposite, BASE. These are design philosophies, not exact specifications, and the contrast is easy to overstate.

ACIDMeaningBASEMeaning
AtomicityAll the writes of a transaction apply, or noneBasically AvailableThe system keeps answering, perhaps with degraded or partial data, even during failures
ConsistencyA transaction moves the database from one valid state to another (constraints hold)Soft stateState may change over time without new input, as replicas converge
IsolationConcurrent transactions behave as if they ran one after another (to a chosen isolation level)Eventually consistentIf writes stop, all replicas converge to the same value, with no bound on how long
DurabilityOnce committed, a write survives crashes

Two cautions. First, BASE does not mean "no durability": a write that was acknowledged is normally still safely stored (on at least one node, often more). What is relaxed is agreement between replicas, not safety of the write. Second, "ACID" is not binary either: isolation has levels (read committed, snapshot, serializable), and several NoSQL databases do offer ACID transactions within a single item, a single partition or, since recent versions, across documents (chapter 9).

QuestionTypical ACID relational answerTypical BASE-style answer
A write is acknowledged. Can another client read the old value?No, on a single node (and for replicas that are synchronous)Yes, for a short time, on some replicas
Two clients update the same item at onceLocks or snapshot isolation serialise themLast write wins, or version conflict detection, or a merge function
A node is unreachableWrites may block or fail until failoverOther nodes keep serving; the node catches up later
Where does an integrity rule live?In the schema (foreign key, unique, check)In the application, plus a unique key where the store supports it
Bank vs social feed: A bank transfer must never show money in both accounts or neither: ACID. A "like" counter that shows 1,203 on one phone and 1,205 on another for a few seconds harms no one: BASE is fine and far cheaper. Decide per piece of data which kind it is; one system often holds both.
04
Chapter Four - Consistency Models

What a Read Is Allowed to Return

"Consistency" is a spectrum of promises about which value a read may return after writes. The diagram shows the situation that makes the spectrum necessary: replicas receive the same write at different moments, and a read can land on any of them.

After a write, three replicas catch up at different times
Writer Replica 1 (leader) Replica 2 Replica 3 put x = 2 acknowledged after the leader has it x = 2 x = 2 x = 2 replication lag differs per replica read: x = 2 read: x = 1 (stale) read: x = 2 A reader that saw x = 2 from Replica 2 and then x = 1 from Replica 3 went backwards in time: a monotonic-reads violation. The data is not lost, only unevenly spread for a moment.
ModelPromiseWhat you can still seeTypical cost
Strong / linearizableEvery read returns the latest completed write; all clients see one order of operationsNothing odd: behaves like one copyCoordination on every operation: latency, lower availability under partition
SequentialAll clients see operations in one agreed order, but not necessarily the real-time latestA reader may lag behind real timeSlightly cheaper than linearizable
CausalIf B was caused by (or depends on) A, everyone sees A before BUnrelated writes may appear in different orders on different replicasNeeds tracking of dependencies
Read-your-writesAfter you write, your reads see that writeOther users may still see old dataRoute the session to a fresh replica, or carry a version token
Monotonic readsOnce you have seen a value, you never see an older oneYou may be behind other users, but never go backwardsPin a session to one replica, or track versions seen
EventualIf writes stop, replicas convergeStale reads, reordering, disappearing then reappearing valuesCheapest and fastest

Linearizable and eventual are the two ends; the middle models are the useful ones for applications. Read-your-writes and monotonic reads are called session guarantees and fix the cases users actually notice: posting a comment and not seeing it, or refreshing a page and watching it revert.

Concrete knobs in real products
ProductHow you choose
DynamoDBReads are eventually consistent by default; set ConsistentRead to true for a strongly consistent read of the table (costs more read capacity). Global secondary indexes only support eventually consistent reads (AWS Developer Guide, read consistency page)
CassandraPer-request consistency level (ONE, QUORUM, ALL, ...), chosen separately for reads and writes
MongoDBRead concern and write concern per operation, plus the read preference (primary or secondaries)
RedisReplication to replicas is asynchronous by default; reading from a replica can be stale. WAIT can block until a number of replicas acknowledge but does not make it linearizable
โš  Reading from a replica to "scale reads" without telling the user story
  • A user saves a profile, the page reloads, a read hits a replica that has not caught up, and the old name appears. Support tickets read "my change did not save". Either read your own writes from the leader for a short window after a write, or send a version/timestamp token and wait for a replica that has reached it.
05
Chapter Five - Replication

Keeping Copies: Leaders, Followers and Quorums

Replication keeps several copies of each piece of data for durability and availability, and lets reads spread across machines. How the copies are kept in step decides most of a database's behaviour. There are three main shapes.

TopologyHow a write flowsStrengthsWeaknessesExamples
Leader-follower (single leader)All writes go to one leader, which streams changes to followersSimple; writes are ordered; easy to reason about conflicts (there are none)The leader is a write bottleneck; failover takes time and can lose recent writes if replication was asynchronousPostgreSQL and MySQL replication, MongoDB replica sets, Redis replication
Multi-leaderSeveral leaders (often one per region) accept writes and replicate to each otherWrites stay local to a region; survives a region lossConcurrent writes to the same item in different regions conflict and must be resolvedMulti-region setups such as DynamoDB global tables (last writer wins)
LeaderlessThe client (or a coordinator node) sends each write to several replicas and reads from severalNo failover step; tolerates node failures smoothly; tunable consistencyConflicts and stale reads handled by the application/database logic; needs repair processesCassandra, Riak, ScyllaDB (all in the Dynamo tradition)
Quorums: R + W > N

In a leaderless system with N replicas of each item, a write is considered successful once W replicas acknowledge it, and a read asks R replicas and returns the freshest answer. If **R + W > N**, the set of replicas read must overlap the set written in at least one node, so the read sees the latest acknowledged write. The overlap guarantee is the whole idea.

Quorum replication: N = 3, W = 2, R = 2
Writer Replica 1 v2 Replica 2 v2 Replica 3 v1 (missed it) write v2: acked write v2: acked lost or slow Reader v2 v1 asks 2 of 3, keeps the highest version: v2 Any 2 replicas read overlap any 2 replicas written in at least one, so the reader always meets a node holding v2.
Worked example (an example, not a benchmark)
NWRR + WOverlap guaranteed?Behaviour
3224Yes (4 greater than 3)The common "quorum" setting. Survives one node down for reads and writes
3314YesFast reads, but one slow or dead node blocks every write
3134YesFast writes, but one dead node blocks every read
3112NoFastest and the most available; stale reads are routine
5336YesSurvives two nodes down; costs more storage and traffic

The quorum overlap is a statement about sets of nodes, and the hands-on in chapter 11 enumerates it exhaustively. In practice R + W > N is a necessary ingredient, not a full guarantee of linearizability. Concurrent writes, sloppy quorums, clock-based conflict resolution and partial write failures can still produce surprises, which is why Cassandra documentation, for example, talks about "strong consistency" with quorum reads and writes under stated conditions rather than as an unconditional guarantee.

The repair mechanisms (qualitatively)
MechanismWhat it doesWhen it happens
Hinted handoffIf a replica is unreachable during a write, another node stores a "hint" (the missed write) and delivers it when the replica returnsWrite path; helps short outages. Hints are kept for a limited time, so a long outage needs a full repair
Read repairA read that contacts several replicas notices a stale one and writes the freshest value back to itRead path; fixes data that is actually read
Anti-entropy repairA background process compares replicas (typically using hash trees) and syncs differencesScheduled maintenance; fixes data nobody reads

Cassandra's architecture documentation describes hinted handoff in the write path and read repair in the read path, following the Dynamo paper. The exact timings, time-to-live of hints and tuning flags are product- and version-specific, so check the documentation for the version you run.

โš  Assuming a quorum makes the database a serial machine
  • With R + W > N you will not see an older value than the last acknowledged write, but a write that failed halfway (acknowledged by fewer than W replicas) can still be visible to some reads and later disappear or be propagated. The client received an error, yet the value may "come back". Design writes to be safe to retry.
06
Chapter Six - Partitioning

Splitting Data Across Machines

Replication copies the same data to several machines. Partitioning (also called sharding) splits different data across machines so no single machine holds or serves everything. Every scale-out database does both: data is cut into partitions, and each partition is replicated.

Hash or range
Range vs hash partitioning for a time-ordered key
Range partitioning by timestamp Node 1 Jan - Mar Node 2 Apr - Jun Node 3 Jul - now ALL new writes Range scans ("all of April") touch one node, but the newest range is a hot spot. Hash partitioning on the key Node 1 hash slice 1 Node 2 hash slice 2 Node 3 hash slice 3 writes spread evenly across all nodes A range scan ("all of April") must ask every node, because neighbouring keys are scattered. Compromise: hash a prefix, sort within it (partition key + sort key), as DynamoDB and Cassandra do.
StrategyHow a key finds its partitionGood atBad at
RangePartitions own contiguous key ranges (A to H, I to Q, ...)Range scans and ordered reads; easy to split a busy rangeHot spots when keys are sequential (timestamps, auto-increment ids)
HashA hash of the key picks the partitionEven load; no hot spot from key shapeRange scans must visit all partitions; ordering across keys is lost
Hash + sort keyHash the partition key, keep rows sorted by a sort key inside the partitionOne partition holds everything for one entity, sorted: "all orders of customer 42 by date"A single huge partition key (one big customer) is still a hot spot
Hot spots

Even with a hash, a hot key can overload one partition: a celebrity profile, a viral product, a counter that every request increments, a "current date" key. The database cannot spread a single key over many machines, so the remedy is in the data model: add a random or computed suffix to spread the key (product-7#0 to product-7#9) and sum or merge on read, cache the hot item, or batch counter updates.

Consistent hashing

The naive scheme, hash(key) mod N, reassigns almost every key when N changes. Consistent hashing places both nodes and keys on a ring; a key belongs to the first node clockwise from it. When a node joins or leaves, only the keys in its arc move.

Consistent hashing: a key belongs to the next node clockwise
hash ring 0 to 2^32 A B C k1 k2 k3 D Rule: hash the node name and the key onto the same ring. A key is stored on the first node clockwise from it. Before D joins: k1 goes to B, k2 goes to C, k3 goes to A. Add node D between B and C: only k2 moves (from C to D). k1 and k3 stay put. About 1/N of the keys move, not nearly all of them. Why not hash(key) mod N? Going from N to N+1 changes the answer for most keys, so nearly all data would have to move. Virtual nodes: each machine owns many small points on the ring, not one. Load evens out and a new machine takes a few slices from every peer.
TermMeaningWhy it matters
Token / ring positionA node's point (or points) on the ringDecides which keys it owns
Virtual nodes (vnodes)Many small ring positions per physical machineSpreads load evenly, lets bigger machines own more points, and makes a rebuild draw from many peers in parallel. Cassandra's documentation describes vnodes following the Dynamo paper
Preference list / replica setThe next few distinct nodes clockwise, which also store copiesReplication placement falls out of the same ring
RebalancingMoving partitions to a new node, or off a leaving oneMust run while serving traffic, throttled so it does not starve normal requests

Not every database uses a literal ring. Fixed-count partition schemes (a database that creates, say, a few hundred logical partitions up front and assigns them to nodes, as Redis Cluster does with 16384 hash slots) give the same property: adding a node moves whole partitions, not individual keys. DynamoDB hides partitioning entirely and splits partitions for you. The concept, not the mechanism, is what carries across products.

โš  Picking a partition key by what is convenient
  • A key such as status (three values) or date (one "today") puts most traffic on a few partitions, whatever hashing you use. A good partition key has high cardinality (many distinct values), is known on every request, and spreads both data size and request rate evenly.
07
Chapter Seven - Secondary Indexes

Querying by Something Other Than the Key

A partitioned store finds data fast by its partition key. Asking "all orders with status SHIPPED" is a different question: the answer is scattered across every partition. A secondary index makes such a query possible, and how it is partitioned decides its cost.

Local secondary indexGlobal secondary index
Where the index entries liveIn the same partition as the data they point to (each partition indexes only its own rows)Partitioned separately by the indexed value, so one lookup finds all matches on one index partition
Write costCheap: one partition updates both the item and its indexDearer: the index entry may live on a different machine than the item; usually updated asynchronously
Read by indexed valueScatter-gather: ask every partition and merge (slow and the tail latency of the slowest partition)One partition: fast
ConsistencyCan be consistent with the itemOften eventually consistent with the item (a just-written item may be missing from the index for a moment)
Typical productsCassandra secondary indexes (local to each node); the MongoDB index on a shard (each shard has its own); Elasticsearch shards each index their own documentsDynamoDB global secondary indexes; Elasticsearch and OpenSearch index per shard but route queries to all shards

DynamoDB offers both under these names: local secondary indexes share the partition key of the table (and need the table's partition key plus a different sort key), while global secondary indexes can have an entirely different key and have their own provisioned or on-demand capacity. Names differ elsewhere, but the trade is the same.

The cost of an index
  • Every write is amplified: an item with three indexes is written four times (once to the table, once per index).
  • Storage grows with each index, and a global index copies the projected attributes.
  • An index can lag: a global index that updates asynchronously can show a stale view, so never use it to check a uniqueness rule.
  • Low-cardinality indexes help little: indexing a boolean returns half the table.
Often the better answer is a second copy

Instead of an index on customerId for orders keyed by orderId, many NoSQL designs keep a second table or item collection keyed by customer and write both. It is the denormalisation idea of the next chapter applied to access paths: one table per question.

08
Chapter Eight - Data Modelling

Model From the Queries, Not From the Entities

This is the biggest mental shift from relational design. In a relational database you model the data (entities, normal forms) and write whatever queries you later need. In most NoSQL databases you list the queries first and shape the data so each query reads one place.

Relational habitNoSQL habit
Start fromEntities and relationshipsA written list of access patterns ("get order with items by order id", "list a customer's orders, newest first")
DuplicationAvoided: normalise, one fact in one placeAccepted: the same fact is copied into every shape that needs it
JoinsAt query timeAt write time (precompute) or in the application
Schema changeMigration of the tableNew code writes new shape; old items migrated lazily or in the background
Cost modelQuery planner decides; you tune indexesYou decide the access path; a missing path means a full scan or no way at all
Embedding or referencing
ChoiceUse whenBenefitRisk
Embed the child inside the parent (an order document contains its line items)The child is read with the parent, belongs to one parent, and the set is boundedOne read, atomic update of the whole itemUnbounded growth (a document or item has a size limit); updating a part still rewrites the whole item in many stores
Reference by id (an order holds product ids)The child is shared by many parents, changes independently, or is large/unboundedNo duplication; independent updatesNeeds a second read (an application-side join) and the referenced item can disappear
Embed a snapshot of a referenced item (the product name and price at the time of the order)The historical value is itself meaningfulOrder history stays correct when the product changesDeliberate duplication: the snapshot must not be "fixed" later
A denormalised order: one read answers "show my order" (illustrative JSON)
{ "orderId": "o-1001", "customerId": "c-42", "placedAt": "2026-10-31T09:15:00Z", "status": "SHIPPED", "shipTo": { "city": "Pune", "pin": "411001" }, "items": [ { "sku": "p-7", "name": "Keyboard", "unitPrice": 2499, "qty": 1 }, { "sku": "p-9", "name": "Cable", "unitPrice": 199, "qty": 2 } ], "version": 4 }
One-to-many and many-to-many
RelationshipRelationalNoSQL options
One to few (an order and its items)Child table + joinEmbed the children
One to many (a customer and thousands of orders)Child table + foreign keyReference: a separate item per order, sharing the partition key customerId and sorted by date, so one range query lists them
Many to many (students and courses)Junction tableTwo lists (each side stores ids of the other) updated together, or an adjacency list: one item per link, indexed from both sides
Hierarchies (categories, org charts)Self-join or recursive queryStore the full path or the parent id; or use a graph database when traversal is the main query
Schema-on-read and schema evolution

A schemaless store does not remove the schema, it moves it into the application code. Every reader must cope with every shape that ever existed. Practise it deliberately:

  • Put a **schemaVersion** field on every item, and let readers upgrade old shapes on read (and optionally write them back).
  • Make new fields optional with defaults; never rely on a field being present in old items.
  • Prefer additive changes; for renames and splits, write both shapes for a while, then backfill, then drop the old one.
  • Where the database supports it, add validation (MongoDB supports JSON Schema validation on collections) to catch typos, because a store that accepts anything also accepts custmerId.
โš  "Schemaless means we do not need to design"
  • It means the design is not enforced for you. Teams that skip it end up with one collection holding five incompatible shapes, queries that need if chains in every reader, and a migration nobody dares to run. The data model is the most expensive thing to change later in a NoSQL system, because the access patterns are baked into the layout.
09
Chapter Nine - Transactions, Time and Ordering

What Is Lost, What You Get Back, and What Clocks Cannot Do

Transactions and constraints
CapabilityTypical NoSQL positionHow to cope
Atomic update of one itemAlmost always provided (one document, one row, one key)Put data that must change together inside one item
Conditional write (compare-and-set)Widely available: DynamoDB condition expressions, MongoDB update filters, Redis WATCH/Lua, Cassandra lightweight transactionsThe building block for optimistic concurrency and uniqueness
Multi-item transactionVaries. DynamoDB has TransactWriteItems; MongoDB supports multi-document transactions on replica sets and sharded clusters (since 4.0 and 4.2); Cassandra has only lightweight, single-partition transactions; many key-value stores have noneCheck the product documentation for limits and cost; keep transactions small and rare
Foreign keysNot enforcedApplication checks, cleanup jobs, accept dangling references
Unique constraintsOnly on the key (or a unique index, where offered)Use the unique value as the key, or a conditional "insert if absent"
Isolation levelsRarely configurableDesign so concurrent writers cannot conflict, or detect conflicts with versions
Optimistic concurrency with a version

Two users edit the same item. Without protection, the second write silently overwrites the first (a lost update). The standard fix needs no locks: every item carries a version, and a write succeeds only if the version is still the one the writer read.

Compare-and-set in pseudocode (the same idea as a DynamoDB condition expression or a MongoDB filter on version)
read item -> { "name": "Keyboard", "price": 2499, "version": 4 } compute new value -> price = 2299 write if version == 4, set version = 5 success -> done condition failed -> someone else wrote first: re-read, re-apply the change, retry

In a Spring application with MongoDB or JPA this appears as an @Version field and an optimistic-locking exception to catch and retry; in DynamoDB it is a ConditionExpression on the version attribute. For the same idea in relational databases see the SQL section; for the transaction boundaries on the Spring side see Spring: Transactions.

Time and ordering

A distributed database needs to decide which of two writes to the same item is "newer". Wall clocks are the obvious tool and the dangerous one: clocks on different machines drift and jump, even with synchronisation, so "later" by clock is not "later" in reality.

ApproachHow it decidesHazard
Last write wins (LWW)Highest timestamp survives, the other write is discardedA machine with a fast clock wins every time; concurrent updates silently vanish; fine for idempotent whole-value overwrites, wrong for counters and merges
Version number / CASA write must name the version it is based onNeeds a single authority per item (leader); the loser must retry
Version vectors (vector clocks)Each replica keeps a counter; comparing the vectors shows whether one write happened-before the other or they were concurrentConcurrent versions are both kept ("siblings") and the application or a merge function must reconcile them. Riak exposed this directly; the Dynamo paper describes it
CRDTs (conflict-free replicated data types)Data types (counters, sets) designed so concurrent updates merge deterministicallyLimited to types that have a merge; Redis Enterprise and Riak offer some
Single leader per itemOne node orders all writesGives order, costs cross-region latency
โš  Last-write-wins on data that cannot be overwritten
  • Two services add an item to the same shopping cart at nearly the same time; each reads the cart, appends, and writes the whole list back. LWW keeps one write and the other item vanishes. Symptom: "my item disappeared". Fix: model the cart as separate items (one per product), use a set-add operation, or write with a version condition and retry.
10
Chapter Ten - Capacity, Anti-cases and Migration

Sizing, When Not to Use NoSQL, and Moving Over Safely

Capacity and cost thinking

Scale-out databases make cost visible in terms of what you ask, not just what you store. Do the arithmetic before choosing.

QuestionWhy it decides the design
How many reads per second and writes per second at peak?Sets the number of partitions or provisioned capacity; managed stores bill by request or by provisioned throughput
How large is an item, and how large will the data be in a year?Item size limits exist (check the product), and larger items cost more per read and write
Read:write ratio and read-after-write needs?Decides replicas, caching, and whether consistent reads are affordable
Is traffic even or spiky? Any hot keys?Hot keys cap throughput regardless of cluster size; spiky load favours on-demand or autoscaling
How many access patterns, and which need an index or a copy?Each index or duplicate multiplies write cost and storage
What latency and consistency does each pattern truly need?Strong reads cost more and are slower than eventual ones

Worked example (illustrative numbers): 5,000 writes per second of 2 KB items is about 10 MB per second of ingest, roughly 860 GB per day before replication; with N = 3 copies that is about 2.6 TB per day of raw storage and three times the network traffic inside the cluster. If each item also has two global secondary indexes, writes become three physical writes before replication. Numbers like these quickly separate "a single PostgreSQL" from "a partitioned store" and show how fast storage grows. Always compute the volume before arguing about brands.

When NOT to use NoSQL
SituationWhy a relational database is usually better
You do not know your queries yet (early product, exploratory reporting)Ad-hoc SQL and joins let you explore; a NoSQL layout must be rebuilt when questions change
Strong multi-row integrity is the core (ledgers, bookings, inventory with reservations)ACID transactions, constraints and serializable isolation are exactly what is needed
Data is highly relational (many joins across many entities)Denormalising everything multiplies writes and invites inconsistency
The whole dataset fits on one machine and traffic is moderateOne good PostgreSQL or MySQL server with replicas is simpler, cheaper and well understood
Heavy analytics and reportingColumnar warehouses and SQL engines are built for scans and aggregations; key-value stores are not
The team has no experience with the chosen storeOperational mistakes (hot keys, bad partition key, wrong consistency) are expensive and slow to fix
"Because it is web scale"Almost no application is. Choose for a measured need, and note modern relational databases also scale reads with replicas and writes with sharding when truly needed
A sensible default

Start with a relational database as the system of record. Add a NoSQL store for a specific need (a cache, a search index, a high-volume key-value path, a document store for variable data) and keep the reasoning in a short design note. See the choosing guide and, later in this section, polyglot persistence.

Migrating from relational
ApproachHowCaution
StranglerMove one bounded slice (one table or one query path) to the new store, route that traffic, repeatPick a slice with few joins; keep the old path working until the new one is proven
Dual writeThe application writes to both old and new storesDangerous as a plain approach: two writes with no shared transaction drift apart on any failure (the dual-write problem)
Change data capture + outboxCapture changes from the relational database log or an outbox table and apply them to the new store asynchronouslyNeeds idempotent consumers and a plan for lag; see Event-Driven Patterns
Backfill then cut overBulk-load history, replay changes, compare, switch reads, then writesShadow-read and compare results before trusting the new store; keep a rollback
โš  Dual-writing from application code and trusting it
  • A crash between the two writes, a timeout that was actually a success, or a retry that duplicates, leaves the stores different, and nothing tells you. If you must dual-write, make the second write asynchronous and idempotent via an outbox, and run a comparison job that reports mismatches.
11
Chapter Eleven - Hands-On

A Thought Experiment: Simulating Quorums in Plain Java

Reading R + W > N is easy to nod at and hard to feel. This small program models N = 3 replicas of one item. Each replica holds a versioned value. A writer updates W replicas to version 2 (the others have not received it yet), then a reader asks R replicas and returns the highest version it sees. The program tries every possible choice of which replicas were written and which were read, and counts the reads that returned the stale value.

QuorumSim.java (plain Java 17+, no libraries; compiled with JDK 21)
import java.util.*; public class QuorumSim { record Versioned(int version, String value) {} static final int N = 3; static final Versioned[] replicas = new Versioned[N]; static void reset() { Arrays.fill(replicas, new Versioned(1, "v1")); } // The write is acknowledged once W replicas have it. The others have not // received it yet (replication is asynchronous or the messages are delayed). static void write(Set<Integer> reachedReplicas, Versioned v) { for (int i : reachedReplicas) replicas[i] = v; } // The read asks R replicas and returns the answer with the highest version. static Versioned read(Set<Integer> askedReplicas) { Versioned best = null; for (int i : askedReplicas) { if (best == null || replicas[i].version() > best.version()) best = replicas[i]; } return best; } static List<Set<Integer>> subsets(int size) { List<Set<Integer>> out = new ArrayList<>(); for (int mask = 0; mask < (1 << N); mask++) { if (Integer.bitCount(mask) != size) continue; Set<Integer> s = new TreeSet<>(); for (int i = 0; i < N; i++) if ((mask & (1 << i)) != 0) s.add(i); out.add(s); } return out; } public static void main(String[] args) { System.out.println("N=" + N + ". Every replica starts with v1. A client writes v2 to W replicas,"); System.out.println("then reads from R replicas. We try EVERY possible choice of replicas.\n"); System.out.printf("%-3s %-3s %-5s %-12s %-22s %s%n", "W", "R", "R+W", "R+W > N ?", "stale reads / combos", "verdict"); for (int w = 1; w <= N; w++) { for (int r = 1; r <= N; r++) { int stale = 0, total = 0; for (Set<Integer> writeSet : subsets(w)) { for (Set<Integer> readSet : subsets(r)) { reset(); write(writeSet, new Versioned(2, "v2")); total++; if (read(readSet).version() != 2) stale++; } } boolean overlap = r + w > N; System.out.printf("%-3d %-3d %-5d %-12s %-22s %s%n", w, r, r + w, overlap, stale + " / " + total, stale == 0 ? "always sees v2" : "can return v1 (stale)"); } } System.out.println("\nOne concrete trace, W=2 R=2 (R+W=4 > 3):"); reset(); write(Set.of(0, 1), new Versioned(2, "v2")); System.out.println(" replicas after write : " + Arrays.toString(replicas)); System.out.println(" read from {1,2} : " + read(new TreeSet<>(List.of(1, 2))) + " <- replica 1 is in both sets"); System.out.println("\nOne concrete trace, W=1 R=1 (R+W=2 <= 3):"); reset(); write(Set.of(0), new Versioned(2, "v2")); System.out.println(" replicas after write : " + Arrays.toString(replicas)); System.out.println(" read from {2} : " + read(Set.of(2)) + " <- replica 2 never saw the write"); } }

Compile and run it with javac QuorumSim.java then java QuorumSim. The output below is the real output of that run:

Output (real run)
N=3. Every replica starts with v1. A client writes v2 to W replicas, then reads from R replicas. We try EVERY possible choice of replicas. W R R+W R+W > N ? stale reads / combos verdict 1 1 2 false 6 / 9 can return v1 (stale) 1 2 3 false 3 / 9 can return v1 (stale) 1 3 4 true 0 / 3 always sees v2 2 1 3 false 3 / 9 can return v1 (stale) 2 2 4 true 0 / 9 always sees v2 2 3 5 true 0 / 3 always sees v2 3 1 4 true 0 / 3 always sees v2 3 2 5 true 0 / 3 always sees v2 3 3 6 true 0 / 1 always sees v2 One concrete trace, W=2 R=2 (R+W=4 > 3): replicas after write : [Versioned[version=2, value=v2], Versioned[version=2, value=v2], Versioned[version=1, value=v1]] read from {1,2} : Versioned[version=2, value=v2] <- replica 1 is in both sets One concrete trace, W=1 R=1 (R+W=2 <= 3): replicas after write : [Versioned[version=2, value=v2], Versioned[version=1, value=v1], Versioned[version=1, value=v1]] read from {2} : Versioned[version=1, value=v1] <- replica 2 never saw the write
Reading the result
RowWhat it shows
W=1 R=1 (R+W=2)6 of 9 combinations return stale data. A write that reached one replica is invisible to a read that asks a different one. This is eventual consistency in its plainest form
W=2 R=1 and W=1 R=2 (R+W=3, equal to N)Still stale in some combinations: R + W = N is not enough; the sets can miss each other by exactly one node
W=2 R=2 (R+W=4)Zero stale reads in all 9 combinations: any two replicas must share at least one. This is the usual "quorum" setting
W=3 R=1 and W=1 R=3Also consistent, but each shifts the cost: reads are cheap and writes need every replica, or the reverse

Notice what the program does not model, because the gaps are the lesson: it has no failures during a write (a write that reached only some replicas but was never acknowledged), no concurrent writers, no clocks, and no repair. Each of those is a place where a real system can still surprise you, and where hinted handoff, read repair and version vectors earn their keep.

Try it yourself
  • Change N to 5 and the loop bounds, and find the smallest W and R that give zero stale reads. Then add a second writer thread that writes version 3 to a different replica set and see which version a quorum read returns without a conflict rule.
  • Extend read so that it also repairs the stale replicas it noticed (read repair), then run a second read and observe that the stale count drops.
12
Chapter Twelve - Interview

Questions You Should Be Able to Answer

Crisp answers for the questions that come up in interviews on this topic.

  • Q: State the CAP theorem precisely.
    In a replicated system, when a network partition occurs, you must choose between consistency (linearizability: every read sees the latest write) and availability (every request to a live node gets a non-error answer). Partition tolerance is not a free choice, since networks fail.
  • Q: Why is "CA" not a real option for a distributed database?
    A system that gives up partition tolerance has to stop working when the network splits, which in practice means it is not distributed (a single node) or it is not actually available during partitions. Partitions happen whether you plan for them or not.
  • Q: What does PACELC add?
    Even without a partition there is a trade between latency and consistency: waiting for more replicas to acknowledge makes writes and reads slower but more consistent. It describes the everyday behaviour that CAP ignores.
  • Q: Compare ACID and BASE.
    ACID gives atomic, isolated, durable transactions with enforced constraints, at the cost of coordination. BASE (basically available, soft state, eventually consistent) accepts temporary disagreement between replicas for availability and speed. The choice is per piece of data, and many databases offer both within limits.
  • Q: What is eventual consistency, and what can a user see?
    If writes stop, all replicas converge. Meanwhile reads can return old values, go backwards in time, or miss a user's own write. Session guarantees (read-your-writes, monotonic reads) remove the most visible anomalies.
  • Q: Explain R + W > N.
    With N replicas, a write acknowledged by W and a read from R replicas must overlap in at least one replica when R + W is greater than N, so the read sees the latest acknowledged write. It is necessary for strong reads in a leaderless system but not sufficient against concurrent or partly failed writes.
  • Q: What are hinted handoff and read repair?
    Hinted handoff: another node holds a write meant for a temporarily down replica and hands it over when it returns. Read repair: a read that finds replicas disagreeing writes the freshest value back to the stale ones. Both reduce divergence; anti-entropy repair covers data that is never read.
  • Q: Hash or range partitioning?
    Hash for even load and point lookups; range for ordered scans but with hot-spot risk on sequential keys. A common compromise is a hashed partition key with a sorted key inside the partition.
  • Q: What is consistent hashing and why use virtual nodes?
    Nodes and keys are placed on a ring and a key belongs to the next node clockwise, so adding or removing a node moves only about 1/N of the keys. Virtual nodes give each machine many ring positions, which evens out load and spreads rebuild work across peers.
  • Q: Local vs global secondary index?
    A local index lives with each partition and is cheap to write but a query by the indexed value must ask every partition. A global index is partitioned by the indexed value, so a lookup is fast, but writes are costlier and the index is often eventually consistent.
  • Q: How do you model one-to-many in a document or key-value store?
    Embed when the children are few, bounded and read with the parent. Reference when they are many or shared: separate items sharing the parent's partition key and sorted by a sort key, so a single range query returns them.
  • Q: How do you prevent lost updates without transactions?
    Optimistic concurrency: keep a version on the item and make the write conditional on the version read. A failed condition means someone else wrote first, so re-read and retry.
  • Q: Why is last-write-wins dangerous?
    Clocks drift, so the "latest" timestamp may not be the latest event, and concurrent updates are silently discarded. It is acceptable only for idempotent whole-value overwrites. Otherwise use versions, version vectors with merge logic, or CRDTs.
  • Q: When would you choose a relational database over NoSQL?
    When queries are unknown or ad-hoc, when multi-row transactions and constraints are central, when data is highly relational, or when it fits on one server. NoSQL is justified by a measured need: scale, latency, flexible schema, or a specific data shape.
  • Q: How would you migrate a service from a relational database to a NoSQL store safely?
    Strangle one slice at a time, load history by backfill, keep stores in sync with change data capture or an outbox rather than naive dual writes, compare results in shadow reads, cut reads over first and writes next, and keep a rollback path.
13
Chapter Thirteen - Summary

Cheat Sheet and Recap

Cheat sheet
ConceptRuleRemember
CAPDuring a partition choose consistency or availabilityPartition tolerance is not optional
PACELCElse (no partition): latency or consistencyThe everyday trade is latency
ConsistencyLinearizable, causal, read-your-writes, monotonic, eventualSession guarantees fix most user-visible oddities
QuorumR + W > N makes read and write sets overlapR + W = N is not enough
RepairHinted handoff, read repair, anti-entropyQualitative, product- and version-specific
PartitioningHash for even load, range for scansAvoid sequential partition keys; split hot keys
Consistent hashingKey goes to next node clockwise; use vnodesAdding a node moves about 1/N of keys
Secondary indexLocal: cheap write, scatter read. Global: fast read, costly writeIndexes multiply write cost
ModellingList the queries first; denormalise; embed or referenceSchema moves into the application
ConcurrencyVersion + conditional writeAvoid last-write-wins for mergeable data
MigrationStrangler + outbox/CDC, not naive dual writeCompare before you cut over
โˆ‘ NoSQL fundamentals - what to remember
  • NoSQL is a set of trades, not an upgrade: scale-out, latency and flexible schemas bought with joins, transactions and strong consistency by default.
  • CAP is about partitions: you choose C or A while the network is split. PACELC adds the latency-versus-consistency trade of normal operation.
  • Consistency is a spectrum. Know the session guarantees; they are what users notice.
  • Quorums work through overlap (R + W > N); they are not a full serial guarantee.
  • Partition key design decides everything: high cardinality, known on every request, even load.
  • Model from queries, denormalise on purpose, and keep a version in every item.
  • Start relational, and add a NoSQL store for a named, measured need.
Version note

Checked on 2026-10-31: DynamoDB read consistency (eventually consistent by default; strongly consistent optional; global secondary indexes eventually consistent only) against the AWS DynamoDB Developer Guide; Cassandra's description of hinted handoff, read repair and virtual nodes against the Apache Cassandra architecture documentation. The Java program was compiled with JDK 21 and its output is a real run. Product limits (item sizes, transaction item counts, hint lifetimes) are deliberately not quoted here: see each product's limits page for your version.

Next in this section
  • Choosing a Database - the map of database families and a decision guide by access pattern.
  • 02 Redis - the in-memory data-structure store, used as a cache and more.
  • 03 DynamoDB - partition keys, sort keys, indexes and capacity modes, applied.
  • Event-Driven Patterns - the outbox and idempotency patterns used for safe migration.