Articles

Tune Sharding Performance: Reduce Inter-Shard Traffic (2026)

April 18, 2026 7 min read

Sharding doesn’t magically make data independent; it just moves the problem of finding data from a single giant table to a distributed system where finding data means talking to multiple machines.

Let’s say we have a users table sharded by user_id. We want to find all users in the eu-west-1 region.

-- This query might look simple, but it's a performance killer
SELECT * FROM users WHERE region = 'eu-west-1';

If region isn’t part of the sharding key, this query will be a "scatter-gather" operation. The database will have to ask every single shard if it has any users in eu-west-1. Each shard will scan its local data and respond. This generates a ton of network traffic and CPU load on each shard, even if most shards have zero results.

The goal is to make this query hit only the shards that actually contain data for eu-west-1.

The Problem: Non-Sharded Filters

The core issue is querying on fields that are not part of your sharding key. When you filter on a non-sharded field, the database has no efficient way to know which shard(s) to ask. It has to ask all of them.

Common Causes and Fixes

  1. Querying on frequently filtered, non-sharded columns:

    • Diagnosis: Look at your slow query logs. Identify queries that are taking a long time and see the WHERE clauses. If a column appears frequently in these filters and is not in your sharding key, that’s your culprit. For example, if you shard by tenant_id and frequently query WHERE status = 'active', you have a problem.
    • Fix:
      • Option A (Best): Re-shard. If a column is critical for filtering and performance, consider making it part of your sharding key, or even the primary sharding key. This is the most disruptive but most effective fix.
        • Example: If you shard by user_id but frequently filter by country_code, and country_code has a reasonable distribution of values (e.g., not 99% of users in one country), consider sharding by a composite key like (country_code, user_id) or using a derived sharding key based on country_code.
        • Why it works: The database can now directly route queries for a specific country_code to the relevant shard(s) without asking others.
      • Option B (Good): Create a secondary index. If re-sharding isn’t feasible, create a secondary index on the frequently filtered column.
        • Example: CREATE INDEX idx_users_region ON users (region);
        • Why it works: The database can use the index to find relevant rows without scanning the whole table on each shard. However, for cross-shard queries involving this index, it still might need to consult multiple shards, but the index lookup on each shard is much faster than a full table scan. This is less effective than re-sharding if your queries must be routed to a single shard.
      • Option C (Workaround): Denormalize or use a materialized view. If the filter is on a dimension that doesn’t change often, you might denormalize.
        • Example: If you have users and orders, and you shard orders by user_id but want to find orders by user_country, you could store user_country directly in the orders table. Or, create a materialized view that pre-aggregates or pre-filters data for specific regions.
        • Why it works: The data is colocated with the query’s filter criteria, eliminating the need for cross-shard lookups.

  2. Joins involving non-sharded columns:

    • Diagnosis: Similar to the above, look for slow queries involving JOIN clauses where columns used in the ON condition are not part of the sharding key for either table.
    • Fix:
      • Option A (Best): Ensure join keys are sharded together. If table A is sharded by user_id and table B is sharded by order_id, but you often join A and B on user_id, you have a problem. The optimal solution is to shard both tables using the same key (user_id in this case).
        • Why it works: When you join on the sharding key, the database knows that rows with the same user_id will reside on the same shard in both tables, allowing for efficient local joins.
      • Option B (Good): Broadcast small tables. If one table is small and frequently joined with a large sharded table, consider "broadcasting" the small table to every shard.
        • Example: If you have a user_profiles table (small, sharded by user_id) and a user_activity table (large, sharded by user_id), and you often join them, you can configure user_profiles to be broadcast.
        • Why it works: Each shard of user_activity will have a full copy of user_profiles, allowing local joins. This incurs storage overhead but eliminates cross-shard join traffic.
      • Option C (Workaround): Re-architecting queries. Sometimes, the join can be rewritten to avoid cross-shard operations, perhaps by pulling data from one side and processing it on the other. This is highly application-specific.

  3. Aggregation queries across shards:

    • Diagnosis: Queries using COUNT(*), SUM(), AVG() on columns that are not part of the sharding key, especially when run across a large dataset.
    • Fix:
      • Option A (Best): Pre-aggregate or use materialized views. For common aggregations, create separate tables or materialized views that store aggregated data, potentially sharded by the dimension you want to aggregate on.
        • Example: Instead of SELECT COUNT(*) FROM events WHERE event_type = 'click', have a daily_event_summary table sharded by event_date which stores pre-calculated counts per event_type.
        • Why it works: The aggregation is performed incrementally or periodically on subsets of data, and the final query hits a pre-computed result set.
      • Option B (Workaround): Optimize the scatter-gather. If pre-aggregation isn’t an option, ensure the underlying data is indexed to make the scan on each shard as fast as possible. The scatter-gather itself is unavoidable, but you can speed up the "gather" part on each shard.

  4. Inadequate Sharding Strategy:

    • Diagnosis: Your sharding key was chosen without considering common query patterns. Perhaps you chose user_id because it’s unique, but most queries are by tenant_id or region.
    • Fix: Re-evaluate your sharding strategy. This is a major undertaking but often necessary for long-term scalability.
      • Composite Sharding: Use a combination of keys, e.g., (tenant_id, user_id).
      • Hash-based Sharding: If a field has skewed distribution, hashing its value can distribute it more evenly.
      • Range-based Sharding: Useful for time-series data or ordered data where you want to query ranges.
      • Why it works: A well-chosen sharding key directly routes queries to specific shards or a small subset of shards, minimizing inter-shard communication.

  5. Cross-Shard Transactions:

    • Diagnosis: Applications that require updates to multiple rows that might reside on different shards within a single ACID transaction. This forces the system to coordinate across shards, incurring significant overhead and potential for deadlocks.
    • Fix:
      • Option A (Best): Avoid them. Re-architect your application logic to perform operations that are entirely contained within a single shard whenever possible.
      • Option B (Workaround): Two-Phase Commit (2PC). If absolutely necessary, use a distributed transaction coordinator, but be aware of the performance implications. Most modern sharded databases try to abstract this away or discourage its use for performance reasons.
      • Why it works: 2PC ensures atomicity across distributed participants, but the coordination protocol is inherently slow and complex.

  6. Improperly configured routing layer:

    • Diagnosis: If you use a separate query router or proxy (like ProxySQL, Vitess, etc.), ensure it’s correctly configured to understand your sharding scheme and can accurately direct queries. Misconfigurations here can lead to unnecessary scatter-gather operations.
    • Fix: Review your router’s configuration files. Ensure table mappings, sharding keys, and connection pools are set up according to your database’s sharding topology.
      • Example: In Vitess, check keyspace.json and shard.json definitions to ensure sharding_column and column_vindexes are correctly defined for your tables.
      • Why it works: A correctly configured router can identify the target shard(s) based on the query and sharding schema, sending the query only to those shards.

After fixing these, the next error you might see is related to insufficient memory on individual shards to hold hot data or process complex queries, or perhaps a bottleneck in network bandwidth between your application and the shards if you’ve reduced inter-shard traffic but increased direct shard traffic.

Want structured learning?

Take the full Sharding course →