Interpretation of PolarDB-X source code: the life of DDL

本記事では、DDL エンジンの視点から、DDL エンジンのアーキテクチャと実装、および DDL エンジンと DDL ジョブの間のインタラクションロジックを紹介します。本記事を読む前に、以下の記事を一読することをお勧めします。
・「DDL の生涯（パート 1）」
・「PolarDB-X DDL は ACID も追求する？」
DDL エンジン関連の概念
DDL ジョブ
DDL ジョブは DDL エンジン内の概念で、論理的な DDL を記述するために使用されます。DDL エンジンでは、1 つの DDL ジョブが 1 つの論理 DDL に対応し、DDL ジョブには論理 DDL の実行に必要な一連のアクションが含まれています。したがって、DDL エンジンフレームワークの下で、開発者が新しい論理 DDL をサポートすることは、本質的に新しい DDL ジョブを定義することです。
DDL 開発者は静的な DDL ジョブを定義します。ただし、DDL ジョブは実行時に状態属性も持ちます。この属性は主に DDL エンジンによって管理されます。もちろん、ユーザーは限定的な DDL 運用保守コマンドを実行して DDL ジョブのステータスを管理し、DDL の実行プロセスを管理することもできます。次の図は DDL ジョブの状態遷移図です。図の太い黒線は DDL ジョブ実行の初期状態と最終状態を示しています。実行可能な運用保守コマンドは、各 DDL ジョブの状態間の接続線上にマークされています。
DDL タスク
DDL タスクは DDL ジョブ内部の一連の動作のカプセル化です。たとえば、metaDb の読み書き、メモリ内での計算、プロセス間通信、物理 DDL を DN に送信して実行させるなどが該当します。これらの動作はそれぞれ DDL タスクとしてカプセル化されます。したがって、1 つの DDL ジョブはいくつかの DDL タスクで構成されます。これらのタスクは、DDL エンジンによって一定の順序でスケジューリングおよび実行される必要があります。DDL 開発者は、PolarDB-X の DDL エンジンが提供する DAG グラフフレームワークを使用して、タスク間の依存関係と実行順序を記述できます。DDL エンジンフレームワークの下で、開発者が新しい DDL ジョブを定義することは、本質的にいくつかの DDL タスクを定義し、それらを DAG グラフで組み合わせることです。
DDL タスクは、DDL エンジンが DDL の近似原子性を実現するための重要な手段であり、DDL の原子性は DDL エンジンが追求する目標です。論理 DDL の実行には一連の操作が伴い、原子性はそれらの操作がすべて有効になるか、またはすべて有効にならないかのいずれかであることを要求します。具体的には、DDL エンジンは各 DDL タスクに冪等性を要求し、各タスクには対応する逆冪等メソッドが必要です（このメソッドはタスクのロールバック時に DDL エンジンによって呼び出されます）。DDL エンジンは DDL の実行前に、その DDL の DDL タスクで構成される DAG グラフを生成して MetaDb に永続化します。これは DDL の原子性を保証するアンドゥログに相当します。DDL エンジンは DAG グラフに従ってタスクを順次実行し、DDL ジョブ全体が正常に実行されるか、完全にロールバックされるまで処理を続けます。
Worker と Leader
DDL エンジンの視点では、CN ノードは Worker ノードと Leader ノード（クラスター内に 1 つだけ存在）に分類されます。Worker ノードはユーザーから送信された DDL リクエストの受信を担当します。受信したリクエストに対して簡単なローカル検証を実行し、DDL を DDL ジョブに変換して MetaDb にプッシュした後、Leader ノードに通知して MetaDb から DDL タスクをプルさせます。
Leader ノードは DDL の実行を担当します。MetaDb から DDL ジョブをプルした後、DAG グラフの形式に復元し、ジョブ内のタスクに対してトポロジカルソートを実行してから、一定の並列度でタスクをスケジューリングして実行します。
DDL エンジンのソースコードディレクトリ
以降の説明のため、本記事ではまず DDL エンジンのソースコードのディレクトリ構成を説明します。PolarDB-X の DDL エンジンのソースコードは com.alibaba.polardbx.executor.ddl.newengine に配置されており、各モジュールの説明は以下の通りです。
サブディレクトリまたは主要クラス
機能
job
ジョブおよびタスクオブジェクトの定義
dag
ノードとグラフの定義、トポロジカルソート、DAG のメンテナンスと更新を含む、汎用 DAG およびトポロジカルソートの実装
meta
GMS 内の永続オブジェクトの読み書きインターフェイス。永続オブジェクトにはジョブとタスクのステータス、システムリソース（永続読み書きロック）を含む
sync
Leader ノードと Follower ノード間の情報同期を実現する同期インターフェイスの提供
utils
スレッド、スレッド間通信、およびスレッドプールのカプセル化
serializable
ジョブおよびタスクオブジェクトのシリアル化インターフェイス
DdlEngineDagExecutor
タスクスケジューリング、タスクステータス監視、および例外処理のメインロジックを含むジョブのエグゼキュータ
DdlEngineScheduler
ジョブを実行キューに投入し、ジョブのエグゼキュータを呼び出すジョブのスケジューラ
DdlEngineRequester
DDL リクエストを処理する DDL エンジンのエントリポイント。DDL ジョブを永続化し、Leader ノードに通知して DDL リクエストを処理させる
example
次に、DDL エンジンの視点から、論理 DDL が DDL エンジンによってどのようにスケジューリングおよび実行されるかを説明します。
DDL タスクのスケジューリング
クライアント側の MySQL Client から DDL 文が送信されると、Worker ノードが DDL 文を受信し、簡単なオプティマイザを通じて解析して LogicalPlan を取得した後、対応する DDL ハンドラに割り当てます。DDL ハンドラは DDL ジョブの生成を担当します。次に、DDL ハンドラの公共基底クラスのインターフェイス com.alibaba.polardbx.executor.handler.ddl.LogicalCommonDdlHandler#handleDdlRequest が DDL リクエストを処理します。この関数は com.alibaba.polardbx.executor.ddl.newengine.DdlEngineRequester#execute メソッドを呼び出し、生成された DDL ジョブと DDL の実行に必要なコンテキストを MetaDB に書き込み、Leader ノードに通知して処理させます。これで Worker ノードの処理は完了です。DDL がブロッキングの場合、Worker ノードは Leader が DDL の実行を完了するのを待ってからクライアントに応答を返します。ノンブロッキングの場合、Worker ノードはそのまま応答を返します。
Leader ノードでは、com.alibaba.polardbx.executor.ddl.newengine.DdlEngineScheduler#ddlDispatcherThread と com.alibaba.polardbx.executor.ddl.newengine.DdlEngineScheduler#ddlSchedulerThread の 2 つのスレッドが動作しており、それぞれインスタンスレベルの DdlJobDispatcher とスキーマレベルの DdlJobScheduler に対応します。DdlJobDispatcher はグローバルで一意な Ddl リクエストキューから Ddl リクエストを取り出し、スキーマレベルの DDL ジョブキューに割り当てます。DdlJobScheduler はスキーマレベルで動作し、スキーマレベルの Ddl ジョブキューから DDL ジョブを継続的に消費します。この処理中、DdlJobScheduler はスキーマレベルのセマフォを使用して DDL ジョブの並列消費の並列度を制御します（同じスキーマでの最大スレッド数は 10 です）。DdlJobScheduler が DDL ジョブを消費することは、実質的にスキーマレベルの DDL ジョブキューから DDL ジョブを取り出し、DdlJobExecutor（ジョブレベル）に割り当てることです。DdlJobExecutor は DDL ジョブを DdlEngineDagExecutor に渡す役割を担います。これで DDL ジョブは正式に DDL エンジン内のエグゼキュータ DdlEngineDagExecutor に移され、後者が DDL ジョブの実行を引き継ぎます。
補足として、上記の説明から分かるように、DDL エンジンは複数の DDL の同時実行をサポートしています。同じリソースを必要とする DDL 間の相互排他を確保するため、DDL エンジンは永続読み書きロック機構を提供しています。DDL 開発者は DDL ジョブを定義する際に、DDL が必要とするスキーマとテーブルのリソースを事前に宣言するだけで済みます。DDL の実行時、DDL エンジンは com.alibaba.polardbx.executor.ddl.newengine.DdlEngineRequester#execute が DDL ジョブを生成して MetaDB に保存する前に、DDL ジョブが必要とするリソースに応じて読み書きロックを取得します。
DDL タスクの実行
DdlEngineDagExecutor は DDL タスクの実行を担当します。restoreAndRun メソッドを呼び出して MetaDb から DDL ジョブをプルし、DAG 形式に復元します。次に run メソッドを呼び出し、DDL ジョブの現在の状態に応じた対応するコールバックメソッドを実行します。
public class DdlEngineDagExecutor {

public static void restoreAndRun(String schemaName, Long jobId, ExecutionContext executionContext){
boolean restoreSuccess = DdlEngineDagExecutorMap.restore(schemaName, jobId, executionContext);
DdlEngineDagExecutor dag = DdlEngineDagExecutorMap.get(schemaName, jobId);
dag. run();
}

private void run() {
// Start the job state machine.
if (ddlContext. getState() == DdlState. QUEUED) {
onQueued();
}
if (ddlContext. getState() == DdlState. RUNNING) {
onRunning();
}
if (ddlContext. getState() == DdlState. ROLLBACK_RUNNING) {
onRollingBack();
}
// Handle the terminated states.
switch (ddlContext. getState()) {
case ROLLBACK_PAUSED:
case PAUSED:
onTerminated();
break;
case ROLLBACK_COMPLETED:
case COMPLETED:
onFinished();
break;
default:
break;
}
}
}
com.alibaba.polardbx.executor.ddl.newengine.DdlEngineDagExecutor#run は、DDL ジョブの現在の状態に応じて対応するコールバックメソッドを実行します。これは本質的に DDL ジョブの状態遷移図上を遷移するプロセスです。
DDL ジョブの初期状態は一般的に QUEUED で、DDL エンジンによってスキーマレベルのキューにスケジュールされたことを示します。このとき、run メソッドはこの状態に基づいて onQueued() メソッドを呼び出します。onQueued() メソッドの役割は、DDL ジョブのステータスを RUNNING に変更することです。
DDL ジョブの現在の状態が RUNNING の場合、run メソッドは onRunning コールバックメソッドを呼び出し、DAG グラフの依存関係に従って DDL ジョブ内のタスクを実行します。
private void onRunning() {
while (true) {
if (hasFailureOnState(DdlState. RUNNING)) {
if (waitForAllTasksToStop(50L, TimeUnit. MILLISECONDS)) {
LOGGER.info(String.format("JobId:[%s], all tasks stopped", ddlContext.getJobId()));
return;
} else {
continue;
}
}
if (executingTaskScheduler. isAllTaskDone()) {
updateDdlState(DdlState. RUNNING, DdlState. COMPLETED);
return;
}
if (executingTaskScheduler. hasMoreExecutable()) {
// fetch & execute next batch
submitDdlTask(executingTaskScheduler. pollBatch(), true, executingTaskScheduler);
continue;
}
//get some rest
sleep(50L);
}
onRunning の処理フローは以下の通りです。
・まず現在の DDL ジョブのステータスが RUNNING かどうかを確認し、そうでない場合はそのまま戻ります。
・現在の DAG グラフ上に未実行のタスクノードがあるかどうかを確認し、ない場合はジョブのステータスを COMPLETED に更新して戻ります。
・現在の DAG グラフ上に実行可能なタスクがある場合、トポロジカルソートを使用して DAG グラフからすべての実行可能なタスクを抽出し、並列度の制限に従って submitDdlTask メソッドを呼び出して並列実行します。なお、タスクが正常に実行されない場合もあります。タスクの実行に失敗した場合、submitDdlTask メソッドはタスクの開発者が事前定義した障害戦略に従って、現在の DDL ジョブのステータスを変更します。最も典型的なケースでは、タスクが失敗すると現在の DDL ジョブのステータスを PAUSED または ROLLBACK_RUNNING に変更します。詳細なエラー処理とリカバリ機構については次のセクションで説明します。
DDL ジョブの状態が ROLLBACK_RUNNING の場合、run メソッドは onRollingBack() コールバックメソッドを呼び出して DDL のロールバックを実行します。関連コードは以下の通りです。
private void onRollingBack() {
if (。allowRollback()) {
updateDdlState(DdlState. ROLLBACK_RUNNING, DdlState. ROLLBACK_PAUSED);
return;
}

reverseTaskDagForRollback();

// Rollback the tasks.
while (true) {
if (hasFailureOnState(DdlState. ROLLBACK_RUNNING)) {
if (waitForAllTasksToStop(50L, TimeUnit. MILLISECONDS)) {
LOGGER.info(String.format("JobId:[%s], all tasks stopped", ddlContext.getJobId()));
return;
} else {
continue;
}
}
if (reveredTaskScheduler. isAllTaskDone()) {
updateDdlState(DdlState. ROLLBACK_RUNNING, DdlState. ROLLBACK_COMPLETED);
return;
}
if (reveredTaskScheduler. hasMoreExecutable()) {
// fetch & execute next batch
submitDdlTask(reveredTaskScheduler.pollBatch(), false, reveredTaskScheduler);
continue;
}
//get some rest
sleep(50L);
}
}
onRollingBack の処理フローは以下の通りです。
・まず現在の DAG グラフの実行進行状況でロールバックが許可されているかどうかを確認します（フェイルポイントタスクを過ぎるとロールバックは許可されません）。ロールバックできない場合は、現在の DDL ジョブのステータスを PAUSED にして終了します。
・DDL ジョブのステータスが ROLLBACK_RUNNING の場合、他に実行中のタスクが存在する可能性があります。このとき、DDL エンジンは新しいタスクの実行開始を許可せず、実行中のタスクの成功または失敗を待ちます。これで DDL ジョブは一貫性のある状態に達します。
・一貫性のある状態に達した後、ロールバック処理を開始できます。まず DAG グラフのすべての有向エッジを反転させ、DDL ジョブ全体の実行順序を反転させます。次に反転した DAG グラフに基づいてトポロジカルソートを実行し、以前に実行された、または実行中だが未完了のタスクを取り出し、それらの逆冪等メソッドを実行します。
・DAG グラフに実行可能なタスクノードがなくなった時点で、DDL ジョブのステータスを ROLLBACK_COMPLETED に設定し、ロールバックが成功となります。
他の状態のコールバック関数のロジックは比較的シンプルなため、ここでの詳細な説明は省略します。興味のある読者は自行コードを参照してください。
エラー処理とリカバリ
DDL エンジンが追求する目標の 1 つは DDL の原子性です。DDL の実行中に一部のタスクが失敗した場合、DDL エンジンは適切な措置を講じて、DDL ジョブを完全に未実行または正常に実行された状態（つまり状態遷移図上の最終状態）にする必要があります。DDL エンジンが採用している方法は、タスクに DdlExceptionAction 属性を追加することです。この属性は、タスク実行時に例外が発生した際に DDL エンジンがどのように処理するかを指示するために使用されます。DDL 開発者は DDL タスクを定義する際にこの属性を設定できます。
DdlExceptionAction には合計 4 つの値があります。
・TRY_RECOVERY_THEN_PAUSE：タスクの実行中に例外が発生した場合、3 回リトライします。それでも失敗した場合は、そのタスクが属する DDL ジョブのステータスを PAUSED に設定します。
・ROLLBACK：タスクの実行中に例外が発生した場合、そのタスクが属する DDL ジョブのステータスを ROLLBACK_RUNNING に設定し、DDL エンジンがそのステータスに従って DDL をロールバックします。
・TRY_RECOVERY_THEN_ROLLBACK：タスクの実行中に例外が発生した場合、3 回リトライします。それでも失敗した場合は、そのタスクが属する DDL ジョブのステータスを ROLLBACK_RUNNING に設定し、DDL エンジンが DDL をロールバックします。
・PAUSE：タスクの実行中に例外が発生した場合、そのタスクが属する DDL ジョブのステータスを PAUSED に設定します。
一般的に、PAUSED 状態は DDL ジョブが最終状態に達しておらず、開発者の介入が必要なことを意味します。これは例外発生後に回復できないタスクや、外部に影響を与えてロールバックできなくなったタスクに対して使用されます。前者の例として、drop table コマンドが挙げられます。メタデータ削除や物理テーブル削除のタスクが一度実行されると、削除前の状態に戻すことはできません。この場合、タスクが失敗して 3 回のリトライ後も失敗すると、DDL ジョブは PAUSED 状態に移行します。後者の例として、PolarDB-X のほとんどの DDL ジョブには CDC マーク付きタスクが含まれており、これは外部にバイナリログを生成するために使用されます。このタスクの実行完了は、外部が既に対応する DDL バイナリログを取得できることを意味するため、ロールバックできません。
まとめ
本記事では、DDL エンジンの視点から、DDL エンジンのアーキテクチャと実装、および DDL エンジンと DDL ジョブの間のインタラクションロジックを紹介しました。PolarDB-X のソースコード分析の詳細については、今後の記事にご注目ください。