# 機能設計書 37-SpillableStateBackend

## 概要

本ドキュメントは、Apache FlinkのSpillableStateBackend（ヒープスピラブルステートバックエンド）機能について記載する。この機能は、Copy-on-Write方式のSkipListベースStateMapを実装し、効率的なスナップショットとメモリ管理を実現する。

### 本機能の処理概要

SpillableStateBackendは、CopyOnWriteSkipListStateMapを中心としたステート管理機能を提供する。SkipList（スキップリスト）データ構造を採用し、ステートをシリアライズしてAllocatorから割り当てられたメモリ空間に格納する。Copy-on-Write方式により、スナップショット中の更新と読み取りの競合を回避する。

**業務上の目的・背景**：大規模ステートを扱う場合、ヒープベースのステートバックエンドではGC負荷が高くなる。SpillableStateBackendは、ステートをシリアライズしてオフヒープまたは管理されたメモリ領域に格納することで、GCプレッシャーを軽減する。また、Copy-on-Write方式により、スナップショット時のブロッキングを最小化する。

**機能の利用シーン**：
- 大規模ステートでのGC負荷軽減が必要な場合
- 高頻度チェックポイントと低レイテンシが必要な場合
- メモリ効率を最大化したいストリーム処理
- Copy-on-Write方式のスナップショットが必要な場合

**主要な処理内容**：
1. SkipListベースのステートマップ管理
2. Copy-on-Writeによる並行スナップショット
3. Allocatorによるメモリ空間管理
4. 論理削除と物理削除の分離
5. バージョン管理によるスナップショット整合性

**関連システム・外部連携**：
- Allocator（メモリ空間割り当て）
- CheckpointStorage
- TypeSerializer

**権限による制御**：特定のロールによる制御なし。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は画面とは直接関連しない（バックエンド処理） |

## 機能種別

状態管理 / メモリ管理 / Copy-on-Writeスナップショット

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| keySerializer | TypeSerializer<K> | Yes | キーシリアライザ | 非null |
| namespaceSerializer | TypeSerializer<N> | Yes | 名前空間シリアライザ | 非null |
| stateSerializer | TypeSerializer<S> | Yes | ステートシリアライザ | 非null |
| spaceAllocator | Allocator | Yes | メモリ空間アロケータ | 非null |
| numKeysToDeleteOneTime | int | No | 一度に削除するキー数 | 0以上（デフォルト3） |
| logicalRemovedKeysRatio | float | No | 論理削除キー比率 | 0.0-1.0（デフォルト0.2） |

### 入力データソース

- Flink設定
- TypeSerializerの登録

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Snapshot | CopyOnWriteSkipListStateMapSnapshot | スナップショット |
| State | S | ステート値 |

### 出力先

- CheckpointStorage

## 処理フロー

### 処理シーケンス

```
1. CopyOnWriteSkipListStateMap初期化
   └─ Allocator、LevelIndexHeader、シリアライザ設定

2. ステート操作
   └─ put/get/remove
   └─ キーをシリアライズしてSkipListで検索

3. 更新操作
   └─ スナップショット中はCopy-on-Write
   └─ スナップショットなしは直接置換

4. スナップショット
   └─ バージョン増加
   └─ CopyOnWriteSkipListStateMapSnapshot作成

5. スナップショットリリース
   └─ バージョン管理更新
   └─ 不要バージョンの値をプルーニング
```

### フローチャート

```mermaid
flowchart TD
    A[ステート操作] --> B{操作タイプ?}
    B -->|get| C[SkipList検索]
    C --> D[ノード取得]
    D --> E[値デシリアライズ]
    B -->|put| F[キーシリアライズ]
    F --> G{キー存在?}
    G -->|Yes| H{スナップショット中?}
    H -->|Yes| I[Copy-on-Write更新]
    H -->|No| J[直接置換]
    G -->|No| K[新規ノード作成]
    B -->|remove| L{スナップショット中?}
    L -->|Yes| M[論理削除]
    L -->|No| N[物理削除]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-37-01 | Copy-on-Write | スナップショット中の更新はCopy-on-Write | スナップショット中 |
| BR-37-02 | 論理削除 | スナップショット中は論理削除のみ | スナップショット中 |
| BR-37-03 | 遅延物理削除 | スナップショット完了後に物理削除 | スナップショット完了後 |
| BR-37-04 | バージョン管理 | 各値にバージョンを付与 | 常時 |

### 計算ロジック

- size() = totalSize - logicallyRemovedNodes.size()
- 論理削除トリガー: logicallyRemovedNodes.size() > totalSize * logicalRemovedKeysRatio

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | RDBMSは使用しない |

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

該当なし（インメモリSkipList）

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ALLOC | FlinkRuntimeException | メモリ割り当て失敗 | メモリ設定確認 |
| VERSION | IllegalStateException | バージョンオーバーフロー | 再起動 |
| GUARD | RuntimeException | リソースガード取得失敗 | 競合確認 |

### リトライ仕様

特になし（メモリ操作は即時失敗）

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

- Copy-on-Write方式による楽観的並行制御
- ResourceGuardによるスナップショット同期

## パフォーマンス要件

- SkipListによるO(log n)検索
- Copy-on-Write方式で読み取りはロックフリー

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

メモリ上のデータは暗号化されない

## 備考

- JDK ConcurrentSkipListMapのアルゴリズムを参考
- ランダムレベル生成はXor-shiftアルゴリズム使用

---

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

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

### 推奨読解順序

#### Step 1: メインクラスを理解する

CopyOnWriteSkipListStateMapのデータ構造と基本操作を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | CopyOnWriteSkipListStateMap.java | `flink-statebackend-heap-spillable/src/main/java/org/apache/flink/runtime/state/heap/` | SkipListベースStateMap |

**読解のコツ**:
- **62-71行目**: クラスコメントで設計意図を把握
- **70-71行目**: StateMapを継承、AutoCloseableを実装
- **82-89行目**: シリアライザとアロケータの保持
- **96-103行目**: バージョン管理（stateMapVersion、highestRequiredSnapshotVersionPlusOne）
- **114-116行目**: totalSize、logicallyRemovedNodes

#### Step 2: ステート操作を理解する

get/put/remove操作の実装を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | CopyOnWriteSkipListStateMap.java | 同上 | 197-263行目 |

**読解のコツ**:
- **197-211行目**: get()、containsKey()
- **213-231行目**: put()、putAndGetOld()
- **233-249行目**: remove()、removeAndGetOld()
- **356-383行目**: putValue() Copy-on-Write判定

#### Step 3: Copy-on-Write実装を理解する

スナップショット時の更新処理を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | CopyOnWriteSkipListStateMap.java | 同上 | 599-656行目 |

**読解のコツ**:
- **599-624行目**: updateValueWithCopyOnWrite() 新値を作成し古値にリンク
- **635-656行目**: updateValueWithReplace() 直接置換

#### Step 4: スナップショット管理を理解する

スナップショット作成とリリースを確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | CopyOnWriteSkipListStateMap.java | 同上 | 1217-1264行目 |
| 4-2 | CopyOnWriteSkipListStateMapSnapshot.java | 同パス | スナップショット実装 |

**読解のコツ**:
- **1217-1241行目**: stateSnapshot() バージョン増加、スナップショット作成
- **1245-1264行目**: releaseSnapshot() バージョン管理更新

#### Step 5: Allocatorインターフェースを理解する

メモリ空間管理を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | Allocator.java | `flink-statebackend-heap-spillable/.../space/` | メモリ割り当てAPI |

**読解のコツ**:
- **24行目**: Closeable実装
- **33行目**: allocate(size) メモリ割り当て
- **40行目**: free(address) メモリ解放
- **48行目**: getChunkById() チャンク取得

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

```
CopyOnWriteSkipListStateMap
    │
    ├─ StateMap<K, N, S> 継承
    │
    ├─ get(key, namespace)
    │      ├─ getKeySegment() → skipListKeySerializer.serializeToSegment()
    │      ├─ getNode() → iterateAndProcess()
    │      └─ helpGetState() → skipListValueSerializer.deserializeState()
    │
    ├─ put(key, namespace, state)
    │      ├─ getKeySegment()
    │      ├─ skipListValueSerializer.serialize()
    │      └─ putValue()
    │             ├─ updateValueWithCopyOnWrite() [スナップショット中]
    │             │      └─ spaceAllocator.allocate()
    │             └─ updateValueWithReplace() [通常]
    │
    ├─ remove(key, namespace)
    │      └─ removeNode()
    │             ├─ 論理削除 [スナップショット中]
    │             │      └─ logicallyRemovedNodes.add()
    │             └─ doPhysicalRemove() [通常]
    │                    └─ spaceAllocator.free()
    │
    └─ stateSnapshot()
           ├─ tryToDeleteNodesPhysically()
           ├─ resourceGuard.acquireResource()
           ├─ stateMapVersion++
           └─ CopyOnWriteSkipListStateMapSnapshot
```

### データフロー図

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

キー/名前空間 ──────────▶ SkipListKeySerializer ────────▶ シリアライズ済みキー
                               │
                               ▼
                          SkipList検索
                               │
                    ┌──────────┴──────────┐
                    ▼                     ▼
               ノード存在              ノード不在
                    │                     │
                    ▼                     ▼
               値取得/更新            新規ノード作成
                    │                     │
        ┌───────────┴───────────┐         │
        ▼                       ▼         │
スナップショット中          通常時         │
        │                       │         │
        ▼                       ▼         │
Copy-on-Write              直接置換       │
   新値作成                               │
        │                       │         │
        └───────────┬───────────┘         │
                    ▼                     │
              Allocator ◀─────────────────┘
                    │
                    ▼
            MemorySegment
                    │
                    ▼
             Chunk管理
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| CopyOnWriteSkipListStateMap.java | `flink-statebackend-heap-spillable/.../heap/` | メインクラス | SkipListベースStateMap |
| CopyOnWriteSkipListStateMapSnapshot.java | 同パス | スナップショット | スナップショット実装 |
| SkipListKeySerializer.java | 同パス | シリアライザ | キーシリアライズ |
| SkipListValueSerializer.java | 同パス | シリアライザ | 値シリアライズ |
| SkipListUtils.java | 同パス | ユーティリティ | SkipList操作補助 |
| LevelIndexHeader.java | 同パス | インターフェース | レベルインデックス |
| OnHeapLevelIndexHeader.java | 同パス | 実装 | ヒープ上レベルインデックス |
| NodeStatus.java | 同パス | Enum | ノード状態（PUT/REMOVE） |
| Allocator.java | `.../heap/space/` | インターフェース | メモリ割り当て |
| Chunk.java | `.../heap/space/` | インターフェース | メモリチャンク |
| SpaceUtils.java | `.../heap/space/` | ユーティリティ | アドレス計算 |
