# 機能設計書：UDF（ユーザー定義関数）

## 1. 機能概要

### 1.1 処理概要

UDF（User Defined Function）は、Apache Flink Table API/SQLにおいてユーザーが独自の関数を定義・登録するための機能である。`UserDefinedFunction`を基底クラスとし、複数の関数タイプをサポートする。

主要な関数タイプ：
1. **ScalarFunction**: 単一値を返すスカラー関数
2. **TableFunction**: 複数行を返すテーブル関数
3. **AggregateFunction**: 集計関数
4. **TableAggregateFunction**: 複数行を返すテーブル集計関数
5. **AsyncScalarFunction**: 非同期スカラー関数

### 1.2 業務上の目的・役割

| 項目 | 内容 |
|------|------|
| 目的 | 標準関数では対応できないビジネスロジックの実装 |
| 役割 | カスタム変換・集計処理をSQL/Table APIで利用可能にする |
| 主な利用場面 | 業務固有の計算、外部サービス連携、カスタム集計 |

### 1.3 利用シーン・ユースケース

1. **カスタム文字列処理**
   - 業務固有のフォーマット変換
   - 暗号化・マスキング処理

2. **外部サービス連携**
   - 非同期APIコール
   - エンリッチメント処理

3. **カスタム集計**
   - 統計関数の実装
   - 複雑な集計ロジック

4. **テーブル展開**
   - ネストデータの展開
   - 複数行生成

## 2. 入出力仕様

### 2.1 関数タイプ別仕様

#### 2.1.1 ScalarFunction

| 項目 | 仕様 |
|------|------|
| 入力 | 0個以上のスカラー値 |
| 出力 | 単一スカラー値 |
| 評価メソッド | `eval(...)` |
| FunctionKind | SCALAR |

```java
// 基本的なScalarFunction例
class SumFunction extends ScalarFunction {
    public Integer eval(Integer a, Integer b) {
        return a + b;
    }
}
```

#### 2.1.2 TableFunction

| 項目 | 仕様 |
|------|------|
| 入力 | 0個以上のスカラー値 |
| 出力 | 0行以上の行（collect()で出力） |
| 評価メソッド | `eval(...)` |
| FunctionKind | TABLE |

```java
// 基本的なTableFunction例
class SplitFunction extends TableFunction<String> {
    public void eval(String str) {
        for (String s : str.split(" ")) {
            collect(s);  // 行を出力
        }
    }
}
```

#### 2.1.3 AggregateFunction

| 項目 | 仕様 |
|------|------|
| 入力 | 複数行のスカラー値 |
| 出力 | 単一スカラー値 |
| 必須メソッド | `createAccumulator()`, `accumulate()`, `getValue()` |
| オプションメソッド | `retract()`, `merge()` |
| FunctionKind | AGGREGATE |

```java
// 基本的なAggregateFunction例
class CountFunction extends AggregateFunction<Long, CountAccumulator> {
    public CountAccumulator createAccumulator() {
        return new CountAccumulator();
    }

    public void accumulate(CountAccumulator acc, Integer value) {
        if (value != null) {
            acc.count++;
        }
    }

    public Long getValue(CountAccumulator acc) {
        return acc.count;
    }
}
```

#### 2.1.4 TableAggregateFunction

| 項目 | 仕様 |
|------|------|
| 入力 | 複数行のスカラー値 |
| 出力 | 0行以上の行 |
| 必須メソッド | `createAccumulator()`, `accumulate()`, `emitValue()` または `emitUpdateWithRetract()` |
| FunctionKind | TABLE_AGGREGATE |

#### 2.1.5 AsyncScalarFunction

| 項目 | 仕様 |
|------|------|
| 入力 | CompletableFuture<T> + 0個以上のスカラー値 |
| 出力 | 非同期完了するスカラー値 |
| 評価メソッド | `eval(CompletableFuture<T>, ...)` |
| FunctionKind | ASYNC_SCALAR |

```java
// 基本的なAsyncScalarFunction例
class AsyncLookupFunction extends AsyncScalarFunction {
    public void eval(CompletableFuture<String> future, String key) {
        // 非同期処理
        future.complete(result);
    }
}
```

## 3. 処理フロー

### 3.1 UDF継承階層

```mermaid
classDiagram
    class UserDefinedFunction {
        <<abstract>>
        +open(FunctionContext)
        +close()
        +getTypeInference(DataTypeFactory)
        +functionIdentifier()
        +isDeterministic()
    }

    class ScalarFunction {
        <<abstract>>
        +eval(...)
        +getKind() SCALAR
    }

    class AsyncScalarFunction {
        +eval(CompletableFuture, ...)
        +getKind() ASYNC_SCALAR
    }

    class TableFunction~T~ {
        <<abstract>>
        #collect(T)
        +setCollector(Collector)
        +eval(...)
        +finish()
        +getKind() TABLE
    }

    class ImperativeAggregateFunction~T,ACC~ {
        <<abstract>>
        +createAccumulator() ACC
    }

    class AggregateFunction~T,ACC~ {
        <<abstract>>
        +accumulate(ACC, ...)
        +getValue(ACC) T
        +retract(ACC, ...)
        +merge(ACC, Iterable)
        +getKind() AGGREGATE
    }

    class TableAggregateFunction~T,ACC~ {
        <<abstract>>
        +accumulate(ACC, ...)
        +emitValue(ACC, Collector)
        +emitUpdateWithRetract(ACC, RetractableCollector)
        +getKind() TABLE_AGGREGATE
    }

    UserDefinedFunction <|-- ScalarFunction
    UserDefinedFunction <|-- AsyncScalarFunction
    UserDefinedFunction <|-- TableFunction
    UserDefinedFunction <|-- ImperativeAggregateFunction
    ImperativeAggregateFunction <|-- AggregateFunction
    ImperativeAggregateFunction <|-- TableAggregateFunction
```

### 3.2 UDF実行フロー

```mermaid
sequenceDiagram
    participant User as ユーザーコード
    participant TE as TableEnvironment
    participant Planner as Planner
    participant TIE as TypeInferenceExtractor
    participant Runtime as Runtime

    User->>TE: createTemporarySystemFunction()
    TE->>TIE: getTypeInference()
    TIE-->>TE: TypeInference
    Note over TE: 関数登録

    User->>TE: SQLクエリ実行
    TE->>Planner: プラン生成
    Planner->>Runtime: コード生成
    Runtime->>Runtime: open(FunctionContext)
    Runtime->>Runtime: eval() / accumulate()
    Runtime->>Runtime: close()
```

### 3.3 型推論フロー

```mermaid
flowchart TD
    subgraph Extraction["型抽出"]
        UDF[UserDefinedFunction] --> |リフレクション| TIE[TypeInferenceExtractor]
        DTH[DataTypeHint] --> TIE
        FH[FunctionHint] --> TIE
    end

    subgraph Inference["型推論"]
        TIE --> TI[TypeInference]
        TI --> ITS[InputTypeStrategy]
        TI --> OTS[OutputTypeStrategy]
    end

    subgraph Validation["検証"]
        ITS --> |引数検証| Valid[Validation]
        OTS --> |戻り値型決定| Valid
    end
```

## 4. 業務ルール・条件

### 4.1 ライフサイクルメソッド

| メソッド | 説明 | 呼び出しタイミング |
|---------|------|-------------------|
| `open(FunctionContext)` | 初期化処理 | タスク開始時 |
| `close()` | 終了処理 | タスク終了時 |
| `finish()` | データ処理終了時（TableFunctionのみ） | 入力データ終了時 |

### 4.2 FunctionContext機能

| メソッド | 説明 |
|---------|------|
| `getTaskInfo()` | タスク情報取得 |
| `getMetricGroup()` | メトリクスグループ取得 |
| `getCachedFile(name)` | 分散キャッシュファイル取得 |
| `getJobParameter(key, default)` | ジョブパラメータ取得 |
| `getExternalResourceInfos(name)` | 外部リソース情報取得 |
| `getUserCodeClassLoader()` | ユーザーコードクラスローダー取得 |

### 4.3 型ヒントアノテーション

| アノテーション | 用途 |
|---------------|------|
| `@DataTypeHint` | データ型の明示的指定 |
| `@FunctionHint` | 入力・出力・アキュムレータ型の指定 |
| `@DataTypeHint(inputGroup = InputGroup.ANY)` | 任意の型を受け入れ |

```java
// 型ヒントの使用例
@FunctionHint(
    input = @DataTypeHint("INT"),
    output = @DataTypeHint("STRING")
)
public class ConvertFunction extends ScalarFunction {
    public String eval(Object o) {
        return o.toString();
    }
}
```

### 4.4 deterministic特性

```java
// isDeterministic()のデフォルトはtrue
// falseを返すと定数式削減が無効化される
@Override
public boolean isDeterministic() {
    return false;  // 非決定的関数
}
```

### 4.5 カタログ保存要件

| 要件 | 説明 |
|------|------|
| デフォルトコンストラクタ | 必須 |
| インスタンス化可能 | ランタイムでの生成が必要 |
| 無状態（匿名関数） | transientとstaticフィールドのみ許可 |

### 4.6 AggregateFunction特有のメソッド

| メソッド | 必須 | 用途 |
|---------|------|------|
| `createAccumulator()` | Yes | 空のアキュムレータ作成 |
| `accumulate()` | Yes | 値の蓄積 |
| `getValue()` | Yes | 結果の取得 |
| `retract()` | No | リトラクション処理（OVER句で必要） |
| `merge()` | No | アキュムレータのマージ（セッションウィンドウで必要） |

### 4.7 データビュー

| ビュー | 用途 |
|--------|------|
| `ListView<T>` | 大量データのリスト保存 |
| `MapView<K, V>` | 大量データのマップ保存 |

```java
// ListView使用例
public class MyAggFunction extends AggregateFunction<String, MyAccumulator> {
    public static class MyAccumulator {
        public ListView<String> list = new ListView<>();
    }
}
```

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

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

```
UserDefinedFunction (基底クラス)
├── ScalarFunction
│   ├── eval(...) → 単一値を返す
│   ├── getKind() → FunctionKind.SCALAR
│   └── getTypeInference() → TypeInferenceExtractor.forScalarFunction()
│
├── AsyncScalarFunction
│   ├── eval(CompletableFuture, ...) → 非同期完了
│   ├── getKind() → FunctionKind.ASYNC_SCALAR
│   └── getTypeInference() → TypeInferenceExtractor.forAsyncScalarFunction()
│
├── TableFunction<T>
│   ├── eval(...) → collect()で複数行出力
│   ├── collect(T) → 行の出力
│   ├── finish() → 処理終了時
│   ├── getKind() → FunctionKind.TABLE
│   └── getTypeInference() → TypeInferenceExtractor.forTableFunction()
│
├── ImperativeAggregateFunction<T, ACC>
│   ├── createAccumulator() → アキュムレータ作成
│   │
│   ├── AggregateFunction<T, ACC>
│   │   ├── accumulate(ACC, ...) → 値の蓄積
│   │   ├── getValue(ACC) → 結果取得
│   │   ├── retract(ACC, ...) → リトラクション
│   │   ├── merge(ACC, Iterable) → マージ
│   │   ├── getKind() → FunctionKind.AGGREGATE
│   │   └── getTypeInference() → TypeInferenceExtractor.forAggregateFunction()
│   │
│   └── TableAggregateFunction<T, ACC>
│       ├── accumulate(ACC, ...) → 値の蓄積
│       ├── emitValue(ACC, Collector) → 結果出力
│       ├── emitUpdateWithRetract(ACC, RetractableCollector) → 増分出力
│       ├── getKind() → FunctionKind.TABLE_AGGREGATE
│       └── getTypeInference() → TypeInferenceExtractor.forTableAggregateFunction()
│
FunctionContext (実行コンテキスト)
├── getTaskInfo()
├── getMetricGroup()
├── getCachedFile()
├── getJobParameter()
├── getExternalResourceInfos()
└── getUserCodeClassLoader()
```

### 5.2 データフロー図

```
[ユーザー定義関数クラス]
        │
        ▼
[TableEnvironment.createTemporarySystemFunction()]
        │
        ▼
[TypeInferenceExtractor]
        │
        ├── リフレクション解析
        ├── @DataTypeHint解析
        └── @FunctionHint解析
        │
        ▼
[TypeInference]
        │
        ├── InputTypeStrategy
        └── OutputTypeStrategy
        │
        ▼
[Planner]
        │
        ├── 関数解決
        └── コード生成
        │
        ▼
[Runtime]
        │
        ├── open(FunctionContext)
        ├── eval() / accumulate()
        └── close()
```

### 5.3 関連ファイル一覧

| ファイルパス | 役割 |
|-------------|------|
| `flink-table-common/.../UserDefinedFunction.java` | UDF基底クラス |
| `flink-table-common/.../ScalarFunction.java` | スカラー関数基底クラス |
| `flink-table-common/.../AsyncScalarFunction.java` | 非同期スカラー関数基底クラス |
| `flink-table-common/.../TableFunction.java` | テーブル関数基底クラス |
| `flink-table-common/.../AggregateFunction.java` | 集計関数基底クラス |
| `flink-table-common/.../TableAggregateFunction.java` | テーブル集計関数基底クラス |
| `flink-table-common/.../ImperativeAggregateFunction.java` | 集計関数共通基底クラス |
| `flink-table-common/.../FunctionContext.java` | 関数実行コンテキスト |
| `flink-table-common/.../FunctionKind.java` | 関数種別列挙型 |
| `flink-table-common/annotation/DataTypeHint.java` | データ型ヒントアノテーション |
| `flink-table-common/annotation/FunctionHint.java` | 関数ヒントアノテーション |
| `flink-table-common/.../TypeInferenceExtractor.java` | 型推論抽出 |

### 5.4 コードリーディング手順

#### Step 1: UserDefinedFunction基底クラスの理解

```java
// UserDefinedFunction.java (34-58行目)
/**
 * Base class for all user-defined functions.
 *
 * A runtime implementation might be called at two different stages:
 * - During planning (constant expression reduction)
 * - During runtime (cluster execution)
 */
@PublicEvolving
public abstract class UserDefinedFunction implements FunctionDefinition, Serializable {
```

#### Step 2: ライフサイクルメソッド

```java
// UserDefinedFunction.java (73-87行目)
/**
 * Setup method for user-defined function.
 */
public void open(FunctionContext context) throws Exception {
    // do nothing
}

/**
 * Tear-down method for user-defined function.
 */
public void close() throws Exception {
    // do nothing
}
```

#### Step 3: ScalarFunctionの実装

```java
// ScalarFunction.java (32-42行目)
/**
 * Base class for a user-defined scalar function. A user-defined scalar function maps
 * zero, one, or multiple scalar values to a new scalar value.
 *
 * An evaluation method must be declared publicly and named "eval".
 */
@PublicEvolving
public abstract class ScalarFunction extends UserDefinedFunction {
    @Override
    public final FunctionKind getKind() {
        return FunctionKind.SCALAR;
    }
}
```

#### Step 4: TableFunctionの実装

```java
// TableFunction.java (136-198行目)
@PublicEvolving
public abstract class TableFunction<T> extends UserDefinedFunction {
    private transient Collector<T> collector;

    protected final void collect(T row) {
        collector.collect(row);
    }

    @Override
    public final FunctionKind getKind() {
        return FunctionKind.TABLE;
    }
}
```

#### Step 5: AggregateFunctionの実装

```java
// AggregateFunction.java (192-208行目)
@PublicEvolving
public abstract class AggregateFunction<T, ACC> extends ImperativeAggregateFunction<T, ACC> {
    public abstract T getValue(ACC accumulator);

    @Override
    public final FunctionKind getKind() {
        return FunctionKind.AGGREGATE;
    }
}
```

#### Step 6: FunctionContextの機能

```java
// FunctionContext.java (42-50行目)
/**
 * A FunctionContext allows to obtain global runtime information about the context
 * in which the user-defined function is executed.
 *
 * The information includes the metric group, distributed cache files, and global job parameters.
 */
@PublicEvolving
public class FunctionContext {
```

## 6. 関連機能・API

### 6.1 関連機能

| 機能名 | 関連種別 | 説明 |
|--------|---------|------|
| Table API Java（No.13） | 利用元 | Table APIからのUDF呼び出し |
| 組み込み関数（No.16） | 類似機能 | システム提供の標準関数 |
| SQLパーサー（No.18） | 連携機能 | UDF呼び出しの解析 |
| SQLプランナー（No.19） | 連携機能 | UDF実行計画の生成 |

### 6.2 使用例

```java
// ScalarFunction登録と使用
TableEnvironment tEnv = TableEnvironment.create(...);
tEnv.createTemporarySystemFunction("my_func", MyScalarFunction.class);
tEnv.sqlQuery("SELECT my_func(col1, col2) FROM my_table");

// TableFunction使用（LATERAL TABLE）
tEnv.createTemporarySystemFunction("split", SplitFunction.class);
tEnv.sqlQuery("SELECT a, s FROM my_table, LATERAL TABLE(split(a)) AS T(s)");

// AggregateFunction使用
tEnv.createTemporarySystemFunction("my_agg", MyAggFunction.class);
tEnv.sqlQuery("SELECT my_agg(amount) FROM orders GROUP BY customer_id");

// Table APIでの使用
table.select(call(MyScalarFunction.class, $("col1"), $("col2")));
table.joinLateral(call(SplitFunction.class, $("text")).as("word"));
```

## 7. 設計上の考慮事項

### 7.1 パフォーマンス考慮事項

| 考慮点 | 説明 | 推奨対応 |
|--------|------|---------|
| オブジェクト生成 | eval()内でのオブジェクト生成を最小化 | 再利用可能なオブジェクトを使用 |
| 定数式削減 | 定数引数の関数は事前評価 | isDeterministic()を適切に設定 |
| 状態管理 | アキュムレータのサイズ | ListView/MapViewを使用 |
| 非同期処理 | 外部サービス呼び出し | AsyncScalarFunctionを使用 |

### 7.2 型安全性

| 観点 | 説明 |
|------|------|
| 自動型推論 | リフレクションで入出力型を抽出 |
| 型ヒント | @DataTypeHint/@FunctionHintで補完 |
| getTypeInference()オーバーライド | 高度なカスタマイズ |

### 7.3 制約事項

| 制約 | 説明 |
|------|------|
| メソッド名 | eval, accumulate等は固定名 |
| 可視性 | publicで宣言必須 |
| 静的メソッド不可 | インスタンスメソッド必須 |
| デフォルトコンストラクタ | カタログ保存に必須 |

## 8. 用語集

| 用語 | 説明 |
|------|------|
| UserDefinedFunction | UDFの基底クラス |
| ScalarFunction | 単一値を返すスカラー関数 |
| TableFunction | 複数行を返すテーブル関数 |
| AggregateFunction | 集計関数 |
| TableAggregateFunction | 複数行を返すテーブル集計関数 |
| AsyncScalarFunction | 非同期スカラー関数 |
| FunctionContext | 関数の実行コンテキスト |
| Accumulator | 集計関数の中間状態 |
| TypeInference | 型推論情報 |
| DataTypeHint | データ型指定アノテーション |
| FunctionHint | 関数ヒントアノテーション |
| ListView/MapView | 大量データ用のステート対応コレクション |
