# バッチ設計書 20-BatchExecSort

## 概要

本ドキュメントは、Apache Flink Table Plannerにおけるソート処理のバッチ実行ノードの設計を記述したものである。

### 本バッチの処理概要

BatchExecSortは、入力データを指定されたソートキーに基づいて完全にソートするバッチ実行ノードである。制限（LIMIT）なしで全データを出力する。ORDER BY句に対応する。

**業務上の目的・背景**：ソート処理は、データ分析において最も基本的かつ重要な操作の一つである。結果セットを特定の順序で表示したい場合、後続の処理（マージ結合、ソート集約など）がソート済みデータを必要とする場合、ランキングやTop-N処理の前処理として使用される。本ノードは、大規模データセットでもメモリに収まらない場合は外部ソート（ディスクスピル）を行い、確実にソート結果を出力する。

**バッチの実行タイミング**：Flink SQLクエリの実行時、ORDER BY句を含むクエリにおいて呼び出される。LIMITがない場合に使用され、LIMITがある場合はBatchExecSortLimitが使用される。

**主要な処理内容**：
1. 入力ストリームからデータを受信
2. SortSpecに基づいてソートキーと順序を決定
3. インメモリソートを実行
4. メモリ不足の場合はディスクにスピル
5. 複数のスピルファイルをマージ
6. ソート済みデータを出力

**前後の処理との関連**：入力としてExchange（データパーティショニング）やCalc（フィルタ・射影）ノードを受け取る。出力は、後続のソート集約（BatchExecSortAggregate）やユーザーへの結果表示に使用される。

**影響範囲**：Flink Table APIおよびSQL APIを使用するすべてのバッチジョブにおいて、ORDER BY句を使用するクエリに影響する。

## バッチ種別

ソート処理

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | 随時（クエリ実行時） |
| 実行時刻 | N/A |
| 実行曜日 | N/A |
| 実行日 | N/A |
| トリガー | Flink SQLクエリの実行 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力データストリーム | 入力ストリームが存在すること |
| ソート仕様 | SortSpecでソートキーと順序が定義されていること |
| メモリ/ディスク | ソート処理に必要なリソースが確保されていること |

### 実行可否判定

Flink PlannerがORDER BY句を検出し、LIMITがない場合に選択される。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| sortSpec | SortSpec | Yes | N/A | ソート仕様（ソートキー、昇順/降順、NULL順序） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 入力ストリーム | RowData | ソート対象のデータストリーム |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| ソート済みストリーム | RowData | 指定された順序でソートされたデータストリーム |

### 出力ファイル仕様

ファイル出力なし（ストリーム出力のみ、スピルファイルは一時的）

## 処理フロー

### 処理シーケンス

```
1. 入力変換の取得
   └─ 入力ストリームをTransformationとして取得
2. SortCodeGeneratorの構築
   └─ 入力型とSortSpecを基にコードジェネレーターを構築
3. ソート演算子の生成
   └─ NormalizedKeyComputerの生成
   └─ RecordComparatorの生成
   └─ SortOperatorの構築
4. 設定の適用
   └─ TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES: 最大ファイルハンドル数
   └─ TABLE_EXEC_SPILL_COMPRESSION_ENABLED: スピル圧縮の有効化
   └─ TABLE_EXEC_SPILL_COMPRESSION_BLOCK_SIZE: 圧縮ブロックサイズ
   └─ TABLE_EXEC_SORT_ASYNC_MERGE_ENABLED: 非同期マージの有効化
5. メモリ設定の取得
   └─ TABLE_EXEC_RESOURCE_SORT_MEMORYで管理メモリを設定
6. Transformationの作成
   └─ オペレーターとメモリ設定を適用
7. ソート処理の実行
8. 結果の出力
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[入力ストリーム取得]
    B --> C[SortCodeGenerator構築]
    C --> D[NormalizedKeyComputer生成]
    D --> E[RecordComparator生成]
    E --> F[SortOperator構築]
    F --> G[設定適用]
    G --> H[Transformation作成]
    H --> I[インメモリソート開始]
    I --> J{メモリ不足?}
    J -->|No| K[ソート完了]
    J -->|Yes| L[ディスクにスピル]
    L --> M[スピルファイルマージ]
    M --> K
    K --> N[ソート済みデータ出力]
    N --> O[バッチ終了]
```

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

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

本バッチノードはインメモリ処理であり、直接的なデータベース操作は行わない。

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| N/A | N/A | N/A | データベース操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| OutOfMemoryError | メモリ不足 | 管理メモリ不足かつスピル失敗 | TABLE_EXEC_RESOURCE_SORT_MEMORYを増加 |
| IOException | スピルエラー | ディスクへのスピル失敗 | ディスク容量を確認、TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLESを調整 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | Flinkジョブ設定に依存 |
| リトライ間隔 | Flinkジョブ設定に依存 |
| リトライ対象エラー | 一時的なリソース不足 |

### 障害時対応

ジョブ失敗時は、Flink Checkpointから再開するか、ジョブを最初から再実行する。一時スピルファイルは自動的にクリーンアップされる。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | バッチ全体（Flinkジョブ単位） |
| コミットタイミング | ジョブ完了時 |
| ロールバック条件 | ジョブ失敗時 |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | データサイズに依存 |
| 目標処理時間 | O(N log N)のソート複雑度 |
| メモリ使用量上限 | TABLE_EXEC_RESOURCE_SORT_MEMORY設定値 |

## 排他制御

同一ジョブ内で複数のタスクが並列実行される。各タスクは独立したソート処理を行い、明示的な排他制御は不要。グローバルソートが必要な場合は、並列度1で実行される。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | Transformation作成時 | ノード情報、ソート仕様 |
| 進捗ログ | 処理中 | 処理件数、スピル回数 |
| 終了ログ | 処理完了時 | 処理完了ステータス |
| エラーログ | エラー発生時 | エラー詳細、スタックトレース |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理時間 | ジョブ設定に依存 | Flink監視システム |
| メモリ使用量 | 管理メモリ上限 | Flink監視システム |
| スピル回数 | 多発時 | Flink監視システム |

## 備考

- ExecNodeMetadataにより、batch-exec-sort（version 1）として登録
- consumedOptionsとして以下の設定を使用：
  - TABLE_EXEC_SORT_MAX_NUM_FILE_HANDLES: スピル時の最大ファイルハンドル数
  - TABLE_EXEC_SORT_ASYNC_MERGE_ENABLED: 非同期マージの有効化
  - TABLE_EXEC_SPILL_COMPRESSION_ENABLED: スピル圧縮の有効化
  - TABLE_EXEC_SPILL_COMPRESSION_BLOCK_SIZE: 圧縮ブロックサイズ
  - TABLE_EXEC_RESOURCE_SORT_MEMORY: ソート用管理メモリ
- SortOperatorを使用し、外部ソート（External Sort）をサポート
- NormalizedKeyComputerで効率的なキー比較を実現
- Flink 2.0以降で使用可能（minPlanVersion、minStateVersion = v2_0）
