# 機能設計書：型変換（Table API / DataStream API Bridge）

## 1. 機能概要

### 1.1 処理概要

型変換機能は、Apache FlinkにおいてTable APIとDataStream APIの間でデータを相互変換するためのブリッジ機能である。`StreamTableEnvironment`を通じて、DataStreamからTableへ、またはTableからDataStreamへの双方向変換を実現する。

主要な変換メソッド：
1. **fromDataStream()**: DataStream → Table（追加専用ストリーム）
2. **toDataStream()**: Table → DataStream（追加専用ストリーム）
3. **fromChangelogStream()**: DataStream<Row> → Table（変更ログストリーム）
4. **toChangelogStream()**: Table → DataStream<Row>（変更ログストリーム）

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

| 項目 | 内容 |
|------|------|
| 目的 | Table APIとDataStream APIの統合使用を可能にする |
| 役割 | 両APIの強みを活かしたハイブリッドなデータ処理パイプラインの構築 |
| 主な利用場面 | DataStream処理後のSQL集計、Table処理後の低レベル操作 |

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

1. **DataStream処理後のSQL集計**
   - DataStreamで前処理した後、SQLで集計
   - 複雑なウィンドウ処理とSQL結合の組み合わせ

2. **Table処理後の低レベル操作**
   - SQL/Table APIで変換後、ProcessFunctionで詳細処理
   - 外部システムとのカスタム連携

3. **変更ログ処理**
   - CDC（Change Data Capture）ストリームの処理
   - 更新・削除を含むデータの処理

## 2. 入出力仕様

### 2.1 入力仕様

#### 2.1.1 StreamTableEnvironment作成

```java
// StreamExecutionEnvironmentから作成
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

// EnvironmentSettings指定
StreamTableEnvironment tEnv = StreamTableEnvironment.create(
    env,
    EnvironmentSettings.newInstance()
        .inStreamingMode()
        .build());
```

#### 2.1.2 fromDataStream パラメータ

| パラメータ | 型 | 必須 | 説明 |
|-----------|-----|------|------|
| dataStream | DataStream<T> | Yes | 変換元DataStream |
| schema | Schema | No | カスタムスキーマ定義 |

#### 2.1.3 fromChangelogStream パラメータ

| パラメータ | 型 | 必須 | 説明 |
|-----------|-----|------|------|
| dataStream | DataStream<Row> | Yes | 変更ログストリーム |
| schema | Schema | No | カスタムスキーマ定義 |
| changelogMode | ChangelogMode | No | 変更種別の指定 |

### 2.2 出力仕様

#### 2.2.1 toDataStream 出力

| メソッド | 出力型 | 説明 |
|---------|-------|------|
| toDataStream(Table) | DataStream<Row> | Row型DataStream |
| toDataStream(Table, Class<T>) | DataStream<T> | 指定クラスのDataStream |
| toDataStream(Table, AbstractDataType<?>) | DataStream<T> | 指定データ型のDataStream |

#### 2.2.2 toChangelogStream 出力

| メソッド | 出力型 | 説明 |
|---------|-------|------|
| toChangelogStream(Table) | DataStream<Row> | RowKind付きDataStream |
| toChangelogStream(Table, Schema) | DataStream<Row> | スキーマ指定付き |
| toChangelogStream(Table, Schema, ChangelogMode) | DataStream<Row> | ChangelogMode指定付き |

## 3. 処理フロー

### 3.1 fromDataStream 変換フロー

```mermaid
sequenceDiagram
    participant DS as DataStream<T>
    participant STE as StreamTableEnvironment
    participant TI as TypeInformation<T>
    participant DT as DataType
    participant Table as Table

    DS->>STE: fromDataStream(dataStream)
    STE->>TI: 型情報抽出
    TI->>DT: DataType変換
    Note over DT: CompositeType → フラット化<br/>その他 → RAW型
    DT->>Table: Table生成
    Table-->>STE: Table返却
```

### 3.2 toDataStream 変換フロー

```mermaid
sequenceDiagram
    participant Table as Table
    participant STE as StreamTableEnvironment
    participant Planner as Planner
    participant Trans as Transformation
    participant DS as DataStream<T>

    Table->>STE: toDataStream(table, Class)
    STE->>Planner: 変換実行計画生成
    Note over Planner: 追加専用チェック
    Planner->>Trans: Transformation生成
    Trans->>DS: DataStream生成
    DS-->>STE: DataStream<T>返却
```

### 3.3 ChangelogStream 変換フロー

```mermaid
flowchart TD
    subgraph FromChangelog["fromChangelogStream"]
        DS1[DataStream<Row>] --> |RowKind評価| TE1[Table]
        TE1 --> |INSERT, UPDATE, DELETE| Table1[Table]
    end

    subgraph ToChangelog["toChangelogStream"]
        Table2[Table] --> |変更ログ生成| DS2[DataStream<Row>]
        DS2 --> |RowKind付与| Output[Output]
    end
```

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

### 4.1 型変換ルール

| TypeInformation | 変換後DataType | 備考 |
|-----------------|---------------|------|
| CompositeType | 第1レベルでフラット化 | Tuple, Row, POJO等 |
| 未サポート型 | RAW型 | 黒箱として扱う |
| プリミティブ型 | 対応DataType | INT → INT等 |

### 4.2 RowKind セマンティクス

| RowKind | 意味 | fromDataStream | fromChangelogStream |
|---------|------|----------------|---------------------|
| INSERT | 追加 | 必須 | 対応 |
| UPDATE_BEFORE | 更新前 | 非対応 | 対応 |
| UPDATE_AFTER | 更新後 | 非対応 | 対応 |
| DELETE | 削除 | 非対応 | 対応 |

### 4.3 ChangelogMode 種別

```java
// 追加専用
ChangelogMode.insertOnly()

// 全変更対応（デフォルト）
ChangelogMode.all()

// Upsertモード（UPDATE_BEFOREなし）
ChangelogMode.upsert()
```

### 4.4 スキーマカスタマイズ

```java
// カラムの型オーバーライド
Schema.newBuilder()
    .column("f1", "DECIMAL(10, 2)")
    .column("f0", "STRING")
    .build()

// タイムスタンプとウォーターマーク
Schema.newBuilder()
    .columnByMetadata("rowtime", "TIMESTAMP_LTZ(3)")
    .watermark("rowtime", "SOURCE_WATERMARK()")
    .build()

// 計算カラム
Schema.newBuilder()
    .columnByExpression("c1", "f1 + 42")
    .build()
```

### 4.5 制約事項

| 制約 | toDataStream | toChangelogStream |
|------|-------------|-------------------|
| 更新テーブル | 非対応（例外発生） | 対応 |
| 削除レコード | 非対応 | 対応 |
| 主キー必要性 | 不要 | Upsertモードで必要 |

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

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

```
StreamTableEnvironment (インターフェース)
├── create(StreamExecutionEnvironment) → StreamTableEnvironmentImpl.create()
├── create(StreamExecutionEnvironment, EnvironmentSettings)
│
├── fromDataStream(DataStream<T>) → Table
├── fromDataStream(DataStream<T>, Schema) → Table
│   └── 型情報抽出 → スキーマ生成 → SourceTransformation
│
├── fromChangelogStream(DataStream<Row>) → Table
├── fromChangelogStream(DataStream<Row>, Schema) → Table
├── fromChangelogStream(DataStream<Row>, Schema, ChangelogMode) → Table
│   └── RowKind評価 → 変更ログ処理 → Table
│
├── toDataStream(Table) → DataStream<Row>
├── toDataStream(Table, Class<T>) → DataStream<T>
├── toDataStream(Table, AbstractDataType<?>) → DataStream<T>
│   └── 追加専用チェック → Transformation → DataStream
│
├── toChangelogStream(Table) → DataStream<Row>
├── toChangelogStream(Table, Schema) → DataStream<Row>
├── toChangelogStream(Table, Schema, ChangelogMode) → DataStream<Row>
│   └── 変更ログ生成 → RowKind付与 → DataStream
│
└── createTemporaryView(String, DataStream<T>) → void
    └── fromDataStream → registerCatalogView
```

### 5.2 データフロー図

```
[DataStream API]                    [Table API]
     │                                   │
     │ fromDataStream()                  │
     ├──────────────────────────────────>│
     │                                   │
     │           SQL/Table変換           │
     │                                   │
     │ toDataStream()                    │
     │<──────────────────────────────────┤
     │                                   │

変更ログストリーム:
[DataStream<Row>]                  [Table]
     │ (RowKind: INSERT,              │
     │  UPDATE_BEFORE,                │
     │  UPDATE_AFTER,                 │
     │  DELETE)                       │
     │ fromChangelogStream()          │
     ├──────────────────────────────>│
     │                                │
     │ toChangelogStream()            │
     │<──────────────────────────────┤
```

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

| ファイルパス | 役割 |
|-------------|------|
| `flink-table-api-java-bridge/.../StreamTableEnvironment.java` | ブリッジAPIインターフェース |
| `flink-table-api-java-bridge/.../internal/StreamTableEnvironmentImpl.java` | ブリッジAPI実装 |
| `flink-table-common/.../Schema.java` | スキーマ定義クラス |
| `flink-table-common/.../ChangelogMode.java` | 変更モード定義 |
| `flink-types/.../Row.java` | Row型定義 |
| `flink-types/.../RowKind.java` | 行種別列挙型 |

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

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

```java
// StreamTableEnvironment.java (46-66行目)
/**
 * This table environment is the entry point and central context for creating Table and SQL API
 * programs that integrate with the Java-specific DataStream API.
 *
 * A stream table environment is responsible for:
 * - Convert a DataStream into Table and vice-versa.
 * - Connecting to external systems.
 * - Registering and retrieving Tables and other meta objects from a catalog.
 */
@PublicEvolving
public interface StreamTableEnvironment extends TableEnvironment {
```

#### Step 2: fromDataStreamの動作理解

```java
// StreamTableEnvironment.java (122-147行目)
/**
 * Column names and types of the Table are automatically derived from the
 * TypeInformation of the DataStream. If the outermost record's TypeInformation
 * is a CompositeType, it will be flattened in the first level.
 *
 * Since the DataStream API does not support changelog processing natively,
 * this method assumes append-only/insert-only semantics.
 */
<T> Table fromDataStream(DataStream<T> dataStream);
```

#### Step 3: toDataStreamの制約理解

```java
// StreamTableEnvironment.java (389-399行目)
/**
 * Since the DataStream API does not support changelog processing natively,
 * this method assumes append-only/insert-only semantics during the
 * table-to-stream conversion. The records of class Row will always
 * describe RowKind.INSERT changes.
 * Updating tables are not supported by this method and will produce an exception.
 */
```

#### Step 4: ChangelogStream変換の理解

```java
// StreamTableEnvironment.java (239-264行目)
/**
 * Converts the given DataStream of changelog entries into a Table.
 * Compared to fromDataStream, this method consumes instances of Row
 * and evaluates the RowKind flag that is contained in every record during runtime.
 */
Table fromChangelogStream(DataStream<Row> dataStream);
```

## 6. 関連機能・API

### 6.1 関連機能

| 機能名 | 関連種別 | 説明 |
|--------|---------|------|
| Table API Java（No.13） | 前提機能 | Table APIの基盤 |
| DataStream変換（No.7） | 前提機能 | DataStream APIの基盤 |
| Watermark（No.10） | 連携機能 | タイムスタンプ・ウォーターマーク伝播 |

### 6.2 使用例

```java
// DataStream → Table
DataStream<Tuple2<String, Integer>> stream = ...;
Table table = tEnv.fromDataStream(stream);

// スキーマ指定付き変換
Table tableWithSchema = tEnv.fromDataStream(stream,
    Schema.newBuilder()
        .columnByExpression("total", "$f1 * 2")
        .build());

// Table → DataStream
DataStream<Row> result = tEnv.toDataStream(table);

// 型指定変換
DataStream<User> users = tEnv.toDataStream(table, User.class);

// 変更ログストリーム処理
DataStream<Row> changelog = ...;  // RowKind付き
Table changelogTable = tEnv.fromChangelogStream(changelog);

// 変更ログ出力
DataStream<Row> changes = tEnv.toChangelogStream(table);
```

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

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

| 考慮点 | 説明 | 推奨対応 |
|--------|------|---------|
| 型変換オーバーヘッド | 毎レコードで型変換が発生 | POJOやRow型を使用 |
| シリアライゼーション | 境界でシリアライゼーション発生可能 | 同一型を維持 |
| 追加専用制約 | toDataStreamは追加専用のみ | toChangelogStreamを検討 |

### 7.2 制約事項

1. **toDataStreamの制約**
   - 更新テーブルは変換不可
   - 削除レコードは出力不可
   - RowKindは常にINSERT

2. **型導出の制約**
   - CompositeTypeは第1レベルのみフラット化
   - 未サポート型はRAW型として扱われる
   - ネストフィールドはアクセス不可

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

| 観点 | リスク | 対策 |
|------|--------|------|
| 型安全性 | 型キャスト失敗 | 明示的な型指定 |
| データ整合性 | RowKind不整合 | ChangelogMode明示 |

## 8. 用語集

| 用語 | 説明 |
|------|------|
| StreamTableEnvironment | DataStreamとTableの相互変換を提供する環境 |
| fromDataStream | DataStreamからTableへの変換メソッド |
| toDataStream | TableからDataStreamへの変換メソッド |
| ChangelogStream | RowKindフラグ付きの変更ログストリーム |
| RowKind | 行の変更種別（INSERT, UPDATE_BEFORE, UPDATE_AFTER, DELETE） |
| ChangelogMode | 変更ログストリームで許可される変更種別の定義 |
| CompositeType | 複合型（Tuple, Row, POJO等） |
| RAW型 | Flinkが直接扱えない型を黒箱として扱う型 |
