A single database server eventually runs out of capacity, availability or geographic reach. A distributed database spreads data over several machines or sites while presenting it as one system. This is the theory behind sharded MySQL, MongoDB replica sets, Cassandra, DynamoDB and Google Spanner, and it explains why scaling a database is harder than scaling a web server. This article revisits the five units of the JNTUK course.
Unit 1: Introduction and architecture
A distributed database system (DDBS) is a collection of logically related data stored at several sites connected by a network, managed by a distributed DBMS (DDBMS) that hides the distribution from users.
Why distribute
| Reason | Benefit |
|---|---|
| Scalability | More machines, more capacity |
| Availability and reliability | One site failing does not stop the system |
| Performance and locality | Data near its users, parallel processing |
| Organisational fit | Branches or regions own their data |
Costs: complexity, network delay, harder consistency, security and administration.
Transparency
Users should not need to know the details:
| Transparency | Meaning |
|---|---|
| Distribution | Fragmentation and location are hidden |
| Replication | Copies are hidden |
| Network | Network details are hidden |
| Failure | Site failures are masked |
| Language / DBMS | A uniform interface |
Architectures
| Architecture | Description |
|---|---|
| Shared memory / shared disk / shared nothing | Parallel database designs; shared-nothing scales best and is used by modern distributed stores |
| Client-server | Clients talk to one or more database servers |
| Peer-to-peer | Every site is equal |
| Multidatabase / federated | Autonomous, possibly different, databases integrated through a mediator |
| Homogeneous vs heterogeneous | Same DBMS everywhere, or a mix |
A reference architecture has a global schema, a fragmentation schema, an allocation schema and local schemas.
Today: most production systems are shared-nothing: each node has its own CPU, memory and disk, as in Cassandra, MongoDB shards or CockroachDB. A cloud database with a read replica in another region is a small distributed database.
Unit 2: Distributed database design: fragmentation and replication
Design answers two questions: how to split the data and where to put the pieces.
Fragmentation
| Type | Splits by | Example |
|---|---|---|
| Horizontal | Rows | Customers of the south region at site A, others at site B |
| Vertical | Columns | Keep the key with each part; frequently used columns together |
| Hybrid (mixed) | Both | Vertical split, then horizontal |
Correct fragmentation follows three rules: completeness (every data item is in some fragment), reconstruction (the original can be rebuilt, by union for horizontal and join for vertical) and disjointness (no overlap, except the key in vertical fragments). Derived horizontal fragmentation splits a table by the fragmentation of a related table.
Allocation and replication
Allocation places fragments at sites to cut communication and balance load.
| Strategy | Meaning | Trade-off |
|---|---|---|
| No replication | One copy of each fragment | Simple; no fault tolerance |
| Full replication | A copy at every site | Fast reads, high availability; costly updates |
| Partial replication | Some fragments are copied | A compromise |
Sharding
Sharding is horizontal fragmentation in practice. The shard key decides placement:
| Method | Idea | Problem |
|---|---|---|
| Range | Key ranges per shard | Hot spots (for example, latest dates) |
| Hash | Hash of the key picks the shard | Even spread; range queries are costly |
| Directory / lookup | A table maps keys to shards | Extra lookup; flexible |
| Consistent hashing | Keys and nodes on a ring | Adding a node moves only a small share of keys |
A poor shard key creates hot shards, and resharding is expensive.
Today: MySQL sharding (Vitess), MongoDB shard keys, Cassandra partition keys and DynamoDB partition keys are all this unit. Choosing the key is the central design decision, and “cross-shard joins and transactions are costly” follows from it.
Unit 3: Distributed query processing and optimization
A query on global relations is translated into operations on fragments at different sites.
Steps
- Query decomposition: parse, validate, normalise and convert to relational algebra.
- Data localisation: replace global relations with their fragments and simplify, removing fragments that cannot match (fragment elimination).
- Global optimisation: choose the join order, join methods and sites.
- Local optimisation: each site optimises its sub-query.
Cost model
Distributed cost = local processing (CPU and I/O) plus communication cost (data transferred and the number of messages), and communication usually dominates. The aim is to move as little data as possible.
Join strategies
| Strategy | Idea |
|---|---|
| Ship whole relation | Send one relation to the other site and join there |
| Fetch as needed | Request matching tuples one by one (many messages) |
| Semi-join | Send only the join-column values of R to S’s site; S returns the matching tuples; reduces transfer. R ⋉ S = π(R) joined with S |
| Bloom join | Send a compact Bloom filter instead of the join column |
| Parallel joins | Hash-partition both relations and join the partitions in parallel |
Parallelism appears as inter-query (many queries at once), intra-query (one query in parts: pipelined and partitioned parallelism) and the speed-up and scale-up measures.
Today: distributed SQL engines (Presto/Trino, Spark SQL, BigQuery, Redshift) do exactly this: partition data, push filters down to the nodes that hold it and shuffle only what is needed. Slow cross-region queries are almost always a data-movement problem.
Unit 4: Distributed transactions and concurrency control
A distributed transaction touches data at several sites but must still be atomic (all or nothing) and consistent.
Two-phase commit (2PC)
A coordinator drives all participants:
| Phase | Steps |
|---|---|
| 1. Prepare (voting) | Coordinator asks all to prepare; each logs its state and votes yes or no |
| 2. Commit or abort (decision) | If all vote yes, the coordinator commits and tells everyone; otherwise it aborts |
2PC guarantees atomicity but is blocking: if the coordinator fails after participants vote yes, they must wait, holding locks. Three-phase commit (3PC) reduces blocking but is rarely used. Modern systems use consensus protocols (Paxos, Raft) to make the coordinator itself fault-tolerant, and sagas (sequences of local transactions with compensating actions) for long-running work across services.
Distributed concurrency control
| Method | Idea |
|---|---|
| Distributed two-phase locking | Locks at each site; primary copy or majority locking with replicas |
| Timestamp ordering | Global, unique timestamps (often Lamport or hybrid logical clocks) |
| Optimistic | Validate at commit time |
| MVCC | Readers use snapshots |
Distributed deadlock arises across sites, since no site sees the entire wait-for graph. It is handled by centralised or distributed detection (edge chasing) or by timeouts. Reliability uses write-ahead logs, checkpoints and recovery protocols after site and network failures, including network partitions.
Today: Google Spanner uses 2PC over Paxos groups with synchronised clocks (TrueTime), and CockroachDB and TiDB use Raft. The Raft consensus algorithm is also inside etcd, the store behind Kubernetes. Microservice teams use sagas and the outbox pattern, because distributed 2PC across services is impractical.
Unit 5: Replication, consistency, CAP and NoSQL
Replication
| Model | Description | Trade-off |
|---|---|---|
| Primary-backup (leader-follower) | Writes go to the leader and are copied to followers | Simple; failover needed |
| Multi-leader | Several nodes accept writes | Write availability across regions; conflict resolution |
| Leaderless (quorum) | Any node accepts writes; reads and writes use quorums | Highly available |
| Synchronous vs asynchronous | Wait for replicas, or do not | Safety versus latency; async allows replication lag |
With N replicas, a write quorum W and a read quorum R, reads see the latest write if R + W > N.
Consistency models
| Model | Meaning |
|---|---|
| Strong (linearizable) | Every read sees the latest write |
| Sequential, causal | Order is respected for related operations |
| Eventual | Replicas converge if updates stop; reads may be stale |
| Read-your-writes, monotonic reads | Session guarantees |
The CAP theorem and PACELC
During a network partition, a distributed system can offer only Consistency or Availability, not both (CAP). PACELC adds that even without a partition there is a trade-off between latency and consistency. ACID systems favour consistency; BASE systems (Basically Available, Soft state, Eventually consistent) favour availability. Conflict resolution uses last-write-wins, vector clocks or CRDTs (conflict-free replicated data types).
NoSQL families
| Family | Examples | Typical use |
|---|---|---|
| Key-value | Redis, DynamoDB | Caching, sessions, simple lookups |
| Document | MongoDB, Couchbase | Flexible records |
| Wide-column | Cassandra, HBase | Huge write-heavy workloads |
| Graph | Neo4j | Relationships |
| NewSQL / distributed SQL | Spanner, CockroachDB, TiDB, Aurora | SQL and ACID at scale |
Today: the CAP and quorum ideas explain Cassandra’s tunable consistency (
QUORUM,ONE), MongoDB’s write concerns and replica sets, Redis replication, and why a read replica may return stale data. Backups, failover tests and “split brain” prevention are the operations side of this unit.
One-minute recap
| Unit | Key idea | You meet it in |
|---|---|---|
| 1. Architecture | Transparency, shared-nothing, federated databases | Cassandra, MongoDB, cross-region replicas |
| 2. Fragmentation and replication | Horizontal, vertical, hybrid; sharding and shard keys | Vitess, Mongo and DynamoDB keys, hot shards |
| 3. Query processing | Minimise data movement; semi-joins | Trino, Spark SQL, BigQuery, Redshift |
| 4. Distributed transactions | 2PC, Paxos and Raft, sagas | Spanner, etcd, microservice sagas |
| 5. Replication, CAP, NoSQL | Quorums, eventual consistency, BASE | Replica lag, Cassandra consistency levels |
Next in Engineering Revisited: Embedded Systems.
Written by Claude, Anthropic’s AI assistant, as part of Engineering Revisited, a quick refresher of the B.Tech subjects one at a time.