Distributed & NoSQL Databases

The core problem: a single machine has bounded disk, RAM, and CPU. When your dataset outgrows one machine, you must spread it across many. Partitioning (also called sharding) is the process of dividing data into disjoint subsets, each living on a different node.

Range partitioning assigns consecutive key ranges to nodes. If you store user records by user ID, node 1 might hold IDs 0–999999, node 2 holds 1000000–1999999, and so on. This is simple and supports range queries efficiently — to find all users with IDs 500–700, you talk to exactly one node. The danger is hot spots: if IDs are assigned sequentially and new users are always young accounts, all writes land on the last node. You must monitor and rebalance continuously.

Hash partitioning routes a key to hash(key) mod N where N is the number of nodes. This distributes writes uniformly but destroys range-query locality. Worse, when you add a node (N becomes N+1), almost every key remaps. A 10-node cluster adding an 11th node remaps ~90% of all keys — a catastrophic migration storm.

Consistent hashing solves the remapping problem. Imagine arranging all hash outputs (say, 0 to 2^64 − 1) around a circle. Each node occupies a point on the circle. Each key maps to the first node clockwise from its hash. When you add a node, only the keys that now fall between the new node and its predecessor move — about 1/N of the total, not N-1/N. Real systems (Amazon Dynamo, Apache Cassandra) use virtual nodes (vnodes): each physical machine owns dozens of positions on the ring, smoothing load imbalance and making data migration finer-grained.

Directory-based sharding (used by MongoDB's mongos router) maintains an explicit routing table: a metadata service maps key ranges to shards. This gives maximum flexibility but introduces a single point of coordination. Most real systems trade off between these approaches.

Secondary indexes and sharding interact painfully. A document database might shard by document ID, but if you need a secondary index on email, that index must either be scatter-gathered across all shards (global index, slower writes) or maintained locally per shard (local index, full-scatter reads).

# Consistent hashing: minimal Python sketch
import hashlib, bisect

class ConsistentHashRing:
    def __init__(self, nodes, vnodes=150):
        self.ring = {}          # hash -> node
        self.sorted_keys = []
        for node in nodes:
            for i in range(vnodes):
                h = self._hash(f"{node}:{i}")
                self.ring[h] = node
                self.sorted_keys.append(h)
        self.sorted_keys.sort()

    def _hash(self, key):
        return int(hashlib.md5(key.encode()).hexdigest(), 16)

    def get_node(self, key):
        h = self._hash(key)
        idx = bisect.bisect_right(self.sorted_keys, h) % len(self.sorted_keys)
        return self.ring[self.sorted_keys[idx]]

ring = ConsistentHashRing(["node-A", "node-B", "node-C"])
print(ring.get_node("user:12345"))   # deterministic, balanced
# Adding node-D: only ~25% of keys migrate, not ~75%

Why replicate? Two reasons: durability (if one machine dies, data is not lost) and availability/throughput (read from nearby replicas; spread read load). Replication is fundamentally about keeping multiple copies of data in sync across machines that can fail independently and communicate only over unreliable networks.

Single-leader replication (the default in PostgreSQL streaming replication, MySQL binlog replication, Kafka) designates one node as the leader or primary. All writes go to the leader; the leader appends them to a replication log and streams changes to followers (replicas). Reads can be served from any follower. The log is the source of truth — this is the same idea as a database WAL (write-ahead log). The hard question is what to do when the leader fails. Automatic failover requires followers to detect the failure (via heartbeats), elect a new leader, and inform clients — all while avoiding split-brain (two nodes both believe they are leader).

Replication lag and consistency models matter enormously. A follower might be seconds behind the leader. If a user writes a post and immediately reads it back from a stale replica, they see nothing — a violation of read-your-writes consistency. Systems offer a spectrum:

• Eventual consistency: all replicas converge eventually, but no timing guarantee.
• Monotonic reads: once you read a value, future reads will not return an older one.
• Read-your-writes (session consistency): you always see your own writes.
• Strong consistency / linearizability: every operation appears to take effect instantaneously at some point between its invocation and response, as if there is only one copy.

Multi-leader replication allows multiple nodes to accept writes, then reconcile. This is powerful for geographically distributed setups (each datacenter has a leader) but creates write conflicts: two users simultaneously editing the same document on different leaders. You must define a conflict-resolution strategy — last-write-wins (LWW), merge, or application-defined CRDTs.

Leaderless replication (Dynamo, Cassandra, Riak) lets clients write to any node. Consistency is enforced through quorum reads and writes: if N total replicas exist, a write to W replicas and a read from R replicas guarantees overlap if W + R > N. Setting W=R=ceil((N+1)/2) gives a majority quorum. With N=3, W=R=2: any read will see at least one node that has the write. But quorums do not provide linearizability in the presence of network partitions — subtle races still occur.

Vector clocks and version vectors track causality across replicas. Instead of a single timestamp, each node maintains a vector of logical clocks, one per replica. When two values have causally unrelated version vectors, they are concurrent — the database must surface both to the client for reconciliation.

# Quorum logic pseudocode for leaderless replication
# N replicas, write to W, read from R nodes

N, W, R = 3, 2, 2   # standard quorum: W + R > N

def write(key, value, version):
    acks = 0
    for replica in all_replicas(key):
        if replica.store(key, value, version):  # may time out
            acks += 1
    if acks >= W:
        return OK          # durably written
    else:
        return ERROR       # partial write — must be repaired

def read(key):
    responses = []
    for replica in all_replicas(key):
        responses.append(replica.fetch(key))   # (value, version)
    if len(responses) >= R:
        # Return value with highest version; kick off read-repair
        latest = max(responses, key=lambda r: r.version)
        read_repair(key, latest, responses)    # heal stale replicas
        return latest.value

# W + R > N guarantees at least one node in the read set
# saw the latest write — pigeonhole principle

Statement (informal): In a distributed data system, you can guarantee at most two of the following three properties simultaneously:

• C — Consistency (linearizability): every read receives the most recent write or an error.
• A — Availability: every request to a non-crashed node receives a non-error response.
• P — Partition tolerance: the system continues operating even when the network drops or delays messages between nodes.

The catch — P is not optional. Networks partition. Switches fail, cables are cut, datacenter links go dark. Any distributed system must handle partitions or it is not a distributed system at all. So the real trade-off is: when a partition occurs, do you sacrifice C or A?

CA systems (traditional single-node RDBMS, or multi-node systems that stop accepting writes during a partition) choose consistency over availability. PostgreSQL replication that halts writes when the leader cannot contact a quorum of replicas is CA.

CP systems (HBase, Zookeeper, etcd, Spanner) choose consistency over availability. During a partition, minority-side nodes refuse to serve requests rather than return stale data. This is the right choice when correctness is paramount (bank ledgers, leader election).

AP systems (Cassandra, DynamoDB, CouchDB) choose availability over consistency. Every node keeps accepting reads and writes during a partition. After the partition heals, conflicts are resolved, but you may temporarily read stale values.

The PACELC extension (Daniel Abadi, 2012) adds the missing dimension: even when there is No partition (the E-L-C part), you still face a latency/consistency trade-off. A CP system that must contact a quorum on every read is slower (higher latency) than an AP system that reads from any local replica. PACELC gives a 2×2 grid:

• PA/EL (e.g., Dynamo, Cassandra): AP during partition, low latency otherwise.
• PC/EC (e.g., Spanner): CP during partition, strong consistency (high latency) otherwise.
• PC/EL (e.g., PNUTS from Yahoo): CP during partition, best-effort consistency for latency.

Why the theorem matters even when partitions are rare. Most of the time your network is fine. But during a 5-minute partition at 3 AM, your system reveals which choice it made. CP systems become partially unavailable; AP systems accumulate divergence that must be reconciled. Neither is wrong — but knowing which you have determines how you design your application logic.

Common misconception: CAP's "C" is linearizability, which is a specific, strong definition. Many systems advertise "consistency" meaning something weaker (e.g., causal consistency, read-your-writes). Do not conflate them.

# Illustrating the partition choice: CP vs AP behavior

# During a network partition between Node A and Node B:

# --- CP behavior (e.g., etcd) ---
def cp_read(node_A, key):
    if node_A.can_reach_quorum():
        return node_A.linearizable_read(key)  # contact quorum, slow
    else:
        raise ServiceUnavailable()  # refuse rather than return stale

# --- AP behavior (e.g., Cassandra with ONE consistency level) ---
def ap_read(node_A, key):
    # Always respond, even if partitioned from other nodes
    return node_A.local_read(key)  # may be stale, but available

# The application chooses at query time in Cassandra:
# session.execute(query, consistency_level=ConsistencyLevel.QUORUM)  # CP-like
# session.execute(query, consistency_level=ConsistencyLevel.ONE)     # AP-like

The simplest possible data model: a global, distributed hash map. Every value is addressed by a key; the value is opaque bytes (from the database's perspective). Operations are GET(key), SET(key, value), DELETE(key), and sometimes atomic variants like CAS(key, old, new) (compare-and-swap).

Where they shine. Key-value stores are blazingly fast because they do almost no work per operation — no schema enforcement, no relational joins, no complex query planning. Redis serves millions of operations per second from a single node. DynamoDB handles trillions of requests per year at Amazon.

Redis is an in-memory key-value store with rich data structures: strings, lists, sets, sorted sets (ZSets), hashes, streams, and HyperLogLog sketches. The sorted set (ZSET) — a set where each member has a floating-point score, and members are ordered by score — is particularly powerful for leaderboards, rate limiters, and time-series queries. Redis Cluster partitions data across nodes using a fixed 16384-slot hash space.

DynamoDB is AWS's fully managed key-value/document store. Its design descends directly from the Amazon Dynamo paper (2007), which pioneered consistent hashing, vector clocks, and leaderless replication in a production system. DynamoDB uses a composite primary key (partition key + optional sort key), which gives it ordered access within a partition — bridging pure key-value and document semantics.

Use cases: session storage, cache layers (Redis as a cache in front of a relational DB), real-time leaderboards, feature flags, pub/sub messaging (Redis Streams), shopping carts (DynamoDB), rate limiting (Redis atomic increment with TTL).

Limitations: no secondary indexes without extra machinery, no rich querying, no joins. You must design access patterns before modeling data, not after.

import redis

r = redis.Redis(host='localhost', port=6379)

# Simple KV
r.set('session:abc123', '{"user_id": 42, "role": "admin"}', ex=3600)  # TTL 1 hour
data = r.get('session:abc123')

# Sorted set: leaderboard
r.zadd('leaderboard:weekly', {'alice': 9850, 'bob': 7200, 'carol': 9900})
top3 = r.zrevrange('leaderboard:weekly', 0, 2, withscores=True)
# [('carol', 9900.0), ('alice', 9850.0), ('bob', 7200.0)]

# Atomic rate limiter (sliding window)
def is_rate_limited(user_id, limit=100, window_seconds=60):
    key = f'ratelimit:{user_id}'
    pipe = r.pipeline()
    pipe.incr(key)
    pipe.expire(key, window_seconds)
    count, _ = pipe.execute()
    return count > limit

Document databases store semi-structured data as documents — typically JSON or BSON objects with nested fields and arrays. Unlike relational tables, documents do not need to share a schema. A users collection might have some documents with an address sub-object and others without, without ALTER TABLE nightmares.

MongoDB is the canonical document store. Documents live in collections (analogous to tables). You query with a rich filter language that supports equality, comparisons, array membership, text search, and aggregation pipelines. MongoDB supports multi-key indexes, geospatial indexes, and text indexes. Writes can be single-document atomic (always) or multi-document transactional (since MongoDB 4.0, using WiredTiger snapshots).

The document model and joins. The killer feature is embedding: instead of a foreign key to a comments table, you store comments as an array inside the blog post document. A single read fetches the post and all its comments. The trade-off: if comments are shared across posts, embedding duplicates them. Document databases offer references (manual joins) but you pay for them in application round-trips or lookup stages in the aggregation pipeline.

Schema design patterns specific to document DBs:

• Outlier pattern: most documents have N comments, but a few viral posts have 100,000. Store the first K comments inline; overflow into a linked extended document.
• Bucket pattern: instead of one document per IoT sensor reading, bucket readings by hour into one document. Reduces document count, improves range scans.
• Polymorphic pattern: heterogeneous object types (cats, dogs, birds) share a collection; each has a type discriminator field and different optional fields.

CouchDB takes a different approach: it uses HTTP/REST natively, each document has a revision ID for MVCC, and it offers offline-first sync (PouchDB in the browser replicates to/from CouchDB). This makes it ideal for edge/mobile use cases.

When documents win over relational: data is naturally hierarchical (product catalogs with varying attributes, medical records with flexible fields), schema evolves rapidly, reads tend to fetch a whole entity rather than joining fragments, and your team needs to iterate quickly without migrations.

// MongoDB — aggregation pipeline (JavaScript/Mongo shell syntax)
db.orders.aggregate([
  // Stage 1: filter to the last 30 days
  { $match: { created_at: { $gte: new Date(Date.now() - 30*86400*1000) } } },

  // Stage 2: group by product, sum revenue
  { $group: {
      _id: "$product_id",
      total_revenue: { $sum: { $multiply: ["$quantity", "$unit_price"] } },
      order_count:   { $sum: 1 }
  }},

  // Stage 3: join to products collection
  { $lookup: {
      from: "products",
      localField: "_id",
      foreignField: "_id",
      as: "product_info"
  }},

  // Stage 4: sort by revenue descending
  { $sort: { total_revenue: -1 } },
  { $limit: 10 }
])

Wide-column stores are often misnamed — they are not column-oriented analytics databases (like Parquet, Redshift, ClickHouse). They are row-oriented stores where each row can have a different set of columns, and columns are grouped into column families that are stored together on disk.

Google Bigtable (2006) invented this model. A Bigtable table is indexed by (row_key, column_family:column_qualifier, timestamp). The timestamp axis gives built-in versioning. Because rows are sorted by row_key and column families are co-located on disk, range scans over row_key prefixes are efficient.

Apache Cassandra (Facebook, 2008; open-sourced) combined Bigtable's data model with Dynamo's distributed architecture (consistent hashing, leaderless replication, tunable consistency). In Cassandra's CQL (Cassandra Query Language), a table has a partition key (determines which nodes hold the data) and an optional clustering key (determines sort order within a partition). This two-level key is the central design primitive.

HBase is an open-source Bigtable implementation on top of HDFS. It is strongly consistent (uses ZooKeeper for leader election), unlike Cassandra's tunable consistency model. HBase powers parts of Facebook Messenger's message storage.

Why wide-column for time-series and events? Consider storing every click event for every user. The partition key is user_id; the clustering key is event_timestamp DESC. Fetching the last 100 events for a user is a single-partition range scan — extremely fast. You can store millions of columns per row (one per event) without pre-declaring them. This pattern is the canonical use case for wide-column stores.

The data-modeling discipline: in Cassandra you must know your queries before you design your schema. You create one table per query pattern. Denormalization is expected and correct. If you need to look up a user's events by both user_id and session_id, you maintain two tables. Storage is cheap; cross-partition joins are not supported.

-- Cassandra CQL: designing for time-series access

-- Table 1: recent events per user (most recent first)
CREATE TABLE user_events (
    user_id   UUID,
    ts        TIMESTAMP,
    event     TEXT,
    payload   TEXT,
    PRIMARY KEY (user_id, ts)
) WITH CLUSTERING ORDER BY (ts DESC)
  AND default_time_to_live = 2592000;  -- 30-day TTL

-- Insert
INSERT INTO user_events (user_id, ts, event, payload)
VALUES (uuid(), toTimestamp(now()), 'click', '{"btn":"buy"}')

-- Efficient single-partition range scan:
SELECT * FROM user_events
WHERE user_id = :uid
  AND ts > '2026-05-01'
LIMIT 100;

-- Table 2: events per session (separate table, denormalized)
CREATE TABLE session_events (
    session_id UUID,
    ts         TIMESTAMP,
    event      TEXT,
    PRIMARY KEY (session_id, ts)
) WITH CLUSTERING ORDER BY (ts DESC);

Some data is fundamentally about relationships. In a relational database, finding friends-of-friends in a social network requires a self-join on a users table and a join on an edges table — and each hop multiplies the cost. After three or four hops, the query planner drowns. Graph databases are designed precisely for this pattern.

The property graph model (used by Neo4j, Amazon Neptune, ArangoDB) represents data as:

• Nodes with labels (:Person, :Movie) and properties ({name: "Alice", age: 30}).
• Directed edges with a type (:KNOWS, :ACTED_IN) and properties ({since: 2020}).

Graph databases store adjacency lists natively. Traversing one hop from a node is O(degree), not O(n log n) as in a relational join over millions of rows. This is called index-free adjacency in Neo4j's architecture.

Cypher query language (Neo4j's declarative language, now an open standard via openCypher) uses ASCII-art patterns. Finding mutual friends of Alice and Bob:

MATCH (alice:Person {name:'Alice'})-[:KNOWS]-(mutual)-[:KNOWS]-(bob:Person {name:'Bob'})
RETURN mutual.name

The pattern -[:KNOWS]- matches the relationship bidirectionally. No explicit JOIN syntax; the traversal is expressed as a graph pattern.

RDF and SPARQL are an alternative (used in academic and semantic-web contexts). Data is stored as subject-predicate-object triples. SPARQL is the query language. This model is more flexible but less performance-optimized than property graphs.

Use cases where graphs shine: social networks (friend recommendations, influence analysis), fraud detection (rings of accounts linked through shared phone numbers, addresses, devices), knowledge graphs (Google's Knowledge Graph, Wikidata), network/IT infrastructure dependency mapping, recommendation engines (collaborative filtering as graph traversal).

When graphs are overkill: your data has few relationships; most queries are simple lookups by ID; your team does not have graph modeling expertise. A relational database with proper indexing handles most "moderate relationship" workloads fine.

// Neo4j Cypher — fraud ring detection
// Find accounts sharing >=2 contact points with a flagged account

MATCH (flagged:Account {fraud: true})
      -[:SHARES_DEVICE|SHARES_PHONE|SHARES_ADDRESS]-(shared)
      -[:SHARES_DEVICE|SHARES_PHONE|SHARES_ADDRESS]-(suspect:Account)
WHERE suspect <> flagged
  AND suspect.fraud = false
WITH suspect, count(DISTINCT shared) AS shared_signals
WHERE shared_signals >= 2
RETURN suspect.account_id, shared_signals
ORDER BY shared_signals DESC
LIMIT 50

// In Python via neo4j driver:
// with driver.session() as session:
//     result = session.run(cypher_above)
//     for record in result:
//         print(record['suspect.account_id'], record['shared_signals'])

Replication and sharding raise an immediate question: how do multiple nodes agree on anything — who is the leader, whether a transaction committed, what value a key has — when messages can be delayed, duplicated, or lost?

The consensus problem: N nodes each propose a value; the system must decide on exactly one value, and all non-crashed nodes must agree. This must work despite arbitrary message delays and node crashes (but not Byzantine faults, which are harder).

Paxos (Lamport, 1989, widely published 1998) is the foundational consensus algorithm. It operates in two phases: Prepare (a proposer asks acceptors for a promise not to accept older proposals) and Accept (the proposer sends the chosen value; a majority of acceptors must respond). Paxos is notoriously hard to understand and harder to implement correctly.

Raft (Ongaro and Ousterhout, 2014) was designed to be understandable. It decomposes consensus into three sub-problems: leader election, log replication, and safety. Every Raft cluster has a single leader that manages all writes; followers replicate the leader's log. If the leader fails, followers time out and elect a new one via randomized timeouts. etcd (the backing store for Kubernetes) and CockroachDB use Raft.

Two-phase commit (2PC) is the classic distributed transaction protocol. A coordinator asks all participants to prepare (write to a temporary area, lock resources, and promise they can commit). If all respond YES, the coordinator sends commit; if any say NO or time out, it sends abort. 2PC is blocking: if the coordinator crashes after sending PREPARE but before COMMIT, participants hold locks indefinitely — a correctness hazard. Three-phase commit (3PC) adds a pre-commit phase to reduce blocking under certain failure models, but it is not widely used because it does not handle network partitions either.

Distributed MVCC and snapshot isolation let multiple transactions run concurrently without locking. Spanner uses TrueTime (GPS + atomic clocks) to assign globally consistent timestamps to transactions, enabling external consistency across datacenters — something previously thought to require synchronous coordination on every transaction.

# Raft leader election — state machine pseudocode

class RaftNode:
    state: "follower" | "candidate" | "leader"
    current_term: int
    voted_for: Optional[NodeId]
    election_timeout: float   # randomized 150-300ms

    def on_election_timeout(self):
        # Follower converts to candidate
        self.state = "candidate"
        self.current_term += 1
        self.voted_for = self.id
        votes = 1   # vote for self
        self.reset_election_timer()
        for peer in self.peers:
            reply = peer.request_vote(self.current_term, self.id,
                                      self.last_log_index, self.last_log_term)
            if reply.vote_granted:
                votes += 1
        if votes > len(self.peers) // 2:  # majority
            self.become_leader()
        # else: remain candidate until timeout fires again

    def on_append_entries(self, msg):
        # Receiving a heartbeat from a valid leader resets timer
        if msg.term >= self.current_term:
            self.state = "follower"
            self.current_term = msg.term
            self.reset_election_timer()

For two decades, the conventional wisdom was: "If you need scale, give up ACID. If you need ACID, stay on one machine." NewSQL systems rejected this compromise. They asked: can we build a system that is distributed, horizontally scalable, and still provides full ACID transactions with SQL?

Google Spanner (2012) was the answer. Spanner shards data using range partitioning with automatic rebalancing, replicates each shard using Paxos across multiple datacenters, and achieves external consistency (the strongest possible consistency — stronger than serializable) using TrueTime: GPS receivers and atomic clocks in each datacenter allow Spanner to bound clock uncertainty to ~7ms, enabling globally ordered transactions without a global lock. Spanner serves both OLTP and limited OLAP workloads for Google's core products.

CockroachDB is an open-source Spanner-inspired system. It uses a layered architecture: a distributed key-value store (using Raft per shard for consensus), a SQL layer on top, and MVCC for concurrency. It achieves serializable isolation by default. CockroachDB stores all data in RocksDB (a key-value LSM-tree) on each node.

TiDB (PingCAP) is another open-source NewSQL system, popular in Asia, that separates storage (TiKV, a Rust key-value store using Raft) from compute (TiDB SQL layer). It is MySQL-compatible, making migration from MySQL straightforward.

The LSM-tree: RocksDB and many distributed databases use a Log-Structured Merge-tree for local storage. Writes go to an in-memory memtable (sorted), flushed to sorted string tables (SSTables) on disk, and periodically compacted (merged and re-sorted) to maintain read performance. LSM-trees are write-optimized (sequential disk I/O, no random writes) at the cost of read amplification (a key may be in multiple levels) and write amplification (data is rewritten during compaction). B-trees (used by PostgreSQL, MySQL InnoDB) are read-optimized but slower for write-heavy workloads.

When NewSQL is the answer: you have outgrown a single PostgreSQL node, you need the relational model and SQL queries, your workload is OLTP (many small, concurrent transactions), and you cannot tolerate eventual consistency in your data model. If you just need read scaling, read replicas + PgBouncer + Citus may be sufficient.

-- CockroachDB: distributed SQL with serializable isolation
-- This behaves like PostgreSQL but scales horizontally

-- Table is automatically sharded by primary key across nodes
CREATE TABLE accounts (
    id      BIGINT PRIMARY KEY DEFAULT unique_rowid(),
    user_id BIGINT NOT NULL,
    balance DECIMAL(18,2) NOT NULL,
    INDEX (user_id)
);

-- Serializable transaction: transfer between accounts
-- CockroachDB uses optimistic concurrency; retries on conflict
BEGIN;
  UPDATE accounts SET balance = balance - 100
  WHERE id = 1 AND balance >= 100;

  UPDATE accounts SET balance = balance + 100
  WHERE id = 2;

  -- If concurrent tx modified either row, COMMIT retries
COMMIT;

-- CockroachDB's retry loop in Go:
-- err := crdb.ExecuteTx(ctx, db, nil, func(tx *sql.Tx) error {
--     return transferFunds(tx, from, to, amount)
-- })

This is the most practically important question in database architecture, and it is too often answered by hype rather than analysis. Let us be rigorous.

Start with relational. Relational databases (PostgreSQL, MySQL) are extraordinarily capable. PostgreSQL supports JSONB (binary JSON with indexing), full-text search, geospatial queries (PostGIS), time-series (TimescaleDB extension), materialized views, and window functions. For the vast majority of applications — even those with tens of millions of rows and thousands of concurrent users — a well-tuned PostgreSQL instance with read replicas is the right answer. It is ACID, has decades of operational tooling, and has a query optimizer that handles complex patterns you never thought to optimize manually.

Legitimate reasons to leave SQL:

1. Data volume and write throughput that genuinely exceeds one machine. You have benchmarked, added replicas, and still cannot keep up. Cassandra, DynamoDB, or a NewSQL system like CockroachDB may be needed.
2. Data model fundamentally ill-suited to relations. Highly variable schema (product catalog with hundreds of different attribute shapes) → document store. Deep multi-hop relationship traversal (>3 hops at scale) → graph database. Append-only time-series at very high ingestion rates → InfluxDB or TimescaleDB.
3. Operational requirements. Fully managed, zero-ops → DynamoDB. Multi-region active-active with low latency → Cassandra or Spanner.
4. Specific consistency requirements. Eventual consistency acceptable for a shopping cart; strict serializability required for a payment ledger.

Illegitimate reasons to leave SQL (anti-patterns):

• "MongoDB is more flexible." Flexibility in schema often moves validation bugs from the database layer to application code, making them harder to catch.
• "NoSQL is faster." Redis is faster than PostgreSQL for cache hits, but PostgreSQL with a proper index is faster than a naive MongoDB scan.
• "Everyone is using Cassandra." Cargo culting. Cassandra's model requires you to pre-define every query pattern; for exploratory data access, PostgreSQL is far superior.
• Premature optimization: building a distributed database architecture before you have evidence of a scaling problem.

The polyglot persistence pattern: mature systems often use multiple databases, each for what it is best at. PostgreSQL as the primary relational store, Redis for caching and rate limiting, Elasticsearch for full-text search, Neo4j for the social graph. The cost is operational complexity and data synchronization. Use CDC (Change Data Capture) with tools like Debezium to stream changes from PostgreSQL into secondary stores consistently.

-- PostgreSQL can do more than you think.
-- JSONB with indexing:
CREATE TABLE products (
    id    SERIAL PRIMARY KEY,
    data  JSONB
);
CREATE INDEX ON products USING GIN(data);

-- Query: find all products where color is 'red'
SELECT id, data->>'name'
FROM products
WHERE data @> '{"attributes": {"color": "red"}}';

-- Window function: running total per user (no ORM does this well)
SELECT user_id, ts, amount,
       SUM(amount) OVER (PARTITION BY user_id ORDER BY ts) AS running_total
FROM transactions;

-- Recursive CTE: org hierarchy (graph-like query in SQL)
WITH RECURSIVE org AS (
    SELECT id, name, manager_id, 0 AS depth
    FROM employees WHERE manager_id IS NULL
  UNION ALL
    SELECT e.id, e.name, e.manager_id, o.depth + 1
    FROM employees e JOIN org o ON e.manager_id = o.id
)
SELECT * FROM org ORDER BY depth, name;