# 機能設計書 91-Declarative Pipelines

## 概要

本ドキュメントは、Apache SparkのSQL Pipelines機能（Declarative Pipelines）の設計について記述する。宣言的なパイプライン定義によるETLワークフローの構築と実行を提供する機能である。

### 本機能の処理概要

**業務上の目的・背景**：データエンジニアリングにおけるETLパイプラインの構築は、従来命令的なコードで記述する必要があり、テーブル間の依存関係の管理やエラーハンドリングの実装が煩雑であった。Declarative Pipelinesは、テーブル・ビュー・フロー（データ変換）を宣言的に定義し、依存関係の自動解決、トポロジカル順序での実行、リトライ制御を自動化することで、ETLワークフローの開発生産性と信頼性を向上させる。

**機能の利用シーン**：データウェアハウスやデータレイクにおいて、複数のソーステーブルから変換処理を経て最終的な分析用テーブルを生成するETLパイプラインを構築する場面で利用される。ストリーミングテーブルとマテリアライズドビューの両方をサポートし、バッチ処理と増分処理の両方に対応する。

**主要な処理内容**：
1. GraphRegistrationContextを通じたテーブル、ビュー、フロー（データ変換）の宣言的登録
2. DataflowGraphによるグラフ構造の構築とトポロジカルソートによる依存関係解決
3. PipelineExecutionによるグラフの解決・バリデーション・実行制御
4. TriggeredGraphExecutionによるトポロジカル順序でのフロー実行と並行制御
5. DatasetManagerによるカタログへのテーブル・ビューのマテリアライゼーション
6. 指数バックオフ付きリトライ機構による障害復旧

**関連システム・外部連携**：Spark SQLのカタログシステム（TableCatalog V2 API）と連携し、テーブルメタデータの管理を行う。Structured Streamingエンジンを利用したストリーミングフロー実行にも対応する。

**権限による制御**：カタログレベルでのテーブル操作権限に依存する。パイプライン自体には独自の権限制御は実装されていない。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 画面機能マッピングに本機能に対応する画面定義なし |

## 機能種別

データ連携 / ETLパイプライン実行

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| tables | Seq[Table] | Yes | パイプラインで定義されるテーブル一覧 | 少なくとも1つのテーブルまたはPersistedViewが必要 |
| views | Seq[View] | No | パイプラインで定義されるビュー一覧 | TemporaryViewまたはPersistedView |
| sinks | Seq[Sink] | No | パイプラインで定義されるシンク一覧 | フォーマットとオプションを指定 |
| flows | Seq[UnresolvedFlow] | Yes | データ変換を定義するフロー一覧 | 各テーブル・ビューに少なくとも1つのフローが必要 |
| defaultCatalog | String | Yes | デフォルトカタログ名 | 有効なカタログ名 |
| defaultDatabase | String | Yes | デフォルトデータベース名 | 有効なデータベース名 |
| fullRefreshTables | GraphFilter | No | フルリフレッシュ対象テーブル | resetAllowedプロパティがtrueである必要あり |

### 入力データソース

SQL文やDataFrame APIを通じたユーザー定義のデータ変換ロジック（FlowFunction）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| マテリアライズされたテーブル | カタログテーブル | フロー実行結果が書き込まれたテーブル |
| パーシステッドビュー | カタログビュー | DDLで作成されたビュー |
| PipelineEvent | イベント | パイプライン実行の進捗・結果イベント |

### 出力先

Spark SQLカタログ（テーブル・ビュー）、PipelineEventコールバック

## 処理フロー

### 処理シーケンス

```
1. GraphRegistrationContext: テーブル・ビュー・フローの登録
   └─ registerTable/registerView/registerFlow で各要素を登録
2. toDataflowGraph: DataflowGraph の構築
   └─ 重複チェック、空パイプラインチェック
3. PipelineExecution.startPipeline / runPipeline
   └─ resolveGraph: グラフの解決とバリデーション
      └─ DataflowGraphTransformer で各ノードを処理
      └─ validate: トポロジカルソート、スキーマ検証等
4. State.reset: フルリフレッシュ対象テーブルの状態リセット
5. DatasetManager.materializeDatasets: テーブル・ビューのカタログマテリアライゼーション
   └─ テーブルの作成/更新、スキーママージ
   └─ ビューの作成/更新
6. TriggeredGraphExecution.start: トポロジカル実行開始
   └─ フローのキューイング
   └─ トポロジカル順序でのフロー実行
7. topologicalExecution: メインループ
   └─ 親フローの完了確認 → フロー開始
   └─ フロー成功/失敗の記録
   └─ リトライ制御（指数バックオフ）
   └─ 下流フローのスキップ
8. onCompletion: 終了コールバック
```

### フローチャート

```mermaid
flowchart TD
    A[パイプライン開始] --> B[グラフ解決・バリデーション]
    B -->|失敗| C[エラーイベント発行]
    B -->|成功| D[テーブル・ビューのマテリアライゼーション]
    D --> E[TriggeredGraphExecution開始]
    E --> F{未完了フローあり?}
    F -->|No| G[完了判定・終了]
    F -->|Yes| H{実行可能フローあり?}
    H -->|Yes| I[フロー開始]
    I --> J{フロー結果}
    J -->|成功| K[SUCCESSFUL記録]
    J -->|失敗| L{リトライ可能?}
    L -->|Yes| M[指数バックオフ後リトライ]
    L -->|No| N[下流フローSKIPPED]
    K --> F
    M --> F
    N --> F
    H -->|No| O[ポーリング待機]
    O --> F
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-91-01 | フロー実行順序 | フローはトポロジカル順序で実行される | 常時 |
| BR-91-02 | 並行実行制限 | 同時実行フロー数はmaxConcurrentFlows設定に従う | 常時 |
| BR-91-03 | ONCEフロー | ONCEフローはフルリフレッシュ時に1回のみ実行 | once=true |
| BR-91-04 | リセット可否 | resetAllowed=falseのテーブルはフルリフレッシュ不可 | フルリフレッシュ指定時 |
| BR-91-05 | スキーマ推論 | テーブルスキーマは指定がなければフローの出力スキーマから推論 | specifiedSchema未指定時 |
| BR-91-06 | 下流スキップ | フローが最終的に失敗すると下流フローはSKIPPEDになる | リトライ上限超過時 |

### 計算ロジック

リトライ待機時間の計算: `ExponentialBackoffStrategy`により、`stepSize * 2^(numFailures-1)` で指数的に増加し、`maxTime`を上限とする。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| テーブルマテリアライゼーション | ユーザー定義テーブル | CREATE/ALTER | カタログにテーブルを作成・更新 |
| フルリフレッシュ | ユーザー定義テーブル | TRUNCATE | テーブルデータの全削除 |
| ビューマテリアライゼーション | ユーザー定義ビュー | CREATE VIEW | カタログにビューを作成 |
| フロー実行 | ユーザー定義テーブル | INSERT/MERGE | フローの変換結果をテーブルに書き込み |

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

#### ユーザー定義テーブル（動的）

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| CREATE | フローから推論されたスキーマ | DataflowGraph.inferredSchema | スキーマ未指定時 |
| ALTER | スキーマ変更カラム | diffSchemasで差分検出 | ストリーミングテーブルの増分更新時 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| RUN_EMPTY_PIPELINE | AnalysisException | テーブルもPersistedViewもSinkも登録されていない | テーブルまたはビューを定義する |
| PIPELINE_DUPLICATE_IDENTIFIERS.FLOW | AnalysisException | 同一識別子のフローが複数定義 | フロー名を一意にする |
| PIPELINE_DUPLICATE_IDENTIFIERS.OUTPUT | AnalysisException | 同一識別子のテーブル/ビュー/シンクが重複 | 識別子を一意にする |
| PIPELINE_DATASET_WITHOUT_FLOW | AnalysisException | テーブル/ビューに対応するフローがない | フローを定義する |
| QueryExecutionFailure | RunTerminationReason | フローがリトライ上限超過 | フローのロジックを修正する |

### リトライ仕様

- 非致命的エラーの場合、指数バックオフ付きで自動リトライ
- 最大リトライ回数: `spark.sql.pipelines.maxFlowRetryAttempts` で設定
- 最大リトライ時間: `watchdogMaxRetryTimeInSeconds` で設定
- 最小リトライ間隔: `watchdogMinRetryTimeInSeconds` で設定

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

各フロー実行はStructured StreamingまたはバッチSQLのトランザクション保証に従う。パイプライン全体としてのトランザクション保証はなく、フロー単位での一貫性が保証される。

## パフォーマンス要件

- フローの並行実行数は `maxConcurrentFlows` で制御
- ストリーミングフローはAvailableNowトリガーで利用可能データをすべて処理
- ポーリング間隔は `streamStatePollingInterval` で設定

## セキュリティ考慮事項

- カタログレベルのアクセス制御に依存
- SQL設定はフローレベルで分離可能（sqlConf）

## 備考

Spark 4.0で導入された新機能。`sql/pipelines/` パッケージ配下に実装されている。

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

まず、パイプラインを構成するグラフ要素（テーブル、ビュー、フロー、シンク）の型定義を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | elements.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/elements.scala` | GraphElement, Input, Output, Table, View, Sink等のトレイトとケースクラスの定義 |
| 1-2 | Flow.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala` | Flow, FlowFunction, ResolvedFlow, StreamingFlow, CompleteFlow等のフロー型階層 |

**読解のコツ**: `elements.scala`のトレイト階層（GraphElement -> Input/Output -> Table/View/Sink）を先に理解し、次にFlow.scalaでフローの状態遷移（UnresolvedFlow -> ResolvedFlow/ResolutionFailedFlow）を押さえる。

#### Step 2: エントリーポイントを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | GraphRegistrationContext.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphRegistrationContext.scala` | グラフ要素の登録とDataflowGraph構築 |
| 2-2 | PipelineExecution.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineExecution.scala` | パイプライン実行のエントリーポイント |

**主要処理フロー**:
1. **41-43行目** (GraphRegistrationContext): registerTable/registerView/registerFlowでグラフ要素を登録
2. **71-91行目** (GraphRegistrationContext): toDataflowGraphで重複チェック後DataflowGraphを構築
3. **46-62行目** (PipelineExecution): startPipelineでグラフ解決→マテリアライゼーション→実行開始
4. **65-85行目** (PipelineExecution): runPipelineで同期的に実行完了まで待機

#### Step 3: グラフ構造を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | DataflowGraph.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala` | グラフ構造の中核。flows/tables/sinks/viewsの管理、解決、バリデーション |

**主要処理フロー**:
- **33-39行目**: DataflowGraphケースクラスの定義。GraphOperationsとGraphValidationsをミックスイン
- **48-52行目**: materializedFlowsの算出（出力先がOutputに含まれるフローのみ）
- **174-182行目**: inferredSchemaの算出（全フローのスキーマをマージ）
- **231-238行目**: resolve()でDataflowGraphTransformerとCoreDataflowNodeProcessorを使用してグラフ解決

#### Step 4: 実行制御を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | GraphExecution.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala` | フロー実行の基底クラス。FlowPlannerとStreamListenerの管理 |
| 4-2 | TriggeredGraphExecution.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala` | トポロジカル順序実行の具体実装 |

**主要処理フロー**:
- **95-128行目** (TriggeredGraphExecution): start()でフローのキューイングとトポロジカル実行スレッド起動
- **157-263行目** (TriggeredGraphExecution): topologicalExecution()メインループ。状態遷移（QUEUED->RUNNING->SUCCESSFUL/TERMINATED_WITH_ERROR）管理
- **282-336行目** (TriggeredGraphExecution): recordFailed()でリトライ判定と下流フローのスキップ処理

#### Step 5: マテリアライゼーションを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | DatasetManager.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala` | テーブル・ビューのカタログマテリアライゼーション |

**主要処理フロー**:
- **75-130行目**: materializeDatasets()でテーブル作成/更新、ビュー公開
- **246-338行目**: materializeTable()でカタログAPIを使用したテーブル操作

### プログラム呼び出し階層図

```
PipelineExecution.startPipeline / runPipeline
    |
    +-- resolveGraph()
    |     +-- DataflowGraph.resolve()
    |     |     +-- DataflowGraphTransformer
    |     |           +-- CoreDataflowNodeProcessor.processNode
    |     +-- DataflowGraph.validate()
    |           +-- validateSuccessfulFlowAnalysis()
    |           +-- validateGraphIsTopologicallySorted()
    |           +-- validateMultiQueryTables()
    |
    +-- State.reset() [フルリフレッシュ時]
    |
    +-- DatasetManager.materializeDatasets()
    |     +-- materializeTable() [各テーブル]
    |     +-- materializeViews() [各PersistedView]
    |
    +-- TriggeredGraphExecution.start()
          +-- topologicalExecution() [別スレッド]
                +-- planAndStartFlow() [各フロー]
                |     +-- FlowPlanner.plan()
                |     +-- FlowExecution.executeAsync()
                +-- recordSuccess() / recordFailed()
                +-- onCompletion()
```

### データフロー図

```
[入力]                       [処理]                           [出力]

GraphRegistrationContext --> DataflowGraph.resolve()     --> マテリアライズされた
  (テーブル/ビュー/フロー)       |                              テーブル・ビュー
                             DatasetManager              --> カタログメタデータ
                               |
                             TriggeredGraphExecution     --> PipelineEvent
                               (トポロジカル実行)              (進捗/結果イベント)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| elements.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/elements.scala` | ソース | グラフ要素の型定義 |
| Flow.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/Flow.scala` | ソース | フローの型定義と状態管理 |
| DataflowGraph.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraph.scala` | ソース | グラフ構造の中核 |
| GraphRegistrationContext.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphRegistrationContext.scala` | ソース | グラフ要素の登録 |
| PipelineExecution.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineExecution.scala` | ソース | パイプライン実行制御 |
| GraphExecution.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphExecution.scala` | ソース | フロー実行の基底クラス |
| TriggeredGraphExecution.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/TriggeredGraphExecution.scala` | ソース | トポロジカル実行 |
| DatasetManager.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala` | ソース | テーブル・ビューマテリアライゼーション |
| FlowPlanner.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowPlanner.scala` | ソース | フロー実行計画の生成 |
| FlowExecution.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala` | ソース | フロー実行の具体処理 |
| DataflowGraphTransformer.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DataflowGraphTransformer.scala` | ソース | グラフ変換処理 |
| CoreDataflowNodeProcessor.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/CoreDataflowNodeProcessor.scala` | ソース | ノード解決処理 |
| GraphOperations.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphOperations.scala` | ソース | グラフ操作（DFS、上流/下流探索） |
| GraphValidations.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/GraphValidations.scala` | ソース | グラフバリデーション |
| State.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/State.scala` | ソース | 状態管理 |
| PipelineUpdateContext.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/PipelineUpdateContext.scala` | ソース | パイプライン更新コンテキスト |
| PipelineEvent.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/logging/PipelineEvent.scala` | ソース | イベントモデル |
| FlowProgressEventLogger.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/logging/FlowProgressEventLogger.scala` | ソース | フロー進捗イベントロガー |
| BackoffStrategy.scala | `sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/util/BackoffStrategy.scala` | ソース | リトライバックオフ戦略 |
