# 通知設計書 89-SparkListenerConnectOperationStarted

## 概要

本ドキュメントは、Apache Spark Connect における `SparkListenerConnectOperationStarted` イベント通知の設計を定義する。本イベントはConnectリクエストの受信後、分析や実行の前段階で発火する。

### 本通知の処理概要

`SparkListenerConnectOperationStarted` は、Spark Connectにおいてクライアントからのリクエスト（ExecutePlanRequest）を受信した直後に、ExecuteEventsManagerのpostStartedメソッドから発火されるイベントである。このイベントはオペレーションの識別情報、ユーザー情報、ステートメントテキスト等の豊富な情報を含む。分析（Analysis）や実行（Execution）が行われる前の段階で発火されるため、リクエストの受付ログとして機能する。

**業務上の目的・背景**：Spark Connectはクライアントからのプラン実行リクエストをgRPC経由で受け取り処理する。リクエスト受信の事実を即座に記録することで、実行パイプラインの各段階のレイテンシ計測が可能となり、問題発生時のトラブルシューティングに活用できる。また、UI上で現在実行中のオペレーション一覧を表示するための基盤情報を提供する。

**通知の送信タイミング**：ExecuteEventsManager.postStartedメソッドにおいて、オペレーションのステータスがPendingからStartedに遷移する際に発火される。assertStatusにより、Pending状態からのみStartedへの遷移が許可される。

**通知の受信者**：SparkListenerインターフェースを実装したすべてのリスナーが受信対象となる。主な受信者はSparkConnectServerListener（onOtherEvent経由でonOperationStartedが呼ばれ、LiveExecutionDataの作成が行われる）。

**通知内容の概要**：jobTag（Sparkジョブタグ）、operationId（36文字UUID）、eventTime（イベント時刻）、sessionId、userId、userName、statementText（プランのテキスト表現）、sparkSessionTags（ユーザー設定タグ）、extraTags、planRequest（オプション）を含む。

**期待されるアクション**：リスナーはオペレーション開始を検知し、UIにオペレーション情報を追加する。セッションごとの実行数カウンタを増分する。

## 通知種別

アプリ内通知（Sparkリスナーバス経由のイベント駆動型通知）

## 送信仕様

### 基本情報

| 項目 | 内容 |
|-----|------|
| 送信方式 | 非同期（LiveListenerBus経由） |
| 優先度 | 中 |
| リトライ | なし（イベント配信失敗時はドロップ） |

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

SparkContextのLiveListenerBusに登録されたすべてのSparkListenerに対してブロードキャスト配信される。onOtherEventメソッドを通じてSparkConnectServerListenerに配信され、onOperationStartedハンドラが呼ばれる。

## 通知テンプレート

### メール通知の場合

本イベントはメール通知ではなく、プログラム内イベントとして配信される。メールテンプレートは該当しない。

### 本文テンプレート

```
イベントデータ構造:
- jobTag: {Sparkジョブタグ（セッション・リクエスト横断でユニーク）}
- operationId: {36文字UUID}
- eventTime: {イベント生成時刻（ミリ秒）}
- sessionId: {セッション識別子}
- userId: {ユーザー識別子}
- userName: {ユーザー名}
- statementText: {Connectリクエストプランのテキスト表現}
- sparkSessionTags: {SparkSession.addTagで設定されたタグのSet}
- extraTags: {追加メタデータ（デフォルト: 空Map）}
- planRequest: {ExecutePlanRequest（オプション）}
```

### 添付ファイル

該当なし（プログラム内イベントのため）

## テンプレート変数

| 変数名 | 説明 | データ取得元 | 必須 |
|--------|------|-------------|-----|
| jobTag | Sparkジョブタグ | executeHolder.jobTag | Yes |
| operationId | オペレーション識別子（UUID） | executeHolder.operationId | Yes |
| eventTime | イベント生成時刻 | clock.getTimeMillis() | Yes |
| sessionId | セッション識別子 | executeHolder.request.getSessionId | Yes |
| userId | ユーザー識別子 | request.getUserContext.getUserId | Yes |
| userName | ユーザー名 | request.getUserContext.getUserName | Yes |
| statementText | プランのテキスト表現 | ProtoUtils.abbreviate(plan) | Yes |
| sparkSessionTags | ユーザー設定タグ | executeHolder.sparkSessionTags | Yes |
| extraTags | 追加メタデータ | デフォルト空Map | No |
| planRequest | ExecutePlanRequest | executeHolder.request | No |

## 送信トリガー・条件

### トリガー一覧

| トリガー種別 | トリガーイベント | 送信条件 | 説明 |
|------------|----------------|---------|------|
| オペレーション開始 | ExecuteEventsManager.postStarted呼び出し | ステータスがPendingの場合 | ExecuteHolder初期化時にpostStartedが呼ばれる |

### 送信抑止条件

| 条件 | 説明 |
|-----|------|
| ステータスがPending以外 | assertStatusにより、Pending以外の状態からStartedへの遷移はIllegalStateExceptionがスローされる |
| セッションが未開始 | sessionHolder.eventManager.statusがStartedでない場合、IllegalStateExceptionがスローされる |

## 処理フロー

### 送信フロー

```mermaid
flowchart TD
    A[gRPCリクエスト受信] --> B[ExecuteHolder作成]
    B --> C[ExecuteEventsManager.postStarted]
    C --> D{status == Pending & session == Started?}
    D -->|Yes| E[status = Started]
    D -->|No| F[IllegalStateException]
    E --> G[プランのテキスト変換 ProtoUtils.abbreviate]
    G --> H[SparkListenerConnectOperationStarted生成]
    H --> I[event.planRequest = Some request]
    I --> J[listenerBus.post]
    J --> K[SparkConnectServerListener.onOperationStarted]
    K --> L[LiveExecutionData作成 totalExecution++]
```

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

### 参照テーブル一覧

該当なし（インメモリイベントバス経由の通知であり、RDBは使用しない）

### 更新テーブル一覧

| テーブル名 | 操作 | 概要 |
|-----------|------|------|
| ElementTrackingStore (KVStore) | INSERT | ExecutionInfo（オペレーション情報）をKVStoreに書き込む |
| ElementTrackingStore (KVStore) | UPDATE | SessionInfoのtotalExecutionを増分する |

#### ExecutionInfo (KVStore)

| 操作 | 項目（カラム名） | 更新値 | 備考 |
|-----|-----------------|-------|------|
| INSERT | jobTag | ジョブタグ | LiveExecutionDataとして作成 |
| INSERT | statement | ステートメントテキスト | e.statementText |
| INSERT | sessionId | セッションID | e.sessionId |
| INSERT | startTimestamp | 開始時刻 | e.eventTime |
| INSERT | userId | ユーザーID | e.userId |
| INSERT | operationId | オペレーションID | e.operationId |
| INSERT | state | STARTED | ExecutionState.STARTED |

## エラー処理

### エラーケース一覧

| エラー種別 | 発生条件 | 対処方法 |
|----------|---------|---------|
| 不正状態遷移 | status != Pendingの場合 | IllegalStateExceptionがスローされる |
| セッション未開始 | sessionStatus != Startedの場合 | IllegalStateExceptionがスローされる |
| プラン型不明 | OpTypeCaseがCOMMAND/ROOT以外の場合 | UnsupportedOperationExceptionがスローされる |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 0（リトライなし） |
| リトライ間隔 | 該当なし |
| リトライ対象エラー | 該当なし |

## 配信設定

### レート制限

| 項目 | 内容 |
|-----|------|
| 1分あたり上限 | 制限なし（リクエスト受信時に1回発火） |
| 1日あたり上限 | 制限なし |

### 配信時間帯

制限なし。リクエスト受信時に随時発火される。

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

本イベントにはstatementText（プランのテキスト表現）が含まれ、SQLクエリやデータ操作の内容が含まれうる。statementTextはUtils.redactにより、sessionState.conf.stringRedactionPatternに基づいてリダクション処理が行われる。また、ProtoUtils.abbreviateにより最大65535文字、ネスティング最大8レベルに制限される。userIdとuserNameはユーザー識別情報であり、適切なアクセス制御が必要。

## 備考

- planRequestフィールドは@JsonIgnoreアノテーション付きであり、JSON/イベントログへのシリアライズ時には除外される。
- statementTextのサイズ上限はExecuteEventsManager.MAX_STATEMENT_TEXT_SIZE（65535文字）で定義されている。
- ネスティングレベル上限はExecuteEventsManager.MAX_STATEMENT_NESTING_LEVEL（8レベル）で定義されている。
- sparkSessionTagsはSparkSession.addTag APIでユーザーが設定したタグであり、オペレーションのフィルタリングやグルーピングに使用できる。

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | ExecuteEventsManager.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteEventsManager.scala` | 行317-333: SparkListenerConnectOperationStartedケースクラスの定義と全フィールドを確認する |
| 1-2 | ExecuteEventsManager.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteEventsManager.scala` | 行36-47: ExecuteStatus列挙（Pending, Started, Analyzed等）を確認する |

**読解のコツ**: planRequestフィールドは@JsonIgnoreが付いているため、イベントログには含まれない。case classのコンストラクタパラメータとは別に、varフィールドとして定義されている点に注意。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ExecuteEventsManager.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteEventsManager.scala` | 行111-141: postStartedメソッドの全体フローを理解する |

**主要処理フロー**:
1. **行112**: assertStatus(List(ExecuteStatus.Pending), ExecuteStatus.Started)で遷移検証
2. **行113-121**: リクエストからプランを取得（CommandまたはRoot）
3. **行123-138**: SparkListenerConnectOperationStartedイベントの構築
4. **行130-137**: Utils.redactとProtoUtils.abbreviateでstatementTextを安全に変換
5. **行139**: event.planRequest = Some(request)でプランリクエストを設定
6. **行140**: listenerBus.post(event)でイベント発火

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SparkConnectServerListener.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerListener.scala` | 行121: onOtherEventでSparkListenerConnectOperationStartedをマッチング |
| 3-2 | SparkConnectServerListener.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerListener.scala` | 行173-193: onOperationStartedメソッドでLiveExecutionData作成とセッションのtotalExecution増分を確認する |

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

```
ExecuteHolder初期化
    |
    +-- ExecuteEventsManager.postStarted() [ExecuteEventsManager.scala:111]
           |
           +-- assertStatus(Pending, Started) [ExecuteEventsManager.scala:274]
           |
           +-- request.getPlan.getOpTypeCase (Command/Root判定) [ExecuteEventsManager.scala:115-121]
           |
           +-- Utils.redact + ProtoUtils.abbreviate (statementText生成) [ExecuteEventsManager.scala:130-137]
           |
           +-- listenerBus.post(SparkListenerConnectOperationStarted) [ExecuteEventsManager.scala:140]
                  |
                  +-- SparkConnectServerListener.onOtherEvent() [SparkConnectServerListener.scala:118]
                         |
                         +-- onOperationStarted() [SparkConnectServerListener.scala:173]
                                |
                                +-- getOrCreateExecution() [SparkConnectServerListener.scala:331]
                                +-- updateLiveStore(state=STARTED) [SparkConnectServerListener.scala:182]
                                +-- sessionData.totalExecution += 1 [SparkConnectServerListener.scala:187]
```

### データフロー図

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

ExecutePlanRequest ───▶ ExecuteEventsManager.postStarted()       ───▶ SparkListenerConnectOperationStarted
(gRPCリクエスト)            |                                            |
                            +-- Plan解析 (Command/Root)                  +-- LiveListenerBus
                            +-- statementText生成 (redact+abbreviate)    |
                            +-- event構築                                +-- SparkConnectServerListener
                            +-- listenerBus.post()                       |   (LiveExecutionData作成)
                                                                         +-- KVStore (ExecutionInfo保存)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ExecuteEventsManager.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteEventsManager.scala` | ソース | イベントクラス定義とイベント発火元 |
| SparkConnectServerListener.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/ui/SparkConnectServerListener.scala` | ソース | UI用のリスナー実装 |
| ExecuteEventsManagerSuite.scala | `sql/connect/server/src/test/scala/org/apache/spark/sql/connect/service/ExecuteEventsManagerSuite.scala` | テスト | イベント発火のテスト |
| ProtoUtils.scala | `sql/connect/common/src/main/scala/org/apache/spark/sql/connect/common/ProtoUtils.scala` | ソース | プランのテキスト変換ユーティリティ |
