Distributed Database Systems

1. Why distributed?

Week 1 講過 Scale up（買更勁的一部機）同 Scale out（加多幾部機）。Distributed database 就係 scale out 用喺 database 上面。Slide 2 列出四個原因：

2. Distributed database system 與 Location transparency

Distributed database system：Host the database on multiple machines (nodes)。Goal：As if it were a single, centralized database，即係 user 用起上嚟要好似只得一個 database，唔使理 data 實際放喺邊部機。

Example concern：Location transparency，user 唔需要知道 data 喺邊個 location。Slide 4 的 idea：Map the domain name to different locations (IP addresses) depending on the requester's location。例如同一個 db.example.com，香港 user 解析到香港 node 的 IP，歐洲 user 解析到歐洲 node 的 IP（即 DNS-based routing）。User 只記一個名，系統自動帶佢去最近的 site。

3. Challenges

Designing, implementing, and maintaining distributed systems is inherently difficult。三大挑戰：

4. Fragmentation and Replication

Fragmentation is also known as Sharding：Partitioning a table into smaller, non-overlapping subsets。設計時要決定：

• What data to store（點樣切）
• Where to store it（放喺邊個 site）
• How many copies to keep（Replication，見 4.4）

4.1 Types of Fragmentation

4.2 Example data（slides 8–10）

Table T（10 個 employees）：

EmployeeID	Name	Department	Salary	StartDate	IsFullTime
101	Alice Johnson	Sales	$60,000	2022-08-15	True
102	Bob Smith	Marketing	$75,000	2021-05-20	True
103	Charlie Brown	Engineering	$95,000	2023-01-10	True
104	Diana Prince	HR	$55,000	2024-03-01	False
105	Ethan Hunt	Engineering	$110,000	2020-11-25	True
106	Fiona Glenanne	Sales	$62,000	2023-09-01	False
107	George Costanza	Marketing	$70,000	2022-04-12	True
108	Hannah Baker	HR	$58,000	2024-01-20	True
109	Ian Malcolm	Engineering	$105,000	2021-07-01	True
110	Jessica Day	Sales	$59,000	2023-05-18	False

Horizontal Fragmentation（slide 9）：按 rows 切開兩份。Fragment 1 = EmployeeID 101–105（$\text{EmployeeID} \le 105$），Fragment 2 = 106–110（$\text{EmployeeID} > 105$）。每個 fragment 都有齊全部 6 個 columns。

Vertical Fragmentation（slide 10）：按 columns 切開兩份。Fragment A = (EmployeeID, Name, Department, Salary)，Fragment B = (EmployeeID, StartDate, IsFullTime)。每個 fragment 都有齊 10 個 rows。

-- Horizontal fragmentation：按條件切 rows
CREATE TABLE T_site1 AS SELECT * FROM T WHERE EmployeeID <= 105;
CREATE TABLE T_site2 AS SELECT * FROM T WHERE EmployeeID > 105;
-- 重組：UNION（fragments 不重疊，用 UNION ALL 已足夠）
SELECT * FROM T_site1 UNION ALL SELECT * FROM T_site2;

-- Vertical fragmentation：按 columns 切，每份都要保留 key
CREATE TABLE T_A AS SELECT EmployeeID, Name, Department, Salary FROM T;
CREATE TABLE T_B AS SELECT EmployeeID, StartDate, IsFullTime FROM T;
-- 重組：JOIN on key
SELECT * FROM T_A JOIN T_B ON T_A.EmployeeID = T_B.EmployeeID;

4.3 Example queries：HF 定 VF 好？（slide 11）

4.4 Comparisons：HF vs VF 優點（slide 12）

4.5 Trade-offs of Replication（slide 13）

5. Overview of Distributed Query Processing

Goal：Transform a high-level query (e.g., SQL) into a series of local operations and inter-site data transfers that minimize cost。

5.1 Steps of Distributed Query Processing（slide 15）

5.2 Example 1：WHERE EmployeeID = 107（slides 16–17）

用返 HF 例子：Site 1 放 $\text{EmployeeID} \le 105$，Site 2 放 $\text{EmployeeID} > 105$。User 對住 global view（成個 T）落 query：

SELECT * FROM T WHERE EmployeeID = 107

• Query Decomposition / Localization：107 > 105 → Only Site 2 is relevant，只將 query send 去 Site 2
• Local Execution：Site 2 執行，得到 (107, George Costanza, Marketing, $70,000, 2022-04-12, True)
• Result Reconstruction：Not necessary as there is only one local result

5.3 Example 2：Sales 的 full-time 員工（slide 18）

SELECT Name, Salary FROM T
WHERE Department = 'Sales' AND IsFullTime = True

• Query Decomposition：條件同 EmployeeID 無關，冇辦法排除任何 site → Send the query to both sites
• Local Execution：Site 1 → (Alice Johnson, $60,000)；Site 2 → Fiona（False）同 Jessica（False）都唔係 full-time → 空
• Result Reconstruction：Use “Union” to get the final result → 只得 Alice Johnson, $60,000

6. Distributed transactions

Challenge：A transaction that involves data located at more than one site。Example：Transferring money between two accounts held at different sites（A 戶口喺 Site 1 扣錢、B 戶口喺 Site 2 加錢）。

Goal：Ensuring all sites agree on the outcome (commit or abort) despite network partitions or site failures。即係唔可以 Site 1 扣咗錢但 Site 2 冇加到（違反 Atomicity）。

7. Two-phase Locking (2PL) in distributed environment

Purpose：to achieve ACID in RDBMS。重溫 Week 3：growing phase 只攞 lock、shrinking phase 只放 lock；Readlock 之間 no conflict，涉及 Writelock 就 conflict。問題係：data 有幾個 copies 分散喺唔同 node，邊個負責批 lock？ 三種做法：

以下三個例子都用同一個 setting：transaction T1 = Read(x), Write(y)；x 放喺 Node 1, 3, 7；y 放喺 Node 2, 5, 8。

（已對照 slides 21、23、25 嘅箭咀圖，下面步驟同圖一致。Slide 21：1 Readlock(x) 同 3 Writelock(y) 都指向 Centralized node；2 Read(x) 指向 Node 1；4 Write(y) 指向 Node 2、Node 5、Node 8。Slide 23：1 Readlock(x) 同 2 Read(x) 都指向 Node 3；3 Writelock(y) 指向 Node 8；4 Write(y) 指向 Node 8、Node 2、Node 5。Slide 25：1 Readlock(x) 指向 Node 3 同 Node 1；2 Read(x) 指向 Node 1；3 Writelock(y) 指向 Node 2 同 Node 5；4 Write(y) 指向 Node 8、Node 2、Node 5。）

7.1 Centralized 2PL（slides 21–22）

假設 Node 1 同 Node 2 最近 user，另有一個 Centralized node 做 lock manager：

1. Readlock(x) → 問 Centralized node
2. Read(x) → 去最近的 copy（Node 1）讀
3. Writelock(y) → 問 Centralized node
4. Write(y) → 寫晒 y 的所有 copies：Node 2, 5, 8（write 要更新所有 replicas）

Q: What is the read cost if no locking is required? 只需直接去最近的 Node 1 讀 x，一個來回就搞掂。呢種「先唔鎖，照做」的做法叫 optimistic concurrency control（Let's go without locking）。加咗 locking 就要多一程去 Centralized node。

7.2 Primary copy 2PL（slides 23–24）

x 的 manager 係 Node 3，y 的 manager 係 Node 8：

1. Readlock(x) → 問 x 的 primary copy Node 3
2. Read(x) → Read from the primary copy only（Node 3）；other copies serve as backup
3. Writelock(y) → 問 y 的 primary copy Node 8
4. Write(y) → 寫 Node 8 同其他 copies（Node 2, 5）

7.3 Distributed 2PL（slides 25–26）

假設 Node 1, 2, 3, 5 比較近。x 有 3 個 copies，majority = 2；y 都係：

1. Readlock(x) → 問最近的兩個 x copies：Node 1 同 Node 3（超過一半 = 2/3 grant 就得）
2. Read(x) → 喺最近的 copy 讀（Node 1）
3. Writelock(y) → 問最近的兩個 y copies：Node 2 同 Node 5
4. Write(y) → 寫晒 Node 2, 5, 8

7.4 三種 2PL 一覽

7.5 Deadlock detection（slide 27）

Except Centralized 2PL, deadlock detection becomes complex。Example：

• T1 is waiting for T2 at Site 1 (item x)
• T2 is waiting for T1 at Site 2 (item y)

Site 1 的 local wait-for graph 只有 T1 → T2；Site 2 只有 T2 → T1。No single site is able to realize the deadlock loop without communication：要將兩邊合併成 global wait-for graph，先見到 cycle T1 → T2 → T1。

If communication is required，就有新問題：When do the sites communicate?（幾耐交換一次？太密 = network 負擔，太疏 = deadlock 拖好耐先發現）What data is exchanged?（成個 local graph？定只係部分 edges？）

8. Timestamp-based concurrency control

Modern high-performance distributed database systems prefer time-based methods。

Requirement：Each transaction gets a unique timestamp。兩種派 timestamp 的方法：

• Centralized：A single site is responsible to assign a timestamp to each transaction（又係 single point of failure / bottleneck）
• Distributed：Combine local timestamp with unique site identifier

$\|$ 係 concatenation（駁埋）。例如 Site 3 喺 local time 1005 開始一個 transaction，Site 7 都喺 1005 開始一個：兩個 TS 分別係 1005‖3 同 1005‖7，因為 Site_ID 唔同，所以一定 unique。Local timestamp 放前面（high-order），所以比較先睇時間，時間一樣先睇 Site_ID。

8.1 Wound-Wait Protocol（slide 29）

When T1 requests a lock on a data item that is currently held by T2：

• If T1 is older (smaller timestamp) → wound：Kill T2 (T2 rollbacks)
• If T1 is younger (larger timestamp) → wait

As the older never waits for a younger, a deadlock can never happen。

8.2 Wait-Die Protocol（slide 30）

Same rule（older has priority）。When T1 requests a lock held by T2：

• If T1 is older (smaller timestamp) → wait
• If T1 is younger (larger timestamp) → die（T1 rollbacks）

As the younger never waits for an older, a deadlock can never happen。

8.3 Comparisons（slide 31）

9. Timestamp Ordering (T/O) Concurrency Control

Each transaction is assigned a timestamp；each data item X is associated with two timestamp values：

9.1 T/O Concurrency Rules（slides 33–34）

Let $t$ be the timestamp of the transaction.

9.2 Worked example（自擬）

開始時 $RTS(X) = 5$，$WTS(X) = 3$。依次處理：

Operation	檢查	結果	$RTS(X)$	$WTS(X)$
READ(X), $t=4$	$4 < 3$? No	OK	$\max(5,4)=5$	3
READ(X), $t=2$	$2 < 3$? Yes	Die	5	3
WRITE(X), $t=4$	$4 < RTS=5$? Yes	Die	5	3
WRITE(X), $t=6$	$6 < 5$? No；$6 < 3$? No	OK，寫入	5	6
WRITE(X), $t=5$	$5 < 5$? No；$5 < 6$? Yes	Ignore（Thomas Write Rule）	5	6
READ(X), $t=7$	$7 < 6$? No	OK	7	6

9.3 Analysis of T/O（slide 35）

Cascading abort：T2 讀咗 T1 寫的值，之後 T1 被 rollback，咁 T2 讀到的係「從未正式存在」的值，T2 都要 abort；如果 T3 又讀咗 T2 寫的值，就一路連鎖落去。

10. Multiversion Timestamp Ordering (MVTO) 與 MVCC

Instead of a single copy of the data, MVTO keeps all the versions of the data（每個 version 標住寫入佢的 transaction timestamp）。

10.1 Example（slides 37–38）

X 的 versions：

t	value
1	10
3	30
5	20
8	15

10.2 Analysis of MVTO（against Basic T/O）（slide 39）

10.3 Multi-Version Concurrency Control (MVCC)（slides 40–42）

• MVCC used by many modern database systems：Oracle、PostgreSQL、MySQL (InnoDB)
• MVCC in distributed database systems：BrightDB、CockroachDB、Google Cloud Spanner、TiDB、YugabyteDB、MongoDB、CouchDB

Example：MVCC in PostgreSQL（slide 42）：

• MVCC handles read-write conflicts（reader 讀舊 version，唔使等 writer）
• Locking is used for write-write conflicts
• Locking is also applied to：Schema changes（the entire table is locked）；Explicit record locking（the SELECT FOR UPDATE statement）

-- PostgreSQL：明確鎖住要改的 record，防止其他 transaction 同時修改
BEGIN;
SELECT * FROM account WHERE id = 1 FOR UPDATE;
UPDATE account SET balance = balance - 100 WHERE id = 1;
COMMIT;

11. Raft algorithm：Consensus

Purpose：Handle failure in a distributed environment → Fault-tolerance protocol → Increased Availability（Ongaro & Ousterhout, 2014, “In Search of an Understandable Consensus Algorithm”）。Existing databases using Raft：BrightDB、MongoDB、TiDB、YugabyteDB。

11.1 Raft mechanism（slide 44）

• A single Raft group is responsible for a single partition/shard
• Key idea：There is a leader of the nodes；the nodes elect one unique leader；other nodes are followers

11.2 Handling request (normal situation)（slide 45）

例如一個 group 有 1 個 leader + 2 個 followers（共 3 個 nodes）：leader 自己加一個 follower ack 就係 2/3 majority，即可宣佈完成；就算第三個 node 壞咗都唔影響。

11.3 Detecting problems（slide 46）

• Heartbeat：The leader periodically sends heartbeats to all followers to maintain authority
• When the leader is found “dead”（followers 收唔到 heartbeat）→ a new leader is elected
• When a failed node recovers, it synchronizes data from the group

12. Distributed database features：CAP

12.1 Consistency in CAP vs ACID（slide 48）

12.2 CAP Theorem（slide 49）

12.3 Category of database systems（slide 50）

12.4 Which system is CP / AP?（slide 51）

In fact, many systems can be configured to adapt to either goal! Example：MongoDB

13. From ACID to BASE (for AP)

Not every application requires ACID。Discussion：Is it a big deal when a transaction reads an old value? Example scenario：application 由 database summarize 新聞，生成一篇「summary news」。如果漏咗幾條最新新聞，問題大唔大？其實唔大，下次再 summarize 就會包埋。

14. NewSQL system

NewSQL：A new class of modern distributed RDBMS。The aim is to achieve：

• High scalability
• Distributed ACID（essentially CP）
• Use the familiar SQL language

自我測驗 Self-test

Q1. 用 slide 8 的 employee table：(a) SELECT Name, StartDate FROM T WHERE EmployeeID = 103；(b) SELECT AVG(Salary) FROM T。各自 HF 定 VF 較好？點解？

(a) HF 較好：用 EmployeeID 可以 localize 到 Site 1（$103 \le 105$），一個 site 就攞到成條 record。VF 下 Name 喺 Fragment A、StartDate 喺 Fragment B，要讀兩個 fragments 再 JOIN。(b) VF 較好：只需 Salary 一個 column，但要掃晒所有 rows。VF 只讀 (EmployeeID, Name, Department, Salary) fragment，唔使讀其他 columns；HF 要兩個 site 都掃全部 columns，再合併兩邊的 sum 同 count。

Q2. 點解 HF 的 reconstruction 比 VF 簡單？VF 每個 fragment 一定要包含咩？

HF 用 UNION（將 rows 疊埋），VF 要用 JOIN（按 key 配對 rows），JOIN 貴好多。VF 每個 fragment 都要包含 primary key（例如 EmployeeID），否則冇辦法 JOIN 返原本的 table。

Q3. 列出 Replication 的 3 個 benefits 同 2 個 costs。

Benefits：Availability（一個 copy 壞咗仲有其他）、Reduced Latency（讀最近的 copy）、Parallelism（read queries 可以 parallel）。Costs：Update Overhead（每次 update 要 propagate 去所有 copies，network traffic 增加）、Concurrency Control（保持所有 copies 一致好複雜，涉及 Consistency vs. Availability）。

Q4. Distributed query processing 的 dominant cost 係咩？列出四個步驟。

Network communication（latency and volume），所以目標係 minimize data transfer。四步：Query Decomposition → Distributed Optimization and Localization → Local Execution → Result Reconstruction。

Q5. x 放喺 Node 1, 3, 7。用 Distributed 2PL 讀 x，最少要幾多個 nodes grant readlock？相比 Centralized 2PL 有咩好處同壞處？

超過一半：3 個 copies 要 2 個 grant。好處：冇 single point of failure，improved availability；可以揀近的 nodes，避免 manager 好遠。壞處：messages 大幅增加、implementation 複雜、deadlock detection 困難。Centralized 2PL 則容易 detect deadlock、最易實作，但有 congestion、scale 受 central node 限制、single point of failure。

Q6. 點解除咗 Centralized 2PL，distributed deadlock detection 都好難？用 slide 例子解釋。

T1 喺 Site 1 等 T2（item x），T2 喺 Site 2 等 T1（item y）。每個 site 的 local wait-for graph 都只有一條 edge，冇 cycle；要合併兩個 site 的資料先見到 T1 → T2 → T1。所以 sites 之間要 communicate，又引出「幾時溝通」同「交換咩 data」的問題。Centralized 2PL 所有 lock 資訊都喺一個 node，所以容易。

Q7. $TS(T_A) = 10$，$TS(T_B) = 20$。$T_B$ 持有 X 的 lock，$T_A$ 申請 X 的 lock。分別用 Wound-Wait 同 Wait-Die，會發生咩？反過來（$T_A$ 持有、$T_B$ 申請）又點？

$T_A$ 較 old（timestamp 細）。$T_A$ 申請：Wound-Wait → $T_A$ wound $T_B$，$T_B$ rollback；Wait-Die → $T_A$ wait。$T_B$ 申請：Wound-Wait → $T_B$（younger）wait；Wait-Die → $T_B$ die（rollback）。兩種 protocol 被 rollback 的都係 younger 的 $T_B$。

Q8. Wound-Wait 同 Wait-Die 邊個 abort rate 較低？邊個 abort 較平？點解？

Wound-Wait abort rate 較低，但 abort 較貴：被 wound 的 transaction 可能已經做咗好多嘢。Wait-Die abort rate 較高，但 abort 較平：younger 喺一開始申請 lock 時就 die，浪費的工作少。兩者都可以 prevent deadlock，但都有 unnecessary abort；2PL 冇 unnecessary abort，但可能 deadlock。

Q9. Basic T/O：$RTS(X) = 8$，$WTS(X) = 6$。依次處理 (a) WRITE(X) $t=7$；(b) READ(X) $t=5$；(c) WRITE(X) $t=9$；(d) WRITE(X) $t=8$。

(a) $7 < RTS = 8$ → Die（younger 已讀咗 X）。(b) $5 < WTS = 6$ → Die（younger 已寫咗 X）。(c) $9 \ge 8$ 而且 $9 \ge 6$ → 寫入，$WTS(X) = 9$。(d) $8 < RTS = 8$? No；$8 < WTS = 9$? Yes → Thomas Write Rule，ignore（do nothing）。最後 $RTS = 8$，$WTS = 9$。

Q10. MVTO：X 的 versions 有 (t=1, 10)、(t=3, 30)、(t=5, 20)、(t=8, 15)。READ(X) $t=4$ 同 $t=9$ 各讀到咩？如果 last read timestamp = 6，WRITE(X) $t=5$ 會點？

$t=4$ → $\le 4$ 最新的係 t = 3 → 30。$t=9$ → t = 8 → 15。WRITE $t=5$：$5 < 6$，timestamp 6 的 transaction 已經讀咗 → abort。（Read never aborts，但 write abort 仍然存在，呢個就係 MVTO 的 weakness。）

Q11. CAP 的 C 同 ACID 的 C 有咩分別？點解 traditional centralized DBMS 叫做 CA？

CAP 的 C 講 data freshness：每次 read 都攞到最新的 write（或 error）。ACID 的 C 講 data correctness：transaction 保持 database 喺 consistent state。Centralized DBMS 冇 network partition 的問題（唔係 distributed，冇 P），所以可以同時有 C 同 A。

Q12. MongoDB 點樣可以 configure 成 CP 或 AP？AP setting 有咩風險？

CP：write majority setting，writes 要去 majority sites；majority 溝通唔到時系統 unavailable。AP：minimal write setting，只需 primary site acknowledge；primary 斷線時，另一個 site 可能頂上做新 primary，但佢手上係舊 data，transaction 可能讀到舊值，所以 ACID is not guaranteed。

Q13. BASE 三個字代表咩？舉一個適合同一個唔適合 BASE 的應用。

Basically Available（盡量用所有 nodes 提供讀寫，但 read 可能唔係最新）、Soft-state（未收斂前只能以某個概率知道 state）、Eventually consistent（系統運作夠耐後，之後所有 reads 回傳同一個值）。適合：新聞摘要，漏幾條新聞問題唔大。唔適合：bank account transactions，必須用 ACID。

Q14. Raft 的目的係咩？佢會唔會分散 database workload？描述 normal request 的流程。

目的係 fault tolerance / handle failure，提高 availability。佢唔會分散 workload，leader 仲要負擔額外的 synchronization 工作。流程：client 只同 leader 溝通 → leader 處理 request → update 傳去 followers → 收到 majority acknowledgement 先正式宣佈完成。Leader 定期發 heartbeat；followers 發現 leader「死咗」就選新 leader；恢復的 node 從 group 同步 data。

一頁總結 Cheat sheet