# 機能設計書：Table API Java

## 1. 機能概要

### 1.1 処理概要

Table API Javaは、Apache Flinkにおけるリレーショナルクエリを記述するためのJava言語向けAPIである。SQLライクな操作を提供しつつ、Javaの型安全性を活かしたプログラミングモデルを実現する。

本機能は以下のコンポーネントで構成される：
1. **TableEnvironment**: Table APIプログラムのエントリーポイントと中央コンテキスト
2. **Table**: データ変換パイプラインを記述する中核抽象
3. **Expressions**: 式を構築するためのDSL（Domain Specific Language）

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

| 項目 | 内容 |
|------|------|
| 目的 | 宣言的なリレーショナルクエリによるデータ処理の簡素化 |
| 役割 | ストリーム/バッチ処理の統一的なAPIの提供 |
| 主な利用場面 | ETL処理、データ分析、リアルタイム集計、データ変換 |

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

1. **リアルタイムデータ分析**
   - ストリームデータに対するリアルタイム集計
   - ウィンドウ集約とグループ化

2. **ETLパイプライン**
   - 複数ソースからのデータ結合
   - フィルタリング、変換、集約の組み合わせ

3. **バッチ処理**
   - 大規模データセットの分析
   - 履歴データの集計レポート生成

4. **インタラクティブ分析**
   - アドホッククエリの実行
   - 結果のローカル取得と確認

## 2. 入出力仕様

### 2.1 入力仕様

#### 2.1.1 TableEnvironment作成

| パラメータ | 型 | 必須 | 説明 |
|-----------|-----|------|------|
| settings | EnvironmentSettings | No | 環境設定（ストリーム/バッチモード、プランナー設定等） |
| configuration | Configuration | No | 設定パラメータ |

```java
// 環境設定からの作成
TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.newInstance()
    .inStreamingMode()
    .build());

// Configurationからの作成
TableEnvironment tEnv = TableEnvironment.create(new Configuration());
```

#### 2.1.2 Tableの取得

| メソッド | 説明 |
|---------|------|
| from(String path) | カタログからテーブルを取得 |
| fromValues(Object... values) | 値からテーブルを作成 |
| fromValues(AbstractDataType<?> rowType, Object... values) | 型指定で値からテーブルを作成 |
| sqlQuery(String query) | SQLクエリ結果をTableとして取得 |

### 2.2 出力仕様

#### 2.2.1 Table操作結果

| 出力 | 型 | 説明 |
|------|-----|------|
| Table | Table | 変換後のテーブル |
| TableResult | TableResult | 実行結果（collect(), print()等） |
| StatementSet | StatementSet | 複数INSERT文のバッチ実行 |

## 3. 処理フロー

### 3.1 全体アーキテクチャ

```mermaid
flowchart TD
    subgraph API["Table API Layer"]
        TE[TableEnvironment]
        T[Table]
        EX[Expressions]
    end

    subgraph Internal["Internal Layer"]
        TEI[TableEnvironmentImpl]
        TI[TableImpl]
        P[Parser]
        PL[Planner]
    end

    subgraph Catalog["Catalog Layer"]
        CM[CatalogManager]
        FC[FunctionCatalog]
        MM[ModuleManager]
    end

    subgraph Execution["Execution Layer"]
        EX2[Executor]
        TR[Transformation]
        JOB[Job]
    end

    TE --> TEI
    T --> TI
    TEI --> CM
    TEI --> FC
    TEI --> MM
    TEI --> P
    TEI --> PL
    PL --> TR
    TR --> EX2
    EX2 --> JOB
```

### 3.2 Table変換フロー

```mermaid
sequenceDiagram
    participant User as ユーザーコード
    participant Table as Table
    participant Planner as Planner
    participant Optimizer as Optimizer
    participant Executor as Executor

    User->>Table: select(), filter(), join()等
    Table->>Table: QueryOperation生成
    User->>Table: execute()
    Table->>Planner: 実行計画生成
    Planner->>Optimizer: 最適化
    Optimizer->>Executor: Transformation生成
    Executor->>User: TableResult返却
```

### 3.3 Expression評価フロー

```mermaid
flowchart LR
    A["$('colA')"] --> B[ApiExpression]
    B --> C[UnresolvedRef]
    C --> D[ExpressionResolver]
    D --> E[ResolvedExpression]
    E --> F[CodeGen]
```

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

### 4.1 テーブル操作ルール

| カテゴリ | 操作 | 説明 |
|---------|------|------|
| 射影 | select() | カラム選択と式評価 |
| フィルタ | filter(), where() | 条件によるレコード絞り込み |
| 集約 | groupBy() + select() | グループ化と集計関数適用 |
| 結合 | join(), leftOuterJoin()等 | 複数テーブルの結合 |
| 集合 | union(), intersect(), minus() | 集合演算 |
| ソート | orderBy() | 結果のソート |
| 制限 | offset(), fetch(), limit() | 結果件数の制限 |

### 4.2 Expression DSLルール

```java
// カラム参照
$("columnName")  // または col("columnName")

// リテラル
lit(42)
lit("string")

// 演算
$("a").plus($("b"))
$("a").isEqual($("b"))
$("name").like("%pattern%")

// 集計
$("value").sum()
$("value").avg()
$("value").count()

// 関数呼び出し
call("myFunction", $("arg1"), $("arg2"))
```

### 4.3 結合条件

| 結合タイプ | メソッド | 備考 |
|-----------|---------|------|
| INNER JOIN | join() | フィールドは重複不可（asでリネーム必要） |
| LEFT OUTER JOIN | leftOuterJoin() | null check有効必須 |
| RIGHT OUTER JOIN | rightOuterJoin() | null check有効必須 |
| FULL OUTER JOIN | fullOuterJoin() | null check有効必須 |
| LATERAL JOIN | joinLateral() | TableFunctionとの結合 |

### 4.4 ウィンドウ操作

```java
// Tumbling Window
table.window(Tumble.over(lit(10).minutes()).on($("rowtime")).as("w"))
    .groupBy($("w"), $("key"))
    .select($("key"), $("value").sum());

// Sliding Window
table.window(Slide.over(lit(10).minutes()).every(lit(5).minutes()).on($("rowtime")).as("w"))
    .groupBy($("w"))
    .select($("value").avg());

// Session Window
table.window(Session.withGap(lit(10).minutes()).on($("rowtime")).as("w"))
    .groupBy($("w"), $("key"))
    .select($("key"), $("value").count());
```

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

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

```
TableEnvironment (インターフェース)
├── create(EnvironmentSettings) → TableEnvironmentImpl.create()
├── from(String) → カタログからテーブル取得
├── fromValues(...) → ValuesOperationファクトリ
├── sqlQuery(String) → Parser.parse() → QueryOperation
├── executeSql(String) → Parser.parse() → 各種Operation実行
├── createTemporaryTable() → CatalogManager登録
├── createTemporaryView() → QueryOperationCatalogView
├── createTemporarySystemFunction() → FunctionCatalog登録
└── createStatementSet() → StatementSet作成

Table (インターフェース)
├── select(Expression...) → ProjectOperation
├── filter(Expression) → FilterOperation
├── where(Expression) → FilterOperation（filterのエイリアス）
├── groupBy(Expression...) → GroupedTable
├── join(Table) → JoinOperation
├── leftOuterJoin(Table) → JoinOperation (LEFT OUTER)
├── union(Table) → SetOperation
├── orderBy(Expression...) → SortOperation
├── offset(int) → LimitOperation
├── fetch(int) → LimitOperation
├── window(GroupWindow) → GroupWindowedTable
├── execute() → TableResult
└── executeInsert(String) → TableResult

Expressions (ユーティリティクラス)
├── $(String) → ApiExpression (カラム参照)
├── col(String) → ApiExpression（$のエイリアス）
├── lit(Object) → ApiExpression (リテラル)
├── row(...) → ApiExpression (行コンストラクタ)
├── call(String, ...) → ApiExpression (関数呼び出し)
└── withAllColumns() → ApiExpression（全カラム選択）
```

### 5.2 データフロー図

```
[ユーザーAPI呼び出し]
        │
        ▼
[Table.select/filter/join等]
        │
        ▼
[QueryOperation生成]
        │
        ├──────────────────────────────┐
        │                              │
        ▼                              ▼
[Table.execute()]              [Table.executeInsert()]
        │                              │
        ▼                              ▼
[Planner.translate()]          [Planner.translate()]
        │                              │
        ▼                              ▼
[Transformation生成]           [Transformation生成]
        │                              │
        ▼                              ▼
[Executor.execute()]           [Executor.execute()]
        │                              │
        ▼                              ▼
[TableResult]                  [TableResult]
(collect/print)                (INSERT実行)
```

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

| ファイルパス | 役割 |
|-------------|------|
| `flink-table-api-java/.../TableEnvironment.java` | Table APIのエントリーポイントインターフェース |
| `flink-table-api-java/.../Table.java` | テーブル変換パイプラインの中核インターフェース |
| `flink-table-api-java/.../Expressions.java` | Expression DSLのユーティリティクラス |
| `flink-table-api-java/.../internal/TableEnvironmentImpl.java` | TableEnvironmentの実装クラス |
| `flink-table-api-java/.../ApiExpression.java` | Expression APIのラッパークラス |
| `flink-table-api-java/.../GroupedTable.java` | groupBy後のテーブル |
| `flink-table-api-java/.../WindowedTable.java` | ウィンドウ定義後のテーブル |
| `flink-table-api-java/.../StatementSet.java` | 複数ステートメントのバッチ実行 |
| `flink-table-common/.../ResolvedSchema.java` | 解決済みスキーマ |
| `flink-table-planner/.../Planner.java` | クエリ実行計画生成 |

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

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

```java
// flink-table-api-java/.../TableEnvironment.java (48-77行目)
/**
 * A table environment is the base class, entry point, and central context for creating Table and
 * SQL API programs.
 *
 * A table environment is responsible for:
 * - Connecting to external systems.
 * - Registering and retrieving Tables and other meta objects from a catalog.
 * - Executing SQL statements.
 * - Offering further configuration options.
 */
@PublicEvolving
public interface TableEnvironment {
    static TableEnvironment create(EnvironmentSettings settings) {
        return TableEnvironmentImpl.create(settings);
    }
}
```

#### Step 2: Tableインターフェースの理解

```java
// flink-table-api-java/.../Table.java (36-94行目)
/**
 * The Table object is the core abstraction of the Table API. Similar to how the DataStream
 * API has DataStreams, the Table API is built around Tables.
 *
 * A Table object describes a pipeline of data transformations. It does not contain the
 * data itself in any way.
 */
@PublicEvolving
public interface Table extends Explainable<Table>, Executable {
    ResolvedSchema getResolvedSchema();
    Table select(Expression... fields);
    Table filter(Expression predicate);
    // ...
}
```

#### Step 3: Expressionsユーティリティの理解

```java
// flink-table-api-java/.../Expressions.java (59-95行目)
/**
 * Entry point of the Table API Expression DSL such as: $("myField").plus(10).abs()
 */
@PublicEvolving
public final class Expressions {
    // カラム参照
    public static ApiExpression $(String name) {
        return new ApiExpression(unresolvedRef(name));
    }

    // リテラル
    public static ApiExpression lit(Object v) {
        return new ApiExpression(valueLiteral(v));
    }
}
```

#### Step 4: 主要な変換操作の確認

```java
// Table.java - select操作 (118-134行目)
Table select(Expression... fields);

// Table.java - filter操作 (189-204行目)
Table filter(Expression predicate);

// Table.java - join操作 (252-267行目)
Table join(Table right);

// Table.java - groupBy操作 (223-239行目)
GroupedTable groupBy(Expression... fields);

// Table.java - orderBy操作 (607-627行目)
Table orderBy(Expression... fields);
```

## 6. 関連機能・API

### 6.1 関連機能

| 機能名 | 関連種別 | 説明 |
|--------|---------|------|
| Table API Scala（No.14） | 言語バインディング | Scala言語向けTable API |
| 型変換（No.15） | ブリッジ機能 | DataStreamとの相互変換 |
| 組み込み関数（No.16） | 関数提供 | 標準関数の提供 |
| UDF（No.17） | 拡張機能 | ユーザー定義関数 |
| SQLパーサー（No.18） | SQL連携 | SQL文の解析 |
| SQLプランナー（No.19） | 実行計画 | クエリ最適化と実行計画生成 |

### 6.2 主要APIリファレンス

#### TableEnvironment

```java
// テーブル取得
Table from(String path)
Table fromValues(Object... values)
Table sqlQuery(String query)

// テーブル登録
void createTemporaryView(String path, Table table)
void createTemporaryTable(String path, TableDescriptor descriptor)

// 関数登録
void createTemporarySystemFunction(String name, Class<? extends UserDefinedFunction> functionClass)
void createTemporaryFunction(String path, UserDefinedFunction functionInstance)

// SQL実行
TableResult executeSql(String statement)

// カタログ操作
void useCatalog(String catalogName)
void useDatabase(String databaseName)
Optional<Catalog> getCatalog(String catalogName)

// ステートメントセット
StatementSet createStatementSet()
```

#### Table

```java
// 変換操作
Table select(Expression... fields)
Table filter(Expression predicate)
Table where(Expression predicate)
GroupedTable groupBy(Expression... fields)
Table distinct()

// 結合操作
Table join(Table right)
Table join(Table right, Expression joinPredicate)
Table leftOuterJoin(Table right, Expression joinPredicate)
Table rightOuterJoin(Table right, Expression joinPredicate)
Table fullOuterJoin(Table right, Expression joinPredicate)
Table joinLateral(Expression tableFunctionCall)

// 集合操作
Table union(Table right)
Table unionAll(Table right)
Table intersect(Table right)
Table minus(Table right)

// ソート・制限
Table orderBy(Expression... fields)
Table offset(int offset)
Table fetch(int fetch)
Table limit(int fetch)

// ウィンドウ
GroupWindowedTable window(GroupWindow groupWindow)

// 実行
TableResult execute()
TableResult executeInsert(String tablePath)
```

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

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

| 考慮点 | 説明 | 推奨対応 |
|--------|------|---------|
| プロジェクションプッシュダウン | 必要なカラムのみ選択 | select()で必要カラムを明示 |
| フィルタプッシュダウン | フィルタ条件を早期適用 | filter()を変換チェーン前半に配置 |
| 結合順序 | 小さいテーブルを左側に | 結合順序の最適化 |
| ウィンドウサイズ | メモリ使用量に影響 | 適切なウィンドウサイズ設定 |

### 7.2 型安全性

```java
// 型情報の明示的指定
Table result = tEnv.fromValues(
    DataTypes.ROW(
        DataTypes.FIELD("id", DataTypes.BIGINT()),
        DataTypes.FIELD("name", DataTypes.STRING())
    ),
    row(1L, "Alice"),
    row(2L, "Bob")
);
```

### 7.3 エラーハンドリング

| 例外 | 発生条件 | 対応 |
|------|---------|------|
| ValidationException | スキーマ不一致、無効な操作 | スキーマ確認、API使用方法確認 |
| TableException | 実行時エラー | ログ確認、リソース設定見直し |
| SqlParserException | SQL構文エラー | SQL文の修正 |

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

| 観点 | リスク | 対策 |
|------|--------|------|
| SQLインジェクション | 動的SQL構築時の脆弱性 | パラメータ化クエリの使用 |
| 認証情報 | コネクタ設定での平文保存 | 環境変数やシークレット管理の活用 |

## 8. 用語集

| 用語 | 説明 |
|------|------|
| TableEnvironment | Table APIプログラムのエントリーポイントと中央コンテキスト |
| Table | データ変換パイプラインを記述する抽象。SQLのビューに相当 |
| Expression | カラム参照、リテラル、関数呼び出しなどを表す式 |
| ApiExpression | ユーザー向けExpression APIのラッパークラス |
| QueryOperation | テーブル操作を表す内部表現 |
| ResolvedSchema | 解決済みの型情報を持つスキーマ |
| Catalog | テーブル、関数、データベースのメタデータストア |
| Planner | クエリの実行計画を生成するコンポーネント |
| StatementSet | 複数のINSERT文をバッチ実行するためのコンテナ |
