# 機能設計書 23-データソースV2 API

## 概要

本ドキュメントは、Apache SparkのデータソースV2 APIの機能設計について記述する。データソースV2 APIは、プラグイン可能なデータソースインターフェースを提供し、外部データストアとの読み書きを標準化されたAPIで実現する。

### 本機能の処理概要

データソースV2 APIは、Spark SQL Data Sources API V2（DSv2）として、従来のV1 APIの制限を克服するために設計された新しいデータソースインターフェースである。カタログ統合、カラムプルーニング、述語プッシュダウン、マイクロバッチストリーミング対応等の高度な機能をプラグイン可能な形式で提供する。

**業務上の目的・背景**：V1 APIでは、データソースの実装が複雑で、カタログ統合やストリーミング対応が困難であった。V2 APIは、Table、ScanBuilder、WriteBuilder等の明確なインターフェース階層を提供し、データソース開発者が段階的に高度な機能を実装できるようにする。また、物理的なスキャン・書き込み処理の分離により、オプティマイザとの連携が改善される。

**機能の利用シーン**：カスタムデータソースの開発、外部データストア（クラウドストレージ、NoSQL DB、メッセージキュー等）との統合、ストリーミングソース/シンクの実装、マルチカタログ対応のデータアクセス。

**主要な処理内容**：
1. TableProviderインターフェースによるテーブルメタデータの提供
2. ScanBuilderによる読み取りスキャンの構築（カラムプルーニング、フィルタプッシュダウン対応）
3. WriteBuilderによる書き込み処理の構築（バッチ・ストリーミング対応）
4. CatalogPluginによるマルチカタログ対応
5. BatchScanExecによるバッチスキャンの物理実行
6. DataSourceRDDによる分散読み取りの実行
7. WriteToDataSourceV2Execによるデータ書き込みの物理実行

**関連システム・外部連携**：各種外部データストア（HDFS, S3, JDBC, Kafka等）、CatalogManager（マルチカタログ管理）、Catalystオプティマイザ（プッシュダウン最適化）

**権限による制御**：CatalogPlugin実装によりカタログレベルでのアクセス制御が可能。テーブル操作はTableCatalogインターフェースを通じて権限チェックを委譲できる。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 14 | SQL All Executions（SQL実行一覧） | 結果表示画面 | V2データソースを使用したSQL実行の表示 |
| 15 | SQL Execution Detail（SQL実行詳細） | 結果表示画面 | V2データソーススキャンの物理計画表示 |

## 機能種別

データ連携 / プラグインフレームワーク / CRUD操作

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| TableProvider実装クラス | Class | Yes | データソースプロバイダのクラス名 | ServiceLoaderで検出可能であること |
| table | Identifier | Yes | テーブル識別子 | 有効なカタログ.スキーマ.テーブル形式 |
| schema | StructType | No | ユーザー指定スキーマ | 有効なStructType |
| options | CaseInsensitiveStringMap | No | データソースオプション | データソース固有 |

### 入力データソース

- 外部データストア（TableProvider実装による）
- カタログメタストア（CatalogPlugin経由）
- Hadoop互換ファイルシステム（FileDataSourceV2経由）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| InputPartition[] | Array[InputPartition] | 読み取りパーティション配列 |
| InternalRow | Iterator[InternalRow] | スキャン結果の行データ |
| WriterCommitMessage | WriterCommitMessage | 書き込みコミットメッセージ |

### 出力先

- DataFrame/Dataset（読み取り結果）
- 外部データストア（書き込み処理）

## 処理フロー

### 処理シーケンス

```
1. DataSourceV2Strategyによる論理計画から物理計画への変換
   └─ V2テーブルスキャン/書き込みのSparkPlanノード生成
2. ScanBuilder構築とプッシュダウン最適化
   └─ SupportsPushDownFilters, SupportsPushDownRequiredColumns等の適用
3. InputPartitionの計画
   └─ Scan.planInputPartitions()による分散パーティションの決定
4. PartitionReaderFactoryによるリーダー生成
   └─ 各パーティションに対するPartitionReader生成
5. DataSourceRDDによる分散読み取り実行
   └─ 各Executorでのパーティションデータ読み取り
```

### フローチャート

```mermaid
flowchart TD
    A[SQL/DataFrame操作] --> B[論理計画生成]
    B --> C[DataSourceV2Strategy]
    C --> D{読み取り/書き込み?}
    D -->|読み取り| E[ScanBuilder構築]
    E --> F[プッシュダウン最適化]
    F --> G[Scan.planInputPartitions]
    G --> H[BatchScanExec実行]
    H --> I[DataSourceRDD]
    D -->|書き込み| J[WriteBuilder構築]
    J --> K[BatchWrite生成]
    K --> L[WriteToDataSourceV2Exec]
    L --> M[DataWritingSparkTask]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-23-01 | フォールバック | V2データソースがサポートしない操作はV1にフォールバック | V1FallbackWriters適用時 |
| BR-23-02 | プッシュダウン | SupportsPushDown*インターフェースに応じて最適化が適用される | スキャン構築時 |
| BR-23-03 | パーティショニング | InputPartitionの数がスキャンの並列度を決定 | planInputPartitions時 |
| BR-23-04 | コミットプロトコル | 全DataWriterの成功後にBatchWrite.commit()が呼ばれる | 書き込み完了時 |

### 計算ロジック

プッシュダウンされたフィルタと必要カラムに基づき、データソース側で読み取りデータを制限する。これにより不要なデータ転送を削減し、I/O効率を向上させる。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| scan | V2テーブル | SELECT | InputPartition経由でのデータ読み取り |
| write | V2テーブル | INSERT/UPDATE/DELETE | BatchWrite/StreamingWrite経由での書き込み |
| DDL | V2テーブル | CREATE/ALTER/DROP | TableCatalog経由でのメタデータ操作 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TABLE_OR_VIEW_NOT_FOUND | 分析エラー | テーブルがカタログに存在しない | テーブル名とカタログの確認 |
| DATA_SOURCE_NOT_FOUND | 構成エラー | TableProviderが見つからない | プロバイダクラスのクラスパス確認 |
| WRITE_ABORTED | 書き込みエラー | DataWriter.write()で例外発生 | BatchWrite.abort()が呼ばれる |

### リトライ仕様

書き込み失敗時はタスクレベルのリトライが適用される。abort()呼び出し後のクリーンアップはデータソース実装に委譲される。

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

BatchWriteはcommit/abortセマンティクスを持つ。全てのDataWriterTaskが成功した場合にcommit()、いずれかが失敗した場合にabort()が呼ばれる。

## パフォーマンス要件

- カラムプルーニングとフィルタプッシュダウンによるI/O削減
- パーティション単位の並列読み取り
- ColumnarBatch対応によるベクトル化処理

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

- CatalogPluginによるカタログレベルのアクセス制御
- データソース接続時の認証情報管理はプロバイダ実装に委譲

## 備考

- Spark 2.3でV2 APIの初版が導入、Spark 3.0で大幅に拡張された
- V1データソースとの後方互換性はFallBackFileSourceV2等で維持

---

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

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

### 推奨読解順序

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

V2 APIのインターフェース階層を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | TableProvider.java | `sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableProvider.java` | データソースプロバイダの基本インターフェース |
| 1-2 | Table.java | `sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java` | テーブルインターフェース |
| 1-3 | Scan.java | `sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/Scan.java` | スキャンインターフェース |

**読解のコツ**: V2 APIはJavaインターフェースとして定義されている。SupportsRead, SupportsWrite等のmixinインターフェースで機能を段階的に追加する設計パターンを理解することが重要。

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

DataSourceV2StrategyがV2テーブル操作を物理計画に変換する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | DataSourceV2Strategy.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala` | V2テーブルの論理計画から物理計画への変換ルール |

#### Step 3: 読み取り処理を理解する

BatchScanExecとDataSourceRDDによるデータ読み取り。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | BatchScanExec.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala` | バッチスキャン物理実行ノード |
| 3-2 | DataSourceRDD.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala` | 分散読み取りRDD |
| 3-3 | FileScanBuilder.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala` | ファイルベーススキャンビルダー |

#### Step 4: 書き込み処理を理解する

WriteToDataSourceV2Execによるデータ書き込み。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | WriteToDataSourceV2Exec.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala` | 書き込み物理実行ノード |
| 4-2 | V2Writes.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala` | V2書き込みルール |

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

```
DataSourceV2Strategy
    │
    ├─ 読み取り系
    │      ├─ BatchScanExec
    │      │      ├─ Scan.planInputPartitions()
    │      │      └─ DataSourceRDD
    │      │             └─ PartitionReader.next()/get()
    │      │
    │      └─ FileDataSourceV2
    │             └─ FileScanBuilder
    │
    └─ 書き込み系
           ├─ WriteToDataSourceV2Exec
           │      ├─ BatchWrite.createBatchWriterFactory()
           │      └─ DataWritingSparkTask
           │             └─ DataWriter.write()/commit()
           │
           └─ V1FallbackWriters（V1互換）
```

### データフロー図

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

外部データストア ────▶ TableProvider ──────▶ Table
                          │                    │
                          ▼                    ▼
                    ScanBuilder           WriteBuilder
                          │                    │
                          ▼                    ▼
                    InputPartition[]      BatchWrite
                          │                    │
                          ▼                    ▼
                    DataSourceRDD        DataWriter ────▶ 外部データストア
                          │
                          ▼
                    DataFrame/Dataset
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| DataSourceV2Strategy.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala` | ソース | V2テーブル物理計画変換 |
| BatchScanExec.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/BatchScanExec.scala` | ソース | バッチスキャン実行 |
| DataSourceRDD.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceRDD.scala` | ソース | 分散読み取りRDD |
| WriteToDataSourceV2Exec.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/WriteToDataSourceV2Exec.scala` | ソース | 書き込み実行 |
| FileDataSourceV2.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileDataSourceV2.scala` | ソース | ファイルベースV2データソース |
| FileScanBuilder.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/FileScanBuilder.scala` | ソース | ファイルスキャンビルダー |
| V2Writes.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2Writes.scala` | ソース | V2書き込みルール |
| V1FallbackWriters.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V1FallbackWriters.scala` | ソース | V1フォールバック |
