# 機能設計書 22-JDBCドライバ

## 概要

本ドキュメントは、Apache FlinkのJDBCドライバ機能（flink-sql-jdbc-driver モジュール）の設計仕様を記述する。JDBCドライバは、標準的なJDBC APIを通じてFlinkのSQLゲートウェイに接続し、SQLクエリを実行するためのクライアントライブラリである。

### 本機能の処理概要

FlinkJDBCドライバは、java.sql.Driverインターフェースを実装し、標準的なJDBC接続を通じてFlink SQLゲートウェイにアクセスする機能を提供する。

**業務上の目的・背景**：既存のJDBCベースのツール（DBeaver、DataGrip、JasperReports等）やアプリケーションからFlinkにアクセスできるようにすることで、Flinkの導入障壁を下げる。開発者は既存のJDBCスキルを活用してFlinkのストリーム/バッチ処理にアクセスでき、BIツールとの連携も容易になる。

**機能の利用シーン**：
- SQLクライアントツール（DBeaver、DataGrip、Squirrel SQL等）からのFlink接続
- Javaアプリケーションからの標準JDBC APIによるクエリ実行
- レポーティングツールからのデータ取得
- ETLツールからのFlink連携

**主要な処理内容**：
1. JDBC URL解析とSQLゲートウェイへの接続確立（FlinkDriver, DriverUri）
2. Connection、Statement、ResultSetの標準JDBCインターフェース実装
3. SQLクエリの実行とResult Setの取得（FlinkStatement, FlinkResultSet）
4. DatabaseMetaDataによるカタログ・スキーマ・テーブル情報の取得
5. 接続プロパティと設定の管理

**関連システム・外部連携**：
- Flink SQLゲートウェイ（REST API経由）
- JDBCクライアントツール
- Javaアプリケーション

**権限による制御**：JDBCドライバ自体は認証機構を持たない。SQLゲートウェイ側のセキュリティ設定に依存。

## 関連画面

本機能は画面機能マッピングには直接含まれていない。JDBCクライアントツールの画面から利用される。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | JDBCクライアントツールからのFlink接続 |

## 機能種別

データ連携 / API提供 / ドライバ

## 入力仕様

### 入力パラメータ

#### 接続URL

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| url | String | Yes | JDBC接続URL | jdbc:flink://で開始 |
| host | String | Yes | SQLゲートウェイホスト | 有効なホスト名/IP |
| port | int | Yes | SQLゲートウェイポート | 1-65535 |
| catalog | String | No | 初期カタログ | URLパス内で指定 |
| database | String | No | 初期データベース | URLパス内で指定 |

#### 接続プロパティ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| properties | Properties | No | 接続プロパティ | キー=値形式 |

#### SQL実行

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| sql | String | Yes | 実行するSQL文 | 有効なFlink SQL |

### 入力データソース

- JDBC接続URL（文字列）
- 接続プロパティ（Properties）
- SQL文（文字列）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| Connection | java.sql.Connection | データベース接続 |
| Statement | java.sql.Statement | SQL文実行用ステートメント |
| ResultSet | java.sql.ResultSet | クエリ結果セット |
| ResultSetMetaData | java.sql.ResultSetMetaData | 結果セットのメタデータ |
| DatabaseMetaData | java.sql.DatabaseMetaData | データベースメタデータ |

### 出力先

- JDBCクライアントアプリケーション（オブジェクト参照）

## 処理フロー

### 処理シーケンス

```
1. DriverManager.getConnection(url, properties)
   └─ FlinkDriver.connect() が呼び出される

2. FlinkDriver がDriverUri を解析
   └─ ホスト、ポート、カタログ、データベースを抽出

3. FlinkConnection を作成
   └─ Executor を初期化してSQLゲートウェイに接続

4. FlinkStatement でSQL実行
   └─ executor.executeStatement() でクエリ実行

5. FlinkResultSet で結果取得
   └─ StatementResult から行データを読み取り
```

### フローチャート

```mermaid
flowchart TD
    A[DriverManager.getConnection] --> B[FlinkDriver.connect]
    B --> C[DriverUri.create]
    C --> D{URL解析成功?}
    D -->|Yes| E[FlinkConnection作成]
    D -->|No| F[SQLException]

    E --> G[Executor初期化]
    G --> H[SQLゲートウェイ接続]

    H --> I[Connection返却]

    I --> J[createStatement]
    J --> K[FlinkStatement作成]

    K --> L[executeQuery/execute]
    L --> M[executor.executeStatement]
    M --> N[FlinkResultSet作成]
    N --> O[結果返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-001 | バッチモード専用 | バッチモードクエリのみ完全サポート | 全クエリ実行時 |
| BR-002 | 非スレッドセーフ | Connection、Statementはスレッドセーフではない | マルチスレッド環境 |
| BR-003 | URL形式 | jdbc:flink://host:port[/catalog[/database]][?properties] | 接続時 |
| BR-004 | プロパティ重複禁止 | URLとPropertiesで同じキーは指定不可 | 接続時 |

### 計算ロジック

特になし（SQLクエリの実行はSQLゲートウェイに委譲）

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

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

本機能は直接データベースを操作しない。SQLゲートウェイ経由でFlinkジョブとして処理される。

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

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| SQLException | システムエラー | URL解析失敗 | URL形式を確認 |
| SQLException | システムエラー | 接続失敗 | ホスト・ポートを確認 |
| SQLException | ビジネスエラー | SELECTでない文にexecuteQuery使用 | execute()を使用 |
| SQLFeatureNotSupportedException | 機能エラー | 未サポート機能の呼び出し | サポート機能を使用 |

### リトライ仕様

- 接続失敗時は再接続を試行
- クエリ失敗時は新しいStatementで再実行

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

JDBCドライバはトランザクションをサポートしない。autoCommitは常にtrue。getTransactionIsolation()はTRANSACTION_NONEを返す。

## パフォーマンス要件

- 接続確立: SQLゲートウェイのセッション作成時間に依存
- クエリ実行: SQLゲートウェイの実行時間に依存

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

- JDBCドライバ自体は認証機構を持たない
- SQLゲートウェイとの通信はHTTP（非暗号化）がデフォルト
- 本番環境ではネットワークレベルでのセキュリティ対策を推奨

## 備考

- FlinkのINSERT文はジョブIDを結果セットとして返す
- ストリーミングクエリを実行する場合、更新・削除の結果に注意が必要
- PreparedStatementは未サポート

---

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

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

### 推奨読解順序

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

JDBCドライバで使用される主要なデータ構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | DriverUri.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/DriverUri.java` | JDBC URLの解析ロジック |
| 1-2 | DriverInfo.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/DriverInfo.java` | ドライババージョン情報 |
| 1-3 | ColumnInfo.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/ColumnInfo.java` | カラム情報の構造 |

**読解のコツ**: DriverUriはjdbc:flink://形式のURLを解析し、ホスト・ポート・カタログ・データベース・プロパティを抽出する。URLのパス部分は/catalog/databaseの形式。

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

JDBCドライバの登録と接続確立のフローを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | FlinkDriver.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkDriver.java` | Driverインターフェース実装 |

**主要処理フロー**:
1. **42-50行目**: staticブロックでDriverManagerへ自動登録
2. **53-55行目**: connect()でDriverUri解析とFlinkConnection作成
3. **58-60行目**: acceptsURL()でjdbc:flink:プレフィックスチェック

#### Step 3: Connection実装を理解する

FlinkConnectionの構造と主要メソッドを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | FlinkConnection.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkConnection.java` | Connection実装 |
| 3-2 | BaseConnection.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/BaseConnection.java` | Connection基底クラス |

**主要処理フロー**:
- **56-69行目**: コンストラクタでExecutor初期化、カタログ・データベース設定
- **79-84行目**: createStatement()でFlinkStatement作成
- **134-138行目**: getMetaData()でFlinkDatabaseMetaData取得
- **110-127行目**: close()でStatement・Executor終了

#### Step 4: Statement実装を理解する

FlinkStatementの構造とSQL実行ロジックを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | FlinkStatement.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkStatement.java` | Statement実装 |
| 4-2 | BaseStatement.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/BaseStatement.java` | Statement基底クラス |

**主要処理フロー**:
- **57-67行目**: executeQuery()でSELECTクエリ実行
- **121-131行目**: execute()で任意SQL実行
- **133-138行目**: executeInternal()で実際の実行（executor.executeStatement）

#### Step 5: ResultSet実装を理解する

FlinkResultSetの構造と結果取得ロジックを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 5-1 | FlinkResultSet.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkResultSet.java` | ResultSet実装 |
| 5-2 | BaseResultSet.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/BaseResultSet.java` | ResultSet基底クラス |
| 5-3 | FlinkResultSetMetaData.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkResultSetMetaData.java` | ResultSetMetaData実装 |

#### Step 6: DatabaseMetaData実装を理解する

データベースメタデータ取得機能を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 6-1 | FlinkDatabaseMetaData.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkDatabaseMetaData.java` | DatabaseMetaData実装 |
| 6-2 | DatabaseMetaDataUtils.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/utils/DatabaseMetaDataUtils.java` | メタデータ取得ユーティリティ |

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

```
DriverManager.getConnection(url, properties)
    │
    └─ FlinkDriver.connect(url, properties)
           │
           ├─ DriverUri.create(url, properties)
           │      └─ URL解析（host, port, catalog, database, properties）
           │
           └─ new FlinkConnection(DriverUri)
                  │
                  ├─ Executor.create()
                  │      └─ SQLゲートウェイへの接続
                  │
                  ├─ createStatement()
                  │      └─ new FlinkStatement(connection)
                  │             │
                  │             ├─ executeQuery(sql)
                  │             │      └─ executor.executeStatement(sql)
                  │             │             └─ new FlinkResultSet(result)
                  │             │
                  │             └─ execute(sql)
                  │                    └─ executor.executeStatement(sql)
                  │
                  └─ getMetaData()
                         └─ new FlinkDatabaseMetaData(url, connection, statement)
```

### データフロー図

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

JDBC URL ─────────────▶ FlinkDriver.connect ─────────▶ Connection
                              │
                              ▼
                        DriverUri.create
                              │
                    ┌─────────┴─────────┐
                    ▼                   ▼
              host:port           catalog/database
                    │                   │
                    └─────────┬─────────┘
                              │
                              ▼
                      FlinkConnection
                              │
                              ▼
                         Executor
                              │
                              ▼
                    SQLゲートウェイ(REST API)


SQL文 ─────────────────▶ FlinkStatement.execute ─────▶ ResultSet
                              │                            │
                              ▼                            ▼
                    executor.executeStatement    FlinkResultSet
                              │                     (イテレータ)
                              ▼
                       StatementResult
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| FlinkDriver.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkDriver.java` | ソース | JDBCドライバメインクラス |
| DriverUri.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/DriverUri.java` | ソース | JDBC URL解析 |
| FlinkConnection.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkConnection.java` | ソース | Connection実装 |
| FlinkStatement.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkStatement.java` | ソース | Statement実装 |
| FlinkResultSet.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkResultSet.java` | ソース | ResultSet実装 |
| FlinkDatabaseMetaData.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkDatabaseMetaData.java` | ソース | DatabaseMetaData実装 |
| FlinkResultSetMetaData.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/FlinkResultSetMetaData.java` | ソース | ResultSetMetaData実装 |
| BaseConnection.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/BaseConnection.java` | ソース | Connection基底クラス |
| BaseStatement.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/BaseStatement.java` | ソース | Statement基底クラス |
| BaseResultSet.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/BaseResultSet.java` | ソース | ResultSet基底クラス |
| DriverInfo.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/DriverInfo.java` | ソース | ドライババージョン情報 |
| DriverUtils.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/utils/DriverUtils.java` | ソース | ユーティリティ関数 |
| DatabaseMetaDataUtils.java | `flink-table/flink-sql-jdbc-driver/src/main/java/org/apache/flink/table/jdbc/utils/DatabaseMetaDataUtils.java` | ソース | メタデータユーティリティ |
