Articles

Load Test Sharded Databases: Validate Distribution (2026)

April 18, 2026 7 min read

Sharded databases don’t just distribute your data; they distribute your problems, and load testing is how you find out where they’re hiding.

Let’s say we’re load testing a sharded PostgreSQL setup using pg_shard (or a similar sharding solution like Citus). Our goal isn’t just to see if the aggregate throughput is high enough, but to ensure that individual shards aren’t becoming bottlenecks.

Here’s what a typical transaction might look like in our load test:

-- Insert a new customer record
INSERT INTO customers (customer_name, email, signup_date)
VALUES ('Alice Smith', 'alice.smith@example.com', NOW());

-- Fetch a customer by their ID
SELECT * FROM customers WHERE customer_id = 12345;

-- Update a customer's email
UPDATE customers
SET email = 'alice.smith.new@example.com'
WHERE customer_id = 12345;

These are simple operations, but when executed at scale across potentially hundreds or thousands of shards, the distribution of the workload becomes critical.

The Problem: Uneven Distribution

The core issue when load testing sharded databases is uneven distribution. If your sharding key isn’t well-chosen, or if your data has inherent skew, certain shards will end up with a disproportionately large share of the data and, consequently, the workload. This leads to:

  • Hotspots: A few shards become overwhelmed while others sit idle.
  • Performance Degradation: Queries hitting hot shards slow down significantly, impacting overall application performance.
  • Resource Exhaustion: CPU, memory, or I/O on specific nodes spike, leading to failures.

How to Validate Distribution During Load Testing

You need to actively monitor the distribution of requests and data across your shards.

1. Monitor Query Latency Per Shard

Diagnosis: During your load test, run queries directly against each shard to measure their individual latency for representative operations. You can use PostgreSQL’s pg_stat_statements or client-side instrumentation. For example, to check average query duration for a specific query type on a shard:

-- Connect to a specific shard (e.g., shard_1)
SELECT
    avg(total_exec_time) AS avg_duration_ms
FROM pg_stat_statements
WHERE query LIKE 'INSERT INTO customers%';

Fix: If you find significant latency differences (e.g., one shard’s INSERT queries average 500ms while others are 50ms), you need to re-evaluate your sharding strategy. This usually means:

  • Changing the Sharding Key: If customer_id is causing skew (e.g., sequential IDs are routed to early shards), consider a sharding key that distributes data more randomly or based on a dimension less prone to skew (e.g., a hashed version of customer_id or a region column if applicable).
  • Rebalancing Data: If your sharding solution supports it, trigger a rebalance operation to redistribute data and workload more evenly. For Citus, this might involve SELECT rebalance_table_வது('customers');.
  • Partitioning within Shards: For extremely skewed data within a shard, consider adding local partitioning on that shard based on a different key.

Why it works: This directly targets the symptom of hotspots – slow queries on specific shards. By identifying and fixing the slowest shards, you improve the overall performance ceiling.

2. Track Row Counts Per Shard

Diagnosis: Periodically check the number of rows in your sharded table on each individual shard.

-- Connect to a specific shard (e.g., shard_1)
SELECT count(*) FROM customers;

Or, if your sharding solution provides a meta-table:

-- Example for Citus
SELECT shardid, count(*) FROM citus_shards WHERE logical_table_name = 'customers' GROUP BY shardid ORDER BY shardid;

Fix: If row counts are wildly divergent (e.g., one shard has 10 million rows, others have 1 million), it indicates a sharding key that isn’t distributing data uniformly. The fix is the same as for query latency:

  • Re-evaluate Sharding Key: Choose a key that offers better cardinality and distribution.
  • Rebalance: Use your sharding solution’s rebalancing tools.

Why it works: Uneven row counts are a direct indicator of poor data distribution, which is the root cause of most sharding performance issues.

3. Monitor CPU/Memory/IO Per Node

Diagnosis: Use system monitoring tools (e.g., top, htop, Prometheus Node Exporter, Datadog) to observe resource utilization on the individual database nodes hosting your shards.

Fix: If a particular node consistently shows higher CPU, memory, or I/O usage than others, it’s likely a hot shard or a collection of hot shards residing on that node. The solution involves addressing the underlying uneven workload:

  • Identify Hot Shards: Correlate the high resource usage on a node with the specific tables and queries identified in the previous steps.
  • Adjust Sharding Key or Rebalance: As described above, these are the primary methods to move the workload.
  • Scale Node Resources (Temporary): In some cases, you might temporarily scale up the resources of the overloaded node, but this is a band-aid if the distribution is fundamentally broken. The long-term fix is distribution.

Why it works: Resource contention on specific nodes is the ultimate consequence of uneven distribution. Addressing the root cause (distribution) prevents these resource bottlenecks.

4. Analyze Query Plans for Shard-Specific Optimizations

Diagnosis: Examine EXPLAIN ANALYZE output for queries that are slow. Look for differences in query plans across shards. A common issue is when a query that should be routed to a single shard ends up broadcasting to all shards, or vice-versa.

-- Execute on the coordinator node for a query that hits a sharded table
EXPLAIN ANALYZE SELECT * FROM customers WHERE customer_id = 12345;

Fix:

  • Ensure Proper Sharding Key Usage: Verify that your application is using the sharding key correctly in WHERE clauses. If the sharding key is customer_id, queries must include customer_id to be routed efficiently.
  • Check for Broadcasts: If you see "Broadcast" in the EXPLAIN ANALYZE output for a query that should be shard-local, it means the query planner couldn’t determine the shard and sent it everywhere, wasting resources. This often happens when joins involve non-sharded tables or when the sharding key isn’t present in the WHERE clause.
  • Optimize Joins: If joining sharded tables, ensure join conditions align with sharding keys or consider colocation (if your sharding solution supports it) to keep related data on the same nodes.

Why it works: Query plans reveal how the database intends to execute your query. Deviations from the ideal, shard-local execution indicate inefficiencies that directly impact performance.

5. Monitor Network Traffic Between Nodes

Diagnosis: If your sharded database involves a coordinator node and multiple worker nodes (like Citus), monitor network traffic. Excessive traffic between the coordinator and workers, or between workers themselves, can indicate inefficient query routing or data shuffling.

Fix:

  • Reduce Cross-Shard Joins: Re-architect queries to minimize joins that require data from multiple shards if those shards are on different nodes.
  • Colocation: If your sharding solution supports it, colocate related sharded tables on the same nodes. This keeps the data for joins together, reducing network hops.
  • Optimize Broadcast Queries: As mentioned, ensure queries are routed to specific shards when possible, rather than broadcasting.

Why it works: Network traffic is a direct cost of distributed systems. High network I/O often signals that data is being moved unnecessarily between nodes, which is a symptom of poor distribution or query design.

6. Check for Lock Contention on Specific Shards

Diagnosis: Monitor lock wait times and the types of locks held on each shard.

-- Connect to a specific shard
SELECT
    pid,
    usename,
    application_name,
    client_addr,
    wait_event_type,
    wait_event,
    state
FROM pg_stat_activity
WHERE wait_event IS NOT NULL;

Fix: Lock contention on a shard is usually a secondary effect of a hot shard. If a shard is receiving a high volume of writes or updates, it’s more prone to locks blocking other operations. The fix is to alleviate the load on that shard:

  • Distribute Writes: Ensure write operations are evenly distributed across shards.
  • Optimize Transactions: Keep transaction durations short to minimize lock holding times.
  • Review VACUUM and ANALYZE: Ensure these background processes are running efficiently and not causing contention themselves.

Why it works: While not a direct measure of distribution, lock contention is a critical performance killer that is exacerbated by hot shards. Addressing the hot shard problem will often resolve lock contention.

The Next Hurdle

After you’ve achieved balanced distribution and consistent per-shard performance, your next load testing challenge will likely be optimizing distributed transactions and complex distributed joins.

Want structured learning?

Take the full Sharding course →