Cross-Shard Queries

Sharding works beautifully when queries hit a single shard. But what happens when you need data from multiple shards? Welcome to the hardest part of sharding.

The Problem: Scatter-Gather

When a query doesn't include the shard key, you must query all shards:

# Query WITH shard key - goes to one shard
SELECT * FROM orders WHERE user_id = 12345;
# Router: user_id 12345 is on Shard 2. Query Shard 2.

# Query WITHOUT shard key - goes to ALL shards
SELECT * FROM orders WHERE total > 1000;
# Router: I don't know which shards have orders > $1000. Query ALL shards.

Scatter-Gather Flow

                    ┌─────────────────┐
                    │     Client      │
                    │ SELECT * WHERE  │
                    │  status='paid'  │
                    └────────┬────────┘
                             │
                    ┌────────▼────────┐
                    │  Query Router   │
                    │ (Scatter query) │
                    └────────┬────────┘
                             │
        ┌────────────────────┼────────────────────┐
        │                    │                    │
        ▼                    ▼                    ▼
   ┌─────────┐          ┌─────────┐          ┌─────────┐
   │ Shard 1 │          │ Shard 2 │          │ Shard 3 │
   │ Query   │          │ Query   │          │ Query   │
   └────┬────┘          └────┬────┘          └────┬────┘
        │                    │                    │
        └────────────────────┼────────────────────┘
                             │
                    ┌────────▼────────┐
                    │  Query Router   │
                    │ (Gather results)│
                    └────────┬────────┘
                             │
                    ┌────────▼────────┐
                    │     Client      │
                    │  Final Results  │
                    └─────────────────┘

Performance Impact

Challenge 1: JOINs Across Shards

In a non-sharded database, JOINs are efficient - all data is local. In a sharded database, JOINed tables may be on different servers.

-- This was fast on a single database:
SELECT users.name, orders.total
FROM users
JOIN orders ON users.id = orders.user_id
WHERE orders.status = 'completed';

-- In a sharded database:
-- users might be on Shard 1
-- orders might be on Shard 2
-- Now you need network round-trips!

Solutions for JOINs

Challenge 2: Aggregations

How do you compute COUNT(*), SUM(), or AVG() across all shards?

-- This query must hit all shards:
SELECT COUNT(*) FROM orders WHERE status = 'completed';
SELECT AVG(total) FROM orders WHERE created_at > '2024-01-01';

Aggregation Strategies

def count_completed_orders():
    # Map: Query each shard
    shard_counts = []
    for shard in all_shards:
        count = shard.query("SELECT COUNT(*) FROM orders WHERE status='completed'")
        shard_counts.append(count)

    # Reduce: Combine results
    total_count = sum(shard_counts)
    return total_count

def average_order_total():
    # Map: Get sum and count from each shard
    shard_results = []
    for shard in all_shards:
        result = shard.query("SELECT SUM(total), COUNT(*) FROM orders")
        shard_results.append(result)

    # Reduce: Calculate global average
    total_sum = sum(r.sum for r in shard_results)
    total_count = sum(r.count for r in shard_results)
    return total_sum / total_count  # Can't just average the averages!

Pre-Computed Aggregations

For frequently-needed aggregations, maintain summary tables:

# Real-time aggregation table (updated on each order)
order_stats = {
    "total_orders": 1_500_000,
    "total_revenue": 45_000_000,
    "orders_by_status": {
        "completed": 1_200_000,
        "pending": 200_000,
        "cancelled": 100_000
    }
}

# Update on each order change
def on_order_completed(order):
    stats.increment("total_orders")
    stats.increment("total_revenue", order.total)
    stats.increment("orders_by_status.completed")

Challenge 3: Global Sorting and Pagination

How do you get "top 10 orders by total" across all shards?

SELECT * FROM orders ORDER BY total DESC LIMIT 10;

The N + K Problem

You can't just get top 10 from each shard. What if all top 10 globally are on one shard?

Naive approach (wrong):
Shard 1 top 10: $100, $95, $90...
Shard 2 top 10: $500, $450, $400...  <- All global top 10 are here!

Correct approach:
Get top K from each shard, then sort globally.
For top 10 with 100 shards: fetch 10 from each = 1000 rows to sort.

def get_top_orders(limit=10):
    # Fetch top K from each shard
    all_candidates = []
    for shard in all_shards:
        top_from_shard = shard.query(
            f"SELECT * FROM orders ORDER BY total DESC LIMIT {limit}"
        )
        all_candidates.extend(top_from_shard)

    # Sort all candidates and take top K
    all_candidates.sort(key=lambda x: x.total, reverse=True)
    return all_candidates[:limit]

Deep Pagination Problem

Pagination gets worse as you go deeper:

Page 1 (OFFSET 0, LIMIT 10): Fetch 10 from each shard, merge, return 10
Page 100 (OFFSET 990, LIMIT 10): Fetch 1000 from each shard, merge, skip 990, return 10

With 100 shards, page 100 requires fetching 100,000 rows!

• Use cursor-based pagination instead of offset
• Limit maximum page depth
• Use search engines (Elasticsearch) for deep pagination needs

Design Patterns for Minimizing Cross-Shard Queries

Pattern 1: Entity Groups

Group related entities that are always accessed together:

# E-commerce: User entity group
# All sharded by user_id - always on same shard
- users
- user_addresses
- user_payment_methods
- orders (for this user)
- order_items (for this user's orders)

# Query within entity group = single shard
# Query across users = scatter-gather

Pattern 2: Reference Tables

Keep small, frequently-JOINed tables replicated on all shards:

# Replicated on every shard (not sharded):
- countries (200 rows)
- product_categories (50 rows)
- tax_rates (100 rows)

# Now JOINs with these tables are always local!
SELECT o.*, c.name as country_name
FROM orders o
JOIN countries c ON o.country_code = c.code
# countries is local to every shard - fast JOIN!

Pattern 3: CQRS (Command Query Responsibility Segregation)

Separate write and read models:

Real-World Example: Slack

Slack shards by workspace (team). All data for a workspace is co-located:

Key Takeaways

Next up: Sharding Assessment - Test your understanding with real-world scenarios.