notexOPENOverview
Notex/Systems/Sharding and Secondary Indexes
08.09LESSON 09 OF 10 · 6 MIN READ

Sharding and Secondary Indexes

Always free to read

The sharding schemes we have discussed so far rely on the client knowing the partition key for any record it wants to access. This is most easily achieved in a key-value data model, where the partition key is the first part of the primary key (or the entire primary key), so we can use the partition key to determine the shard and thus route reads and writes to the node that is responsible for that key.

The situation becomes more complicated if secondary indexes are involved (see “Multicolumn and Secondary Indexes”). A secondary index usually doesn’t identify a record uniquely but rather is a way of searching for occurrences of a particular value: find all actions by user 123, find all articles containing the word hogwash, find all cars whose color is red, and so on.

Key-value stores often don’t have secondary indexes, but they are a standard feature of relational databases and common in document databases. This type of indexing is also the raison d’être of full-text search engines such as Solr and Elasticsearch. The problem with secondary indexes is that they don’t map neatly to shards. There are two main approaches to sharding a database with secondary indexes: local and global.

Local Secondary Indexes

In the first indexing approach, each shard independently maintains its own secondary indexes, covering only the records in that shard. It doesn’t care what data is stored in other shards. Whenever you write to the database—to add, remove, or update a record—you need to deal with only the shard containing the record that you are writing. For that reason, this type of secondary index is known as a local index. In an information retrieval context, it’s also known as a document-partitioned index.

For example, imagine you are operating a website for selling used cars. Each listing has a unique ID, and you use that ID as the partition key for sharding, as illustrated in Figure 7-9 (IDs 0 to 499 in shard 0, IDs 500 to 999 in shard 1, etc.). If you want to let users search for cars, allowing them to filter by color and by make, you need secondary indexes on color and make (in a document database these would be fields; in a relational database they would be columns). If you have declared the index, the database can perform the indexing automatically. For example, whenever a red car is added to the database, the database shard automatically adds its ID to the list of IDs for the index entry color:red. that list of IDs is also called a postings list.

Diagram showing local secondary indexes for two database shards, highlighting how searches for red cars access specific IDs within each shard's index.

Figure 7-9. With local secondary indexes, each shard indexes only the records it contains.

Warning

If your database supports only a key-value model, you might be tempted to implement a secondary index yourself by creating a mapping from values to IDs in application code. If you go down this route, you need to take great care to ensure that your indexes remain consistent with the underlying data. Race conditions and intermittent write failures (where some changes were saved but others weren’t) can very easily cause the data to go out of sync—see “The need for multi-object transactions”.

When reading from a local secondary index, if you already know the partition key of the record you’re looking for, you can just perform the search on the appropriate shard. Moreover, if you want only some results and don’t need all of them, you can send the request to any shard. However, if you want all the results and don’t know their partition key in advance, you will need to send the query to all shards and combine the results you get back, because the matching records might be scattered across all the shards. In Figure 7-9, for example, red cars appear in both shard 0 and shard 1.

This approach to querying a sharded database can make read queries on secondary indexes quite expensive. Even if you query the shards in parallel, it is prone to tail latency amplification (see “Use of Response Time Metrics”). It also limits the scalability of your application: adding more shards lets you store more data, but it doesn’t increase your query throughput if every shard has to process every query anyway.

Nevertheless, local secondary indexes are widely used—for example, MongoDB, Riak, Cassandra, Elasticsearch, SolrCloud, and VoltDB all use local secondary indexes.

Global Secondary Indexes

Rather than each shard having its own local secondary index, we can construct a global index that covers data in all shards. However, we can’t just store that index on one node, since it would likely become a bottleneck and defeat the purpose of sharding. A global index must also be sharded, but it can be sharded differently from the primary-key index.

Figure 7-10 illustrates what this could look like. The IDs of red cars from all shards appear under color:red in the index, but the index is sharded so that colors starting with the letters a to r appear in shard 0 and colors starting with s to z appear in shard 1. The index on the make of car is partitioned similarly (with the shard boundary being between f and h).

Diagram showing a global secondary index for car IDs, sharded by attributes such as color and make, across two primary key index shards.

Figure 7-10. A global secondary index reflects data from all shards and is itself sharded by the indexed value

This kind of index is also called term-partitioned. Recall from “Full-Text Search” that in full-text search, a term is a keyword in a text that you can search for. Here we generalize it to mean any value that you can search for in the secondary index.

The global index uses the term as the partition key, so that when you’re looking for a particular term or value, you can figure out which shard you need to query. Again, a shard can contain a contiguous range of terms (as in Figure 7-10), or you can assign terms to shards based on a hash of the term.

Global indexes have the advantage that a query with a single condition (such as color = red) needs to read from only a single shard to fetch the postings list. However, if you want to fetch records and not just IDs, you still have to read from all the shards that are responsible for those IDs.

If you have multiple search conditions or terms (e.g., searching for cars of a certain color and a certain make, or searching for multiple words occurring in the same text), those terms will likely be assigned to different shards. To compute the logical AND of the two conditions, the system needs to find all the IDs that occur in both of the postings lists. That’s no problem if the postings lists are short, but if they are long, it can be slow to send them over the network to compute their intersection.

Another challenge with global secondary indexes is that writes are more complicated than with local indexes, because writing a single record might affect multiple shards of the index (every term in the document might be on a different shard). This makes it harder to keep the secondary index in sync with the underlying data. One option is to use a distributed transaction to atomically update the shards storing the primary record and its secondary indexes.

Global secondary indexes are used by CockroachDB, TiDB, and YugabyteDB; DynamoDB supports both local and global secondary indexes. In the case of DynamoDB, writes are asynchronously reflected in global indexes, so reads from a global index may be stale (this is similar to the situation discussed in “Problems with Replication Lag”). Nevertheless, global indexes are useful if read throughput is higher than write throughput, and if the postings lists are not too long.

Progress is saved on this device.
KEEP EXPLORING

One note leads to another.

Back to the library →
⌕
Search across the entire library ⌘ K