Zenn
📑

[論文紹介] TiDB:a Raft-based HTAP database

2021/02/24に公開
DB
tech

今回はTiDB（PingCAP）

久しぶりに論文紹介シリーズの第2弾である。
今回は分散DBのど真ん中、PingCAPが開発しているTiDBについての論文「TiDB:a Raft-based HTAP database」(VLDB2020)を紹介する。
この論文に関しては、PingCAP社が自身のブログでも解説している。

TiDBとは

念のため、TiDBとは何なのかを触れておこう。
一言でいうと、「MySQL互換のNewSQL（＝分散SQLデータベース）」である。

NewSQLとは何かについての説明は今回記事では省略するが、過去に書いたこちらの入門編やこちらの詳解編に解説をしている。

TiDBはMySQLと互換性を持つだけでなく、、今回の論文に示されているように、OLAP用途の機能強化を行っており、これもまたMySQLの弱点を補強する良い方向性と言える。この辺りのNewSQLの機能強化については、他製品も含めてこちらにまとめているのでOLAP機能以外にもどのような開発が行われているかを確認してもらいたい。

なお、MySQLにはOracleからHeatwaveという機能が出てきており、これは当論文のRelated Workで紹介されているOracle Database In-Memoryに近い構成を取るものとして、分析機能の強化で注目されている。

この論文を読んで（個人的な感想）

TiDBをはじめとしたPingCAPのOSSは、単体のソフトウェアとしての稼働よりも、様々な用途を想定したデータストアを形成するエコシステムの一部であることが強く意識されている。

例えば、クエリエンジンとしては以下の2つを同時に用いることができ、

  • TiDB：MySQL互換のSQLインターフェース
  • TiSpark：Spark互換

ストレージエンジンとしては下記2つの間で一貫性を保った状態でデータを保持することができる。

  • TiKV：ACIDトランザクションを提供する分散KV (※CNCF Graduated Project)
  • TiFlash：列形式のデータストア

これらのエコシステム間の関係を示した構成図は下記の通りとなる。

<TiDB 4.0 HTAP architecture>

"How We Build an HTAP Database That Simplifies Your Data Platform"より引用

このような構成を取ることで、「replicated state machine-based consensus algorithmsを拡張した、HTAP向けデータベース」が実現されており、これは過去に例のない新規性の高いものであることが当論文では主張されている。

上記の構成に見られるデュアルフォーマット（行と列、それぞれのストレージ）を持つデータベースは他にもあるが、それらの間の一貫性（最新性と言い換えても良い）をRaftを用いたレプリケーションで担保する点に、TiDBエコシステムの大きなユニーク性がある。

一方でRaftは万能ではなく、通信のオーバーヘッドによるレイテンシの増加が発生するなど、性能にセンシティブなデータストアとして利用する際には最適化の余地が多く残されている。

そうした最適化について、MULTI-RAFT STORAGEのセクションでLeader-Follwerによるコンセンサスベースのレプリケーションを行うTiKVと、それらと非同期でデータを受け取るTiFlashのアーキテクチャが詳細に解説されている。TiFlashの実装についてもClickHouse同等のDeltaTreeに関する内容が紹介されており、TiKVや他のNewSQLストレージエンジンで用いられるLSMツリーとのRead/Write特性の違いについて理解が進む内容となっている。

ここまでのReadやWriteパスの最適化については、CockroachDBやYugabyteDBなどでも共通した、Raftを用いた分散データベースのノウハウが習得できる。

またHTAP Enginesのセクションでは、Snapshot Isolationの分離レベルを実現し、楽観的/悲観的の2つのトランザクションタイプを選択可能なエンジンがどのように実装されているかを解説している。TiDBの特徴として、他のNewSQLのようにHLCではなく、centralized clockであるTSOを利用したPercolatorモデルを用いることから、タイムスタンプを発行するPDがボトルネックとはならない点についても、検証が行われている。

EXPERIMENTSのセクションでは、OLTP、OLAP、そしてHTAP（OLTP+OLAP）のワークロードにおいて、CockroachDBやMemSQL等の他のDBMSとの比較を行っている。
OLTPにおける楽観的/悲観的ロックの性能差や、リソースを膨大に消費する分析クエリがトランザクションクエリに影響を及ぼす程度、非同期であるLearnerへのデータ同期のラグに関する検証など、見るべき点はここでも非常に多い。
一方でこの論文に限らず、他DBMSとの比較は容易ではなく（そもそものアーキテクチャの違いや検証者の習熟度等から）、TiDBを中心とした参考指標程度に捉えるのが妥当と思われる。

RELATED WORKには、TiDBと同様のコンセプトを持つ分散SQLデータベースはもちろん、OracleやSAP HANAなどの商用DBや、インメモリによる分析クエリの高速化を目指したDBMSやHTAPを目標としたいくつかのDBMSが紹介されている。
これらを把握することはTiDBが目指すHTAPの現状の課題と改善の方向性を理解する上の助けとなるだろう。

また、仕方ないこととは言え、当論文ではRaftベースの分散SQLデータベースについてのデメリットはほぼ触れられていない。例えば、既に述べたようなレイテンシの課題を既存RDBMS（つまりMySQL）と比較するような検証は含まれず、更に最低3つのレプリカが必要なLeader-Follower構成や可用性を考えると2つは必要なLearnerのレプリカ等で膨らむデータ容量の増加についても言及はない。
この辺りは今回論文を理解した上で、利用者が適切に判断をする必要がある。

論文のほぼ全訳

以降では論文全文の（一部要約しながらも）日本語訳を試みる。
訳文中で私個人の意見を述べる際には※～私見～というフォーマットでコメントしている。

全訳掲載の是非についてはPingCAP社へ確認済みで快諾頂いております。

ABSTRACT/INTRODUCTION

HTAP（OLTP+OLAP）のデータベースでは、トランザクションクエリと分析クエリを分離して処理し、それらの間の干渉を取り除く必要がある。
※これができない場合、一般的に長時間実行されるOLAPクエリが投入された際に、 レイテンシが重要なOLTPが使い物にならなくなってしまう。

これを実現するには、2つのタイプのクエリに最適化された異なるレプリカを管理する必要があるが、分散したレプリカに一貫性のあるビューを提供することは困難である。さらに一貫性に加えて、高可用性も担保するレプリケーションの仕組みを構築することは困難を極める。

こうした課題に対処するため、ステートマシンベースのコンセンサスアルゴリズムを拡張して、HTAPワークロードへの一貫したレプリカの提供を提案する。

このアイデアに基づいた、RaftベースのHTAPデータベースであるTiDBを紹介する。 ここでは、行ストアと列ストアで構成されるmulti-Raft ストレージシステムをデザインする。行ストアはRaftアルゴリズムに基づいて構築され、可用性の高いトランザクションをサポートする。その上で、RaftログをLearnerに非同期的でレプリケートし、行形式を列形式に変換して列ストアを形成している。

RDBMSは、トランザクション保証やSQLインターフェイスなどで未だに人気が高いが、旧来のRDBMSはスケーラビリティと高可用性を提供できていない。 そのため、2000年代の初め頃からインターネット上のアプリケーションはGoogle Bigtable やDynamoDBのようなNoSQLシステムを選んでいた。NoSQLは、一貫性の要件を緩和し、高いスケーラビリティと、キーと値のペアなどの多様なデータモデルを提供した。

しかし、多くのアプリケーションには強力なトランザクション、データ整合性、およびSQLインターフェイスも必要であるため、NewSQLシステムが登場する。CockroachDBやGoogleSpannerなどのNewSQLは、NoSQLのようなRead/Write双方の高いスケーラビリティを実現し、ACIDトランザクションを担保する。

さらに、SQLベースのOLAPシステムは、SQL-on-Hadoopシステムなど多く開発されている。 それらは複雑性も持ち込むこととなったが、最新バージョンのデータをリアルタイムで分析するモチベーションから、ハイブリッドOLTPおよびOLAP（HTAP）システムが生まれた。 HTAPでは、NewSQLシステムのように、スケーラビリティ、高可用性、および地域を越えた一貫性を実装する必要がある。

さらに、HTAPシステムはデータの鮮度と分離という2つの要件を共に実現し、OLTPおよびOLAPのスループットとレイテンシを保証する必要がある。従来のETLによるデータ更新ではこの要件を満たすことができない。

ここでいう分離とは、OLTPクエリとOLAPクエリの処理を分離して、それぞれ独立したパフォーマンスを保証することである。 一部のインメモリデータベース（HyPerなど）では、分析クエリで同じサーバー上のトランザクション処理から最新バージョンのデータを読み取ることができるが、分離が実現できていない。これはHyPerおよびSAP HANAでHTAPベンチマークであるCH-benCHmarkを実行することで研究されている。この研究では、分析クエリを同時に実行すると、達成可能な最大のOLTPスループットが大幅に低下した。
SAP HANAのスループットは少なくとも3倍、HyPerは少なくとも5倍減少した。 同様の結果がMemSQLでも確認されている。さらに、インメモリデータベースは、シングルサーバに展開されている場合、高可用性とスケーラビリティを提供できない。

分離を実現する際の本質的な難しさは、単一システム内でOLTPワークロードからOLAPへ一貫性のあるレプリケーションを行うことにある。この際には高可用性も重要であり、それはPaxosやRaftなどのアルゴリズムを使用して実現できる。これらのコンセンサスアルゴリズムを拡張して、HTAPワークロードに一貫性のあるレプリカを提供するアイデアは、論文著者達の知る限り、このアイデアはこれまで研究されていない。

今回の研究の要点は以下となる。

  • コンセンサスアルゴリズムに基づいてHTAPシステムを構築することを提案し、RaftベースのHTAPデータベースであるTiDBを実装した。これはオープンソースプロジェクトで、HTAPワークロードに高い可用性、一貫性、スケーラビリティ、データの鮮度、および分離を提供する。
  • リアルタイムOLAPクエリ向けのカラムナストアを生成するために、RaftアルゴリズムにLearnerの役割を導入する。
  • multi-Raft ストレージシステムを実装し、その読み取りと書き込みを最適化して、スケーリングした際も高いパフォーマンスを提供する。
  • 大規模なHTAPクエリ用にSQLエンジンを調整する。このエンジンは、行ベースのストアと列ストアのどちらを使用するかを最適に選択できる。
  • HTAPベンチマークであるCH-benCHmarkを使用して、OLTP、OLAP、およびHTAPに関するTiDBのパフォーマンスを評価するための実験を実施する。

RAFT BASED HTAP/ARCHITECTURE

RaftやPaxosなどのコンセンサスアルゴリズムは、一貫性を保持し、スケーラブルで、可用性の高い分散システムを構築するための基盤として利用されている。今回のTiDBエコシステム全体としては、HTAPワークロードの様々なノードにステートマシンベースのレプリケーションを行う。これによりOLTPとOLAPのリソース分離と、OLAP向けストレージが新鮮かつ一貫性の取れたデータを持つことを同時に実現する。このようにコンセンサスアルゴリズムを使用してHTAPデータベースを構築した研究は過去にないと主張されている。

TiDBではデータは、まずOLTP向けに行形式で複数のRaftグループ（各グループがLeader/Followerから構成される）に分割して格納される。Learnerはこれらのグループに追加され、Leaderからのデータを非同期に受け取り、列形式に変換する。ここで、LearnerをLeader/Followerとは別の物理ノードに配置することで、リソース分離が実現できる。

※この構成を説明したのが下図であり、クォラムと非同期（Asynchronous）のレプリケーションの違いを示している。

通常のRaftグループでは、各FollowerはLeaderに昇格してRead/Writeを処理できる。仕組み上、Followerの追加はリソース分離に繋がらないだけでなく、Leaderがクライアントに応答する前に、より大きなクォラムの応答を待つ必要が生じるためにパフォーマンスを劣化させる。 これを解決するために、RaftコンセンサスアルゴリズムにLearnerを導入した。LearnerはLeader選出に参加せず、ログレプリケーションのクォラムにもならない。 LeaderからLearnerへのログレプリケーションは非同期で、クライアントへの応答時間には影響しない。LeaderとLearnerが持つデータ間の強い一貫性は、読み取り時間中に保証され、ログレプリケーションのラグは通常低くなっている。
※LearnerをRaftに導入したのはTiDBの貢献なのだろうか。[etcd Learner](https://etcd.io/docs/current/learning/design-learner/)にも解説がある。

こうしたデザインにおいて、行形式と列形式でデータの一貫性が保たれるため、クエリオプティマイザはいずれかまたは両方のストレージにアクセスする実行計画を作成するという最適化も可能となる。

OLTPで求められるのは効率的なデータ更新で、一方OLAPに求められるのは結合や集計等に必要な列形式のアクセス（と大きな範囲でレコード取得）である。行形式ではインデックスを活用して効率的なアクセスを実現するが、列形式ではデータ圧縮とベクトル化された処理による効率化が行われる。

Raftを拡張して、HTAPデータベースの一貫性とリソース分離の要件を満たすために、次のようなエンジニアリング上の課題を克服した。

  1. 基本的なRaftプロセスでは、リクエストは順次処理され、クライアントに応答する前にRaftノードのクォラムで承認される。このプロセスにはI/Oが含まれるため、レイテンシを増大させる。この課題を解消するため、並行性の高いRead/WriteをサポートするスケーラブルなRaftストレージシステムを構築し、大規模なデータセットを処理するLeaderがボトルネックとならないように、データを分散するためのパーティション戦略が求められる。
  2. データの一貫性を保つため、低レイテンシでログをLearnerに同期させる必要があり、必要なトランザクションログが非常に大きくなる可能性がある。これらのログはLearnerで素早くリプレイする必要があり、その中で列形式に変換する際に、スキーマ不一致によるエラーが発生する場合がある。
  3. OLTPとOLAPの双方で効率的に処理を行い、かつパフォーマンスも保証する必要がある。これらは相互に影響を与えてはいけないし、オーバーヘッドを削減するために、行形式のストアと列形式のストアの両方で最適な実行計画を選択する必要がある。

構造面から見ると、TiDBはMySQLプロトコルをサポートし、MySQL互換クライアントからアクセスできる。そして、分散ストレージレイヤー、配置ドライバー（PD）、および計算エンジンレイヤーの3つのコアコンポーネントを持つ。

分散ストレージレイヤーは、行ストア（TiKV）と列ストア（TiFlash）で構成される。 TiKVに格納されているデータはソートされたKey-Valueマップである。各タプルは、キーと値のペアにマップされ、キーはテーブルIDと行ID、値は実際の行データである。たとえば、4列のタプルは次のようにエンコードされる。

Key:{table{tableID}_record{rowID}}
Value: {col0, col1, col2, col3}

スケールアウトするには、rangeに基づくパーティション戦略を使用して、大きなKey-Valueマップを多数の連続するrangeに分割する。 各rangeは region と呼ばれる。regionは可用性を高めるために複数のレプリカを持ち、Raftコンセンサスアルゴリズムにより、レプリカ間（Raftグループを形成する）の一貫性が維持される。RaftグループのLeaderは、TiKVからTiFlashにデータを非同期でレプリケートする。

配置ドライバー（PD）はregionの管理を担当し、各キーの物理的配置や、region間でキーを自動的に移動させるなどの作業を行う。 PDはタイムスタンプオラクル（timestamp oracle）でもあり、厳密に増加し、グローバルに一意なタイムスタンプを発光する。このタイムスタンプは、トランザクションIDとしても機能する。 PDはステートレスで、起動時にTiKVノードから（または他にPDがいればそこからも）必要なすべてのデータを収集する。
TiDBの起動シーケンスを考えたことがなかったが、TiKV全体からメタデータをPDに集めるとした場合、起動時間がかなりかかりそうではある。

計算エンジンレイヤもステートレスであり、スケールアウトが可能である。SQLエンジンとして、コストベースのクエリオプティマイザと分散クエリエグゼキュータが含まれる。TiDBは、トランザクション処理をサポートするために、Percolatorに基づく2PCプロトコルを実装している。

上記のコンポーネントに加えて、TiDBはSparkとも統合される。これは、TiDBとHadoop分散ファイルシステム（HDFS）に格納されたデータを統合する際に有用となる。

MULTI-RAFT STORAGE

TiDBの分散ストレージレイヤーは、行形式のTiKVと列形式のTiFlashで構成される。大きなテーブルは、TiKVに保存されるregionに分割される。各regionは、Raftコンセンサスアルゴリズムを使用してレプリカ間の一貫性と高可用性を実現する。 テーブルスキャンに最適化するために、TiFlashへのレプリケート時に複数のregionを1つのパーティションにマージできる。このように複数のRaftグループが分散ストレージレイヤーのデータを管理するため、これをmulti-Raftストレージと呼ぶ。以降ではHTAPデータベースとしての最適化に焦点を当てて、TiKVとTiFlashについて詳しく説明する。

行形式のストレージ（TiKV）

これまで見てきたようにregionは、Raftを使用してTiKVサーバー間でレプリカされる。各TiKVサーバーは、各regionのLeaderまたはFollowerとなる。各TiKVサーバーでは、データとメタデータは、組込み式のKey-ValueストアであるRocksDBに永続化される。regionには最大サイズがあり、デフォルトでは96MBとなる。LeaderとなるTiKVサーバーは、対応するregionのRead/Writeリクエストを処理する。

基本的なRaftプロセスはLeaderとFollower間で以下の手順で実行される。

（基本的なRaftプロセス）
1. regionのLeaderは、SQLエンジンからリクエストを受ける。
2. Leaderはリクエストをログに追加。
3. Leaderは、新しいログエントリをFollowerに送信、Followerはそれを自身のログに追加。
4. LeaderはFollowerの応答を待つ。ノードのクォラムが応答したら、Leaderはコミット。
5. Leaderは結果をクライアントに送信。

このプロセスではデータの整合性と高可用性が保証されるが、手順がシーケンシャルに実行されるため、効率的とは言えず、大きなI/Oオーバーヘッド（ディスクとネットワークの両方）が発生する可能性がある。

LeaderとFollower間の最適化

上記のプロセスでは、2番目と3番目のステップには依存関係がないため、Leaderはログをローカルに追加し、同時にFollowerに送信できる。Leaderでログの追加が失敗しても、Followerのクォラムがログを正常に追加した場合にはコミットが可能である。
3番目のステップでは、Followerにログを送信する際、Leaderはそれをバッファリングしてまとめてフォロワーに送信できる。ログを送信した後、LeaderはFollowerの応答を待つ必要がなく、 成功を仮定して予測したログインデックスを用いて更にログを送信する。エラーが発生した場合、Leaderはログインデックスを調整し、レプリケーションのリクエストを再送する。
4番目のステップでは、この段階では一貫性のリスクがないため、コミットされたログエントリを適用するLeaderを別のスレッドで非同期に処理できる。

これらの最適化に基づいて、複数のクライアントからのリクエストは並行処理してスループットを向上させたRaftプロセスは、以下のようになる。

（最適化後のRaftプロセス）
1. LeaderはSQLエンジンからのリクエストを受ける。
2. 対応するログをFollowerに送信し、並行してログをローカルに追加。
3. Leaderはクライアントからのリクエストを受け続け、2.を繰り返す。
4. Leaderはログをコミットし、別スレッドで適用。
5. ログを適用した後、結果をクライアントに送信。

Readの高速化手法

TiKVのLeaderに対するReadはlinearizableである。regionのLeaderから時刻tに値が読み取られる場合、Leaderはt以降のReadに対して以前のバージョンの値を返してはならないことを意味する。これはReadごとにログエントリを発行し、そのエントリがコミットされるのを待つというRaftプロセスによって実現できる。しかし、このプロセスはログのをレプリケート時に通信のオーバーヘッドが発生する。パフォーマンスを向上させるために、ログ同期フェーズを回避する必要がある。
Raftでは、Leaderがデータの書き込みに成功するとサーバ間でログを同期せずに、Leaderは読み取り要求に応答できる。しかし、Leaderはサーバ間で移動される場合があるため、それを前提に、TiKVでは次のような読み取り最適化を実装している。

最初のアプローチは、read indexと呼ばれる。LeaderがReadリクエストに応答すると、現在のコミットインデックスをローカルのread indexとして記録し、フォロワーにハートビートメッセージを送信して自身がLeaderであることを確認する。依然としてLeaderであれば、適用済みのインデックスがread index以上になると、その値を返すことができる。このアプローチはReadのパフォーマンスを向上させるが、通信のオーバーヘッドは多少発生する。

もう一つのアプローチはlease readであり、read indexによって引き起こされるハートビートの通信を削減する。LeaderとFollowerは、Leaderが変更されないようにFollowerが再選出を提案しないリース期間に合意する。リース期間中、LeaderはFollowerに確認することなく、任意のReadに応答できる。 このアプローチは、各ノードのCPUクロックに差があまり生じない場合に上手く機能する。

Leaderだけでなく、FollowerもクライアントからのReadに応答することができ、これをfollower readと呼ぶ。FollowerはReadリクエストを受け取ると、Leaderに最新のread indexを問い合わせる。Followerのローカルで適用済みのインデックスがread indexと同じかそれ以上であればその値をクライアントに返すことができるが、そうでなければ、ログが適用されるのを待つ必要がある。follower readは、リクエストが集中するregionのLeaderの負荷を軽減してReadのパフォーマンスを向上させるとともに、Followerの追加で更なるパフォーマンスの改善が可能となる。

regionの管理

regionはサーバ間で分散して格納され、サーバ構成とデータサイズは動的に変化するため、一部のサーバのディスクが過剰に使用される（他のサーバは空いている）状態が発生する。
サーバ間でregionをバランシングするために、PDはレプリカの数と配置に制約をつけてregionをスケジュールする。重要な制約の一つは、高可用性のために、異なるTiKVインスタンス上にregionの少なくとも3つのレプリカを配置することである。
PDはハートビートを通じてサーバから特定の情報を収集するとともに、各サーバのワークロードを監視し、アプリケーションに影響を与えることなくホットなRegionを異なるサーバに移行する。
regionを維持するにはハートビートの送信やメタデータの管理が必要だが、Raftグループにワークロードがない場合、ハートビートは不要となる。regionのビジー率に応じて、ハートビートを送信する頻度を調整し、ネットワークやノードの過負荷が生じるのを避けることができる。

regionにアクセスが集中し、Read/Writeの性能を保証できなくなることがある。そうしたホットなregionや大規模になったregionは、負荷をより良く分散させるために小さなregionに分割すべきである。また逆に、多くのregionが小さくてアクセスされることはほとんどない場合、管理負荷を下げるために小さなリージョンをマージする必要もある。（region間のソート順を維持するために、キー空間内の隣接するregionのみがマージ可能である）これらの分割とマージはPDからのコマンド発行で動的に行われる。

分割操作ではあるregionをいくつかの新しい小さなregionに分割し、それぞれが元のregionのrangeをカバーする。一番右のrangeをカバーするregionは、元のregionのRaftグループを再利用する。他のregionでは新しいRaftグループが使われる。分割操作は以下のように通常のRaftプロセスに似ている。

1. PDがあるregionのLeaderに分割のコマンドを発行。
2. Leaderは分割コマンドを受けると、それをログに変換してFollowerにレプリケート。
　（ログには分割コマンドのみが含まれる。）
3. クォラムが応答するとLeaderはコマンドをコミットして、Raftグループに適用。
　適用処理では、元のregionのrangeとエポックのメタデータを更新。
　残りのrangeをカバーするために新しいregionを作成。
　このコマンドはアトミックに適用され、ディスクまで同期される。
4. 分割されたregionの各レプリカに対して、Raftステートマシンを作成。
　元のregionのLeaderは分割結果をPDに報告して、分割処理が完了する。

分割処理は、過半数のノードが分割ログをコミットしたときに成功する。すべてのノードがregionの分割を完了させる必要はなく、他のRaftログをコミットするのと似ている。region分割はメタデータの変更のみが必要でオーバーヘッドは低い。分割コマンドが終了した後、PDの定期的なロードバランシングにより、新たに分割されたregionはサーバ間で移動することができる。隣接する２つのregionのマージは、分割の反対となる。

列形式のストレージ（TiFlash）

上記のようにTiKVからのReadを最適化しても、そもそも行形式のデータは大量のデータを高速に分析する用途にはあまり適していない。そこで、TiDBにカラムナストア(TiFlash)を組み込む。TiFlashはLearnerノードで構成されており、Raftグループからのログを受けて、行形式のタプルを列データに変換する。

ユーザーは、SQL文で簡単にテーブルの列形式のレプリカを設定できる。

ALTER TABLE x SET TiFLASH REPLICA n. 
-- xはテーブル名、nはレプリカ数

列形式のレプリカを追加することは、テーブルに非同期のカラムナインデックスを追加するのと似ている。TiFlash のテーブルはパーティションに分割され、それぞれのパーティションは、TiKVのいくつかの連続したregionに従ってrangeをカバーする。パーティションが大きければ、範囲スキャンが容易となる。

TiFlashインスタンスを初期化する際、関連するregionのLeaderは、新しいLearnerにデータをレプリカを開始する。同期の高速化のため、データが多すぎる場合はLeaderはそのスナップショットを送信する。初期化が完了すると、TiFlashインスタンスは Raftグループからの更新をリッスンする。Learnerがログのパッケージを受け取ると、ログのリプレイ、データフォーマットの変換、ローカルストレージ内の参照値の更新などを含めて、ログをローカルのステートマシンに適用する。

ログのリプレイ、スキーマの同期

RaftのアルゴリズムによってLearnerが受け取るログはlinearizableである。コミット済みデータでlinearizableを維持するため、それらのログは先入れ先出し（FIFO）でリプレイされる。ログ再生には3つのステップがある。

  1. ログの圧縮 トランザクションのログは、prewritten/committed/rollbackedの3つの状態に分類される。rollbackedのデータはディスクに書き込む必要がないため、圧縮処理では、不要なprewrittenのログを削除し、有効なログだけをバッファに格納する。
  2. タプルのデコード バッファ内のログを行形式のタプルにデコードし、トランザクションに関する冗長な情報を削除する。そして、デコードされたタプルを行バッファに入れる。
  3. データ形式の変換 行バッファ内のデータがサイズ制限を超えるか、またはその持続時間が制限を超える場合、これらの行形式タプルは列形式のデータに変換され、ローカルパーティションのデータプールに書き込まれる。変換ではTiKVと周期的に同期されるローカルキャッシュのスキーマを参照する。

ログのリプレイでは、各Raftログの項目を*transaction ID-operation type[transaction status][@start ts][#commit ts]operation data*のように抽象化する。典型的なDMLでは、operation typeにはinsert、update、deleteが含まれる。transaction statusはprewritten/committed/rollbackedのいずれでもよい。operation dataは具体的にinsertやupdateされたデータ、deleteされたキーとなる。最終的に変換された列形式のデータはDeltaTreeに追加される。

タプルをリアルタイムで列形式に変換するために、Learnerは最新のスキーマを認識している必要がある。このスキーマフルなプロセスは、タプルをバイト配列としてエンコードするTiKVのスキーマレスな操作とは異なる。最新のスキーマ情報はTiKVに格納されており、TiFlashがそれをリクエストする回数を減らすため、Learnerはスキーマキャッシュを管理する。

キャッシュはスキーマの同期機能によってTiKVのスキーマと同期する。キャッシュされたスキーマが古い場合、デコードされたデータとローカルスキーマの間にミスマッチが生じ、データを再変換する必要がある。スキーマを同期する頻度とミスマッチが生じる回数の間にはトレードオフがあり、それを解消するために以下の2段階の戦略を採用する。

  • 定期的な同期 スキーマの同期機能は定期的にTiKVから最新スキーマを取得し、ローカルキャッシュに変更を適用する。
  • 強制的な同期 スキーマ同期機能は不一致のスキーマを検出した場合、TiKVから最新スキーマを取得する。これは列番号がタプルとスキーマの間で異なる場合や、列の値がオーバーフローした場合に起動される。
    ※TiKVのスキーマ変更時にそれをTiFlashにプッシュしたりしないのだろうか？

Columnar Delta Tree

列形式のデータのWriteとReadを効率的に行うために、新しい列形式のストレージエンジンであるDeltaTreeを設計し、差分（delta）の更新を追記していき、後からパーティションごとに以前の安定バージョンとマージする。
図4に示すように、DeltaTreeには更新のDeltaとStableのデータが別々に格納される。Stableのスペースでは、パーティションのデータはチャンクとして格納され、それぞれがパーティションのタプルのより小さなrangeをカバーし、これらの行形式タプルは列ごとに格納される。これに対してDeltaのスペースには、TiKVが生成した順序でdeltaが直接追記される。TiFlashの列形式データのフォーマットはParquetに似ており、列のチャンクに行グループを格納する。異なる点として、Parquetが1つのファイルに全てを格納するのと対照的に、TiFlashは行グループの列データとそのメタデータを別のファイルに格納してそれらを同時に更新する。TiFlashでは、データファイルを一般的なLZ4で圧縮し、ディスクサイズを節約する。

図4 The columnar delta tree

新しいdeltaは、挿入されたデータまたは削除された範囲のアトミックなバッチである。これらのdeltaはメモリにキャッシュされて後にディスクに永続化されるが、順序通りに格納されているため、write-aheadログ（WAL）の機能も実現している。deltaは通常は沢山の小さなファイルに格納されるため、読み込み時に大きなIOオーバーヘッドが発生する。このコストを削減するため、定期的に小さなdeltaをより大きなものに圧縮し、より大きなdeltaをディスクにフラッシュして、以前に永続化された小さなものを置き換える。受け取ったdeltaのメモリ内コピーは、最新データの読み取りを容易にし、古いデルタが制限サイズに達した場合、それらは削除される。
※DeltaTreeは一般的概念なのか、今回TiDBエコシステムでデザインされたものなのか分かっていない。

ある特定のタプルの最新データを読み込む場合、関連するdeltaがどこに含まれるか事前に分からないため、すべてのdeltaファイルをStableのタプルとマージする必要があり、読み取り増幅（read amplification）が発生する。更にdeltaファイルには、無駄なデータが含まれている可能性があり、スペース増幅（space amplification）が発生し、Stableのデータとのマージが遅くなる。そのため、定期的にdeltaファイルをStableのスペースにマージする。各deltaファイルとそれに関連するチャンクはメモリに読み込まれ、マージされる。deltaに挿入されたタプルはStableのチャンクに追加され、修正されたタプルは元のタプルを置き換え、削除されたタプルは移動されます。マージされたチャンクは、ディスク内の元のチャンクをアトミックに置き換える。
※ここではDeltaTreeがLSM Treeのような追記型のディスク構造ではないと言っているように見える。

マージすると、Deltaスペースで関連するキーの順序が崩れるため、コストが生じる。また、このような非順序性は、Readに最新データを返すためにStableなチャンクにdeltaをマージする処理を遅くする。そこで、Deltaスペース上にB+木インデックスを構築する。更新の各deltaは、そのキーとタイムスタンプによって順序付けられたB+ツリーに挿入される。 この順序性は、あるキーの範囲で行う更新を効率的に行ったり、Deltaスペース内の単一のキーを探し出したりするのに有用である。また、B+木の順序付けされたデータは、Stableなチャンクとのマージが容易である。

TiFlashにおいて、ログ構造化マージ(LSM)ツリーとDeltaTreeの性能を比較するための実験を行った。 TiKVノードを3つ、TiFlashノードを1つ設定している。TiKV上でSysbenchの書き込みワークロードを実行し、TiFlash上で次のSQLを実行する。

select count(id), count(k) from sbtest1

compactionの大きな書き込み増幅（write amplification）を避けるために、levelスタイルのcompactionではなく、universal compactionを使用してLSMストレージエンジンを実装した。この実装は列指向OLAPデータベースのClickHouseでも採用されている。
DeltaTreeからのReadは、トランザクションのワークロードと同様に、1億タプル、2億タプルのどちらでも、LSMツリーよりも約2倍高速である。これは、DeltaTreeでは、ReadがB+ツリーでインデックス化されているdeltaファイルの最大1レベルにアクセスするだけなのに対し、LSMツリーでは多くの重複するファイルにアクセスするためである。deltaファイルの比率がほぼ同じであるため、異なる書き込みワークロードの下でもパフォーマンスはほぼ安定している。DeltaTreeの書き込み増幅率(16.11)はLSMツリー(4.74)よりも大きいが、それも許容範囲内である。
※DeltaTreeはLSMツリーのReadの弱点を補っているが、書込みは劣化するようだ。

Readのプロセス

follower readと同様に、LearnerはSnapshot Isolationを提供するので、特定のタイムスタンプでTiFlashからデータを読み出すことができる。Readのリクエストを受けると、LearnerはLeaderにread indexのリクエストを送り、必要なタイムスタンプをカバーする最新のデータを取得する。LeaderはログをLearnerに送り返し、Learnerはログをリプレイして保存する。ログがDeltaTreeに書き込まれると、そこからデータが読み込まれ、クライアントへ応答する。

HTAP ENGINES

大規模なトランザクションと分析クエリの並列処理という課題を解決するため、それらを実行し評価するためのSQLエンジンを提供する。
このSQLエンジンは、Percolatorモデルを適用し、分散クラスタに楽観的および悲観的なロックを実装する。SQLエンジンは、ルールとコストベースのオプティマイザ、インデックス、ストレージ層へのプッシュダウンを用いて、分析クエリを高速化する。また、Hadoopのエコシステムと接続し、OLAP機能を強化するためにTiSparkを実装している。HTAPリクエストは、分離されたストレージとエンジンで別々に処理することができ、特にSQLエンジンとTiSparkは、行形式と列形式の両方のストアを同時に使用して、最適な結果を得ることができる。

トランザクション処理

TiDBは、Snapshot Isolation(SI)またはRepeatable Read(RR)の分離レベルでACIDトランザクションを提供する。
SIはトランザクション内の各リクエストが一貫したバージョンのデータを読み取ることを可能にし、RRはトランザクション内の異なるステートメントが同じキーに対して異なる値を読み取る可能性があっても繰り返しの読み取り(すなわち、同じタイムスタンプを持つ2つの読み取り)は常に同じ値を読み取ることを意味する。TiDBの実装は、マルチバージョン同時実行制御（MVCC）に基づいており、read-writeのロックを回避し、write-writeの競合から保護します。TiDBでは、トランザクションはSQLエンジン、TiKV、PDの間で協調的に行われ、トランザクション中の各コンポーネントの役割は以下の通りとなる。

  • SQLエンジン トランザクションを調整する。クライアントからのWriteとReadを受けて、データをkey-valueの形式に変換し、2フェーズコミット（2PC）を使用してTiKVにトランザクションを書き込む。
  • PD 論理的なregionと物理的な配置を管理し、グローバルで厳密に増加するタイムスタンプ（timestamp oracle）を提供する。
  • TiKV 分散トランザクションのインターフェースを提供し、MVCCを実装して、ディスクにデータを永続化する。

TiDBは、楽観的ロックと悲観的ロックの両方を実装している。これらは1つのキーを主キーとして選択し、それをトランザクションの状態を表すために使用するPercolatorモデルと、トランザクションを行うために基本となる2PCに対応している。

1. クライアントのbeginを受けて、SQLエンジンはPDに開始タイムスタンプ(start_ts)を要求。
2. SQLエンジンはTiKVからデータを読み取りローカルメモリに書き込むことでDMLを実行。
   TiKVはstart_ts以前の最新のcommit_tsでデータを提供。
3. SQLエンジンがcommitを受けると、2PCプロトコルを開始。
   主キーをランダムに選んですべてのキーを並列にロックし、TiKVにprewriteを送る。
4. prewriteが成功したら、SQLエンジンはPDにのcommit_tsを要求。
   TiKVにcommitを送り、TiKVはcommit後にSQLエンジンに応答。
5. SQLエンジンはクライアントに成功を返す。
6. SQLエンジンはセカンダリキーをcommit。
   TiKVに更にcommitを送って、非同期かつ並列にロックをクリアする。

楽観的なトランザクションと悲観的なトランザクションの主な違いは、ロックを取得するタイミングとなる。楽観的なトランザクションでは、prewriteの段階（上記の3.）で段階的にロックが取得される。一方、悲観的なトランザクションでは、prewrite以前にDMLを実行しながらロックを取得していく（2.の一部）ため、prewriteが開始されれば、他のトランザクションとの競合によりトランザクションが失敗することはない。(それでもネットワークのパーティションなどの問題で失敗することはあり得る)。

悲観的なトランザクションでキーをロックする場合、SQLエンジンはfor_update_tsと呼ばれる新しいタイムスタンプを取得する。 SQLエンジンがロックを取得できない場合、トランザクション全体をロールバックして再試行するのではなく、そのロックから始まるトランザクションを再試行することができる。データを読み込む際、TiKVはキーのどの値が読み込まれるかを決定するために、start_tsではなくfor_update_tsを使用する。このようにして、悲観的なトランザクションは、ランザクションの部分的な再試行でもRRの分離レベルを維持する。
悲観的なトランザクションでは、ユーザはRead Committed(RC)分離レベルのみを要求することもできる。これにより、トランザクション間の競合が減り、パフォーマンスが向上するが、分離されたトランザクションが減少することを犠牲にしている。実装の違いとしては、RRではReadが他のトランザクションによってロックされたキーにアクセスしようとした場合、TiKVは競合を報告しなければならない。RCでは、Readのためにロックを無視することができる。

TiDBは、中央集権的なロックマネージャーなしで分散トランザクションを実装している。ロックはTiKVに格納され、高いスケーラビリティと可用性を実現する。 また、SQLエンジンとPDはOLTPリクエストを処理するためにスケーラブルな構成となっている。
※PDはスケーラブルなのか、この論文では言及がないか。

タイムスタンプはPDから払い出される。各タイムスタンプには、物理と論理の両時刻が含まれており、物理時刻はミリ秒精度の現在時刻を指し、論理時間は18ビットで表される。そのため理論的には、PDは1ミリ秒あたり2^18のタイムスタンプを割り当てることができる。実際に、タイムスタンプの割り当てには数サイクルのコストがかかるだけで、1秒間に約100万個のタイムスタンプを生成することができる。クライアントはオーバーヘッド、特にネットワークのレイテンシを償却するため、一括でタイムスタンプを要求する。現在のところ、我々の実験でも、実運用の環境でも、タイムスタンプの取得はパフォーマンスのボトルネックにはなっていない。

分析処理

このセクションでは、OLAPクエリの最適化について、オプティマイザ・インデックス・プッシュダウンを含む、カスタマイズされたSQLエンジンとTiSparkでの最適化について説明する。

SQLエンジンでのクエリ最適化

TiDBは、クエリ最適化の2つのフェーズを持つオプティマイザを実装している。1つは論理的なプランを生成するクエリのルールベース最適化（RBO）、もう1つは論理的なプランを物理的なプランに変換するコストベース最適化（CBO）である。RBOは不要な列の切り取り、射影（projection）の排除、述語のプッシュダウン、述語の導出、定数の折りたたみ、"group by"や外部結合の排除、サブクエリのネスト解除など、豊富な変換ルールを持つ。CBOは実行コストに応じて候補プランの中から最もコストの低いプランを選択する。TiDBはTiKVとTiFlashの2つのデータストアを持つため、テーブルのスキャンには以下の3つのオプションがある。

  • TiKVで行形式でテーブルスキャン
  • TiKVでインデックススキャン
  • TiFlashで列形式でテーブルスキャン

インデックスはデータベースのクエリパフォーマンスを向上させるために重要で、通常はポイント取得（※いわゆるユニークスキャン）や範囲検索で使用され、ハッシュ結合やマージ結合のためのより安価なデータスキャンのパスを提供する。TiDBは分散環境で動作するようにスケーラブルなインデックスを実装している。インデックス管理には大量のリソースを消費し、オンライントランザクションや分析処理に影響を与える可能性があるため、バックグラウンドで非同期にインデックスの構築や削除を行う。インデックスはデータと同じようにregionに分割され、key-valueの形でTiKVに格納される。

一意なキーインデックス上のインデックス項目は、次のようにエンコードされる。

Key: {table{tableID} index{indexID} indexedColValue}
Value: {rowID}

非一意なインデックス上のインデックス項目は、以下のようにデコードされる。

Key: {table{tableID} index{indexID} indexedColValue rowID}
Value: {null}

インデックスを使用するには、インデックスに関連したregionを見つけるための二分探索が必要となる。インデックス選択の安定性を高め、物理的な最適化のオーバーヘッドを減らすために、skyline pruningのアルゴリズムを用いて、候補インデックスを排除する。異なるクエリ条件にマッチする複数の候補インデックスがある場合、部分的な結果（すなわち、評価済みのrowIDのセット）をマージして、正確な結果セットを得る。
※skyline pruningとは何だろう。

物理的なプラン(CBOの結果)は、SQLエンジンレイヤでpulling iteratorモデルを使用して実行される。ここでいくつかの処理をストレージレイヤにプッシュダウンすることで、さらに最適化できる。ストレージレイヤで処理を実行するコンポーネントはcoprocessorと呼ばれ、異なるサーバ上で実行計画のサブツリーを並列処理する。これにより、ストレージレイヤからエンジンレイヤに送信するタプルの数を減らすことができる。例えば、coprocessorでフィルタを評価し、受け入れられたタプルのみがエンジンレイヤに送信される。coprocessorは論理演算、算術演算、および他の一般的な関数を評価できる。場合によっては、集約やTopNを処理することができる。coprocessorは、演算をベクトル化することでパフォーマンスをさらに向上させることができる。行全体をイテレーションするのではなくバッチ処理し、データを列ごとに整理することで、より効率的なイテレーションを行う。

TiSpark

TiDBがHadoopエコシステムに接続できるように、TiDBはmulti-RaftストレージにTiSparkを追加した。TiSparkはSQLに加えて、機械学習ライブラリなどの強力な計算をサポートし、TiDBの外部からのデータを処理することができる。
TiSparkでは、SparkライバがTiKVからメタデータを読み込み、テーブルスキーマやインデックス情報を含むSparkカタログを構築する。Sparkドライバは、データベースの一貫したスナップショットを取得し、TiKVからMVCCのデータを読み取るためにPDにタイムスタンプを要求する。SQLエンジンと同様に、Sparkドライバはストレージレイヤのcoprocessorに処理をプッシュダウンし、インデックスを使用することができる。これはSparkオプティマイザによって生成されたプランを修正することで実行される。また、TiKVとTiFlashからデータを読み取り、Sparkワーカー向けに行を組み立てるために、読み取り操作をカスタマイズしている。例えば、TiSparkは複数のTiDBのregionから同時に読み込み、ストレージレイヤからインデックスデータを並行で取得できる。特定バージョンのSparkへの依存しないように、これらの機能の殆どは追加パッケージで実装されている。
TiSparkでは大容量データをトランザクションで読み込むこともサポートしており、そこでは2フェーズコミットとテーブルロックを行う。

分離と調整

リソースの分離は、トランザクションクエリのパフォーマンスを保証する際に効果的である。分析クエリはCPU・メモリ・I/Oなどリソースを大量消費することが多いため、トランザクションクエリと一緒に実行されると、そこに深刻な遅延が発生する可能性がある。TiDBでこの問題を回避するために、分析クエリとトランザクションクエリを別々のエンジンサーバにスケジュールし、TiKVとTiFlash（※ストレージエンジン）も分離して配置します。トランザクションクエリは主にTiKVにアクセスするのに対し、分析クエリは主にTiFlashにアクセスする。 RaftによってTiKVとTiFlashの間で一貫性を維持するオーバーヘッドが低いため、TiFlashで分析クエリを実行してもトランザクション処理のパフォーマンスにはほとんど影響しない。

データはTiKVとTiFlashの間で一貫しているため、クエリはTiKVまたはTiFlashのいずれかからReadが可能となる。その結果、クエリオプティマイザはより多くの物理的プランから選択することができ、最適なプランはTiKVとTiFlashの両方からReadする可能性がある。TiKVのテーブルアクセスは行スキャンとインデックススキャンを提供し、TiFlashは列スキャンをサポートする。

これらの3つのアクセスパスは、実行コストとデータの順序性が異なる。行スキャンと列スキャンは主キーによる順序となるが、インデックススキャンはキーのエンコーディングによっていくつかの順序を取り得る。異なるパスのコストは、平均的なタプル/カラム/インデックスのサイズと推定されるタプル/region数に依存する。データスキャンのI/Oオーバヘッドやファイルシークのコストも合わせると、クエリオプティマイザは、式(1)に従って最適なアクセスパスを選択する。式(2)に示すように、行スキャンのコストは連続した行データをスキャンし、regionファイルをシークすることに由来する。列スキャンのコスト（式(3)）は、m列のスキャンの総和である。インデックスの作成された列が表スキャンに必要な列を満たさない場合、インデックススキャン（式(4)）は、インデックスファイルをスキャンするコストとデータファイルをスキャンするコスト（すなわちdouble read）を考慮しなければならない。double readはタプルのランダムスキャンとなり、これは式(5)にあるように多くのファイルを必要とする。

式は論文より引用

クエリオプティマイザが行ストアと列ストアの両方を選択し、同じクエリで異なるテーブルにアクセスする場合の例として以下の例を考えてみよう。

select T.*, S.a from T join S on T.b=S.b where T.a between 1 and 100

これは典型的な結合クエリで、TとSは行ストアの列aにインデックスを持ち、列形式のレプリカも存在する。ここではインデックスを使用して行ストアからTにアクセスし、列ストアからSにアクセスするのが最適となる。 これはクエリにはTで指定された範囲検索（と全てタプル）が必要であり、インデックスを介してタプル単位でデータにアクセスする方が列ストアよりも低コストとなる。一方、Sの2つの列の全てのタプルをフェッチする場合、列ストアを利用した方が低コストである。
※行ストアと列ストアのデータを結合可能である。

TiKVとTiFlashの連携はパフォーマンスの分離を保証する。分析クエリのために、小さな範囲検索またはポイント取得のスキャンのみがfollower readを介してTiKVにアクセスできるが、これはLeaderにはほとんど影響がない。また分析クエリ向けのTiKVのテーブルアクセスサイズをデフォルトで最大500 MBに制限している。 トランザクションクエリは、一意性などのいくつかの制約をチェックするためにTiFlashから列データにアクセスすることがある。実際には、特定テーブルに対して複数の列データレプリカを設定し、1つのレプリカをトランザクションクエリ専用としている。トランザクションクエリを別のサーバで処理し、分析クエリへの影響を回避する。

EXPERIMENTS

まず、TiDBのOLTPとOLAPの能力を別々に評価する。
OLAPではTiKVとTiFlashを用いるSQLエンジンの性能を調査し、TiSparkを他のOLAPシステムと比較する。次に、TiKVとTiFlash間のログレプリケーションの遅延を含むTiDBのHTAPパフォーマンスを測定する。最後に分離の観点からTiDBとMemSQLを比較する。

評価環境

クラスタは以下の通り。
6台のサーバのクラスタ上で包括的な評価を行う。それぞれ188GBのメモリとIntel Xeon E5-2630v4を2つ、すなわち2つのNUMAノードを持つ。各プロセッサは10個の物理コア（20スレッド）と25MBの共有L3キャッシュを持っています。サーバーはCentOS 7.6.1810、10Gbpsイーサネットで接続される。

ワークロードは以下の通り。
CH-benCHmarkを使用したOLTPとOLAPのハイブリッドワークロードを用いる。 ソースコードはオンラインで公開済み。ベンチマークは標準的なOLTP（TPC-C）とOLAP（TPC-H)のベンチマークで構成されている。TPC-Cベンチマークは修正なしのバージョンから構築されている。OLAP部分としてはTPC-Hにインスパイアされた22の分析クエリが含まれ、そのスキーマはTPC-HをCH-benCHmark向けに修正し、さらにTPC-Hにはない3つのリレーションが追加されている。2つのワークロードは複数クライアントから同時に実行される。スループットは、それぞれクエリー/秒（QPS）またはトランザクション/秒（TPS）で測定される。CH-benCHmarkでは、データの単位をウェアハウス（warehouses）と呼び、これはTPC-Cと同様である。100個のウェアハウスは約70GBのメモリを必要とする。

OLTPパフォーマンス

CHbenCHmarkのOLTP部分、すなわちTPC-Cベンチマークの下で、TiDBのOLTPパフォーマンスを楽観的または悲観的ロックの両方で評価する。ここではTiDBのパフォーマンスを、別の分散NewSQLデータベースであるCockroachDB（CRDB）と比較している。 CRDBは6台のサーバ上に配置され、一方TiDBはSQLエンジンとTiKVが6台のサーバー上に配置されて、それらのインスタンスはそれぞれのサーバー上で別々に2つのNUMAノードにバインドされている。PDは6台のサーバーのうち3台に配置される。リクエストをバランスさせるために、TiDBとCRDBの両方にHAProxyを用いてアクセスする。様々な数のクライアントを使用して、50、100、200のウェアハウスでスループットと平均レイテンシを測定している。
※"their instances are bound to the two NUMA nodes separately"とあることから、NUMAノード単位で、つまりCPUのPinningを行ってTiDBとTiKVのインスタンスを配置している。

PingCAP社による修正後のFigure7、論文中のものとは一部異なる

100ウェアハウスのスループットと200ウェアハウスのスループットの傾向は、50ウェアハウスのケースと異なっている。50ウェアハウスでは、256クライアント以下で楽観的/悲観的ロックの両方で、TiDBのスループットがクライアント数に応じて増加している。256クライアント以上では、悲観的ロックの場合のスループットは安定した状態を維持した後に低下し始めるのに対し、楽観的ロックの場合のスループットは512クライアントで最大値に達した後に低下する。100ウェアハウスと200ウェアハウスのTiDBのスループットは増加し続けている。この結果は、同時実行性が高くデータサイズが小さい場合に、リソースの競合が最も激しくなることが予想される。

50ウェアハウスでは楽観的ロックよりも悲観的ロックの方が性能が良いことを示しているが、200ウェアハウスでは逆になる。このような性能の変化は100ウェアハウスでも同様である。これは、悲観的ロックは小さいデータセットで競合が多い場合にうまく動作するが、リソース競合が少ない場合には楽観的ロックの方がうまく動作するからである。

ほとんどの場合、TiDBのスループットはCRDBよりも高く、特に大規模なウェアハウスで楽観的ロックを使用する場合、高いパフォーマンスが得られている。フェアな比較のために悲観的ロックを使用しても（CRDBは常に悲観的なロッキングを使用している）、TiDBの性能は依然として高くなっている。TiDBのパフォーマンス上の優位性は、トランザクション処理とRaftアルゴリズムの最適化によるものと考えている。
※上に提示した修正後のFigure7を参照のこと。本文にある通り、100/200のウェアハウスで楽観的ロックを利用した際の性能が高い。またCRDBは楽観的ロックがデフォルトのはずであり、20.1以降のバージョンで悲観的ロックをサポートする。

また最大スループットに達した後、多くのリクエストがより長い時間待たなければならないため、クライアントで高いレイテンシが発生することも示されている。同様にウェアハウスの数が少ない場合の待ち時間の増加も説明できる。特定のクライアントでは、スループットが高いほどTiDBとCRDBの待ち時間が少なくなり、同じ結果は50/100ウェアハウスでも見られる。

PDからのタイムスタンプの取得がボトルネックになる可能性があるので、その性能も評価する。
持続的にタイムスタンプへ要求を行うために1,200のクライアントを用いた。TiDBをエミュレートして、各クライアントはPDにタイムスタンプの要求をバッチ送信する。6台のサーバ構成では、それぞれが1秒間に602,594個のタイムスタンプを受け取ることができ、これはTPC-Cベンチマークを実行しているときに必要なレートの100倍以上である。TPC-Cを実行しているとき、TiDBはサーバ1台あたり1秒に最大6,000のタイムスタンプを要求する。サーバの数を増やすと、各サーバが受けるタイムスタンプの数は減るが、合計のタイムスタンプ数はほぼ同じである。待ち時間についても、1ミリ秒や2ミリ秒のリクエストはごく一部であり、以上よりPDからタイムスタンプを取得する処理は、現時点でTiDBのパフォーマンス上のボトルネックにはなっていないと結論づけられる。

OLAPパフォーマンス

TiDBのOLAPパフォーマンスを2つの観点から評価する。
第一にCH-benCHmarkのOLAP部分において、100ウェアハウスを用いて、行ストアと列ストアのいずれかを最適に選択するSQLエンジンの性能を評価する。ここでは以下3種類のストレージを設定した。

  • TiKV-only
  • TiFlash-only
  • TiKV & TiFlash

各クエリの平均実行時間（5回ずつ実行）をした結果、1つのストアからのみデータを取得する場合にはTiKVとTiFlashの優劣は付かないが、TiKV & TiFlashからデータを取得するケースでは常により良いパフォーマンスが得られた。

Q8、Q12、Q22では興味深い結果が見られる。TiKV-onlyのケースは、Q8とQ12ではTiFlash-onlyのケースよりも高速だが、Q22では時間がかかる。TiKV & TiFlashのケースでは、TiKV-only/TiFlash-onlyのケースよりもパフォーマンスが良い。

Q12は2テーブルの結合が含まれるが、ストレージタイプごとに物理的な実装が異なる。TiKV-onlyのケースではインデックス結合が使われ、テーブル:ORDER_LINEからいくつかの評価済みタプルをスキャンし、インデックスを使用してテーブル:OORDERを検索する。インデックス読み取りのコストは非常に低く、2つのテーブルから必要列をスキャンするTiFlash-onlyのケースでハッシュ結合を取るよりも優れている。TiKV & TiFlashの場合、TiFlashからORDER_LINEをスキャンし、TiKVでインデックスを使用してOORDERを検索する低コストなインデックス結合を使用するため、コストはさらに削減される。TiKV & TiFlashの場合、列ストアを読み込むことで、TiKV-onlyの場合より実行時間が半分に短縮される。

Q22では*exists()*のサブクエリがアンチセミジョインに変換される。TiKV-onlyの場合はインデックス結合を、TiFlash-onlyの場合はハッシュ結合を使用する。しかしQ12と異なり、インデックス結合がハッシュ結合よりも高コストとなる。TiFlashから内部表をフェッチし、TiKVからのインデックスを使って外部表を検索する場合はインデックス結合のコストが下がるため、ここでもTiKV & TiFlashのケースが最も時間がかからない。

問8はより複雑で、9つのテーブルの結合が含まれる。TiKV-onlyのケースでは、2つのインデックスマージ結合と6つのハッシュ結合を行い、さらにインデックスを使用して2つのテーブル（CUSTOMERとOORDER）を検索する。このプランの所要時間は1.13秒で、TiFlash-onlyの場合の8つのハッシュ結合の所要時間1.64秒よりも優れる。このオーバーヘッドはTiKV & TiFlashのケースではさらに削減され、6つのハッシュ結合でTiFlashからのデータをスキャンする以外は、物理的なプランはほとんど変更されず、実行時間は0.55秒に短縮された。これら3つのクエリでは、TiKV-only、またはTiFlash-onlyを使用するとパフォーマンスが変化し、両方を組み合わせることで最良の結果が得られる。

Q1、Q4、Q6、Q11、Q13、Q14、Q19では、TiFlash-onlyがTiKV-onlyよりもパフォーマンスが良く、TiKV & TiFlashはTiFlash-onlyと同等だが、7つのクエリで理由が異なる。Q1とQ6は主に1テーブルの集約で構成されているため、TiFlashの列ストアの利点が活かされる。Q4とQ11は同一の物理プランが実行される。しかし、TiFlashからのデータスキャンはTiKVよりも低コストのため、TiFlash-onlyの実行時間が短く最適な選択となる。Q13、Q14、Q19には2テーブルの結合が含まれ、ハッシュ結合が実行される。TiKV-onlyではハッシュテーブルを探索する際にインデックス読み取りを用いてますが、TiFlashからのスキャンよりも実行時間が長くなる。

Q9はmulti-joinのクエリです。TiKV-onlyではインデックスを使って一部のテーブルでインデックスマージ結合を行いますが、これはTiFlashでハッシュ結合を行うよりも低コストになる。Q7、Q20、Q21も同様となる。22のTPC-Hクエリのうち、残りの8つのクエリは3つのストレージタイプで同等のパフォーマンスを示している。

さらに、500のウェアハウスを持つCH-benCHmarkの22の分析クエリを使用して、TiSparkをSparkSQL、PrestoDB、Greenplumと比較する（各データベースは6台のサーバーを利用）。SparkSQLとPrestoDBでは、データはHiveに列形式のparquetファイルとして格納される。TiSparkのパフォーマンスは同じエンジンを使用しているためSparkSQLに匹敵するが、圧縮されたparquetファイルのスキャンが低コストのため、SparkSQLはTiSparkをやや上回るケースが多い。しかし、TiSparkが多くの処理をストレージレイヤにプッシュダウンする場合には、その優位性が相殺されてしまうこともある。TiSparkとPrestoDBおよびGreenplumの比較は、SparkSQL（TiSparkの基礎となるエンジン）と他の2つのエンジンの比較となる。しかし、これは本稿の範囲外であり、詳細な議論は行わない。
※論文で提示されているFigure9を確認すると、Greenplumの性能が全般的に高いように見える。

HTAPパフォーマンス

トランザクション処理（TP）と分析処理（AP）の性能を調べるとともに、CH-benCHmark全体をベースとしたハイブリッドワークロードを用いて、トランザクションクライアント（TC）と分析クライアント（AC）を分けてTiDBを評価した。これらの実験は100ウェアハウスで行い、データがTiKVにロードされ、同時にTiFlashにレプリケートされる。TiKVは3つのサーバ上に配置され、TiDBのSQLエンジンインスタンスからアクセスされる。TiFlashは他の3つのサーバに配置され、TiSparkインスタンスと共に配置される。各ベンチマークは10分間実行され、3分間のウォームアップ期間がある。
今回評価ではTPおよびAPのスループットと平均レイテンシを測定する。

様々な数のTP/APクライアントでOLTPのスループットと平均遅延を測定すると、スループットはTPのクライアント数が多いほど増加するが、512クライアント未満で最大値に達する。TPクライアントの数が同じ場合、APクライアントの数が増えると、それがない場合と比較して、TPのスループットが最大10%低下する。これはTiKVとTiFlash間のログレプリケーションが、後述のMemSQLのパフォーマンスとは対照的に、高いレベルで分離を達成していることを示す。
トランザクションの平均レイテンシは上限なしで増加する。これは多くのクライアントがリクエストを発行してもすぐに完了することができず、待ちが発生するためである。

OLAPのスループットとレイテンシの結果から、APリクエストに対するTPの影響が分かる。APクエリは高コストでリソースを奪い合うため、APスループットは16のAPクライアントですぐに最大に達する。このような競合はAPクライアント数が増えるほどスループットを低下させる。ここで同じ数のAPクライアントではスループットはほとんど変わらず、せいぜい5%程度しか低下しない。これはTPがAPの実行に大きく影響しないことを示す。分析クエリの平均レイテンシの増加は、クライアントが増えた際の待ち時間の増加に起因する。

ログレプリケーションの遅延

リアルタイムの分析処理を実現するためには、トランザクションの更新が即座にTiFlashで表示される必要がある。 このデータの鮮度はTiKVとTiFlash間のログレプリケーション遅延によって決定される。トランザクションクライアントと分析クライアントを異なる数でCH-benCHmarkを実行している間にログレプリケーション時間を測定し、10分間のベンチマーク実行中の遅延状況を記録して10秒ごとの平均遅延を算出する。また、10分間のログレプリケーション遅延の分布も示す。

10ウェアハウスではログレプリケーションの遅延は常に300ms以下であり、ほとんどの遅延は100ms以下である。100ウェアハウスで遅延は増加するが、ほとんどが1,000ミリ秒以下であることを示している。遅延の分布を詳細に確認すると、10ウェアハウスではクライアント数によらず、ほぼ99%のクエリで500ms未満であった。100ウェアハウスの場合、2つの分析クライアントでは約99%、32 の分析クライアントでは85%のクエリが1,000ms未満であった。これらの指標は、HTAPワークロードにおいてTiDB が約1秒のデータ鮮度を保証できることを強調する。
※トランザクションの更新がほぼ1秒以内にTiFlashで読み取り可能ということ。

遅延時間はデータサイズに関係していることもわかる。データが多いほど同期されるログが多くなるため、ウェアハウスが多ければ多いほど、遅延時間は大きくなる。さらに遅延は分析クエリの数にも依存するが、TPクライアントの数が多いほど遅延は少なくなる。例えば、32のACは2つのACよりも多くの遅延を引き起こすが、AP分析クライアントの数が同じであれば遅延はさほど変わらない。より正確な結果を表4に示す。100ウェアハウスと2つのACでは80%以上のクエリが100ミリ秒以下になるが、32のACでは50%以下のクエリが100ミリ秒以下になる。これは分析クエリが多いほど、ログレプリケーションの頻度が高くなるためである。

MeMSQLとの比較

CH-benCHmarkを用いてTiDBとMemSQL7.0を比較する。 これはOLTPやOLAPの性能よりも、現状のHTAPシステムの分離に関する課題の明確化を目的とする。MemSQLはトランザクションとリアルタイム分析の両方を大規模に処理する分散型リレーショナルデータベースである。MemSQLはマスター1台、アグリゲータ1台、リーフ4台の計6台のサーバーに配置される。100ウェアハウスをMemSQLにロードし、様々な数のAPクライアントとTPクライアントで、5分間のウォームアップ後に10分間のベンチマークを実行した。

評価の結果は、ワークロードの干渉がMemSQLのパフォーマンスに大きな影響を与えることを示す。特に、APクライアントの数が増えると、トランザクションスループットは著しく（5倍以上）低下する。APのスループットもTPクライアントの数が増えると低下するが、トランザクションクエリは分析クエリのような膨大なリソースを必要としないため、それほど顕著な影響は見られない。

HTAPシステムを構築するための一般的なアプローチは、既存のデータベースからの進化、オープンソースの分析システムの拡張、またはスクラッチからの構築などがある。TiDBはスクラッチから構築であり、アーキテクチャやデータ構成、エンジン、一貫性の保証などの点で他システムとは異なる。

既存のデータベースからの進化
成熟したデータベースの中には、既存の製品をベースにHTAPソリューションを提供できるものがあり、特に分析クエリの高速化に力を注いでいる。これらのデータベースは、一貫性と高可用性を個別に達成するために、個別のアプローチを採用している。一方、TiDBはデータの一貫性と高可用性を達成するために、Raftのログレプリケーションの恩恵を受けている。

Oracleは、業界初のデュアルフォーマットのインメモリRDBMSとして2014年にDatabase In-Memoryオプションを導入した。 このオプションは、通常のトランザクション処理のパフォーマンスを損なうことなく（あるいは改善することなく）、分析クエリのワークロードでパフォーマンスの壁を破ることを目的としている。列形式のストレージは読み取り専用のスナップショットで、ある時点で一貫しており、完全オンラインの再配置メカニズムを使用して更新される。Oracleの最近の活動としては、分散アーキテクチャの高可用性を強化し、耐障害性の高い分析クエリの実行を実現している。

SQL ServerはApolloという列形式のストレージエンジンと、トランザクションワークロード用のHekatonインメモリエンジンの2つの特化したストレージエンジンをコアに統合している。データ移行タスクは、定期的にHekatonテーブルの末尾から圧縮された列ストアにデータをコピーする。SQL Server は列ストアインデックスとバッチ処理を使用して分析クエリを効率的に処理し、データスキャンにSIMDを利用する。

SAP HANAは異なるOLAPクエリとOLTPクエリを効率的にサポートし、それぞれに異なるデータ構造を用いる。OLAPをスケールするために、行形式ストアのデータを非同期で分散された列形式のストアにコピーする。このアプローチはMVCCのデータに1秒未満の可視性を実現する。しかし、エラー処理や一貫性を保つために多くの労力を必要な上に、重要な点として、トランザクションエンジンは単一のノードにしか配置されておらず、高可用性に欠ける。

オープンソースシステムの変革
Apache Sparkはデータ分析のためのオープンソースのフレームワークでで、HTAPを実現するためにはトランザクションモジュールが必要となる。以降に挙げる多くのシステムはこれに従う。TiDBはTiSparkが拡張機能であり、Sparkへの依存は大きくない。TiDBはTiSparkを介さずとも独立したHTAPデータベースとなっている。

WildfireはSparkをベースにHTAPエンジンを構築している。分析とトランザクションの両方のリクエストを、同じ列形式のデータ構造であるParquet上で処理する。同時更新にはlast-write-winsの考えを取り入れ、読み取りではSnapshot Isolationを採用する。高可用性のために、コンセンサスアルゴリズムは用いずに、シャードログを複数のノードにレプリケートする。分析クエリとトランザクションクエリは別のノードで処理できるが、最新の更新処理には顕著な遅延が生じる。Wildfireは、大規模なHTAPワークロードのために統合されたマルチバージョンとマルチゾーンのインデックス手法を用いる。

SnappyDataは、OLTP、OLAP、ストリーム分析のための統合プラットフォームである。高スループットな分析エンジン（Spark）と、スケールアウトされたインメモリのトランザクションストア（GemFire） を統合している。最新の更新は行形式で保存され、分析クエリ向けに列形式に変換される。トランザクションはGemFireのPaxos実装を使用した2PCプロトコルに従っており、クラスタ全体でのコンセンサスと一貫したビューを保証する。

スクラッチでの構築
多くの新しいHTAPシステムは、インメモリコンピューティングを利用してパフォーマンス、最適なデータストレージ、可用性を向上させることを含むハイブリッドワークロードのさまざまな検討を行ってきたが、TiDBとは異なり、高可用性・一貫性・スケーラビリティ・データの鮮度・分離を同時に提供することはできていない。

MemSQLは、スケール可能なインメモリOLTPと高速な分析クエリの両方のエンジンを持つ。MemSQLはテーブルを行または列形式で保存し、データの一部を行形式で保持し、ディスクにデータを書き込む際に高速な分析のために列形式に変換することができる。繰り返し実行されるクエリを低レベルのマシンコードにコンパイルして分析クエリを高速化し、多くのロックフリー構造を使用してトランザクション処理を支援する。しかし、HTAPワークロードを実行しているときに、OLAPとOLTPでパフォーマンスを分離することはできない。

HyPerは、分析ワークロードのスナップショット分離を提供するために、オペレーティングシステムのフォークのシステムコールを使用している。新しいバージョンではMVCC実装を採用し、直列化可能性、高速なトランザクション処理やスキャンを提供する。ScyPerはHyPerを拡張し、論理的または物理的なREDOログを使用して更新を伝播することで、分析クエリをリモートレプリカにスケール可能にしています。

BatchDBは、HTAPワークロード用に設計されたインメモリデータベースエンジンで、用途の異なるレプリカを使用したプライマリとセカンダリのレプリケーションに基づいており、それぞれが特定のワークロード（すなわち、OLTPやOLAP）用に最適化されている。トランザクション用エンジンと分析エンジン間の負荷の相互作用を最小限に抑え、HTAPワークロードの厳しいSLAの下でリアルタイム分析を可能にする。行フォーマットのレプリカ上で分析クエリを実行し、高可用性を実現したものではないことに注意が必要である。

Lineage-based data store(L-Store) は、更新が容易なLineage-based storage architectureを導入することで、単一のエンジンでのリアルタイムな分析処理とトランザクション処理を組み合わせている。このストレージは、書き込みに最適化された列形式のものから読み取りに最適化された列形式のレイアウトへの安定したデータを連携するために、ネイティブのマルチバージョンの列形式のストレージモデルを介して競合のない更新メカニズムを可能にしている。

Pelotonは、self-driving SQLデータベース管理システムで、実行時にHTAPワークロード用のデータ生成に対応させようとしている。リアルタイム分析をサポートするために、ロックフリーのマルチバージョン同時実行制御を使用しているが、設計上はシングルノードのインメモリデータベースである。

Cockroach DBは、高可用性、データの一貫性、スケーラビリティ、そして分離を提供する分散型SQLデータベースである。TiDBと同様にRaftアルゴリズム上に構築され、分散トランザクションをサポートする。Snapshot Isolationより強力な分離レベルである直列化可能性を提供する。しかし、専用のOLAPやHTAP機能はサポートしていない。

CONCLUSION

TiDBは、分散型の行形式のデータストアであるTiKVの上に構築され、Raftアルゴリズムを使用している。TiKVからのログを非同期でレプリケートし、行形式のデータを列形式に変換するリアルタイム分析のためのLearnerを導入している。TiKVとTiFlash間のこのようなログレプリケーションは、わずかなオーバーヘッドでリアルタイムデータの一貫性を提供する。TiKVとTiFlashは、異なる物理リソースに配置して、トランザクションと分析クエリの両方を効率的に処理できる。TiKVとTiFlashは、トランザクションクエリと分析クエリの両方でテーブルをスキャンする際に、TiDBによって最適に選択される。評価結果は、HTAPベンチマークであるCH-benCHmarkの下でTiDBが良好に動作することを示している。
TiDBは、NewSQLシステムをHTAPシステムに進化させるための汎用的なソリューションを提供する。

Discussion