Cross-Shard Queries: Handling Data That Spans Multiple Shards

The Query That Took 45 Minutes on Pinterest

Pinterest shards their MySQL database by user_id. Loading a user's boards, pins, and followers is blazing fast - single shard, single query. But their data team needed to answer: 'What are the top 100 most-pinned images globally?' This query had to scan ALL shards, extract top images from each, merge the results, and sort globally. With 4,000+ shards, the scatter-gather took 45 minutes. Their solution: a secondary analytics pipeline using Apache Spark that pre-computes global aggregations nightly, plus a Redis cache for real-time trending. The primary shard key (user_id) stayed optimal for the 99% case, while the 1% of cross-shard queries got their own dedicated infrastructure.

Scatter-Gather: How Cross-Shard Queries Work

Scatter-Gather Pattern:

  Client: SELECT COUNT(*) FROM orders WHERE date = '2024-01-15'

  Coordinator Node
    |---- Query ----> Shard 1: COUNT(*) = 1,234
    |---- Query ----> Shard 2: COUNT(*) = 987
    |---- Query ----> Shard 3: COUNT(*) = 1,456
    |---- Query ----> Shard 4: COUNT(*) = 1,102
    |                 (all execute in parallel)
    |<--- Results ---- (wait for ALL shards)
    |
    Merge: 1,234 + 987 + 1,456 + 1,102 = 4,779
    Return: 4,779

  Latency = MAX(shard_latencies) + merge_time
  Not SUM - parallel execution helps.
  But: one slow shard delays EVERYTHING.

  Problematic Queries (require cross-shard):
  - Global aggregations: SUM, COUNT, AVG across all data
  - Global sorting: ORDER BY + LIMIT across shards
  - JOINs between tables on different shard keys
  - UNIQUE constraints across shards
  - Full-text search across all data

Strategies to Avoid Cross-Shard Queries

Strategy Comparison:

  Strategy              Cost          Freshness      Best For
  ------------------    ----------    -----------    -----------------------
  Denormalization       Extra storage Real-time      Frequently accessed joins
  Secondary index       Extra system  Near-real-time Search/filter queries
  Materialized views    Compute cost  Periodic       Aggregation/analytics
  Broadcast tables      Memory        Real-time      Small reference data
  Application joins     Code complex  Real-time      Ad-hoc queries

  1. DENORMALIZATION:
  Instead of: JOIN orders ON users.id = orders.user_id
  Store: user_name, user_email directly in orders table.
  Each shard has everything it needs. No cross-shard join.

  2. SECONDARY INDEX (Elasticsearch):
  Shard by user_id (PostgreSQL)  ->  CDC  ->  Elasticsearch
  User queries: hit PostgreSQL (fast, single shard)
  Search queries: hit Elasticsearch (not sharded by user_id)

  3. MATERIALIZED VIEWS:
  Nightly job: compute global top products from all shards
  Store result in a central (non-sharded) table.
  Dashboard reads from materialized view, never scatter-gathers.

  4. BROADCAST TABLES:
  Small tables (countries, currencies, config) replicated to ALL shards.
  JOIN with broadcast table = local operation, no cross-shard.

  5. APPLICATION-LEVEL JOIN:
  Step 1: Query shard A for user data
  Step 2: Query shard B for order data
  Step 3: Join in application code
  Flexible but pushes complexity to the application.

Cross-Shard Transactions: The Worst Case

Two-Phase Commit (2PC) for Cross-Shard Transactions:

  Coordinator: 'Transfer $100 from User A (Shard 1) to User B (Shard 2)'

  Phase 1 - PREPARE:
  Coordinator --> Shard 1: 'Can you debit $100?'  --> 'YES' (locks row)
  Coordinator --> Shard 2: 'Can you credit $100?'  --> 'YES' (locks row)

  Phase 2 - COMMIT:
  Coordinator --> Shard 1: 'COMMIT'  --> done (releases lock)
  Coordinator --> Shard 2: 'COMMIT'  --> done (releases lock)

  Problems:
  - Blocking: if coordinator dies between phases, shards hold locks forever
  - Latency: 2 network round-trips minimum
  - Availability: fails if ANY shard is unreachable
  - Performance: locks held across network round-trips

  Better alternatives:
  1. Saga pattern: compensating transactions instead of distributed locks
  2. Design shard key to co-locate related data (avoid the problem)
  3. Eventual consistency with idempotent operations

Interview Tip

When designing a sharded system, always address cross-shard queries proactively. The interviewer wants to see you acknowledge the trade-off: your shard key optimizes for the 90% case, and cross-shard queries are the cost. Present your strategy: (1) choose shard key to minimize cross-shard queries for the most critical paths, (2) use Elasticsearch as a secondary index for search/filter queries, (3) use materialized views for analytics/aggregations, (4) use broadcast tables for small reference data. If asked about cross-shard transactions, explain why 2PC is problematic and propose Sagas as the alternative. The key insight: if you need frequent cross-shard queries, your shard key might be wrong.

<

/blockquote>

Key Takeaway

Cross-shard queries are expensive scatter-gather operations bounded by your slowest shard. Minimize them through proper shard key selection, denormalization, secondary indexes (Elasticsearch), and materialized views. Avoid cross-shard transactions entirely - use Sagas instead. Design your sharding strategy around your most critical queries, and accept that rare cross-shard queries will be slower.