# バッチ設計書 14-BatchExecOverAggregate

## 概要

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

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

BatchExecOverAggregateは、SQL OVER句によるウィンドウ集約を実行するバッチ実行ノードである。ROW型およびRANGE型のウィンドウフレームをサポートし、LEAD、LAG、SUM OVER、COUNT OVERなどの分析関数を処理する。

**業務上の目的・背景**：オーバーウィンドウ集約は、行単位でウィンドウを定義し、その範囲内でデータを集約する処理である。移動平均、累積合計、前後の行との比較（LEAD/LAG）など、データ分析において頻繁に使用される。GROUP BY集約と異なり、各行に対して集約結果を付加するため、元のデータ粒度を維持したまま分析が可能である。

**バッチの実行タイミング**：Flink SQLクエリの実行時、OVER句を含むクエリにおいて呼び出される。SUM(x) OVER (PARTITION BY a ORDER BY b ROWS BETWEEN ...)のような構文が使用された場合に実行される。

**主要な処理内容**：
1. パーティション（PARTITION BY）でソートされた入力ストリームを受信
2. パーティション内でウィンドウフレーム（ROWS/RANGE）に基づいて処理
3. ウィンドウタイプに応じた処理モード選択：
   - INSENSITIVE: 境界に依存しない集約
   - ROW: 行ベースのウィンドウフレーム
   - RANGE: 値ベースのウィンドウフレーム
   - OFFSET: LEAD/LAG関数
4. バッファリングが必要な場合はBufferDataOverWindowOperatorを使用
5. バッファリング不要の場合はNonBufferOverWindowOperatorを使用
6. 各行に対して集約結果を計算し出力

**前後の処理との関連**：入力としてSort（ソート）やExchange（データパーティショニング）ノードを受け取る。BatchExecOverAggregateBaseを継承し、共通のウィンドウ処理ロジックを再利用する。

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

## バッチ種別

集計処理（ウィンドウ集約）

## 実行スケジュール

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

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力データストリーム | パーティションキーとオーダーキーでソート済みの入力ストリーム |
| OverSpec | ウィンドウ定義（パーティション、ソート、グループ、境界）が設定されていること |

### 実行可否判定

Flink Plannerが、クエリにOVER句が含まれ、Java/Scala UDFが使用されている場合に選択する。Python UDFの場合はBatchExecPythonOverAggregateが使用される。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| overSpec | OverSpec | Yes | N/A | OVER句の仕様（パーティション、ソート、グループ、集約関数） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| ソート済み入力ストリーム | RowData | パーティション・オーダーでソートされたデータ |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 出力ストリーム | RowData | 入力データにウィンドウ集約結果を追加したレコード |

### 出力ファイル仕様

ファイル出力なし（ストリーム出力のみ）

## 処理フロー

### 処理シーケンス

```
1. 入力変換の取得
   └─ ソート済み入力ストリームをTransformationとして取得
2. パーティションコンパレータ生成
   └─ パーティションキーでの比較器を生成
3. バッファリング要否判定
   └─ needBufferData()でウィンドウタイプに応じた判定
4. オペレーター選択と生成
   └─ バッファリング不要: NonBufferOverWindowOperator
   └─ バッファリング要: BufferDataOverWindowOperator + OverWindowFrame[]
5. OverWindowFrameの生成（バッファリング要の場合）
   └─ ウィンドウモードに応じたフレームを生成
   └─ OFFSET: OffsetOverFrame（LEAD/LAG）
   └─ ROW/RANGE: 各種OverFrame
6. Transformationの作成
   └─ オペレーターを使用してTransformationを作成
7. 結果の出力
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[ソート済み入力取得]
    B --> C[パーティションコンパレータ生成]
    C --> D{バッファリング要否}
    D -->|不要| E[NonBufferOverWindowOperator]
    D -->|要| F[OverWindowFrame生成]
    F --> G{ウィンドウモード}
    G -->|OFFSET| H[OffsetOverFrame]
    G -->|ROW| I[RowXxxOverFrame]
    G -->|RANGE| J[RangeXxxOverFrame]
    G -->|INSENSITIVE| K[InsensitiveOverFrame]
    H --> L[BufferDataOverWindowOperator]
    I --> L
    J --> L
    K --> L
    E --> M[Transformation作成]
    L --> M
    M --> N[ウィンドウ集約実行]
    N --> O[結果出力]
    O --> P[バッチ終了]
```

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

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

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | 未サポートウィンドウ | サポートされていないウィンドウ定義 | OVER句の構文を確認 |
| RuntimeException | オフセット型エラー | LEAD/LAGのオフセットが不正な型 | オフセットはLONG/INT/SHORTのみ |
| OutOfMemoryError | メモリ不足 | バッファデータがメモリ超過 | TABLE_EXEC_RESOURCE_EXTERNAL_BUFFER_MEMORYを増加 |

### リトライ仕様

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

### 障害時対応

ジョブ失敗時は、Flink Checkpointから再開するか、ジョブを最初から再実行する。

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

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

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | データサイズに依存 |
| 目標処理時間 | クエリ複雑度とデータサイズに依存 |
| メモリ使用量上限 | TABLE_EXEC_RESOURCE_EXTERNAL_BUFFER_MEMORY（バッファリング要の場合） |

## 排他制御

同一ジョブ内で複数のタスクが並列実行される。各タスクは独立して処理を行い、明示的な排他制御は不要。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | Transformation作成時 | ノード情報、ウィンドウ定義 |
| 進捗ログ | 処理中 | 処理件数、パーティション数 |
| 終了ログ | 処理完了時 | 処理完了ステータス |
| エラーログ | エラー発生時 | エラー詳細、スタックトレース |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理時間 | ジョブ設定に依存 | Flink監視システム |
| バッファサイズ | 管理メモリ上限 | Flink監視システム |

## 備考

- ExecNodeMetadataにより、batch-exec-over-aggregate（version 1）として登録
- consumedOptionsとしてexternal-buffer-memory設定を使用
- BatchExecOverAggregateBaseを継承し、ウィンドウモード判定などの共通ロジックを再利用
- 多様なOverWindowFrameをサポート：
  - UnboundedOverWindowFrame: 無制限ウィンドウ
  - RowUnboundedPrecedingOverFrame: ROW型の前方無制限
  - RowUnboundedFollowingOverFrame: ROW型の後方無制限
  - RowSlidingOverFrame: ROW型のスライディング
  - RangeUnboundedPrecedingOverFrame: RANGE型の前方無制限
  - RangeUnboundedFollowingOverFrame: RANGE型の後方無制限
  - RangeSlidingOverFrame: RANGE型のスライディング
  - InsensitiveOverFrame: 境界非依存
  - OffsetOverFrame: LEAD/LAG用
