# 機能設計書 20-Catalystオプティマイザ

## 概要

本ドキュメントは、Apache Spark SQLのCatalystオプティマイザについて、論理実行プランの最適化ルールエンジンの設計仕様を記述した機能設計書である。

### 本機能の処理概要

解析済みの論理実行プラン（LogicalPlan）に対して、コストベース最適化（CBO）やルールベース最適化（RBO）を含む一連の最適化ルールを適用し、実行効率の高い等価な論理プランに変換する。RuleExecutorフレームワークに基づくバッチ処理方式で、複数の最適化ルールを組み合わせて反復適用する。

**業務上の目的・背景**：SQLクエリの実行効率はクエリプランの品質に大きく依存する。同じ結果を返すクエリでも、フィルタの適用順序、ジョイン方法、カラムプルーニング等により実行時間が大幅に異なる。Catalystオプティマイザは、豊富な最適化ルールを自動適用することで、ユーザーが記述したクエリを最適な実行プランに変換する。

**機能の利用シーン**：(1) spark.sql()でのSQLクエリ最適化、(2) DataFrame APIのクエリ最適化、(3) Structured Streamingのクエリ最適化、(4) CTEやサブクエリの最適化。

**主要な処理内容**：
1. defaultBatches定義による最適化バッチの構成
2. FixedPoint/Once戦略によるルールの反復適用制御
3. 述語プッシュダウン（PushDownPredicates）による効率的なフィルタリング
4. カラムプルーニング（ColumnPruning）による不要カラムの削除
5. 定数畳み込み（ConstantFolding）による定数式の事前計算
6. コストベースジョインリオーダリング（CostBasedJoinReorder）
7. サブクエリの最適化（OptimizeSubqueries, RewritePredicateSubquery）
8. プラン検証（validatePlanChanges）による最適化結果の整合性確認

**関連システム・外部連携**：Catalyst Analyzer（入力の解析済みプラン提供）、SparkPlanner（出力の最適化済みプランを物理プランに変換）、CatalogManager（カタログ情報へのアクセス）

**権限による制御**：特になし

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 16 | SQL Query Plans（SQL実行プラン） | 補助機能 | 最適化済み論理プランをOptimized Logical Planとして表示 |

## 機能種別

SQL処理基盤 / クエリ最適化

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| plan | LogicalPlan | Yes | 解析済み論理プラン | resolved状態であること |
| spark.sql.optimizer.maxIterations | Int | No | 最大反復回数 | デフォルト: 100 |
| spark.sql.optimizer.excludedRules | String | No | 除外するルール名 | カンマ区切り |

### 入力データソース

Catalyst Analyzerから出力された解析済みLogicalPlan。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| optimizedPlan | LogicalPlan | 最適化済み論理プラン |

### 出力先

SparkPlannerに渡され、物理実行プランに変換される。

## 処理フロー

### 処理シーケンス

```
1. Optimizer.execute(plan)呼び出し
   └─ RuleExecutor.execute()で全バッチを順次実行
2. Finish Analysis
   └─ 解析フェーズの残処理を完了
3. Rewrite With expression
   └─ With式の書き換え
4. Eliminate Distinct / Inline CTE / Union最適化
   └─ 初期最適化バッチ
5. LocalRelation early
   └─ 空リレーションの伝播による早期簡略化
6. Subquery最適化
   └─ サブクエリの最適化（相関サブクエリの引き上げ等）
7. Operator Optimization (FixedPoint)
   └─ メインの反復最適化ループ
   │  ├─ PushDownPredicates（述語プッシュダウン）
   │  ├─ ColumnPruning（カラムプルーニング）
   │  ├─ ConstantFolding（定数畳み込み）
   │  ├─ BooleanSimplification（ブーリアン簡略化）
   │  └─ 50以上のルール
8. Join Reorder (CBO)
   └─ コストベースジョインリオーダリング
9. RewriteSubquery
   └─ サブクエリの最終書き換え
10. NormalizeFloatingNumbers
    └─ 浮動小数点の正規化
```

### フローチャート

```mermaid
flowchart TD
    A[解析済みLogicalPlan入力] --> B[Finish Analysis]
    B --> C[Rewrite With expression]
    C --> D[Eliminate Distinct / Inline CTE]
    D --> E[Union / LocalRelation early]
    E --> F[Pullup Correlated Expressions]
    F --> G[Subquery最適化]
    G --> H[Replace Operators]
    H --> I[Aggregate最適化]
    I --> J[Operator Optimization FixedPoint]
    J --> K{収束?}
    K -->|No| J
    K -->|Yes| L[Pre CBO Rules]
    L --> M[Early Filter and Projection Push-Down]
    M --> N[Join Reorder CBO]
    N --> O[Eliminate Sorts]
    O --> P[Decimal Optimizations]
    P --> Q[LocalRelation]
    Q --> R[RewriteSubquery]
    R --> S[NormalizeFloatingNumbers]
    S --> T[最適化済みLogicalPlan出力]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-20-01 | FixedPoint反復 | 最大反復回数（デフォルト100回）まで変更がなくなるまで反復 | Operator Optimizationバッチ |
| BR-20-02 | Once実行 | 一部のバッチ（Infer Filters等）は1回のみ実行 | excludedOnceBatchesで定義 |
| BR-20-03 | ルール除外 | spark.sql.optimizer.excludedRulesで特定ルールを除外可能 | 設定時 |
| BR-20-04 | 除外不可ルール | nonExcludableRulesに列挙されたルールは除外できない | 常時 |
| BR-20-05 | プラン検証 | 各ルール適用後にvalidatePlanChangesで整合性を検証 | 全ルール適用後 |
| BR-20-06 | excludedOnceBatches | PartitionPruning, RewriteSubquery, Extract Python UDFs, Infer Filtersは一度だけ実行 | 特定バッチ |

### 計算ロジック

**FixedPoint収束判定**: ルール適用前後でプランが同一（参照等価性）であれば収束とみなす。最大反復回数（spark.sql.optimizer.maxIterations）を超えた場合も停止。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし（プラン変換のみ） |

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

該当なし

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | SparkException | プラン検証失敗（属性の不整合等） | 例外スロー |
| - | TreeNodeException | ルール適用中のエラー | 例外を上位に伝播 |

### リトライ仕様

リトライは行わない。最適化エラーは即座に報告。

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

該当なし

## パフォーマンス要件

最適化処理はドライバー側で実行される。FixedPointの最大反復回数でCPU消費を制限。大規模なクエリプランでは最適化時間が増加する可能性がある。

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

特になし

## 備考

- SparkOptimizer（sql/core側）がOptimizerを継承し、追加の最適化ルールとバッチを定義
- extendedOperatorOptimizationRulesメソッドをオーバーライドすることでカスタムルールを追加可能
- preCBORules, earlyScanPushDownRules等のフックポイントも提供
- TreePatternによるルール適用対象の効率的なフィルタリングが行われる

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | LogicalPlan.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala` | 論理プランのノード型階層 |
| 1-2 | Rule.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/Rule.scala` | Rule[LogicalPlan]の定義 |
| 1-3 | RuleExecutor.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala` | Batch, Strategy(Once/FixedPoint)の定義 |

**読解のコツ**: RuleExecutorはBatch単位でRuleを順次適用する。各Batchには実行戦略（Once: 1回、FixedPoint: 収束まで反復）が設定されている。TreePatternによるパターンマッチングで、不要なルール適用をスキップする仕組みが導入されている。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | Optimizer.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala` | メインクラス。defaultBatches()の構成を確認 |

**主要処理フロー**:
1. **51-52行目**: Optimizerクラス定義（RuleExecutor[LogicalPlan]を継承）
2. **54-64行目**: validatePlanChangesでプラン検証（軽量/通常）
3. **66-71行目**: excludedOnceBatchesの定義
4. **73-76行目**: fixedPointの設定（maxIterations）
5. **100-280行目**: defaultBatches()で全最適化バッチを定義
6. **101-162行目**: operatorOptimizationRuleSet（50以上のルール）
7. **164-174行目**: operatorOptimizationBatchの構成（Infer Filters含む）
8. **176-276行目**: batches定義（全バッチの実行順序）
9. **290-299行目以降**: nonExcludableRulesの定義

#### Step 3: 主要な最適化ルールを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Optimizer.scala | 同上 | PushDownPredicates, ColumnPruning, ConstantFolding等の個別ルール実装 |
| 3-2 | CostBasedJoinReorder.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala` | コストベースジョインリオーダリング |

#### Step 4: プラン検証層を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | LogicalPlanIntegrity.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/LogicalPlanIntegrity.scala` | validateOptimizedPlanの実装 |

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

```
QueryExecution.optimizedPlan
    │
    └─ Optimizer.execute(analyzedPlan)
         │
         ├─ RuleExecutor.execute()
         │    │
         │    ├─ Batch("Finish Analysis", FixedPoint(1))
         │    │    └─ FinishAnalysis
         │    │
         │    ├─ Batch("Rewrite With expression", fixedPoint)
         │    │    └─ RewriteWithExpression
         │    │
         │    ├─ Batch("Inline CTE", Once)
         │    │    └─ InlineCTE
         │    │
         │    ├─ Batch("Operator Optimization", fixedPoint)
         │    │    ├─ PushDownPredicates
         │    │    ├─ ColumnPruning
         │    │    ├─ ConstantFolding
         │    │    ├─ BooleanSimplification
         │    │    ├─ NullPropagation
         │    │    ├─ ReorderJoin
         │    │    ├─ EliminateOuterJoin
         │    │    └─ ... (50+ rules)
         │    │
         │    ├─ Batch("Join Reorder", FixedPoint(1))
         │    │    └─ CostBasedJoinReorder
         │    │
         │    └─ Batch("RewriteSubquery", Once)
         │         └─ RewritePredicateSubquery
         │
         └─ validatePlanChanges()
              └─ LogicalPlanIntegrity.validateOptimizedPlan()
```

### データフロー図

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

解析済み              ───▶ Optimizer.execute()         ───▶ 最適化済み
LogicalPlan                 │                                LogicalPlan
                            ├─ ルールベース最適化 (RBO)
                            │   ├─ 述語プッシュダウン
                            │   ├─ カラムプルーニング
                            │   ├─ 定数畳み込み
                            │   └─ サブクエリ書き換え
                            │
                            ├─ コストベース最適化 (CBO)
                            │   └─ ジョインリオーダリング
                            │
                            └─ プラン検証
                                └─ 属性整合性チェック
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| Optimizer.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala` | ソース | オプティマイザのメインクラス・ルール定義 |
| RuleExecutor.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala` | ソース | ルール実行フレームワーク |
| Rule.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/Rule.scala` | ソース | Ruleトレイト定義 |
| CostBasedJoinReorder.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/CostBasedJoinReorder.scala` | ソース | コストベースジョインリオーダリング |
| LogicalPlanIntegrity.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/LogicalPlanIntegrity.scala` | ソース | プラン整合性検証 |
| expressions/ | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala` | ソース | 式レベルの最適化ルール |
| joins.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala` | ソース | ジョイン最適化ルール |
| subquery.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala` | ソース | サブクエリ最適化ルール |
