# 機能設計書 47-Protobuf

## 概要

本ドキュメントは、Apache Flink の flink-protobuf モジュールが提供する Protocol Buffers シリアライゼーション/デシリアライゼーション機能の設計を記述する。

### 本機能の処理概要

Protobuf フォーマット機能は、Google Protocol Buffers 形式のデータを読み書きするためのシリアライザとデシリアライザを提供する。コード生成（codegen）アプローチにより、高性能な変換処理を実現する。

**業務上の目的・背景**：Protocol Buffers は Google が開発したバイナリシリアライゼーションフォーマットである。スキーマ定義（.proto ファイル）に基づく型安全なデータ交換が可能で、JSON や XML と比較して高速かつコンパクトなデータ表現を提供する。マイクロサービス間通信やストリーミングデータ処理で広く使用される。

**機能の利用シーン**：
- gRPC 通信のデータ処理
- Kafka メッセージのシリアライゼーション
- マイクロサービス間のデータ交換
- 高性能ストリーム処理
- 型安全なデータパイプライン

**主要な処理内容**：
1. PbRowDataDeserializationSchema: Protobuf バイト列から RowData への変換
2. PbRowDataSerializationSchema: RowData から Protobuf バイト列への変換
3. ProtoToRowConverter: コード生成によるデシリアライザ
4. RowToProtoConverter: コード生成によるシリアライザ
5. PbSchemaValidationUtils: スキーマ検証

**関連システム・外部連携**：
- Kafka コネクタ
- gRPC サービス
- ファイルシステムコネクタ

**権限による制御**：本機能はデータフォーマット処理のため、権限制御は利用するコネクタに委譲される。

## 関連画面

本機能はバックエンドのデータフォーマット処理機能であり、直接関連する画面はない。

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

## 機能種別

- データ変換処理
- シリアライゼーション/デシリアライゼーション

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| message-class-name | String | Yes | Protobuf メッセージクラスの完全修飾名 | クラスパスに存在すること |
| ignore-parse-errors | boolean | No | パースエラー無視 | デフォルト: false |
| read-default-values | boolean | No | デフォルト値読み取り | デフォルト: false |
| write-null-string-literal | String | No | null 文字列リテラル | デフォルト: "" |

### 入力データソース

- デシリアライゼーション: Protobuf バイナリ形式の byte[] データ
- シリアライゼーション: Flink の RowData 内部データ構造

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| rowData | RowData | Flink 内部データ形式に変換されたデータ |
| pbBytes | byte[] | Protobuf 形式にシリアライズされたバイト列 |

### 出力先

- デシリアライゼーション: Flink の下流オペレーター
- シリアライゼーション: 外部システム（Kafka 等）

## 処理フロー

### フローチャート

```mermaid
flowchart TD
    subgraph Deserialization
        A[Protobuf byte[]] --> B[PbRowDataDeserializationSchema]
        B --> C[ProtoToRowConverter]
        C --> D[Generated Code]
        D --> E[RowData]
    end

    subgraph Serialization
        F[RowData] --> G[PbRowDataSerializationSchema]
        G --> H[RowToProtoConverter]
        H --> I[Generated Code]
        I --> J[Protobuf byte[]]
    end
```

### データフロー図

```mermaid
flowchart LR
    subgraph CodeGeneration
        Proto[.proto Schema] --> Desc[Descriptors.Descriptor]
        Desc --> Gen[Code Generator]
        Gen --> Class[Generated Class]
    end

    subgraph Runtime
        Input[byte[]] --> Parse[parseFrom]
        Parse --> Msg[Message Object]
        Msg --> Decode[Generated decode Method]
        Decode --> Row[RowData]
    end
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-47-01 | パースエラー処理 | ignore-parse-errors=true でエラー行を null として返却 | デシリアライゼーション時 |
| BR-47-02 | デフォルト値読み取り | read-default-values=true でフィールド未設定時にデフォルト値を使用 | デシリアライゼーション時 |
| BR-47-03 | proto3 プリミティブ | proto3 ではプリミティブ型は常にデフォルト値 | proto3 使用時 |
| BR-47-04 | null 文字列リテラル | write-null-string-literal で配列/マップ内の null 表現を指定 | シリアライゼーション時 |
| BR-47-05 | スキーマ検証 | クライアント側でスキーマ整合性を事前検証 | 初期化時 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| IOException | パースエラー | Protobuf 形式不正 | ignore-parse-errors 設定確認 |
| FlinkRuntimeException | シリアライゼーションエラー | RowData 変換失敗 | スキーマ定義確認 |
| PbCodegenException | コード生成エラー | コンパイル失敗 | スキーマとRowType の整合性確認 |
| ClassNotFoundException | クラス未発見 | message-class-name 不正 | クラスパス確認 |

## 備考

- コード生成（codegen）アプローチで高性能を実現
- proto2/proto3 両方をサポート
- proto3 では 3.15 未満のバージョンで read-default-values=true が必要な場合あり

---

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

### 推奨読解順序

#### Step 1: ファクトリクラス

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | PbFormatFactory.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbFormatFactory.java` | エントリーポイント |

**主要処理フロー**:
- **42行目**: IDENTIFIER = "protobuf"
- **45-49行目**: createDecodingFormat() でデシリアライザ生成
- **52-56行目**: createEncodingFormat() でシリアライザ生成
- **58-72行目**: buildConfig() で設定構築
- **80-84行目**: requiredOptions() で必須オプション定義

#### Step 2: 設定オプション

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | PbFormatOptions.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbFormatOptions.java` | 設定定義 |

**主要設定**:
- **26-32行目**: MESSAGE_CLASS_NAME（必須）
- **34-39行目**: IGNORE_PARSE_ERRORS（デフォルト: false）
- **41-50行目**: READ_DEFAULT_VALUES（デフォルト: false）
- **51-57行目**: WRITE_NULL_STRING_LITERAL（デフォルト: ""）

#### Step 3: デシリアライゼーション

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | PbRowDataDeserializationSchema.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java` | デシリアライザ |
| 3-2 | ProtoToRowConverter.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/ProtoToRowConverter.java` | コード生成変換 |

**デシリアライゼーション処理**:
- **51-61行目**: コンストラクタでスキーマ検証
- **64-66行目**: open() で ProtoToRowConverter 初期化
- **69-78行目**: deserialize() でバイト列を RowData に変換
- **73-74行目**: ignore-parse-errors=true で null 返却

**コード生成処理**:
- **59-130行目**: コンストラクタでコード生成・コンパイル
- **75-114行目**: codegen アペンダーでコード構築
- **119-123行目**: 動的コンパイル
- **132-135行目**: convertProtoBinaryToRow() で変換実行

#### Step 4: シリアライゼーション

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | PbRowDataSerializationSchema.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/serialize/PbRowDataSerializationSchema.java` | シリアライザ |

**シリアライゼーション処理**:
- **46-52行目**: コンストラクタでスキーマ検証
- **55-57行目**: open() で RowToProtoConverter 初期化
- **65-71行目**: serialize() で RowData を Protobuf バイト列に変換

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

```
PbFormatFactory
├── createDecodingFormat()
│   └── PbDecodingFormat
│       └── createRuntimeDecoder()
│           └── PbRowDataDeserializationSchema
│               └── ProtoToRowConverter
│                   ├── PbCodegenDeserializeFactory
│                   │   └── PbCodegenRowDeserializer
│                   └── PbCodegenUtils.compileClass()
└── createEncodingFormat()
    └── PbEncodingFormat
        └── createRuntimeEncoder()
            └── PbRowDataSerializationSchema
                └── RowToProtoConverter
                    ├── PbCodegenSerializeFactory
                    │   └── PbCodegenRowSerializer
                    └── PbCodegenUtils.compileClass()
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| PbFormatFactory.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbFormatFactory.java` | ソース | ファクトリ |
| PbFormatOptions.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/PbFormatOptions.java` | ソース | 設定オプション |
| PbRowDataDeserializationSchema.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/PbRowDataDeserializationSchema.java` | ソース | デシリアライザ |
| PbRowDataSerializationSchema.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/serialize/PbRowDataSerializationSchema.java` | ソース | シリアライザ |
| ProtoToRowConverter.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/deserialize/ProtoToRowConverter.java` | ソース | デシリアライズ変換 |
| RowToProtoConverter.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/serialize/RowToProtoConverter.java` | ソース | シリアライズ変換 |
| PbSchemaValidationUtils.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/util/PbSchemaValidationUtils.java` | ソース | スキーマ検証 |
| PbCodegenUtils.java | `flink-formats/flink-protobuf/src/main/java/org/apache/flink/formats/protobuf/util/PbCodegenUtils.java` | ソース | コード生成ユーティリティ |
