# 機能設計書 21-SQLゲートウェイ

## 概要

本ドキュメントは、Apache FlinkのSQLゲートウェイ機能（flink-sql-gateway モジュール）の設計仕様を記述する。SQLゲートウェイは、外部クライアントからFlinkに対してSQLクエリを実行するための統合インターフェースを提供するサービスである。

### 本機能の処理概要

SQLゲートウェイは、REST APIを通じてSQLクエリの実行、セッション管理、およびカタログ操作を可能にする中間層サービスである。

**業務上の目的・背景**：分散ストリーム処理システムであるFlinkに対して、標準的なSQL構文でクエリを投入する手段を提供することで、開発者やデータアナリストがFlinkを容易に利用できるようにする。従来はFlink CLIや独自のクライアントコードが必要であったが、SQLゲートウェイにより標準的なREST APIやJDBC接続でFlinkにアクセス可能となる。

**機能の利用シーン**：
- BIツールやデータ分析プラットフォームからFlinkへのSQL接続
- アドホッククエリの実行と結果取得
- マテリアライズドテーブルの定期リフレッシュ管理
- SQLスクリプトのアプリケーションモードでのデプロイ

**主要な処理内容**：
1. セッションの開始・終了・設定管理（SessionManager）
2. SQL文の実行とオペレーション管理（OperationManager）
3. クエリ結果の取得とストリーミング（ResultFetcher）
4. カタログ・データベース・テーブル・関数のメタデータ取得
5. マテリアライズドテーブルのリフレッシュトリガー
6. SQLスクリプトのクラスタへのデプロイ

**関連システム・外部連携**：
- REST APIクライアント（HTTPプロトコル経由）
- JDBCドライバ（flink-sql-jdbc-driver経由）
- Flinkクラスター（JobManager/TaskManager）
- Quartzスケジューラ（マテリアライズドテーブルの定期実行）

**権限による制御**：現時点では認証・認可の明示的な制御は実装されていないが、エンドポイントレベルでのアクセス制御が可能。

## 関連画面

本機能は画面機能マッピングには直接含まれていないが、以下の画面と関連する可能性がある。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | REST API経由でのSQLクエリ実行基盤として機能 |

## 機能種別

データ連携 / API提供 / セッション管理

## 入力仕様

### 入力パラメータ

#### セッション操作

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| sessionName | String | No | セッション名 | 任意文字列 |
| properties | Map<String, String> | No | セッション設定 | Flink設定キー |
| endpointVersion | EndpointVersion | No | エンドポイントバージョン | サポートされたバージョン |

#### SQL実行

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| sessionHandle | SessionHandle | Yes | セッション識別子 | UUID形式 |
| statement | String | Yes | SQL文 | 有効なSQL構文 |
| executionTimeoutMs | long | No | 実行タイムアウト | 非負整数 |
| executionConfig | Configuration | No | 実行設定 | Flink設定 |

#### 結果取得

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| sessionHandle | SessionHandle | Yes | セッション識別子 | UUID形式 |
| operationHandle | OperationHandle | Yes | オペレーション識別子 | UUID形式 |
| token | long | No | 結果取得位置トークン | 非負整数 |
| maxRows | int | No | 最大取得行数 | 正整数 |

### 入力データソース

- HTTPリクエスト（REST APIエンドポイント経由）
- セッション設定（メモリ内SessionContext）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| SessionHandle | SessionHandle | 作成されたセッションのハンドル |
| OperationHandle | OperationHandle | 実行されたオペレーションのハンドル |
| ResultSet | ResultSet | クエリ実行結果（スキーマ＋データ） |
| OperationInfo | OperationInfo | オペレーションステータス情報 |
| GatewayInfo | GatewayInfo | ゲートウェイのバージョン情報 |

### 出力先

- HTTPレスポンス（JSON形式）
- Flinkクラスター（ジョブサブミット時）

## 処理フロー

### 処理シーケンス

```
1. クライアントがREST APIにリクエスト送信
   └─ HttpリクエストをNettyサーバーで受信

2. RestHandlerがリクエストをディスパッチ
   └─ OpenSessionHandler, ExecuteStatementHandler等が処理

3. SqlGatewayServiceImplが実際の処理を実行
   └─ SessionManagerでセッション管理
   └─ OperationManagerでオペレーション管理

4. 結果をクライアントに返却
   └─ JSON形式でHTTPレスポンス送信
```

### フローチャート

```mermaid
flowchart TD
    A[REST APIリクエスト受信] --> B{リクエストタイプ判定}
    B -->|OpenSession| C[SessionManager.openSession]
    B -->|ExecuteStatement| D[OperationManager.submitOperation]
    B -->|FetchResults| E[OperationManager.fetchResults]
    B -->|CloseSession| F[SessionManager.closeSession]

    C --> G[Session作成]
    G --> H[SessionHandle返却]

    D --> I[Executor.executeStatement]
    I --> J[OperationHandle返却]

    E --> K[ResultFetcher.fetchResults]
    K --> L[ResultSet返却]

    F --> M[Session終了]
    M --> N[Acknowledge返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | セッション数上限 | SQL_GATEWAY_SESSION_MAX_NUMで設定した数以上のセッションは作成不可 | 新規セッション作成時 |
| BR-002 | セッションアイドルタイムアウト | SQL_GATEWAY_SESSION_IDLE_TIMEOUTを超えてアイドル状態のセッションは自動クローズ | 定期チェック時 |
| BR-003 | オペレーションステータス管理 | オペレーションはPENDING/RUNNING/FINISHED/ERROR/CANCELEDのいずれかの状態を持つ | オペレーション実行中 |

### 計算ロジック

特になし（SQLクエリの実行はFlink Table/SQLエンジンに委譲）

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

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

本機能は直接データベースを操作しない。SQLクエリの実行結果はFlinkジョブとして処理される。

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| SqlGatewayException | システムエラー | セッション/オペレーション操作失敗 | 詳細メッセージを確認し再試行 |
| SessionNotFoundException | ビジネスエラー | 存在しないセッションへのアクセス | 新規セッションを作成 |
| OperationNotFoundException | ビジネスエラー | 存在しないオペレーションへのアクセス | 有効なオペレーションハンドルを使用 |
| UnsupportedOperationException | 機能エラー | タイムアウト機構は未サポート | executionTimeoutMsに0以下を指定 |

### リトライ仕様

- セッション切断時は新規セッションを作成して再実行
- オペレーション失敗時は新規オペレーションとして再実行

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

SQLゲートウェイ自体はステートレスであり、トランザクション管理はFlinkジョブ実行レベルで行われる。

## パフォーマンス要件

- セッション作成: 100ms以内
- SQL実行（非同期サブミット）: 500ms以内
- 結果フェッチ: データ量に依存

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

- REST APIエンドポイントへのネットワークレベルでのアクセス制御を推奨
- 現時点では認証・認可機構は未実装
- 本番環境ではSSL/TLSによる通信暗号化を推奨

## 備考

- バッチモードクエリのみが完全サポート
- ストリーミングクエリは更新・削除結果の扱いに注意が必要

---

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

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

### 推奨読解順序

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

SQLゲートウェイで使用される主要なデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SessionHandle.java | `flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/session/SessionHandle.java` | セッション識別子の構造 |
| 1-2 | OperationHandle.java | `flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/operation/OperationHandle.java` | オペレーション識別子の構造 |
| 1-3 | ResultSet.java | `flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/results/ResultSet.java` | クエリ結果の構造 |
| 1-4 | SessionEnvironment.java | `flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/session/SessionEnvironment.java` | セッション環境設定 |

**読解のコツ**: UUIDベースのハンドルがセッションとオペレーションの識別に使用されている。ResultSetはスキーマ（ResolvedSchema）とデータ（List<RowData>）を保持する。

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

SQLゲートウェイの起動とREST APIエンドポイントの初期化を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SqlGateway.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/SqlGateway.java` | メインエントリーポイント |
| 2-2 | SqlGatewayRestEndpoint.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/SqlGatewayRestEndpoint.java` | REST APIエンドポイント定義 |

**主要処理フロー**:
1. **94-96行目**: main()メソッドでstartSqlGateway()を呼び出し
2. **99-147行目**: startSqlGateway()でDefaultContext読み込み、SessionManager作成、gateway.start()実行
3. **63-78行目**: start()でSessionManager起動とエンドポイント初期化

#### Step 3: サービスインターフェースを理解する

SqlGatewayServiceの役割と実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | SqlGatewayService.java | `flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/SqlGatewayService.java` | サービスインターフェース定義 |
| 3-2 | SqlGatewayServiceImpl.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java` | サービス実装 |

**主要処理フロー**:
- **76-83行目**: openSession - セッション作成
- **202-226行目**: executeStatement - SQL文実行
- **229-256行目**: fetchResults - 結果取得

#### Step 4: セッション管理を理解する

セッションのライフサイクル管理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | SessionManagerImpl.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/SessionManagerImpl.java` | セッション管理実装 |
| 4-2 | Session.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/Session.java` | セッションエンティティ |
| 4-3 | SessionContext.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/context/SessionContext.java` | セッションコンテキスト |

**主要処理フロー**:
- **81-115行目**: start() - セッションクリーンアップスケジューラとワーカースレッドプール初期化
- **142-167行目**: openSession() - 新規セッション作成

#### Step 5: オペレーション管理を理解する

SQL実行オペレーションの管理を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | OperationManager.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java` | オペレーション管理 |
| 5-2 | OperationExecutor.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java` | オペレーション実行 |
| 5-3 | ResultFetcher.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/result/ResultFetcher.java` | 結果取得 |

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

```
SqlGateway (main)
    │
    ├─ SessionManager.create()
    │      └─ SessionManagerImpl
    │             └─ operationExecutorService (ThreadPool)
    │
    ├─ SqlGatewayServiceImpl
    │      ├─ openSession() -> SessionManager.openSession()
    │      │                        └─ Session.open()
    │      │
    │      ├─ executeStatement() -> Session.getOperationManager()
    │      │                             └─ OperationManager.submitOperation()
    │      │                                    └─ OperationExecutor.executeStatement()
    │      │
    │      └─ fetchResults() -> OperationManager.fetchResults()
    │                                └─ ResultFetcher.fetchResults()
    │
    └─ SqlGatewayRestEndpoint
           ├─ OpenSessionHandler
           ├─ ExecuteStatementHandler
           ├─ FetchResultsHandler
           └─ CloseSessionHandler
```

### データフロー図

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

HTTPリクエスト ───────▶ SqlGatewayRestEndpoint ───▶ HTTPレスポンス
(JSON)                       │                        (JSON)
                             │
                             ▼
                    SqlGatewayServiceImpl
                             │
           ┌─────────────────┼─────────────────┐
           │                 │                 │
           ▼                 ▼                 ▼
    SessionManager    OperationManager   ResultFetcher
           │                 │                 │
           ▼                 ▼                 ▼
       Session        OperationExecutor    ResultStore
           │                 │                 │
           └────────────────▼──────────────────┘
                             │
                             ▼
                    Flink Table/SQL Engine
                             │
                             ▼
                      Flinkクラスター
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SqlGateway.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/SqlGateway.java` | ソース | メインエントリーポイント |
| SqlGatewayService.java | `flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/SqlGatewayService.java` | ソース | サービスインターフェース |
| SqlGatewayServiceImpl.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/SqlGatewayServiceImpl.java` | ソース | サービス実装 |
| SqlGatewayRestEndpoint.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/rest/SqlGatewayRestEndpoint.java` | ソース | REST APIエンドポイント |
| SessionManagerImpl.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/session/SessionManagerImpl.java` | ソース | セッション管理 |
| OperationManager.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java` | ソース | オペレーション管理 |
| OperationExecutor.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationExecutor.java` | ソース | オペレーション実行 |
| ResultFetcher.java | `flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/result/ResultFetcher.java` | ソース | 結果取得 |
| SqlGatewayServiceConfigOptions.java | `flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/config/SqlGatewayServiceConfigOptions.java` | ソース | 設定オプション定義 |
