# 機能設計書 52-S3（Presto）

## 概要

本ドキュメントは、Apache Flink の Amazon S3 ファイルシステム（Prestoベース）の機能設計書である。flink-s3-fs-presto モジュールが提供する S3 ストレージとの連携機能について、処理内容、設定方法、およびコードの構造を詳細に記載する。

### 本機能の処理概要

**業務上の目的・背景**：Apache Flink は複数の S3 ファイルシステム実装を提供している。Prestoベースの実装は、Facebook の Presto プロジェクトで開発された PrestoS3FileSystem を活用し、特定のユースケースにおいて Hadoop S3A とは異なる特性を持つ。軽量な実装が求められる場合や、Presto との互換性が重要な場合に選択される。

**機能の利用シーン**：
- 軽量なS3アクセスが必要な場合
- Prestoとの互換性が求められる場合
- s3p:// スキームを使用したい場合
- Hadoop S3A の代替が必要な場合

**主要な処理内容**：
1. S3FileSystemFactory による FileSystem インスタンスの生成
2. Presto S3 設定の読み込みとマッピング
3. PrestoS3FileSystem の初期化とラッピング
4. FlinkS3PrestoFileSystem によるカスタム削除処理（バグ回避）
5. 再帰的削除のカスタム実装

**関連システム・外部連携**：
- Amazon S3 / S3互換ストレージ
- Facebook Presto S3 ライブラリ
- AWS SDK for Java

**権限による制御**：AWS IAM によるアクセス制御。アクセスキー/シークレットキーまたは IAM ロールによる認証をサポート。

## 関連画面

本機能は画面を持たないバックエンド機能である。

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | - |

## 機能種別

データ連携 / ファイルシステム抽象化

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| presto.s3.access-key | String | Yes（環境変数未使用時） | AWSアクセスキー | 非空文字列 |
| presto.s3.secret-key | String | Yes（環境変数未使用時） | AWSシークレットキー | 非空文字列 |
| presto.s3.path-style-access | Boolean | No | パススタイルアクセスの有効化 | true/false |
| s3.entropy.key | String | No | エントロピー注入キー | 文字列 |
| s3.entropy.length | Integer | No | エントロピー文字数 | 正の整数 |

### 入力データソース

- Flink Configuration（flink-conf.yaml）
- 環境変数

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| FileSystem | org.apache.flink.core.fs.FileSystem | Flinkファイルシステムインターフェース実装 |

### 出力先

Flink ランタイムへの FileSystem インスタンス提供

## 処理フロー

### 処理シーケンス

```
1. FileSystemFactory.create(URI) が呼び出される
   └─ URIスキーム（s3:// または s3p://）を検証
2. HadoopConfigLoader で設定を読み込み
   └─ Flink設定からPresto設定へのマッピング
3. PrestoS3FileSystem インスタンスを生成
   └─ new PrestoS3FileSystem()
4. PrestoS3FileSystem.initialize(URI, Configuration) で初期化
   └─ デフォルトでs3.amazonaws.comを使用
5. FlinkS3PrestoFileSystem でラップして返却
   └─ カスタム削除処理の提供
```

### フローチャート

```mermaid
flowchart TD
    A[開始: create URI] --> B{URIスキーム確認}
    B -->|s3:// or s3p://| C[HadoopConfigLoader実行]
    B -->|その他| Z[エラー: 非対応スキーム]
    C --> D[PrestoS3FileSystem生成]
    D --> E[getInitURI でURI正規化]
    E --> F{authority有り?}
    F -->|Yes| G[fsUri使用]
    F -->|No| H[s3://s3.amazonaws.com使用]
    G --> I[PrestoS3FileSystem.initialize]
    H --> I
    I --> J[FlinkS3PrestoFileSystem生成]
    J --> K[終了: FileSystem返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-52-01 | スキームマッピング | s3:// と s3p:// を PrestoS3FileSystem にマッピング | 常時 |
| BR-52-02 | 設定プレフィックス | s3., presto.s3. プレフィックスの設定を読み込み | 設定読み込み時 |
| BR-52-03 | デフォルトエンドポイント | authority未指定時は s3.amazonaws.com を使用 | URI解決時 |
| BR-52-04 | 再帰的削除 | Prestoのバグ回避のためカスタム実装を使用 | delete(recursive=true)時 |

### 計算ロジック

特になし

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

本機能はデータベース操作を行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | 接続エラー | S3エンドポイントに接続できない | ネットワーク設定を確認 |
| AccessDeniedException | 認証エラー | 認証情報が無効または権限不足 | 認証設定を確認 |
| FlinkRuntimeException | URI構文エラー | 無効なURIが指定された | URIフォーマットを確認 |

### リトライ仕様

Presto S3 ファイルシステムの標準リトライポリシーに従う

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

オブジェクトストレージのため、トランザクションは非対応。

## パフォーマンス要件

- Prestoの軽量S3実装を使用
- RecoverableWriterはサポートしない（getS3AccessHelper()がnullを返す）

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

- 認証情報は環境変数での提供を推奨
- 設定ファイルへの認証情報直接記載は非推奨
- HTTPS接続の使用を推奨

## 備考

- Presto S3 ファイルシステムの再帰的削除にはバグがあり（GitHub issue #17416）、FlinkS3PrestoFileSystem でカスタム実装を提供
- RecoverableWriter は非サポート（S3AccessHelperがnull）

---

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

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

### 推奨読解順序

#### Step 1: ファクトリ実装を理解する

S3ファイルシステムファクトリの実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | S3FileSystemFactory.java | `flink-filesystems/flink-s3-fs-presto/src/main/java/org/apache/flink/fs/s3presto/S3FileSystemFactory.java` | ファクトリのメイン実装 |
| 1-2 | S3PFileSystemFactory.java | `flink-filesystems/flink-s3-fs-presto/src/main/java/org/apache/flink/fs/s3presto/S3PFileSystemFactory.java` | s3p:// スキーム用サブクラス |

**主要処理フロー**:
1. **40-46行目**: 設定プレフィックスとミラーキーの定義
2. **48-50行目**: コンストラクタでHadoopConfigLoaderを初期化
3. **53-55行目**: getScheme()で"s3"を返却
4. **68-87行目**: createFlinkFileSystem()でFlinkS3PrestoFileSystemを生成
5. **90-92行目**: createHadoopFileSystem()でPrestoS3FileSystemを生成
6. **95-108行目**: getInitURI()でデフォルトエンドポイント設定
7. **111-114行目**: getS3AccessHelper()でnullを返却（RecoverableWriter非サポート）

**読解のコツ**: getS3AccessHelper()がnullを返すことで、RecoverableWriterが使用できないことを理解する。

#### Step 2: カスタムファイルシステム実装を理解する

Prestoバグ回避のためのカスタム実装を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | FlinkS3PrestoFileSystem.java | `flink-filesystems/flink-s3-fs-presto/src/main/java/org/apache/flink/fs/s3presto/FlinkS3PrestoFileSystem.java` | カスタム削除処理 |

**主要処理フロー**:
- **39-40行目**: GitHub issue #17416 へのリファレンス（バグの説明）
- **63-72行目**: delete()メソッドのオーバーライド
- **74-109行目**: deleteRecursively()の再帰的削除実装
- **121-141行目**: deleteObject()の単一オブジェクト削除（存在確認付き）

**読解のコツ**: Prestoは空ディレクトリと削除エラーの区別ができないため、削除後に存在確認を行うワークアラウンドを理解する。

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

```
S3FileSystemFactory.create(URI)
    │
    ├─ createHadoopConfigLoader()
    │      └─ HadoopConfigLoader(prefixes, mirroredKeys, ...)
    │
    ├─ createHadoopFileSystem()
    │      └─ new PrestoS3FileSystem()
    │
    ├─ getInitURI(fsUri, hadoopConfig)
    │      └─ createURI("s3://s3.amazonaws.com") [if no authority]
    │
    ├─ PrestoS3FileSystem.initialize(URI, Configuration)
    │
    └─ createFlinkFileSystem(fs, ...)
           └─ new FlinkS3PrestoFileSystem(fs, ...)
                  │
                  ├─ delete(path, recursive)
                  │      └─ deleteRecursively(path) [if recursive]
                  │             └─ deleteObject(path)
                  │
                  └─ deleteObject(path)
                         └─ super.delete(path, false)
                         └─ exists(path) [error handling]
```

### データフロー図

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

URI (s3://bucket/path) ──▶ S3FileSystemFactory ──────▶ FlinkS3PrestoFileSystem
                               │
Flink Configuration ────────▶ HadoopConfigLoader ───▶ Presto Configuration
                               │
AWS Credentials ────────────▶ PrestoS3FileSystem ───▶ S3 API Calls
                               │
delete(recursive=true) ─────▶ FlinkS3PrestoFileSystem
                               │
                          deleteRecursively() ───────▶ 個別オブジェクト削除
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| S3FileSystemFactory.java | `flink-filesystems/flink-s3-fs-presto/src/main/java/org/apache/flink/fs/s3presto/S3FileSystemFactory.java` | ソース | s3://スキーム用ファクトリ |
| S3PFileSystemFactory.java | `flink-filesystems/flink-s3-fs-presto/src/main/java/org/apache/flink/fs/s3presto/S3PFileSystemFactory.java` | ソース | s3p://スキーム用ファクトリ |
| FlinkS3PrestoFileSystem.java | `flink-filesystems/flink-s3-fs-presto/src/main/java/org/apache/flink/fs/s3presto/FlinkS3PrestoFileSystem.java` | ソース | カスタムファイルシステム |
| HadoopConfigLoader.java | `flink-filesystems/flink-hadoop-fs/src/main/java/org/apache/flink/runtime/util/HadoopConfigLoader.java` | ソース | 設定ローダー |
| AbstractS3FileSystemFactory.java | `flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/AbstractS3FileSystemFactory.java` | ソース | 基底ファクトリクラス |
| org.apache.flink.core.fs.FileSystemFactory | `flink-filesystems/flink-s3-fs-presto/src/main/resources/META-INF/services/` | 設定 | SPIサービス登録 |
