# バッチ設計書 2-BatchExecBoundedStreamScan

## 概要

本ドキュメントは、Apache Flink Table APIにおけるBatchExecBoundedStreamScanバッチ実行ノードの設計仕様を記載する。このノードは、既存の有界DataStreamをTable APIのバッチ処理パイプラインに接続するための変換処理を担当する。

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

BatchExecBoundedStreamScanは、Flinkの有界DataStreamからデータを読み取り、Table APIのバッチ実行プランに統合するための実行ノードである。DataStreamからRowData形式への変換を行い、後続のテーブル操作ノードへデータを提供する。

**業務上の目的・背景**：既存のDataStream APIプログラムとTable APIを連携させる際に必要となる。レガシーなストリーム処理コードをTable SQLクエリと組み合わせて使用する場合や、カスタムデータソースからのデータをテーブル操作で処理する場合に使用される。

**バッチの実行タイミング**：Flinkジョブのバッチ実行時に、DataStreamからの入力ノードとして実行される。入力DataStreamは有界（Bounded）である必要がある。

**主要な処理内容**：
1. 入力DataStreamのTransformationを取得
2. 内部変換が必要か判定（時間属性フィールドの有無、型変換の必要性）
3. 必要に応じてRowData形式への変換処理を適用
4. フィールドインデックスに基づくデータマッピング

**前後の処理との関連**：前提となるバッチはなく、DataStreamからの入力として機能する。後続にはTable APIの各種処理ノード（Calc、Join、Aggregate等）が接続される。

**影響範囲**：入力DataStreamの型定義とフィールドマッピングが後続処理に影響。時間属性の扱いにも関与する。

## バッチ種別

データソース（DataStream変換）

## 実行スケジュール

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

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| 有界DataStream | 入力DataStreamが有界であること |
| 型定義 | DataStreamの型がTable APIと互換性があること |
| フィールドマッピング | fieldIndexesが正しく定義されていること |

### 実行可否判定

入力DataStreamが利用可能であり、出力RowTypeへの変換が可能であること。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| dataStream | DataStream<?> | Yes | - | 入力となる有界DataStream |
| sourceType | DataType | Yes | - | ソースデータの型定義 |
| fieldIndexes | int[] | Yes | - | フィールドインデックスマッピング |
| qualifiedName | List<String> | Yes | - | テーブルの完全修飾名 |
| outputType | RowType | Yes | - | 出力行の型定義 |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| DataStream | 任意のDataStream型 | 有界DataStreamからのデータ入力 |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| Transformation<RowData> | RowData | 変換されたレコードをRowData形式で後続ノードへ出力 |

### 出力ファイル仕様

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

## 処理フロー

### 処理シーケンス

```
1. translateToPlanInternal呼び出し
   └─ PlannerBaseとExecNodeConfigを受け取る
2. ソースTransformation取得
   └─ dataStream.getTransformation()を呼び出し
3. 内部変換必要性判定
   └─ needInternalConversion()で判定
4. 変換処理適用（必要時）
   └─ ScanUtil.convertToInternalRowを呼び出し
5. Transformation返却
   └─ 変換済みまたは元のTransformationを返却
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[DataStreamからTransformation取得]
    B --> C{内部変換が必要?}
    C -->|Yes| D[convertToInternalRow呼び出し]
    C -->|No| E[元のTransformationを使用]
    D --> F[Transformation返却]
    E --> F
    F --> G[バッチ終了]
```

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

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

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| データ読み取り | DataStream | SELECT相当 | DataStreamからデータを読み取り |

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

データベース操作は行わず、DataStreamからの読み取りのみ。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| ClassCastException | 型変換エラー | ソース型と出力型の不一致 | 型定義を確認・修正 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | 該当なし（同期処理） |
| リトライ間隔 | 該当なし |
| リトライ対象エラー | 該当なし |

### 障害時対応

ジョブ失敗時はFlinkのリスタート機構に従い再実行される。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | 該当なし |
| コミットタイミング | 該当なし（読み取り専用） |
| ロールバック条件 | 該当なし |

## パフォーマンス要件

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

## 排他制御

並列実行可能。各タスクは独立してデータを処理する。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | 処理開始時 | テーブル名、並列度 |
| 終了ログ | 処理完了時 | 処理件数 |
| エラーログ | エラー発生時 | 例外情報 |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 処理レコード数/秒 | ジョブ定義依存 | Flink Metrics |

## 備考

- MultipleTransformationTranslatorインターフェースを実装
- 時間属性フィールドがある場合は変換が必要
- ScanUtil.needsConversionで型変換の必要性を判定
