# 機能設計書：SQLプランナー

## 1. 機能概要

### 1.1 処理概要

SQLプランナー（Planner）は、Apache Flink Table/SQLにおいて、解析されたSQL（SqlNode）やTable API操作（Operation）から実行計画を生成し、最適化を行い、最終的にFlink Transformationに変換する機能である。

主要なコンポーネント：
1. **Planner**: プランナーインターフェース
2. **PlannerBase**: プランナー基底実装（Scala）
3. **StreamPlanner**: ストリーミングモード用プランナー
4. **BatchPlanner**: バッチモード用プランナー
5. **Optimizer**: 最適化エンジン

### 1.2 業務上の目的・役割

| 項目 | 内容 |
|------|------|
| 目的 | SQL/Table API操作を効率的な実行計画に変換 |
| 役割 | クエリ最適化・コード生成・Transformation生成 |
| 主な利用場面 | SQLクエリ実行、Table API実行、EXPLAIN出力 |

### 1.3 利用シーン・ユースケース

1. **クエリ実行**
   - SQLクエリの実行計画生成
   - Table API操作の変換

2. **最適化**
   - ルールベース最適化
   - コストベース最適化
   - 共通サブグラフ最適化

3. **実行計画表示**
   - EXPLAIN文の実行
   - AST/物理計画/実行計画の表示

## 2. 入出力仕様

### 2.1 Plannerインターフェース

| メソッド | 入力 | 出力 | 説明 |
|---------|------|------|------|
| getParser() | - | Parser | SQLパーサー取得 |
| translate() | List<ModifyOperation> | List<Transformation<?>> | 実行計画変換 |
| explain() | List<Operation>, ExplainFormat, ExplainDetail... | String | 実行計画説明 |
| loadPlan() | PlanReference | InternalPlan | コンパイル済プラン読み込み |
| compilePlan() | List<ModifyOperation> | InternalPlan | プランコンパイル |
| translatePlan() | InternalPlan | List<Transformation<?>> | コンパイル済プラン変換 |

### 2.2 ExplainDetail種別

| 種別 | 説明 |
|------|------|
| ESTIMATED_COST | 推定コスト表示 |
| CHANGELOG_MODE | Changelog mode表示（ストリーミング） |
| PLAN_ADVICE | 最適化アドバイス表示 |
| JSON_EXECUTION_PLAN | JSON形式の物理実行計画 |

### 2.3 ExplainFormat種別

| 種別 | 説明 |
|------|------|
| TEXT | テキスト形式 |
| JSON | JSON形式（将来対応） |

## 3. 処理フロー

### 3.1 translate処理フロー

```mermaid
sequenceDiagram
    participant TE as TableEnvironment
    participant P as Planner
    participant O as Optimizer
    participant ENG as ExecNodeGraphGenerator
    participant T as Transformation

    TE->>P: translate(modifyOperations)
    P->>P: translateToRel(operations)
    Note over P: ModifyOperation → RelNode
    P->>O: optimize(relNodes)
    Note over O: ルールベース最適化<br/>コストベース最適化
    O-->>P: optimizedRelNodes
    P->>ENG: translateToExecNodeGraph()
    ENG-->>P: ExecNodeGraph
    P->>P: translateToPlan(execGraph)
    Note over P: ExecNode → Transformation
    P-->>TE: List<Transformation>
```

### 3.2 プランナー継承階層

```mermaid
classDiagram
    class Planner {
        <<interface>>
        +getParser() Parser
        +translate(List~ModifyOperation~) List~Transformation~
        +explain(List~Operation~, ExplainFormat, ExplainDetail...) String
        +loadPlan(PlanReference) InternalPlan
        +compilePlan(List~ModifyOperation~) InternalPlan
        +translatePlan(InternalPlan) List~Transformation~
    }

    class PlannerBase {
        <<abstract>>
        #executor Executor
        #tableConfig TableConfig
        #moduleManager ModuleManager
        #functionCatalog FunctionCatalog
        #catalogManager CatalogManager
        #isStreamingMode boolean
        +getParser() Parser
        +translate() List~Transformation~
        #getOptimizer() Optimizer
        #getTraitDefs() RelTraitDef[]
        #translateToPlan(ExecNodeGraph) List~Transformation~
    }

    class StreamPlanner {
        -isStreamingMode = true
        +getOptimizer() StreamCommonSubGraphBasedOptimizer
        +getTraitDefs() TraitDef[]
    }

    class BatchPlanner {
        -isStreamingMode = false
        +getOptimizer() BatchCommonSubGraphBasedOptimizer
        +getTraitDefs() TraitDef[]
        +getExecNodeGraphProcessors() Seq
    }

    Planner <|.. PlannerBase
    PlannerBase <|-- StreamPlanner
    PlannerBase <|-- BatchPlanner
```

### 3.3 最適化フロー

```mermaid
flowchart TD
    subgraph Input["入力"]
        MO[ModifyOperation] --> RN[RelNode]
    end

    subgraph Optimization["最適化"]
        RN --> RBO[ルールベース最適化]
        RBO --> CBO[コストベース最適化]
        CBO --> CSG[共通サブグラフ最適化]
    end

    subgraph Output["出力"]
        CSG --> ORN[Optimized RelNode]
        ORN --> ENG[ExecNodeGraph]
        ENG --> TF[Transformation]
    end
```

## 4. 業務ルール・条件

### 4.1 StreamPlannerのTraitDef

| TraitDef | 説明 |
|----------|------|
| ConventionTraitDef | 実行規約（物理/論理） |
| FlinkRelDistributionTraitDef | データ分散特性 |
| MiniBatchIntervalTraitDef | ミニバッチ間隔 |
| ModifyKindSetTraitDef | 変更種別セット（INSERT/UPDATE/DELETE） |
| UpdateKindTraitDef | 更新種別（BEFORE/AFTER） |
| DuplicateChangesTraitDef | 重複変更特性 |

### 4.2 BatchPlannerのTraitDef

| TraitDef | 説明 |
|----------|------|
| ConventionTraitDef | 実行規約（物理/論理） |
| FlinkRelDistributionTraitDef | データ分散特性 |
| RelCollationTraitDef | ソート順序 |

### 4.3 BatchPlannerのExecNodeGraphProcessor

| Processor | 説明 |
|-----------|------|
| DeadlockBreakupProcessor | デッドロック回避 |
| DynamicFilteringDependencyProcessor | 動的フィルタリング依存 |
| MultipleInputNodeCreationProcessor | 複数入力ノード作成 |
| ForwardHashExchangeProcessor | ハッシュ交換最適化 |
| AdaptiveJoinProcessor | 適応型JOIN最適化 |

### 4.4 PlannerBaseの主要メソッド

| メソッド | 説明 |
|---------|------|
| beforeTranslation() | 変換前処理 |
| translateToRel() | ModifyOperation → RelNode |
| optimize() | RelNode最適化 |
| translateToExecNodeGraph() | RelNode → ExecNodeGraph |
| translateToPlan() | ExecNodeGraph → Transformation |
| afterTranslation() | 変換後処理 |

### 4.5 ランタイムモード検証

```scala
// StreamPlanner
override def beforeTranslation(): Unit = {
  val runtimeMode = getTableConfig.get(ExecutionOptions.RUNTIME_MODE)
  if (runtimeMode != RuntimeExecutionMode.STREAMING) {
    throw new IllegalArgumentException("...")
  }
}

// BatchPlanner
override def beforeTranslation(): Unit = {
  val runtimeMode = getTableConfig.get(ExecutionOptions.RUNTIME_MODE)
  if (runtimeMode != RuntimeExecutionMode.BATCH) {
    throw new IllegalArgumentException("...")
  }
}
```

### 4.6 EXPLAIN出力形式

```
== Abstract Syntax Tree ==
[RelNode AST表示]

== Optimized Physical Plan ==
[最適化後の物理計画]

== Optimized Execution Plan ==
[ExecNodeグラフ表示]

== Physical Execution Plan ==  (JSON_EXECUTION_PLANオプション時)
[StreamGraph JSON]
```

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

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

```
Planner (インターフェース)
├── getParser() → Parser
│   └── ParserFactory.create() → Parser
│
├── translate(List<ModifyOperation>) → List<Transformation>
│   ├── beforeTranslation()
│   ├── translateToRel(operation) → RelNode
│   ├── optimize(relNodes) → optimizedRelNodes
│   │   └── Optimizer.optimize()
│   ├── translateToExecNodeGraph(optimizedRelNodes) → ExecNodeGraph
│   │   └── ExecNodeGraphGenerator.generate()
│   ├── translateToPlan(execGraph) → transformations
│   │   ├── StreamExecNode.translateToPlan()  [Streaming]
│   │   └── BatchExecNode.translateToPlan()   [Batch]
│   └── afterTranslation()
│
├── explain(operations, format, extraDetails...) → String
│   ├── getExplainGraphs(operations)
│   │   ├── sinkRelNodes
│   │   ├── optimizedRelNodes
│   │   ├── execGraph
│   │   └── streamGraph
│   └── フォーマット出力
│
└── compilePlan(modifyOperations) → InternalPlan
    └── ExecNodeGraph → ExecNodeGraphInternalPlan

PlannerBase
├── StreamPlanner
│   ├── getOptimizer() → StreamCommonSubGraphBasedOptimizer
│   ├── getTraitDefs() → [Convention, Distribution, MiniBatch, ModifyKindSet, UpdateKind, DuplicateChanges]
│   └── getExecNodeGraphProcessors() → Seq.empty
│
└── BatchPlanner
    ├── getOptimizer() → BatchCommonSubGraphBasedOptimizer
    ├── getTraitDefs() → [Convention, Distribution, Collation]
    └── getExecNodeGraphProcessors() → [Deadlock, DynamicFiltering, MultipleInput, ForwardHash, AdaptiveJoin]
```

### 5.2 データフロー図

```
[SQL/Table API操作]
        │
        ▼
[ModifyOperation]
        │
        ▼ translateToRel()
[RelNode (論理計画)]
        │
        ▼ optimize()
[Optimized RelNode (物理計画)]
        │
        ▼ translateToExecNodeGraph()
[ExecNodeGraph]
        │
        ▼ ExecNodeGraphProcessor (Batch only)
[Processed ExecNodeGraph]
        │
        ▼ translateToPlan()
[Transformation]
        │
        ▼
[StreamGraph / JobGraph]
```

### 5.3 関連ファイル一覧

| ファイルパス | 役割 |
|-------------|------|
| `flink-table-api-java/.../Planner.java` | プランナーインターフェース |
| `flink-table-planner/.../PlannerBase.scala` | プランナー基底クラス |
| `flink-table-planner/.../StreamPlanner.scala` | ストリーミングプランナー |
| `flink-table-planner/.../BatchPlanner.scala` | バッチプランナー |
| `flink-table-planner/.../PlannerContext.java` | プランナーコンテキスト |
| `flink-table-planner/.../DefaultPlannerFactory.java` | プランナーファクトリ |
| `flink-table-planner/.../FlinkPlannerImpl.scala` | Calciteプランナー実装 |
| `flink-table-planner/.../StreamCommonSubGraphBasedOptimizer.scala` | ストリーム最適化 |
| `flink-table-planner/.../BatchCommonSubGraphBasedOptimizer.scala` | バッチ最適化 |
| `flink-table-planner/.../ExecNodeGraphGenerator.scala` | ExecNodeグラフ生成 |

### 5.4 コードリーディング手順

#### Step 1: Plannerインターフェースの理解

```java
// Planner.java (34-55行目)
/**
 * This interface serves two purposes:
 * - SQL parser via getParser() - transforms a SQL string into Table API specific objects
 * - relational planner - provides a way to plan, optimize and transform tree of
 *   ModifyOperation into a runnable form (Transformation)
 */
@Internal
public interface Planner {
    Parser getParser();
    List<Transformation<?>> translate(List<ModifyOperation> modifyOperations);
    String explain(List<Operation> operations, ExplainFormat format, ExplainDetail... extraDetails);
}
```

#### Step 2: PlannerBaseの構造

```scala
// PlannerBase.scala (74-101行目)
/**
 * Implementation of a Planner.
 * @param executor instance of Executor
 * @param tableConfig mutable configuration
 * @param moduleManager manager for modules
 * @param functionCatalog catalog of functions
 * @param catalogManager manager of catalog meta objects
 * @param isStreamingMode Determines if the planner should work in batch or streaming mode
 */
abstract class PlannerBase(
    executor: Executor,
    tableConfig: TableConfig,
    val moduleManager: ModuleManager,
    val functionCatalog: FunctionCatalog,
    val catalogManager: CatalogManager,
    isStreamingMode: Boolean,
    classLoader: ClassLoader) extends Planner {
```

#### Step 3: translate処理フロー

```scala
// PlannerBase.scala (175-188行目)
override def translate(modifyOperations: util.List[ModifyOperation]): util.List[Transformation[_]] = {
  beforeTranslation()
  if (modifyOperations.isEmpty) {
    return List.empty[Transformation[_]].asJava
  }
  val relNodes = modifyOperations.asScala.map(translateToRel)
  val optimizedRelNodes = optimize(relNodes)
  val execGraph = translateToExecNodeGraph(optimizedRelNodes, isCompiled = false)
  val transformations = translateToPlan(execGraph)
  afterTranslation()
  transformations
}
```

#### Step 4: StreamPlannerの特性

```scala
// StreamPlanner.scala (47-74行目)
class StreamPlanner(...) extends PlannerBase(..., isStreamingMode = true, ...) {
  override protected def getTraitDefs: Array[RelTraitDef[_ <: RelTrait]] = {
    Array(
      ConventionTraitDef.INSTANCE,
      FlinkRelDistributionTraitDef.INSTANCE,
      MiniBatchIntervalTraitDef.INSTANCE,
      ModifyKindSetTraitDef.INSTANCE,
      UpdateKindTraitDef.INSTANCE,
      DuplicateChangesTraitDef.INSTANCE
    )
  }

  override protected def getOptimizer: Optimizer = new StreamCommonSubGraphBasedOptimizer(this)
}
```

#### Step 5: BatchPlannerの特性

```scala
// BatchPlanner.scala (49-88行目)
class BatchPlanner(...) extends PlannerBase(..., isStreamingMode = false, ...) {
  override protected def getTraitDefs: Array[RelTraitDef[_ <: RelTrait]] = {
    Array(
      ConventionTraitDef.INSTANCE,
      FlinkRelDistributionTraitDef.INSTANCE,
      RelCollationTraitDef.INSTANCE
    )
  }

  override protected def getOptimizer: Optimizer = new BatchCommonSubGraphBasedOptimizer(this)

  override def getExecNodeGraphProcessors: Seq[ExecNodeGraphProcessor] = {
    val processors = new util.ArrayList[ExecNodeGraphProcessor]()
    processors.add(new DeadlockBreakupProcessor())
    // ... additional processors
    processors
  }
}
```

## 6. 関連機能・API

### 6.1 関連機能

| 機能名 | 関連種別 | 説明 |
|--------|---------|------|
| Table API Java（No.13） | 利用元 | Table操作からのプランナー呼び出し |
| SQLパーサー（No.18） | 前段処理 | SQL解析結果を受け取る |
| SQLクライアント（No.20） | 利用元 | SQL CLI経由のプランナー呼び出し |

### 6.2 使用例

```java
// TableEnvironment経由での暗黙的使用
TableEnvironment tEnv = TableEnvironment.create(...);

// EXPLAIN
String explanation = tEnv.explainSql("SELECT * FROM my_table",
    ExplainDetail.ESTIMATED_COST,
    ExplainDetail.CHANGELOG_MODE);

// クエリ実行（内部でtranslate呼び出し）
TableResult result = tEnv.executeSql("INSERT INTO sink SELECT * FROM source");

// コンパイル済プラン
CompiledPlan plan = tEnv.compilePlanSql("INSERT INTO sink SELECT * FROM source");
plan.writeToFile("/path/to/plan.json");

// プランの復元と実行
CompiledPlan loadedPlan = tEnv.loadPlan(PlanReference.fromFile("/path/to/plan.json"));
tEnv.executePlan(loadedPlan);
```

## 7. 設計上の考慮事項

### 7.1 拡張性

| 観点 | 説明 |
|------|------|
| TraitDef拡張 | 新しい特性定義の追加が可能 |
| Optimizer拡張 | カスタム最適化ルールの追加が可能 |
| ExecNodeGraphProcessor拡張 | 後処理プロセッサの追加が可能 |

### 7.2 パフォーマンス考慮事項

| 考慮点 | 説明 |
|--------|------|
| 共通サブグラフ最適化 | 複数クエリの共通部分を共有 |
| コストベース最適化 | 統計情報に基づく最適計画選択 |
| コード生成 | JITコンパイルによる高速実行 |

### 7.3 制約事項

| 制約 | 説明 |
|------|------|
| ランタイムモード | Streaming/Batchは実行時に変更不可 |
| コンパイル済プラン | バッチモードでは非対応 |
| EXPLAIN PLAN_ADVICE | バッチモードでは非対応 |

## 8. 用語集

| 用語 | 説明 |
|------|------|
| Planner | SQLプランナーインターフェース |
| PlannerBase | プランナー基底実装 |
| StreamPlanner | ストリーミングモードプランナー |
| BatchPlanner | バッチモードプランナー |
| Optimizer | 最適化エンジン |
| RelNode | 関係演算ノード（Calcite） |
| ExecNodeGraph | 実行ノードグラフ |
| Transformation | Flinkランタイム変換 |
| TraitDef | 関係特性定義 |
| ExecNodeGraphProcessor | 実行グラフ後処理プロセッサ |
| InternalPlan | コンパイル済内部プラン |
