INTRODUCTION

➢ Distributed processing system refers to a number of autonomous

processing elements (or computing devices) that are not necessarily
❖ Distributed
homogeneous, interconnected by a computer network and that
Data cooperate in performing their assigned tasks.
Processing ➢ The “processing element” referred to a computing device that can
execute a program on its own.

 The following 1. Processing logic: Computing devices are distributed.

2. Function: Various functions of a computer system may be assigned
are examples
to various pieces of hardware or software.
of what is
3. Data: Data may be distributed to a number of processing sites.
being 4. Control: The control of the execution of tasks might be distributed
distributed: instead of being performed by one computer system.

❖ What is a Distributed Database System?

➢ A distributed database (DDB) is a collection of

Distributed database
multiple, logically interrelated databases distributed
over a computer network.
system (DDBS) =
➢ A distributed database management system (D– DDB + D–DBMS
DBMS) is the software that manages the DDB and
provides an access mechanism that makes this
distribution transparent to the users.  The two important
➢ A DDBS is not a “collection of files” that can be terms in these
individually stored at each node of a computer definitions are
network. “logically interrelated”
➢ To form a DDBS, files should not only be logically and “distributed over a
related, but there should be structured among the computer network”
files, and access should be via a common interface.

❖ What is not a DDBS?

1- A timesharing It is the sharing of a computing resource among many users by means
computer of multiprogramming and multi-tasking at the same time by allowing a
system large number of users to interact concurrently with a single computer.
2- A loosely or ▪ Multiprocessor systems should not be considered as DDBSs.
tightly coupled Database systems that run over multiprocessor systems are called
multiprocessor parallel database systems.
system ▪ A multiprocessor system design is rather symmetrical, consisting of a
number of identical processor and memory components, and
controlled by one or more copies of the same operating system that
is responsible for a strict control of the task assignment to each
processor.
▪ This is not true in distributed computing systems, where
heterogeneity (‫ )عدم تجانس‬of the operating system as well as the
hardware is quite common.
3- A database system which resides at one of the nodes of a network of computers

A centralized database on a network node. Despite

of the existence of the network ( ‫على الرغم من وجود‬ A distributed DBMS environment
‫)الشبكة‬, the database is centrally stored and where data is distributed over
managed by one computer system (site 2) and all number of sites over a network.
the requests are routed to that site

➢ In distributed database systems, data delivery refer to delivering data

❖ Data
from the sited store the data to the sites posted the query. Data
Delivery delivery alternatives are divided according to three dimensions:
Alternatives 1. Delivery Modes 2. Frequency 3. Communication Methods

① Delivery Modes
A. A. Pull-only: ▪ The transfer of data from servers to clients is initiated by a client pull.
When a client request is received at a server, the server responds by
locating the requested information.
▪ Updates are carried out at a server without notification to clients.
B. Push- ▪ The transfer of data from servers to clients is initiated by a server push in
only: the absence of any specific request from clients.
▪ The usefulness of server push depends heavily upon the accuracy of a
server to predict the needs of clients.
C. Hybrid: ▪ Combines the client-pull and server-push mechanisms.
▪ The transfer of information from servers to clients is first initiated by a
client pull (by posing the query), and the subsequent transfer of updated
information to clients is initiated by a server push.

② Frequency

A. Periodic: ▪ Data are sent from the server to clients at regular intervals.
▪ Both pull and push can be performed in periodic fashion. Periodic
delivery is carried out on a regular and pre-specified repeating
schedule
B. Conditional: ▪ Data are sent from servers whenever certain conditions are satisfied.
▪ Such conditions can be as simple as a given time span or as
complicated as event-condition-action rules.
▪ Conditional delivery is mostly used in the hybrid or push-only delivery
systems.
C. Ad-hoc or ▪ Ad-hoc delivery is irregular and is performed mostly in a pure pull-
irregular: based system.
▪ Data are pulled from servers to clients in an ad-hoc fashion whenever
clients request it

③ Communication Methods

A. Unicast: ▪ The communication from a server to a client is one-to-one; the

server sends data to one client using a particular delivery mode
with some frequency.
B. In one-to-many: ▪ The server sends data to a number of clients.

❖ Promises of DDBSs

① Transparent management of distributed, fragmented, and replicated data:

 Transparency refers to separation of
the higher-level semantics of a system
from lower-level implementation
issues. In other words, a transparent
system “hides” the implementation
details from users.

Consider the employee database

example For this query:

 For a system to deal with this type of query over a distributed, fragmented and replicated
database, it needs to be able to deal with a number of different types of transparencies:

A. Data ▪ When a user application is written, it should not be concerned

Independence: with the details of physical data organization or distribution.
▪ Therefore, the user application should not need to be modified
when data organization changes occur due to performance
considerations.
B. Network ▪ Where the user should be protected from the operational details
Transparency or of the network; possibly even hiding the existence of the network.
Distribution ▪ Then there would be no difference between database applications
Transparency: that would run on a centralized database and those that would run
on a distributed database.
C. Replication ▪ For performance, reliability, and availability reasons, it is usually
Transparency: desirable to be able to distribute data in a replicated fashion
across the machines on a network.
▪ The replication transparency refers to that the users should not be
aware of the existence or the way of managing of copies of the
data, and the user should act as if there is a single copy of the data
D. Fragmentation ▪ It is commonly desirable to divide each database relation into
Transparency smaller fragments and treat each fragment as a separate database
object or another relation. This is commonly done for reasons of
performance, availability, and reliability.
▪ The fragmentation transparency refers to finding a query
processing strategy based on the fragments rather than the base
relations, even though the queries are specified on the base
relation.

② ➢ Distributed DBMSs are intended to improve reliability since

Improved they have replicated components and, thereby eliminate
single points of failure.
reliability/availability
➢ The failure of a single site, or the failure of a communication
through distributed link which makes one or more sites unreachable, is not
transactions: sufficient to bring down the entire system.

➢ The improved performance of distributed DBMSs is based on two

points; first, a distributed DBMS fragments the conceptual database,
enabling data to be stored in close proximity to its points of use
③
(called data localization). This has two potential advantages:
Improved 1. Each site handles only a portion of the database, contention for
performance: CPU and I/O services is not as severe as for centralized databases.
2. Localization reduces remote access delays that are usually
involved in wide area networks.

④ ➢ In a distributed environment, it is much easier to accommodate

Easier and more increasing database sizes; expansion can usually be handled by
adding processing and storage power to the network.
economical
➢ One aspect of easier system expansion is economics. It normally
system costs much less to put together a system of “smaller” computers
expansion: with the equivalent power of a single big machine.

❖ Complications Introduced by Distribution

 The problems encountered in centralized database systems take on additional

complexity in a distributed environment, even though the basic underlying principles
are the same:
➢ First, data may be replicated in a distributed ➢ Second, if some sites fail or if
environment due to reliability and efficiency some communication links fail
considerations. making some of the sites
➢ A distributed database can be designed so that the unreachable, while an update is
entire database, or portions of it; reside at being executed, the system must
different sites of a computer network. make sure that the effects will be
➢ It is not essential that every site on the network reflected on the data residing at
contain the database; it is only essential that there the failing or unreachable sites as
be more than one site where the database resides. soon as the system can recover
➢ The distributed database system is responsible for: from the failure.
▪ Choosing one of the stored copies of the requested
➢ The third point is about the
data for access in case of retrievals.
synchronization of transactions
▪ Making sure that the effect of an update is
on multiple sites is considerably
reflected on each and every copy of that data
harder than for a centralized
item.
system.

❖ Distributed DBMS Architecture

The architecture of a system defines its structure.

This means that the components of the system are
identified, the function of each component is
specified, and the interrelationships and
interactions among these components are defined.

 The possible ways in which a distributed DBMS

may be architected:
 A classification organizes the systems as
characterized with respect to (1) the autonomy of
local systems, (2) their distribution, and (3) their
heterogeneity

A. Autonomy ▪ Refers to the distribution of control, not of data. It indicates the

degree to which individual DBMSs can operate independently
B. Distribution ▪ Distribution dimension of the taxonomy deals with data. Of
course, it considers the physical distribution of data over multiple
sites; the user sees the data as one logical pool.
C. Heterogeneity ▪ Heterogeneity may occur in various forms in distributed systems,
ranging from hardware heterogeneity and differences in
networking protocols to variations in data managers.
▪ The important ones we concern here is relate to data models,
query languages, and transaction management protocols:
▪ Representing data with different modeling tools creates
heterogeneity because of the inherent expressive powers and
limitations of individual data models.
▪ Heterogeneity in query languages not only involves the use of
completely different data access paradigms in different data
models (set-at-a-time access in relational systems versus record-
at-a-time access in some object-oriented systems), but also covers
differences in languages even when the individual systems use the
same data model.

❖ Client/Server Systems

➢ The general idea is very simple: it distinguishes the functionality that needs to be
provided and divide these functions into two classes: server functions and client
functions.
➢ This provides a two level architecture which makes it easier to manage the complexity of
modern DBMSs and the complexity of distribution.

Distributed Database
Servers architecture

Where each
application server is
dedicated to one or a
few applications,
while database
servers operate in the
multiple server
fashion.

Client/Server Reference
Architecture
 Overview of Relational DBMS

➢ A database is “a structured collection of data related to some real-life phenomena that

we are trying to model”.
➢ A relational database is “one where the database structure is in the form of tables”.

➢ In the first relation EMP, ENO represents the relation primary key and the values of ENO
come from the domain of all valid employee numbers, say D1

➢ Note that each attribute of each relation does not ➢ Sometimes, there may be
have to come from a distinct domain. Various more than one possibility for
attributes within a relation or from a number of the key. These alternative keys
relations may be defined over the same domain. are called candidate keys, and
➢ Each relation in any database must have a key. The one of the candidate keys is
key of a relation is “the minimum non-empty subset chosen as the primary key.
of its attributes such that the values of the ➢ The number of attributes of a
attributes comprising the key uniquely identify each relation defines its degree,
tuple of the relation”. The attributes that make up and the number of tuples of
key are called prime attributes. Each relation has at the relation defines its
least one primary key. cardinality.

❖ Normalization

➢ The aim of normalization is to” eliminate various anomalies (or undesirable aspects) of a
relation in order to obtain better relations”.
➢ The following four problems might exist in a relation scheme:
 1. Repetition ▪ Certain information may be repeated unnecessarily.
Anomaly: ▪ For example: we have to repeat the projects details (PNAME, BUDGET) for
every employee in the EMP relation.
▪ This repetition cause redundancy, a waste of storage and complicated
update
2. Update ▪ As a consequence of the repetition of data, performing updates may be
Anomaly troublesome.
▪ For example: if the budget of a project changes, multiple tuples have to
be updated to reflect this change.
3. Insertion ▪ It may not be possible to add new tuple to the EMP1 relation.
Anomaly ▪ For example: when a new project is added and no employees are assigned
to this project yet. We can’t add the project data to the EMP1 relation; as
this will cause part of the primary key to have NULL value (no EMP
number) which is not valid.
4. Deletion ▪ This is the converse of the insertion anomaly.
Anomaly ▪ For example: If a project will be terminated, so the project details will be
deleted, and all the employees’ details of those assigned to this project
will be deleted too.
Let X, Y are two attributes or two sets of attributes in a relation R, If for
Functional each value of X in R, there is only one associated Y value, we say that “X
Dependency functionally determines Y” or that “Y is functionally dependent on X”
(FD) .Notational, this is shown as X → Y. By default the primary key of a relation
functionally determines all the non-key attributes of that relation.

▪ Normalization transforms relation schemes into ones

The following shows
without these problems.
the relations schemes
▪ A relation with one or more of the above mentioned
resulted from applying
anomalies is split into two or more relations of a higher
the normalization rules
normal form.
on the EMP1:
▪ A relation is said to be in a normal form if it satisfies the
conditions associated with that normal form.
▪ For a relation to be in 1NF: this means the relation
doesn’t contain any repeating group.
▪ For a relation to be in 2NF: it means the relation is in
the 1NF and, no non-key attributes partially depends on
the primary key attributes.
▪ For a relation to be in 3NF: commonly known as the
Boyce-Codd normal form (BCNF), it means the relation
is in the 2NF and, no non-key attributes depends on any
other non-primary key attribute.
 ❖ Relational Data Languages

➢ Data manipulation languages developed for the relational model (query languages)
fall into two fundamental groups: relational algebra languages and relational calculus
languages.

A. Relational Algebra

➢ Relational algebra consists of a set of operators that operate on relations. Each operator
takes one or two relations as operands and produces a relation as a result, which, in turn,
may be an operand to another operator.
➢ There are five fundamental relational algebra operators: selection, projection, union, set
difference, and Cartesian product. The first two of these operators are unary operators,
and the last three are binary operators.
➢ The additional operators that can be defined in terms of these fundamental operators:
intersection, theta- join, natural join, semi-join and division.

1. Selection ▪ Produces a horizontal subset of the operand relation

▪ General form:
2. Projection ▪ Produces a vertical slice of a relation (number of attributes)

▪ General form:
▪ Note: projection can generate duplicate tuples. Commercial systems (and
SQL) allow this and provide Projection with duplicate elimination or
Projection without duplicate elimination
3. Union

▪ General form:
 ▪ Two relations R and S are union compatible if and only if they are of the
same degree and the i-th attribute of each relation is defined over the
same domain
4. Set
Difference

▪ General Form:
5. Cartesian
(Cross) ▪ Given relations:
Product
▪ Cartesian (cross) product:
▪ The result of R × S is a relation of degree (k1+ k2) and consists of all (n1*
n2)-tuples where each tuple is a concatenation of one tuple of R with one
tuple of S. Example: EMP × PAY
6. Intersection
▪ General form: R, S union-compatible
7. θ-Join

▪ General form:
▪ A derivative of Cartesian product:
▪ Example:

▪ Note that:
8. Division

▪ Given relations:

▪ Example:
Types of Join:

1. Outer-Join:
▪ Ensures that tuples from one or both relations that do
not satisfy the join condition still appear in the final
result with other relation’s attribute values set to NULL
▪ Left outer join - Right outer join - Full outer join
▪ Example: EMP left outer join with ASG on ENO

2. Semi-join:

Example:

B. Relational Calculus

➢ In relational calculus-based languages, instead of specifying how to obtain the result,

one specifies what the result is by stating the relationship that is supposed to hold
for the result.
➢ There are many languages that are based on relational calculus, the most popular
ones being SQL.
 ❖ Review of Computer Networks

➢ Computer network is an interconnected collection of autonomous computers that are

capable of exchanging information among them.
➢ Types of Networks:

① According to scale (geographic distribution)

A. Wide area network (WAN): C. Local area network B. Metropolitan

▪ Distance between any two nodes > 20km (LAN): area
and can go as high as thousands of kms ▪ Limited in geographic network
▪ Long delays due to distance traveled scope (usually < 2km) (MAN):
▪ Heterogeneity of transmission media ▪ Speeds 10-1000 Mbps ▪ In between
▪ Speeds of 150Mbps to 10Gbps (OC192 ▪ Short delays and low LAN and
on the backbone) noise WAN

② According to A. Irregular: No regularity in the interconnection (Internet)

B. Bus: Typical in LANs (Ethernet)
Topology
C. Star D. Ring E. Mesh

③ According to physical communication schemes employed

A. Point-to-point (unicast) : B. Broadcast (multi-point):

▪ One or more (direct or indirect) links between ▪ Messages are transmitted over a
each pair of nodes shared channel and received by all the
▪ Communication always between two nodes nodes
▪ Receiver and sender are identified by their ▪ Each node checks the address and if it
addresses included in the message header not the intended recipient, ignores
▪ Message may follow one of many links ▪ Multi-cast: special case
between the sender and receiver using ▪ Message is sent to a subset of the
switching or routing nodes

Distributed Database Design

➢ Making decisions about the placement Framework of
of data and programs across the sites Distribution
of a computer network as well as
possibly designing the network itself.
➢ In Distributed DBMS: placement of the
distributed DBMS software; and
placement of the applications that run
on the database

➢ The organization of distributed systems can be investigated along three orthogonal

dimensions:

A. Level of ▪ Refers to sharing of both data and the applications. There are three
sharing possibilities:-
First, there is no sharing: Where each application and its data execute
at one site, and there is no communication with any other program or
access to any data file at other sites.
Second; the level of data sharing: all the programs are replicated at all
the sites, but data files are not. Accordingly, user requests are handled
at the site where they originate and the necessary data files are
moved around the network.
Finally, in data-plus-program sharing: both data and programs may be
shared, meaning that a program at a given site can request a service
from another program at a second site, which, in turn, may have to
access a data file located at a third site
B. Access ▪ The access patterns of user requests may be static, so that they do
Pattern not change over time, or dynamic.
Behavior
C. The level of ▪ One possibility, of course, is that the designers do not have any
knowledge information about how users will access the database.
about the ▪ Another alternative is that the designers have complete
access information, where the access patterns can reasonably be predicted,
pattern or partial information, where there are deviations from the
behavior predictions.

➢ Two major strategies that have been identified for designing distributed databases are
the top-down approach and the bottom-up approach.
Top-down: mostly in designing systems from scratch - mostly in homogeneous systems
Bottom-up: when the databases already exist at a number of sites

❖ Top-Down Design Process

➢ The activity begins with a requirements analysis
that defines the environment of the system
➢ The requirements study also specifies the
objectives of a distributed DBMS. These
objectives are defined with respect to
performance, reliability and availability,
economics, and flexibility.
➢ The requirements document is input to two
parallel activities: view design and conceptual
design.
➢ The view design activity deals with defining the
interfaces for end users. The conceptual design
is the process by which the enterprise is
examined to determine entity types and
relationships among these entities.
➢ There is a relationship between the conceptual
The distribution design activity
design and the view design: the conceptual
consists of two steps:
design can be interpreted as being an
1. Fragmentation
integration of user views.
2. Allocation.
➢ The definition of global conceptual schema
(GCS) is identical to that in a centralized ➢ The last step in the design process is
database design. the physical design, which maps the
➢ The global conceptual schema (GCS) and access local conceptual schemas to the
pattern information collected as a result of physical storage devices available at
view design are inputs to the distribution the corresponding sites.
design step. ➢ The inputs to this process are the
➢ The objective at this stage is to design the local local conceptual schema and the
conceptual schemas (LCSs) by distributing the access pattern information about the
entities over the sites of the distributed system. fragments in them.

❖ Distribution Design Issues

➢ The relations in a database schema are usually decomposed into smaller fragments, for
better understanding we need to answer the following questions:
1. Why fragment at all? 2. How should we fragment? 3. How much should we fragment?
4. Is there any way to test the correctness of decomposition?
5. How should we allocate?
6. What is the necessary information for fragmentation and allocation?
 ① Why fragment at all ? Can't we just distribute relations?

➢ A relation is not a suitable unit, for a number of reasons: ➢ Fragmentation permits a

1- Application views are usually subsets of relations. number of transactions to
Therefore, the locality of accesses of applications is execute concurrently.
defined not on entire relations but on their subsets. ➢ In addition, the
2- if the applications that have views defined on a given fragmentation of relations
relation reside at different sites, two alternatives can be typically results in the
followed, with the entire relation being the unit of parallel execution of a
distribution. Either the relation is not replicated and is single query by dividing it
stored at only one site, or it is replicated at all or some of into a set of subqueries
the sites where the applications reside. that operate on fragments.
The first results in an unnecessarily high volume of ➢ Thus fragmentation
remote data accesses. typically increases the
The latter has unnecessary replication, which causes level of concurrency and
problems in executing updates and may not be desirable therefore the system
if storage is limited. throughput.

Problems with fragmentations

1- If the applications have conflicting requirements that prevent decomposition of the

relation into mutually exclusive fragments → Those applications whose views are defined
on more than one fragment may suffer performance degradation; as it might be necessary
to retrieve data from two fragments and then take their join, which is costly. Minimizing
distributed joins is a fundamental fragmentation issue.
2- Semantic data control, specifically to integrity checking. As a result of fragmentation,
attributes participating in a dependency may be decomposed into different fragments
that might be allocated to different sites → In this case, even the simpler task of checking
for dependencies would result in chasing after data in a number of sites.

➢ Fragmentation refers to ways of dividing a relation into smaller ones.

② How There are two main alternatives for this:
should we ▪ Dividing the relation horizontally or Dividing it vertically.
fragment? ▪ The fragmentation may be nested; if the nesting fragmentations are of
different types, it’s called hybrid fragmentation.

③ Degree of Fragmentation
➢ The extent to which the database should be fragmented is an important decision that
affects the performance of query execution.
➢ The degree of fragmentation goes from (one extreme) that is not to fragment at all, to
(the other extreme), to fragment to the level of individual tuples (in the case of horizontal
fragmentation) or to the level of individual attributes (in the case of vertical
fragmentation).

④ Correctness Rules of Fragmentation

➢ There are three rules to apply during fragmentation together, to ensure that the
database doesn’t have any semantic change during or after fragmentation:

Completeness If a relation instance R is decomposed into fragments FR ={R1;R2;

……;Rn}, each data item that can be found in R can also be found in
one or more of Ri’s.
Reconstruction If relation R is decomposed into fragments R1, R2, ..., Rn, then there
should exist some relational operator ∇ such that R = ∇1 ≤ I ≤ nRi.
Disjointness ▪ If relation R is decomposed into fragments R1, R2, ..., Rn, and data
item di is in Rj, then di should not be in any other fragment Rk (k ≠ j ).
▪ This criterion ensures that the horizontal fragments are disjoint.
▪ If relation R is vertically decomposed; its primary key attributes are
typically repeated in all its fragments (for reconstruction). Therefore,
in case of vertical partitioning, disjointness is defined only on the
non-primary key attributes of a relation.
⑤ Allocation Alternatives

➢ When the database is fragmented properly, we have to decide about the allocation of
the fragments to various sites on the network.
➢ The replication alternatives of allocation are:

1- Non-replicated →
◦ Partitioned : each fragment resides
at only one site
2- Replicated → The reasons for
replication are reliability and
efficiency of read-only queries.
◦ Fully replicated: each fragment at
each site.
◦ Partially replicated: each
Comparison of Replication Alternatives
fragment at some of the sites.
 ❖ The various types of fragmentation

① Horizontal Fragmentation It refers to dividing the

relations horizontally by tuples
➢ Horizontal fragmentation partitions a relation along its
tuples. Thus each fragment has a subset of the tuples
of the relation.

Example:
➢ The PROJ relation is horizontally fragmented into
PROJ1, PROJ2 as described below:
➢ PROJ1 : projects with budgets less than $200,000
➢ PROJ2 : projects with budgets greater than or equal to
$200,000

There are two versions of horizontal partitioning: primary and derived.

Primary Horizontal Fragmentation

➢ Is performed using predicates that are defined on that relation.
➢ Is defined by a selection operation on the owner relations of a database schema.

Example:
The relation PROJ is fragmented into horizontal fragments
PROJ1 and PROJ2. The fragments are defined as follows:

Example:
The relation PROJ can also be fragmented based on
the project location. The resulting fragments are:

Minterm Predicates
➢ A horizontal fragment Ri of relation R consists of all the tuples of R that satisfy a minterm
predicate mi.
➢ Hence, given a set of minterm predicates M, there are as many horizontal fragments of
relation R as there are minterm predicates.
➢ This set of horizontal fragments is also commonly referred to as the set of minterm
fragments.
➢ Minterm predicates consists of set of simple predicates that will form the minterm
predicates. Two important aspects of simple predicates Pr are their completeness, and
their minimality.
➢ Pr should be complete: each tuple of a relation has the same access probability for each
application
➢ Pr should be minimal: there should be at least one application that accesses fi and f j
differently. The simple predicate should be relevant in determining a fragmentation.

Derived horizontal fragmentation

➢ Derived horizontal fragmentation is the partitioning of a relation that result from

predicates being defined on another relation.
➢ A derived horizontal fragmentation is defined on a member relation of a link according to
a selection operation specified on its owner.
➢ It is important to remember two points:
1. The link between the owner and the member relations is defined as an equi-join.
2. An equi-join can be implemented by means of semijoins. We want to partition a member
relation according to the fragmentation of its owner, but we also want the resulting
fragment to be defined only on the attributes of the member relation.

Example:
▪ Consider link L1: owner(L1) = PAY and The result of
member(L1) = EMP. this
▪ Then we can group engineers into two fragmentation
groups according to their salary: those
making less than or equal to $30,000, and
those making more than $30,000. The two
fragments EMP1 and EMP2 are defined as
follows:

Derived fragmentation may follow a chain

where one relation is fragmented as a result
of another one’s design and it causes the
fragmentation of another relation (the chain
PAY→EMP→ASG).

It refers to dividing the relations

② Vertical fragmentation
vertically by attributes
 Example:
▪ The PROJ relation is vertically fragmented into PROJ1,
PROJ2 as described below:
▪ PROJ1: information about project budgets
▪ PROJ2: information about project names and locations

③ Hybrid fragmentation
Nested fragmentation
with different types.

Query Processing
➢ Query processing is a critical performance issue, ➢ Many important functions
so it has received (and continues to receive) characterize this mapping:
considerable attention in both centralized and 1- The calculus query must be
distributed DBMSs. decomposed into a sequence of
➢ But, the query processing problem is much more relational operators called an
difficult in distributed environments; this is algebraic query.
because a larger number of parameters affect 2- The data accessed by the query
the performance of distributed queries. must be localized so that the
➢ Distributed relations are divided and distributed operators on relations are
as fragments. Distributed database design is of translated to bear on local data
major importance for query processing since the (fragments).
definition of fragments is based on the objective 3- The algebraic query on fragments
of increasing reference locality, and sometimes must be extended with
parallel execution for the most important communication operators and
queries. optimized with respect to a cost
➢ The role of a distributed query processor is to function to be minimized. This
map a high-level query on a distributed database cost function typically refers to
(a set of global relations) expressed in relational computing resources such as disk
calculus into a sequence of database operators I/Os, CPUs, and communication
(of relational algebra) on relation fragments. networks

❖ Query Processing Problem

The main function of a relational query processor is to transform a high-
level query (in relational calculus) into an equivalent lower-level query (in
some variation of relational algebra).

➢ The low-level query actually implements the

execution strategy for the query.
➢ The transformation must achieve both
correctness and efficiency: It is correct if the
➢ Apply the following simple user
low-level query has the same semantics as the
query:
original query, that is, if both queries produce
➢ “Find the names of employees who
the same result.
are managing a project”. The
➢ The well-defined mapping from relational
expression of the query in relational
calculus to relational algebra makes the
calculus using the SQL syntax is:
correctness issue easy. But producing an
efficient execution strategy is more involved.
➢ A relational calculus query may have many
equivalent and correct transformations into
relational algebra. Since each equivalent ➢ Two equivalent relational algebra
execution strategy can lead to very different queries that are correct
consumptions of computer resources, the main transformations of the query above
difficulty is to select the execution strategy that are:
minimizes resource consumption.

Strategy ① Strategy ② avoids

Cartesian product, so
Strategy ② it may be “better”

➢ In a distributed system, relational algebra is ➢ Example: Show the importance of site

not enough to express execution strategies. selection and communication for a
It must be supplemented with operators for chosen relational algebra query against a
exchanging data between sites. fragmented database
➢ Besides the choice of ordering relational ➢ For the following query:
algebra operators, the distributed query
processor must also select the best sites to ➢ Assume that relations EMP and ASG are
process data, and possibly the way data horizontally fragmented as follows:
should be transformed.
➢ This increases the solution alternatives
from which to choose the distributed
execution strategy, making distributed
query processing more difficult.
 Distribution of fragments over sites

Two equivalent distributed

execution strategies for this query:

➢ To evaluate the resource consumption of these two

strategies, we use a simple cost model with the following
assumptions:
▪ a tuple access, denoted by tupacc, is 1 unit
▪ a tuple transfer, denoted tuptrans, is 10 units
▪ cardinality of relations EMP and ASG have 400 and 1000
tuples, respectively,
▪ There are 20 managers in relation ASG.
▪ Data is uniformly distributed among sites.
▪ Relations ASG and EMP are locally clustered on attributes
RESP and ENO, respectively.
▪ There is direct access to tuples of ASG (respectively, EMP)
based on the value of attribute RESP (respectively, ENO)

➢ The total cost of strategy A can be ➢ The total cost of strategy B can be derived as
derived as follows: follows:
1. Produce ASG' by selecting ASG 1. Transfer EMP to site 5 requires 400 *
requires (10+10) * tupacc = 20 tuptrans = 4000
2. Transfer ASG' to the sites of EMP 2. Transfer ASG to site 5 requires 1000 *
requires (10+10) *tuptrans = 200 tuptrans = 10000
3. Produce EMP' by joining ASG' and EMP 3. Produce ASG' by selecting ASG requires
requires (10+10) * tupacc * 2= 40 1000 * tupacc = 1000
4. Transfer EMP' to result site requires 4. Join EMP and ASG' requires 400 * 20
(10+10) *tuptrans = 200 *tupacc = 8000
5. The total cost = 20+ 200+ 40+ 200= 460 5. The total cost = 4000+10000+1000+8000=
▪ The join of ASG' and EMP (step 3) 23000
exploit the index on ENO of EMP. Thus, ▪ We assume that the access methods to
EMP is accessed only once for each relations EMP and ASG based on attributes
tuple of ASG'. RESP and ENO are lost because of data
▪ Strategy A is better by a factor of 50, transfer.
which is quite significant. It also ▪ We assume that the join of EMP and ASG' in
provides better distribution of work step 4 is done by the default nested loop
among sites. algorithm (it performs the Cartesian product
of the two input relations).
 The difference between the costs of the two strategies would be higher with slower
communication and/or higher degree of fragmentation

❖ Query optimization objectives

➢ An important aspect of query processing is ➢ In a distributed database system, the

query optimization. As many execution total cost to be minimized includes CPU,
strategies are correct transformations of I/O, and communication costs.
the same high-level query, the one that ➢ The CPU cost is incurred when
optimizes (minimizes) resource performing operators on data in main
consumption should be retained. memory.
➢ A good measure of resource consumption is ➢ The I/O cost is the time necessary for
the total cost that will be incurred in disk accesses. This cost can be minimized
processing the query. The total cost is the by reducing the number of disk accesses
sum of all times incurred in processing the through fast access methods to the data
operators of the query at various sites. and efficient use of main memory
➢ Another important measure is the response (buffer management).
time of the query, which is the time ➢ The communication cost is the time
elapsed for executing the query. needed for exchanging data between
➢ An important note here is that the total sites participating in the execution of the
cost and the response time measures are query. This cost is incurred in processing
not equal. the messages and in transmitting the
➢ This is because many operators can be data on the communication network.
executed in parallel at different sites, so ➢ But The I/O and CPU cost are the only
the response time of a query may be factors considered in centralized DBMSs.
significantly less than its total cost.

❖ Complexity of Relational Operations

➢ The simplest way of defining complexity is

in terms of relation cardinalities
independent of physical implementation
details such as fragmentation and storage
structures.
➢ The complexity of unary and binary
operators in the order of increasing
complexity, and thus of increasing
execution time: Where n refers to the
relation’s cardinality
 ❖ Types of Optimization

➢ Query optimization aims at choosing the “best” ➢ To avoid the high cost of
point in the solution space of all possible execution exhaustive search →
strategies. 1- They try to find a very good
➢ The first method for query optimization is to solution, not necessarily the best
search the solution space, predict the cost of each one, but avoid the high cost of
strategy, and select the strategy with minimum optimization, in terms of memory
cost. and time consumption.
➢ The problem is that the solution space can be 2- The use of heuristics, whose
large; so there may be many equivalent strategies, effect is to restrict the solution
even with a small number of relations. space so that only a few
➢ The problem becomes worse as the number of strategies are considered. An
relations or fragments increases (becomes greater important heuristic in distributed
than 5 or 6). systems is to replace join
➢ Having high optimization cost is not necessarily operators by combinations of
bad, particularly if query optimization is done once semi-joins to minimize data
for many subsequent executions of the query. communication.

❖ Optimization Timing

➢ A query may be optimized at different times ➢ The main advantage over static query
relative to the actual time of query optimization is that the actual sizes of
execution. intermediate relations are available to
➢ Optimization can be done statically before the query processor, thereby minimizing
executing the query or dynamically as the the probability of a bad choice. The
query is executed. main shortcoming is that query
➢ Static query optimization is done at query optimization, an expensive task, must
compilation time. Dynamic query be repeated for each execution of the
optimization proceeds at query execution query.
time. ➢ Hybrid query optimization attempts to
➢ At any point of execution, the choice of the provide the advantages of static query
best next operator can be based on accurate optimization while avoiding the issues
knowledge of the results of the operators generated by inaccurate estimates.
executed previously. ➢ The approach is basically static, but
➢ Database statistics are not needed to dynamic query optimization may take
estimate the size of intermediate results. place at run time when a high difference
However, they may still be useful in between predicted sizes and actual size
choosing the first operators. of intermediate relations is detected.
 ❖ Layers of Query Processing

➢ The problem of query processing can itself be decomposed into several sub-problems;
corresponding to various layers:
➢ A generic layering scheme for query processing is shown where each layer solves a well-
defined sub-problem.

➢ The input is a query on global data expressed in

relational calculus. This query is posed on global
relations, meaning that data distribution is hidden.
➢ Four main layers are involved in distributed query
processing:
▪ The first three layers:
1. Map the input query into an optimized distributed
query execution plan.
2. Perform the functions of query decomposition, data
localization, and global query optimization.
3. These layers are performed by a central control site and
use schema information stored in the global directory.
▪ The fourth layer performs distributed query execution by
executing the plan and returns the answer to the query.

① Query Decomposition

➢ The first layer decomposes the calculus query into an algebraic query on global relations.
➢ The information needed for this transformation is found in the global conceptual schema
describing the global relations. The information about data distribution is this layer.
➢ Query decomposition can be viewed as four successive steps which are: normalization,
analysis, simplification, and restructuring

A. Normalization ▪ The calculus query is rewritten in a normalized form that is suitable

for subsequent manipulation.
▪ This step involves the manipulation of the query quantifiers and of
the query qualification by applying logical operator priority.
B. Analysis ▪ The normalized query is analyzed semantically so that incorrect
queries are detected and rejected as early as possible
C. Simplification ▪ The correct query (still expressed in relational calculus) is simplified.
This refers to elimination of redundant predicates.
 ▪ Note that redundant queries are likely to arise when a query is the
result of system transformations applied to the user query.
D. Restructuring ▪ The calculus query now is restructured as an algebraic query.
▪ Note that several algebraic queries can be derived from the same
calculus query, and that some algebraic queries are “better” than
others.
▪ The quality of an algebraic query is defined in terms of expected
performance. The traditional way to do this transformation toward a
“better” algebraic specification is to start with an initial algebraic
query and transform it in order to find a “good” one.
▪ The algebraic query generated by this layer is good in the sense that
the worse executions are typically avoided. This query is generally far
from providing an optimal execution, since information about data
distribution and fragment allocation are not used yet

② Data Localization

➢ The input to the second ➢ Generating a query on fragments is done in two steps:
layer is an algebraic query 1- The query is mapped into a fragment query by
on global relations. The main substituting each relation by its reconstruction
role of the second layer is to program (also called materialization program).
localize the query’s data 2- The fragment query is simplified and restructured to
using data distribution produce another “good” query. Simplification and
information in the fragment restructuring may be done according to the same
schema rules used in the decomposition layer.

③ Global Query Optimization

➢ The input to the third layer is an algebraic query on fragments. The goal of query
optimization is to find an execution strategy for the query which is close to optimal.
➢ An execution strategy for a distributed query can be described with relational algebra
operators and communication primitives (send/receive operators) for transferring data
between sites.
➢ The previous layers have already optimized the query, For example: by eliminating
redundant expressions.
➢ However, this optimization is independent of fragment characteristics such as fragment
allocation and cardinalities. In addition, communication operators are not yet specified. By
permuting (‫ )تبديل‬the ordering of operators within one query on fragments, many
equivalent queries may be found.
➢ Query optimization consists of finding the “best” ordering of operators in the query,
including communication operators that minimize a cost function.
➢ To select the ordering of operators it is necessary to predict execution costs of alternative
candidate orderings.
➢ Determining execution costs before query execution (static optimization) is based on
fragment statistics and the formulas for estimating the cardinalities of results of relational
operators.
➢ An important aspect of query optimization is the join ordering; a basic technique for
optimizing a sequence of distributed join operators is through the semijoin operator. The
semijoin in a distributed system reduces the size of the join operands and then the
communication cost

➢ The last layer is performed by all the sites having fragments

involved in the query.
④ Distributed ➢ Each subquery executing at one site, called a local query, is then
Query optimized using the local schema of the site and executed.
Execution ➢ At this time, the algorithms to perform the relational operators
may be chosen. Local optimization uses the algorithms of
centralized systems

Query Decomposition and Data Localization

➢ Query decomposition maps a ❖ Query Decomposition
distributed calculus query into an
algebraic query on global relations. ➢ The query decomposition is the first phase
➢ The resultant query is subsequently of query processing. It transforms a
optimized, as the details of the relational calculus query into a relational
processing environment are added to algebra query.
the query. ➢ Both input and output queries in this phase
➢ Data localization takes as input the refer to global relations, without
decomposed query on global relations knowledge of the distribution of relations.
and applies data distribution ➢ So, query decomposition is the same for
information to the query in order to centralized and distributed systems.
localize its data.
➢ Data localization determines which There are four major steps in the query
fragments are involved in the query and decomposition phase:
thereby transforms the distributed A) Normalization. B) Analysis.
query into a fragment query. C) Elimination of redundancy. D) Rewriting.
 ① Normalization

➢ The goal of this step is to transform the

query to a normalized form to facilitate
further processing.
➢ With relational languages such as SQL, the
most important transformation is that of
the query qualification (the WHERE clause)
which may be complex.
➢ There are two possible normal forms for
the predicate: one giving precedence to the
AND (^) and the other to the OR (v).

The transformation of the quantifier-free predicate is straightforward using the well-

known equivalence rules for logical operations (^,v and ¬):

Note:
OR's mapped into union.
AND's mapped into join or selection.

▪ Example: Let us consider the following query on the

engineering database that we have been referring to:
▪ “Find the names of employees who have been working
on project P1 for 12 or 24 months”. The query
expressed in SQL is:

The qualification in conjunctive normal form is In the latter form, treating the two
conjunctions independently may
lead to redundant work if common
While the qualification in disjunctive normal subexpressions are not eliminated
form is
② Analysis
 ➢ Query analysis enables rejection of normalized queries for which further processing is
either impossible or unnecessary.
➢ The main reasons for rejection are that the query is type incorrect or semantically
incorrect.
➢ When one of these cases is detected, the query is simply returned to the user with an
explanation. Otherwise, query processing is continued.

A. Type incorrect B. Semantically incorrect

▪ If any of its attribute or relation ▪ Components do not contribute in any way to the
names are not defined in the generation of the result
global schema ▪ Only a subset of relational calculus queries can be
▪ If operations are applied to tested for correctness
attributes of the wrong type ▪ Those that do not contain disjunction and negation

▪ Example: consider the following SQL query on the

engineering database:
▪ It is type incorrect for two reasons: First, attribute E# is
not declared in the schema. Second, the operation “>200”
is incompatible with the type string of ENAME.

➢ To detect that the query is semantically incorrect a query graph and join graph are used.
➢ In a query graph, one node indicates the result relation, and any other node indicates an
operand relation. An edge between two nodes one of which does not correspond to the
result represents a join, whereas an edge whose destination node is the result represents
a project.
➢ In the join graph only the joins are considered. It’s useful in the query optimization
phase.

▪ Example: Let us consider the following query:

▪ “Find the names and responsibilities of programmers who have been working on the
CAD/CAM project for more than 3 years”. The query expressed in SQL is:

Shows the query

graph for the
query in the
example and its
corresponding
join graph
▪ Example: Let us consider the
following SQL query:

Shows the query graph for this example, it’s is

disconnected, which indicates that the query is
semantically incorrect.

➢ There are basically three solutions to the problem:

1) Reject the query
2) Assume that there is an implicit Cartesian product between relations ASG and PROJ
3) Infer (using the schema) the missing join predicate ASG.PNO = PROJ.PNO which
transforms the query into that of the previous example.

③ Elimination of Redundancy (Simplification)

Example:

The previous SQL query can be

simplified using the previous
rules to become:
 ④ Rewriting (restructuring)

➢ The last step of query decomposition is rewriting the query in relational algebra. Here
we represent the relational algebra query graphically by an operator tree.
➢ An operator tree is a tree in which a leaf node is a relation stored in the database, and
a non-leaf node is an intermediate relation produced by a relational algebra operator.
The sequence of operations is directed from the leaves to the root, which represents
the answer to the query.

The query “Find the names of employees other than J. Doe who
Example: worked on the CAD/CAM project for either one or two years”
whose SQL expression is:

Fig [3] The

operator
tree for
this query.

➢ By applying transformation rules, many different trees may be found equivalent to the
one produced by the method described above.
➢ We now present the six most useful equivalence rules, which concern the basic
relational algebra operators:
 The application of these six The restructuring of the tree in Fig[3] leads to the tree
rules enables the generation of in Fig [5].
many equivalent trees. The resulting tree is good in the sense that repeated
For instance, the tree in Fig [4] is access to the same relation (as in Fig [3]) is avoided and
equivalent to the one in Fig [3]. that the most selective operations are done first.

Fig [4]
Rewritten
Operator
Tree

❖ Localization of Distributed Data

➢ We presented general techniques for decomposing and restructuring queries expressed

in relational calculus. These global techniques apply to both centralized and distributed
DBMSs and do not take into account the distribution of data. This is the role of the
localization layer.
➢ The localization layer translates an algebraic query on global relations into an algebraic
query expressed on physical fragments. Localization uses information stored in the
fragment schema.
➢ A naive way to localize a distributed query is to generate a query where each global
relation is substituted by its localization program. This can be viewed as replacing the
leaves of the operator tree of the distributed query with subtrees corresponding to the
localization programs. We call the query obtained this way the localized query.

Example: ▪ ASG fragmented into ASG1

and ASG2 as follows:
▪ Assume EMP is fragmented
into EMP1, EMP2, and
EMP3 as follows: ▪ Replace EMP by (EMP1 U
EMP2 U EMP3) and ASG
by (ASG1 U ASG2) in any
query
 Reduction for Primary Horizontal Fragmentation

Example: ➢ The localization program for the horizontally fragmented

relation is the union of the fragments. In our example we have
▪ Relation EMP (ENO, EMP = EMP1 U EMP2 U EMP3
ENAME, TITLE) is split ➢ The reduction of queries on horizontally fragmented relations
into three horizontal consists primarily of determining, after restructuring the
fragments EMP1, subtrees, those that will produce empty relations, and
EMP2, and EMP3 as removing them.
defined in the ➢ Horizontal fragmentation can be exploited to simplify both
previous example. selection and join operations.

① Reduction with Selection ② Reduction with Join

▪ Example: We now illustrate reduction ▪ Joins on horizontally fragmented relations can

by horizontal fragmentation using the be simplified when the joined relations are
following example query: fragmented according to the join attribute.
▪ The simplification consists of distributing joins
over unions and eliminating useless joins. The
distribution of join over union can be stated
as:
▪ Applying the naive approach to
localize EMP from EMP1, EMP2, and where Ri are
fragments of R and S is a relation.
EMP3 gives the localized query of Fig
▪ Example: Assume EMP is fragmented as before
[6].
and ASG is fragmented as follows:
▪ By commuting the selection with the
union operation, it is easy to detect
that the selection predicate
contradicts the predicates of EMP1
▪ Reduction is done by Distributing join over
and EMP3, thereby producing empty
unions as in Fig [7].
relations.
 Reduction for Vertical Fragmentation

➢ Find useless (not empty) intermediate relations. Relation R defined over attributes

A = {A1, ..., An} vertically fragmented as Ri

Example: Applying the reduction rule we get the reduced query in Fig [8].

Fig [8]
Reduction for
Vertical
Fragmentation

▪ By commuting the projection with the join (i.e., projecting on ENO, ENAME), we can see
that the projection on EMP2 is useless because ENAME is not in EMP2. Therefore, the
projection needs to apply only to EMP1.

Reduction for Derived Fragmentation

➢ Reduction is done by distribute joins over unions, and Example:

apply the join reduction for horizontal fragmentation.

▪ Given a one-to-many relationship from EMP to ▪ The localization program for a

ASG, relation ASG (ENO, PNO, RESP, DUR) can horizontally fragmented relation is the
be indirectly fragmented according to the union of the fragments. In this
following rules: example, we have
ASG = ASG1 UASG2.
▪ And the predicates of ASG1 and EMP2
conflict; thus we have
ASG1 ⋈ EMP2 = ∅

▪ Apply the following query Fig [9] The

localized
query

The following figures show the steps

to produce the reduced query:
 Fig [10] query after pushing Fig [11] applying
selection down joins over union

➢ The last step to produce the reduced query

in fig [12] is to eliminate the empty
intermediate relations (left sub-tree) as we
mentioned before ASG1 ⋈ EMP2 = ∅

➢ The reduced query is always preferable to

the localized query because the number of
partial joins usually equals the number of
fragments of R. Fig [12] the reduced query

Query Optimization

➢ The optimizer is the one responsible of

finding an “optimal” ordering of
operations for a given query.
➢ Query optimization refers to the process
of producing a query execution plan
➢ (QEP) which represents an execution
strategy for the query. This QEP
minimizes the objective cost function.
➢ A query optimizer is the software
module that performs query
optimization; it consists of three
components: a search space, a cost
model, and a search strategy. Query Optimization Process
 ① The Search space ② The Cost model

➢ The search space is the set of alternative execution ➢ The cost model predicts the
plans that represent the input query. These plans cost of a given execution plan.
are equivalent, in the sense that they yield the same To be accurate, the cost model
result, but they differ in the execution order of must have good knowledge
operations and the way these operations are about the distributed
implemented, and in their performance. execution environment.
➢ The search space is obtained by applying
transformation rules discussed before ③ The Search strategy
➢ Search space characterized by alternative execution.
It Focus on join trees ➢ The search strategy explores
➢ For N relations, there are O(N!) equivalent join trees the search space and selects
that can be obtained by applying commutativity and the best plan, using the cost
associativity rules. model.

An important question is how to “move” in the search space. There are two options:

① Deterministic ② Randomized

▪ Start from base relations and build ▪ Search for optimalities around a particular
plans by adding one relation at each starting point
step ▪ Trade optimization time for execution time
▪ Dynamic programming: breadth-first ▪ Better when > 10 relations
▪ Greedy: depth-first ▪ Iterative improvement

Example:

Consider the following ▪ Fig represents three equivalent join trees for that query,
query: which are obtained by exploiting the associativity of binary
operators.
▪ Each of these join trees can be assigned a cost based on the
estimated cost of each operator.
▪ Join tree (c) which starts with a Cartesian product may have
a much higher cost than the other join trees.
 Equivalent
Join Trees

➢ Investigating a large search space may ➢ Another important restriction is with respect
make optimization time prohibitive, to the shape of the join tree. Two kinds of
sometimes much more expensive than join trees are usually distinguished: linear
the actual execution time. versus bushy trees
➢ Therefore, query optimizers typically ➢ A linear tree is a tree such that at least one
restrict the size of the search space operand of each operator node is a base
they consider: relation. By considering only linear trees, the
➢ The first restriction is to use heuristics: size of the search space is reduced to O(2N).
The most common heuristic is to ➢ A bushy tree is more general and may have
perform selection and projection when operators with no base relations as operands
accessing base relations. Another (both operands are intermediate relations).
common heuristic is to avoid Cartesian However, in a distributed environment,
products that are not required by the bushy trees are useful in exhibiting
query. parallelism.

The Two Major

Shapes of Join Trees

❖ Join Ordering in Distributed Queries

➢ Ordering joins is an important aspect of centralized queryoptimization. Join ordering in a

distributed context is even more important since joins between fragments may increase
the communication time.
➢ Two basic approaches exist to order joins in distributed queries: One tries to optimize the
ordering of joins directly, whereas the other replaces joins by combinations of semijoins
in order to minimize communication costs.
➢ For a join query between two relations (R ⋈ S); where R and S are relations stored at
different sites.
➢ The obvious choice of the relation to transfer is to send the smaller relation to the site of
the larger one, which gives rise to two possibilities. To make this choice we need to
evaluate the sizes of R and S. Transfer of Operands in Binary
 ➢ We now consider the case where there are more than two relations to join.
➢ As in the case of a single join, the objective of the join-ordering algorithm is to transmit
smaller operands. The difficulty stems from the fact that the join operations may reduce
or increase the size of the intermediate results. Thus, estimating the size of join results is
mandatory, but also difficult. Example:

Consider the following query 1. EMP → site 2; Site 2 computes EMP’ = EMP ⋈ ASG
expressed in relational algebra: EMP’ → site 3; Site 3 computes EMP’ ⋈ PROJ.
PROJ ⋈PNO ASG ⋈ENO EMP 2. ASG → site 1; Site 1 computes EMP’ = EMP ⋈ ASG
EMP’ → site 3; Site 3 computes EMP’ ⋈ PROJ.
3. ASG → site 3; Site 3 computes ASG’ = ASG ⋈ PROJ
ASG’ →site 1; Site 1 computes ASG’ ⋈ EMP.
4. PROJ → site 2; Site 2 computes PROJ’ = PROJ ⋈
ASG; PROJ’ → site 1; Site 1 computes PROJ’ ⋈
EMP.
To implement this query there 5. EMP → site 2; PROJ→ site 2; Site 2 computes EMP
are many alternatives: ⋈ PROJ ⋈ ASG

➢ To select one of these programs, the ▪ Example: Assume that relations EMP,
following sizes must be known or ASG, and PROJ of the following query are
predicted: size (EMP), size (ASG), size stored as follows, where relation EMP is
(PROJ), size (EMP ⋈ ASG), and size (ASG ⋈ fragmented “Names of employees
PROJ). working on the CAD/CAM project”
➢ Furthermore, if it is the response time that
is being considered, the optimization must
take into account the fact that transfers
can be done in parallel with strategy 5.
➢ An alternative to enumerating all the
solutions is to use heuristics that consider
▪ There are several possible strategies,
only the sizes of the operand relations by
including the following:
assuming
1. Execute the entire query (EMP ⋈ ASG
▪ For example: the cardinality of the resulting
⋈ PROJ) by moving EMP1 and ASG to
join is the product of operand cardinalities.
site 2.
In this case, relations are ordered by
2. Execute (EMP ⋈ASG) 1 PROJ by moving
increasing sizes and the order of execution
(EMP1 ⋈ASG) and ASG to site 2, and so
is given by this ordering and the join graph.
on.
For instance, the order (EMP, ASG, PROJ)
▪ The choice between the possible
could use strategy 1, while the order (PROJ,
strategies requires an estimate of the
ASG, EMP) could use strategy 4.
size of the intermediate results.
▪ For example: if size (EMP1 ⋈ ASG) > size (EMP1), strategy 1 is preferred to strategy 2.
Therefore, an estimate of the size of joins is required. Example:

Let us consider the query ▪ With a point–to–point network, the best strategy is
PROJ ⋈ ASG, where PROJ and to send each PROJi to site 3, which requires a
ASG are fragmented. Assume transfer of 3000 kbytes, versus 6000 kbytes if ASG is
that the allocation of fragments sent to sites 1,2, and 4.
and their sizes are as follows: ▪ However, with a broadcast network, the best
strategy is to send ASG (in a single transfer) to sites
1, 2, and 4, which incurs a transfer of 2000 kbytes.
The latter strategy is faster and maximizes response
time because the joins can be done in parallel.

Introduction to Transaction Management

➢ The concept of a transaction is used in database systems as a basic unit of consistent

and reliable computing.
➢ Thus queries are executed as transactions once their execution strategies are
determined and they are translated into primitive database operations.
➢ We differentiate between database consistency and transaction consistency:
➢ A database is in a consistent state if it obeys all of the consistency (integrity)
constraints defined over it. State changes occur due to modifications, insertions, and
deletions (together called updates). Of course, we want to ensure that the database
never enters an inconsistent state. Note that the database can be temporarily
inconsistent during the execution of a transaction. The important point is that the
database should be consistent when the transaction terminates.
➢ Transaction consistency refers to the actions of concurrent transactions. We would like
the database to remain in a consistent state even if there are a number of user requests
that are concurrently accessing (reading or updating) the database. Transaction
management deals with the problems of always keeping the database in a consistent
state even when concurrent accesses and failures occur.

A transaction
Model
 ❖ Definition of a Transaction

➢ A transaction is a unit of consistent and reliable computation. A transaction is

considered to be made up of a sequence of read and write operations on the database,
together with computation steps.
➢ A transaction takes a database, performs an action on it, and generates a new version
of the database, causing a state transition.

Example: This query can be specified, using the embedded SQL

notation, as a transaction by giving it a name (e.g.,
Consider the following SQL BUDGET UPDATE) and declaring it as follows:
query for increasing by 10% the
budget of the CAD/CAM project:

Example:
Let us consider a simplified version of a typical
reservation application, where a travel agent
Let us assume that there is a FLIGHT
enters the flight number, the date, and a
relation that records the data about
customer name, and asks for a reservation. The
each flight, a CUST relation for the
transaction to perform this function can be
customers who book flights, and an FC
implemented as follows, where database accesses
relation indicating which customers
are specified in embedded SQL notation:
are on what flights. Let us also assume
that the relation definitions are as
follows (where the underlined
attributes constitute the keys:
 Termination A transaction always terminates, even when there are failure. If
the transaction can complete its task successfully, we say that
Conditions of
the transaction commits. If, a transaction stops without
Transactions completing its task, we say that it aborts.

Abort Commit

When a transaction is ➢ The importance of commit is twofold:

aborted, its execution 1) The commit command signals to the DBMS that the effects
is stopped and all of its of that transaction should now be reflected in the database,
already executed thereby making it visible to other transactions that may
actions are undone by access the same data items.
returning the database 2) The point at which a transaction is committed is a “point of
to the state before no return.” The results of the committed transaction are
their execution. This is now permanently stored in the database and cannot be
also known as rollback. undone.

Example:

▪ Let us return to our reservation

system example. One thing we
did not consider is that there may
not be any free seats available on
the desired flight
▪ In this example the first SQL
statement gets the STSOLD and
CAP into the two variables temp1
and temp2. These two values are
then compared to determine if
any seats are available.
▪ The transaction either aborts if
there are no free seats, or
updates the STSOLD value and
inserts a new tuple into the FC
relation to represent the seat that
was sold.
 The consistency and reliability aspects of transactions are due to four
❖ Properties of properties: (1) atomicity, (2) consistency, (3) isolation, and (4)
Transactions durability. Together, these are commonly referred to as the ACID
properties of transactions.

① Atomicity ② Consistency

➢ Atomicity refers to the fact that a transaction is treated ➢ The consistency of a

as a unit of operation. Therefore, either all the transaction is simply its
transaction’s actions are completed, or none of them correctness. In other words,
are. This is also known as the “all-or-nothing property. a transaction is a correct
➢ Atomicity requires that if a transaction is interrupted program that maps one
by a failure, its partial results must be undone. consistent database state
➢ The activity of preserving the transaction's atomicity in to another.
presence of transaction aborts due to input errors, ➢ There are four levels of
system overloads, or deadlocks is called transaction consistency: Based on the
recovery. concept of dirty data, the
➢ The activity of ensuring atomicity in the presence of four levels are defined as
system crashes is called crash recovery. follows:

Degree 0 Transaction T does not overwrite dirty data of other transactions

T does not overwrite dirty data of other transactions
Degree 1
T does not commit any writes before EOT
T does not overwrite dirty data of other transactions
Degree 2 T does not commit any writes before EOT
T does not read dirty data from other transactions
T does not overwrite dirty data of other transactions
Degree 3 T does not commit any writes before EOT
T does not read dirty data from other transactions
Other transactions do not dirty any data read by T before T completes.

➢ The point in defining multiple levels of consistency is to provide application

programmers the flexibility to define transactions that operate at different levels.
➢ Consequently, while some transactions operate at Degree 3 consistency level, others
may operate at lower levels.

③ Isolation The next three phenomena are specified when

isolation is not maintained correctly:
Dirty Read: There are four levels
Dirty data refer of Isolation levels:
to data items
whose values
have been 1) Read Uncommitted:
modified by a For transactions
transaction that operating at this level,
has not yet In this example T2 read a value of item x all three phenomena
committed. which has been modified by uncommitted are possible.
transaction T2 2) Read Committed:
Non-repeatable Fuzzy reads and
or Fuzzy read: phantoms are
Two reads possible, but dirty
within the same reads are not.
transaction T1
3) Repeatable Read: Only
returns different
phantoms possible.
values
4) Anomaly Serializable:
Phantom T1 searches the database according to a None of the
predicate while T2 inserts new tuples that phenomena are
satisfy the predicate possible.

Durability refers to that property of transactions which ensures

④ Durability that once a transaction commits, its results are permanent and
cannot be erased from the database.

❖ Types of transactions

① Based on transaction ② Based on Organization

duration of read and write actions

A. Online (short-life transactions) A. Two-step: All the read actions are performed
• Very short execution time (seconds) before any write action
• Access to small portion of the B. Restricted: restricted so that a data item has to
database be read before it can be written
B. batch (long-life transactions) C. Action model: Which consists of the restricted
• Longer execution time (minutes or class with the further restriction that each
hours) (read, write) pair be executed together.
• Access to large portion of the
database Consider the following example:
 The following is how to define transaction T according to the previous transaction models:

Two step
Restricted
Action

Distributed Concurrency Control

❖ Concurrency Control

▪ The problem of synchronizing Anomalies:

concurrent transactions such that ▪ Lost updates: The effects of some transactions
the consistency of the database is are not reflected on the database.
maintained while, at the same time, ▪ Inconsistent retrievals: A transaction, if it reads
maximum degree of concurrency is the same data item more than once, should
achieved. always read the same value.

❖ Execution History (or Schedule)

▪ An order in which the operations of a

set of transactions are executed.
▪ A history (schedule) can be defined as a
partial order over the operations of a
set of transactions.

❖ Formalization of History Complete Schedule – Example

❖ Schedule Definition
A schedule is a prefix of a
complete schedule such
that only some of the
operations and only some
of the ordering
relationships are included

❖ Serial History

▪ All the actions of a transaction occur consecutively.

▪ No interleaving of transaction operations.
▪ If each transaction is consistent (obeys integrity
rules), then the database is guaranteed to be
consistent at the end of executing a serial history

❖ Serializable History

▪ Transactions execute concurrently, but the net effect of the resulting history upon the
database is equivalent to some serial history.
▪ Equivalent with respect to what?
▪ Conflict equivalence: the relative order of execution of the conflicting operations
belonging to unaborted transactions in two histories are the same.
▪ Conflicting operations: two incompatible operations (e.g., Read and Write) conflict if they
both access the same data item.
▪ Incompatible operations of each transaction is assumed to conflict; do not change their
execution orders.
▪ If two operations from two different transactions conflict, the corresponding transactions
are also said to conflict

❖ Serializability in
Distributed DBMS

▪ Somewhat more involved. Two

histories have to be considered:
 Local histories
 Global history
▪ For global transactions (i.e., ❖ Global Non-serializability
global history) to be serializable,
two conditions are necessary:
 Each local history should be
serializable.
 Two conflicting operations
should be in the same relative
order in all of the local histories
where they appear together.

❖ Concurrency Control Algorithms ❖ Locking-Based Algorithms

1) Pessimistic ▪ Transactions indicate their intentions by

a. Two-Phase Locking-based (2PL) requesting locks from the scheduler (called
- Centralized (primary site) 2PL lock manager).
- Primary copy 2PL ▪ Locks are either read lock (rl) [also called
- Distributed 2PL shared lock] or write lock (wl) [also called
b. Timestamp Ordering (TO) exclusive lock]
- Basic TO ▪ Read locks and write locks conflict (because
- Multiversion TO Read and Write operations are incompatible)
- Conservative TO
c. Hybrid
2) Optimistic
a. Locking-based ▪ Locking works nicely to allow concurrent
b. Timestamp ordering-based processing of transactions.

▪ A Transaction locks an object before using it.

❖ Two-Phase
▪ When an object is locked by another transaction, the requesting
Locking transaction must wait.
(2PL) ▪ When a transaction releases a lock, it may not request another lock.

❖ Strict
2PL
 ❖ Centralized 2PL

▪ There is only one 2PL scheduler in

the distributed system.
▪ Lock requests are issued to the
central scheduler.

❖ Distributed 2PL

▪ 2PL schedulers are placed at each site.

Each scheduler handles lock requests
for data at that site.
▪ A transaction may read any of the
replicated copies of item x, by obtaining
a read lock on one of the copies of x.
Writing into x requires obtaining write
locks for all copies of x.

❖ Distributed 2PL Execution ❖ Timestamp Ordering

❖ Conservative Timestamp Ordering

▪ Basic timestamp ordering tries to execute an operation as soon as it receives it

- progressive
- too many restarts since there is no delaying
▪ Conservative timestamping delays each operation until there is an assurance that it will
not be restarted
▪ Assurance?
- No other operation with a smaller timestamp can arrive at the scheduler
- Note that the delay may result in the formation of deadlocks
 ❖ Multiversion ❖ Optimistic Concurrency Control
Timestamp Algorithms
Ordering

▪ Transaction execution model: divide into subtransactions each of which execute at a site
Tij: transaction Ti that executes at site j
▪ Transactions run independently at each site until they reach the end of their read phases
▪ All subtransactions are assigned a timestamp at the end of their read phase
▪ Validation test performed during validation phase. If one fails, all rejected.

❖ Optimistic CC Validation Test

❖ Deadlock

▪ A transaction is deadlocked if it is
blocked and will remain blocked until
there is intervention.
▪ Locking-based CC algorithms may cause
deadlocks.
▪ TO-based algorithms that involve waiting
may cause deadlocks.
▪ Wait-for graph
- If transaction Ti waits for another
transaction Tj to release a lock on an
entity, then Ti → Tj in WFG.
 ❖ Local versus
Global WFG

❖ Deadlock
Management

▪ Ignore: Let the application programmer deal with it, or restart the system
▪ Prevention: Guaranteeing that deadlocks can never occur in the first place. Check
transaction when it is initiated. Requires no run time support.
▪ Avoidance: Detecting potential deadlocks in advance and taking action to insure that
deadlock will not occur. Requires run time support.
▪ Detection and Recovery: Allowing deadlocks to form and then finding and breaking them.
As in the avoidance scheme, this requires run time support.

❖ Deadlock Prevention
▪ Evaluation:
▪ All resources which may be needed by a transaction - Reduced concurrency due
must be predeclared. to preallocation
- The system must guarantee that none of the resources - Evaluating whether an
will be needed by an ongoing transaction. allocation is safe leads to
- Resources must only be reserved, but not necessarily added overhead.
allocated a priori - Difficult to determine
- Unsuitability of the scheme in database environment (partial order)
- Suitable for systems that have no provisions for - No transaction rollback or
undoing processes. restart is involved.

▪ Transactions are not required to request resources a priori.

▪ Transactions are allowed to proceed unless a requested
resource is unavailable.
❖ Deadlock ▪ In case of conflict, transactions may be allowed to wait for a
Avoidance fixed time interval.
▪ Order either the data items or the sites and always request
locks in that order.
▪ More attractive than prevention in a database environment.
 ❖ Wait-Die Algorithm ❖ Wound-Wait Algorithm

▪ Transactions are allowed to wait freely.

▪ Wait-for graphs and cycles.
❖ Deadlock ▪ Topologies for deadlock detection algorithms:
Detection - Centralized
- Distributed
- Hierarchical

❤ ‫أسأل هللا لي ولكم دوام التوفيق والنجاح‬