# 機能設計書 78-PyFlink Table

## 概要

本ドキュメントは、Apache Flink の PyFlink Table API に関する機能設計書である。flink-python モジュールで提供される、Python から Flink の Table API および SQL を利用するためのラッパーライブラリについて記載する。

### 本機能の処理概要

PyFlink Table API は、Python から Flink の Java Table API を透過的に利用するためのラッパーレイヤーである。SQL と表形式データ操作を Python から利用可能にし、宣言的なデータ処理パイプラインの構築を支援する。

**業務上の目的・背景**：
- Python ユーザーが SQL でストリーム/バッチ処理を利用可能にする
- 宣言的なデータ変換 API の提供
- カタログ・メタデータ管理機能の提供

**機能の利用シーン**：
- SQL によるデータ変換・分析
- ETL パイプラインの構築
- Python UDF を含むテーブル処理

**主要な処理内容**：
1. TableEnvironment による実行環境・カタログ管理
2. Table による宣言的変換処理（select, filter, join 等）
3. SQL クエリの実行
4. Python UDF（スカラ関数、テーブル関数、集約関数）の登録と実行

**関連システム・外部連携**：
- Py4J（Python-Java ブリッジ）
- カタログ（Hive, JDBC 等）
- コネクタ（Kafka, FileSystem 等）
- PyFlink DataStream API

**権限による制御**：カタログレベルでの権限管理

## 関連画面

本機能はバックエンドライブラリであり、直接関連する画面はない。

## 機能種別

クライアントライブラリ / Table API・SQL

## 入力仕様

### TableEnvironment 設定パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| environment_settings | EnvironmentSettings/Configuration | Yes | 環境設定 | - |

### Table 変換メソッドパラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| fields | Expression | Yes (select) | 選択フィールド | - |
| predicate | Expression[bool] | Yes (filter/where) | フィルタ条件 | - |
| right | Table | Yes (join) | 結合テーブル | - |
| join_predicate | Expression[bool] | No (join) | 結合条件 | - |
| path | str | Yes (from_path) | テーブルパス | SQL識別子形式 |

### 入力データソース

- from_path(): カタログ登録テーブル
- from_descriptor(): TableDescriptor からのテーブル
- sql_query(): SQL クエリ結果

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Table | Table | 変換後のテーブル |
| TableResult | TableResult | SQL 実行結果 |
| StatementSet | StatementSet | 複数 DML のバッチ実行 |

### 出力先

- execute_insert(): テーブルシンク
- execute(): TableResult として取得
- to_data_stream(): DataStream への変換

## 処理フロー

### 処理シーケンス

```
1. 環境構築
   └─ TableEnvironment.create(EnvironmentSettings)
   └─ Py4J 経由で Java TableEnvironment 取得

2. カタログ・テーブル設定
   └─ create_catalog() / use_catalog() / use_database()
   └─ create_temporary_table() / create_temporary_view()
   └─ from_path() / from_descriptor()

3. テーブル変換処理定義
   └─ select() / filter() / where()
   └─ join() / left_outer_join() / full_outer_join()
   └─ group_by() / window() / order_by()

4. UDF 登録
   └─ create_temporary_function() / create_temporary_system_function()
   └─ udf() / udtf() / udaf() / udtaf()

5. 実行
   └─ execute() / execute_insert() / execute_sql()
   └─ TableResult 取得
```

### フローチャート

```mermaid
flowchart TD
    subgraph Python [Python Layer]
        A[TableEnvironment]
        B[Table]
        C[GroupedTable]
        D[WindowedTable]
        E[StatementSet]
    end

    subgraph Py4J [Py4J Bridge]
        F[Java Gateway]
    end

    subgraph Java [Java Layer]
        G[J_TableEnvironment]
        H[J_Table]
        I[Catalog]
        J[Planner]
    end

    A -->|wrap| F
    F -->|create| G
    B -->|wrap| F
    F -->|transform| H
    G -->|manage| I
    G -->|use| J
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-78-01 | パス形式 | [[catalog.]database.]table 形式 | from_path/create_temporary_view |
| BR-78-02 | 一時オブジェクト | 一時オブジェクトは永続オブジェクトをシャドウ可能 | create_temporary_* |
| BR-78-03 | テーブル同一環境 | 結合するテーブルは同一 TableEnvironment | join系メソッド |
| BR-78-04 | UDF 決定性 | is_deterministic() で結果の決定性を宣言 | UDF 登録時 |

### 計算ロジック

特になし

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

カタログ経由でのメタデータ管理を行う。

### カタログ操作

| 操作 | メソッド | 説明 |
|-----|---------|------|
| カタログ作成 | create_catalog() | CatalogDescriptor で作成 |
| カタログ取得 | get_catalog() | 名前でカタログ取得 |
| カタログ切替 | use_catalog() | デフォルトカタログ設定 |
| DB切替 | use_database() | デフォルトDB設定 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| CatalogException | カタログエラー | カタログ/DBが存在しない | パスを確認 |
| ValidationException | 検証エラー | 重複名でモジュール登録 | 名前を変更 |
| TableException | テーブルエラー | テーブルが見つからない | 登録を確認 |
| AttributeError | 属性エラー | 存在しないカラム参照 | スキーマを確認 |

### リトライ仕様

ライブラリ側でのリトライ機能はない。

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

本機能はトランザクション管理を行わない。

## パフォーマンス要件

- Planner による最適化
- StatementSet による複数 DML のバッチ実行
- Python UDF は Apache Beam ランナーで実行

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

- カタログ経由での権限管理
- Python UDF コードのシリアライズと転送

## 備考

- Table API と SQL は相互に変換可能
- DataStream API との相互変換サポート

---

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

### 推奨読解順序

#### Step 1: 実行環境を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | table_environment.py | `flink-python/pyflink/table/table_environment.py` | TableEnvironment ラッパー |

**主要処理フロー（table_environment.py）**:
- **64-95行目**: クラス定義と Javadoc - TableEnvironment の役割説明
- **97-105行目**: コンストラクタ - Java オブジェクトのラップ
- **107-126行目**: create() - 静的ファクトリメソッド
- **128-137行目**: create_catalog() - カタログ作成
- **154-166行目**: get_catalog() - カタログ取得
- **168-179行目**: load_module() - モジュールロード
- **236-276行目**: create_temporary_system_function() - Python UDF 登録
- **432-462行目**: create_temporary_table() - 一時テーブル作成
- **495-529行目**: from_path() - テーブル読み込み
- **560-567行目**: list_catalogs() - カタログ一覧
- **600-608行目**: list_tables() - テーブル一覧
- **815-836行目**: sql_query() - SQL クエリ実行
- **838-853行目**: execute_sql() - SQL 文実行
- **855-866行目**: create_statement_set() - StatementSet 作成
- **878-925行目**: use_catalog() - カタログ切替
- **937-984行目**: use_database() - DB 切替
- **998-1099行目**: create_temporary_view() - 一時ビュー作成

#### Step 2: Table 変換を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | table.py | `flink-python/pyflink/table/table.py` | Table ラッパー |

**主要処理フロー（table.py）**:
- **49-101行目**: Table クラス定義と Javadoc
- **103-105行目**: コンストラクタ - _j_table と _t_env 保持
- **110-123行目**: __getattr__() - カラムアクセス（tab.a 形式）
- **125-139行目**: select() - 選択操作
- **141-157行目**: alias() - フィールド名変更
- **159-172行目**: filter() - フィルタ操作
- **189-202行目**: group_by() - グループ化
- **204-215行目**: distinct() - 重複除去
- **217-242行目**: join() - 内部結合
- **244-271行目**: left_outer_join() - 左外部結合
- **298-321行目**: full_outer_join() - 完全外部結合
- **416-435行目**: minus() - 差集合
- **459-476行目**: union() - 和集合
- **497-517行目**: intersect() - 積集合
- **541-557行目**: order_by() - ソート
- **559-580行目**: offset() - オフセット
- **582-605行目**: fetch() - 行数制限
- **632-669行目**: window() - グループウィンドウ
- **671-699行目**: over_window() - オーバーウィンドウ

#### Step 3: UDF を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | udf.py | `flink-python/pyflink/table/udf.py` | UDF 定義 |

**主要処理フロー（udf.py）**:
- **34-68行目**: FunctionContext - UDF コンテキスト
- **70-106行目**: UserDefinedFunction 基底クラス - open/close/is_deterministic
- **108-122行目**: ScalarFunction - スカラ UDF
- **125-139行目**: TableFunction - テーブル UDF
- **146-199行目**: ImperativeAggregateFunction - 集約 UDF 基底

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

```
TableEnvironment (Python)
    │
    ├─ create(EnvironmentSettings)
    │      └─ gateway.jvm.TableEnvironment.create()
    │
    ├─ from_path(path)
    │      └─ j_tenv.from(path)
    │             └─ Table(j_table, self)
    │
    ├─ Table
    │      │
    │      ├─ select(*fields)
    │      │      └─ j_table.select(to_expression_jarray(fields))
    │      │
    │      ├─ filter(predicate)
    │      │      └─ j_table.filter(_get_java_expression(predicate))
    │      │
    │      ├─ join(right, predicate)
    │      │      └─ j_table.join(right._j_table, predicate)
    │      │
    │      └─ group_by(*fields)
    │             └─ GroupedTable(j_table.groupBy())
    │
    ├─ sql_query(query)
    │      └─ j_tenv.sqlQuery(query)
    │             └─ Table(j_table, self)
    │
    ├─ execute_sql(stmt)
    │      └─ j_tenv.executeSql(stmt)
    │             └─ TableResult(j_result)
    │
    └─ create_temporary_function(name, func)
           └─ func._java_user_defined_function()
           └─ j_tenv.createTemporaryFunction(name, java_func)
```

### データフロー図

```
[Python Application]                [Py4J Gateway]              [Java Runtime]

        │                                │                           │
        │   TableEnvironment.create()    │                           │
        └───────────────────────────────▶│                           │
                                         │   create(settings)        │
                                         └──────────────────────────▶│
                                                                     │
                                         ◀───────────────────────────┤
        TableEnvironment                 │                           │
        ◀────────────────────────────────┤                           │
        │                                │                           │
        │   from_path("table")           │                           │
        └───────────────────────────────▶│                           │
                                         │   from("table")           │
                                         └──────────────────────────▶│
                                                                     │
                                                          Catalog lookup
                                                                     │
                                         ◀───────────────────────────┤
        Table                            │                           │
        ◀────────────────────────────────┤                           │
        │                                │                           │
        │   select() / filter() / join() │                           │
        └───────────────────────────────▶│                           │
                                         │  transform operations     │
                                         └──────────────────────────▶│
                                                                     │
                                                              Planner
                                                          (optimization)
                                                                     │
                                         ◀───────────────────────────┤
        Table                            │                           │
        ◀────────────────────────────────┤                           │
        │                                │                           │
        │   execute_sql(INSERT INTO ...) │                           │
        └───────────────────────────────▶│                           │
                                         │   executeSql()            │
                                         └──────────────────────────▶│
                                                                     │
                                                              Job Execution
                                                                     │
                                         ◀───────────────────────────┤
        TableResult                      │                           │
        ◀────────────────────────────────┤                           │
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| table_environment.py | `flink-python/pyflink/table/table_environment.py` | ソース | TableEnvironment ラッパー |
| table.py | `flink-python/pyflink/table/table.py` | ソース | Table ラッパー |
| udf.py | `flink-python/pyflink/table/udf.py` | ソース | UDF 定義 |
| environment_settings.py | `flink-python/pyflink/table/environment_settings.py` | ソース | 環境設定 |
| table_config.py | `flink-python/pyflink/table/table_config.py` | ソース | テーブル設定 |
| table_descriptor.py | `flink-python/pyflink/table/table_descriptor.py` | ソース | テーブル記述子 |
| table_result.py | `flink-python/pyflink/table/table_result.py` | ソース | 実行結果 |
| statement_set.py | `flink-python/pyflink/table/statement_set.py` | ソース | 複数 DML バッチ |
| catalog.py | `flink-python/pyflink/table/catalog.py` | ソース | カタログ |
| schema.py | `flink-python/pyflink/table/schema.py` | ソース | スキーマ定義 |
| types.py | `flink-python/pyflink/table/types.py` | ソース | データ型 |
| expressions.py | `flink-python/pyflink/table/expressions.py` | ソース | 式ヘルパー |
| expression.py | `flink-python/pyflink/table/expression.py` | ソース | 式クラス |
| window.py | `flink-python/pyflink/table/window.py` | ソース | ウィンドウ定義 |
| functions.py | `flink-python/pyflink/table/functions.py` | ソース | 組み込み関数 |
| module.py | `flink-python/pyflink/table/module.py` | ソース | モジュール |
