The Past and Present Lives of PolarDB Parallel Query

Background

1. PolarDB

クラウドの台頭は、かつては保守的だったデータベース市場に新たな発展の機会をもたらした。Gartner の予測によると、2022 年までに全データベースの 75% がクラウドプラットフォームにデプロイまたは移行されるとされている。クラウドサービスプロバイダーには、市場で逆転を図る絶好の機会がある。AWS が re:Invent 2020 で発表した Babelfish を見れば、データベース市場に対する同社の野心がどれほど大きいかがわかるだろう。

AWS が 2017 年に発表した Aurora に関する論文 [1] は、クラウドネイティブリレーショナルデータベースの開発トレンドを牽引してきた。中国で最も早くクラウドコンピューティングに参入した Alibaba Cloud も、2018 年に独自のクラウドネイティブリレーショナルデータベースをリリースした。PolarDB は Aurora の設計思想と一致している。PolarDB はクラウド上のインフラストラクチャと深く統合されており、クラウド固有のスケーラビリティ、弾力性、高可用性を顧客に提供することを目指している。同時に、低いレスポンスレイテンシと高い同時実行数、スループットを実現しており、その基本アーキテクチャは以下の通りである。

基盤の分散共有ストレージは、単一マシンのストレージ容量制限を突破し、ユーザーのデータ量の増加に応じて自動的に弾性拡張できる。コンピュートレイヤーは 1 書き込み複数読み取りの典型的なトポロジーを採用し、RDMA が提供する高速リモートアクセス機能を利用して、コンピュートとストレージの分離による追加のネットワークオーバーヘッドを相殺している。

2 課題

上記の図からわかるように、ストレージレイヤーはシングルマシンを大幅に上回るデータ容量（現在 128 TB）を可能にしており、実際の運用では単一テーブルの容量が数 TB に達するケースも見られる。これは従来の MySQL のマスター・スレーブレプリケーション設定では想像もできなかったことだ。同時に、多くのユーザーが統計やレポートなどのビジネスデータに対するリアルタイム分析を求めている。しかし、MySQL に対する一般的な印象は、小規模なトランザクション処理は高速で同時実行数も高いが、分析能力は弱いというものだ。こうしたリアルタイム分析クエリにどう対応すべきだろうか。

3 対応策

まず述べておきたいのは、インターネットの発展とデータ量の爆発的な増加に伴い、一定のデータ分析能力と異種データ処理能力は、トランザクションデータベースにとって標準的な装備となっている点だ。MySQL コミュニティもバージョン 8.0 で独自のクエリ処理機能を強化しており、サブクエリの変換、ハッシュ結合、ウィンドウ関数のサポートなどが含まれる。同時に、PolarDB MySQL のオプティマイザーチームも、統計情報の強化やより多くのサブクエリ変換、クエリキャッシュなど、複雑なクエリの処理能力向上に多大な取り組みを行ってきた。

並列クエリ（Parallel Query）は、PolarDB MySQL がローンチ当初から搭載しているクエリ加速機能である。本質的には、MySQL のクエリ実行がシングルスレッドであり、現代のマルチコア・大容量メモリハードウェアリソースを十分に活用できていないという核心的な問題を解決する。マルチスレッドによる並列実行により、IO と CPU 計算を含む処理時間を短縮し、応答時間を大幅に削減する。ユーザーにとって、1 コアで 10 分かけるよりも 10 コアで 1 分で完了できるクエリの方が、はるかに意味がある。また、成熟した商用データベースはすべて並列クエリ機能を備えている。


並列クエリの概要

1 特徴

並列クエリは、PolarDB MySQL のコンピュートレイヤーにおいて最も重要かつ複雑な機能コンポーネントと言える。PolarDB のローンチ以来、長年にわたり安定してオンラインで稼働しており、継続的に進化を続けている。以下の特徴を備えている。

MySQL コードベースに完全に基づいており、ネイティブ MySQL と 100% 互換である。これには以下が含まれる。

構文の互換性
データ型の互換性
動作の互換性
追加コスト 0、製品リリースに標準搭載

追加のストレージリソース不要
追加のコンピュートノード不要
メンテナンスコスト 0、通常のクエリと使用方法に違いはないが、応答がより高速である

クラスターにデプロイされ、すぐに使える
ビジネスへの侵入なし
単一のパラメーター設定（並列度）
リアルタイム分析、PolarDB のネイティブな一部として、REDO 物理レプリケーションの低レイテンシの恩恵を受ける

基盤のトランザクションデータを統合
送信して確認
優れたパフォーマンス、PQ の継続的な改善により、分析オペレーターと複雑なクエリ構造のサポート能力が継続的に向上している

全オペレーターの並列化
効率的なパイプライン
複雑な SQL 構造のサポート
安定性と信頼性、エンタープライズ向け機能として疑いの余地はない

MySQL テストフレームワークの拡張
長年のオンライン運用の蓄積
包括的な診断システム

上記は宣伝文句のように聞こえるかもしれないが、まさに並列クエリのコアコンピタンスである。

2 進化

並列クエリの機能は継続的に蓄積されている。初期の PQ1.0 から PQ2.0 へと進化し、ノード間並列処理の研究開発段階に入り、まもなくオンラインでリリースされる予定である。ここではノード間並列処理の能力については紹介せず、既存のオンライン稼働状況に焦点を当てる。

PQ1.0

最初にリリースされた並列クエリ機能の基本的な考え方は、計算のプッシュダウンであり、できるだけ多くの計算を複数のワーカーに分散して並列で完了させることで、IO などの重い操作を同時に実行できるようにする。ただし、一般的な share-nothing 分散データベースとは異なる。基盤の共有ストレージにより、PolarDB の並列処理におけるデータ分割は物理的ではなく論理的である。各ワーカーはテーブルデータの全体量を参照できる。エグゼキューター部分については、後ほど論理分割について紹介する。

並列分割のプランの典型的な形式は以下の通りである。

いくつかの特徴が見て取れる。

実行モードはシンプルなスキャッターギャザーであり、単一のプランスライスで、複数のワーカーが同じ機能を完了してリーダーに集約する
オペレーターを可能な限りワーカーにプッシュダウンする
リーダーはプッシュダウンできない計算の完了を担当する
この方式は多くのオンラインスロークエリ問題を解決し、良好な高速化効果を得ることができるが、一定の制約もある。

プラン形式が単一であるため、オペレーターの並列モードも単一になる。たとえば、group by と集約は、2 段階集約（ワーカーで部分集約、リーダーで最終集約）でしか完了できない
集約操作がリーダー側で完了する場合、distinct、ウィンドウ関数、order by などがあると、リーダーでのみ完了でき、シングルポイントのボトルネックを形成する
データスキューが存在する場合、一部のワーカーに処理対象のデータがなくなり、並列スケーラビリティが低下する
さらに、実装面でも改善の余地がある。たとえば、一部のオペレーターが並列化に対応しておらず、一部の複雑なクエリネスト構造が並列化に対応していない
全体的に見ると、PQ1.0 の並列形式は PostgreSQL コミュニティの方式に類似しており、まだ改善の余地がある。商用データベースの並列形式はより柔軟で複雑であるからだ。

PQ2.0

PQ2.0 は前述の制約を補い、実行モードを SQL Server に合わせて、より強力な多段並列処理を実現している。

典型的なプランは以下の通りである。

まず目にする変化は、ここに複数のワーカーグループが存在することである。PQ2.0 の実行計画は多段的であり、プランはいくつかのプランスライスに分割される。各スライスは 1 グループのワーカーが並列で完了し、中間結果はデータ交換チャネルを通じてスライス間で渡され、後続スライスのパイプライン実行がトリガーされる。主な改善点は以下の通りである。

新しいコストベース並列オプティマイザーは、統計情報とコストに基づいて最適なプラン形式を決定する

すべてのオペレーターに対する並列サポート。前述の複雑な多層ネスト構造を含む、すべての構造を完全に並列化できる

エクスチェンジオペレーターの導入。シャッフルやブロードキャストなどのデータ分散操作をサポートする

一定の自己適応機能を導入。並列最適化が完了した後でも、リソース負荷状況に応じて動的な調整が可能。たとえば、シリアルへのロールバックや並列度の低減など

これらの変化は何を意味するのか。シンプルで実践的な例を見てみよう。


上記のシンプルなクエリに対して、最適化後、PQ1.0 は図中の実行計画を生成する。

結合テーブルセットの中から、論理シャーディングに使用できるテーブルを探す。3 つのテーブルで十分なシャードを分割できない場合は、最も多くのシャードを分割できるものを選ぶ。たとえば、ここで t2 が選択されたが、12 個のシャードに分割できても、並列度 16 の要件を満たせず、4 個のワーカーがデータを読み取れずアイドル状態になった。
集約操作は、まずワーカーで部分集約を行い、リーダーで集約する。各ワーカーでのグループ集約が十分でない場合、リーダーは依然として多数のグループを受け取り、重い集約計算を処理しなければならない。リーダーの計算が遅いと、ワーカーからのデータ受信が追いつかず、ワーカーの実行速度に逆圧をかけ、クエリ全体の低速化を招く。

PQ2.0 の実行計画は以下の通りである

データシャーディングは t2 に限定されるが、12 個のワーカーは t1 join t2 の操作のみを完了すればよい。結合完了後、データ量は一般に増加し、より多くの中間結果がシャッフル（リパーティション）を通じて後続のスライスに再分散され、より高い並列度で t3 との結合を完了できる
各ワーカーがローカル集約を完了した後、まだ多くのグループが残っている場合は、group by キーに基づいてシャッフルを行い、データを次のスライスに分散できる。次のワーカーグループが重い集約操作を並列で完了し、後続の order by の部分的なソートを行い、最終的なリーダーはマージソートの集約のみを実行すればよい

これにより、シングルポイントのボトルネックとデータ量不足によるスケーラビリティの問題が解決され、線形アクセラレーションを実現する。

なぜ線形スケーリングがそれほど重要なのか。

上記の図からわかるように、並列度が上がるにつれて E2E の応答時間が線形に減少する。これは顧客にとって 2 つの重要な効果をもたらす。

ビジネスの成長とデータの拡大に合わせて、並列度を上げることでマッチングするコンピュートリソースを利用し、安定的で予測可能なクエリパフォーマンスを継続的に獲得できる
一貫して高速な分析時間が迅速なビジネス判断を促進し、急速に変化する市場環境でビジネスの競争力を維持する
完全な線形アクセラレーションは、Parallel RT = Serial RT / CPU cores であるが、これは現実的ではない

3 アーキテクチャ

並列クエリコンポーネントの全体アーキテクチャは以下の通りである

コア部分は上から下へ 3 つのレイヤーで構成されている。

コストベース並列オプティマイザー。MySQL オプティマイザーフレームワークに組み込まれ、並列最適化部分を完了する

並列プランジェネレーター。抽象並列プラン記述に基づいて、ワーカーが実行可能な物理実行計画を生成する

並列エグゼキューター。オペレーター内の並列関数やデータ分散機能などを含む並列実行コンポーネント

各コンポーネントの実装については後ほど詳しく説明する。

4 パフォーマンス

個人的な記事であるため、具体的な実行時間はここでは伏せる（オンラインで検索可能）。主に PQ2.0 のクエリ高速化能力を見ていく。ここでは並列度を 32 としている（Q6/Q12 のスピードアップ比が 32 を超える理由については、後ほど詳しく触れる）。

全体の結果は以下の通りである。100% の SQL が高速化可能で、総合スピードアップ比は 18.8 倍である。

5 使用方法
使いやすさの観点から見ると、ユーザーは 1 つのパラメーターを設定するだけで並列クエリを有効にできる。


並列実行計画を確認したい場合は、通常のクエリと同様に EXPLAIN / EXPLAIN FORMAT=TREE を実行するだけでよい。


EXPLAIN には並列関連情報を表示するための拡張が加えられており、コスト、並列モード、分散方法などが含まれる。

並列クエリの実装

以上が技術的な詳細を含まない一般的な内容である。次の章からは各モジュールに順次深入りしていく。

1 並列オプティマイザー

PQ2.0 ではプラン形式が多様化するため、単純なルールと単純な統計情報のみに依存した分割プランでは、最適な解を得ることが難しい。そのため、コストベースの並列オプティマイザーを再実装した。

基本的な流れは、MySQL のシリアル最適化の後に、さらに並列分割を実行するというものである。ここで、なぜ Greenplum のように最適化プロセス中でシリアルと並列の実行戦略を統合的に考慮しないのかと疑問に思う方もいるだろう。その理由は、MySQL の最適化プロセスにはサブステップ間の明確な境界がなく、深く再帰的な結合順序付けアルゴリズムとその中に埋め込まれたセミ結合最適化戦略の選択が、コードロジックと構造をより複雑にしているため、ネイティブコードへの大規模な侵入なしでは統合最適化を実現できないからである。コミュニティコードを大きく損なうと、後続のバージョンイテレーションに追従できなくなり、コミュニティの恩恵を受けられなくなる。

そのため、2 段階の最適化プロセスを採用している。これは業界でも一般的な手法であり、Spark、CockroachDB、SQL Server PDW、Oceanbase なども同様の方式を採用している。

コストモデルの強化

コストベースの最適化であるため、各オペレーターの並列実行に関するコスト情報を取得できるようにする必要がある。そのため、PolarDB は統計情報の大幅な強化を行った。

統計情報の自動更新
シリアル最適化プロセスで、並列実行向けの補強を行う。たとえば、テーブルスキャン方法の変更など。これが、上記のパフォーマンスデータで Q6/Q12 が超線形スピードアップ比を示す理由である
全オペレーターの統計情報導出 + コスト計算。一連のコスト式とカーディナリティ推定導出メカニズムで補完

ここでは統計情報強化の効果のみを示す。メリットは並列クエリだけでなく、シリアル実行にも及ぶ。

適応型実行ポリシー

初期バージョンでは、シリアル最適化と並列最適化の間、および並列最適化と並列プラン生成の間に一定の結合度があり、並列度の高すぎる設定により多数のワーカースレッドが同時に消費され、CPU 使用率が急上昇する問題があった。新しい並列オプティマイザーはこれらの問題を解決した。

シリアル最適化と並列最適化を分離。並列最適化は抽象オペレーターツリーを再構築し、これを入力として列挙を開始する
並列最適化と並列プラン生成を分離。最適化結果はプランサブフラグメントの抽象記述であり、これをプラン生成に出力する
これにより実行戦略の柔軟性が可能になり、リソース不足時にシリアルへの復帰、並列度の低減、またはスケジューリングキューでのリソース待ち合わせが許可される

コストベースの網羅的列挙

これは比較的大きなトピックである。一般的に、並列最適化は動的計画法に基づくボトムアップの網羅的列挙プロセスである。実装の思路は SQL Server PDW 論文 [2] を参考にしている。サブごとに、可能な並列実行方法とデータ分散方法を列挙し、出力データの物理プロパティ（distribution + order）に基づいて物理等価クラスを構築することで、ローカルプルーニングを行い、ローカル部分問題の最適解を取得して上位レイヤーに渡し、最終的にルートオペレーターでグローバル最適解を取得する。

下図は、オペレーター t1 NLJ t2 の列挙プロセスの簡単な例である。

全体の列挙が完了すると、プラン空間にデータ分散 Exchange Enforcers を含む一連の物理オペレーターツリーが生成され、コストに基づいて最適なツリーを選択できる。その後、Enforcer をサブプランの分割点としてサブプランを構築し、一連の実行計画の抽象記述をプランジェネレーターに出力する。

2 並列プラン生成

実装の観点から見ると、並列プラン生成はコンポーネント全体の中で最も複雑度が高く、最も落とし穴の多い部分と言える。ここでは物理プランクローンのメカニズムを採用している。つまり、オプティマイザーが生成した並列プラン記述に従って、各プランフラグメントの物理実行計画を元のシリアルプランからクローンする。

なぜこの方法を採用するのか。それは MySQL 自体のメカニズムに関連している。MySQL の最適化と実行は密接に結合されており、明確な境界がない。つまり、最適化プロセス中に関連する実行構造が構築される。そのため、独立したプラン記述に基づいて各物理実行構造を直接構築する方法がなく、シリアルプランから「クローン」するしかない。これがすべての複雑さの根源と言える。

MySQL の実行構造は非常に複雑である。式（Item）とクエリブロック（SELECT_LEX）の相互参照、内部クエリと外部クエリの関連付け（Item_ref）などが、このタスクをさらに困難にしている。この過程で、チームは MySQL の最適化実行構造について深い理解を得て、コミュニティの多くのバグも発見した。

上図のシンプルなクエリを例に説明する

コミュニティはイテレーターモデルに基づいてエグゼキューターをリファクタリングしたが、本質的には物理実行計画は依然として QEP_TAB で構成されるシーケンスであり、group by と集約は tmp table1 で完了し、order by は tmp table2 で完了する。

プラン生成には 2 つの核心的な操作がある。

シリアル物理プランとサブスライスの記述に基づいて、対応する構造を各ワーカースレッドにクローンする。上図の右下部分に示すように、ワーカーで実行される t1 join t2 とプッシュダウン集約操作がクローンされる。

参照関係の修正

元のシリアルプランはリーダープランに変換する必要があるため、不要な実行構造を置き換え、一部の参照関係を調整する必要がある。上図の右上部分に示すように、t1 join t2 と一部の集約操作がプッシュダウンされたため、リーダーは不要な構造を削除し、ワーカーから渡されたデータをコレクターテーブルから読み取るように置き換える必要がある。同時に、後続のステップで参照される t1/t2 テーブルの構造を、コレクターテーブルを参照する対応する構造に変換する必要がある。

ここでは最もシンプルな例のみを示しており、サブクエリや多段プランは含まれていない。実際のプロジェクト実装コストはこれよりはるかに高い。

3 並列エグゼキューター

PQ はオペレーター内の一連の並列メカニズムを実装している。たとえば、テーブルの論理パーティショニングと並列スキャン、並列ハッシュ結合などがあり、並列実行を可能にし、パフォーマンスをさらに向上させる。また、さまざまなサブクエリ処理メカニズムも実装している。ここでは代表的なものを選んで紹介する。

並列スキャン

PolarDB は共有ストレージであり、すべてのデータはすべてのノードから可視である。これはシャーディング分散システムとは異なる。どのワーカーがどのデータを処理するかを事前に決定できないため、論理パーティショニング方式を採用している。

btree レベルでデータを多数の小さなフラグメントに分割し、異なるワーカーが異なるフラグメントを担当して並列実行をトリガーする。以下にいくつかの最適化ポイントを示す。

粒度の細かい分割を行い、シャード数 >> ワーカー数とした上で、ワーカーがラウンドロビンでシャードを「取得」して実行する。これは共有ストレージシステムの天然のメリットである
分割時にリーフノードまで降りる必要はなく、ページを最小パーティション単位として使用することで、初期パーティション速度を高速化する

ハッシュ結合はコミュニティ 8.0 で導入された分析クエリ高速化機能であり、バージョンが進化するにつれてセミハッシュ結合、アンチハッシュ結合、レフトハッシュ結合などをサポートしている。PolarDB もこれらのパッチを導入して完全なハッシュ結合機能を実現し、さまざまな並列実行戦略を実装している。

並列ハッシュ結合は build と probe の両フェーズでサポートされている

build フェーズでは、複数のワーカーが同じ共有ロックフリーハッシュテーブルにデータを挿入する
probe フェーズでは、複数のワーカーがハッシュテーブルを並列で検索する
2 つのフェーズ間に重複がないため、フェーズ全体の並列化が実現する。ただし、並列ハッシュ結合にも固有の問題がある。たとえば、共有ハッシュテーブルが大きすぎてディスクへのスピル問題が発生することや、ロックフリーの parallel insert でも同期プリミティブによるキャッシュ無効化が生じることなどがある

パーティションハッシュ結合

パーティションハッシュ結合は上記の問題を回避できるが、データシャッフルのオーバーヘッドが発生する。


build と probe の両側で join キーに基づいてシャッフルを実行し、データをターゲットパーティションに分散させる
各パーティション内で、build 側が小規模なハッシュテーブルを構築する
各パーティション内で、probe 側が対応するハッシュテーブルを検索する
このようにして、各パーティション内で同一配置結合が完了し、各ハッシュテーブルが小さくなることでディスクへのスピルを回避できる。さらに、build 時の競合問題もない

上記 2 つの方式のどちらが優れているかは、並列オプティマイザーがコストに基づいて判断する。

サブクエリの並列化 - pushdown exec

ここで、サブクエリは式の一部であり、select list、where、having などの句に存在できる。
相関サブクエリの場合、並列化する唯一の方法は、外層が依存するデータ（テーブル）をワーカーにプッシュダウンし、各ワーカー内で完全に実行することである。ただし、外層が並列化されているため、各ワーカーでのサブクエリ実行回数は依然として比例して減少する。

EXISTS サブクエリは各ワーカーに完全にクローンされ、WHERE 条件の評価に合わせて繰り返しトリガーされる。

サブクエリの並列化 - pushdown shared

この並列サブクエリは式の一部にも、派生テーブルにもなり得る。

大まかに言えば、この並列方式は非相関サブクエリに適しているため、事前に並列でマテリアライズして一時結果テーブルを形成できる。後続の外層が並列実行される際に、各ワーカーはサブクエリを参照する際に直接そのテーブルから並列で結果データを取得できる。


さらに、オンラインユーザーのレポートクエリでは、派生テーブルの多層ネストが非常に一般的なクエリモードである。この種の SQL に対して、プッシュダウン共有戦略は並列実行のパフォーマンスを向上できる。以下の例を示す。

上図の各色の四角は 1 層のクエリブロックを表し、多層派生テーブルのネストロジックを構成している。一部の層は UNION ALL で集約され、一部の層は複数のテーブル（派生テーブルを含む）の結合になっている。このようなクエリに対して、MySQL は各派生テーブルに必要なマテリアライズを実行し、外層で一時結果テーブルを形成して後続の計算に参加する。PQ2.0 はこの一般的なクエリモードに対してより汎用的なサポートを提供し、各層のクエリの実行が並列で実行されることで、線形アクセラレーション効果の達成を目指している。

エクスチェンジ

効率的で柔軟な実行計画を生成するには、データ分散コンポーネントが不可欠である。現在、PolarDB はシャッフル、ブロードキャスト、ギャザーの 3 種の分散方式をサポートしており、ロックフリー共有リングバッファを使用して、パイプラインモードで効率的なデータ転送を実現している。

下図はシャッフル（リパーティション）の基本形式を示している

ここまでで、並列クエリのオンラインバージョンの機能と実装について概説した。

成熟したエンタープライズ向け機能として、チームは製品の使いやすさを向上させる一連の補助ツールを実装しており、監視、介入、フィードバックの機能を実現しているが、紙面の都合でここでは紹介しない。

今後の計画

ここでの今後の計画は正確ではない。チームは既にノード間並列処理について多くの作業を行い、開発サイクルの終盤に入っているためである。ノード間並列処理は、大規模データに対する複雑なクエリ機能を新たなレベルに引き上げる。

* ノード間のコンピュートリソースを開放し、より高いコンピューティング並列度を実現する
* IO と CPU におけるシングルノードのボトルネックを突破し、分散ストレージの高いスループットを最大限に活用する
* グローバルノード管理とリソースビューを組み合わせて、グローバルコンピュートリソースを均等に分散・スケジューリングし、ロードバランシングを実現しながらクエリパフォーマンスを保証する
* グローバル整合性ビューと連携して、トランザクションデータの正確な読み取りを確保する