Discussion 10
Parallel Query Processing
Announcements
Agenda
Types of Parallelism
Types of Query Parallelism
Types of Intra-query Parallelism
Inter-operator Parallelism
Inter-operator Parallelism
Intra-operator Parallelism
Worksheet #1
What is the difference between inter- and intra- query parallelism?
Worksheet #1
What is the difference between inter- and intra- query parallelism?
Inter-query parallelism operates between multiple queries, rather than within a single query, whereas intra-query parallelism operates within a single query (parallelism of the operators that make up the query).
Partitioning Schemes
Network cost
Partitioning data
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 0 KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 0 KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 0 KB
Assume the size of each tuple is X KB.
Total network cost: 0 KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
7
9
2
12
11
m1
10
Assume the size of each tuple is X KB.
Total network cost: X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 2X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 2X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 2X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 3X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 3X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 4X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 4X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 5X KB
Example: Range Partitioning
m2
m3
Assign data to 3 partitions using range partitioning
[0, 4] → 1
[5, 9] → 2
[10, 14] → 3
2
10
7
9
2
12
11
m1
Assume the size of each tuple is X KB.
Total network cost: 5X KB
Example: Hash Partitioning
2
10
7
9
2
12
11
m1
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
m2
m3
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
2 % 3 = 2
2
10
7
9
2
12
11
m1
m2
m3
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
2 % 3 = 2
2
10
7
9
2
12
11
m2
m3
m1
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
10 % 3 = 1
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
10 % 3 = 1
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
7 % 3 = 1
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
7 % 3 = 1
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
2 % 3 = 2
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
2 % 3 = 2
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
12 % 3 = 0
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
11 % 3 = 2
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
11 % 3 = 2
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
9 % 3 = 0
10
7
9
2
12
11
m2
m3
m1
2
Example: Hash Partitioning
Assign data to 3 partitions using hash partitioning
h(k) = k % 3
10
7
9
2
12
11
m2
m3
m1
2
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
10
7
9
2
12
11
m1
m2
m3
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
10
7
9
2
12
11
m1
m2
m3
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
10
7
9
2
12
11
m1
m2
m3
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
7
9
2
12
11
m1
m2
m3
10
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
7
9
2
12
11
m1
m2
m3
10
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
9
2
12
11
m1
m2
m3
10
7
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
9
2
12
11
m1
m2
m3
10
7
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
9
2
12
11
m1
m2
m3
10
7
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
9
2
11
m1
m2
m3
10
7
12
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
9
2
11
m1
m2
m3
10
7
12
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
9
2
11
m1
m2
m3
10
7
12
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
9
2
11
m1
m2
m3
10
7
12
Example: Round Robin Partitioning
Assign data to 3 partitions using round robin partitioning
2
9
2
11
m1
m2
m3
10
7
12
Partitioning Comparison
Hash
hash function divides data
2
2
m1
m2
m3
Range�key divides data
by range
12
9
2
m1
m2
m3
10
2
7
2
9
2
11
m1
m2
m3
10
7
12
Round Robin
no key
10
7
9
12
11
11
Partitioning data
Worksheet #2
What are the advantages and disadvantages of organizing data by keys?
Worksheet #2
What are the advantages and disadvantages of organizing data by keys?
Advantages: because data is organized by keys, search and update operations (which require searching on the key) can be done more efficiently, since we have some sense of where the data must be (if it exists).
Disadvantages: we must maintain the organization, which adds overhead to insertions and updates.
Worksheet #3a
All of the data for a relation with N pages starts on one machine, and we would like to partition the data onto M machines.
How much data (in KB) would be sent over the network to partition the data through each of the following: range, hash, and round-robin partitioning?
Assume that the size of each page is S (in KB). Also, assume we use uniform hash functions and are able to construct ranges that have the same number of values in them.
Worksheet #3a
How much data (in KB) would be sent over the network to partition the data through each of the following: range, hash, and round-robin partitioning?
Worksheet #3b
If there are no assumptions about hash functions or data ranges, what is the best and worst case in terms of network I/O cost for the three partitioning schemes?
Worksheet #3b
If there are no assumptions about hash functions or data ranges, what is the best and worst case in terms of network I/O cost for the three partitioning schemes?
For round-robin partitioning, the best/worst case is still S ∗ N ∗ (M−1) / M KB, since the data will still be divided evenly.
For range and hash partitioning, the best case will be 0 KB, if all the data stays on the current machine. The worst case will be S∗N KB, in the scenario where the current machine retains no data and sends all of it to the other machines.
Parallel Query Operators
Parallel Sorting
Parallel Sort Merge Join
Parallel Hashing
Parallel Hash Joins
Phase 1: Partition R and S across different machines
Phase 2: Within each machine, perform local (grace) hash joins
Join Pipelining
Symmetric Hash Joins
Symmetric Hash Joins
Symmetric Hash Join Example:
B=7
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
1
2
3
1
2
3
Output
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
1
2
3
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Symmetric Hash Join Example:
B=7
1
2
3
1
2
3
Output
Assign colors to 3 partitions using our hash function:
{P} → 1, {B} → 2, {R} → 3
Table R partition 1
Table S
partition 1
Worksheet #4a
Given m=3 machines with B=5 buffer pages each, along with N=63 pages of data that don’t contain duplicates.
In the best case, what is the number of passes needed to sort the data? (Range/hash partitioning counts as 1 pass)
Worksheet #4a
With m=3 machines, B=5 buffer pages each, N=63 pages of data that don’t contain duplicates: in the best case, what is the number of passes needed to sort the data?
# of passes to sort the data: (# of passes to partition the data) + (# of passes to sort each partition)
1 pass: Range partitioning data (Best case: data is uniformly partitioned, so each machine gets 21 pages. )
3 passes: Executing external sorting on each machine
Overall: 4 passes
Worksheet #4a
With m=3 machines, B=5 buffer pages each, N=63 pages of data that don’t contain duplicates: in the best case, what is the number of passes needed to sort the data?
Generally, it will take 1 +⎡1 + logB-1 ⎡N/mB⎤⎤passes to sort the data using parallel sorting in the best case.
Worksheet #4b
What is the number of passes needed to hash the data?
Examine the best case scenario, where the data will be uniformly distributed under the given hash functions.
Worksheet #4b
What is the number of passes needed to hash the data? Examine the best case scenario, where the data will be uniformly distributed under the given hash functions.
1 pass: Hash partitioning data (in best case, each machine gets 21 pages)
3 passes: Executing external hashing on each machine
Overall: 4 passes
Worksheet #4c
Assume that relation R has R pages of data, and relation S has S pages of data. If we have m machines with B buffer pages each, what is the number of passes in order to perform sort merge join (in terms of R, S, m, and B) in the best case?
Consider reading over either relation to be a pass.
Worksheet #4c
Assume that relation R has R pages of data, and relation S has S pages of data. If we have m machines with B buffer pages each, what is the number of passes in order to perform sort merge join (in terms of R, S, m, and B) in the best case?
Consider reading over either relation to be a pass.
(1 pass/table to partition across machines) + (number of passes needed to sort R) + (number of passes to sort S) + (1 final merge sort pass, going through both tables)
2 + ⌈1 + logB-1⌈R/(mB)⌉⌉ + ⌈1 + logB-1⌈S/(mB)⌉⌉ + 2
Worksheet #4d
Can you use pipeline parallelism to implement this join (sort merge join from part c)?
Worksheet #4d
Can you use pipeline parallelism to implement this join (sort merge join from part c)?
No, the sorting pass must complete before the merge pass can begin.
Parallel Aggregation
Parallel Aggregation
Asymmetric Shuffles
Asymmetric Shuffles
Broadcast joins
Worksheet #5a
Relation R has 10,000 pages, round-robin partitioned across 4 machines (M1, M2, M3, M4).
Relation S has 10 pages, all of which are only stored on M1. We want to join R and S on the condition R.col = C.col.
Assume the size of each page is 1 KB.
What type of join would be best in this scenario, and why?
Worksheet #5a
Relation R has 10,000 pages, round-robin partitioned across 4 machines (M1, M2, M3, M4).
Relation S has 10 pages, all of which are only stored on M1. We want to join R and S on the condition R.col = C.col.
Assume the size of each page is 1 KB.
What type of join would be best in this scenario, and why?
Broadcast join, because it is cheaper to send relation S to every machine rather than to partition R based on col.
Worksheet #5b
Relation R has 10,000 pages, round-robin partitioned across 4 machines (M1, M2, M3, M4).
Relation S has 10 pages, all of which are only stored on M1. We want to join R and S on the condition R.col = C.col.
Assume the size of each page is 1 KB.
How many KB of data must be sent over the network to join R and S?
Worksheet #5b
Relation R has 10,000 pages, round-robin partitioned across 4 machines (M1, M2, M3, M4).
Relation S has 10 pages, all of which are only stored on M1. We want to join R and S on the condition R.col = C.col.
Assume the size of each page is 1 KB.
How many KB of data must be sent over the network to join R and S?
The amount of data sent over the network is the amount of data required to send all pages of S to every machine that does not have it (M2, M3, and M4): 3 * 10 = 30 KB
Worksheet #5c
Relation R has 10,000 pages, round-robin partitioned across 4 machines (M1, M2, M3, M4).
Relation S has 10 pages, all of which are only stored on M1. We want to join R and S on the condition R.col = C.col.
Assume the size of each page is 1 KB.
Would the amount of data sent over the network change if R was hash partitioned among the 4 machines rather than round-robin partitioned? What if R was instead range partitioned? Assume each machine contains at least 1 tuple of R after partitioning.
Worksheet #5c
Would the amount of data sent over the network change if R was hash partitioned among the 4 machines rather than round-robin partitioned? What if R was instead range partitioned? Assume each machine contains at least 1 tuple of R after partitioning.
Attendance Link