# バッチ設計書 29-BatchExecCorrelate

## 概要

本ドキュメントは、Apache Flink Table APIにおけるBatchExecCorrelateバッチ実行ノードの設計仕様を記述したものです。このノードはJava/Scalaユーザー定義テーブル関数（UDTF）との結合処理を実行します。

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

BatchExecCorrelateは、SQL文のLATERAL TABLE句を実行するためのバッチ実行ノードです。入力テーブルの各行に対してユーザー定義テーブル関数（UDTF）を呼び出し、その結果と入力行を結合します。INNER JOINまたはLEFT OUTER JOINセマンティクスをサポートします。

**業務上の目的・背景**：複雑なデータ変換、配列やマップの展開、外部データソースとの動的な結合など、1行の入力から複数行を生成する必要があるビジネスシナリオで使用されます。例えば、カンマ区切り文字列の分割、JSON配列の展開、時系列データの補間などの処理に適しています。

**バッチの実行タイミング**：Flink SQLクエリの実行時、オプティマイザがLATERAL TABLE句またはテーブル関数呼び出しを検出した際に自動的に実行計画に組み込まれます。

**主要な処理内容**：
1. 入力データストリームからレコードを受信
2. 各レコードに対してUDTFを呼び出し
3. UDTFの結果（0行以上）と入力レコードを結合
4. 結合結果を下流に出力

**前後の処理との関連**：任意のデータソースや変換処理からデータを受け取り、UDTF結合後のデータを後続の処理（Filter、Join、Aggregate、Sink等）に渡します。

**影響範囲**：出力レコード数はUDTFの出力に依存し、入力レコード数より多くなることも少なくなることもあります。

## バッチ種別

テーブル関数結合処理

## 実行スケジュール

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

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力データ | 有効なRowData形式のデータストリーム |
| UDTF登録 | テーブル関数がカタログに登録済み |
| 入力プロパティ | 単一の入力ストリーム |

### 実行可否判定

入力プロパティが1つ、UDTFの呼び出し式が有効であれば実行可能です。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| joinType | FlinkJoinType | Yes | - | 結合タイプ（INNER/LEFT） |
| invocation | RexCall | Yes | - | UDTF呼び出し式 |
| condition | RexNode | No | null | 結合条件（WHERE句） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| inputEdge | RowData | 上流の実行ノードからの入力データストリーム |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 下流ノード | RowData | 入力レコード + UDTF出力 |

### 出力ファイル仕様

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

## 処理フロー

### 処理シーケンス

```
1. オペレータ生成
   └─ CorrelateCodeGeneratorでCorrelateオペレータのコードを動的生成
   └─ UDTF呼び出しコードを生成
   └─ 結合条件の評価コードを生成（conditionがある場合）
   └─ コンパイルしてTransformationを作成
2. 入力レコード処理
   └─ UDTF呼び出し
   └─ 各UDTF出力に対して入力レコードと結合
   └─ 結合条件の評価（conditionがある場合）
3. 出力処理
   └─ INNER JOIN: UDTFが出力を生成した場合のみ出力
   └─ LEFT JOIN: UDTFが出力を生成しない場合でも入力を出力（NULL埋め）
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[CorrelateCodeGenerator実行]
    B --> C[オペレータ生成・コンパイル]
    C --> D[入力レコード受信]
    D --> E[UDTF呼び出し]
    E --> F{UDTF出力あり?}
    F -->|Yes| G[入力 + UDTF出力を結合]
    G --> H{condition指定?}
    H -->|Yes| I[条件評価]
    H -->|No| J[下流に送信]
    I --> K{条件=true?}
    K -->|Yes| J
    K -->|No| L[次のUDTF出力へ]
    F -->|No| M{LEFT JOIN?}
    M -->|Yes| N[入力 + NULL出力]
    M -->|No| O[スキップ]
    N --> J
    L --> P{次のUDTF出力?}
    P -->|Yes| G
    P -->|No| Q{次の入力レコード?}
    O --> Q
    J --> P
    Q -->|Yes| D
    Q -->|No| R[バッチ終了]
```

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

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

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

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

N/A

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| RuntimeException | UDTF実行エラー | UDTF内で例外発生 | UDTF実装確認 |
| CodeGenerationException | コード生成エラー | 不正なUDTF呼び出し式 | クエリの修正 |

### リトライ仕様

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

### 障害時対応

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

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

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

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 入力件数 × UDTFの平均出力件数 |
| 目標処理時間 | UDTFの複雑度に依存 |
| メモリ使用量上限 | UDTFの出力サイズに依存 |

## 排他制御

同一ジョブ内での並列実行はFlinkにより管理されます。ステートレスな処理であるため、並列度に関わらず独立して処理可能です。

## ログ出力

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

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 出力件数 | Flink Metricsで監視 | Flink WebUI |
| UDTF呼び出し時間 | アプリケーション定義 | Flink Metrics |

## 備考

- BatchExecCorrelateはCommonExecCorrelateを継承しており、ストリーム処理版のStreamExecCorrelateと共通の基底クラスを持ちます
- retainHeader=falseで動作し、レコードヘッダ情報は保持されません
- TableStreamOperatorを基底クラスとして使用します
- FlinkJoinTypeでINNER JOIN（デフォルト）またはLEFT OUTER JOINを指定可能
- conditionパラメータはオプションであり、UDTF出力に対する追加フィルタ条件を指定できます
- CorrelateCodeGeneratorによりUDTF呼び出しと結合処理のコードが動的生成されます
- CORRELATE_TRANSFORMATIONという名前でTransformationが生成されます
- Python UDTFの場合はBatchExecPythonCorrelateが使用されます
