Overview on Query Processing
In a Distributed Database Management System (DDBMS), query processing is the
multi-step process of converting a high-level query (like SQL) into an efficient set of operations
executed across multiple networked sites. Unlike centralized systems, distributed query
processing must account for communication costs (data transfer over the network), which
often dominate the total execution time.
Query Processing Problem
In a distributed database, data is stored across multiple sites (locations).
The query processing problem is: How to execute a user query efficiently when data is
distributed across different sites?
Challenges:
Data is fragmented (split into parts)
Data may be replicated
Sites are connected by a network (slow compared to memory/CPU)
Example:
SELECT Name
FROM Employee E, Department D
WHERE [Link] = [Link] AND [Link] = 'Kabul';
Problem:
Employee table may be in Site A
Department table may be in Site B
System must decide:
Where to execute joins?
What data to transfer?
Objectives of Query Processing
To transform a high-level query into an efficient execution strategy that minimizes
resource consumption across multiple sites. While traditional centralized systems focus on CPU
and I/O costs, distributed systems prioritize minimizing communication costs due to the
high overhead of moving data over a network.
The main goal is: Minimize total cost
Cost includes:
1. Communication cost (MOST important)
2. Local processing cost (CPU, disk I/O)
3. Response time
Objectives:
Reduce data transfer between sites
Use parallelism
Choose best execution plan
Complexity of Relational Algebra Operations
The complexity of relational algebra operations refers to the computational resources
(time and space) required to execute them. In distributed systems, this complexity is further
amplified by the need for data movement between sites.
Different operations have different costs.
Key operations:
1. Selection (σ)
Low cost if indexed, Otherwise full scan
2. Projection (π)
Requires duplicate removal → expensive
3. Join (⨝) ❗ (Most expensive)
Requires combining tables
In distributed DB → data movement required
Join strategies:
Nested-loop join
Hash join
Merge join
In distributed systems: Join cost = local cost + communication cost
Assignment No. 1
About Big-O notation on relational algebra in DB.
Characterization of Query Processors
A query processor can be described based on several factors:
Languages
Query is written in SQL
Internally converted to Relational Algebra
Example:
SELECT * FROM Student WHERE Age > 20;
➡ Converted to:
σ Age > 20 (Student)
Types of Optimization
Optimizers generally use these three main methods to select a plan:
1. Heuristic (Rule-Based):
Uses a set of predefined rules and algebraic transformations (e.g., "perform selections and
projections as early as possible") to simplify the query. It is fast and simple but may miss the
truly optimal plan.
Mean:
Perform selection early
Reduce data size
2. Cost-based:
Evaluates several execution plans and assigns a "cost" to each based on CPU, I/O, and
communication overhead. It selects the plan with the lowest estimated cost.
Mean:
Uses statistics
Chooses lowest-cost plan
3. Adaptive (Dynamic) Optimization:
Mid-query, the system can stop and recalculate a plan if the actual intermediate results differ
significantly from the optimizer's predictions
Optimization Timing
Optimization can happen at different stages of the query lifecycle:
1. Static optimization: Performed at compile time. The optimizer uses database statistics
(like cardinality) to estimate sizes of intermediate results and generate a plan before the
query runs.
Mean done: Before execution
2. Dynamic optimization: Occurs during execution time. The system chooses the next
operation based on the actual results of previously executed steps. This is more accurate
because it doesn't rely on estimates, but it adds overhead to the query execution process.
Statistics
Used for cost estimation:
Table size
Number of tuples
Index availability
Data distribution
Example: If a table has 1 million rows → avoid full scan
Decision Sites
Which node in the network is responsible for generating the global execution plan. This
characterization defines how control is delegated across the system.
Where should optimization happen?
Centralized → one site decides
Distributed → multiple sites cooperate
Exploitation of Network Topology
Network structure affects performance:
1. Star network
2. Ring network
3. Fully connected
System tries to: Send data through shortest/cheapest paths
Exploitation of Replicated Fragments
If data exists in multiple sites:
✔ Choose the nearest copy
Example:
Customer table replicated in Site A & B
Query from Site B → use local copy
Use of Semi-joins:
A semi-join (R ⋉ S) is a reduction operator used to decrease the amount of data
transferred between sites during a join operation. Unlike a standard join, which
combines full records from both tables, a semi-join returns only the rows from the
first table R that have a match in the second table S.
Semijoin reduces data transfer ❗
Instead of sending full table: Only send matching attributes
Example:
Send only DeptIDs instead of full Department table
SQL equivalent:
Semijoins are usually written using EXISTS or IN:
SELECT *
FROM Students s
WHERE EXISTS (
SELECT 1
FROM Enrollments e
WHERE [Link] = [Link]
);
Reduces communication cost significantly
Layers of Query Processing
Query processing is done in 4 layers:
1. Query Decomposition
Steps:
Parse SQL query
Convert to relational algebra
Simplify query
Example:
SELECT Name FROM Employee WHERE Salary > 5000;
➡ Converted to:
σ Salary > 5000 (Employee)
The first layer treats the database as a single centralized unit. It focuses on the logical
correctness and simplification of the query.
Normalization: Standardizes the query into a canonical form (e.g., Conjunctive or
Disjunctive Normal Form).
Analysis: Checks for syntactic and semantic errors (e.g., verifying that tables and
attributes exist in the global schema).
Redundancy Elimination: Removes unnecessary predicates or redundant
operations.
Rewriting: Converts the SQL query into a relational algebra expression, often
represented as a query tree.
2. Data Localization
Replace global tables with fragmented tables
Example:
Employee → Employee1 (Site A), Employee2 (Site B)
✔ Query rewritten to access fragments
In this stage, the system uses the fragmentation schema to map global relations to the
actual physical fragments stored across different sites.
The global query is rewritten into a fragment query.
The system identifies which specific sites hold the necessary data fragments and
adjusts the query to target them specifically.
3. Global Query Optimization
This layer finds the most efficient execution strategy by selecting the best ordering of
operations and determining where those operations should occur.
Cost Estimation: The optimizer calculates a cost function based on CPU time, I/O
cost, and—most importantly—communication cost.
Join Strategies: Decisions are made on how to join data from different sites.
Techniques like Semi-joins are often used to reduce the size of data before it is
transmitted across the network.
Execution Plan: The output is a distributed execution plan that includes specific
communication operators (like send and receive).
Goal:
👉 Find best execution plan across sites
Decisions:
Where to perform joins?
Which site sends data?
Order of operations?
Distributed Query Execution (Local Optimization and
Execution)
The final layer occurs at the individual local sites.
Each site receives its specific sub-query and optimizes it using its own local query
optimizer, much like a centralized DBMS.
The sites execute their portions of the plan, and the results are then transmitted and
assembled at the query-initiating site to produce the final answer for the user.
Final step:
✔ Execute query plan
Includes:
Data transfer between sites
Local query execution
Combining results