# 機能設計書 28-JDBCデータソース

## 概要

本ドキュメントは、Apache SparkにおけるJDBC経由での外部RDBMSへの読み書き機能の設計について記述する。複数のデータベース方言（MySQL, PostgreSQL, Oracle等）に対応し、分散並列読み取りを提供する。

### 本機能の処理概要

SparkのJDBCデータソースは、JDBC（Java Database Connectivity）標準インターフェースを通じて外部リレーショナルデータベースとの読み書きを提供する。JdbcDialectによるデータベース固有の型変換・SQL方言対応、パーティション分割による並列読み取り、述語プッシュダウンによる効率的なクエリ実行を実現する。

**業務上の目的・背景**：企業のデータは多くの場合RDBMSに格納されている。SparkのJDBCデータソースにより、これらの既存データ資産をSparkの分散処理エンジンで直接分析できる。JdbcDialect機構により、各データベースの特性に応じた最適な型マッピングとSQL生成を自動的に行う。

**機能の利用シーン**：RDBMSからのデータ読み取りと分析、ETLパイプラインでのRDBMSデータの取り込み、分析結果のRDBMSへの書き戻し、RDBMSテーブルのSpark SQLからの直接クエリ。

**主要な処理内容**：
1. JDBC接続の確立とコネクションプール管理（JdbcConnectionProvider）
2. JdbcDialectによるデータベース方言対応（型変換、SQL生成）
3. パーティション分割による並列JDBC読み取り
4. 述語プッシュダウンとカラムプルーニング
5. JdbcSQLQueryBuilderによるSELECTクエリ生成
6. JDBCを通じたデータ書き込み（バッチINSERT）
7. テーブル作成/削除等のDDL操作

**関連システム・外部連携**：MySQL, PostgreSQL, Oracle, SQL Server, DB2, Derby, H2, Teradata, Snowflake, Databricks等の各種RDBMS

**権限による制御**：JDBC接続時のデータベースユーザー認証による。JDBC URLにユーザー名/パスワードを指定、またはConnectionProvider経由でKerberos認証等を使用。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 22 | ThriftServer Overview（ThriftServer概要） | 参照画面 | JDBC経由でのSQL実行（ThriftServerはJDBC接続を受け付ける） |

## 機能種別

データ連携 / CRUD操作 / 外部システム連携

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| url | String | Yes | JDBC接続URL | 有効なJDBC URL形式 |
| dbtable / query | String | Yes | テーブル名またはサブクエリ | テーブルが存在すること |
| user | String | No | データベースユーザー名 | 有効なユーザー名 |
| password | String | No | データベースパスワード | - |
| driver | String | No | JDBCドライバクラス名 | クラスパスに存在 |
| partitionColumn | String | No | パーティションカラム | 数値型カラムであること |
| lowerBound | Long | No | パーティション下限値 | partitionColumnと同時指定 |
| upperBound | Long | No | パーティション上限値 | partitionColumnと同時指定 |
| numPartitions | Int | No | パーティション数 | 正の整数 |
| fetchsize | Int | No | JDBCフェッチサイズ | 正の整数 |
| batchsize | Int | No | 書き込みバッチサイズ（デフォルト: 1000） | 正の整数 |
| pushDownPredicate | Boolean | No | 述語プッシュダウン（デフォルト: true） | true/false |

### 入力データソース

- 外部RDBMS（JDBC接続経由）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| DataFrame | Dataset[Row] | JDBCテーブルから読み込まれたデータ |

### 出力先

- DataFrame/Dataset（読み取り結果）
- 外部RDBMS（書き込み結果）

## 処理フロー

### 処理シーケンス

```
1. JDBCOptions解析とドライバ登録
   └─ DriverRegistry.register()でJDBCドライバをロード
2. JdbcDialect検出
   └─ JDBC URLに基づいてデータベース固有のDialectを自動検出
3. 読み取り時: JdbcSQLQueryBuilder によるSELECTクエリ生成
   └─ カラムプルーニング、述語プッシュダウン、パーティション条件の適用
4. パーティション分割
   └─ partitionColumn/lowerBound/upperBound/numPartitionsに基づく範囲分割
5. 並列JDBC読み取り
   └─ 各パーティションが独立してJDBC接続を確立しデータ取得
6. 書き込み時: JdbcUtils.saveTable()
   └─ バッチINSERTによるデータ書き込み
```

### フローチャート

```mermaid
flowchart TD
    A[spark.read.jdbc] --> B[JDBCOptions解析]
    B --> C[JdbcDialect検出]
    C --> D{パーティション指定?}
    D -->|Yes| E[パーティション分割]
    D -->|No| F[単一パーティション]
    E --> G[JdbcSQLQueryBuilder]
    F --> G
    G --> H[述語プッシュダウン]
    H --> I[カラムプルーニング]
    I --> J[並列JDBC読み取り]
    J --> K[JdbcDialect.getCatalystType型変換]
    K --> L[DataFrame]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-28-01 | Dialect自動検出 | JDBC URLからデータベース種別を判定しDialectを選択 | JDBC接続時 |
| BR-28-02 | 並列読み取り | partitionColumn/lowerBound/upperBound/numPartitionsで範囲パーティション | パーティション指定時 |
| BR-28-03 | バッチ書き込み | batchsizeごとにバッチINSERTを実行 | 書き込み時 |
| BR-28-04 | 型マッピング | JdbcDialect.getCatalystType()でJDBC型をSpark型に変換 | 読み取り時 |
| BR-28-05 | SaveMode | Overwrite/Append/ErrorIfExists/Ignoreで書き込みモードを制御 | 書き込み時 |

### 計算ロジック

パーティション分割: partitionColumnの値域を[lowerBound, upperBound)としてnumPartitions個の等間隔範囲に分割する。各パーティションは`WHERE partitionColumn >= low AND partitionColumn < high`条件のSQLクエリを実行する。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| read | 外部RDBMSテーブル | SELECT | パーティション分割による並列読み取り |
| write (Append) | 外部RDBMSテーブル | INSERT | バッチINSERTによるデータ追加 |
| write (Overwrite) | 外部RDBMSテーブル | TRUNCATE + INSERT | テーブル切り詰め後にデータ挿入 |
| createTable | 外部RDBMS | CREATE TABLE | テーブル作成 |

### テーブル別操作詳細

#### 外部RDBMSテーブル

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| SELECT | プッシュダウンされたカラム | WHERE句にプッシュダウンされたフィルタ | JdbcSQLQueryBuilder生成 |
| INSERT | DataFrameのスキーマカラム | DataFrameの各行データ | batchsize毎にバッチ実行 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| CONNECTION_FAILURE | 接続エラー | JDBC接続の確立に失敗 | 接続URL/認証情報の確認 |
| TABLE_NOT_FOUND | テーブルエラー | 指定テーブルが存在しない | テーブル名の確認 |
| DRIVER_NOT_FOUND | ドライバエラー | JDBCドライバがクラスパスに存在しない | ドライバJARの配置確認 |
| TYPE_MAPPING_ERROR | 型変換エラー | JDBCの型がSpark型にマッピングできない | JdbcDialectの型定義確認 |
| BATCH_WRITE_FAILURE | 書き込みエラー | バッチINSERTの実行に失敗 | データ内容・制約の確認 |

### リトライ仕様

JDBC接続のリトライはConnectionProvider実装に委譲される。タスクレベルのリトライによりJDBC読み取りの再試行が行われる。

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

書き込み時は各パーティション（タスク）ごとにJDBCトランザクションが管理される。isolationLevel設定によりトランザクション分離レベルを制御可能。Overwriteモードではテーブルの切り詰めと挿入がトランザクション内で実行される。

## パフォーマンス要件

- パーティション分割による並列読み取りでスループット向上
- 述語プッシュダウンによるRDBMS側でのデータフィルタリング
- fetchsizeによるネットワークラウンドトリップの最適化
- batchsizeによるバッチINSERT最適化

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

- JDBC接続のユーザー名/パスワードによる認証
- JdbcConnectionProviderによるKerberos認証対応
- パスワードのマスク処理（Spark UIでの表示時）
- SSL/TLS暗号化接続のサポート（JDBC URL設定による）

## 備考

- JdbcDialectはServiceLoaderによる拡張が可能
- AggregatedDialectにより複数のDialectを組み合わせ可能

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | JdbcDialects.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala` | JdbcDialect基底クラス、JdbcType定義 |

**読解のコツ**: JdbcDialectsは@DeveloperApi注釈付きの拡張ポイント。getCatalystType()でJDBC型→Spark型、getJDBCType()でSpark型→JDBC型の変換を行う。62行目のJdbcType case classが型マッピングの基本構造。

**主要処理フロー**:
- **62行目**: JdbcType case class定義（databaseTypeDefinition, jdbcNullType）
- **64-80行目**: JdbcDialect抽象クラスのScaladoc - データベース方言管理の設計意図

#### Step 2: 各種データベースDialectを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | MySQLDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala` | MySQL固有の型変換・SQL生成 |
| 2-2 | PostgresDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala` | PostgreSQL固有の型変換・SQL生成 |
| 2-3 | OracleDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala` | Oracle固有の型変換・SQL生成 |

#### Step 3: JDBC接続とクエリ生成を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | JdbcConnectionProvider.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcConnectionProvider.scala` | JDBC接続プロバイダ |
| 3-2 | JdbcSQLQueryBuilder.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcSQLQueryBuilder.scala` | SQLクエリビルダー |

#### Step 4: JDBC操作ユーティリティを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | JdbcUtils.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JdbcUtils.scala` | JDBC操作ユーティリティ |
| 4-2 | JDBCOptions.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/jdbc/JDBCOptions.scala` | JDBCオプション定義 |

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

```
DataFrameReader.jdbc() / spark.read.format("jdbc")
    │
    ├─ JDBCOptions解析
    │      └─ DriverRegistry.register()
    │
    ├─ JdbcDialects.get(url)
    │      ├─ MySQLDialect
    │      ├─ PostgresDialect
    │      ├─ OracleDialect
    │      ├─ MsSqlServerDialect
    │      ├─ DB2Dialect
    │      ├─ DerbyDialect
    │      ├─ H2Dialect
    │      ├─ TeradataDialect
    │      ├─ SnowflakeDialect
    │      └─ DatabricksDialect
    │
    ├─ JdbcSQLQueryBuilder
    │      ├─ カラムプルーニング
    │      └─ 述語プッシュダウン
    │
    └─ JdbcUtils
           ├─ getSchema() - スキーマ取得
           ├─ readTable() - テーブル読み取り
           └─ saveTable() - テーブル書き込み
```

### データフロー図

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

JDBC URL/認証情報 ──▶ JdbcConnectionProvider ──▶ Connection
                          │
テーブル名/クエリ ──▶ JdbcSQLQueryBuilder ────▶ SELECTクエリ
                          │
外部RDBMS ────────▶ JdbcDialect型変換 ────────▶ DataFrame
                          │
DataFrame ─────────▶ JdbcUtils.saveTable ─────▶ 外部RDBMS
                          │                        (バッチINSERT)
                          └─ JdbcDialect.getJDBCType
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| JdbcDialects.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcDialects.scala` | ソース | JDBC方言基底クラス |
| MySQLDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/MySQLDialect.scala` | ソース | MySQL方言 |
| PostgresDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/PostgresDialect.scala` | ソース | PostgreSQL方言 |
| OracleDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/OracleDialect.scala` | ソース | Oracle方言 |
| MsSqlServerDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/MsSqlServerDialect.scala` | ソース | SQL Server方言 |
| DB2Dialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/DB2Dialect.scala` | ソース | DB2方言 |
| DerbyDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/DerbyDialect.scala` | ソース | Derby方言 |
| H2Dialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/H2Dialect.scala` | ソース | H2方言 |
| TeradataDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/TeradataDialect.scala` | ソース | Teradata方言 |
| SnowflakeDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/SnowflakeDialect.scala` | ソース | Snowflake方言 |
| DatabricksDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/DatabricksDialect.scala` | ソース | Databricks方言 |
| AggregatedDialect.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/AggregatedDialect.scala` | ソース | 複合方言 |
| JdbcConnectionProvider.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcConnectionProvider.scala` | ソース | 接続プロバイダ |
| JdbcSQLQueryBuilder.scala | `sql/core/src/main/scala/org/apache/spark/sql/jdbc/JdbcSQLQueryBuilder.scala` | ソース | SQLクエリビルダー |
