# バッチ設計書 24-BatchExecCalc

## 概要

本ドキュメントは、Apache Flink Table APIにおけるBatchExecCalcバッチ実行ノードの設計仕様を記述したものです。このノードは射影（SELECT）とフィルタリング（WHERE）を組み合わせた計算処理を実行します。

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

BatchExecCalcは、SQL文のSELECT句（射影）とWHERE句（フィルタリング）を実行するためのバッチ実行ノードです。入力レコードに対して列の選択、式の評価、条件によるフィルタリングを行います。コード生成技術を使用して、各クエリに最適化されたオペレータを動的に生成します。

**業務上の目的・背景**：データ変換パイプラインにおいて、必要な列の抽出、計算式の適用、条件によるデータフィルタリングは最も基本的かつ頻繁に使用される処理です。BatchExecCalcはこれらの処理を効率的に実行し、不要なデータの伝搬を防ぎ、後続処理の負荷を軽減します。

**バッチの実行タイミング**：Flink SQLクエリの実行時、オプティマイザがProjection（射影）やFilter（フィルタ）を含む論理計画を物理計画に変換する際に自動的に実行計画に組み込まれます。

**主要な処理内容**：
1. 入力データストリームからレコードを受信
2. フィルタ条件が指定されている場合、条件を評価してマッチしないレコードを除外
3. 射影リストに基づいて出力レコードを構築（列選択、式評価）
4. 変換後のレコードを下流に出力

**前後の処理との関連**：データソース（TableSourceScan等）から読み込まれたデータを受け取り、Join、Aggregate、Sinkなどの後続処理にデータを渡します。複数のCalcノードが連続する場合、オプティマイザにより1つのノードにマージされることがあります。

**影響範囲**：入力スキーマと出力スキーマが異なる場合があり、後続の処理は出力スキーマに合わせて設計される必要があります。

## バッチ種別

データ変換処理（射影・フィルタリング）

## 実行スケジュール

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

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力データ | 有効なRowData形式のデータストリーム |
| 射影リスト | 1つ以上のRexNode式 |
| 入力プロパティ | 単一の入力ストリーム |

### 実行可否判定

入力プロパティが1つ、射影リストが有効であれば実行可能です。条件（WHERE句）はオプションであり、nullの場合はフィルタリングなしで射影のみを実行します。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| projection | List<RexNode> | Yes | - | 射影式のリスト |
| condition | RexNode | No | null | フィルタ条件（WHERE句） |

### 入力データソース

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

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 下流ノード | RowData | 変換後のレコード（新しいスキーマ） |

### 出力ファイル仕様

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

## 処理フロー

### 処理シーケンス

```
1. オペレータ生成
   └─ CalcCodeGeneratorでCalcオペレータのコードを動的生成
   └─ コンパイルしてCodeGenOperatorFactoryを作成
2. 入力レコード処理
   └─ 条件評価（conditionがnullでない場合）
   └─ 条件がfalseの場合はスキップ
   └─ 射影式を評価して出力レコードを構築
3. 出力処理
   └─ 変換後のレコードを下流に送信
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[CalcCodeGenerator実行]
    B --> C[オペレータ生成・コンパイル]
    C --> D[入力レコード受信]
    D --> E{condition指定?}
    E -->|Yes| F[条件評価]
    E -->|No| G[射影式評価]
    F --> H{条件=true?}
    H -->|Yes| G
    H -->|No| I[レコードスキップ]
    G --> J[出力レコード構築]
    J --> K[下流に送信]
    I --> L{次のレコード?}
    K --> L
    L -->|Yes| D
    L -->|No| M[バッチ終了]
```

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

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

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

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

N/A

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| RuntimeException | 式評価エラー | 式の評価中に例外発生 | 入力データと式の整合性確認 |
| CodeGenerationException | コード生成エラー | 不正な式によるコード生成失敗 | クエリの修正 |

### リトライ仕様

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

### 障害時対応

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

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

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

## パフォーマンス要件

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

## 排他制御

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

## ログ出力

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

## 監視・アラート

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

## 備考

- BatchExecCalcはCommonExecCalcを継承しており、ストリーム処理版のStreamExecCalcと共通の基底クラスを持ちます
- retainHeader=falseで動作し、レコードヘッダ情報は保持されません
- supportFusionCodegen()がtrueを返すため、オペレータフュージョン最適化に対応しています
- CalcFusionCodegenSpecにより、複数のCalcオペレータを1つの融合オペレータにマージできます
- TableStreamOperatorを基底クラスとして使用し、テーブルストリーム処理の標準的なライフサイクルに従います
- Apache CalciteのRexNode（行式）を使用して射影とフィルタを表現しています
