# バッチ設計書 5-BatchExecSink

## 概要

本ドキュメントは、Apache Flink Table APIにおけるBatchExecSinkバッチ実行ノードの設計仕様を記載する。このノードは、DynamicTableSinkで定義された外部シンクにデータを書き込むバッチ処理を担当する。

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

BatchExecSinkは、Flinkバッチ実行プランにおいて、処理結果を外部システム（データベース、ファイルシステム、メッセージキュー等）に書き込むための実行ノードである。DynamicTableSinkSpec経由で定義されたシンクコネクタを使用してデータを出力する。

**業務上の目的・背景**：ETL処理やデータ分析パイプラインの最終段階として、処理結果を永続化する。データウェアハウスへのロード、レポートファイルの生成、他システムへのデータ連携などに使用される。

**バッチの実行タイミング**：Flinkジョブのバッチ実行時に、実行プランの終端ノードとして実行される。上流の全処理が完了した後にデータ書き込みが行われる。

**主要な処理内容**：
1. 入力Transformationからデータを受け取る
2. 制約検証（NOT NULL、型長さ制限）の適用
3. 主キーに基づくパーティショニング（必要時）
4. DynamicTableSinkを使用してシンクTransformationを生成
5. 行レベル更新/削除のサポート

**前後の処理との関連**：上流のBatchExecCalc、BatchExecHashJoin、BatchExecHashAggregate等の処理結果を受け取る。後続処理はなく、データパイプラインの終端となる。

**影響範囲**：外部システムへのデータ書き込みに直接影響。トランザクション整合性、重複排除、エラー時のリカバリに注意が必要。

## バッチ種別

データシンク（出力処理）

## 実行スケジュール

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

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| テーブルシンク定義 | DynamicTableSinkSpecで有効なシンクが定義されていること |
| シンク接続 | 外部シンクへの接続が確立可能であること |
| 入力データ | 上流処理からの有効な入力Transformationが存在すること |

### 実行可否判定

DynamicTableSinkがSinkRuntimeProviderを提供可能であること。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| tableSinkSpec | DynamicTableSinkSpec | Yes | - | テーブルシンクの仕様 |
| inputProperty | InputProperty | Yes | - | 入力プロパティ |
| outputType | LogicalType | Yes | - | 出力型定義 |
| tableConfig | ReadableConfig | Yes | - | テーブル設定 |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| Transformation<RowData> | RowData | 上流処理からの入力データ |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 外部シンク | コネクタ依存 | DynamicTableSinkで定義されたシンクへ出力 |

### 出力ファイル仕様

コネクタ実装依存。ファイルシンクの場合は設定に従う。

## 処理フロー

### 処理シーケンス

```
1. translateToPlanInternal呼び出し
   └─ PlannerBaseとExecNodeConfigを受け取る
2. 入力Transformation取得
   └─ getInputEdges().get(0).translateToPlan
3. DynamicTableSink取得
   └─ tableSinkSpec.getTableSinkを呼び出し
4. シンクTransformation生成
   └─ createSinkTransformationを呼び出し
5. 制約検証適用
   └─ applyConstraintValidations
6. パーティショニング適用（必要時）
   └─ applyKeyBy（主キーがある場合）
7. シンクプロバイダ適用
   └─ applySinkProvider
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[入力Transformation取得]
    B --> C[DynamicTableSink取得]
    C --> D[制約検証適用]
    D --> E{主キーあり?}
    E -->|Yes| F[KeyByパーティショニング]
    E -->|No| G[パーティショニングなし]
    F --> H[シンクTransformation生成]
    G --> H
    H --> I[シンク出力]
    I --> J[バッチ終了]
```

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

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

| 処理 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| データ書き込み | 外部テーブル | INSERT/UPDATE/DELETE | シンクコネクタ経由で書き込み |

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

#### 外部シンクテーブル

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| INSERT | 全カラム | 入力データの値 | 通常モード |
| UPDATE_AFTER | 変更カラム | 入力データの値 | 行レベル更新時 |
| DELETE | 主キー | 削除対象 | 行レベル削除時 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | 並列度エラー | 無効な並列度設定 | 正の整数を設定 |
| TableException | 主キー不足 | 並列度変更時に主キーなし | 主キーを定義 |

### リトライ仕様

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

### 障害時対応

ジョブ失敗時はチェックポイントからの再実行。Exactly-onceセマンティクスはシンクコネクタのサポートに依存。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | コネクタ実装依存 |
| コミットタイミング | ジョブ完了時またはチェックポイント時 |
| ロールバック条件 | ジョブ失敗時（コネクタサポート依存） |

## パフォーマンス要件

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

## 排他制御

並列実行可能。主キーに基づくパーティショニングで順序保証を提供。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| 開始ログ | シンク開始時 | シンク名、並列度 |
| 進捗ログ | 書き込み中 | 処理済みレコード数 |
| 終了ログ | シンク完了時 | 総書き込み件数 |
| エラーログ | エラー発生時 | 例外情報 |

## 監視・アラート

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

## 備考

- FlinkVersion v2.0以降で使用可能
- ExecNodeMetadataのname: "batch-exec-sink"
- 消費オプション: table.exec.sink.not-null-enforcer, table.exec.sink.type-length-enforcer
- 行レベル更新/削除はRowLevelUpdateSpec/RowLevelDeleteSpecで制御
- isBounded=trueで有界データ出力を示す
- ChangelogMode.insertOnly()で挿入のみモード
