Query routing in sharded systems isn’t about finding the right shard; it’s about the cost of finding the right shard.

Let’s watch a query hit a sharded database. Imagine we have a users table sharded by user_id.

SELECT * FROM users WHERE user_id = 12345;
  1. Client sends query: The application connects to a "router" or "coordinator" node. This node doesn’t hold data itself but knows how to find it.
  2. Router parses query: It breaks down the SQL into its components: SELECT, FROM users, WHERE user_id = 12345.
  3. Router identifies shard key: It sees user_id is the column used for sharding.
  4. Router consults shard map: It looks up its internal mapping (often a hash ring or a direct lookup table) to determine which shard(s) hold data for user_id = 12345. Let’s say it’s shard S5.
  5. Router dispatches query: It forwards the exact same SELECT statement to S5.
  6. Shard S5 executes: S5 runs the query against its local data and returns the result.
  7. Router returns result: The coordinator sends the result back to the client.

This seems straightforward, but the devil is in the details of steps 2-4.

The Problem: Ambiguity and Overhead

The core challenge is how the router parses and dispatches effectively.

  • Parsing: If the query is complex, contains multiple conditions, or uses functions on the shard key, the router might not be able to determine the exact shard(s) upfront. This forces it to send the query to all shards and then aggregate results, which is incredibly inefficient.
  • Dispatching: Even if the shard is identified, managing the shard map and ensuring the query goes to the correct instance of a shard (e.g., the primary for writes, a replica for reads) adds complexity. Network latency, node failures, and dynamic resharding all complicate this.

Key Components and Their Roles

  • Query Parser: This is standard SQL parsing, but with an awareness of sharding. It needs to recognize the shard key column and any operations applied to it.
  • Shard Mapper/Directory: This is the central piece. It stores the mapping between shard keys (or ranges of keys) and the actual physical shard locations. This map needs to be highly available and frequently updated.
  • Dispatcher: This component takes the identified shard(s) and routes the query to the appropriate node(s). It handles connections, retries, and potentially load balancing across replicas.

How to Control It: Configuration Levers

  1. Shard Key Definition: This is fundamental. When you create a sharded table, you define user_id as the shard key.

    CREATE TABLE users (
        user_id BIGINT PRIMARY KEY,
        username VARCHAR(255),
        email VARCHAR(255)
    ) SHARD BY HASH(user_id);
    

    The SHARD BY HASH(user_id) clause tells the system this is the key. The choice of HASH versus RANGE sharding also dramatically impacts routing. HASH distributes evenly but makes range queries painful; RANGE makes range queries efficient but can lead to hot shards.

  2. Shard Map Configuration: The shard map itself is often managed by a separate service (like ZooKeeper, etcd, or a dedicated discovery service) or embedded within the coordinator nodes. You configure how these coordinators discover and update their shard maps. This often involves pointing them to a central configuration store. For example, a coordinator might be configured with:

    shard_map_service:
      type: etcd
      endpoints: "etcd-1:2379,etcd-2:2379"
      key_prefix: "/sharded_db/users/"
    

    This tells the coordinator to fetch its shard map for the users table from etcd under the /sharded_db/users/ path.

  3. Query Rewrite Rules: Some systems allow you to define rules that rewrite queries to be more shard-friendly. For instance, if a query uses DATE(timestamp_col) and timestamp_col is the shard key, the system might rewrite it to timestamp_col BETWEEN '2023-10-26 00:00:00' AND '2023-10-26 23:59:59' if it knows the shard key is date-based, allowing it to target specific date shards.

Seeing It in Action: A "Broadcast" Query

Consider a query that cannot be routed to a single shard:

SELECT COUNT(*) FROM users WHERE username LIKE 'a%';

If username is not the shard key and user_id is, the router cannot determine which shard(s) contain usernames starting with 'a' without scanning.

What happens:

  1. The router parses the query.
  2. It recognizes username LIKE 'a%' is not a direct shard key lookup.
  3. It consults the shard map and sees shards S1 through S10 exist for the users table.
  4. Crucially: Instead of picking one shard, it sends the exact same SELECT COUNT(*) FROM users WHERE username LIKE 'a%'; query to every single shard (S1, S2, …, S10).
  5. Each shard executes the query locally and returns its count (e.g., S1 returns 1500, S2 returns 1200, etc.).
  6. The router receives all these individual counts and sums them up to produce the final result.

This "broadcast" or "scatter-gather" pattern is the fallback when shard-aware routing fails. It’s a powerful way to ensure correctness but comes with a significant performance penalty because every data node is involved in processing a query that might only need data from one or two.

The Counterintuitive Part

Many assume that in a sharded system, queries are always sent to a single, precisely identified shard. The reality is that the system prefers to do this, but it will always broadcast a query to all shards if it cannot determine the specific shard(s) containing the relevant data. This is a critical design choice: correctness over potential performance gains from avoiding broadcast. The system prioritizes giving you the right answer, even if it has to ask everyone.

The next challenge you’ll face is how to optimize these broadcast queries when they become frequent.

Want structured learning?

Take the full Sharding course →