# バッチ設計書 19-BatchExecPythonOverAggregate

## 概要

本ドキュメントは、Apache Flink Table Plannerにおける、Pythonユーザー定義集約関数（UDAF）を使用したオーバーウィンドウ集約処理のバッチ実行ノードの設計を記述したものである。

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

BatchExecPythonOverAggregateは、Pythonで実装されたユーザー定義集約関数（UDAF）を使用してOVER句によるウィンドウ集約を実行するバッチ実行ノードである。ROW型およびRANGE型のウィンドウフレームをサポートする。

**業務上の目的・背景**：OVER句によるウィンドウ集約とPython UDAFの組み合わせにより、移動平均、累積合計、ランキングなどの分析処理にPythonの豊富なライブラリを活用できる。特にPandasのrolling関数やexpanding関数と同様の処理をFlink SQLで表現可能になる。

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

**主要な処理内容**：
1. ソート済み入力ストリームからデータを受信
2. OverSpecからウィンドウグループ情報を抽出
3. 各グループのウィンドウモード（ROW/RANGE）と境界を計算
4. Apache Arrowフォーマットでデータをバッチ化
5. Pythonワーカープロセスにデータを転送
6. Python側でウィンドウごとにUDAFを実行
7. 結果をArrowフォーマットで受信
8. 入力データに集約結果を追加して出力

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

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

## バッチ種別

集計処理（Pythonオーバーウィンドウ集約）

## 実行スケジュール

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

## 実行条件

### 前提条件

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

### 実行可否判定

Flink PlannerがPython UDAFを含むOVER句クエリを検出した場合に自動的に選択される。

## 入力仕様

### 入力パラメータ

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

### 入力データソース

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

## 出力仕様

### 出力データ

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

### 出力ファイル仕様

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

## 処理フロー

### 処理シーケンス

```
1. 入力変換の取得
   └─ ソート済み入力ストリームをTransformationとして取得
2. グループ情報の抽出
   └─ OverSpecからグループリストを取得
3. 各グループの処理
   └─ ウィンドウモード（ROW/RANGE）の判定
   └─ 境界値（lowerBoundary, upperBoundary）の計算
   └─ 集約関数の収集
4. Python設定の抽出
   └─ CommonPythonUtil.extractPythonConfiguration()でPython設定を取得
5. Python関数情報の抽出
   └─ extractPythonAggregateFunctionInfosFromAggregateCall()で関数情報を抽出
6. 投影コード生成
   └─ UdafInputProjection, GroupKey, GroupSetの投影を生成
7. Pythonオペレーターの構築
   └─ BatchArrowPythonOverWindowAggregateFunctionOperatorを構築
8. 管理メモリの宣言
9. Transformationの作成
10. 結果の出力
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[ソート済み入力取得]
    B --> C[グループ情報抽出]
    C --> D[各グループのウィンドウモード判定]
    D --> E{ROW or RANGE}
    E -->|ROW| F[行ベースの境界計算]
    E -->|RANGE| G[値ベースの境界計算]
    F --> H[境界リストに追加]
    G --> H
    H --> I[Python設定抽出]
    I --> J[Python関数情報抽出]
    J --> K[投影コード生成]
    K --> L[Pythonオペレーター構築]
    L --> M{管理メモリ使用?}
    M -->|Yes| N[ManagedMemoryUseCase.PYTHON宣言]
    M -->|No| O[Transformation作成]
    N --> O
    O --> P[Arrow形式でデータ転送]
    P --> Q[Python UDAF実行]
    Q --> R[結果受信]
    R --> S[結果出力]
    S --> T[バッチ終了]
```

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

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

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

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | Pythonオペレーター構築失敗 | リフレクションエラー | Python環境とPyFlinkのバージョンを確認 |
| TableException | 未サポートウィンドウグループ | サポートされていないROW/RANGEウィンドウ | ウィンドウ定義を確認 |
| PythonException | Python実行時エラー | Python UDAF内でのエラー | UDAFのコードをデバッグ |

### リトライ仕様

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

### 障害時対応

ジョブ失敗時は、Flink Checkpointから再開するか、ジョブを最初から再実行する。Pythonワーカーのエラーログも確認する。

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

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

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | データサイズに依存 |
| 目標処理時間 | クエリ複雑度とデータサイズに依存（Pythonオーバーヘッドあり） |
| メモリ使用量上限 | Pythonワーカーの管理メモリ設定に依存 |

## 排他制御

同一ジョブ内で複数のタスクが並列実行される。各タスクは独立したPythonワーカープロセスを持ち、明示的な排他制御は不要。

## ログ出力

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

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理時間 | ジョブ設定に依存 | Flink監視システム |
| Pythonワーカー状態 | プロセス異常 | Flink監視システム |

## 備考

- BatchExecOverAggregateBaseを継承し、ウィンドウモード判定などの共通ロジックを再利用
- BatchArrowPythonOverWindowAggregateFunctionOperatorを使用（リフレクションで動的ロード）
- 境界値の特殊値：
  - Long.MIN_VALUE: UNBOUNDED PRECEDING
  - Long.MAX_VALUE: UNBOUNDED FOLLOWING
- ウィンドウモードの判定：isRangeWindows配列で管理
- aggWindowIndex: 各集約関数が属するウィンドウグループのインデックス
- ソートキーの昇順/降順を考慮（sortSpec.getAscendingOrders()[0]）
- 3種類の投影を生成：UdafInputProjection、GroupKey、GroupSet
