# 機能設計書 31-UDF/UDAF/UDTFサポート

## 概要

本ドキュメントは、Apache Sparkにおけるユーザー定義関数（UDF: スカラー関数、UDAF: 集約関数、UDTF: テーブル生成関数）の登録と実行機構に関する機能設計書である。Spark SQLエンジン上でユーザーが独自の処理ロジックを関数として定義・登録し、SQLクエリやDataFrame APIから呼び出す仕組みを詳細に記述する。

### 本機能の処理概要

**業務上の目的・背景**：Spark SQLの組み込み関数だけでは対応できない業務固有のデータ変換・集計・行展開ロジックが必要な場合がある。UDF/UDAF/UDTFサポートは、ユーザーがScala、Java、Pythonで独自の関数を実装し、SQLクエリ内でシームレスに利用できるようにすることで、Sparkの拡張性を飛躍的に向上させる。これにより、ETL処理やデータ分析パイプラインにおいて、組み込み関数では実現困難なカスタム処理を標準的なSQLインターフェースから実行可能になる。

**機能の利用シーン**：データエンジニアが業務固有のデータ変換（住所正規化、独自エンコーディングなど）をSQLクエリ内で行いたい場合、データサイエンティストがカスタム集約（重み付き平均、カスタムウィンドウ関数など）を適用したい場合、テーブル生成関数でJSON解析結果を複数行に展開したい場合などに利用される。

**主要な処理内容**：
1. UDF（スカラー関数）の登録：`spark.udf.register()` またはプログラムAPI `functions.udf()` を通じてユーザー定義のスカラー関数を登録し、SQL式やDataFrame列操作で使用可能にする
2. UDAF（集約関数）の登録：`Aggregator[IN, BUF, OUT]` インターフェースを実装した集約関数を `functions.udaf()` で登録し、GROUP BYやウィンドウ関数として使用可能にする
3. UDTF（テーブル生成関数）の登録：`spark.udtf.register()` を通じてPython UDTFを登録し、FROM句のテーブル関数として使用可能にする
4. Java UDFクラスのリフレクションベース登録：`UDF0`〜`UDF22` インターフェースを実装したJavaクラスを動的にロードして登録する
5. Python UDFのシリアライゼーションと実行：PythonワーカープロセスとのUDF連携を管理する
6. Catalyst内部での型変換：ユーザー定義関数の入出力をCatalyst内部表現（InternalRow）とScala/Java型の間で変換する

**関連システム・外部連携**：Catalyst optimizer（関数解決・最適化）、FunctionRegistry（関数カタログ管理）、TableFunctionRegistry（テーブル関数カタログ管理）、Python Worker（PySpark UDF実行）

**権限による制御**：関数の登録はSparkSession単位のスコープで管理される。永続関数としてHiveメタストアに登録する場合はカタログ権限が必要となる。一時関数は登録したセッション内でのみ有効。

## 関連画面

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

## 機能種別

計算処理 / データ変換 / API拡張

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| name | String | Yes | 登録する関数名 | 空文字不可、既存関数との重複時は上書き |
| f (UDF) | FunctionN | Yes | スカラー関数の実装（0〜22引数） | Scalaの場合TypeTag必須、nullセーフ考慮 |
| agg (UDAF) | Aggregator[IN, BUF, OUT] | Yes | 集約関数の実装 | zero/reduce/merge/finish/bufferEncoder/outputEncoder必須 |
| udtf (UDTF) | UserDefinedPythonTableFunction | Yes | Pythonテーブル生成関数 | returnType定義必須 |
| returnType | DataType | No | Java UDFの戻り値型（null時は推論） | Spark DataTypeとして有効な型 |
| className | String | No | Java UDF/UDAFクラスの完全修飾名 | クラスパスに存在すること |

### 入力データソース

- プログラムAPI（Scala/Java/Python）からの関数オブジェクト
- SQLの`CREATE FUNCTION`文による関数定義
- Java UDFクラスのリフレクションによるロード（ArtifactManager経由）

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| UserDefinedFunction | UserDefinedFunction | 登録済みUDFオブジェクト（Column式として使用可能） |
| FunctionRegistry登録 | Unit | FunctionRegistryへの関数ビルダー登録 |
| TableFunctionRegistry登録 | Unit | TableFunctionRegistryへのUDTFビルダー登録 |

### 出力先

- SparkSessionのFunctionRegistry（一時関数）
- SparkSessionのTableFunctionRegistry（一時テーブル関数）
- Hiveメタストア（永続関数として登録する場合）

## 処理フロー

### 処理シーケンス

```
1. UDF登録リクエスト受信
   └─ spark.udf.register(name, f) または functions.udf(f) が呼び出される
2. 関数オブジェクトのラッピング
   └─ SparkUserDefinedFunction として入力/出力エンコーダーと共にラップ
3. Expression Builder の構築
   └─ FunctionRegistryに登録するための Seq[Expression] => Expression ビルダーを生成
4. パラメータ数バリデーション（オプション）
   └─ validateParameterCount=true の場合、引数数の不一致を検出
5. FunctionRegistry への登録
   └─ createOrReplaceTempFunction() で一時関数として登録
6. SQL実行時の関数解決
   └─ Catalyst Analyzerが関数名をFunctionRegistryからExpressionに解決
7. Catalyst型変換
   └─ CatalystTypeConvertersで内部表現とScala/Java型を相互変換
8. 実行
   └─ 各Executor上でタスク内で関数が評価される
```

### フローチャート

```mermaid
flowchart TD
    A[UDF登録要求] --> B{関数種別}
    B -->|Scala UDF| C[SparkUserDefinedFunction生成]
    B -->|Java UDF| D[リフレクションでクラスロード]
    B -->|UDAF| E[ScalaAggregator生成]
    B -->|Python UDF| F[UserDefinedPythonFunction生成]
    B -->|UDTF| G[UserDefinedPythonTableFunction生成]
    C --> H[FunctionRegistry登録]
    D --> H
    E --> H
    F --> H
    G --> I[TableFunctionRegistry登録]
    H --> J[SQL/DataFrame APIで使用可能]
    I --> J
    J --> K{実行時}
    K --> L[Catalyst Analyzerが関数解決]
    L --> M[型変換・Expression評価]
    M --> N[結果返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-01 | 関数名上書き | 同名の関数が既に登録されている場合、新しい定義で上書きされる | createOrReplaceTempFunction呼び出し時 |
| BR-02 | 引数数制限 | Scala/Java UDFは最大22引数まで対応 | UDF登録時 |
| BR-03 | 決定性フラグ | UDFはデフォルトでdeterministicとして扱われ、asNondeterministic()で変更可能 | 最適化フェーズで関数の決定性を判定する際 |
| BR-04 | UDAF非推奨 | UserDefinedAggregateFunction APIは3.0.0で非推奨、Aggregator APIに移行を推奨 | UDAF登録時に非推奨警告 |
| BR-05 | セッションスコープ | 一時関数はSparkSession単位で有効 | SparkSession終了時に破棄 |

### 計算ロジック

- **ScalaUDAF**: initialize → update（各入力行） → merge（パーティション間統合） → evaluate（最終結果計算）の4ステップで集約を実行
- **ScalaAggregator**: zero → reduce（各入力行） → merge（パーティション間統合） → finish（最終結果計算）の4ステップで型安全な集約を実行
- **型変換**: CatalystTypeConvertersが`createToCatalystConverter`/`createToScalaConverter`でInternalRowとScala/Java型を双方向変換

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| 関数登録 | FunctionRegistry（インメモリ） | INSERT/UPDATE | 一時関数の登録・上書き |
| 永続関数登録 | Hive Metastore functions テーブル | INSERT | 永続関数のカタログ登録 |

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

#### FunctionRegistry（インメモリカタログ）

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| INSERT/UPDATE | functionName → ExpressionBuilder | 関数名をキーとしてビルダー関数を登録 | ConcurrentHashMapベースで管理 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| WRONG_NUM_ARGS | QueryCompilationError | 登録済みUDFの引数数と実際の引数数が不一致 | 正しい引数数で呼び出す |
| CLASS_NOT_FOUND | QueryCompilationError | registerJava()でクラスが見つからない | クラスパスにJARを追加 |
| NO_PUBLIC_CONSTRUCTOR | QueryCompilationError | UDFクラスに引数なしpublicコンストラクタがない | コンストラクタを追加 |
| DOES_NOT_IMPLEMENT_UDAF | QueryCompilationError | registerJavaUDAF()で指定クラスがUDAFを実装していない | 正しいインターフェースを実装 |
| TOO_MANY_TYPE_ARGS | QueryCompilationError | Java UDFの型引数が23以上 | 22以下に削減 |

### リトライ仕様

UDF登録処理自体にリトライ機構はない。実行時のエラーはタスクリトライの対象となる。

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

一時関数の登録はインメモリ操作でありトランザクション管理の対象外。永続関数の登録はHiveメタストアのトランザクション制御に従う。

## パフォーマンス要件

- UDF登録は即座に完了する（ミリ秒オーダー）
- UDF実行時のオーバーヘッドはCatalyst型変換コストに依存（特にScalaUDAFのRow→InternalRow変換）
- ScalaAggregatorはExpressionEncoder使用により、ScalaUDAFよりも高効率
- Python UDFはJVM-Python間のシリアライゼーションオーバーヘッドがあるため、可能な限りSpark組み込み関数を優先すべき

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

- Java UDFのリフレクションロードではArtifactManagerのClassLoaderを経由し、SparkSessionのセキュリティコンテキスト内で動作する
- Python UDFは隔離されたPythonワーカープロセス内で実行される
- ユーザー定義関数内での外部リソースアクセスはSparkのセキュリティ設定に従う

## 備考

- `UserDefinedAggregateFunction` は Spark 3.0.0 以降非推奨であり、`Aggregator[IN, BUF, OUT]` への移行が推奨されている
- UDTF（テーブル生成関数）はSpark 3.5.0で追加されたPython専用の機能であり、`@Evolving` アノテーションが付与されている
- `ResolveEncodersInScalaAgg` ルールにより、ScalaAggregatorのエンコーダーは論理計画の解決時にバインドされる

---

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

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

### 推奨読解順序

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

UDF/UDAF/UDTFの型階層と内部表現を理解することが最初のステップである。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | UserDefinedFunction.scala | `sql/api/src/main/scala/org/apache/spark/sql/expressions/` | UDFの基底トレイト・抽象クラス |
| 1-2 | SparkUserDefinedFunction.scala | `sql/api/src/main/scala/org/apache/spark/sql/expressions/` | 具体的なUDF実装クラス、inputEncodersの構造 |
| 1-3 | Aggregator.scala | `sql/api/src/main/scala/org/apache/spark/sql/expressions/` | UDAF用Aggregator[IN, BUF, OUT]トレイト |

**読解のコツ**: SparkUserDefinedFunctionは内部的にExpressionEncoderを保持し、Scala/Java型とCatalyst内部型（InternalRow）の相互変換を担う。TypeTagからのエンコーダー自動生成がポイント。

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

UDFの登録APIがどこから呼び出されるかを把握する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | functions.scala | `sql/api/src/main/scala/org/apache/spark/sql/functions.scala` | `udf()` および `udaf()` ファクトリメソッド群 |
| 2-2 | UDFRegistration.scala | `sql/core/src/main/scala/org/apache/spark/sql/classic/UDFRegistration.scala` | `register()` メソッドによるFunctionRegistryへの登録 |
| 2-3 | UDTFRegistration.scala | `sql/core/src/main/scala/org/apache/spark/sql/UDTFRegistration.scala` | UDTF登録のエントリーポイント |

**主要処理フロー**:
1. **47-48行目（UDFRegistration.scala）**: UDFRegistrationクラス定義、FunctionRegistryへの参照保持
2. **86-113行目（UDFRegistration.scala）**: `register(name, udf, source, validateParameterCount)`メソッドでUDAFとUDFを分岐処理
3. **92-94行目（UDFRegistration.scala）**: UserDefinedAggregatorの場合は`ScalaAggregator(udaf, _)`をビルダーとして登録
4. **95-109行目（UDFRegistration.scala）**: SparkUserDefinedFunctionの場合はパラメータ数検証付きで登録
5. **34-49行目（UDTFRegistration.scala）**: Python UDTFを`TableFunctionRegistry.createOrReplaceTempFunction`で登録

#### Step 3: UDAF内部実装を理解する

集約関数の内部実行メカニズムを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | udaf.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala` | ScalaUDAF/ScalaAggregatorの実装 |

**主要処理フロー**:
- **349-359行目**: ScalaUDAFクラス定義 - ImperativeAggregateを継承
- **448-452行目**: `initialize()` - 集約バッファの初期化
- **454-460行目**: `update()` - 入力行ごとのバッファ更新（型変換含む）
- **462-467行目**: `merge()` - パーティション間バッファ統合
- **469-473行目**: `eval()` - 最終結果の計算
- **487-555行目**: ScalaAggregatorクラス定義 - TypedImperativeAggregateを継承、ExpressionEncoder経由で型安全に動作
- **575-585行目**: ResolveEncodersInScalaAgg - Catalystルールとしてエンコーダーの解決とバインドを実行

#### Step 4: Java UDFのリフレクション登録を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | UDFRegistration.scala | `sql/core/src/main/scala/org/apache/spark/sql/classic/UDFRegistration.scala` | registerJava/registerJavaUDAFメソッド |

**主要処理フロー**:
- **122-137行目**: `registerJavaUDAF` - クラスローダーでUDAFクラスをロードしてインスタンス化
- **141-196行目**: `registerJava` - Java UDFインターフェース（UDF0〜UDF22）をリフレクションで判定し、型引数数に基づいて適切なregisterオーバーロードを呼び出す

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

```
SparkSession.udf / functions.udf() / functions.udaf()
    |
    ├─ UDFRegistration.register(name, udf, source)
    |      ├─ UserDefinedAggregator => ScalaAggregator(udaf, children)
    |      ├─ SparkUserDefinedFunction => toScalaUDF(udf, children)
    |      └─ FunctionRegistry.createOrReplaceTempFunction()
    |
    ├─ UDFRegistration.registerJava(name, className, returnType)
    |      ├─ ArtifactManager.classloader.loadClass()
    |      └─ FunctionRegistry.createOrReplaceTempFunction()
    |
    └─ UDTFRegistration.registerPython(name, udtf)
           └─ TableFunctionRegistry.createOrReplaceTempFunction()

[SQL実行時]
Catalyst Analyzer
    └─ FunctionRegistry.lookupFunction(name)
           └─ ExpressionBuilder(children) => ScalaUDAF / ScalaAggregator / ScalaUDF
                  └─ CatalystTypeConverters (型変換)
```

### データフロー図

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

ユーザー定義関数         SparkUserDefinedFunction        FunctionRegistryに登録
(Scala/Java/Python)  ──▶  ラッピング               ──▶  (ExpressionBuilder)
                              │
                              ▼
SQLクエリ /             Catalyst Analyzer               Expression評価
DataFrame API      ──▶  関数名解決               ──▶  (ScalaUDF/ScalaUDAF/
                              │                         ScalaAggregator)
                              ▼                              │
                     CatalystTypeConverters                   ▼
                     (InternalRow ⇔ Scala/Java型)        結果値返却
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| UDFRegistration.scala | `sql/core/src/main/scala/org/apache/spark/sql/classic/UDFRegistration.scala` | ソース | UDF/UDAF登録のメインクラス |
| UDTFRegistration.scala | `sql/core/src/main/scala/org/apache/spark/sql/UDTFRegistration.scala` | ソース | UDTF登録クラス |
| udaf.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/aggregate/udaf.scala` | ソース | ScalaUDAF/ScalaAggregator実行エンジン |
| functions.scala | `sql/api/src/main/scala/org/apache/spark/sql/functions.scala` | ソース | udf()/udaf()ファクトリメソッド |
| FunctionRegistry.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/FunctionRegistry.scala` | ソース | 関数カタログ管理 |
| TableFunctionRegistry.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableFunctionRegistry.scala` | ソース | テーブル関数カタログ管理 |
| UserDefinedPythonFunction.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonFunction.scala` | ソース | Python UDFラッパー |
| UserDefinedPythonTableFunction.scala | `sql/core/src/main/scala/org/apache/spark/sql/execution/python/UserDefinedPythonTableFunction.scala` | ソース | Python UDTFラッパー |
| CatalystTypeConverters.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/CatalystTypeConverters.scala` | ソース | Catalyst内部型とScala/Java型の変換 |
| ExpressionEncoder.scala | `sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/encoders/ExpressionEncoder.scala` | ソース | 型安全なエンコーダー |
