# バッチ設計書 33-BatchExecExchange

## 概要

本ドキュメントは、Apache Flinkのバッチ処理における入力要素のパーティショニング変更を担当する実行ノードである`BatchExecExchange`の設計仕様を定義するものです。

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

BatchExecExchangeは、バッチ処理において入力データのパーティショニング（分散配置）を変更するための実行ノードです。データのシャッフル、ブロードキャスト、フォワードなどの分散戦略を実装し、下流オペレーターが必要とするデータ分散を実現します。

**業務上の目的・背景**：分散バッチ処理において、異なるオペレーター間でデータの分散方法を変更する必要が頻繁に発生します。例えば、結合処理では結合キーに基づいてデータを再分散する必要があり、集約処理ではグループキーに基づく再分散が必要です。また、ブロードキャスト結合では小さいテーブルを全ノードにブロードキャストします。本バッチは、これらのデータ再分散要件を満たすための基盤コンポーネントです。

**バッチの実行タイミング**：クエリプランナーによって、オペレーター間でデータ分散の変更が必要と判断された場合に自動的に挿入されます。JOIN、GROUP BY、DISTINCT等の操作に先行して実行されることが一般的です。

**主要な処理内容**：
1. 入力プロパティから必要な分散戦略（RequiredDistribution）を取得
2. 分散タイプに応じた適切なStreamPartitionerの選択・生成
3. ハッシュ分散の場合はHashCodeGeneratorによるハッシュ関数コード生成
4. PartitionTransformationの構築と接続
5. 交換モード（BATCH/UNDEFINED/PIPELINE）の設定

**前後の処理との関連**：前段として任意のExecNodeからデータを受け取ります。後段としてBatchExecHashJoin、BatchExecHashAggregate、BatchExecSortMergeJoin等、特定のデータ分散を要求するオペレーターに接続されます。

**影響範囲**：データ再分散が必要なすべてのバッチクエリに影響します。特にJOIN、集約、ソート処理のパフォーマンスに直接影響を与えます。

## バッチ種別

データ交換 / パーティショニング

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | クエリ実行時（随時） |
| 実行時刻 | N/A |
| 実行曜日 | N/A |
| 実行日 | N/A |
| トリガー | クエリプラン生成 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 入力存在 | 上流ExecNodeからの入力が存在すること |
| 分散戦略定義 | InputPropertyに有効なRequiredDistributionが定義されていること |

### 実行可否判定

常に実行可能。分散タイプに応じて適切なパーティショナーが選択される。未サポートの分散タイプが指定された場合はTableExceptionがスローされる。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| inputProperty | InputProperty | Yes | - | 入力プロパティ（分散要件を含む） |
| outputType | RowType | Yes | - | 出力行の型定義 |
| requiredExchangeMode | StreamExchangeMode | No | UNDEFINED | 交換モード（BATCH/UNDEFINED/PIPELINE） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 上流ExecNode | RowData | 再分散対象の入力レコード |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 下流ExecNode | RowData | 分散戦略に従って再配置されたレコード |

### 出力ファイル仕様

N/A（メモリ内データ交換のため）

## 処理フロー

### 処理シーケンス

```
1. 分散戦略の取得
   └─ InputPropertyからRequiredDistributionを取得
2. 分散タイプの判定
   └─ ANY / BROADCAST / SINGLETON / HASH / KEEP_INPUT_AS_IS
3. パーティショナーの選択・生成
   └─ ANY: null（パーティショナーなし）
   └─ BROADCAST: BroadcastPartitioner
   └─ SINGLETON: GlobalPartitioner
   └─ HASH: BinaryHashPartitioner（ハッシュコード生成）
   └─ KEEP_INPUT_AS_IS: ForwardPartitioner / ForwardForConsecutiveHashPartitioner
4. 並列度の決定
   └─ SINGLETON: 1
   └─ その他: デフォルト or 入力並列度継承
5. 交換モードの設定
   └─ BATCH / UNDEFINED / PIPELINE
6. PartitionTransformation構築
   └─ パーティショナー、交換モード、並列度の設定
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[RequiredDistribution取得]
    B --> C{分散タイプ?}
    C -->|ANY| D[パーティショナーなし]
    C -->|BROADCAST| E[BroadcastPartitioner]
    C -->|SINGLETON| F[GlobalPartitioner]
    C -->|HASH| G[BinaryHashPartitioner]
    C -->|KEEP_INPUT_AS_IS| H{Strict?}
    H -->|Yes| I[ForwardPartitioner]
    H -->|No| J[ForwardForConsecutiveHashPartitioner]
    D --> K[並列度決定]
    E --> K
    F --> K
    G --> K
    I --> K
    J --> K
    K --> L[交換モード設定]
    L --> M[PartitionTransformation構築]
    M --> N[バッチ終了]
```

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

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

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

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

N/A

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | 未サポート分散タイプ | 未知のDistributionTypeが指定された場合 | サポートされている分散タイプを使用 |
| IllegalArgumentException | 不正な入力分散 | KEEP_INPUT_AS_ISで非HashDistributionが指定された場合 | HashDistributionを使用 |

### リトライ仕様

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

### 障害時対応

クエリプランの見直しが必要。プランナーの最適化設定を確認する。

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

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

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 入力データ量に依存 |
| 目標処理時間 | ネットワーク帯域とデータ量に依存 |
| メモリ使用量上限 | 交換モードBATCHの場合はディスクスピルを使用 |

## 排他制御

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

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| DEBUG | 変換生成時 | 分散戦略と交換モードの詳細 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| シャッフルデータ量 | アプリケーション定義 | Flinkメトリクス |
| ネットワークI/O | アプリケーション定義 | Flinkメトリクス |

## 備考

- Flink 2.0以降で利用可能
- 分散タイプ:
  - **ANY**: 任意の分散（変更なし）
  - **BROADCAST**: 全パーティションに複製
  - **SINGLETON**: 単一パーティションに集約（並列度1）
  - **HASH**: ハッシュキーに基づく分散
  - **KEEP_INPUT_AS_IS**: 入力分散を維持（ForwardまたはForwardForConsecutiveHash）
- 交換モード:
  - **BATCH**: ブロッキング交換（上流完了後に下流開始）
  - **UNDEFINED**: 設定に従う
  - **PIPELINE**: パイプライン交換（上流と下流が同時実行）
