Distributed Secondary Indexing

In a horizontally sharded database, routing queries by the primary Shard Key (e.g. WHERE user_id = 42) is instantaneous because the router knows precisely which node holds the record.

However, real-world applications must frequently query by non-shard attributes:

When a database is sharded by user_id, searching by email requires Distributed Secondary Indexing.

There are two fundamental approaches:

1. Document-Partitioned Local Secondary Indexes (Scatter-Gather)
2. Term-Partitioned Global Secondary Indexes

---

1. Document-Partitioned Local Secondary Indexes

In a Local Secondary Index (LSI) (used by MongoDB, Cassandra Local Indexes, PostgreSQL Citus), each physical shard maintains an index exclusively for the records residing on that local shard.

Write Path (Blazing Fast & Atomic):

• Inserting or updating a record requires modifying only one single shard.
• The row insert and its local index entry commit inside a standard single-node ACID transaction with zero network overhead.

Read Path (The Scatter-Gather Penalty):

• When searching by WHERE email = '[email protected]', the router has no idea which shard holds Alice's record.
• The router must scatter the query to every single shard in the cluster and gather the results!

---

2. The Mathematics of Tail Latency Amplification

Why are scatter-gather queries dangerous in large clusters?

Suppose a cluster has $N$ shards, and each individual shard has a $1\%$ chance of experiencing a temporary latency hiccup (e.g. background garbage collection or SSD write queue flush):

$p=0.01(\text{99th percentile single-node latency})$

The probability that a scatter-gather query hits at least one slow shard is:

$P(\text{Query Stalls})=1-(1-p{)}^N$

In a 100-shard cluster, over 63% of all client requests experience the 99th-percentile worst-case tail latency!

---

3. Term-Partitioned Global Secondary Indexes

In a Global Secondary Index (GSI) (used by Amazon DynamoDB, Google Spanner, CockroachDB), the index itself is partitioned independently by the indexed attribute (e.g. email).

Read Path (Blazing Fast Single-Shard Point Read):

• Querying WHERE email = '[email protected]' routes directly to Index Shard A.
• Index Shard A returns the exact primary key pointer in $<2\text{ms}$ without fanning out to any other nodes!

Write Path (Distributed Complexity):

• Inserting a new user requires writing to two different physical machines:
  1. Data Shard 1 (stores the user row).
  2. Index Shard A (stores the email index entry).
• Synchronous GSI (Spanner / CockroachDB): Uses distributed Two-Phase Commit with Raft/Paxos (higher write latency).
• Asynchronous GSI (DynamoDB): The database writes to Data Shard 1 immediately and replicates to the GSI asynchronously via a change-data-capture stream (Eventual Consistency).

---

4. Architectural Comparison: LSI vs GSI

---

5. Production Failure Postmortem: The 64-Shard Fanout Collapse

Incident Overview:

In 2021, a fintech banking service experienced severe latency degradation during peak morning trading, with p99 API latency climbing from $12\text{ms}$ to over $3,800\text{ms}$.

What Happened:

1. The accounts table was sharded across 64 PostgreSQL instances by account_number.
2. A newly deployed fraud-detection microservice looked up accounts by phone_number using a Local Secondary Index.
3. Every fraud check fanned out 64 parallel queries to all database nodes simultaneously.
4. At 15,000 fraud checks/sec, the database cluster was overwhelmed by 960,000 internal scatter-gather queries per second, exhausting connection pools and disk read queues.

Remediation:

• Provisioned a dedicated Global Secondary Index (GSI) table partitioned by phone_number with change-data-capture (Debezium + Kafka).
• Reduced query fanout from $64\text{ shards}$ to $1\text{ targeted shard}$, dropping p99 latency back to $4\text{ms}$.

---

6. Chapter Summary & Secondary Indexing Rules