Secondary indexes across shards: local or global

After you shard a database (lesson 69), the interviewer asks: "The data is split by user id. How do you find a user by email?" This is a very common follow-up, because the shard key cannot be every column.

A weak answer is "add an index", as if the table were on one server. A strong answer says which shard the query goes to, and what each choice costs in reads, writes and delay. Then the interviewer asks the next question: "What if the user changes the email?" or "How do you keep emails unique?"

At SDE-1 level you should see that a query without the shard key must ask every shard. At SDE-2 level you should compare a local index and a global index with numbers. At SDE-3 level you should explain how the index stays up to date, and how a unique rule works across shards.

A school keeps 20 registers, split by roll number. A parent says: "Find the child whose phone number is this one." The clerk must open all 20 registers. Or the school keeps one extra book that lists phone numbers in order, with the roll number next to each. The book is quick to read, but every new child must now be written in two places.

Here the shard key is the user id. A query by email does not contain it, so no one knows which shard holds the row. An index on email is a secondary index: an index on a column other than the shard key (lesson 36 explains indexes).

Ask every shard (scatter-gather, lesson 69). The query goes to all N shards, and the answer waits for the slowest one. That is N queries for one lookup.

How likely is it that one of them is slow? Say each shard is slow for a share p of its queries. Assumption: shards are slow independently of each other. One shard is not slow with chance 1 - p. All N shards are not slow with chance (1 - p) to the power N. So at least one is slow with chance 1 - (1 - p) to the power N.

Local index. Each shard indexes only its own rows. A write is cheap: one shard, one transaction (all changes together, or none). But a query without the shard key still goes to every shard. Each shard answers fast, but there are still N calls.

Global index. The index is itself split, by the indexed column. It is sharded like any table: hash the email to pick the index shard. The entry "email -> user id" lives on that index shard. A lookup goes to one index shard, which returns the user id. Then it goes to the data shard for the row: two calls.

The cost is on writes. A new user is written on the data shard and on the index shard, so one write touches two shards. A change of email costs more: delete the old entry, add the new one, and change the row. That is three writes on up to three shards.

Keeping a global index up to date. There are two ways.

• In the same step. Both shards change in one transaction. The index is always right, but a transaction across two shards is slow and hard (lesson 69).
• A moment later. The data shard saves first, and a background step updates the index. All copies end up the same after some time (eventual consistency, lesson 66). In the gap, a new user cannot be found by email yet, and a changed email can still be found by its old value. So check the row after you fetch it.

A real example. DynamoDB, a database from Amazon, offers both kinds by name. Its global secondary index is split by its own key and is updated in the background, usually within a fraction of a second. Its local secondary index is different. You can use it only together with the table's own key, so it does not solve this problem (see Sources).

A unique rule across shards. "Email must be unique" cannot be checked inside one shard. One place must decide. Use a lookup table: a small table with the email as its primary key and the user id as its value.

1. Insert into the lookup table first. If the email exists, the insert fails, and you reject the sign-up.
2. Then insert the user.
3. If that second insert fails, remove the lookup row. A clean-up job removes rows that are left over.

This takes two steps with a repair, not one transaction.

Copy extra columns into the index. If the index entry also holds the name, a login lookup needs no second call. The cost is a live copy (lesson 35): when the name changes, two places change.

Adding an index to live data. Build it in the background, fill it in small batches, and switch the queries only when it is complete (lesson 41).

-- Run in SQLite 3.50. In a real system this table is sharded by email.
CREATE TABLE email_lookup (
  email   TEXT PRIMARY KEY,   -- the unique rule lives here
  user_id INTEGER NOT NULL,
  name    TEXT                -- a copied column
);
INSERT INTO email_lookup VALUES ('asha@example.in', 1001, 'Asha');
INSERT INTO email_lookup VALUES ('asha@example.in', 2002, 'Asha K');
-- Error: UNIQUE constraint failed: email_lookup.email
SELECT user_id, name FROM email_lookup WHERE email = 'asha@example.in';
-- 1001 | Asha

A local index is only CREATE INDEX users_email_idx ON users(email); on every shard.

Assumptions (made up to teach with): 20 shards; 100 million users; 2,000 lookups by email per second. Each shard is slow for 1% of queries (p = 0.01), and shards are slow independently. There are 50 new users per second, and an index entry is 50 bytes.

• Local index: each lookup is 20 shard queries. 2,000 x 20 = 40,000 shard queries per second. The chance that at least one shard is slow: 1 - 0.99 to the power 20 = 1 - 0.818 = 18.2% of lookups. If shards slow down together, for example after a bad release, the real chance is lower than 18.2%, but the work is still 20 calls.
• Global index: each lookup is 1 index query and 1 row fetch = 2 queries. 2,000 x 2 = 4,000 queries per second, ten times fewer. This answers the predict box: 20 against 2. The slow chance is about 2 x 1% = 2%.
• The write side: 50 new users per second cost 50 writes with a local index, and 50 x 2 = 100 writes with a global index.
• Size: 100,000,000 x 50 bytes = 5,000,000,000 bytes = 5 GB of index, split over index shards.
• With the name copied into the index, the 2,000 row fetches per second are not needed for a login.

Top: with a local index the lookup goes to every one of the 20 shards, drawn as 20 small boxes, so it costs 20 queries. Bottom: with a global index the lookup goes to one index shard, which finds the user id, and then to one data shard, which holds the row. That costs 2 queries.

• A local index solves cheap writes (50 per second, one shard each). It costs 20 calls for every lookup by email. I would change it when the lookup by email is frequent.
• A global index solves the lookup (2 calls). It costs a second write for each change, and an index that can be a moment old. I would change back when the data changes very often and lookups are rare.
• A global index on a column with few values (such as status) puts all its entries on a few index shards. Use it for columns with many values, such as email.

"I added an index on email, so the lookup is fast." A local index makes each shard fast, but the query still goes to 20 shards.

"A global index is always better, because it needs fewer calls." It needs fewer calls to read, but every write touches two shards. A change of email touches three.

"A global index is always up to date." If it is updated in the background, a new user may not be found for a short time.

1. What is the difference between a local and a global secondary index?
2. There are 50 shards, 1,000 lookups per second, and each shard is slow for 2% of queries, independently. How many shard queries per second does a local index cause? What share of lookups meets at least one slow shard?
3. Sign-up must reject a second account with the same email, and the users are sharded by user id. How do you do it?