Cross-shard queries in HLD - System Design Exercise

Design: Cross-shard Query System

Design focuses on the architecture and flow for executing cross-shard queries efficiently and reliably. Does not cover shard key design or shard rebalancing.

Functional Requirements

FR1: Support queries that need to fetch and combine data from multiple database shards.

FR2: Ensure query results are consistent and accurate across shards.

FR3: Minimize query latency to keep response times under 500ms for 95% of queries.

FR4: Handle up to 10,000 concurrent cross-shard queries.

FR5: Support both read-only and read-write queries spanning multiple shards.

Non-Functional Requirements

NFR1: Data is horizontally partitioned (sharded) across multiple database instances.

NFR2: Each shard holds a subset of the total data with no overlap.

NFR3: System availability target is 99.9% uptime.

NFR4: Network latency between shards can vary up to 50ms.

NFR5: Queries must not overload any single shard or network link.

Think Before You Design

Questions to Ask

❓ Question 1

❓ Question 2

❓ Question 3

❓ Question 4

❓ Question 5

Key Components

Query router or coordinator to parse and distribute queries

Shard databases holding partitioned data

Result aggregator to combine partial results

Caching layer to reduce repeated cross-shard queries

Network communication layer between components

Design Patterns

Scatter-gather query pattern

Two-phase commit for distributed transactions

Query rewriting and pushdown to shards

Result merging and sorting

Caching and partial result reuse

Reference Architecture

Client
  |
  v
Query Router/Coordinator
  |
  |---> Shard 1 DB
  |---> Shard 2 DB
  |---> Shard N DB
  |
  v
Result Aggregator
  |
  v
Client Response

Components

Query Router/Coordinator

Custom service or middleware

Receives client queries, parses them, determines which shards to query, and sends sub-queries.

Shard Databases

Relational or NoSQL databases (e.g., PostgreSQL, MongoDB)

Store partitioned data subsets; execute sub-queries locally.

Result Aggregator

Service or component within Query Router

Collects partial results from shards, merges, sorts, and formats final response.

Cache Layer

In-memory cache (e.g., Redis, Memcached)

Store frequently requested query results or partial results to reduce load and latency.

Network Communication Layer

RPC or REST APIs

Facilitates communication between Query Router and shards.

Request Flow

1. Client sends a cross-shard query to the Query Router.

2. Query Router parses the query and identifies which shards hold relevant data.

3. Query Router rewrites the query into sub-queries targeted to each shard.

4. Sub-queries are sent in parallel to the respective shard databases.

5. Each shard executes its sub-query and returns partial results.

6. Result Aggregator collects all partial results.

7. Aggregator merges, sorts, and processes partial results into a single response.

8. Final combined result is sent back to the client.

9. Optionally, the result is cached for future similar queries.

Database Schema

Data is horizontally partitioned by shard key. Each shard contains the same schema but only a subset of rows. No cross-shard foreign keys exist. Entities include: - Records (sharded by key) - Metadata for shard mapping Relationships are 1:N within shards; cross-shard joins are handled at query layer.

Scaling Discussion

Bottlenecks

Query Router becomes a bottleneck handling all query parsing and coordination.

Network latency and bandwidth between Query Router and shards.

Result Aggregator delays when merging large partial results.

Individual shards overloaded by heavy sub-query load.

Cache invalidation complexity for cross-shard data.

Solutions

Scale Query Router horizontally with load balancers and stateless design.

Use efficient binary protocols and compression to reduce network overhead.

Parallelize result aggregation and stream partial results to client.

Shard data evenly and monitor shard load to rebalance if needed.

Implement fine-grained cache invalidation and TTL policies.

Interview Tips

Time: Spend 10 minutes understanding requirements and clarifying assumptions, 20 minutes designing architecture and data flow, 10 minutes discussing scaling and trade-offs, 5 minutes summarizing.

Explain how query routing and rewriting works for cross-shard queries.

Discuss trade-offs between consistency, latency, and complexity.

Highlight importance of parallelism and result aggregation.

Mention caching strategies and their challenges.

Address scaling bottlenecks and mitigation techniques.

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