Cross-shard queries in HLD - Scalability & System Analysis

Scalability Analysis - Cross-shard queries

Growth Table: Cross-shard Queries Scaling

Users / Data Size	100 Users	10K Users	1M Users	100M Users
Number of shards	1-2 shards	10-20 shards	100-200 shards	1000+ shards
Cross-shard query frequency	Rare, simple queries	Occasional, moderate complexity	Frequent, complex queries	Very frequent, highly complex queries
Latency impact	Negligible	Noticeable delays	Significant latency increase	High latency, possible timeouts
Data consistency challenges	Minimal	Moderate	High	Very high, complex coordination
Network overhead	Low	Moderate	High	Very high
System complexity	Low	Moderate	High	Very high

First Bottleneck: Query Coordination and Latency

As the number of shards grows, the first bottleneck is the query coordinator that must gather and merge data from multiple shards. This causes increased latency and network overhead. The coordinator's CPU and memory can become overwhelmed handling many parallel cross-shard queries. Also, data consistency and transaction management across shards become challenging, slowing down responses.

Scaling Solutions for Cross-shard Queries

• Query Routing: Design queries to target as few shards as possible by improving shard keys and data locality.
• Denormalization & Aggregation: Pre-aggregate or duplicate data to reduce cross-shard joins.
• Asynchronous Processing: Use background jobs or eventual consistency to avoid blocking user queries.
• Distributed Query Engines: Use specialized systems that parallelize and optimize cross-shard queries efficiently.
• Caching: Cache frequent cross-shard query results to reduce load.
• Horizontal Scaling: Add more query coordinators and shard servers to distribute load.
• Sharding Strategy: Revisit shard keys to minimize cross-shard queries by grouping related data.

Back-of-Envelope Cost Analysis

Assuming 1M users with 100 shards:

• Average queries per second (QPS): 10,000
• Cross-shard queries: 20% -> 2,000 QPS involving multiple shards
• Each cross-shard query touches ~5 shards -> 10,000 shard queries/sec
• Network bandwidth: 10,000 shard queries x 10 KB data = ~100 MB/s
• Coordinator CPU: Must handle merging 2,000 complex queries/sec, needs multiple servers
• Storage: Each shard stores ~10,000 users' data, scaling storage horizontally

Interview Tip: Structuring Scalability Discussion

Start by explaining how sharding improves write/read scale but introduces cross-shard query challenges. Identify the coordinator as the bottleneck. Discuss trade-offs between latency, consistency, and complexity. Propose solutions like better shard keys, caching, and distributed query engines. Use numbers to justify your approach and show understanding of system limits.

Self-Check Question

Your database handles 1000 QPS. Traffic grows 10x with many cross-shard queries. What do you do first?

Answer: The first step is to reduce cross-shard query load by improving shard keys to localize queries or add caching to avoid repeated cross-shard fetches. Then, scale query coordinators horizontally to handle increased query merging load.

Key Result

Cross-shard queries become a bottleneck as shards increase, mainly due to query coordination overhead and latency. Solutions focus on minimizing cross-shard queries, caching, and horizontally scaling coordinators.

Practice

1. What is the main purpose of cross-shard queries in a distributed database system?

easy

A. To encrypt data within a shard

B. To backup data from one shard to another

C. To split a large shard into smaller shards

D. To retrieve data from multiple shards to answer a single complex request

Solution

1. Step 1: Understand what shards are

   Shards are parts of a database split to store data separately for scalability.

2. Step 2: Identify the role of cross-shard queries

   Cross-shard queries combine data from multiple shards to answer complex requests that need data from more than one shard.

3. Final Answer:

   To retrieve data from multiple shards to answer a single complex request -> Option D

4. Quick Check:

   Cross-shard queries = combine data from shards [OK]

Hint: Cross-shard queries combine data from many shards [OK]

Common Mistakes:

• Confusing cross-shard queries with data backup
• Thinking cross-shard queries split shards
• Assuming cross-shard queries encrypt data

2. Which component is responsible for splitting and merging queries in cross-shard query processing?

easy

A. Coordinator

B. Shard manager

C. Cache server

D. Load balancer

Solution

1. Step 1: Identify query processing roles

   The coordinator handles query distribution and result merging across shards.

2. Step 2: Match roles to components

   Shard manager manages shards but does not split/merge queries; cache server stores data temporarily; load balancer distributes traffic but not queries.

3. Final Answer:

   Coordinator -> Option A

4. Quick Check:

   Coordinator splits and merges queries [OK]

Hint: Coordinator splits and merges cross-shard queries [OK]

Common Mistakes:

• Confusing shard manager with coordinator
• Thinking cache server merges queries
• Assuming load balancer handles query splitting

3. Consider a system with 3 shards. A cross-shard query is sent to all shards in parallel. If each shard takes 100ms, 120ms, and 80ms respectively to respond, what is the total time to get the combined result?

medium

A. 300ms

B. 80ms

C. 120ms

D. 100ms

Solution

1. Step 1: Understand parallel query execution

   Queries run in parallel, so total time is the longest shard response time.

2. Step 2: Identify the longest response time

   Shard times are 100ms, 120ms, and 80ms. The longest is 120ms.

3. Final Answer:

   120ms -> Option C

4. Quick Check:

   Parallel time = max shard time = 120ms [OK]

Hint: Total time = longest shard response in parallel [OK]

Common Mistakes:

• Adding all shard times instead of taking max
• Choosing shortest shard time
• Confusing sequential with parallel execution

4. A cross-shard query coordinator is returning incomplete results. Which of the following is the most likely cause?

medium

A. Coordinator failed to merge all shard responses

B. Shard keys are not evenly distributed

C. Cache server is overloaded

D. Load balancer is routing traffic incorrectly

Solution

1. Step 1: Analyze incomplete results cause

   Incomplete results usually mean some shard data was not included in the final output.

2. Step 2: Identify coordinator role in merging

   The coordinator merges shard responses; failure here causes missing data.

3. Final Answer:

   Coordinator failed to merge all shard responses -> Option A

4. Quick Check:

   Incomplete results = merge failure [OK]

Hint: Incomplete results often mean merge failure by coordinator [OK]

Common Mistakes:

• Blaming shard key distribution for missing data
• Assuming cache overload causes missing data
• Thinking load balancer affects query merging

5. You want to reduce latency of cross-shard queries in a large distributed system. Which combination of strategies is best?

hard

A. Run queries sequentially on each shard and use a load balancer to distribute traffic

B. Use a coordinator to parallelize queries and implement caching of frequent results

C. Increase shard size and avoid caching to reduce complexity

D. Remove the coordinator and let clients query shards directly

Solution

1. Step 1: Identify strategies to reduce latency

   Parallel query execution via coordinator reduces wait time; caching avoids repeated work.

2. Step 2: Evaluate options

   Use a coordinator to parallelize queries and implement caching of frequent results uses both parallelism and caching, best for latency. Others increase complexity or remove coordination, causing delays.

3. Final Answer:

   Use a coordinator to parallelize queries and implement caching of frequent results -> Option B

4. Quick Check:

   Parallel + caching = lower latency [OK]

Hint: Parallel queries plus caching reduce latency best [OK]

Common Mistakes:

• Thinking bigger shards reduce latency
• Running queries sequentially increases latency
• Removing coordinator causes uncoordinated delays