Distributed Query Planning

In this lesson, you will learn how a query coordinator decomposes a single high-level SQL request into localized sub-plans for parallel execution across distributed shards.

The Decomposition Workflow

When you submit a SQL query to a distributed system, the coordinator doesn't just broadcast the raw text to every node. Instead, it performs Distributed Query Planning. This involves three primary stages:

1. Normalization & Rewriting: The query is parsed and rewritten based on the physical distribution of data (e.g., using shard keys to prune irrelevant nodes).
2. Operator Pushdown: The planner identifies which parts of the query—such as WHERE filters, LIMIT clauses, and partial SUM aggregations—can be pushed down to the shard level to minimize network I/O.
3. Plan Fragmentation: The final plan is fragmented into sub-plans. Each fragment is sent to a specific shard, while the coordinator retains a "Global Execution Plan" to merge the incoming partial results.

Practical Example

Imagine you have a Sales table sharded by region_id. You run: SELECT region_id, SUM(amount) FROM Sales WHERE year = 2023 GROUP BY region_id;

The coordinator realizes that because the table is partitioned by region_id, it can delegate the heavy lifting to the shards:

Sub-plan for Shard N:

-- Pushed down to each individual shard
SELECT region_id, SUM(amount) AS partial_sum 
FROM Sales_Shard_N 
WHERE year = 2023 
GROUP BY region_id;

Global Plan (Coordinator): The coordinator receives the partial_sum results from all shards and performs a final SUM(partial_sum) to produce the aggregate total. By pushing the WHERE clause and the initial GROUP BY logic into the shards, we avoid transferring raw rows over the network, drastically reducing latency.

The Role of Metadata

This planning is only possible because the coordinator maintains a Scalable Metadata Store. It knows exactly which shard holds which range of region_id values. Without accurate metadata, the coordinator would be forced to perform a "scatter-gather" approach, flooding every shard with queries even if the data isn't present, which degrades performance across the entire cluster.

Key Takeaway: Distributed query planning transforms global intent into local action by pushing operators down to the data, ensuring that only necessary, reduced result sets traverse the network.