# 通知設計書 39-SparkListenerSQLAdaptiveExecutionUpdate

## 概要

本ドキュメントは、Apache SparkにおけるSparkListenerSQLAdaptiveExecutionUpdateイベント通知の設計について記述する。本イベントはAdaptive Query Execution（AQE）により物理実行計画が更新された際に発火するイベントである。

### 本通知の処理概要

SparkListenerSQLAdaptiveExecutionUpdateは、AQEがクエリステージの実体化完了後に物理実行計画を再最適化した際に発火するイベント通知である。AdaptiveSparkPlanExec内で、新しいステージが作成された時点で実行計画の更新情報がpostされる。

**業務上の目的・背景**：AQEは実行時のデータ統計情報に基づいて物理実行計画を動的に最適化する機能である。計画が更新されるたびにSQL UIに反映する必要があり、本イベントがその更新情報を伝達する。これにより、開発者はAQEによる計画変更の経過をリアルタイムで追跡できる。

**通知の送信タイミング**：AdaptiveSparkPlanExec内で新しいステージが作成された際（createQueryStages処理後）に、実行計画が変更されていた場合に発火する。ただし、`spark.sql.ui.explainMode`でexplainModeが設定されている場合のみ物理計画の記述が含まれる。

**通知の受信者**：LiveListenerBusに登録された全てのSparkListenerInterface実装。特にSQLAppStatusListenerがonOtherEvent経由で受信する。

**通知内容の概要**：executionId、更新された物理実行計画のテキスト記述、更新されたSparkPlanInfo（計画のツリー構造）が含まれる。

**期待されるアクション**：SQLAppStatusListenerがSparkPlanGraphを再構築してKVStoreを更新する。SQL UIタブで最新の実行計画DAGが表示される。

## 通知種別

アプリ内通知（Sparkイベントバスによるインプロセス非同期イベント配信。onOtherEvent経由）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由、onOtherEventにディスパッチ） |
| 優先度 | 中 |
| リトライ | なし |

### 送信先決定ロジック

SparkListenerBus.doPostEventのデフォルトケース（`case _ => listener.onOtherEvent(event)`）によりonOtherEventに配信。SQLAppStatusListener.onOtherEvent内でパターンマッチされ、onAdaptiveExecutionUpdateが呼び出される。

## 通知テンプレート

### イベントオブジェクト

| 項目 | 内容 |
|-----|------|
| イベントクラス | `SparkListenerSQLAdaptiveExecutionUpdate` |
| パッケージ | `org.apache.spark.sql.execution.ui` |
| 親クラス | `SparkListenerEvent` |
| シリアライズ形式 | JSON（JsonProtocol経由） |

### イベントデータ構造

```scala
@DeveloperApi
case class SparkListenerSQLAdaptiveExecutionUpdate(
  executionId: Long,
  physicalPlanDescription: String,
  sparkPlanInfo: SparkPlanInfo)
  extends SparkListenerEvent
```

### 添付ファイル

該当なし

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| executionId | SQL実行の一意ID | SQLExecution.EXECUTION_ID_KEY (ローカルプロパティ) | Yes |
| physicalPlanDescription | 更新された物理実行計画のテキスト表現 | context.qe.explainString(planDescriptionMode) | Yes |
| sparkPlanInfo | 更新された実行計画のツリー構造 | SparkPlanInfo.fromSparkPlan(context.qe.executedPlan) | Yes |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | AQEによるステージ作成・計画更新 | 新しいステージが作成され、計画が変更された場合 | AdaptiveSparkPlanExec内で発火 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| AQE無効 | `spark.sql.adaptive.enabled=false`の場合はAdaptiveSparkPlanExecが使用されない |
| 計画変更なし | ステージ作成後に計画が変更されていない場合はイベント不要（ただし実装上は新ステージ作成時に常に発行される） |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[AQEによるクエリステージ実体化完了] --> B[物理計画の再最適化]
    B --> C[新しいステージ作成]
    C --> D{新しいサブプランあり?}
    D -->|Yes/No 分岐| E[SparkListenerSQLAdaptiveExecutionUpdate生成]
    E --> F[sc.listenerBus.post]
    F --> G[SQLAppStatusListener.onAdaptiveExecutionUpdate受信]
    G --> H[SparkPlanGraph再構築]
    H --> I[KVStore更新]
```

## データベース参照・更新仕様

### KVStore更新仕様

| ストア | キー | 操作 | 説明 |
|--------|------|------|------|
| ElementTrackingStore | executionId | WRITE | SparkPlanGraphWrapper（更新された計画グラフ）を上書き |
| ElementTrackingStore | executionId | UPDATE | LiveExecutionDataのphysicalPlanDescriptionとmetricsを更新 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| イベント配信失敗 | ListenerBusキュー溢れ | LiveListenerBusのドロップポリシーに従う |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 0 |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（AQEステージ更新毎に発行） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし

## セキュリティ考慮事項

- 物理実行計画のテキスト表現には、テーブル名やカラム名が含まれる可能性がある
- イベントログに記録される場合は、イベントログのアクセス制御に従う

## 備考

- 本イベントはAQEが有効（`spark.sql.adaptive.enabled=true`、Spark 3.2以降デフォルト有効）の場合にのみ発行される
- AQEは実行中に複数回計画を更新する可能性があり、その都度本イベントが発行される
- SparkListenerSQLAdaptiveSQLMetricUpdates（No.40）とは相互排他的ではなく、状況により両方が発行されることがある
- AdaptiveSparkPlanExecの856行目付近でpostされる

---

## コードリーディングガイド

本通知を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SQLListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala` | 33-37行目: SparkListenerSQLAdaptiveExecutionUpdateのcase class定義。3フィールドのシンプルな構造 |

**読解のコツ**: physicalPlanDescriptionは人間可読なテキスト、sparkPlanInfoはプログラムが解析可能なツリー構造である。

#### Step 2: エントリーポイントを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AdaptiveSparkPlanExec.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala` | 856行目付近: SparkListenerSQLAdaptiveExecutionUpdateをpostする箇所 |

**主要処理フロー**:
- **856行目**: `context.session.sparkContext.listenerBus.post(SparkListenerSQLAdaptiveExecutionUpdate(executionId, context.qe.explainString(planDescriptionMode), SparkPlanInfo.fromSparkPlan(context.qe.executedPlan)))`

#### Step 3: リスナー側の処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | 372-391行目: onAdaptiveExecutionUpdateメソッド。SparkPlanGraph再構築とKVStore更新 |

**主要処理フロー**:
1. **376行目**: `SparkPlanGraph(sparkPlanInfo)`で新しいプラングラフを構築
2. **381-385行目**: `SparkPlanGraphWrapper`をKVStoreに書込（上書き）
3. **387-389行目**: LiveExecutionDataのphysicalPlanDescriptionとmetricsを更新

#### Step 4: イベント配信経路を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | 100行目: onOtherEvent経由でSQLモジュールのリスナーに配信 |
| 4-2 | SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | 439行目: onOtherEvent内でSparkListenerSQLAdaptiveExecutionUpdateをパターンマッチ |

### プログラム呼び出し階層図

```
AdaptiveSparkPlanExec (ステージ作成・計画更新)
    |
    +-- createQueryStages() / getFinalPhysicalPlan()
            |
            +-- context.qe.explainString(planDescriptionMode)
            +-- SparkPlanInfo.fromSparkPlan(context.qe.executedPlan)
            +-- context.session.sparkContext.listenerBus.post(
                    SparkListenerSQLAdaptiveExecutionUpdate(...))
                    |
                    +-- SparkListenerBus.doPostEvent()
                            |
                            +-- listener.onOtherEvent()
                                    |
                                    +-- SQLAppStatusListener.onAdaptiveExecutionUpdate()
                                            |
                                            +-- SparkPlanGraph()
                                            +-- kvstore.write(graphToStore)
                                            +-- LiveExecutionData更新
```

### データフロー図

```
[入力]                          [処理]                              [出力]

AQE計画更新            AdaptiveSparkPlanExec             SparkListenerSQLAdaptive
(executedPlan,    ---> ステージ作成・計画更新        --->  ExecutionUpdate イベント
 executionId)                                              |
                                                           v
                                                    LiveListenerBus
                                                           |
                                                    +------+------+
                                                    |             |
                                             SQLAppStatus    EventLogging
                                             Listener        Listener
                                                    |
                                                    v
                                             KVStore (計画グラフ更新)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SQLListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala` | ソース | イベントクラス定義 |
| AdaptiveSparkPlanExec.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala` | ソース | AQE実行・イベント発行元 |
| SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | ソース | イベント受信・KVStore更新 |
| SparkListenerBus.scala | `core/src/main/scala/org/apache/spark/scheduler/SparkListenerBus.scala` | ソース | イベントディスパッチ（onOtherEvent経由） |
| SQLExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/SQLExecution.scala` | ソース | SQL実行管理（executionId管理） |
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | ソース | イベントログへの永続化 |
