# 通知設計書 38-SparkListenerSQLExecutionEnd

## 概要

本ドキュメントは、Apache SparkにおけるSparkListenerSQLExecutionEndイベント通知の設計について記述する。本イベントはSQL実行が終了した際に発火するイベントであり、実行の完了状態やエラー情報を提供する。

### 本通知の処理概要

SparkListenerSQLExecutionEndは、DataFrame/Dataset APIやSQLクエリの実行が完了（成功またはエラー）した際に発火するイベント通知である。SQLExecution.withNewExecutionId0のfinallyブロック内で生成される。

**業務上の目的・背景**：SQL実行の完了を通知し、SQL UIでの実行状態の更新とメトリクスの集約を可能にする。実行の成功・失敗を記録し、エラーメッセージを保持することで、デバッグやパフォーマンス分析に必要な情報を提供する。また、QueryExecutionListenerへの通知もこのイベントを起点として行われる。

**通知の送信タイミング**：SQLExecution.withNewExecutionId0のfinallyブロック内で、body実行の完了後（成功・失敗を問わず）にイベントがpostされる。シャッフルクリーンアップ処理の後に発行される。

**通知の受信者**：LiveListenerBusに登録された全てのSparkListenerInterface実装。特にSQLAppStatusListenerがonOtherEvent経由で受信し、メトリクス集約と状態更新を行う。

**通知内容の概要**：executionId、終了時刻、エラーメッセージ（成功時は空文字列、失敗時はエラー詳細）、クエリIDが含まれる。加えて、@JsonIgnoreフィールドとしてexecutionName、duration、QueryExecution、executionFailure、jobIdsが内部的に保持される。

**期待されるアクション**：SQLAppStatusListenerがメトリクスを非同期で集約し、KVStoreを更新する。SQL UIで実行の完了状態が反映される。QueryExecutionListenerManagerがexecutionName付きの実行について通知を転送する。

## 通知種別

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

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由、onOtherEventにディスパッチ） |
| 優先度 | 高（SQL UI表示・メトリクス集約の基盤） |
| リトライ | なし |

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

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

## 通知テンプレート

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

| 項目 | 内容 |
|-----|------|
| イベントクラス | `SparkListenerSQLExecutionEnd` |
| パッケージ | `org.apache.spark.sql.execution.ui` |
| 親クラス | `SparkListenerEvent` |
| シリアライズ形式 | JSON（JsonProtocol経由、@JsonIgnoreフィールドは除外） |

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

```scala
@DeveloperApi
case class SparkListenerSQLExecutionEnd(
    executionId: Long,
    time: Long,
    errorMessage: Option[String] = None,
    queryId: Option[UUID] = None)
  extends SparkListenerEvent {

  @JsonIgnore private[sql] var executionName: Option[String] = None
  @JsonIgnore private[sql] var duration: Long = 0L
  @JsonIgnore private[sql] var qe: QueryExecution = null
  @JsonIgnore private[sql] var executionFailure: Option[Throwable] = None
  @JsonIgnore private[sql] var jobIds: Set[Int] = Set.empty
}
```

### 添付ファイル

該当なし

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| executionId | SQL実行の一意ID | SQLExecution.withNewExecutionId0内のローカル変数 | Yes |
| time | 終了時刻（ミリ秒） | System.currentTimeMillis() | Yes |
| errorMessage | エラーメッセージ（成功時は Some(""), 失敗時はSome(エラー文字列)） | 例外情報から生成 | No |
| queryId | クエリの一意ID | SQLExecution.withNewExecutionId0内のローカル変数 | No |
| executionName | 実行名（Dataset.withAction等で設定） | SQLExecution.withNewExecutionId0のnameパラメータ | No |
| duration | 実行時間（ナノ秒） | endTime - startTime | No |
| qe | QueryExecutionインスタンス | SQLExecution.withNewExecutionId0のqueryExecutionパラメータ | No |
| executionFailure | 実行失敗の例外 | try-catchで捕捉 | No |
| jobIds | 関連ジョブID集合（テスト用） | dagScheduler.activeQueryToJobs | No |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| 内部処理 | SQL実行完了（成功） | withNewExecutionId0のbody実行完了 | finallyブロックで必ず発行 |
| 内部処理 | SQL実行失敗（例外） | withNewExecutionId0のbody実行で例外発生 | finallyブロックで必ず発行 |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| なし | finallyブロック内で必ず発行される |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[body実行完了/例外発生] --> B[finallyブロック開始]
    B --> C[endTime = System.nanoTime]
    C --> D[errorMessage生成]
    D --> E{シャッフルクリーンアップ必要?}
    E -->|Yes| F[シャッフルファイル除去/マイグレーションスキップ]
    E -->|No| G[SparkListenerSQLExecutionEnd生成]
    F --> G
    G --> H[executionName, duration, qe等を設定]
    H --> I[dagScheduler.cleanupQueryJobs]
    I --> J[sc.listenerBus.post event]
    J --> K[observationManager.tryComplete]
    K --> L[SQLAppStatusListener.onExecutionEnd受信]
    L --> M[completionTime設定]
    M --> N[kvstore.doAsync: メトリクス集約]
    N --> O[LiveExecutionData更新]
```

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

### KVStore更新仕様

| ストア | キー | 操作 | 説明 |
|--------|------|------|------|
| ElementTrackingStore | executionId | UPDATE | completionTime, errorMessage, metricsValuesの更新 |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| SparkThrowable | SQL実行で発生したSparkThrowable | SparkThrowableHelper.getMessageでPRETTY形式のメッセージを生成 |
| その他の例外 | SQL実行で発生した非SparkThrowable例外 | Utils.exceptionStringでスタックトレースを含むメッセージを生成 |
| イベント配信失敗 | ListenerBusキュー溢れ | LiveListenerBusのドロップポリシーに従う |

### リトライ仕様

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

## 配信設定

### レート制限

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

### 配信時間帯

制限なし

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

- errorMessageにはSQLクエリのエラー詳細が含まれる可能性があるが、SparkThrowableの場合はPRETTY形式でフォーマットされる
- @JsonIgnoreフィールドはJSON シリアライズ対象外であり、イベントログには含まれない
- QueryExecutionやThrowableオブジェクトはインプロセスでのみ参照可能

## 備考

- errorMessageがNoneの場合は、古いバージョンのSparkで生成されたイベントログの互換性のため。Spark 3.4以降では常にSome値を持ち、空文字列は成功を示す
- メトリクス集約はkvstore.doAsyncで非同期実行されるため、大規模クエリでもイベント配信をブロックしない
- シャッフルクリーンアップ（RemoveShuffleFiles, SkipMigration）はイベントpost前に実行される

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SQLListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala` | 68-96行目: SparkListenerSQLExecutionEndのcase class定義。@JsonIgnoreフィールドの役割を理解する |

**読解のコツ**: case classのプライマリコンストラクタパラメータのみがJSONシリアライズの対象。varフィールドは`@JsonIgnore`で除外されている。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SQLExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/SQLExecution.scala` | 194-256行目: withNewExecutionId0のfinally部分。イベント生成・配信ロジック |

**主要処理フロー**:
1. **195行目**: `val endTime = System.nanoTime()`
2. **196-201行目**: errorMessage生成（SparkThrowableとその他で分岐）
3. **226-232行目**: `SparkListenerSQLExecutionEnd`インスタンス生成
4. **237-238行目**: executionName, durationの設定
5. **249行目**: `sc.dagScheduler.cleanupQueryJobs(executionId)`
6. **251行目**: `sc.listenerBus.post(event)`でイベント配信

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SQLAppStatusListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLAppStatusListener.scala` | 401-418行目: onExecutionEndメソッド。メトリクス非同期集約とKVStore更新 |

**主要処理フロー**:
1. **403-404行目**: completionTimeとerrorMessageの設定
2. **411-415行目**: `kvstore.doAsync`で非同期にメトリクス集約
3. **412行目**: `aggregateMetrics(exec)`でメトリクス値を計算
4. **414行目**: `exec.endEvents.incrementAndGet()`で完了イベントカウント更新

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

```
SQLExecution.withNewExecutionId0() [finallyブロック]
    |
    +-- errorMessage生成
    +-- シャッフルクリーンアップ
    +-- SparkListenerSQLExecutionEnd生成
    +-- event.executionName = name
    +-- event.duration = endTime - startTime
    +-- event.qe = queryExecution
    +-- dagScheduler.cleanupQueryJobs()
    +-- sc.listenerBus.post(event)
    |       |
    |       +-- SparkListenerBus.doPostEvent()
    |               |
    |               +-- listener.onOtherEvent()
    |                       |
    |                       +-- SQLAppStatusListener.onExecutionEnd()
    |                               |
    |                               +-- completionTime設定
    |                               +-- kvstore.doAsync { aggregateMetrics() }
    |
    +-- observationManager.tryComplete()
```

### データフロー図

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

実行結果                  SQLExecution                     SparkListenerSQLExecutionEnd
(errorMessage,      ---> .withNewExecutionId0()     --->   イベント
 duration,               finallyブロック                     |
 queryExecution)                                            v
                                                     LiveListenerBus
                                                            |
                                                     +------+------+
                                                     |             |
                                              SQLAppStatus    EventLogging
                                              Listener        Listener
                                                     |
                                                     v
                                              KVStore (メトリクス集約)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SQLListener.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/ui/SQLListener.scala` | ソース | イベントクラス定義 |
| SQLExecution.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/SQLExecution.scala` | ソース | イベント生成・配信（finallyブロック） |
| 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経由） |
| EventLoggingListener.scala | `core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala` | ソース | イベントログへの永続化 |
