CockroachDB: The Resilient Geo-Distributed SQL Database

一直以來對CockroachDB(CRDB for short)的設計和實作很感興趣，最近抽時間研究了下，發現其在技術上還是領先了同類NewSQL産品不少的，個人感覺應該是目前最為先進的類Spanner分布式資料庫系統，是以這篇文章會盡可能詳細的讨論下其系統的多個方面，重點是事務和一緻性相關。

paper中針對的是v.19.2.2版本，不過官方文檔中是基于最新的v.21.1.7，兩者在描述上有一些沖突的地方，而官方文檔中會更為詳盡些，是以本文的很多介紹将盡量将paper與官方reference結合，并以reference為準。

介紹

随着時代發展，大型跨國企業的事務型工作負載開始出現跨地理位置的趨勢，同時他們也追求對資料放置位置的細粒度控制以及高性能。一般其需求有如下幾點：

1. 遵循所在地區的資料本地化合規要求，同時盡量保證資料的就近通路以提供高性能
2. 為使用者提供高可用的服務，容忍哪怕是region級的失效
3. 為了簡化上層應用的開發，提供SQL的操作接口和可串行化的事務語義

CRDB是面向具有全球級使用者的企業或組織，基于雲平台提供具有擴充性、高可用性、強一緻性和高性能的OLTP事務型資料庫。如其名字，其具有很強的容災和自動恢複能力。為了滿足以上提到的使用者需求，它提供了以下幾個主要功能：

1. 容錯和高可用： 通過多副本（通常是3-replicas)提供容錯能力，通過自動化的快速恢複實作高可用能力
2. 跨地理位置的分區政策和副本放置政策： CRDB本身是share-nothing架構，可以自動實作水準擴充，它内部基于一些啟發式規則來決定資料的放置方式，使用者也可以為資料設定分區方案，并從分區粒度上控制資料的放置位置
3. 高性能事務： CRDB的事務協定非常嚴格且對性能做了大量優化，支援跨分區事務和可串行化的隔離級别并且不依賴任何特殊的硬體，隻需要正常伺服器和基于軟體的時鐘同步協定，是以可以做到跨雲部署。
4. 有先進的query optimizer和query execution engine，此外一個成熟的資料庫産品所需要的輔助功能如 online schema change / backup and restore / fast import / JSON support等都有支援。

系統架構

本身是share-nothing的架構，每個node同時提供存儲和計算能力。在任一節點内，系統實作了一種分層的架構如下：

• SQL層

負責查詢優化和執行，對下層的讀寫請求則轉換為KV操作，一般情況下，在SQL層看來，下層是一個單體的KV存儲（一些情況下會暴露出分布特性）

• Transaction層

對上層提供事務語義的保證，對到達的KV操作提供隔離性和并發控制等

• Distribution層

對上層提供一個全局單調遞增的key space的抽象，所有的資料包括system data/user data都在這個key space内。CRDB采用了range partition的政策，連續的key range以64MB左右為機關進行切分，每個range也就是複制和遷移的基本資料單元（等同于TiKV中的Region）。

從key到Range的映射在system data中，是一個2層的index，同時在每個節點上都有一份cache，友善快速從key定位對應range。

64MB是一個适當的大小，遷移時不會産生大量快速，同時也可以提供不錯的range scan的資料局部性。 和TiKV一樣，range會根據size自動做split和merge，同時也可以split來避免熱點或存儲的不均衡。

• Replication層

基于Raft完成多副本的共識，每個range構成一個raft group，複制的内容是kv請求執行的結果，基于command形成RSM。

在每個range内，存在一個leaseholder的概念，應該和raft leader是基本等價的，但不同的是leader如果想避免通過raft log做一緻性讀取，需要實作raft算法中的lease read功能，這個leaseholder就是為了這個功能而存在，隻有leaseholder才能進行一緻性讀，并發起寫操作。

在raft協定中，一個raft group中的前後lease間隔必須保證不會重疊，在CRDB中，這一點通過HLC來實作，具體後面請看這篇單獨的文章

通過多副本CRDB實作了容錯能力：

1. 如果是短時間失效例如leader突然crash，raft group會自然選舉出新leader并持續提供服務，重新開機後的原leader會rejoin到group中，并變為follower并追上必要的update logs。
2. 如果是長時間失效，CRDB可以自動的補全under-replicated的Ranges，而新建立replica的放置政策也可以有多種：

1. 1. 手動指定： 對于每個節點可以添加必要的“标簽”：系統配置，所在位置等，在建立table時，可以在schema中加入特殊的"region"列，并建立為分區表，這樣對不同分區可以映射到不同的region中。
   2. 自動決定： CRDB内部會基于一系列啟發式規則，總體上是保證系統負載的均衡。

• Storage層

實作單node内的本地kV存儲，目前使用的是RocksDB。

關于資料的放置政策，paper中具體給出了3種常用政策，基本上都是在性能和容錯能力間做trade-off：

1. Geo-Partitioned Replicas

一個partition内的replica的複制不跨region，這樣一個range的所有副本都固定在一個region内，這樣可以符合資料本地化的合規要求，讀寫性能也會不錯，但無法抵抗region level的失效

例如上圖，同一個region内的所有replica都保留在該region内。

2. Geo-Partitioned Leaseholders

一個partition内的leader都固定在某個region内，但其他replica可以跨range複制，這樣可以有不錯的local read性能，但需要cross-region write，同時也提供了跨region的容災能力

3. Duplicated Indexes

在CRDB中，所有的資料都以索引形式來組織，是以和MySQL類似，主表也是基于primary key的聚簇索引。如果讀負載比較重，可以将二級索引在多個region間建立多份copy，這樣各個region内都可以利用本地index做快速的讀取，缺點則是寫放大（多份index），是以比較适合資料很少變更的情況。

事務

事務是CRDB中最為亮眼的功能，在跨地域部署的前提下，它依然提供了serializable的最強隔離級别，以及近乎linearizability的線性一緻性保障，同時還有不錯的性能，可以說是相當牛逼。

首先介紹下CRDB為了事務的延遲和吞吐，做的幾個主要的優化：

• Write Pipelining

每個針對KV的寫請求到達range leader時，leader會在本地先做evaluation，确定這個request在資料上會産生怎樣的結果，這個結果就是要在多副本間複制的内容，一旦計算完成，就立即向“client”端(gateway node，接收使用者請求的節點）傳回一個叫做provisional ACK的東西，同時異步執行複制，gateway受到ACK後會發送後續的請求，這樣請求的本地執行就和結果的複制并行了起來，形成pipeline。盡可能減少了操作延遲。

當然，gateway node作為事務的coordinator會跟蹤所有in-flight（沒複制完成）的操作集合。

• Parallel commit

最直接的方式，當所有in-flight的寫操作都複制完成時，coordinator發起commit的送出操作并完成複制，此時事務就算是送出完成了，但這意味着更多的round-trip和更大的延遲。為此CRDB采用了一個比較激進的方案：

當最後一個write操作回複provisional ACK後，coordinator立即發起請求，修改事務狀态為staging + [所有pending write集合]。然後進入wait狀态等待所有pending write操作都完成，一旦确認都完成就直接向用戶端告知事務已送出，然後異步的将事務狀态改為committed

這樣來看，實際的write操作和commit操作都形成了流水線，可以很好的減少複制的延遲，理論上如果寫足夠快，2輪的複制延遲（一輪是所有write操作，一輪是staging狀态）内，事務就可以完成送出，當然缺點也很明顯，由于staging -> commit的狀态是異步修改的，有可能其他并發事務在檢視該事務狀态是看到的仍是staging，那麼需要去确定所有pending的write opertions是否已經完成，如果完成就等同于commit狀态。

實際上，通過這種原子性修改事務狀态的方式，也實作了分布式事務的原子送出，所有寫入資料的可見性也是原子性的switch。

CRDB的事務隻支援可串行化的隔離級别，采用了MVTO的并發控制方式，每個事務被配置設定唯一的時間戳，事務基于時間戳的順序建立在串行化曆史中的先後順序，也就是說，一個事務所有的讀寫操作，都在這個時間戳上“原子性瞬時”完成。

關于事務的部分我會更多參考官方文檔的介紹，按照架構層次從上到下介紹整體流程：

• SQL

client的請求首先到達叢集中某個node，稱為gateway node，它負責SQL的解析優化等，并在execution engine中轉換為對于的KV操作，注意這裡沒有range的概念。

• Transaction

事務層會在gateway中建立TxnCoordSender，它負責整個事務的執行，TxnCoordSender基于gateway本地的時間戳為事務設定初始ts(既是read ts也是commit ts)。

事務層會把執行層下發的類似put/get的KV請求，打包為BatchRequest，發送到下層。

為了記錄事務狀态，TxnCoordSender會在事務的第一個write操作所在的range上寫入一條額外的記錄transaction record。這個記錄保持了事務的狀态，變化序列是pending -> staging -> committed/aborted。

同時為了維持事務的活躍性，TxnCoordSender會周期性的下發heartbeat到transaction record。

注：

CRDB對于transaction record的處理有一個lazy的優化，即隻在必要時才寫入事務狀态記錄，由于這隻是個優化，具體細節不再講解，有興趣可以參考

CRDB documents

• Distribution

分布層的處理仍然在gateway node上，會建立DistSender，針對上層下發的BatchRequest，利用前面提到的key -> range的二級索引結構将BatchRequest拆解為針對各個range的BatchRequest，然後并行的向所有range的leader下發，如前面所說的write pipeling，一個BatchRequest一旦收到provisional ACK就立即下發下一個。

• Replication

Range leader在收到request後要依次執行以下操作:

• 檢測rw沖突

為了避免rw沖突，需要檢查目前的write ts是否小于目标key的最近read ts，如果小于則違反了事務的串行化順序（晚的讀操作已經發生，早的寫操作還沒有進行），這是不能允許的，這樣也就保證了

1. 任何已讀的曆史都不會被更改
2. 讀總是擷取最近的已送出版本。

為了能夠檢測該沖突，每個range會維護一個timestamp cache，記錄每個key的最近read ts。每個對key的讀操作都會更新該timestamp cache。同時timestamp cache還負責一件事：記錄一個事務的時間戳是否被"push"過，這個後面會介紹。

2. 加讀寫latch

到latch manager中擷取該key的latch，注意latch不是事務鎖，隻負責規避對同一key的并發操作沖突，操作完成後（複制完成）即可釋放，這樣保證了一個邏輯對象的讀、寫操作是依次完成的。

3. 對請求本地執行(evaluate)

無論請求是read/write，都要先到storage layer對并發事務的write intent做沖突檢測。

write intent是事務尚未送出的寫操作，在事務送出前，它寫入的KV中除了正常的MVCC value外，還包含一個pointer，指向該事務的transaction record，這樣其他事務可以從write intent擷取到事務狀态，如果沒有該pointer，則認為這是一個普通的多版本value，版本由其commit ts決定。

write intent起到了預寫入值 + 獨占鎖的意義，這裡鎖是指事務鎖，任一時刻隻有一個事務能夠寫入write intent，代表了對資料的最新寫入！

假設txnB遇到txnA的write intent，處理流程如下:

txnB讀取/寫入，遇到A的intent，到transaction record上判斷事務的狀态：

• commited，則這已經是一個普通value，幫助消除其pointer資訊
• aborted，忽略該intent并幫助删除
• pending，需要判斷事務活躍性

• • 事務已不活躍（與TxnCoordSender的heartbeat沒有更新），視為aborted處理
  • 事務活躍，這裡要看下write intent的時間戳

• • • write intent的時間戳小于txnB ts

• • • • 如果是read，這裡需要等待，因為read是要保證讀取最近已送出資料的，是以如果跳過該intent讀取更早版本，可能忽略掉這個本應該看到的送出。
      • 如果是write，也需要等待因為無法判斷write intent事務最終會以哪個時間戳送出（可能被向後push)，如果被push到了一個比txnB的ts更晚的時間戳，則txnB目前本質上是應該執行的，是以要等待來判斷會不會出現這種情況。

• • • write intent的時間戳大于txnB ts

• • • • 如果是read，忽略該intent，讀取更早版本資料
      • 如果是write，則目前事務abort掉，以更大的時間戳restart

• staging，先驗證事務是否已送出，如果已送出按commited處理，否則按照pending處理

遇到write intent的情況在CRDB中稱為transaction conflict，在處理完conflict後，之前遇到的intent将不複存在，看下對非write intent的處理：

• read

1. 讀到ts更大的已送出value，跳過
2. 讀到比ts更小的已送出value，則可以讀取
3. 如果已送出value的wts和目前事務的read ts有不确定區間的重合（有偏差的HLC時間戳），則無法确定該value是否可見，向後push讀事務的時間戳到不确定區間之後，然後讀取該資料。

這種向後push時間戳的政策實際是對abort -> restart的一種優化，為了保證事務的原子性，在commit ts被push後，要做一個read refresh的操作：

假設commit ts從ts1被推到ts2，需要驗證[ts1 -> ts2]時間範圍内，之前事務所有已經讀取的值，是否發生了變化！這個從直覺上很好了解，如果值發生了變化，在ts1上發生的事務與ts2上發生的事務就不再等價了，那麼對于已完成的寫操作，為什麼不需要驗證呢？其實也很簡單，因為寫入的都是write intent，而intent本身就代表了對key的最新操作，是獨占的，不會有其他事務有更新的write intent或者write，是以不必驗證。

對于這種由于時間戳的不确定區間而後推事務的情況，read refresh操作是立即進行，驗證成功才能繼續後續操作。

• write

前面已經介紹了rw沖突的檢測，如果發現write ts比timestamp cache中的最近read ts更大，則沒有問題，如果更小，則目前write事務的時間戳要被後推到比read ts更大。

如果遇到的已送出value的ts比目前write ts更小，則沒有問題，正常覆寫。

如果已送出value的ts比目前write ts更大或者兩者處于不确定區間，則類似wr沖突的情況，目前事務的ts要被後推到大于已送出value的ts。

這裡提到的push操作，并不是立即做read refreshing檢測的，而是在事務發起commit時完成（為何不立即檢測？?）

paper中對于事務的協調算法和leader上對操作的處理算法，如下兩圖所示：

可以看到pipelining write的處理和如果出現push(第10行）的refresh驗證

處理流程遵循了上面介紹的get latch -> evaluate -> replicate -> release latch的流程，但沒有展開具體evaluate的過程，不過evaluate傳回的資訊中表明了可能發生時間戳的push。

• commit

1. 首先判斷自己是否被abort了（被更高優先級事務abort或者一度不再活躍），如果abort則結束并清理。
2. 如果判斷自己被push過，則執行read refreshing，成功則繼續，否則事務abort并重新開機。
3. 以上兩個檢查通過後，則進入快速commit流程：

DistSender擷取各個BatchRequest的ACK後，會向TxnCoordSender彙總所有的in-flight write操作和所有讀操作的結果，TxnCoordSender在向transaction record記錄staging狀态時，會把in-flight writes同時記錄下來。當等待确定所有in-flight write複制完成後，向client傳回送出成功。然後異步改變transaction狀态為commit，并對write intent做cleanup，清理Pointer使其稱為普通的MVCC value。

到此主要的事務處理流程已經介紹完了，總的來看，處理流程是比較标準的，先檢測了rw沖突，然後是wr/ww沖突，當并發沖突解決後，再考慮timestamp ordering算法中不允許出現的過晚讀/過晚寫問題，進而保證各個事務對于一個key的操作是依次串行執行，且實體執行順序與事務的時間戳保持一緻，也就是說，對于單key事務來說，CRDB可以提供線性一緻性的最強保證。

時鐘同步與一緻性模型

關于CRDB的HLC-based時鐘機制，以及它所能提供的一緻性保證，在單獨一篇文章中進行了詳細的分析

這裡就不再贅述了，官方有篇不錯的文章也很值得一看：

https://www.cockroachlabs.com/blog/consistency-model/

關于SQL層paper中介紹的不多，總的來說，其對外提供了PostgreSQL的方言和通信協定。

• Query Optimizer

采用了Cascades的優化器架構，并利用DSL定義了一系列的transformation rules，例如：

這樣一條DSL定義了match pattern 和 replace pattern，描述如果operator tree中的算子滿足箭頭左側形式，則可以轉換為箭頭右側的形式。CRDB是用go寫的，DSL也會編譯成go code。

transformation rules分為兩類

1. Normalization rules，也就是重寫，認為這類轉換是一定要執行的，原來的plan tree不再保留，CRDB中目前有290+的Normalization rule
2. Exploration rules，不一定會産生更有計劃，例如join ordering，join method等，這種轉換會保留原始plan tree，并基于cost選擇更優plan，CRDB中目前有29條Exploration rule

實作了統一的search算法，會交錯去apply兩種rules，直到探索了所有轉換。

前面也提到，SQL層一般是不感覺下層的分布資訊的，看做一個單體的KV store，但在optimize時有一些特殊情況：

有些情況下可以利用schema中的partition資訊進行一些特定優化，如：存在idx(region, id)，可以靜态改寫query

SELECT * From t where id = 5;
=>
SELECT * From t where id = 5 AND (region = 'east' or region = 'west');      

使其可以利用上這個idx。

或者在存在duplicate index時，在考慮不同index副本的通路cost時，考慮其分布資訊。

• Query Planning and Execution

執行有兩種模式

1. gateway mode : 所有SQL計算都在gateway節點内完成，這種适合掃描資料量較小的情況，很多TP應用的查詢屬于這類
2. distributed mode：也就是MPP模式，在這種模式下，會有一個專門的physical planning階段，将optimizer産生的單機計劃，轉換為分布式的DAG plan，其中要覺得算子的并行執行方式，以及資料分發方式，類似下圖：

關于CRDB的優化器，youtube上有個不錯的視訊，是本paper的一作Rebecca Taft在cmu的talk，大家有興趣可以看下：

CockroachDB's Query Optimizer (Rebecca Taft, Cockroach Labs)

在每個data stream的内部，CRDB還支援兩種不同的執行模式，根據input cardinality和plan complexity決定：

https://www.youtube.com/watch?v=wHo-VtzTHx0

1. row-at-a-time，經典的iterator模型，支援所有的SQL算子計算
2. vector-at-a-time，向量化執行模型，支援部分SQL算子，當data從KV中讀取出來後，要先轉換為column format的vector，流經各算子後，再發送給end user之間轉換回行格式，每個算子的實作支援了selection vector的輸入。

• Schema change

CRDB的online schema change參考了F1的方案，具體細節paper中沒有講，可以參考

F1的paper

，大概來說，每個schema的變更被拆解為一系列的versions，通過控制叢集在任一時間點上，任兩個node一定處于兩個相鄰versions之間，則可以允許叢集的各個node在不同時間點異步的完成向新schema的轉換，同時仍可對外提供服務。

經驗總結

從2015年成立以來，CRDB Lab在多年的設計、開發中總結了一系列的經驗或者收獲，概述如下：

• Raft Made Live

雖然raft的paper給出了比較明确、詳細的實作細節，但工程實踐中仍然有很多可以優化的點：

1. 将同一個node上所有range leader向follower的heartbeat統一，進而大量減少無數range之間heartbeat帶來的網絡開銷
2. Joint Consensus，raft paper中原始的實作方案在執行group成員變更中有個限制，每次隻能增加或移除1個成員，這樣就會在過程中存在兩種情況：

上面是先做移除，可以看到group内隻有2個member，無法保證quorum

下面是先做增加，group内變為了4個member，這是的quorum變為了3，一旦region3失效了，quorum将被打破。

是以兩種中間狀态都存在可用性問題。

為此CRDB實作了Join Consensus機制，也就是說，在group成員變更時，需要同時滿足old / new兩種configuration，任何寫操作都要同時在兩個raft group下達成quorum才能成功。

具體細節可以參考CRDB官網：

https://www.cockroachlabs.com/blog/joint-consensus-raft/

值得一提的是，TiDB5.0的release中，也實作了Joint Consensus。

• 去除Snapshot Isolation

CRDB事務的設計初衷就是基于MVTO的可串行化事務，支援SI對于他們來說開發成本很高，帶來的性能收益卻沒有那麼大，因為需要對每個read操作加事務鎖(?為啥？)，對原有設計破壞的比較厲害，是以他們放棄了對SI的支援。

• Version Upgrades的坑

在做滾動更新中，有可能一個raft group中的多個node處于不同的軟體版本，在早期實作中複制的是request，然後各個replica各自執行evaluate并apply結果，但這樣可能導緻不同replica得到不同執行結果（不同版本二進制），是以後續改為了先在leader做evaluate，然後複制結果。

• Follow the workload

CRDB提供這種功能，希望能夠讓leader自适應的跟随user位置的變化來調整自身的locality，進而始終保持較低的通路延遲，但實踐證明這個方案很少被使用，使用者決定對于特定負載進行手動的調優并固化下來已經可以達到非常好的效果，而且這種自适應的政策在通用系統中也很難表現良好，經常無法保證性能的可預期性，而這對于使用者來說是至關重要的。

總結

在我來看，CRDB在NewSQL的OLTP型資料庫系統中已經領先了其他競品不少，通過大量的激進優化和開發積累，它建立了很多技術優勢：

1. 比較先進的query optimizer和executor，支援mpp的并行執行方式，單個執行流中也可以執行向量化的執行。
2. 先進的事務模型，提供了最為嚴格的隔離級别和幾乎接近完美的一緻性模型，同時還能利用大量優化來保證足夠好的性能和擴充性，現在看來除了對熱點并發事務的處理不夠理想外，其擴充性和事務吞吐、延遲都還是很不錯的
3. 自動維護能力，包括熱點打散，負載均衡，自動恢複機制等，減輕了運維負擔
4. 跨地理位置部署的能力和靈活的資料放置政策

類似的系統如YugabyteDB，TiDB，甚至Kudu，雖然各自實作了其中的一些功能，甚至實作的更好，但綜合來看，作為一款分布式大規模OLTP資料庫産品，短時間内應該沒有哪款産品可以超越。