# 機能設計書 80-Spark Connectサーバー

## 概要

本ドキュメントは、Apache Spark ConnectサーバーのgRPCサービス基盤の設計を記述する。

### 本機能の処理概要

Spark Connectサーバーは、gRPCプロトコルを介してリモートクライアントからSpark SQLの実行を可能にするサーバーコンポーネントである。クライアントからの実行プラン、分析リクエスト、設定変更、アーティファクト管理、セッション管理などのRPCを受け付け、サーバーサイドのSparkSessionで処理を実行して結果を返す。

**業務上の目的・背景**：従来のSpark実行モデルではクライアントとドライバーが同一JVMで動作する必要があったが、Spark Connectにより、Python/Go/その他の言語クライアントからリモートでSpark処理を実行可能になる。これによりクライアントとドライバーの分離、セッション分離、マルチテナントが実現される。

**機能の利用シーン**：リモートクライアント（PySpark Connect、Spark Connect Go等）からのSQL/DataFrame実行、分散環境でのマルチユーザーSpark共有、ノートブック環境からのSpark接続。

**主要な処理内容**：
1. gRPCサーバーの起動（NettyServerBuilder）
2. SparkConnectService によるRPCハンドリング
3. セッション管理（SparkConnectSessionManager）
4. 実行管理（SparkConnectExecutionManager）
5. アーティファクト管理
6. Reattachable Execution（再接続可能な実行）
7. ストリーミングクエリキャッシュ

**関連システム・外部連携**：
- gRPC/Nettyサーバーフレームワーク
- Protobufプロトコル定義（SparkConnectServiceGrpc）
- Spark SQL実行エンジン
- Spark UI（SparkConnectServerTab）

**権限による制御**：
- PreSharedKeyAuthenticationInterceptorによるトークンベース認証
- SparkConnectInterceptorRegistryによるカスタムインターセプター登録

## 関連画面

本機能に直接関連するUI画面はない（ただしSparkConnectServerTabがSpark UIに統合される）。

## 機能種別

サーバーサービス / gRPCサービス基盤

## 入力仕様

### 入力パラメータ（gRPC RPCメソッド）

| RPCメソッド | リクエスト型 | 説明 |
|------------|------------|------|
| executePlan | ExecutePlanRequest | SparkプランをRDDに変換して実行 |
| analyzePlan | AnalyzePlanRequest | プランの分析（explain、スキーマ取得等） |
| config | ConfigRequest | Spark設定の取得・変更 |
| addArtifacts | AddArtifactsRequest (stream) | JARやファイルのアップロード |
| artifactStatus | ArtifactStatusesRequest | アーティファクトの存在確認 |
| interrupt | InterruptRequest | 実行中のオペレーションの中断 |
| reattachExecute | ReattachExecuteRequest | 切断された実行への再接続 |
| releaseExecute | ReleaseExecuteRequest | 実行リソースの解放 |
| releaseSession | ReleaseSessionRequest | セッションの解放 |
| fetchErrorDetails | FetchErrorDetailsRequest | エラー詳細の取得 |
| cloneSession | CloneSessionRequest | セッションのクローン |

### 設定パラメータ

| 設定キー | デフォルト | 説明 |
|---------|-----------|------|
| spark.connect.grpc.binding.port | - | gRPCバインドポート |
| spark.connect.grpc.binding.address | None | gRPCバインドアドレス |
| spark.connect.grpc.maxInboundMessageSize | - | 最大受信メッセージサイズ |
| spark.connect.grpc.marshaller.recursionLimit | - | Protobuf再帰制限 |
| spark.connect.grpc.debug.enabled | false | デバッグモード |
| spark.connect.grpc.port.maxRetries | - | ポートバインドのリトライ回数 |

## 出力仕様

### 出力データ

| RPCメソッド | レスポンス型 | 説明 |
|------------|------------|------|
| executePlan | ExecutePlanResponse (stream) | 実行結果（Arrow形式データ、メトリクス等） |
| analyzePlan | AnalyzePlanResponse | 分析結果（explainテキスト、スキーマ等） |
| config | ConfigResponse | 設定値 |
| addArtifacts | AddArtifactsResponse | アップロード結果 |
| artifactStatus | ArtifactStatusesResponse | アーティファクト状態 |
| interrupt | InterruptResponse | 中断結果 |
| reattachExecute | ExecutePlanResponse (stream) | 再接続された実行結果 |
| releaseExecute | ReleaseExecuteResponse | 解放結果 |
| releaseSession | ReleaseSessionResponse | セッション解放結果 |
| fetchErrorDetails | FetchErrorDetailsResponse | エラー詳細 |
| cloneSession | CloneSessionResponse | クローンされたセッション情報 |

### 出力先

gRPC StreamObserver経由でクライアントに送信

## 処理フロー

### 処理シーケンス

```
1. サーバー起動 (SparkConnectService.start)
   ├─ sessionManager.initializeBaseSession [ベースSparkSession作成]
   ├─ startGRPCService [gRPCサーバー起動]
   │   ├─ NettyServerBuilder でサーバー構築
   │   ├─ SparkConnectService をサービス登録
   │   ├─ PreSharedKeyAuthenticationInterceptor 登録（設定時）
   │   ├─ カスタムインターセプター登録
   │   └─ server.start()
   ├─ createListenerAndUI [UIタブ作成]
   └─ postSparkConnectServiceStarted [イベント発行]

2. RPC処理フロー (executePlan例)
   ├─ SparkConnectService.executePlan(request, responseObserver)
   ├─ SparkConnectExecutePlanHandler.handle(request)
   │   ├─ getOrCreateIsolatedSession [セッション取得/作成]
   │   ├─ ExecuteKey構築
   │   ├─ executionManager.createExecuteHolderAndAttach
   │   │   ├─ ExecuteHolder作成
   │   │   ├─ executeHolder.start() [実行開始]
   │   │   └─ ExecuteGrpcResponseSender [結果送信]
   │   └─ ErrorUtils.handleError [エラー処理]

3. セッション管理
   ├─ getOrCreateIsolatedSession: ConcurrentHashMapでセッション管理
   ├─ periodicMaintenance: 期限切れセッションの自動クリーンアップ
   └─ closeSession: セッションの明示的解放

4. 実行管理
   ├─ createExecuteHolder: 実行の登録
   ├─ removeExecuteHolder: 実行の解放
   ├─ reattachExecuteHolder: 再接続
   └─ periodicMaintenance: 放棄された実行の自動クリーンアップ

5. サーバー停止 (SparkConnectService.stop)
   ├─ server.shutdown / shutdownNow
   ├─ streamingSessionManager.shutdown()
   ├─ executionManager.shutdown()
   ├─ sessionManager.shutdown()
   └─ uiTab.detach()
```

### フローチャート

```mermaid
flowchart TD
    A[SparkConnectServer.main] --> B[SparkSession構築]
    B --> C[SparkConnectService.start]
    C --> D[gRPCサーバー起動]
    D --> E[UIタブ作成]
    E --> F[server.awaitTermination]

    G[クライアントRPC] --> H{RPCメソッド}
    H -->|executePlan| I[SparkConnectExecutePlanHandler]
    H -->|analyzePlan| J[SparkConnectAnalyzeHandler]
    H -->|config| K[SparkConnectConfigHandler]
    H -->|interrupt| L[SparkConnectInterruptHandler]
    H -->|reattachExecute| M[SparkConnectReattachExecuteHandler]
    I --> N[セッション取得/作成]
    N --> O[ExecuteHolder作成・実行]
    O --> P[ExecuteGrpcResponseSender]
    P --> Q[クライアントに結果送信]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-80-01 | セッション分離 | userId + sessionIdの組み合わせでセッションを分離 | 全RPC |
| BR-80-02 | セッションタイムアウト | 一定時間アクセスのないセッションは自動解放 | periodicMaintenance |
| BR-80-03 | 実行タイムアウト | RPCが接続されていない実行は一定時間後に自動解放 | periodicMaintenance |
| BR-80-04 | 再接続防止 | 閉じたセッション/放棄された実行への再接続を拒否 | tombstoneキャッシュ |
| BR-80-05 | 操作ID重複防止 | 同一operationIdの実行を拒否（リトライ時のBR保証） | createExecuteHolder |
| BR-80-06 | セッションIDフォーマット | sessionIdはUUID形式であること | validateSessionCreate |

### 計算ロジック

- **セッション取得**: `sessionStore.computeIfAbsent(key, _ => callable())` (SparkConnectSessionManager.scala 213行目)
- **セッションタイムアウト判定**: `lastAccessTimeMs + timeoutMs <= nowMs` (336行目)
- **実行タイムアウト判定**: `detachedNs + timeoutNs <= nowNs` (SparkConnectExecutionManager.scala 339行目)

## データベース操作仕様

Spark Connectサーバー自体はデータベースを直接操作しない。クライアントからのSQL実行リクエストを受けて、Spark SQLエンジンに委譲する。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| INVALID_HANDLE.SESSION_CHANGED | セッションエラー | 以前観測されたセッションIDと一致しない | 新しいセッションを作成 |
| INVALID_HANDLE.SESSION_CLOSED | セッションエラー | 既に閉じられたセッションへのアクセス | 新しいセッションを作成 |
| INVALID_HANDLE.SESSION_NOT_FOUND | セッションエラー | 存在しないセッションへのアクセス | 新しいセッションを作成 |
| INVALID_HANDLE.FORMAT | フォーマットエラー | sessionId/operationIdがUUID形式でない | 正しいUUID形式を使用 |
| INVALID_HANDLE.OPERATION_ALREADY_EXISTS | 実行エラー | 同一operationIdの実行が既に存在 | 異なるoperationIdを使用 |
| INVALID_HANDLE.OPERATION_ABANDONED | 実行エラー | 放棄された実行への再リクエスト | 新しいoperationIdで再実行 |
| INVALID_CURSOR.NOT_REATTACHABLE | 実行エラー | 非再接続可能な実行への再接続 | 新しい実行を開始 |

### リトライ仕様

- operationIdの重複チェックにより、クライアントのリトライが二重実行を引き起こさない
- reattachExecuteにより、ネットワーク切断後の再接続が可能
- tombstoneキャッシュにより、完了/放棄された実行への不正な再アクセスを防止

## トランザクション仕様

Spark Connectサーバー自体にはトランザクション機構はない。ConcurrentHashMapのアトミック操作（computeIfAbsent）でスレッドセーフなセッション/実行管理を行う。

## パフォーマンス要件

- NettyServerBuilderによるNIOベースの非同期gRPC通信
- ConcurrentHashMapによるロックフリーなセッション/実行管理
- PeriodicMaintenanceスレッドによるバックグラウンドクリーンアップ
- maxInboundMessageSizeによるメモリ保護
- Protobuf marshallerのrecursionLimit設定

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

- PreSharedKeyAuthenticationInterceptorによるトークンベース認証
- SparkConnectInterceptorRegistryによるカスタムインターセプターの登録
- セッション分離によるマルチテナント環境での隔離
- userId + sessionIdの組み合わせによるセッション識別

## 備考

- SparkConnectServerはmain()メソッドを持つスタンドアロンアプリケーション
- サーバー起動時にSparkListenerConnectServiceStartedイベントを発行
- サーバー停止時にSparkListenerConnectServiceEndイベントを発行
- SQLConf.ARTIFACTS_SESSION_ISOLATION_ENABLED = true で起動（セッション分離有効）

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SessionHolder.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SessionHolder.scala` | SessionKey, SessionHolder, SessionHolderInfoの構造 |
| 1-2 | ExecuteHolder.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/ExecuteHolder.scala` | ExecuteKey, ExecuteHolder, ExecuteInfoの構造 |

**読解のコツ**: SessionKey = (userId, sessionId)、ExecuteKey = (userId, sessionId, operationId) で、セッションと実行の識別子階層を理解すること。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SparkConnectServer.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectServer.scala` | main()メソッド（30-58行目） - SparkSession作成、サービス起動、awaitTermination |

**主要処理フロー**:
- **30行目**: main メソッド
- **33-37行目**: SparkSession構築（ARTIFACTS_SESSION_ISOLATION_ENABLED = true）
- **40行目**: SparkConnectService.start(session.sparkContext)
- **50行目**: server.awaitTermination()

#### Step 3: サービス実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SparkConnectService.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala` | AsyncServiceの実装、各RPCハンドラーへの委譲パターン |

**主要処理フロー**:
- **58行目**: SparkConnectServiceクラス定義（AsyncService + BindableService）
- **70-82行目**: executePlan - SparkConnectExecutePlanHandlerに委譲
- **96-108行目**: analyzePlan - SparkConnectAnalyzeHandlerに委譲
- **117-129行目**: config - SparkConnectConfigHandlerに委譲
- **271-292行目**: bindService - gRPCサービス定義。marshallerのrecursionLimit設定
- **301行目**: SparkConnectServiceコンパニオンオブジェクト
- **380-432行目**: startGRPCService - NettyServerBuilderによるサーバー起動
- **435-448行目**: start - サーバー起動の一連の手順

#### Step 4: セッション管理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SparkConnectSessionManager.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectSessionManager.scala` | ConcurrentHashMapベースのセッション管理、periodicMaintenanceによる自動クリーンアップ |

**主要処理フロー**:
- **40行目**: SparkConnectSessionManagerクラス定義
- **48-49行目**: sessionStore - ConcurrentHashMap[SessionKey, SessionHolder]
- **92-106行目**: getOrCreateIsolatedSession - computeIfAbsentによるスレッドセーフなセッション作成
- **292-316行目**: schedulePeriodicChecks - メンテナンススレッドの起動
- **319-356行目**: periodicMaintenance - 期限切れセッションの検出と解放

#### Step 5: 実行管理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | SparkConnectExecutionManager.scala | `sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectExecutionManager.scala` | ExecuteHolder管理、reattach、periodic maintenance |

**主要処理フロー**:
- **66行目**: SparkConnectExecutionManagerクラス定義
- **69-70行目**: executions - ConcurrentHashMap[ExecuteKey, ExecuteHolder]
- **88-129行目**: createExecuteHolder - 重複チェック付きの実行作成
- **149-176行目**: removeExecuteHolder - 実行の解放とtombstone登録
- **186-211行目**: createExecuteHolderAndAttach - 実行作成 + 実行開始 + gRPCレスポンス送信
- **216-241行目**: reattachExecuteHolder - 再接続処理

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

```
SparkConnectServer.main()
    |
    +-- SparkSession.builder().getOrCreate()
    +-- SparkConnectService.start(sparkContext)
    |       +-- sessionManager.initializeBaseSession()
    |       +-- startGRPCService()
    |       |       +-- NettyServerBuilder.forPort(port)
    |       |       +-- sb.addService(SparkConnectService)
    |       |       +-- sb.intercept(PreSharedKeyAuth) [if configured]
    |       |       +-- server.start()
    |       +-- createListenerAndUI(sc)
    |       +-- postSparkConnectServiceStarted()
    |
    +-- server.awaitTermination()
    |
    +-- [クライアントRPC受信時]
            +-- SparkConnectService.executePlan(request, observer)
                    +-- SparkConnectExecutePlanHandler.handle(request)
                            +-- getOrCreateIsolatedSession(userId, sessionId)
                            +-- executionManager.createExecuteHolderAndAttach(...)
                                    +-- createExecuteHolder(key, request, session)
                                    +-- executeHolder.start()
                                    +-- ExecuteGrpcResponseSender(executeHolder, observer)
```

### データフロー図

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

クライアント                  gRPCサーバー
ExecutePlanRequest -----> SparkConnectService
                              |
                    SparkConnectExecutePlanHandler
                              |
                    セッション取得/作成 (SessionManager)
                              |
                    ExecuteHolder作成 (ExecutionManager)
                              |
                    Spark SQLプラン実行
                              |
                    ExecuteGrpcResponseSender
                              |
                              v
                    ExecutePlanResponse (stream) -----> クライアント
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SparkConnectServer.scala | `sql/connect/server/.../service/SparkConnectServer.scala` | ソース | サーバーエントリーポイント(main) |
| SparkConnectService.scala | `sql/connect/server/.../service/SparkConnectService.scala` | ソース | gRPCサービス実装 |
| SparkConnectSessionManager.scala | `sql/connect/server/.../service/SparkConnectSessionManager.scala` | ソース | セッション管理 |
| SparkConnectExecutionManager.scala | `sql/connect/server/.../service/SparkConnectExecutionManager.scala` | ソース | 実行管理 |
| SessionHolder.scala | `sql/connect/server/.../service/SessionHolder.scala` | ソース | セッションホルダー |
| ExecuteHolder.scala | `sql/connect/server/.../service/ExecuteHolder.scala` | ソース | 実行ホルダー |
| SparkConnectExecutePlanHandler.scala | `sql/connect/server/.../service/SparkConnectExecutePlanHandler.scala` | ソース | executePlanハンドラー |
| SparkConnectAnalyzeHandler.scala | `sql/connect/server/.../service/SparkConnectAnalyzeHandler.scala` | ソース | analyzePlanハンドラー |
| SparkConnectConfigHandler.scala | `sql/connect/server/.../service/SparkConnectConfigHandler.scala` | ソース | configハンドラー |
| SparkConnectAddArtifactsHandler.scala | `sql/connect/server/.../service/SparkConnectAddArtifactsHandler.scala` | ソース | addArtifactsハンドラー |
| SparkConnectInterruptHandler.scala | `sql/connect/server/.../service/SparkConnectInterruptHandler.scala` | ソース | interruptハンドラー |
| SparkConnectReattachExecuteHandler.scala | `sql/connect/server/.../service/SparkConnectReattachExecuteHandler.scala` | ソース | reattachExecuteハンドラー |
| SparkConnectReleaseExecuteHandler.scala | `sql/connect/server/.../service/SparkConnectReleaseExecuteHandler.scala` | ソース | releaseExecuteハンドラー |
| SparkConnectReleaseSessionHandler.scala | `sql/connect/server/.../service/SparkConnectReleaseSessionHandler.scala` | ソース | releaseSessionハンドラー |
| SparkConnectCloneSessionHandler.scala | `sql/connect/server/.../service/SparkConnectCloneSessionHandler.scala` | ソース | cloneSessionハンドラー |
| SparkConnectFetchErrorDetailsHandler.scala | `sql/connect/server/.../service/SparkConnectFetchErrorDetailsHandler.scala` | ソース | fetchErrorDetailsハンドラー |
| SparkConnectInterceptorRegistry.scala | `sql/connect/server/.../service/SparkConnectInterceptorRegistry.scala` | ソース | インターセプター管理 |
| PreSharedKeyAuthenticationInterceptor.scala | `sql/connect/server/.../service/PreSharedKeyAuthenticationInterceptor.scala` | ソース | トークン認証 |
| SparkConnectStreamingQueryCache.scala | `sql/connect/server/.../service/SparkConnectStreamingQueryCache.scala` | ソース | ストリーミングクエリキャッシュ |
| ExecuteEventsManager.scala | `sql/connect/server/.../service/ExecuteEventsManager.scala` | ソース | 実行イベント管理 |
| SessionEventsManager.scala | `sql/connect/server/.../service/SessionEventsManager.scala` | ソース | セッションイベント管理 |
