19.7 Distributed Query Processing

Distributed query processing deals with optimizing queries that access data from multiple sites in a distributed database. Unlike centralized systems, which primarily focus on minimizing disk accesses, distributed systems must also consider:

• Network transmission costs: The cost of sending data between sites.
• Parallel processing benefits: The performance gains from executing parts of a query concurrently at different sites.

The relative costs of network transmission and disk access vary, so a good trade-off is essential.

19.7.1 Query Transformation

Goal: Convert a user’s query into an equivalent form that can be executed efficiently across distributed sites. The presence of data fragmentation and replication significantly complicates this.

Example Scenario

Consider the query “Find all tuples in the account relation.”

1. Replication: If account is replicated, choose the replica with the lowest transmission cost.

2. Fragmentation: If account is fragmented, the optimization is more complex because multiple joins or unions might be needed to reconstruct the relation.

Fragmentation Transparency Example

The user should be able to use the simple query:

SELECT *
FROM account
WHERE branch_name = "Hillside";

If, the account relation is horizontally fragmented, the query optimizer needs to make use of that. For Example: account is defined as: $accoun{t}_1\cup accoun{t}_2$.

The system transforms this into:

${\sigma }_{branch\_name=\text{"Hillside"}}(accoun{t}_1\cup accoun{t}_2)$

Using the query-optimization techniques: ${\sigma }_{branch\_name=\text{"Hillside"}}(accoun{t}_1)\cup {\sigma }_{branch\_name=\text{"Hillside"}}(accoun{t}_2)$.

Since $account1$ contains tuples for “Hillside” and $account2$ for “Valleyview”, the second part of the union is empty.

Final Optimized Strategy: Simply return $accoun{t}_1$.

19.7.2 Simple Join Processing

Goal: Efficiently compute joins across relations stored at different sites.

Example: Three-Relation Join

Consider: $account\bowtie depositor\bowtie branch$

Assume:

• account is at site $S_1$.
• depositor is at site $S_2$.
• branch is at $S_3$.
• The result is needed at $S_1$.

Possible Strategies

1. Ship all to $S_1$: Send copies of all three relations to $S_1$ and perform the join locally. This might require recreating indexes if they exist at other sites, which is expensive.

2. Ship and Join Incrementally:

   • Ship account to $S_2$.
   • Compute $tem{p}_1=account\bowtie depositor$ at $S_2$.
   • Ship $tem{p}_1$ to $S_3$.
   • Compute $tem{p}_2=tem{p}_1\bowtie branch$ at $S_3$.
   • Ship $tem{p}_2$ to $S_1$.

3. Other permutations of the roles of S1, S2, and S3 in the above strategies.

Choosing the Best Strategy

There’s no single best strategy. Factors to consider include:

• Data volumes: How much data needs to be shipped between sites?
• Transmission costs: Cost per block transferred between sites.
• Processing speeds: Relative processing power of each site.
• Presence of Indexes: If present, can improve the computation, if not, will require creation of indexes.

19.7.3 Semijoin Strategy

Goal: Reduce data transmission costs by eliminating tuples that don’t contribute to the final join result before shipping.

The Semijoin Operator ($⋉$)

The semijoin of $r_1$ with $r_2$, denoted as $r_1⋉r_2$, is defined as:

$r_1⋉r_2={\Pi }_{R_1}(r_1\bowtie r_2)$

Where:

• $R_1$ is the schema (set of attributes) of $r_1$.
• The semijoin selects the tuples in $r_1$ that participate in the join with $r_2$.

Semijoin Strategy Example ($r_1\bowtie r_2$)

$r_1$ is stored at site $S_1$, and $r_2$ is stored at site $S_2$. The result should be at $S_1$.

4. Compute temp1: $tem{p}_1\leftarrow {\Pi }_{R_1\cap R_2}(r_1)$ at $S_1$. (Project the common attributes of $r_1$).
5. Ship temp1: Send $tem{p}_1$ from $S_1$ to $S_2$.
6. Compute temp2: $tem{p}_2\leftarrow r_2⋉tem{p}_1$ at $S_2$. (Semijoin of $r_2$ with $tem{p}_1$).
7. Ship temp2: Send $tem{p}_2$ from $S_2$ to $S_1$.
8. Compute Final Result: $r_1\bowtie tem{p}_2$ at $S_1$.

Correctness

The final result ($r_1\bowtie tem{p}_2$) is equivalent to $r_1\bowtie r_2$ because join is associative and commutative, and ${\Pi }_{R_1\cap R_2}(r_1)\bowtie r_1=r_1.$

Efficiency

The strategy to find $r_1\bowtie r_2$ is good if a small fraction of tuples in $r_2$ contribute to the join. In this case, $tem{p}_2$ is much smaller than $r_2$, reducing network transmission costs. The cost of transmitting $tem{p}_1$ is usually outweighed by these savings. The saving is maximum when $r_1$ is the result of the selection.

19.7.4 Join Strategies that Exploit Parallelism

Goal: Perform join operations concurrently at multiple sites to reduce overall query execution time.

Example: Four-Relation Join

Consider: $r_1\bowtie r_2\bowtie r_3\bowtie r_4$

• $r_i$ is stored at site $S_i$.
• Result is needed at $S_1$.

Parallel Join Strategy

1. Parallel Joins:
   • Ship $r_1$ to $S_2$ and compute $r_1\bowtie r_2$ at $S_2$.
   • Simultaneously, ship $r_3$ to $S_4$ and compute $r_3\bowtie r_4$ at $S_4$.
2. Pipelining:
   • $S_2$ sends tuples of $(r_1\bowtie r_2)$ to $S_1$ as they are produced (pipelining).
   • $S_4$ sends tuples of $(r_3\bowtie r_4)$ to $S_1$ as they are produced.
3. Final Join: $S_1$ starts computing $(r_1\bowtie r_2)\bowtie (r_3\bowtie r_4)$ as soon as it receives tuples from $S_2$ and $S_4$.

Advantages

• Concurrency: Multiple joins are computed in parallel.
• Reduced Latency: The final join at $S_1$ can begin before the intermediate joins at $S_2$ and $S_4$ are fully completed, thanks to pipelining.