# バッチ設計書 37-BatchExecDynamicFilteringDataCollector

## 概要

本ドキュメントは、Apache Flinkのバッチ処理におけるダイナミックフィルタリングデータ収集を担当する実行ノードである`BatchExecDynamicFilteringDataCollector`の設計仕様を定義するものです。

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

BatchExecDynamicFilteringDataCollectorは、ダイナミックフィルタリングに使用するフィルタリングデータを収集し、DynamicFilteringDataを構築してソースコーディネーターにDynamicFilteringEventを送信するための実行ノードです。結合処理のビルド側から抽出したキー値を収集し、プローブ側のソーススキャンで使用するフィルタを生成します。

**業務上の目的・背景**：大規模なバッチJOIN処理において、プローブ側（通常はファクトテーブル）のスキャン効率は全体の処理性能に大きく影響します。ダイナミックフィルタリングは、ビルド側（通常はディメンションテーブル）から結合キーの値を収集し、これをブルームフィルター等の形式でプローブ側のソースにプッシュダウンすることで、不要なデータの読み込みを削減します。本バッチは、このフィルタリングデータの収集と配信を担当します。

**バッチの実行タイミング**：ダイナミックフィルタリングが有効なJOINクエリで、ビルド側の処理後に実行されます。クエリプランナーが自動的に挿入します。

**主要な処理内容**：
1. 入力データからダイナミックフィルタリング用のフィールド（結合キー）を抽出
2. 抽出したデータを収集・重複排除
3. 収集データサイズの閾値チェック（デフォルト8MB）
4. 閾値以下の場合：全データを含むDynamicFilteringEventを生成
5. 閾値超過の場合：フラグのみのDynamicFilteringEventを生成（フィルタリング無効化）
6. ソースコーディネーターへのイベント送信

**前後の処理との関連**：前段としてJOINのビルド側からBatchExecExchange等経由でデータを受け取ります。後段としてBatchExecExecutionOrderEnforcerに接続され、フィルタリング完了後にソーススキャンが開始されます。

**影響範囲**：ダイナミックフィルタリングが有効なすべてのバッチJOINクエリに影響します。特にスタースキーマやスノーフレークスキーマにおけるファクト・ディメンション結合の性能に大きく影響します。

## バッチ種別

動的フィルタリング / データ収集

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | クエリ実行時（随時） |
| 実行時刻 | N/A |
| 実行曜日 | N/A |
| 実行日 | N/A |
| トリガー | ダイナミックフィルタリング有効時 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| ダイナミックフィルタリング有効 | クエリでダイナミックフィルタリングが適用されていること |
| 出力フィールド数一致 | outputType.getFieldCount() == dynamicFilteringFieldIndices.size() |
| 並列度1 | 常に並列度1で実行される |

### 実行可否判定

ダイナミックフィルタリングが設定された場合に自動的に挿入される。出力型のフィールド数とフィルタリングフィールドインデックス数が一致しない場合はIllegalArgumentExceptionがスローされる。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| dynamicFilteringFieldIndices | List<Integer> | Yes | - | フィルタリング対象フィールドのインデックス |
| inputProperty | InputProperty | Yes | - | 入力プロパティ |
| outputType | RowType | Yes | - | 出力型（フィルタリングフィールドのみ） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 上流ExecNode | RowData | JOINビルド側からのデータ |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| ExecutionOrderEnforcer | Object | DynamicFilteringEvent（ソースコーディネーター経由） |

### 出力ファイル仕様

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

## 処理フロー

### 処理シーケンス

```
1. 入力変換の取得
   └─ 入力エッジから変換を取得
2. DynamicFilteringDataCollectorOperatorFactoryの生成
   └─ outputType（RowType）の設定
   └─ dynamicFilteringFieldIndicesの設定
   └─ 閾値（TABLE_EXEC_DYNAMIC_FILTERING_THRESHOLD）の設定
3. OneInputTransformationの構築
   └─ 並列度を1に固定
4. データ収集
   └─ 入力データからフィルタリングフィールドを抽出
   └─ 重複排除
5. 閾値チェック
   └─ 収集データ > 閾値? → フラグのみイベント
   └─ 収集データ <= 閾値? → 全データ含むイベント
6. DynamicFilteringEvent送信
   └─ ソースコーディネーターに送信
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[入力変換取得]
    B --> C[DynamicFilteringDataCollectorOperatorFactory生成]
    C --> D[OneInputTransformation構築]
    D --> E[並列度1設定]
    E --> F[データ収集・重複排除]
    F --> G{収集データ > 閾値?}
    G -->|Yes| H[フラグのみイベント生成]
    G -->|No| I[全データ含むイベント生成]
    H --> J[ソースコーディネーターに送信]
    I --> J
    J --> K[バッチ終了]
```

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

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

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

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

N/A

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IllegalArgumentException | フィールド数不整合 | outputType.getFieldCount() != dynamicFilteringFieldIndices.size() | クエリプランの確認 |

### リトライ仕様

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

### 障害時対応

クエリプランの確認とダイナミックフィルタリング設定の見直しが必要。

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

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

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | ビルド側データのユニークキー数 |
| 目標処理時間 | ビルド側処理時間に依存 |
| メモリ使用量上限 | TABLE_EXEC_DYNAMIC_FILTERING_THRESHOLD（デフォルト8MB） |

## 排他制御

並列度1で実行されるため、排他制御は不要。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| DEBUG | イベント送信時 | DynamicFilteringEvent内容（全データ or フラグのみ） |
| WARN | 閾値超過時 | 収集データサイズが閾値を超過した旨 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 収集データサイズ | TABLE_EXEC_DYNAMIC_FILTERING_THRESHOLD | Flinkメトリクス |

## 備考

- @ExecNodeMetadataアノテーションが付与されていない（永続化対象外）
- TABLE_EXEC_DYNAMIC_FILTERING_THRESHOLD設定オプション:
  - キー: `table.exec.dynamic-filtering.threshold`
  - デフォルト: 8MB
  - 説明: この閾値を超えるデータを収集した場合、フラグのみのDynamicFilteringEventを送信し、フィルタリングを事実上無効化
- 並列度は常に1に固定（line 106: `1, true`）
- pekko（旧akka）のフレームサイズ制限（RpcOptions.FRAMESIZE）を考慮した設計
- フレームサイズを超えるデータはOOMやRPCエラーの原因となるため、閾値で制限
- ExecNodeContextにBatchExecTableSourceScanを使用（永続化対象外のため実害なし）
