# バッチ設計書 1-BatchExecTableSourceScan

## 概要

本ドキュメントは、Apache Flink Table APIにおけるBatchExecTableSourceScanバッチ実行ノードの設計仕様を記載する。このノードは、ScanTableSourceで定義された外部データソースからデータを読み取るバッチ処理の基盤となる。

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

BatchExecTableSourceScanは、Flink Table APIのバッチ実行プランにおいて、外部ソースからデータを読み取るための実行ノードである。DynamicTableSourceSpec経由で定義されたテーブルソースから有界データを取得し、後続の処理ノードへRowData形式でデータを提供する。

**業務上の目的・背景**：SQLクエリやTable APIプログラムで外部データソース（データベース、ファイルシステム、メッセージキュー等）からデータを取得する際の入力処理を担当する。ETL処理やデータ分析パイプラインの起点として、様々な外部システムとの接続を抽象化し、統一的なデータ読み取りインターフェースを提供する。

**バッチの実行タイミング**：Flinkジョブのバッチ実行時に、実行プラン内の最初のノードとして実行される。ストリーミングではなくバッチモードでの実行を前提とし、データソースが有界（bounded）であることを要求する。

**主要な処理内容**：
1. DynamicTableSourceSpecからScanTableSourceを取得し、ScanRuntimeProviderを生成
2. InputFormat、Source、DataStreamなどのプロバイダタイプに応じたTransformationを生成
3. 動的フィルタリング（Dynamic Filtering）対応時はBatchExecDynamicFilteringDataCollectorとの連携
4. レガシー変換に対して有界性（Bounded）マークを付与

**前後の処理との関連**：前提となるバッチはなく、データパイプラインの起点となる。後続には、BatchExecCalc（フィルタ・射影）、BatchExecHashJoin（結合）、BatchExecHashAggregate（集約）などが接続される。動的フィルタリング使用時はBatchExecDynamicFilteringDataCollectorが入力となる。

**影響範囲**：テーブルソースの設定、パーティション情報、並列度設定が後続の全処理に影響する。リネージ（データ系譜）情報の生成にも関与。

## バッチ種別

データソース（入力処理）

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | ジョブ実行時（随時） |
| 実行時刻 | ジョブスケジュールに依存 |
| 実行曜日 | 該当なし |
| 実行日 | 該当なし |
| トリガー | Flinkジョブ実行時 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| テーブルソース定義 | DynamicTableSourceSpecで有効なテーブルソースが定義されていること |
| 有界データソース | ScanTableSourceが有界（bounded）データを提供すること |
| ソース接続 | 外部データソースへの接続が確立可能であること |

### 実行可否判定

テーブルソースがScanRuntimeProviderを提供可能であり、かつそのプロバイダが有界データを返すこと。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| tableSourceSpec | DynamicTableSourceSpec | Yes | - | テーブルソースの仕様 |
| outputType | RowType | Yes | - | 出力行の型定義 |
| inputProperty | InputProperty | No | 空リスト | 動的フィルタリング時の入力プロパティ |
| tableConfig | ReadableConfig | Yes | - | テーブル設定 |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 外部テーブルソース | ScanTableSource | DDLで定義されたコネクタ経由のデータソース |
| 動的フィルタ | BatchExecDynamicFilteringDataCollector | 動的フィルタリング使用時のフィルタ条件 |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| Transformation<RowData> | RowData | 読み取ったレコードをRowData形式で後続ノードへ出力 |

### 出力ファイル仕様

ファイル出力はなく、Flink内部のTransformationチェーンへ出力。

## 処理フロー

### 処理シーケンス

```
1. translateToPlanInternal呼び出し
   └─ PlannerBaseとExecNodeConfigを受け取る
2. ScanTableSourceの取得
   └─ tableSourceSpec.getScanTableSourceを呼び出し
3. ScanRuntimeProviderの取得
   └─ tableSource.getScanRuntimeProviderを呼び出し
4. プロバイダタイプ別のTransformation生成
   └─ SourceFunctionProvider/InputFormatProvider/SourceProvider/DataStreamScanProvider対応
5. リネージ情報の設定
   └─ TableSourceLineageVertexを生成・設定
6. レガシーTransformationの有界化
   └─ makeLegacySourceTransformationsBoundedを呼び出し
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[DynamicTableSourceSpec取得]
    B --> C[ScanTableSource生成]
    C --> D[ScanRuntimeProvider取得]
    D --> E{プロバイダタイプ判定}
    E -->|SourceFunctionProvider| F[SourceFunction Transformation生成]
    E -->|InputFormatProvider| G[InputFormat Transformation生成]
    E -->|SourceProvider| H[Source Transformation生成]
    E -->|DataStreamScanProvider| I[DataStream Transformation生成]
    F --> J[リネージ情報設定]
    G --> J
    H --> J
    I --> J
    J --> K[有界マーク付与]
    K --> L[Transformation返却]
    L --> M[バッチ終了]
```

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

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

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| データ読み取り | 外部テーブル | SELECT | コネクタ経由で外部データソースからデータを読み取り |

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

#### 外部テーブルソース

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| SELECT | 全カラム | テーブルソース定義に従う | プッシュダウンフィルタ適用可能 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| UnsupportedOperationException | 未対応プロバイダ | サポート外のRuntimeProviderタイプ | サポートされるプロバイダに変更 |
| TableException | 無効な並列度 | 並列度が0以下に設定 | 正の整数を設定 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | コネクタ実装依存 |
| リトライ間隔 | コネクタ実装依存 |
| リトライ対象エラー | 一時的な接続エラー |

### 障害時対応

ジョブ失敗時はFlinkのチェックポイント/リスタート機構に従い、ソース読み取りから再実行される。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | コネクタ実装依存 |
| コミットタイミング | 読み取り専用のためコミット不要 |
| ロールバック条件 | ジョブ失敗時（読み取りのため影響なし） |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | ソースデータサイズ依存 |
| 目標処理時間 | 並列度・データ量依存 |
| メモリ使用量上限 | TaskManager設定に従う |

## 排他制御

並列実行可能。各タスクは独立したパーティション/スプリットを処理する。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | ソース読み取り開始時 | ソース名、並列度 |
| 進捗ログ | スプリット処理完了時 | 処理済みスプリット数 |
| 終了ログ | 全データ読み取り完了時 | 総処理件数 |
| エラーログ | エラー発生時 | 例外情報、スタックトレース |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理レコード数/秒 | ジョブ定義依存 | Flink Metrics |
| バックプレッシャー | 高 | Flink Dashboard |

## 備考

- FlinkVersion v2.0以降で使用可能
- 動的フィルタリング機能はdynamicFilteringDataListenerIDで管理
- レガシーUIDは無効化（legacyUidsEnabled = false）
