# 通知設計書 40-SparkListenerSQLAdaptiveSQLMetricUpdates

## 概要

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

### 本通知の処理概要

SparkListenerSQLAdaptiveSQLMetricUpdatesは、AQEが新しいサブプランを生成した際に、新しく追加されたメトリクス情報をリスナーに通知するイベントである。AdaptiveSparkPlanExec内で、新しいサブプランのメトリクスが検出された場合にpostされる。

**業務上の目的・背景**：AQEによる計画変更では、新しい物理演算子が追加されることがあり、それに伴い新しいメトリクス（Accumulatorベース）も追加される。これらの新規メトリクスをSQL UIに反映するために、メトリクス情報の更新をリスナーに通知する必要がある。完全な実行計画の再送信よりも軽量なメトリクス差分情報のみを送信する最適化手法である。

**通知の送信タイミング**：AdaptiveSparkPlanExec内で新しいサブプランが作成された際に、そのサブプラン内のメトリクスを抽出してpostする。SparkListenerSQLAdaptiveExecutionUpdate（No.39）の代替として、新しいサブプランがある場合にメトリクス情報のみを通知する。

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

**通知内容の概要**：executionIdと新しいSQLPlanMetric（メトリクス名、AccumulatorID、メトリクスタイプ）のシーケンスが含まれる。

**期待されるアクション**：SQLAppStatusListenerがLiveExecutionDataにメトリクスを追加し、SQL UIでの新しいメトリクスの表示を可能にする。

## 通知種別

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

## 送信仕様

### 基本情報

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

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

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

## 通知テンプレート

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

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

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

```scala
@DeveloperApi
case class SparkListenerSQLAdaptiveSQLMetricUpdates(
    executionId: Long,
    sqlPlanMetrics: Seq[SQLPlanMetric])
  extends SparkListenerEvent
```

### 添付ファイル

該当なし

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| executionId | SQL実行の一意ID | SQLExecution.EXECUTION_ID_KEY (ローカルプロパティ) | Yes |
| sqlPlanMetrics | 新しいSQLメトリクスのリスト | 新しいサブプランのmetrics.values | Yes |

### SQLPlanMetricの構造

| フィールド | 説明 | 型 |
|-----------|------|------|
| name | メトリクス名 | String |
| accumulatorId | Accumulator ID | Long |
| metricType | メトリクスタイプ（"size", "timing", "nsTiming"等） | String |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | AQEによる新サブプラン作成 | 新しいサブプランにメトリクスが存在する場合 | AdaptiveSparkPlanExec内で発火 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| AQE無効 | `spark.sql.adaptive.enabled=false`の場合はAdaptiveSparkPlanExecが使用されない |
| 新しいサブプランなし | 計画更新で新しいサブプランが追加されなかった場合、代わりにSparkListenerSQLAdaptiveExecutionUpdateが発行される |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[AQEによるクエリステージ実体化完了] --> B[物理計画の再最適化]
    B --> C[新しいサブプラン検出]
    C --> D{新しいサブプランあり?}
    D -->|Yes| E[新しいメトリクス抽出]
    E --> F[SparkListenerSQLAdaptiveSQLMetricUpdates生成]
    F --> G[sc.listenerBus.post]
    D -->|No| H[SparkListenerSQLAdaptiveExecutionUpdate発行]
    G --> I[SQLAppStatusListener.onAdaptiveSQLMetricUpdate受信]
    I --> J[LiveExecutionData.addMetrics]
    J --> K[KVStore更新]
    H --> K
```

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

### KVStore更新仕様

| ストア | キー | 操作 | 説明 |
|--------|------|------|------|
| ElementTrackingStore | executionId | UPDATE | LiveExecutionDataのmetricsリストに新メトリクスを追加 |

## エラー処理

### エラーケース一覧

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

### リトライ仕様

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

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし

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

- メトリクス情報はAccumulator IDとメトリクス名のみであり、機密情報は含まない
- イベントログに記録される場合は、イベントログのアクセス制御に従う

## 備考

- 本イベントはSparkListenerSQLAdaptiveExecutionUpdate（No.39）と相互排他的に発行される場合がある。新しいサブプランがある場合は本イベント、ない場合はNo.39が発行される
- AdaptiveSparkPlanExecの849-853行目付近でpostされる
- SQLPlanMetricはメトリクス名、AccumulatorID、メトリクスタイプの3要素から構成される
- メトリクスタイプには"size"（バイト数）、"timing"（ミリ秒）、"nsTiming"（ナノ秒）、"sum"（合計値）等がある

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SQLListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala` | 40-42行目: SparkListenerSQLAdaptiveSQLMetricUpdatesのcase class定義。executionIdとsqlPlanMetricsのリスト |

**読解のコツ**: sqlPlanMetricsはSeq[SQLPlanMetric]型であり、各要素はname、accumulatorId、metricTypeの3フィールドを持つ。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | AdaptiveSparkPlanExec.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/AdaptiveSparkPlanExec.scala` | 849-853行目: 新しいサブプランからメトリクスを抽出してイベントをpostする箇所 |

**主要処理フロー**:
1. **849行目**: `val newMetrics = newSubPlans.flatMap { p => ... }`で新サブプランからメトリクスを抽出
2. **850行目**: `p.flatMap(_.metrics.values.map(m => SQLPlanMetric(m.name.get, m.id, m.metricType)))`
3. **852行目**: `context.session.sparkContext.listenerBus.post(SparkListenerSQLAdaptiveSQLMetricUpdates(executionId, newMetrics))`

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | 393-399行目: onAdaptiveSQLMetricUpdateメソッド。LiveExecutionDataにメトリクスを追加 |

**主要処理フロー**:
1. **394行目**: イベントからexecutionIdとsqlPlanMetricsを抽出
2. **396行目**: `exec.addMetrics(sqlPlanMetrics)`でメトリクスを追加
3. **397-398行目**: `update(exec)`でKVStoreを更新

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | 440行目: onOtherEvent内でSparkListenerSQLAdaptiveSQLMetricUpdatesをパターンマッチ |

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

```
AdaptiveSparkPlanExec (新サブプラン作成)
    |
    +-- newSubPlans.flatMap { metrics抽出 }
    +-- SQLPlanMetric(name, id, metricType)
    +-- context.session.sparkContext.listenerBus.post(
            SparkListenerSQLAdaptiveSQLMetricUpdates(executionId, newMetrics))
            |
            +-- SparkListenerBus.doPostEvent()
                    |
                    +-- listener.onOtherEvent()
                            |
                            +-- SQLAppStatusListener.onAdaptiveSQLMetricUpdate()
                                    |
                                    +-- exec.addMetrics(sqlPlanMetrics)
                                    +-- update(exec)
```

### データフロー図

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

新サブプラン             AdaptiveSparkPlanExec             SparkListenerSQLAdaptive
(metrics.values)   ---> メトリクス抽出               --->  SQLMetricUpdates イベント
                                                           |
                                                           v
                                                    LiveListenerBus
                                                           |
                                                    +------+------+
                                                    |             |
                                             SQLAppStatus    EventLogging
                                             Listener        Listener
                                                    |
                                                    v
                                             LiveExecutionData
                                             (メトリクス追加)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| 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` | ソース | イベント受信・メトリクス追加 |
| 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` | ソース | イベントログへの永続化 |
