# 機能設計書 74-State Processing API

## 概要

本ドキュメントは、Apache Flink の State Processing API に関する機能設計書である。flink-state-processing-api モジュールで提供される、セーブポイントの読み取り・書き込み機能について、その構造、処理フロー、および実装詳細を記載する。

### 本機能の処理概要

State Processing API は、Flink ジョブのセーブポイント内のステートを読み取り・変更・作成するためのライブラリである。DataStream API を使用してバッチ処理としてセーブポイントを操作でき、新規アプリケーションのステートブートストラップや既存ジョブのステートマイグレーションを可能にする。

**業務上の目的・背景**：
- 新規 Flink ジョブに初期状態を設定（状態ブートストラップ）
- 既存ジョブのステート構造変更時のマイグレーション
- セーブポイント内のステートを分析・デバッグ
- オペレーター UID の変更やステートの再編成

**機能の利用シーン**：
- レガシーシステムから Flink への移行時に既存データをステートとして投入
- スキーマ変更に伴うステートマイグレーション
- ジョブの障害調査時にセーブポイント内の状態を確認
- オペレーターの並列度変更に伴うステート再配置

**主要な処理内容**：
1. セーブポイント読み取り（SavepointReader）
   - ListState、UnionState、BroadcastState、KeyedState の読み取り
   - Window State の読み取り
2. セーブポイント書き込み（SavepointWriter）
   - 新規セーブポイントの作成
   - 既存セーブポイントの変更・拡張
   - オペレーター UID の変更

**関連システム・外部連携**：
- Flink State Backend（RocksDB、HashMapStateBackend）
- Flink Checkpoint/Savepoint メカニズム
- Flink DataStream API

**権限による制御**：特になし（ファイルシステム権限に依存）

## 関連画面

本機能はバッチ処理ライブラリであり、直接関連する画面はない。

## 機能種別

ステート管理ライブラリ / セーブポイント操作

## 入力仕様

### 入力パラメータ（SavepointReader）

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| env | StreamExecutionEnvironment | Yes | 実行環境 | null不可 |
| path | String | Yes | セーブポイントパス | 有効なファイルパス |
| stateBackend | StateBackend | No | ステートバックエンド | - |
| identifier | OperatorIdentifier | Yes | オペレーター識別子 | - |

### 入力パラメータ（SavepointWriter）

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| executionEnvironment | StreamExecutionEnvironment | Yes | 実行環境 | null不可 |
| path | String | Yes | 既存セーブポイントパス（fromExistingSavepoint時） | - |
| maxParallelism | int | Yes | 最大並列度（newSavepoint時） | 1 <= x <= 32768 |
| stateBackend | StateBackend | No | ステートバックエンド | - |

### 入力データソース

- 既存セーブポイントファイル（_metadata）
- ブートストラップ用 DataStream

## 出力仕様

### 出力データ（SavepointReader）

| 項目名 | 型 | 説明 |
|--------|-----|------|
| listState | DataStream<T> | ListState 要素のストリーム |
| unionState | DataStream<T> | UnionState 要素のストリーム |
| broadcastState | DataStream<Tuple2<K,V>> | BroadcastState のキー・バリューペア |
| keyedState | DataStream<OUT> | キー付きステートの読み取り結果 |

### 出力データ（SavepointWriter）

| 項目名 | 型 | 説明 |
|--------|-----|------|
| savepoint | ファイル | _metadata ファイルとステートファイル群 |

### 出力先

- 指定されたファイルシステムパス

## 処理フロー

### 処理シーケンス（セーブポイント読み取り）

```
1. セーブポイントメタデータ読み込み
   └─ SavepointLoader.loadSavepointMetadata()
   └─ CheckpointMetadata 取得
   └─ maxParallelism 算出

2. SavepointReader インスタンス作成
   └─ SavepointMetadataV2 作成
   └─ StateBackend 設定

3. ステート読み取り
   └─ readListState() / readUnionState() / readBroadcastState() / readKeyedState()
   └─ OperatorState 取得
   └─ InputFormat 作成（ListStateInputFormat, KeyedStateInputFormat等）
   └─ SourceBuilder.fromFormat() で DataStream 生成
```

### 処理シーケンス（セーブポイント書き込み）

```
1. SavepointWriter インスタンス作成
   └─ newSavepoint() または fromExistingSavepoint()
   └─ SavepointMetadataV2 作成/読み込み

2. オペレーターステート追加
   └─ withOperator(identifier, transformation)
   └─ StateBootstrapTransformation 登録

3. セーブポイント書き込み
   └─ write(path)
   └─ writeOperatorStates() で各オペレーターのステート生成
   └─ MergeOperatorStates で結合
   └─ SavepointOutputFormat で _metadata 出力
```

### フローチャート

```mermaid
flowchart TD
    subgraph Reader [SavepointReader]
        A1[read] --> A2[loadSavepointMetadata]
        A2 --> A3[SavepointMetadataV2作成]
        A3 --> A4{読み取り種別}
        A4 -->|ListState| A5[ListStateInputFormat]
        A4 -->|UnionState| A6[UnionStateInputFormat]
        A4 -->|BroadcastState| A7[BroadcastStateInputFormat]
        A4 -->|KeyedState| A8[KeyedStateInputFormat]
        A5 --> A9[DataStream生成]
        A6 --> A9
        A7 --> A9
        A8 --> A9
    end

    subgraph Writer [SavepointWriter]
        B1{新規/既存?}
        B1 -->|新規| B2[newSavepoint]
        B1 -->|既存| B3[fromExistingSavepoint]
        B2 --> B4[SavepointMetadataV2作成]
        B3 --> B5[loadSavepointMetadata]
        B5 --> B4
        B4 --> B6[withOperator]
        B6 --> B7[StateBootstrapTransformation登録]
        B7 --> B8[write]
        B8 --> B9[writeOperatorStates]
        B9 --> B10[MergeOperatorStates]
        B10 --> B11[SavepointOutputFormat]
        B11 --> B12[_metadata出力]
    end
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-74-01 | maxParallelism 制約 | maxParallelism は 1 以上 32768 以下 | newSavepoint時 |
| BR-74-02 | オペレーター存在確認 | 読み取り時に指定オペレーターが存在する必要がある | 読み取り時 |
| BR-74-03 | UID 変更の適用順序 | UID 変更は他の操作より後に適用される | write()時 |
| BR-74-04 | BroadcastState並列度 | BroadcastState ブートストラップは並列度1 | 書き込み時 |
| BR-74-05 | 最小1オペレーター | セーブポイントには最低1つのオペレーターが必要 | write()時 |

### 計算ロジック

| ロジックNo | ロジック名 | 内容 |
|-----------|-----------|------|
| CL-74-01 | maxParallelism 算出 | セーブポイント内の全オペレーターの maxParallelism の最大値 |
| CL-74-02 | 並列度調整 | 現在の並列度が localMaxParallelism を超える場合、localMaxParallelism に設定 |

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

本機能はデータベース操作を行わない。ステートは Flink State Backend でファイルシステムに永続化される。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | ファイルエラー | セーブポイントパスが無効 | パスを確認 |
| RuntimeException | オペレーター不存在 | 指定UIDのオペレーターが見つからない | UIDを確認 |
| IllegalStateException | 空セーブポイント | オペレーターが0個 | オペレーターを追加 |
| IllegalArgumentException | maxParallelism不正 | 範囲外の値 | 1-32768の値を指定 |
| FlinkRuntimeException | UID変更未適用 | changeOperatorIdentifier()で指定したUIDが存在しない | UIDを確認 |

### リトライ仕様

バッチ処理として実行されるため、ジョブ全体の再実行により対応する。

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

本機能はトランザクション管理を行わない。セーブポイント書き込みは原子的に行われる（_metadata が最後に書き込まれる）。

## パフォーマンス要件

- 大規模ステートの読み取り：並列処理で分散
- メモリ使用量：State Backend の設定に依存

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

- ファイルシステム権限に依存
- 機密データを含むステートの取り扱いに注意

## 備考

- DataStream API を使用したバッチ処理
- Flink 1.9 で導入

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SavepointReader.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointReader.java` | セーブポイント読み取りエントリーポイント |
| 1-2 | SavepointWriter.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointWriter.java` | セーブポイント書き込みエントリーポイント |

**主要処理フロー（SavepointReader.java）**:
- **70-90行目**: read() - セーブポイント読み込みのエントリーポイント
- **162-166行目**: readListState() - ListState 読み取り
- **217-221行目**: readUnionState() - UnionState 読み取り
- **274-285行目**: readBroadcastState() - BroadcastState 読み取り
- **347-384行目**: readKeyedState() - キー付きステート読み取り
- **424-428行目**: window() - ウィンドウステート読み取り

**主要処理フロー（SavepointWriter.java）**:
- **79-82行目**: fromExistingSavepoint() - 既存セーブポイント読み込み
- **126-130行目**: newSavepoint() - 新規セーブポイント作成
- **227-230行目**: removeOperator() - オペレーター削除
- **239-243行目**: withOperator() - オペレーター追加
- **298-302行目**: changeOperatorIdentifier() - UID 変更
- **309-348行目**: write() - セーブポイント書き込み

#### Step 2: ステート変換を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | OperatorTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OperatorTransformation.java` | 変換ビルダーエントリーポイント |
| 2-2 | OneInputStateTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OneInputStateTransformation.java` | 非キー付きステート変換 |
| 2-3 | KeyedStateTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/KeyedStateTransformation.java` | キー付きステート変換 |
| 2-4 | StateBootstrapTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java` | ステートブートストラップ変換 |

**主要処理フロー（StateBootstrapTransformation.java）**:
- **127-148行目**: writeOperatorState() - オペレーターステート書き込み
- **164-204行目**: writeOperatorSubtaskStates() - サブタスクステート書き込み
- **207-238行目**: getConfig() - StreamConfig 設定

#### Step 3: 入力フォーマットを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ListStateInputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/ListStateInputFormat.java` | ListState 読み取り |
| 3-2 | KeyedStateInputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java` | キー付きステート読み取り |
| 3-3 | BroadcastStateInputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/BroadcastStateInputFormat.java` | BroadcastState 読み取り |

#### Step 4: ブートストラップ関数を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | KeyedStateBootstrapFunction.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/KeyedStateBootstrapFunction.java` | キー付きステートブートストラップ |
| 4-2 | StateBootstrapFunction.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/StateBootstrapFunction.java` | 非キー付きステートブートストラップ |
| 4-3 | BroadcastStateBootstrapFunction.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/BroadcastStateBootstrapFunction.java` | BroadcastStateブートストラップ |
| 4-4 | KeyedStateReaderFunction.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/KeyedStateReaderFunction.java` | キー付きステート読み取り関数 |

#### Step 5: 出力処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | SavepointOutputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java` | _metadata 出力 |
| 5-2 | MergeOperatorStates.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/MergeOperatorStates.java` | オペレーターステート結合 |
| 5-3 | OperatorSubtaskStateReducer.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/OperatorSubtaskStateReducer.java` | サブタスクステート集約 |

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

```
// 読み取りフロー
SavepointReader.read(env, path)
    │
    └─ SavepointLoader.loadSavepointMetadata()
           │
           └─ CheckpointMetadata 取得
                  │
                  └─ SavepointMetadataV2 作成

SavepointReader.readKeyedState(identifier, function)
    │
    ├─ metadata.getOperatorState(identifier)
    │
    └─ KeyedStateInputFormat 作成
           │
           └─ SourceBuilder.fromFormat()
                  │
                  └─ DataStream<OUT> 生成

// 書き込みフロー
SavepointWriter.newSavepoint(env, maxParallelism)
    │
    └─ createSavepointMetadata()
           │
           └─ SavepointMetadataV2 作成

SavepointWriter.withOperator(identifier, transformation)
    │
    └─ metadata.addOperator()

SavepointWriter.write(path)
    │
    ├─ writeOperatorStates()
    │      │
    │      └─ for each transformation:
    │             └─ transformation.writeOperatorState()
    │                    │
    │                    └─ writeOperatorSubtaskStates()
    │                           │
    │                           ├─ factory.createOperator()
    │                           ├─ BootstrapStreamTaskRunner
    │                           └─ DataStream<TaggedOperatorSubtaskState>
    │
    ├─ getFinalOperatorStates()
    │      │
    │      └─ existingOperators + newOperatorStates
    │
    └─ GroupReduceOperator(MergeOperatorStates)
           │
           └─ OutputFormatSink(SavepointOutputFormat)
                  │
                  └─ _metadata 書き込み
```

### データフロー図

```
[読み取りフロー]

セーブポイント                SavepointReader              DataStream
(_metadata)
    │                              │                           │
    │   loadSavepointMetadata      │                           │
    └──────────────────────────────▶                           │
                                   │                           │
                   CheckpointMetadata                          │
                   OperatorState[]                             │
                                   │                           │
                        readKeyedState(uid, func)              │
                                   │                           │
                         KeyedStateInputFormat                 │
                                   │                           │
                                   │   createSplits()          │
                                   │   KeyGroupRangeごと分割    │
                                   │                           │
                                   │   nextRecord()            │
                                   │   StateBackendから読み取り │
                                   │                           │
                                   └───────────────────────────▶
                                                               │
                                                    DataStream<OUT>


[書き込みフロー]

DataStream                  StateBootstrapTransformation      セーブポイント
    │                                  │                           │
    │   bootstrapWith(stream)          │                           │
    └──────────────────────────────────▶                           │
                                       │                           │
                                 keyBy / transform                 │
                                       │                           │
                            StateBootstrapTransformation           │
                                       │                           │
                       write(path)     │                           │
                                       │                           │
                            writeOperatorSubtaskStates()           │
                                       │                           │
                       BootstrapStreamTaskRunner                   │
                                       │                           │
                                    snapshot()                     │
                                       │                           │
                       TaggedOperatorSubtaskState                  │
                                       │                           │
                       OperatorSubtaskStateReducer                 │
                                       │                           │
                                 OperatorState                     │
                                       │                           │
                             MergeOperatorStates                   │
                                       │                           │
                             CheckpointMetadata                    │
                                       │                           │
                            SavepointOutputFormat                  │
                                       │                           │
                                       └───────────────────────────▶
                                                                   │
                                                          _metadata + state files
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SavepointReader.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointReader.java` | ソース | セーブポイント読み取り |
| SavepointWriter.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/SavepointWriter.java` | ソース | セーブポイント書き込み |
| OperatorTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OperatorTransformation.java` | ソース | 変換エントリーポイント |
| OneInputStateTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OneInputStateTransformation.java` | ソース | 非キー付き変換 |
| KeyedStateTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/KeyedStateTransformation.java` | ソース | キー付き変換 |
| WindowedStateTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WindowedStateTransformation.java` | ソース | ウィンドウ変換 |
| StateBootstrapTransformation.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/StateBootstrapTransformation.java` | ソース | ブートストラップ変換 |
| OperatorIdentifier.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/OperatorIdentifier.java` | ソース | オペレーター識別子 |
| WindowSavepointReader.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/WindowSavepointReader.java` | ソース | ウィンドウステート読み取り |
| ListStateInputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/ListStateInputFormat.java` | ソース | ListState 入力 |
| UnionStateInputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/UnionStateInputFormat.java` | ソース | UnionState 入力 |
| BroadcastStateInputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/BroadcastStateInputFormat.java` | ソース | BroadcastState 入力 |
| KeyedStateInputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/input/KeyedStateInputFormat.java` | ソース | KeyedState 入力 |
| KeyedStateBootstrapFunction.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/KeyedStateBootstrapFunction.java` | ソース | キー付きブートストラップ |
| StateBootstrapFunction.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/StateBootstrapFunction.java` | ソース | 非キー付きブートストラップ |
| BroadcastStateBootstrapFunction.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/BroadcastStateBootstrapFunction.java` | ソース | Broadcast ブートストラップ |
| KeyedStateReaderFunction.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/functions/KeyedStateReaderFunction.java` | ソース | キー付きステート読み取り |
| SavepointOutputFormat.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/SavepointOutputFormat.java` | ソース | セーブポイント出力 |
| MergeOperatorStates.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/MergeOperatorStates.java` | ソース | ステート結合 |
| OperatorSubtaskStateReducer.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/output/OperatorSubtaskStateReducer.java` | ソース | サブタスク集約 |
| SavepointLoader.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/SavepointLoader.java` | ソース | セーブポイント読み込み |
| SavepointMetadataV2.java | `flink-libraries/flink-state-processing-api/src/main/java/org/apache/flink/state/api/runtime/metadata/SavepointMetadataV2.java` | ソース | メタデータ管理 |
