Intro to Database Systems (15-445/645)

21 Introduction to
Distributed Databases

FALL Andy
2022 Pavlo
 2

ADMINISTRIVIA

Project #4 is due Sun Dec 11th @ 11:59pm

→ Zoom Q&A Session TONIGHT @ 8:00pm

Homework #5 is due Sun Dec 4th @ 11:59pm

15-445/645 (Fall 2022)

 3

PA R A L L E L V S . D I S T R I B U T E D

Parallel DBMSs:
→ Nodes are physically close to each other.
→ Nodes connected with high-speed LAN.
→ Communication cost is assumed to be small.

Distributed DBMSs:
→ Nodes can be far from each other.
→ Nodes connected using public network.
→ Communication cost and problems cannot be ignored.

15-445/645 (Fall 2022)

 4

DISTRIBUTED DBMSs

Use the building blocks that we covered in single-

node DBMSs to now support transaction
processing and query execution in distributed
environments.
→ Optimization & Planning
→ Concurrency Control
→ Logging & Recovery

15-445/645 (Fall 2022)

 5

TO DAY ' S AG E N DA

System Architectures
Design Issues
Partitioning Schemes
Distributed Concurrency Control

15-445/645 (Fall 2022)

 6

S Y S T E M A RC H I T E C T U R E

A distributed DBMS's system architecture specifies

what shared resources are directly accessible to
CPUs.

This affects how CPUs coordinate with each other

and where they retrieve/store objects in the
database.

15-445/645 (Fall 2022)

 7

S Y S T E M A RC H I T E C T U R E

Network
Network
Network

Shared Shared Shared Shared

Everything Memory Disk Nothing

15-445/645 (Fall 2022)

 8

SHARED MEMORY

CPUs have access to common

memory address space via a fast Network
interconnect.
→ Each processor has a global view of all the
in-memory data structures.
→ Each DBMS instance on a processor must
"know" about the other instances.

15-445/645 (Fall 2022)

 9

SHARED DISK

All CPUs can access a single logical

disk directly via an interconnect, but
each have their own private
Network
memories.
→ Can scale execution layer independently
from the storage layer.
→ Must send messages between CPUs to
learn about their current state.

15-445/645 (Fall 2022)

 10

SHARED DISK EXAMPLE

Node
Page ABC Storage
Get Id=101

Application
Server Node

15-445/645 (Fall 2022)

 10

SHARED DISK EXAMPLE

Node
Storage

Get Id=102
Page XYZ
Application
Server Node

15-445/645 (Fall 2022)

 10

SHARED DISK EXAMPLE

Node
Storage

Get Id=101 Page ABC

Node

Application
Server Node

15-445/645 (Fall 2022)

 10

SHARED DISK EXAMPLE

Node
Page ABC Storage
Update 101

Node

Application
Server Node

15-445/645 (Fall 2022)

 14

S H A R E D N OT H I N G

Each DBMS instance has its own Network

CPU, memory, and local disk.
Nodes only communicate with each
other via network.
→ Harder to scale capacity.
→ Harder to ensure consistency.
→ Better performance & efficiency.

15-445/645 (Fall 2022)

 15

S H A R E D N OT H I N G E X A M P L E
Node
P1→ID:1-150

Get Id=200

Application
Server Node
P2→ID:151-300

15-445/645 (Fall 2022)

 15

S H A R E D N OT H I N G E X A M P L E
Get Id=100 Node
Get Id=200 P1→ID:1-150

Get Id=200 Id=200

Application
Server Node
P2→ID:151-300

15-445/645 (Fall 2022)

 15

S H A R E D N OT H I N G E X A M P L E
Node
P1→ID:1-100

Node
P3→ID:101-200
Application
Server Node
P2→ID:201-300

15-445/645 (Fall 2022)

 18

E A R LY D I S T R I B U T E D DATA B A S E S Y S T E M S

MUFFIN – UC Berkeley (1979)

SDD-1 – CCA (1979)
System R* – IBM Research (1984) Stonebraker Bernstein

Gamma – Univ. of Wisconsin (1986)

NonStop SQL – Tandem (1987)
Mohan DeWitt

Gray
15-445/645 (Fall 2022)
 19

DESIGN ISSUES

How does the application find data?

Where does the application send queries?
How to execute queries on distributed data?
→ Push query to data.
→ Pull data to query.
How does the DBMS ensure correctness? Next Class
How do we divide the database across resources?

15-445/645 (Fall 2022)

 20

H O M O G E N O U S V S . H E T E RO G E N O U S

Approach #1: Homogenous Nodes

→ Every node in the cluster can perform the same set of
tasks (albeit on potentially different partitions of data).
→ Makes provisioning and failover "easier".

Approach #2: Heterogenous Nodes

→ Nodes are assigned specific tasks.
→ Can allow a single physical node to host multiple "virtual"
node types for dedicated tasks.

15-445/645 (Fall 2022)

 21

M O N G O D B H E T E RO G E N O U S A RC H I T E C T U R E
Shards (mongod)
Router
(mongos) P1 P2
Get Id=101
Router
(mongos)
⋮
P3 P4
Application
Server
Config Server P1→ID:1-100
(mongod) P2→ID:101-200
P3→ID:201-300
⋮ P4→ID:301-400
15-445/645 (Fall 2022)
 22

DATA T R A N S PA R E N C Y

Applications should not be required to know

where data is physically located in a distributed
DBMS.
→ Any query that run on a single-node DBMS should
produce the same result on a distributed DBMS.

In practice, developers need to be aware of the

communication costs of queries to avoid
excessively "expensive" data movement.

15-445/645 (Fall 2022)

 23

DATA B A S E PA RT I T I O N I N G

Split database across multiple resources:

→ Disks, nodes, processors.
→ Often called "sharding" in NoSQL systems.

The DBMS executes query fragments on each

partition and then combines the results to produce
a single answer.

The DBMS can partition a database physically

(shared nothing) or logically (shared disk).

15-445/645 (Fall 2022)

 24

N A Ï V E TA B L E PA RT I T I O N I N G

Assign an entire table to a single node.

Assumes that each node has enough storage space
for an entire table.

Ideal if queries never join data across tables stored

on different nodes and access patterns are uniform.

15-445/645 (Fall 2022)

 25

N A Ï V E TA B L E PA RT I T I O N I N G

Table1 Table2 Partitions

Table1

Ideal Query: Table2

SELECT * FROM table

15-445/645 (Fall 2022)

 26

V E RT I C A L PA RT I T I O N I N G
CREATE TABLE foo (
Split a table's attributes into separate attr1 INT,
partitions. attr2 INT,
attr3 INT,
Must store tuple information to attr4 TEXT
reconstruct the original record. );

Tuple#1 attr1 attr2 attr3 attr4

Tuple#2 attr1 attr2 attr3 attr4
Tuple#3 attr1 attr2 attr3 attr4
Tuple#4 attr1 attr2 attr3 attr4

15-445/645 (Fall 2022)

 26

V E RT I C A L PA RT I T I O N I N G
CREATE TABLE foo (
Split a table's attributes into separate attr1 INT,
partitions. attr2 INT,
attr3 INT,
Must store tuple information to attr4 TEXT
reconstruct the original record. );

Partition #1 Partition #2
Tuple#1 attr1 attr2 attr3 Tuple#1 attr4
Tuple#2 attr1 attr2 attr3 Tuple#2 attr4
Tuple#3 attr1 attr2 attr3 Tuple#3 attr4
Tuple#4 attr1 attr2 attr3 Tuple#4 attr4

15-445/645 (Fall 2022)

 29

H O R I ZO N TA L PA RT I T I O N I N G

Split a table's tuples into disjoint subsets based on

some partitioning key and scheme.
→ Choose column(s) that divides the database equally in
terms of size, load, or usage.

Partitioning Schemes:
→ Hashing
→ Ranges
→ Predicates

15-445/645 (Fall 2022)

 30

H O R I ZO N TA L PA RT I T I O N I N G

Partitioning Key
Table1 Partitions
101 a XXX 2022-11-29 hash(a)%4 = P2
102 b XXY 2022-11-28 hash(b)%4 = P4 P1 P2
103 c XYZ 2022-11-29 hash(c)%4 = P3
104 d XYX 2022-11-27 hash(d)%4 = P2
105 e XYY 2022-11-29 hash(e)%4 = P1

Ideal Query: P3 P4
SELECT * FROM table
WHERE partitionKey = ?

15-445/645 (Fall 2022)

 31

LO G I C A L PA RT I T I O N I N G
Node Id=1
Id=2 Storage
Get Id=1

Id=1
Id=2
Id=3
Application Id=4
Server Node
Id=3
Id=4

15-445/645 (Fall 2022)

 31

LO G I C A L PA RT I T I O N I N G
Node Id=1
Id=2 Storage

Id=1
Id=2
Get Id=3 Id=3
Application Id=4
Server Node
Id=3
Id=4

15-445/645 (Fall 2022)

 31

LO G I C A L PA RT I T I O N I N G
Node Id=1
Id=2 Storage

Id=1
Get Id=3 Id=2
Get Id=2
Id=3
Application Id=4
Server Node
Id=3
Id=4

15-445/645 (Fall 2022)

 34

P H Y S I C A L PA RT I T I O N I N G
Node
Id=1
Get Id=1 Id=2

Get Id=3
Application
Server Node
Id=3
Id=4

15-445/645 (Fall 2022)

 35

H O R I ZO N TA L PA RT I T I O N I N G

Partitioning Key
Table1 Partitions
101 a XXX 2022-11-29 hash(a)%4 = P2
102 b XXY 2022-11-28 hash(b)%4 = P4 P1 P2
103 c XYZ 2022-11-29 hash(c)%4 = P3
104 d XYX 2022-11-27 hash(d)%4 = P2
105 e XYY 2022-11-29 hash(e)%4 = P1

Ideal Query: P3 P4
SELECT * FROM table
WHERE partitionKey = ?

15-445/645 (Fall 2022)

 35

H O R I ZO N TA L PA RT I T I O N I N G

Partitioning Key
Table1 Partitions
101 a XXX 2022-11-29 hash(a)%5 = P4
102 b XXY 2022-11-28 hash(b)%5 = P3 P1 P2
103 c XYZ 2022-11-29 hash(c)%5 = P5
104 d XYX 2022-11-27 hash(d)%5 = P1
105 e XYY 2022-11-29 hash(e)%5 = P3

Ideal Query: P3 P4
SELECT * FROM table
WHERE partitionKey = ?

15-445/645 (Fall 2022)

 37

CONSISTENT HASHING
10 hash(key1)

P1

P3

P2

15-445/645 (Fall 2022)

0.5
 37

CONSISTENT HASHING
10 hash(key1)

P1

P3

hash(key2)
P2

15-445/645 (Fall 2022)

0.5
 37

CONSISTENT HASHING
10 hash(key1)

P1

P3

hash(key2)
P2

15-445/645 (Fall 2022)

0.5
 37

CONSISTENT HASHING
10

P1

P3 If hash(key)=P4

P2
P4

15-445/645 (Fall 2022)

0.5
 37

CONSISTENT HASHING
10
P5
P1

P3

P6

P2
P4

15-445/645 (Fall 2022)

0.5
 37

CONSISTENT HASHING
10
P5 Replication Factor = 3
P1

P3

P6

P2
P4

15-445/645 (Fall 2022)

0.5
 37

CONSISTENT HASHING
10 hash(key1)
P5 Replication Factor = 3
P1

P3

P6

P2
P4

15-445/645 (Fall 2022)

0.5
 37

CONSISTENT HASHING
10 hash(key1)
P5 Replication Factor = 3
P1

P3

P6

P2
P4

15-445/645 (Fall 2022)

0.5
 45

SINGLE-NODE VS. DISTRIBUTED

A single-node txn only accesses data that is

contained on one partition.
→ The DBMS may not need check the behavior concurrent
txns running on other nodes.

A distributed txn accesses data at one or more

partitions.
→ Requires expensive coordination.

15-445/645 (Fall 2022)

 46

T R A N S AC T I O N C O O R D I N AT I O N

If our DBMS supports multi-operation and

distributed txns, we need a way to coordinate their
execution in the system.

Two different approaches:

→ Centralized: Global "traffic cop".
→ Decentralized: Nodes organize themselves.

15-445/645 (Fall 2022)

 47

T P M O N I TO R S

A TP Monitor is an example of a centralized

coordinator for distributed DBMSs.
Originally developed in the 1970-80s to provide
txns between terminals and mainframe databases.
→ Examples: ATMs, Airline Reservations.

Standardized protocol from 1990s: X/Open XA

15-445/645 (Fall 2022)

 48

C E N T R A L I Z E D C O O R D I N ATO R

Coordinator
Lock Request Partitions

P1 P2

Application
Server P3 P4

15-445/645 (Fall 2022)

 48

C E N T R A L I Z E D C O O R D I N ATO R
P1
Coordinator P2
Lock Request P3 Partitions
P4

P1 P2

Application
Server P3 P4

15-445/645 (Fall 2022)

 48

C E N T R A L I Z E D C O O R D I N ATO R
P1
Coordinator P2
Lock Request P3 Partitions
P4

Acknowledgement P1 P2

Application
Server P3 P4

15-445/645 (Fall 2022)

 48

C E N T R A L I Z E D C O O R D I N ATO R
P1
Coordinator P2
Commit Request P3 Partitions
P4

P1 P2

Application Safe to commit?

Server P3 P4

15-445/645 (Fall 2022)

 48

C E N T R A L I Z E D C O O R D I N ATO R
P1
Coordinator P2
Commit Request P3 Partitions
P4

Acknowledgement P1 P2

Application Safe to commit?

Server P3 P4

15-445/645 (Fall 2022)

 48

C E N T R A L I Z E D C O O R D I N ATO R
P1
Coordinator P2
Commit Request P3 Partitions
P4

Acknowledgement P1 P2

Application Safe to commit?

Server P3 P4

15-445/645 (Fall 2022)

 54

C E N T R A L I Z E D C O O R D I N ATO R

Partitions

Middleware
Commit
Query Requests
Request Safe to commit?
P1 P2

Application P1→ID:1-100
Server P2→ID:101-200 P3 P4
P3→ID:201-300
P4→ID:301-400

15-445/645 (Fall 2022)

 55

D E C E N T R A L I Z E D C O O R D I N ATO R

Leader Node Partitions

Begin Request P1 P2

Application
Server P3 P4

15-445/645 (Fall 2022)

 55

D E C E N T R A L I Z E D C O O R D I N ATO R

Leader Node Partitions

P1 P2

Query Request
Application
Server P3 P4

15-445/645 (Fall 2022)

 55

D E C E N T R A L I Z E D C O O R D I N ATO R

Leader Node Partitions

Commit Request P1 P2

Safe to commit?
Application
Server P3 P4

15-445/645 (Fall 2022)

 58

D I S T R I B U T E D C O N C U R R E N C Y C O N T RO L

Need to allow multiple txns to execute

simultaneously across multiple nodes.
→ Many of the same protocols from single-node DBMSs
can be adapted.
This is harder because of:
→ Replication.
→ Network Communication Overhead.
→ Node Failures.
→ Clock Skew.

15-445/645 (Fall 2022)

 59

DISTRIBUTED 2PL

Set A=2 Set B=7

Application Application
Server Server

A=1 B=8

NETWORK
Node 1 Node 2
15-445/645 (Fall 2022)
 59

DISTRIBUTED 2PL

Set A=2 Set B=7

Application Application
Server Server

A=2
A=1 B=7
B=8

NETWORK
Node 1 Node 2
15-445/645 (Fall 2022)
 59

DISTRIBUTED 2PL

Set A=2 Set B=7

Application Application
Server Set B=9 Set A=0 Server

A=2
A=1 B=7
B=8

NETWORK
Node 1 Node 2
15-445/645 (Fall 2022)
 59

DISTRIBUTED 2PL
Waits-For Graph

Set A=2 Set B=7

T1 T2
Application Application
Server Set B=9 Set A=0 Server

A=2
A=1 B=7
B=8

NETWORK
Node 1 Node 2
15-445/645 (Fall 2022)
 63

CONCLUSION

I have barely scratched the surface on distributed

database systems…

It is hard to get this right.

15-445/645 (Fall 2022)

 64

P RO J E C T # 4 – C O N C U R R E N C Y C O N T RO L

You will add support for concurrent

transactions using two-phase locking
in BusTub!
→ Deadlock Detection
→ Hierarchical Locking (Table, Tuple)
→ Multiple Isolation Levels
→ Aborts/Rollbacks

You do not need to worry about

Prompt: A dramatic and vibrant painting of a giant
eye in the clouds looking down on a field of grazing
sheep with padlocks as their heads.

logging txns to disk.

[Link]
15-445/645 (Fall 2022)
 65

P RO J E C T # 3 – TA S K S

Lock Manager
→ Maintain internal lock table and queues.
→ Track the growing/shrinking phases of txns.
→ Notify waiting txns when their locks are available.

Deadlock Detector:
→ Build the waits-for graph and deterministically identify
what txn to kill off to break deadlocks

Execution Engine
→ Modify Project #3 executors to support txn requests.
15-445/645 (Fall 2022)
 66

P RO J E C T # 3 - L E A D E R B OA R D

We have designed the Terrier benchmark to

measure who has the fastest BusTub
implementation!

Tasks:
→ UpdateExecutor
→ Predicate Pushdown

15-445/645 (Fall 2022)

 67

T H I N G S TO N OT E

Do not change any file other than the ones that

you submit to Gradescope.

Make sure you pull in the latest changes from the

BusTub main branch.

Post your questions on Piazza or come to TA

office hours.

Compare against our solution in your browser!

15-445/645 (Fall 2022)
 68

P L AG I A R I S M WA R N I N G

Your project implementation must be

your own work.
→ You may not copy source code from other
groups or the web.
→ Do not publish your implementation on
Github.

Plagiarism will not be tolerated.

See CMU's Policy on Academic
Integrity for additional information.

15-445/645 (Fall 2022)

 69

NEXT CLASS

Distributed OLTP Systems

Replication
CAP Theorem
Real-World Examples

15-445/645 (Fall 2022)