# 機能設計書 38-ステート管理

## 概要

本ドキュメントは、Apache SparkのStructured Streamingにおけるステート管理機能に関する機能設計書である。ストリーミングクエリの状態（ウィンドウ集約、セッション、結合等）を永続化し、障害復旧時に正確な状態を復元するStateStoreフレームワークを詳細に記述する。

### 本機能の処理概要

**業務上の目的・背景**：ステートフルなストリーム処理（ウィンドウ集約、ストリーム間結合、重複排除等）では、バッチ間で状態を維持する必要がある。StateStoreは、各パーティションの状態をバージョン管理されたキーバリューストアとして提供し、障害発生時には特定バージョンの状態を復元してExactly-once処理保証を実現する。

**機能の利用シーン**：ウィンドウ集約（タンブリング/スライディング/セッションウィンドウ）、ストリーム間ジョイン、mapGroupsWithState/flatMapGroupsWithStateによるカスタムステートフル処理、重複排除。

**主要な処理内容**：
1. StateStore：バージョン管理されたキーバリューストアインターフェース
2. HDFSBackedStateStoreProvider：HDFS/クラウドストレージベースのステートストアプロバイダー
3. RocksDBStateStoreProvider：RocksDB組み込みデータベースベースの高性能ステートストアプロバイダー
4. StateStoreCoordinator：ドライバー上でStateStoreインスタンスの位置情報を管理するコーディネーター
5. StateStoreRDD：ステートストアアクセスを分散タスクに提供するRDD
6. ステートスキーマ互換性検証：StateSchemaCompatibilityChecker
7. RocksDBStateEncoder：RocksDBキー/バリューのエンコーディング
8. OperatorStateMetadata：ステートフルオペレーターのメタデータ管理

**関連システム・外部連携**：HDFS/S3等の分散ストレージ、RocksDB組み込みデータベース、チェックポイント管理

**権限による制御**：ステートストアのバックエンドストレージへのアクセス権限が必要。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 17 | Streaming Query Statistics（ストリーミングクエリ統計） | 補助機能 | State Store関連のカスタムメトリクス（state rows等）を取得して表示 |

## 機能種別

データストレージ / 状態管理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| storeId | StateStoreId | Yes | ステートストア識別子（checkpointLocation, operatorId, partitionId, storeName） | 有効なID |
| version | Long | Yes | アクセス対象のステートバージョン | 0以上 |
| keySchema | StructType | Yes | キーのスキーマ | 有効なStructType |
| valueSchema | StructType | Yes | バリューのスキーマ | 有効なStructType |
| storeConf | StateStoreConf | Yes | ストア設定 | 有効な設定 |
| hadoopConf | Configuration | Yes | Hadoop設定 | 有効な設定 |

### 入力データソース

- ステートフルオペレーターからのキーバリューペア
- 前バージョンのステートデータ（障害復旧時）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| UnsafeRow | UnsafeRow | ステートのキー/バリュー行 |
| Iterator[(UnsafeRow, UnsafeRow)] | Iterator | ステートの全キーバリューペア |
| StateStoreMetrics | StateStoreMetrics | ステートストアのメトリクス |

### 出力先

- ステートフルオペレーター（読み取り）
- バックエンドストレージ（永続化、HDFSまたはRocksDB）

## 処理フロー

### 処理シーケンス

```
1. StateStoreProvider初期化
   └─ StateStore.get() → StateStoreProvider.createAndInit()
2. ステートバージョンロード
   └─ provider.getStore(version) でバージョン指定のストアを取得
3. ステート操作
   └─ get(key)/put(key, value)/remove(key)/iterator() で状態を操作
4. ステートコミット
   └─ store.commit() で現在の状態をバージョン+1として永続化
5. メンテナンス
   └─ doMaintenance() で古いバージョンのクリーンアップ（スナップショット保持）
6. StateStoreCoordinator連携
   └─ ドライバーがストアインスタンスの場所を追跡
```

### フローチャート

```mermaid
flowchart TD
    A[ステートフルオペレーター] --> B[StateStoreRDD]
    B --> C[StateStore.get - Provider取得]
    C --> D{Provider種別}
    D -->|HDFS| E[HDFSBackedStateStoreProvider]
    D -->|RocksDB| F[RocksDBStateStoreProvider]
    E --> G[getStore - バージョンロード]
    F --> G
    G --> H[ステート操作 get/put/remove]
    H --> I[commit - 永続化]
    I --> J[doMaintenance - クリーンアップ]
    J --> K[StateStoreCoordinator - 位置情報更新]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | バージョン管理 | 各コミットで新しいバージョンが作成され、特定バージョンへのロールバックが可能 | commit()時 |
| BR-02 | スキーマ互換性 | ステートスキーマの変更は互換性チェッカーで検証される | ストア再オープン時 |
| BR-03 | メモリ管理 | RocksDBStateStoreProviderはRocksDBMemoryManagerで共有メモリを管理 | RocksDB使用時 |
| BR-04 | エンコーディング | UnsafeRowエンコーディングとAvroエンコーディングから選択可能 | StateStoreEncoding設定 |

### 計算ロジック

- **HDFS Provider**: デルタファイル（変更分）とスナップショットファイル（全量）の組み合わせでバージョン管理
- **RocksDB Provider**: RocksDBのネイティブバージョニングとチェックポイント機能を活用
- **StateStoreCoordinator**: RPC経由でExecutor上のStateStoreインスタンス位置を追跡

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

### 操作別データベース影響一覧

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| get | StateStore | SELECT | キーに対応するバリューの取得 |
| put | StateStore | INSERT/UPDATE | キーバリューペアの挿入/更新 |
| remove | StateStore | DELETE | キーの削除 |
| commit | StateStore | WRITE | 状態の永続化 |

### テーブル別操作詳細

#### StateStore（キーバリューストア）

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| get | key → value | UnsafeRowキーで検索 | キーが存在しない場合null |
| put | key, value | UnsafeRowペアを更新 | 既存キーは上書き |
| remove | key | 指定キーを削除 | キーが存在しない場合は無操作 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| STATE_STORE_UNAVAILABLE | QueryExecutionError | StateStoreの取得に失敗 | ストレージアクセスを確認 |
| SCHEMA_INCOMPATIBLE | StateStoreError | ステートスキーマの互換性なし | スキーマ変更の互換性を確認 |
| VERSION_NOT_FOUND | StateStoreError | 要求されたバージョンが存在しない | チェックポイントを確認 |

### リトライ仕様

StateStore操作はタスクリトライの対象。タスク失敗時は別のExecutorでストアを再ロードして処理を再開。

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

StateStore.commit()はアトミックに実行される。コミット完了前にタスクが失敗した場合、ストアは以前のバージョンに戻る。

## パフォーマンス要件

- RocksDBStateStoreProviderはHDFSBackedよりも大規模ステートで高パフォーマンス
- RocksDBMemoryManagerによりメモリ使用量を制限可能
- prefixScanによる効率的な範囲検索（セッションウィンドウ等で使用）

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

- バックエンドストレージのアクセス権限管理
- ステートデータの暗号化はバックエンドストレージの機能に依存

## 備考

- StateStoreは `sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/state/` 配下に実装
- RocksDBStateStoreProviderがデフォルトの推奨プロバイダー
- StateStoreEncodingとしてUnsafeRowとAvroの2種類をサポート

---

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

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

### 推奨読解順序

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

StateStoreのインターフェースと設定を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StateStore.scala | `sql/core/.../streaming/state/StateStore.scala` | ReadStateStore/StateStoreトレイト、StateStoreProviderトレイト、StateStoreIterator |
| 1-2 | StateStoreConf.scala | `sql/core/.../streaming/state/StateStoreConf.scala` | ストア設定パラメータ |

**読解のコツ**: StateStoreは ReadStateStore（読み取り専用）と StateStore（読み書き）の2レベルに分かれている。StateStoreEncodingはUnsafeRowとAvroの2種類がある（69-79行目）。MaintenanceTaskType（81-87行目）はメンテナンス処理のソースを区別する。

#### Step 2: プロバイダー実装を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | HDFSBackedStateStoreProvider.scala | `sql/core/.../streaming/state/` | デルタ/スナップショットベースの実装 |
| 2-2 | RocksDBStateStoreProvider.scala | `sql/core/.../streaming/state/` | RocksDBベースの高性能実装 |
| 2-3 | RocksDB.scala | `sql/core/.../streaming/state/` | RocksDBラッパー |

#### Step 3: コーディネーターとRDDを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StateStoreCoordinator.scala | `sql/core/.../streaming/state/` | ドライバー側のストアインスタンス管理 |
| 3-2 | StateStoreRDD.scala | `sql/core/.../streaming/state/` | ステートストアアクセスの分散RDD |

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

```
ステートフルオペレーター
    │
    └─ StateStoreRDD
           │
           ├─ StateStore.get(storeProviderId, keySchema, valueSchema, ...)
           │      ├─ StateStoreProvider.createAndInit()
           │      │      ├─ HDFSBackedStateStoreProvider
           │      │      └─ RocksDBStateStoreProvider → RocksDB
           │      └─ provider.getStore(version)
           │
           ├─ store.get(key) / put(key, value) / remove(key)
           ├─ store.commit()
           │
           └─ StateStoreCoordinator
                  └─ reportActiveStoreInstance()
```

### データフロー図

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

キーバリューペア          StateStore                       永続化ストレージ
(UnsafeRow)       ──▶  put/get/remove              ──▶  (HDFS/RocksDB)

前バージョンステート      StateStoreProvider              現バージョンステート
(checkpoint)      ──▶  getStore(version)            ──▶  (in-memory)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StateStore.scala | `sql/core/.../streaming/state/` | ソース | コアインターフェース |
| StateStoreConf.scala | `sql/core/.../streaming/state/` | ソース | 設定管理 |
| HDFSBackedStateStoreProvider.scala | `sql/core/.../streaming/state/` | ソース | HDFSプロバイダー |
| RocksDBStateStoreProvider.scala | `sql/core/.../streaming/state/` | ソース | RocksDBプロバイダー |
| RocksDB.scala | `sql/core/.../streaming/state/` | ソース | RocksDBラッパー |
| RocksDBStateEncoder.scala | `sql/core/.../streaming/state/` | ソース | RocksDBエンコーダー |
| RocksDBMemoryManager.scala | `sql/core/.../streaming/state/` | ソース | RocksDBメモリ管理 |
| RocksDBFileManager.scala | `sql/core/.../streaming/state/` | ソース | RocksDBファイル管理 |
| StateStoreCoordinator.scala | `sql/core/.../streaming/state/` | ソース | ストアコーディネーター |
| StateStoreRDD.scala | `sql/core/.../streaming/state/` | ソース | 分散アクセスRDD |
| StateSchemaCompatibilityChecker.scala | `sql/core/.../streaming/state/` | ソース | スキーマ互換性検証 |
| OperatorStateMetadata.scala | `sql/core/.../streaming/state/` | ソース | オペレーターメタデータ |
| StateStoreChangelog.scala | `sql/core/.../streaming/state/` | ソース | 変更ログ管理 |
