# 機能設計書 33-Protobufデータフォーマット

## 概要

本ドキュメントは、Apache SparkにおけるProtocol Buffersフォーマットのデータ読み書き機能に関する機能設計書である。`from_protobuf`/`to_protobuf`関数によるバイナリProtobufデータとSpark SQL型の相互変換機構、およびスキーマ変換メカニズムを詳細に記述する。

### 本機能の処理概要

**業務上の目的・背景**：Protocol Buffers（Protobuf）はGoogleが開発した高効率なバイナリシリアライゼーションフォーマットであり、gRPCサービス間通信やデータパイプラインで広く採用されている。SparkでのProtobufネイティブサポートにより、Protobufエンコードされたデータを直接SQLクエリで処理可能にし、ETLパイプラインのデータ変換コストを削減する。

**機能の利用シーン**：gRPCサービスからのログデータ（Protobuf形式）をSpark SQLで分析する場合、Kafkaメッセージ（Protobufエンコード）をStructured Streamingで処理する場合、.protoファイル定義に基づいたスキーマ変換が必要な場合。

**主要な処理内容**：
1. `from_protobuf`関数：バイナリProtobufデータをSpark SQL構造体に変換するSQL関数
2. `to_protobuf`関数：Spark SQLデータをバイナリProtobuf形式に変換するSQL関数
3. スキーマ変換：ProtobufメッセージDescriptorとSpark SQL StructTypeの相互変換
4. FileDescriptorSetサポート：コンパイル済み.protoファイルからのDescriptor構築
5. TypeRegistry対応：google.protobuf.Anyフィールドの動的型解決

**関連システム・外部連携**：Protocol Buffersライブラリ（com.google.protobuf）、Kafkaコネクタ（from_protobuf/to_protobuf経由）

**権限による制御**：特別な権限制御はない。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能に直接関連するWeb UI画面はない |

## 機能種別

データ連携 / データ変換

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| data | Binary | Yes | Protobufエンコードされたバイナリデータ | バイナリ型 |
| messageName | String | Yes | Protobufメッセージ名（完全修飾名） | 有効なメッセージ名 |
| binaryFileDescriptorSet | Array[Byte] | No | コンパイル済み.protoファイルのバイナリ | 有効なFileDescriptorSet |
| options | Map[String, String] | No | 変換オプション | mode, convertAnyFieldsToJson等 |

### 入力データソース

- Kafkaメッセージ等のバイナリProtobufデータ
- コンパイル済み.protoファイル（FileDescriptorSet）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| StructType | StructType | from_protobuf変換結果（Spark SQL構造体） |
| Binary | Array[Byte] | to_protobuf変換結果（Protobufバイナリ） |

### 出力先

- Spark SQLカラム値として返却
- Kafkaシンク等（to_protobufバイナリ）

## 処理フロー

### 処理シーケンス

```
1. from_protobuf実行
   └─ ProtobufDataToCatalyst.nullSafeEval() でバイナリ → InternalRow
2. Descriptor構築
   └─ ProtobufUtils.buildDescriptor() でメッセージ定義を構築
3. DynamicMessage解析
   └─ DynamicMessage.parseFrom() でバイナリをProtobufメッセージに変換
4. Unknown Fields検証
   └─ フィールド番号の不整合を検出
5. デシリアライズ
   └─ ProtobufDeserializer.deserialize() でSpark SQL型に変換
6. to_protobuf実行
   └─ CatalystDataToProtobuf.nullSafeEval() でInternalRow → バイナリ
```

### フローチャート

```mermaid
flowchart TD
    A[Protobufバイナリデータ] --> B{操作種別}
    B -->|from_protobuf| C[ProtobufDataToCatalyst]
    B -->|to_protobuf| D[CatalystDataToProtobuf]
    C --> E[ProtobufUtils.buildDescriptor]
    E --> F[DynamicMessage.parseFrom]
    F --> G{Unknown Fields検証}
    G -->|OK| H[ProtobufDeserializer]
    G -->|NG| I[フィールド型不整合エラー]
    H --> J[InternalRow]
    D --> K[ProtobufSerializer]
    K --> L[DynamicMessage.toByteArray]
    L --> M[Binary Output]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | ParseMode制御 | PermissiveMode（不正レコードをnull変換）とFailFastMode（例外発生）を選択可能 | from_protobuf実行時 |
| BR-02 | Unknown Fields検証 | パース結果のunknown fieldsに既知フィールド番号が含まれる場合、スキーマ不整合エラー | from_protobuf実行時 |
| BR-03 | Any型JSON変換 | convertAnyFieldsToJsonオプション有効時、google.protobuf.AnyフィールドをJSON文字列に変換 | TypeRegistry構築時 |
| BR-04 | デフォルト値出力 | emitDefaultValuesオプション有効時、Protobufのデフォルト値を明示的に出力 | ProtobufDeserializer構築時 |
| BR-05 | Enum整数出力 | enumsAsIntsオプション有効時、Enum値を文字列ではなく整数として出力 | ProtobufDeserializer構築時 |

### 計算ロジック

- **from_protobuf**: DynamicMessage.parseFrom(descriptor, binary) → ProtobufDeserializer.deserialize() → InternalRow
- **to_protobuf**: ProtobufSerializer.serialize(input) → DynamicMessage → toByteArray()
- **スキーマ変換**: SchemaConverters.toSqlType()がProtobuf Descriptor → StructType変換を実施

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

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

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

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

データベース操作は発生しない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| MALFORMED_PROTOBUF | QueryExecutionError | FailFastModeで不正なProtobufレコードを検出 | データ修正またはPermissiveModeに変更 |
| PROTOBUF_FIELD_TYPE_MISMATCH | QueryCompilationError | Unknown Fieldsに既知フィールド番号が含まれる | スキーマの整合性を確認 |
| PARSE_MODE_UNSUPPORTED | QueryCompilationError | Permissive/FailFast以外のモード指定 | 有効なモードを指定 |

### リトライ仕様

Spark標準のタスクリトライメカニズムに従う。

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

from_protobuf/to_protobuf関数は行単位の変換処理であり、トランザクション管理の対象外。

## パフォーマンス要件

- DynamicMessageベースの解析を使用（Javaクラス生成ベースより低速だが柔軟）
- binaryFileDescriptorSetが大きい場合の転送コストに注意（TODO: SPARK-43578でブロードキャスト対応予定）

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

- バイナリデータの解析は隔離された環境で実行される
- FileDescriptorSetの信頼性はユーザーの責任

## 備考

- Protobufコネクタは `connector/protobuf/` 配下に独立モジュールとして配置
- SPARK-43578: 大きなbinaryFileDescriptorSetのブロードキャスト対応が将来の改善課題
- ProtobufDataToCatalystはequals/hashCodeをオーバーライドしてbinaryFileDescriptorSetのバイト配列比較に対応

---

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

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

### 推奨読解順序

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

Protobufスキーマの構築とSpark型への変換を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SchemaConverters.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/SchemaConverters.scala` | toSqlType変換ロジック |
| 1-2 | ProtobufUtils.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufUtils.scala` | Descriptor構築、TypeRegistry構築 |
| 1-3 | ProtobufOptions.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/ProtobufOptions.scala` | 変換オプション定義 |

**読解のコツ**: ProtobufUtils.buildDescriptor()はメッセージ名とオプションのFileDescriptorSetからProtobuf Descriptorを構築する。クラスパス上の.protoコンパイル結果とバイナリDescriptorSetの2つのソースに対応。

#### Step 2: from_protobuf/to_protobuf関数を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | ProtobufDataToCatalyst.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufDataToCatalyst.scala` | from_protobuf実装 |
| 2-2 | CatalystDataToProtobuf.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/CatalystDataToProtobuf.scala` | to_protobuf実装 |

**主要処理フロー**:
- **32-38行目（ProtobufDataToCatalyst.scala）**: case class定義、messageName・binaryFileDescriptorSet・optionsを保持
- **49-50行目**: ProtobufUtils.buildDescriptor()でDescriptor構築
- **55-70行目**: TypeRegistry構築（Any型JSON変換オプション対応）
- **92-120行目**: `nullSafeEval` - DynamicMessage.parseFrom() → Unknown Fields検証 → ProtobufDeserializer.deserialize()
- **26-31行目（CatalystDataToProtobuf.scala）**: case class定義
- **44-47行目**: `nullSafeEval` - ProtobufSerializer.serialize() → toByteArray()

#### Step 3: シリアライザ/デシリアライザを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | ProtobufDeserializer.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufDeserializer.scala` | Protobuf → Catalyst変換 |
| 3-2 | ProtobufSerializer.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/ProtobufSerializer.scala` | Catalyst → Protobuf変換 |

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

```
from_protobuf(col, messageName, descriptorSet, options)
    │
    └─ ProtobufDataToCatalyst
           ├─ ProtobufUtils.buildDescriptor(messageName, binaryFileDescriptorSet)
           ├─ ProtobufUtils.buildTypeRegistry(descBytes)
           ├─ DynamicMessage.parseFrom(messageDescriptor, binary)
           ├─ UnknownFields検証
           └─ ProtobufDeserializer.deserialize(result)

to_protobuf(col, messageName, descriptorSet, options)
    │
    └─ CatalystDataToProtobuf
           ├─ ProtobufUtils.buildDescriptor(messageName, binaryFileDescriptorSet)
           ├─ ProtobufSerializer.serialize(input)
           └─ DynamicMessage.toByteArray()
```

### データフロー図

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

Protobufバイナリ          DynamicMessage.parseFrom         InternalRow
(from_protobuf)    ──▶  → ProtobufDeserializer       ──▶  (Spark SQL型)

Spark SQL値              ProtobufSerializer               Protobufバイナリ
(to_protobuf)     ──▶  → DynamicMessage.toByteArray  ──▶  (byte[])
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| ProtobufDataToCatalyst.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/` | ソース | from_protobuf式 |
| CatalystDataToProtobuf.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/` | ソース | to_protobuf式 |
| ProtobufDeserializer.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/` | ソース | デシリアライザ |
| ProtobufSerializer.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/` | ソース | シリアライザ |
| SchemaConverters.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/` | ソース | スキーマ変換 |
| ProtobufUtils.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/` | ソース | Descriptor構築ユーティリティ |
| ProtobufOptions.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/utils/` | ソース | オプション定義 |
| package.scala | `connector/protobuf/src/main/scala/org/apache/spark/sql/protobuf/` | ソース | パブリックAPI |
| functions.scala | `sql/api/src/main/scala/org/apache/spark/sql/protobuf/` | ソース | SQL関数API |
