Distributed Query Processing and Optimization

Introduction

A distributed database stores related data across multiple networked sites, yet users expect queries to behave as if the data were held in one logical database. Distributed query processing is the set of techniques used to transform a global query into a sequence of local operations, data transfers, and combinations of intermediate results. Distributed query optimization selects an execution strategy that minimizes cost while satisfying correctness, consistency, and performance requirements.

The central difficulty is that a distributed query has two major resource dimensions: computation and communication. In a centralized database, reading a relation from disk may dominate cost. In a distributed database, transferring tuples across a network can be substantially more expensive than local computation. Consequently, a plan that is efficient at one site may be inefficient globally.

This chapter develops query decomposition, data localization, distributed joins and semijoins, cost models, statistics, communication-aware optimization, parallel execution, and adaptive optimization.

Learning Objectives

By the end of this chapter, students should be able to:

1. Decompose a global relational query into relational-algebra operations.
2. Localize queries over fragmented and replicated data.
3. Compare distributed join strategies, including semijoin methods.
4. Construct basic communication and execution-cost models.
5. Explain how statistics influence cardinality and cost estimation.
6. Design parallel and pipelined execution strategies.
7. Evaluate adaptive optimization under changing data and network conditions.
8. Assess practical, governance, and reliability implications of distributed query processing.

1. The Distributed Query-Processing Pipeline

A distributed query processor generally performs the following stages:

1. Parsing and validation: The SQL statement is converted into a logical representation and checked against the schema and authorization rules.
2. Query decomposition: Predicates, projections, joins, grouping, and aggregation are represented as a relational-algebra expression.
3. Data localization: References to global relations are replaced by their fragments and, where necessary, replicas.
4. Global optimization: Alternative sites, join orders, data-transfer strategies, and parallel schedules are compared.
5. Local optimization: Each site selects access paths, indexes, join algorithms, and local operator implementations.
6. Execution and coordination: Operators are scheduled, intermediate data is exchanged, and failures or changing conditions are handled.

A logical query describes what must be computed; a physical plan specifies how and where it will be computed.

Important terminology

• Query decomposition: Transformation of a global query into relational-algebra operations.
• Data localization: Substitution of global relations with their physical fragments or replicas.
• Distributed execution plan: A site-aware physical plan including data movement.
• Shipping: Moving data between sites.
• Coordinator: A component that orchestrates distributed execution.
• Exchange operator: An operator that repartitions, broadcasts, gathers, or redistributes tuples.
• Fragment: A logically defined portion of a relation.
• Replica: An additional copy of data stored at another site.

2. Query Decomposition

A relational query should first be simplified before distribution decisions are made. Equivalent transformations can reduce intermediate results and therefore communication.

Consider:

SELECT C.customer_id, O.order_id
FROM Customer C JOIN Orders O
  ON C.customer_id = O.customer_id
WHERE C.region = 'West' AND O.total > 1000;

The corresponding expression is:

A better form pushes selections toward the base relations:

Selection pushdown is valid when a predicate references only attributes of the relation below it. Projection pushdown is valid when all attributes needed by later predicates, joins, grouping, ordering, and final output are retained.

Equivalence transformations

Common transformations include:

• Selection cascade: applying σ_p∧q(R) is equivalent to applying σ_p(σ_q(R)).
• Selection commutativity: σ_p(σ_q(R)) and σ_q(σ_p(R)) produce the same result.
• Join commutativity: R ⋈ S is equivalent to S ⋈ R for an inner join.
• Join associativity: (R ⋈ S) ⋈ T is equivalent to R ⋈ (S ⋈ T) for compatible inner joins.
• Predicate transfer: a predicate on a join key may sometimes be transferred to the other relation when the join semantics guarantee correctness.

Outer joins, null values, duplicate preservation, aggregation, and nondeterministic functions restrict these transformations. An optimizer must therefore use semantic rules, not merely syntactic rearrangement.

3. Data Localization and Fragmentation

A global relation may be fragmented in three principal ways.

Horizontal fragmentation

Horizontal fragmentation divides tuples using predicates. For example:

The fragments should normally be disjoint and collectively complete. If the predicates are mutually exclusive and cover every tuple, reconstruction uses union without duplicate concerns. If fragments overlap, duplicate handling must be specified.

Vertical fragmentation

Vertical fragmentation divides attributes. A key is usually replicated in every vertical fragment so that the original relation can be reconstructed by joining fragments:

The join is correct only when the common identifier uniquely identifies the original tuple and the fragments preserve the necessary rows.

Hybrid fragmentation

Hybrid fragmentation combines horizontal and vertical fragmentation. A relation can first be divided by region and then by attribute groups, or vice versa.

Localization process

Suppose Orders is horizontally fragmented:

A query over Orders is localized by replacing Orders with the union of fragments, then pushing predicates into each fragment. If only West orders can satisfy a predicate, other fragments can be eliminated through fragment pruning.

Replication adds a choice: the optimizer may read the nearest or least-loaded replica, but it must account for freshness, consistency, and failover policy. A stale analytical replica may be acceptable for a dashboard but unacceptable for regulatory reporting.

4. Distributed Join Processing

Joins are usually the dominant source of distributed cost. Let R and S be relations joined on attribute A. The optimizer must decide:

• where the join will execute;
• which relation, if any, will move;
• whether data will be broadcast or repartitioned;
• whether a semijoin can reduce transfer;
• whether the join can execute in parallel;
• how skew will be controlled.

Ship-whole-relation strategies

If R and S are located at different sites, one relation may be transferred to the site of the other:

• Ship R to S’s site and join there.
• Ship S to R’s site and join there.
• Ship both to a third site.

These strategies are simple, but transferring a large relation can be expensive.

Broadcast join

A small relation is copied to every site containing partitions of the large relation. If a small dimension table has size 50 MB and is sent to 20 workers, approximately 1 GB of payload is transmitted, ignoring protocol overhead. Broadcasting is attractive when the replicated relation is small and the large relation is already distributed.

Repartition or hash join

Both inputs are partitioned on the join key using the same partitioning function. Tuples with equal keys are sent to the same worker. Each worker then performs a local join. This supports parallelism but may transmit a large fraction of both relations.

For a partition function h and p workers, worker i receives:

The global join is obtained from the union of local joins:

This assumes the same deterministic partitioning function and compatible join semantics.

Fragment join

If corresponding fragments are co-located and partitioned on the join key, local joins may be performed without network transfer. Physical design can therefore be as important as query-time optimization.

5. Semijoin Optimization

A semijoin transfers only the attributes needed to identify matching tuples before transferring full tuples. For relations R and S joined on R.A = S.A:

A distributed semijoin may proceed as follows:

1. Project S onto the join attribute A.
2. Send the projected keys to the site containing R.
3. Filter R to keys appearing in S.
4. Transfer only the reduced R, or continue with a coordinated join.

The technique is beneficial when the join-key projection is small and eliminates many nonmatching tuples. It can be harmful when most tuples match or when an extra round trip dominates savings.

Worked example: semijoin decision

Assume:

• R contains 1,000,000 tuples of 200 bytes each.
• S contains 100,000 tuples.
• The distinct join-key projection of S contains 10,000 keys of 8 bytes each.
• The semijoin reduces R to 20,000 tuples.
• Ignore headers and protocol overhead.

Shipping R directly requires:

The semijoin sends 80,000 bytes of keys and then 20,000 reduced tuples:

Thus, under these assumptions, the semijoin greatly reduces payload. The conclusion changes if the key set is nearly as large as R or if the reduction is weak.

6. Cost Models for Distributed Optimization

A useful cost model separates local processing from data movement. A simplified elapsed-time model is:

where:

• T_cpu is computation time;
• T_io is local storage and memory-access time;
• T_net is network transmission and startup time;
• T_coord is coordination, synchronization, and scheduling overhead.

For a message of size B bytes, a common network approximation is:

where α is message startup latency in seconds and β is effective bandwidth in bytes per second. This model assumes a relatively stable link and does not fully represent congestion, retransmission, or queueing.

For k messages:

This explains why one large transfer can be preferable to many small transfers, even when total bytes are equal.

A cost optimizer may minimize total resource consumption or estimated elapsed time. These are not identical. A plan that uses more total CPU can still finish sooner through parallelism, whereas a plan with low total work can suffer from a serial network bottleneck.

Example: latency and bandwidth

Suppose a plan sends 100 messages, each 1 MB, over a link with latency α = 2 ms and bandwidth β = 100 MB/s. Using decimal MB for consistency:

If the same payload is sent as one message, the idealized estimate is:

The difference is caused by message startup overhead, not by payload volume.

7. Cardinality Estimation and Statistics

Cost decisions depend on estimates of relation size, selectivity, distinct values, and data distribution. Let N(R) denote the number of tuples in R and s the selectivity of a predicate, defined as the fraction of tuples satisfying it. Then:

This estimate assumes that the selectivity is known or approximated and that the predicate behavior is sufficiently stable.

For an equality join R.A = S.B, a classical estimate is:

where V(R,A) is the number of distinct values of A in R. This assumes approximate uniformity and independence. It can be seriously inaccurate for correlated, skewed, or highly selective data.

Statistics used by optimizers

• relation cardinality;
• page or byte size;
• number of distinct values;
• minimum and maximum values;
• histograms for nonuniform distributions;
• null fractions;
• frequency information for heavy hitters;
• correlation or multicolumn statistics;
• fragment-level and site-level statistics.

Outdated statistics can cause an optimizer to choose a broadcast plan for a relation that has grown too large, or to choose a nested-loop join when a hash join is now preferable. Statistics collection itself consumes storage and processing resources, so update frequency should reflect workload volatility.

8. Parallel and Pipelined Execution

Distributed systems exploit several forms of parallelism:

• Inter-operator parallelism: different operators execute concurrently.
• Intra-operator parallelism: one operator is divided among workers.
• Partition parallelism: different data partitions are processed independently.
• Pipeline parallelism: one operator produces tuples while the next consumes them.

Blocking operators, such as global sorting and some forms of aggregation, must receive substantial or all input before producing final output. Pipelining reduces temporary storage and latency but can propagate failures and backpressure.

Parallel aggregation

A distributed aggregation often uses two stages:

1. Local partial aggregation at each worker.
2. Global aggregation of partial results.

For additive measures such as SUM and COUNT, this is straightforward. For AVG, workers should transmit both sum and count, because:

A simple average of worker averages is incorrect when workers contain different numbers of tuples.

Data skew

Hash partitioning assumes a reasonably balanced key distribution. If one key occurs extremely frequently, one worker may receive most matching tuples, creating a straggler. Remedies include skew-aware partitioning, replicating small heavy-hitter groups, salting keys, or using adaptive repartitioning.

9. Adaptive and Runtime Optimization

Static optimization chooses a plan before execution. Adaptive optimization changes the plan when runtime observations differ materially from estimates. Examples include:

• switching from broadcast to repartitioning when a relation exceeds a threshold;
• changing join algorithms after observing cardinality;
• dynamically allocating workers to a slow partition;
• reordering remaining operations;
• mitigating skew after identifying heavy keys;
• selecting a fresher or less-loaded replica.

Adaptive changes require safe boundaries. A system may pause at a materialization point, preserve already produced results, and re-optimize the remaining subplan. Replanning too frequently can introduce instability and coordination overhead.

A robust optimizer should distinguish between estimation error and transient events. A slow node may indicate hardware or network contention rather than an incorrect logical plan.

10. Illustrative Case: Distributed Retail Analytics

Background

A retail organization stores customer data in Europe, North America, and Asia. Orders are horizontally fragmented by sales region. Product data is replicated to each regional site, while a central warehouse stores historical summaries.

Problem

Analysts request the total value of orders above 500 currency units for customers who purchased products in a selected category during the previous month. A naive plan transfers all regional orders to the central warehouse.

Analysis

The order relation is large, but the category predicate is selective. Product is relatively small and replicated. Customer attributes are needed only for a final regional grouping. Network links to the central warehouse are shared with operational traffic.

Approach

Each regional site performs the following:

1. Prunes order fragments to the requested month.
2. Joins local orders with the local product replica.
3. Pushes the category and value predicates below later operators.
4. Performs local partial aggregation by region and customer segment.
5. Sends only aggregate rows to the central warehouse.
6. The warehouse performs the final aggregation.

If a product replica is stale, the system either routes the query to a consistent replica or labels the result according to the organization’s analytical freshness policy.

Outcome

The plan avoids shipping detailed orders and uses parallel regional computation. Its accuracy depends on correct fragment predicates, synchronized product definitions, and correct aggregation of partial results. Runtime monitoring can detect a region with exceptional order volume and allocate additional workers.

Learning points

The best distributed plan is not necessarily the one with the fewest operators. It is the plan that reduces expensive intermediate data while respecting freshness, consistency, workload isolation, and failure requirements.

11. Practical Design and Governance Considerations

Correctness and consistency

Query optimization must preserve transaction isolation and consistency guarantees. Reading replicas can reduce latency but may return stale values. A distributed aggregation must also define whether data committed during execution is included.

Fault tolerance

Network partitions, worker failures, duplicate messages, and partial results require recovery protocols. Idempotent operators and checkpointed stages can reduce the cost of retry. However, retries may duplicate output unless operators or sinks provide exactly-once or deduplication semantics.

Security and privacy

Data movement can expose sensitive attributes. Projection pushdown is therefore a privacy control as well as a performance technique: do not transfer columns that are not required. Query plans should respect data-residency laws, access policies, encryption requirements, and audit obligations.

AI-assisted optimization

Machine-learning optimizers can learn cardinality models, select join strategies, or predict resource demand. They are useful when traditional statistics fail, but their predictions require monitoring, drift detection, explainability, and safe fallback plans. An opaque model should not silently violate data-governance constraints.

12. Best Practices

1. Push selective predicates and necessary projections toward data sources.
2. Estimate both bytes transferred and message count.
3. Maintain fragment-level, replica-level, and multicolumn statistics.
4. Prefer local joins when physical co-location makes them possible.
5. Broadcast only when the replicated input is reliably small.
6. Use semijoins when projected join keys are substantially smaller than full tuples.
7. Design for skew rather than assuming uniform distributions.
8. Separate logical correctness from physical optimization.
9. Instrument plans with actual cardinalities, bytes, latency, and task imbalance.
10. Use adaptive optimization with controlled replan points.
11. Treat freshness, privacy, and data residency as optimization constraints.
12. Test plans under realistic failures and changing workload conditions.

Challenges and Limitations

No cost model perfectly predicts a distributed environment. Network congestion, cache effects, concurrent workloads, skew, failures, and cloud pricing can invalidate static assumptions. Cardinality errors compound across long operator trees. Replication improves read locality but increases update complexity. Adaptive execution improves resilience but consumes coordination resources and can make performance less predictable. Consequently, production optimization requires continuous measurement rather than reliance on estimates alone.

Chapter Summary

Distributed query processing converts a global query into executable operations across sites. Query decomposition uses relational-algebra equivalences, while data localization maps global relations to fragments and replicas. Selection and projection pushdown, fragment pruning, and local aggregation reduce intermediate data.

Distributed joins require choices among shipping, broadcasting, repartitioning, co-location, and semijoin reduction. Cost models should include CPU, I/O, network latency, bandwidth, coordination, and parallelism. Cardinality estimates depend on selectivity, distinct-value counts, histograms, and assumptions about independence and uniformity. Parallel execution improves throughput but must address blocking operators, stragglers, and skew.

Adaptive optimization responds to runtime evidence, while practical systems must also preserve consistency, security, privacy, fault tolerance, and regulatory compliance. Effective optimization is therefore a combined problem of algebraic correctness, statistical estimation, systems engineering, and operational governance.

Key Terms

Query decomposition: Conversion of a global query into relational-algebra operations.

Data localization: Mapping global relations to physical fragments or replicas.

Horizontal fragmentation: Division of a relation into tuple subsets.

Vertical fragmentation: Division of a relation into attribute subsets.

Replica: An additional copy of stored data.

Fragment pruning: Excluding fragments that cannot satisfy a query predicate.

Semijoin: A join-like operation that returns tuples from one relation having matches in another.

Broadcast join: Sending one input to every worker processing the other input.

Repartition join: Redistributing both inputs by a common join-key partition function.

Cardinality: Number of tuples in a relation or intermediate result.

Selectivity: Fraction of tuples satisfying a predicate.

Distinct-value count: Number of different values of an attribute.

Cost model: Mathematical approximation of execution resource requirements.

Network startup latency: Fixed communication cost per message.

Bandwidth: Rate of data transmission, measured in bytes per second.

Data skew: Uneven distribution of tuples or keys across workers.

Pipelining: Passing produced tuples directly to a subsequent operator.

Adaptive optimization: Runtime modification of an execution plan.

Partial aggregation: Local aggregation performed before global aggregation.

Straggler: A slow worker that delays completion of a parallel stage.

Review Questions

Short Answer Questions

1. What is the difference between a logical query plan and a physical distributed execution plan?
2. Under what condition is selection pushdown valid?
3. Why can a semijoin reduce network traffic?
4. Define data skew and explain its effect on parallel joins.
5. What are the two components of the simplified network model α + B/β?

Long Answer Questions

1. Explain the complete distributed query-processing pipeline from parsing to execution.
2. Compare broadcast, ship-whole-relation, and repartition joins, including their advantages and limitations.
3. Explain horizontal, vertical, and hybrid fragmentation and the conditions needed for correct reconstruction.
4. Discuss cardinality estimation, including selectivity, distinct-value statistics, histograms, and independence assumptions.
5. Explain how consistency, privacy, fault tolerance, and data residency constrain query optimization.

Analytical / Application Questions

1. Relation R contains 500,000 tuples of 400 bytes. A semijoin sends 20,000 eight-byte keys and reduces R to 15,000 tuples. Calculate the payload for direct shipping and semijoin processing, ignoring overhead, and recommend a strategy.
2. Design a distributed plan for joining a large, regionally fragmented fact table with a small replicated dimension table. Justify predicate placement, join method, aggregation strategy, and skew controls.
3. A query optimizer estimated that a relation had 10,000 tuples, but runtime observation found 2,000,000 tuples. Analyze two likely consequences for the original plan and propose an adaptive response.

Multiple Choice Questions with Answers

Question 1

Which transformation most directly reduces the amount of data entering a distributed join?

A. Push a selective predicate below the join

B. Delay all predicates until after the join

C. Duplicate every input at every site

D. Replace a projection with an ordering operation

Correct Answer: A. Push a selective predicate below the join

Question 2

A semijoin is especially beneficial when:

A. Nearly every tuple matches and the key projection is large

B. The join-key projection is small and eliminates many tuples

C. No join predicate exists

D. The network has zero latency

Correct Answer: B. The join-key projection is small and eliminates many tuples

Question 3

In the approximation T_net ≈ α + B/β, what does β represent?

A. Number of participating sites

B. Message startup latency

C. Effective bandwidth

D. Number of distinct join keys

Correct Answer: C. Effective bandwidth

Question 4

For an equality join, the classical cardinality estimate divides by:

A. The sum of relation cardinalities

B. The maximum distinct-value count of the join attributes

C. The network bandwidth

D. The number of output columns

Correct Answer: B. The maximum distinct-value count of the join attributes

Question 5

Which join strategy redistributes both inputs using a common hash function on the join key?

A. Broadcast join

B. Nested local scan

C. Repartition hash join

D. Centralized projection

Correct Answer: C. Repartition hash join

Question 6

A relation contains 1,000,000 tuples of 100 bytes. What is its payload size, ignoring overhead?

A. 1 MB

B. 10 MB

C. 100 MB

D. 1 GB

Correct Answer: C. 100 MB

Question 7

Why can a simple average of worker averages be incorrect?

A. Averages cannot be computed in parallel

B. Workers may contain different numbers of tuples

C. SUM is not associative

D. Network bandwidth changes the mathematical mean

Correct Answer: B. Workers may contain different numbers of tuples

Question 8

Which problem occurs when one hash-partitioned worker receives most tuples for a frequent key?

A. Data skew and a straggler

B. Predicate equivalence

C. Vertical reconstruction

D. Replica freshness

Correct Answer: A. Data skew and a straggler

Question 9

Adaptive optimization changes a plan primarily in response to:

A. Runtime observations that differ from estimates

B. A fixed relational-algebra identity only

C. The number of columns in the schema alone

D. The use of primary keys

Correct Answer: A. Runtime observations that differ from estimates

Question 10

Which is a privacy benefit of projection pushdown?

A. It forces all attributes to be replicated

B. It transfers only required columns across sites

C. It removes the need for authorization

D. It guarantees encryption without configuration

Correct Answer: B. It transfers only required columns across sites