分散実行の概要
OceanBaseデータベースは、Shared-Nothingアーキテクチャに基づく分散システムとして構築されており、分散実行計画の生成と実行機能を備えています。
リレーショナルデータテーブルのデータはパーティション単位でシステム内の各ノードに格納されるため、パーティションをまたがるデータクエリリクエストでは、実行計画が複数のノードのデータを操作できる必要があります。OceanBaseデータベースのオプティマイザーは、クエリとデータの物理的な分散状況に基づいて自動的に分散実行計画を生成します。分散実行計画では、パーティショニングによってクエリ性能が向上します。データベースのリレーショナルテーブルが比較的小さい場合は、パーティショニングは不要です。テーブルが大きい場合は、上位のビジネス要件に基づいて慎重にパーティションキーを選択し、ほとんどのクエリがパーティションキーによるパーティションプルーニングを利用できるようにすることで、データアクセス量を削減する必要があります。
また、関連性のあるテーブルについては、関連キーをパーティションキーとして使用し、同じパーティショニング方式を採用することを推奨します。これにより、テーブルグループを使用して同一のパーティションを同じノードに配置し、ノード間のデータ受け渡しを削減できます。
並列クエリの概要
並列クエリとは、クエリ計画を並列実行することで、各クエリ計画のCPUおよびI/O処理能力を向上させ、単一クエリの応答時間を短縮する技術です。並列クエリ技術は、分散実行計画にもローカルクエリ計画にも適用できます。
単一クエリのアクセスデータが同一ノード上にない場合、データ再分散により関連データを同一ノードに配分して計算を行います。データ再分散のノードを上下の境界として、OceanBaseデータベースの実行計画は垂直方向に複数のDFO（Data Flow Operation）に分割されます。各DFOは指定された並列度のタスクに分割され、並列実行によって実行効率が向上します。
一般的に、並列度が高くなるとクエリの応答時間は短縮され、より多くのCPU、I/O、メモリリソースがクエリコマンドの実行に割り当てられます。大量データのクエリ処理をサポートするDSS（Decision Support Systems）システムやデータウェアハウス型アプリケーションでは、クエリ時間の向上が特に顕著です。
全体として、並列クエリの基本的な考え方は分散実行計画と類似しています。すなわち、実行計画を分解した後、実行計画の各部分を複数の実行スレッドが実行し、一定のスケジューリング手法によって、実行計画のDFO間およびDFO内部での並列実行を実現します。
システムが以下の条件を満たす場合、並列クエリはシステムの処理性能を効果的に向上させることができます：
十分なI/O帯域幅
システムのCPU負荷が低いこと
十分なメモリリソース
システムに追加の並列処理を行うための十分なリソースがない場合、並列クエリを使用したり並列度を高めたりしても実行性能は向上しません。逆に、システムが過負荷の場合、オペレーティングシステムはより多くのスケジューリングを強いられ、例えば実行コンテキストスイッチングによって性能が低下する可能性があります。
通常、DSSシステムでは大量のデータへのアクセスが必要となるため、並列実行によって実行応答時間が向上します。単純なDML操作やデータ量が比較的小さいクエリについては、並列クエリを使用してもクエリ応答時間を顕著に短縮することはできません。
並列クエリと分散クエリの原理
OceanBaseデータベースのデータはシャーディング形式で各ノードに保存され、ノード間はギガビットや10ギガビットネットワークで通信します。通常、各ノードにはobserverと呼ばれるプロセスがデプロイされ、これがOceanBaseデータベースの外部サービス提供の主体となります。以下の図に示すように。
OceanBaseデータベースは、一定の均等化戦略に基づいてデータシャーディングを複数のobserverプロセスに均等に分散します。そのため、並列クエリでは通常、複数のobserverプロセスに同時にアクセスする必要があります。以下の図に示すように。
SQL文のパラレル実行プロセス
ユーザーが指定したSQL文でアクセスするデータが2台以上のOBServerノードにまたがっている場合、パラレル実行が有効になります。ユーザーが接続しているそのOBServerノードは、クエリコーディネーター（QC、Query Coordinator）の役割を担います。実行手順は以下のとおりです：
QCは十分なスレッドリソースを予約します。
QCは、パラレル処理が必要な計画を複数のサブ計画、すなわちDFO（Data Flow Operation）に分割します。各DFOには、直列実行される複数の演算子が含まれます。例えば、あるDFOにはパーティションのスキャン、集計、送信の演算子のタスクが含まれ、別のDFOには収集、集計の演算子などのタスクが含まれます。
QCは一定の論理的順序に従って、DFOを適切なOBServerノードにスケジュールして実行します。OBServerノードでは一時的に補助コーディネーター（SQC、Sub Query Coordinator）が起動し、SQCはそのOBServerノード上で各DFOの実行リソースの申請や実行コンテキスト環境の構築などを担当し、その後DFOを起動して各OBServerノード上でパラレル実行を開始します。
各DFOの実行が完了すると、QCは残りの部分の計算を直列実行します。例えば、パラレルな
COUNTアルゴリズムでは、最終的にQCが各マシン上の計算結果をSUM演算でまとめる必要があります。QCが存在するスレッドは、結果をクライアントに返します。
オプティマイザーはどのようなパラレル計画を生成するかを決定し、QCはその計画を具体的に実行します。例えば、2つのパーティションテーブルのJOINでは、オプティマイザーはルールとコスト情報に基づき、分散型のPARTITION WISE JOIN計画を生成することも、HASH HASHで分散させた分散JOIN計画を生成することもあります。計画が一度確定すると、QCは計画を複数のDFOに分割し、順序立ててスケジュールして実行します。QCの実行手順は以下の図のとおりです。
パラレル度とタスク分割方法
パラレル度（DOP、Degree Of Parallelism）は、1つのDFOを実行するためにいくつのスレッド（Worker）を使用するかを指定できます。現在、OceanBaseデータベースではPARALLELヒントを使用してパラレル度を指定します。パラレル度が決定すると、DOPはDFOを実行する必要がある複数のOBServerノードに分割されます。
スキャンを含むDFOについては、DFOがアクセスする必要があるパーティションと、それらのパーティションがどのOBServerノードに分散しているかを計算し、その後DOPを対応するOBServerに比例して割り当てます。例えば、DOPが6、DFOが120個のパーティションにアクセスする場合、server1に60個のパーティション、server2に40個のパーティション、server3に20個のパーティションがある場合、server1に3つのスレッド、server2に2つのスレッド、server3に1つのスレッドを割り当て、結果として各スレッドが平均20個のパーティションを処理できるようにします。DOPとパーティション数が整数で割り切れない場合、OceanBaseデータベースは長尾をできるだけ短くするために、一定の調整を行います。
各マシンに割り当てられるWorker数がパーティション数を大幅に上回る場合、自動的にパーティション内での並列処理が行われます。各パーティションはマクロブロックを境界として複数のスキャンタスクに分割され、複数のWorkerが実行を競合します。
このような分割能力を抽象化・カプセル化するために、Granuleの概念が導入されました。各スキャンタスクを1つのGranuleと呼び、このスキャンタスクは1つのパーティションをスキャンする場合もあれば、パーティション内のごく小さな範囲をスキャンする場合もあります。以下の図のようになります。
パラレルスケジューリング方法
オプティマイザーがパラレル計画を生成した後、QCはそれを複数のDFOに分割します。以下の図のように、t1テーブルとt2テーブルのHASH JOINは3つのDFOに分割されます。DFO 1とDFO 2はデータのパラレルスキャンを担当し、データを対応するノードにHASHします。DFO 3はHASH JOINを実行し、最終的なHASH結果をQCに集約します。
QCは可能な限り2組のスレッドを使用して計画のスケジューリングを完了するよう努めます。上記の例におけるスケジューリングプロセスは以下のとおりです：
QCはまずDFO 1とDFO 3をスケジューリングします。DFO 1の実行開始後、データのスキャンを開始し、DFO 3にデータを渡します。
DFO 3の実行開始後、最初はHASH JOINでHash Tableを作成するステップでブロックされます。つまり、DFO 1からのデータ収集が完了し、Hash Tableの構築が完了するまで待機します。その後、DFO 3は右側のDFO 2からデータを収集します。この時点でDFO 2はまだスケジューリングされていないため、DFO 3はデータ収集のプロセスで待機します。DFO 1はデータをすべてDFO 3に送信した後、スレッドリソースを解放して終了できます。
スケジューラーがDFO 1のスレッドリソースを回収すると、直ちにDFO 2をスケジューリングします。
DFO 2の実行開始後、DFO 3にデータの送信を開始します。DFO 3はDFO 2から1行のデータを受信するたびにHash Tableで検索し、ヒットした場合は直ちにQCに出力します。QCは結果をクライアントに出力します。
ネットワーク通信方法
関連するChild DFOとParent DFOのペアについて、Child DFOはプロデューサーとしてM個のWorkerスレッドを割り当てられ、Parent DFOはコンシューマーとしてN個のWorkerスレッドを割り当てられます。彼らの間のデータ転送には、M × N個のネットワークチャネルが必要です。以下の図のようになります。
このようなネットワーク通信形態をより深く理解するために、データ転送層 DTL（Data Transfer Layer）の概念を導入します。すなわち、任意の2点間の通信接続はチャネル（Channel）の概念で記述されます。
チャネルは送信側と受信側に分かれており、初期の実装では送信側が受信側に無制限にデータを送信できるようにしていました。しかし、受信側がこれらのデータを即座に処理できない場合、受信側のメモリがオーバーフローする可能性があることが判明しました。そのため、ストリーミング制御ロジックが追加されました。各チャネルの受信側には3つのスロットが予約されており、スロットがデータで満杯になると送信側にデータ送信の一時停止を通知します。受信側でデータが処理されてスロットが空きになると、送信側に送信再開を通知します。