# バッチ設計書 40-BatchExecVectorSearchTableFunction

## 概要

本ドキュメントは、Apache Flinkのバッチ処理におけるベクトル類似度検索を実行する`BatchExecVectorSearchTableFunction`の設計仕様を定義するものです。

### 本バッチの処理概要

BatchExecVectorSearchTableFunctionは、ベクトルデータベースやベクトル検索対応テーブルソースに対して類似度検索（Approximate Nearest Neighbor: ANN検索）を実行するための実行ノードです。SQL構文のVECTOR_SEARCH関数に対応し、同期・非同期の両方の検索処理をサポートします。

**業務上の目的・背景**：AI/ML時代において、ベクトル埋め込み（Embedding）を使用した類似度検索は、セマンティック検索、レコメンデーションシステム、画像/テキスト類似検索などの多くのユースケースで重要な役割を果たします。本バッチは、FlinkのTable API/SQLから直接ベクトル検索を実行し、検索結果を他のデータと結合・処理する機能を提供します。

**バッチの実行タイミング**：VECTOR_SEARCH関数を含むSQLクエリの実行時に、クエリプランナーによって自動的に生成されます。

**主要な処理内容**：
1. ベクトル検索テーブルソースの取得
2. VectorSearchRuntimeProviderから検索関数（VectorSearchFunction/AsyncVectorSearchFunction）の取得
3. VectorSearchCodeGeneratorによる検索コードの生成
4. 同期モード：VectorSearchRunnerとProcessOperatorによる処理
5. 非同期モード：AsyncVectorSearchRunnerとAsyncWaitOperatorによる処理
6. 入力行のベクトルカラムとTopKパラメータを使用してベクトル検索を実行
7. 検索結果と入力行を結合（INNER JOINまたはLEFT OUTER JOIN）して出力

**前後の処理との関連**：前段として任意のBatchExecNodeからの入力データを受け取ります。後段としてBatchExecCalc、BatchExecSink等に接続されます。

**影響範囲**：VECTOR_SEARCH関数を使用するすべてのバッチクエリに影響します。ベクトル検索バックエンドの性能がクエリ全体に影響します。

## バッチ種別

機械学習 / ベクトル検索

## 実行スケジュール

| 項目 | 内容 |
|-----|------|
| 実行頻度 | クエリ実行時（随時） |
| 実行時刻 | N/A |
| 実行曜日 | N/A |
| 実行日 | N/A |
| トリガー | VECTOR_SEARCH関数使用時 |

## 実行条件

### 前提条件

| 条件 | 説明 |
|-----|------|
| ベクトル検索テーブル登録 | VECTOR_SEARCH対象のテーブルがVectorSearchTableSourceとして登録されていること |
| VectorSearchFunctionProvider | プロバイダーがVectorSearchFunctionProviderを実装していること（同期モード） |
| AsyncVectorSearchFunctionProvider | プロバイダーがAsyncVectorSearchFunctionProviderを実装していること（非同期モード） |

### 実行可否判定

ベクトル検索プロバイダーが要求される検索関数（同期/非同期）を提供できる場合のみ実行可能。提供できない場合はTableExceptionがスローされる。

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | デフォルト値 | 説明 |
|-------------|-----|-----|-------------|------|
| tableSourceSpec | VectorSearchTableSourceSpec | Yes | - | ベクトル検索テーブルソース仕様 |
| vectorSearchSpec | VectorSearchSpec | Yes | - | 検索仕様（TopK、検索カラム、結合タイプ等） |
| asyncOptions | FunctionCallUtil.AsyncOptions | No | null | 非同期オプション（null時は同期モード） |
| inputProperty | InputProperty | Yes | - | 入力プロパティ |
| outputType | RowType | Yes | - | 出力型（入力＋検索結果） |

### 入力データソース

| データソース | 形式 | 説明 |
|-------------|------|------|
| 上流ExecNode | RowData | 検索クエリベクトルを含む入力レコード |

## 出力仕様

### 出力データ

| 出力先 | 形式 | 説明 |
|-------|------|------|
| 下流ExecNode | RowData | 入力レコード＋検索結果カラム |

### 出力ファイル仕様

N/A（メモリ内データ処理のため）

## 処理フロー

### 処理シーケンス

```
1. 入力変換の取得
   └─ 入力エッジから変換を取得
2. ベクトル検索テーブルの取得
   └─ tableSourceSpec.getSearchTable
3. VectorSearchRuntimeProviderの取得
   └─ VectorSearchUtil.createVectorSearchRuntimeProvider
4. 検索関数の取得
   └─ 同期: VectorSearchFunctionProvider.createVectorSearchFunction
   └─ 非同期: AsyncVectorSearchFunctionProvider.createAsyncVectorSearchFunction
5. 関数の準備
   └─ UserDefinedFunctionHelper.prepareInstance
6. 検索コード生成
   └─ 同期: VectorSearchCodeGenerator.generateSyncVectorSearchFunction
   └─ 非同期: VectorSearchCodeGenerator.generateAsyncVectorSearchFunction
7. コレクター生成
   └─ VectorSearchCodeGenerator.generateCollector
8. オペレーター構築
   └─ 同期: ProcessOperator(VectorSearchRunner)
   └─ 非同期: AsyncWaitOperatorFactory(AsyncVectorSearchRunner)
9. OneInputTransformation構築
   └─ 並列度は入力と同一
10. 実行（ランタイム）
    └─ 入力行からベクトルとTopK抽出
    └─ ベクトル検索実行
    └─ 結果を出力行に結合（INNER/LEFT JOIN）
```

### フローチャート

```mermaid
flowchart TD
    A[バッチ開始] --> B[入力変換取得]
    B --> C[VectorSearchテーブル取得]
    C --> D[VectorSearchRuntimeProvider取得]
    D --> E{非同期モード?}
    E -->|Yes| F[AsyncVectorSearchFunction取得]
    E -->|No| G[VectorSearchFunction取得]
    F --> H[generateAsyncVectorSearchFunction]
    G --> I[generateSyncVectorSearchFunction]
    H --> J[AsyncWaitOperatorFactory構築]
    I --> K[ProcessOperator構築]
    J --> L[OneInputTransformation構築]
    K --> L
    L --> M[入力行処理]
    M --> N[ベクトル・TopK抽出]
    N --> O[ベクトル検索実行]
    O --> P{結合タイプ?}
    P -->|INNER| Q[内部結合]
    P -->|LEFT| R[左外部結合]
    Q --> S[バッチ終了]
    R --> S
```

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

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

N/A（インメモリ処理のため、ベクトルDBへは検索のみ）

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

N/A

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| TableException | プロバイダー不整合 | 要求された検索関数をプロバイダーが提供しない | テーブルソース実装を確認 |
| RuntimeException | 検索エラー | ベクトル検索中にエラー発生 | ベクトルDBと接続を確認 |
| TimeoutException | タイムアウト | 非同期検索がタイムアウト | asyncTimeoutを調整 |

### リトライ仕様

| 項目 | 内容 |
|-----|------|
| リトライ回数 | N/A（検索エラーは再試行設計に依存） |
| リトライ間隔 | N/A |
| リトライ対象エラー | N/A |

### 障害時対応

ベクトル検索バックエンドの状態、入力ベクトルの妥当性、非同期設定（タイムアウト、バッファ容量）を確認。

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

| 項目 | 内容 |
|-----|------|
| トランザクション範囲 | N/A（ベクトル検索） |
| コミットタイミング | N/A |
| ロールバック条件 | N/A |

## パフォーマンス要件

| 項目 | 内容 |
|-----|------|
| 想定処理件数 | 入力データ量に依存 |
| 目標処理時間 | ベクトル検索バックエンドの性能に依存 |
| メモリ使用量上限 | TopK結果サイズと非同期バッファに依存 |

## 排他制御

並列実行可能。各タスクは独立してベクトル検索を実行。

## ログ出力

| ログ種別 | 出力タイミング | 出力内容 |
|---------|--------------|---------|
| DEBUG | オペレーター生成時 | テーブル識別子と設定情報 |
| ERROR | 検索エラー時 | エラーメッセージとスタックトレース |

## 監視・アラート

| 監視項目 | 閾値 | アラート先 |
|---------|-----|----------|
| 検索レイテンシ | アプリケーション定義 | Flinkメトリクス |
| 検索失敗率 | アプリケーション定義 | Flinkメトリクス |

## 備考

- Flink 2.2以降で利用可能（minPlanVersion = FlinkVersion.v2_2）
- @ExecNodeMetadataアノテーションあり（永続化対応）
- 消費する設定オプション:
  - `table.exec.async-vector-search.max-concurrent-operations`
  - `table.exec.async-vector-search.timeout`
  - `table.exec.async-vector-search.output-mode`
- CommonExecVectorSearchTableFunctionを継承
- SingleTransformationTranslatorを実装
- vectorSearchSpec.getJoinType()でINNER/LEFT JOINを選択
- TopKとsearchColumnsをパラメータとして検索関数に渡す
- isLeftOuterJoinフラグで外部結合時のnull処理を制御
- searchOutputType.getFieldCount()で検索結果のフィールド数を管理
