# バッチ設計書 36-BatchExecMultipleInput

## 概要

本ドキュメントは、Apache Flinkのバッチ処理における複数入力ストリームオペレーターを統合管理する実行ノードである`BatchExecMultipleInput`の設計仕様を定義するものです。

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

BatchExecMultipleInputは、複数のExecNodeから構成されるサブグラフを単一のオペレーターとして実行するための実行ノードです。複数の入力ストリームを受け取り、内部に含まれる複数のオペレーターを効率的に連携実行することで、オペレーター間のシリアライゼーション/デシリアライゼーションオーバーヘッドを削減し、処理性能を向上させます。

**業務上の目的・背景**：複雑なバッチクエリでは、複数のオペレーター（結合、集約、計算等）が連続して実行されることが一般的です。各オペレーター間でデータをシリアライズ/デシリアライズすることはオーバーヘッドとなり、特に大規模データ処理では顕著な性能劣化を引き起こします。本バッチは、関連するオペレーターを単一のMultipleInputオペレーターに統合することで、これらのオーバーヘッドを削減し、処理効率を大幅に向上させます。

**バッチの実行タイミング**：クエリプランナーのMultipleInput最適化フェーズで、連続するオペレーターが融合可能と判断された場合に自動的に生成されます。

**主要な処理内容**：
1. 複数の入力変換（Transformation）の収集と管理
2. 融合コード生成が有効な場合：オペレーター融合コード生成
   - BatchExecInputAdapterの挿入
   - SourceOpFusionCodegenSpecGeneratorの生成
   - OperatorFusionCodegenFactoryの生成
3. 融合コード生成が無効な場合：TableOperatorWrapperGeneratorによるラッパー生成
   - InputSpecの生成
   - BatchMultipleInputStreamOperatorFactoryの構築
4. MultipleInputTransformationの構築
5. リソース（メモリ、並列度）の設定
6. チェイニング戦略（HEAD_WITH_SOURCES）の設定

**前後の処理との関連**：前段として複数のBatchExecExchange等から入力を受け取ります。内部にはBatchExecHashJoin、BatchExecHashAggregate、BatchExecCalc等のメンバーノードを含みます。後段としてBatchExecSink、BatchExecExchange等に接続されます。

**影響範囲**：複数オペレーターの連続実行を含むすべてのバッチクエリに影響します。特に複雑なJOINやサブクエリを含むクエリの性能に大きく影響します。

## バッチ種別

複数入力 / オペレーター融合

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | クエリ実行時（随時） |
| 実行時刻 | N/A |
| 実行曜日 | N/A |
| 実行日 | N/A |
| トリガー | MultipleInput最適化 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力数一致 | inputPropertiesとoriginalEdgesのサイズが一致すること |
| rootNode存在 | 内部サブグラフのルートノードが存在すること |
| メンバーノード存在 | 融合対象のメンバーノードが存在すること |

### 実行可否判定

入力プロパティとオリジナルエッジの数が一致している場合のみ実行可能。不一致の場合はIllegalArgumentExceptionがスローされる。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| inputProperties | List<InputProperty> | Yes | - | 入力プロパティのリスト |
| rootNode | ExecNode<?> | Yes | - | サブグラフのルートノード |
| memberExecNodes | List<ExecNode<?>> | Yes | - | サブグラフのメンバーノード |
| originalEdges | List<ExecEdge> | Yes | - | 入力エッジのリスト |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 複数の上流ExecNode | RowData | 各入力ストリームからのレコード |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 下流ExecNode | RowData | rootNodeの出力型に対応するレコード |

### 出力ファイル仕様

N/A（メモリ内データ処理のため）

## 処理フロー

### 処理シーケンス

```
1. 入力変換の収集
   └─ 各入力エッジから変換を取得
2. 読み取り順序の決定
   └─ InputPropertyのpriorityから順序を取得
3. 実行モード判定
   └─ 融合コード生成有効 && 全メンバーが融合サポート?
4A. 融合コード生成モード
   └─ BatchExecInputAdapterの挿入
   └─ InputSelectionSpecの生成
   └─ SourceOpFusionCodegenSpecGeneratorの生成
   └─ OutputFusionCodegenSpecのラップ
   └─ OperatorFusionCodegenFactoryの生成
4B. 通常モード
   └─ TableOperatorWrapperGeneratorの実行
   └─ BatchMultipleInputStreamOperatorFactoryの生成
5. MultipleInputTransformation構築
   └─ 並列度、最大並列度の設定
   └─ リソース（minResources、preferredResources）の設定
   └─ マネージドメモリの設定
6. チェイニング戦略設定
   └─ ChainingStrategy.HEAD_WITH_SOURCES
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[入力変換収集]
    B --> C[読み取り順序決定]
    C --> D{融合コード生成有効?}
    D -->|Yes| E{全メンバー融合サポート?}
    D -->|No| F[TableOperatorWrapperGenerator]
    E -->|Yes| G[BatchExecInputAdapter挿入]
    E -->|No| F
    G --> H[InputSelectionSpec生成]
    H --> I[FusionCodegenSpecGenerator生成]
    I --> J[OperatorFusionCodegenFactory生成]
    F --> K[BatchMultipleInputStreamOperatorFactory]
    J --> L[MultipleInputTransformation構築]
    K --> L
    L --> M[リソース設定]
    M --> N[チェイニング戦略設定]
    N --> O[バッチ終了]
```

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

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

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

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

N/A

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalArgumentException | 入力不整合 | inputProperties.size() != originalEdges.size() | クエリプランの確認 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | N/A（設計エラーのため） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

### 障害時対応

クエリプランの確認とMultipleInput最適化設定の見直しが必要。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | N/A（オペレーター融合） |
| コミットタイミング | N/A |
| ロールバック条件 | N/A |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 入力データ量とオペレーター数に依存 |
| 目標処理時間 | 個別実行比で大幅改善が期待される |
| メモリ使用量上限 | マネージドメモリとして設定（メンバーオペレーターの合計） |

## 排他制御

並列実行可能。各パーティションは独立して処理される。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| DEBUG | 変換生成時 | 融合モード、メンバーノード情報 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理時間 | アプリケーション定義 | Flinkメトリクス |
| メモリ使用量 | アプリケーション定義 | Flinkメトリクス |

## 備考

- @ExecNodeMetadataアノテーションが付与されていない（永続化対象外）
- TABLE_EXEC_OPERATOR_FUSION_CODEGEN_ENABLEDで融合コード生成を制御
- allSupportFusionCodegenで全メンバーノードの融合サポートを確認
- ChainingStrategy.HEAD_WITH_SOURCESでソースとのチェイニングを有効化
- マネージドメモリはメガバイト単位で設定（memoryWeight << 20）
- getRootNode、getMemberExecNodes、getOriginalEdgesで内部構造にアクセス可能（テスト用）
- 複数の入力を持つため、InputSelectionSpecで読み取り順序を制御
