Distributed Execution of Relational
Operators
Kamal Karlapalem
Distributed Database
• Very large databases cannot fit in one machine or one site
• Data is generated in distributed places and stored locally
• Data can can also be distributed to many sites before processing for
parallel execution
• Distributed database
• Logically integrated – global schema – as global relations – virtual
• Physically stored and managed by different database management systems
Relation distributed
• Horizontal Partition, Sharding, …
• R is partitioned rows wise –
• R is R1, R2,… Rk – k fragments, each fragment has all columns of R
• Example Employee(Eno, Name, Salary, Location), fragmented as EmployeeHyd(Eno,
Name, Salary Location) where location is Hyderabad, and EmployeeMum(Eno, Name,
Salary, Location( where location is Mumbai
• The above is example of range based, or semantics-oriented partitioning,
meaning all rows of a fragment satisfy a select condition
Relation distributed
• Horizontal Partition, Sharding, …
• Other is random partitioning, each row is assigned a random site. Thus, each
site has random set of rows from R
• Hash based partitioning – use a hash function to determine which site a row
goes to.
• To get all rows of R – all the rows at all the sites must be accessed and
a union is to be done.
• R = R1 ∪ R2 ∪ … ∪ Rk
Select σ operator
• σc(R) selects rows of R that satisfy condition c
• σc(R) = σc(R1 ∪ R2 ∪ … ∪ Rk)
= σc(R1) ∪ σc(R2) ∪ … ∪ σc(Rk)
• Hence each of the selects can be done on each of Ri’s and then
an union of results is got.
• But some σc(Rj) need not be done if we know before that Rj
does not have a row that satisfied condition c
Select σ operator
• σc(R) = σc(R1) ∪ σc(R2) ∪ … ∪ σc(Rk)
• Range partitioning – if c has the attribute used in range partitioning,
then some Rj need not be accessed if range of Rj does not overlap the
range of condition c.
• Hash and Random partitioning – the rows that satisfy condition can
be hashed to all the partitions, hence no reduction all R1 to Rk have to
be accessed
Why distributed execution σc(R)
• When R is very large and there are no indices on attributes in c
sequential scan is the only choice. But it takes time.
• Hence partition R quickly to many sites and execute the scan in
parallel.
• R could be an intermediate result or original relation.
• There can be further processing of selected rows which make cost of
the required partitioning to be amortized
Join operator
• Let R at site 1 and S at site 2 be two relations with common attribute
A and we want to compute R⨝R.A=S.A S
• How to compute the R⨝R.A=S.A S?
• Shift R to site 2
• Shift S to site 1
• Shirt R and S to site 3
• Perform local join R⨝R.A=S.A S (assuming there is pace for R and S in
that site) in the corresponding site using the local best join algorithm
Join operator
• Let R at site 1 and S at site 2 be two relations with common attribute
A and we want to compute R⨝R.A=S.A S
• SemiJoin
• Move ⌅A(S) – join column of S is sent to site 1 where R is there.
• Perform ⌅Attrs(R)[R ⨝R.A=S.A ⌅A(S)]- determine those rows of R which will join
with S at site 1
• Move ⌅Attrs(R)[R ⨝R.A=S.A ⌅A(S)] and perform R⨝R.A=S.A S at site 2.
Join Operator
• Let R at site 1 and S at site 2 be two relations with common attribute
A and we want to compute R⨝R.A=S.A S
• Use hashing or range based partitioning, same functions on attributes
A of R and S, and get R1, R2, … Rk, and S1, S2, …, Sk
• Note Ri joins with Si and no other Sj, and same with Si – joins with Ri
and no other Rj
Join Operator
• How to join R1, R2, … Rk, and S1, S2, …, Sk?
• Ship Ri and Si to a site p to execute Ri⨝R.A=[Link]
• Result is ∪i=1,…,k(Ri⨝R.A=[Link])
• In case multiple compute engines are available, each can execute one
Ri⨝R.A=[Link]
• Sometimes R and S are deliberately partitioned and sent to different
sites for faster parallel join execution. This depends on resources
available and the size of the relations.
⌅AttrList(R)
• If attribute list attributes are in one vertical fragment of R, then it is a local
operation.
• If attribute list attributes are in multiple vertical fragments, then join of
these vertical fragments is needed
• If the relation is horizontally fragmented, the project is done on each
horizontal fragment and its union is taken: ∪i=1,…,k[⌅AttrList(Ri)]
• For hybrid fragments, we need to joins and unions depending on how
many vertical fragments the attributes are in.