# バッチ設計書 23-BatchExecRank

## 概要

本ドキュメントは、Apache Flink Table APIにおけるBatchExecRankバッチ実行ノードの設計仕様を記述したものです。このノードはTop-N処理やランク関数を実行し、パーティション内でのランキングを計算します。

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

BatchExecRankは、SQL文のウィンドウ関数（ROW_NUMBER、RANK、DENSE_RANK）とTOP-N処理を効率的に実行するためのバッチ実行ノードです。入力データをパーティションごとにグループ化し、指定されたソート順でランクを計算して、指定されたランク範囲内のレコードのみを出力します。

**業務上の目的・背景**：売上上位顧客の抽出、カテゴリ別のトップ商品の取得、時系列データでの最新レコードの取得など、パーティションごとにランキングを計算し上位N件を取得する必要があるビジネスシナリオにおいて、効率的なTop-N処理を実現します。二段階（ローカル・グローバル）ランク処理をサポートすることで、分散環境でのデータシャッフルを最小化します。

**バッチの実行タイミング**：Flink SQLクエリの実行時、オプティマイザがウィンドウランク関数またはTop-Nパターンを検出した際に自動的に実行計画に組み込まれます。

**主要な処理内容**：
1. ソート済みの入力データストリームからレコードを受信
2. パーティションキーの変化を検出してグループを識別
3. オーダーキーの変化に応じてランク値を更新
4. rankStartからrankEndの範囲内のレコードを出力（オプションでランク値を付加）

**前後の処理との関連**：BatchExecSortまたはBatchExecExchangeにより事前にソートされたデータを受け取ります。InputSortedExecNodeインターフェースを実装しており、入力データがソート済みであることを前提としています。

**影響範囲**：入力データはパーティションキーとオーダーキーでソートされている必要があります。outputRankNumberフラグにより出力スキーマが変更される場合があります。

## バッチ種別

ランキング計算処理

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | クエリ実行時（随時） |
| 実行時刻 | N/A |
| 実行曜日 | N/A |
| 実行日 | N/A |
| トリガー | SQLクエリ実行（Top-N/ウィンドウランク関数） |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力データ | ソート済みのRowData形式のデータストリーム |
| ソート順 | パーティションキー、オーダーキーの順でソート |
| rankEnd | 有効な正の整数値 |

### 実行可否判定

入力データがInputSortedExecNodeの要件を満たすソート済みデータであれば実行可能です。入力がソートされていない場合、結果の正確性は保証されません。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| partitionFields | int[] | Yes | - | パーティションキーのフィールドインデックス配列 |
| sortFields | int[] | Yes | - | ソートキーのフィールドインデックス配列 |
| rankStart | long | Yes | - | 出力するランクの開始値（含む） |
| rankEnd | long | Yes | - | 出力するランクの終了値（含む） |
| outputRankNumber | boolean | Yes | - | ランク値を出力に含めるかどうか |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| inputEdge | RowData | ソート済みの入力データストリーム |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 下流ノード | RowData | ランク範囲内のレコード（オプションでランク列付き） |

### 出力ファイル仕様

N/A（メモリ内処理、下流ノードへのストリーム出力）

## 処理フロー

### 処理シーケンス

```
1. オペレータ初期化
   └─ パーティションコンパレータとオーダーコンパレータの生成
   └─ ランク出力用のJoinedRowData準備（outputRankNumberがtrueの場合）
2. 入力レコード処理
   └─ rowNumをインクリメント
   └─ パーティションキーが変化した場合：rank=1, rowNum=1にリセット
   └─ オーダーキーが変化した場合：rank=rowNumに更新
   └─ rankがrankStart〜rankEnd範囲内の場合に出力
3. 出力処理
   └─ outputRankNumberがtrueの場合：元のレコード + ランク値を結合して出力
   └─ outputRankNumberがfalseの場合：元のレコードをそのまま出力
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[コンパレータ初期化]
    B --> C[入力レコード受信]
    C --> D[rowNum++]
    D --> E{パーティションキー変化?}
    E -->|Yes| F[rank=1, rowNum=1]
    E -->|No| G{オーダーキー変化?}
    G -->|Yes| H[rank=rowNum]
    G -->|No| I{rankStart <= rank <= rankEnd?}
    F --> I
    H --> I
    I -->|Yes| J{outputRankNumber?}
    I -->|No| K[スキップ]
    J -->|Yes| L[レコード+ランク値を出力]
    J -->|No| M[レコードを出力]
    L --> N[lastInput保存]
    M --> N
    K --> N
    N --> O{次のレコード?}
    O -->|Yes| C
    O -->|No| P[バッチ終了]
```

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

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

N/A（インメモリ処理のみ）

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

N/A

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| N/A | 不正結果 | 入力データが未ソート | 事前にソート処理を挿入 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | N/A（Flinkフレームワーク管理） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

### 障害時対応

Flinkのチェックポイント/リスタート機能による自動リカバリに依存します。

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

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

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 入力件数に依存 |
| 目標処理時間 | O(n) ※n=入力件数 |
| メモリ使用量上限 | 最小限（1レコード分のバッファ） |

## 排他制御

同一ジョブ内での並列実行はFlinkにより管理されます。二段階ランク処理では、ローカルステージとグローバルステージで異なる並列度で実行される場合があります。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | オペレータopen時 | 初期化完了 |
| 進捗ログ | N/A | - |
| 終了ログ | 処理完了時 | 処理完了 |
| エラーログ | エラー発生時 | 例外詳細 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理件数 | Flink Metricsで監視 | Flink WebUI |
| 処理時間 | アプリケーション定義 | Flink Metrics |

## 備考

- RankOperatorはOneInputStreamOperatorインターフェースを実装しています
- InputSortedExecNodeインターフェースにより、入力がソート済みであることが保証されます
- コメントにより「operator needn't cache data」と明記されており、メモリ効率が高い設計です
- パーティションコンパレータとオーダーコンパレータは、フィールド変化の検出のみに使用され、実際のソートには使用されません（入力は既にソート済みのため）
- SortUtilのgetAscendingSortSpecにより、コンパレータは昇順ソート仕様で生成されますが、これはフィールド比較のみに使用されます
- 二段階ランク処理により、大規模データセットでのネットワークシャッフルを削減できます
