Cross-shard queries in HLD - System Design Guide

Problem Statement

When data is split across multiple shards, running queries that need to access multiple shards at once can cause slow response times and inconsistent results. Without a proper approach, the system may have to query all shards sequentially or in parallel without coordination, leading to high latency and complex result merging.

Solution

Cross-shard queries coordinate querying multiple shards in parallel and then merge the results to present a unified response. This involves a query router or coordinator that breaks down the query, sends sub-queries to relevant shards, collects partial results, and combines them before returning to the client. This approach balances load and maintains consistency across distributed data.

Architecture

Client App

Client App

↓

Query Router

Query Router

↓

Shard 1

↓

Result Merger & Sort

Result Merger & Sort

↓

Unified Response

Unified Response

This diagram shows a client sending a query to a query router, which splits the query to multiple shards. Each shard processes its part and returns results to a merger component that combines and sorts them before sending a unified response back to the client.

Trade-offs

✓ Pros

→

Enables scalable querying across distributed data by parallelizing requests to shards.

→

Improves response time compared to sequential shard querying.

→

Maintains data consistency by coordinating query execution and result merging.

→

Supports complex queries that span multiple shards transparently to the client.

✗ Cons

→

Increases system complexity due to query routing, coordination, and result merging logic.

→

Merging large result sets can cause memory and CPU overhead on the coordinator.

→

Latency depends on the slowest shard response, potentially causing bottlenecks.

Use when your dataset is sharded for scale and you need to support queries that span multiple shards, especially when read traffic exceeds thousands of requests per second or data size exceeds single-node capacity.

Avoid if your queries mostly target single shards or if your system handles low traffic (under 1,000 queries per second), where the added complexity and overhead of cross-shard coordination outweigh benefits.

Real World Examples

Google

Google Bigtable uses cross-shard queries to scan data distributed across tablets, merging results to provide consistent views for large-scale analytics.

Twitter

Twitter shards user timelines and uses cross-shard queries to aggregate tweets from multiple shards when users follow many accounts.

Amazon

Amazon DynamoDB supports cross-shard queries by routing requests to multiple partitions and merging results for global secondary indexes.

Alternatives

Single-shard queries

Queries are restricted to a single shard, avoiding cross-shard coordination.

Use when: When queries can be designed to target one shard only, reducing complexity and latency.

Denormalization and pre-aggregation

Data is duplicated or aggregated to avoid cross-shard queries by storing combined views.

Use when: When read performance is critical and data update frequency is low enough to allow denormalization.

Distributed SQL engines

Use specialized distributed query engines that handle cross-shard queries with optimized execution plans.

Use when: When complex SQL queries across shards are frequent and require advanced optimization.

Summary

Cross-shard queries solve the problem of querying data distributed across multiple shards by coordinating parallel queries and merging results.

This pattern improves scalability and supports complex queries but adds system complexity and potential latency bottlenecks.

It is best used when data size and query patterns require distributed access beyond single shards.

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