# 機能設計書 45-検索パイプライン管理

## 概要

本ドキュメントは、OpenSearchの検索パイプライン（Search Pipeline）機能について、検索リクエスト・レスポンスを加工するパイプラインのCRUD操作および実行管理を設計書としてまとめたものである。

### 本機能の処理概要

**業務上の目的・背景**：検索クエリの前処理（フィルタ追加、クエリ書き換え等）や検索結果の後処理（フィールド名変更、結果トリミング等）を宣言的に定義・管理する仕組みが必要である。検索パイプラインは、インジェストパイプラインの検索版として、検索リクエストと検索レスポンスに対するプロセッサチェーンを提供する。

**機能の利用シーン**：検索結果のフィルタリング・ソート・フィールド名変更などの定型的な後処理を行う場面、検索リクエストにフィルタクエリを自動付与する場面、検索フェーズ間でのデータ変換を行う場面で利用される。

**主要な処理内容**：
1. パイプラインの作成・更新（PutSearchPipeline）：プロセッサ構成を定義してパイプラインを登録
2. パイプラインの取得（GetSearchPipeline）：登録済みパイプラインの定義を取得
3. パイプラインの削除（DeleteSearchPipeline）：パイプラインを削除
4. パイプラインの実行：検索リクエスト→リクエストプロセッサ→検索実行→レスポンスプロセッサの順でプロセッサチェーンを実行

**関連システム・外部連携**：SearchPipelinePluginインターフェースを実装するプラグインから提供されるプロセッサファクトリと連携する。

**権限による制御**：パイプラインの作成・更新・削除はクラスタマネージャノードへの操作が必要。パイプラインIDは最大512バイトの制限がある。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 21 | 検索 | 補助機能 | 検索パイプラインによるリクエスト・レスポンスの加工処理 |
| 136 | 検索パイプライン取得 | 主機能 | 検索パイプラインの定義を返す処理 |
| 137 | 検索パイプライン作成・更新 | 主機能 | 検索パイプラインを作成または更新する処理 |
| 138 | 検索パイプライン削除 | 主機能 | 検索パイプラインを削除する処理 |

## 機能種別

CRUD操作 / データ連携

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| id | String | Yes（Put/Delete） | パイプラインID | 最大512バイト |
| description | String | No | パイプラインの説明 | - |
| version | Integer | No | パイプラインのバージョン | - |
| request_processors | List | No | リクエストプロセッサ定義 | - |
| response_processors | List | No | レスポンスプロセッサ定義 | - |
| phase_results_processors | List | No | フェーズ結果プロセッサ定義 | - |
| pipeline | String | No（Search） | 検索時に使用するパイプラインID | - |

### 入力データソース

REST API経由でのリクエスト。パイプライン定義はJSON形式で指定する。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| acknowledged | boolean | 操作が承認されたか（Put/Delete） |
| pipelines | List<PipelineConfiguration> | パイプライン定義一覧（Get） |
| SearchResponse | SearchResponse | パイプライン適用後の検索レスポンス（Search実行時） |

### 出力先

REST APIレスポンス（JSON形式）

## 処理フロー

### 処理シーケンス

```
1. パイプライン作成（Put）
   └─ SearchPipelineService.putPipeline() → ClusterState更新
2. パイプライン実行（Search時）
   └─ SearchPipelineService.resolvePipeline() → PipelinedRequest生成
3. リクエスト変換
   └─ Pipeline.transformRequest() → SearchRequestProcessorチェーン実行
4. 検索実行
   └─ 通常の検索処理
5. レスポンス変換
   └─ Pipeline.transformResponseListener() → SearchResponseProcessorチェーン実行
6. フェーズ結果処理
   └─ Pipeline.runSearchPhaseResultsTransformer() → フェーズ間プロセッサ実行
```

### フローチャート

```mermaid
flowchart TD
    A[Search Request] --> B{Pipeline specified?}
    B -->|No| C[Direct Search]
    B -->|Yes| D[Resolve Pipeline]
    D --> E[Request Processors Chain]
    E --> F[Execute Search]
    F --> G[Response Processors Chain]
    G --> H[Return Response]

    I[Put Pipeline API] --> J[Validate Pipeline ID]
    J --> K[Parse Processor Definitions]
    K --> L[Update ClusterState]

    M[Delete Pipeline API] --> N[Remove from ClusterState]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-45-01 | パイプラインIDサイズ制限 | パイプラインIDは最大512バイト | Put操作時 |
| BR-45-02 | NOOPパイプライン | `_none`パイプラインIDは変換を行わないNOOPパイプライン | 検索時 |
| BR-45-03 | アドホックパイプライン | `_ad_hoc_pipeline`は一時的なパイプライン定義に使用 | 検索リクエスト内で定義時 |
| BR-45-04 | プロセッサ失敗時の動作 | ignore_failure=trueまたはverbosePipeline=trueの場合、プロセッサ失敗を無視して次へ進む | プロセッサ実行時 |
| BR-45-05 | ワイルドカードマッチ | パイプラインID取得時にワイルドカード`*`が使用可能 | Get操作時 |

### 計算ロジック

各プロセッサの実行時間はナノ秒単位で計測され、PipelineWithMetricsで統計が収集される。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| Put | ClusterState.Metadata (SearchPipelineMetadata) | INSERT/UPDATE | パイプライン定義をクラスタ状態に保存 |
| Delete | ClusterState.Metadata (SearchPipelineMetadata) | DELETE | パイプライン定義をクラスタ状態から削除 |
| Get | ClusterState.Metadata (SearchPipelineMetadata) | SELECT | パイプライン定義をクラスタ状態から取得 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | OpenSearchParseException | パイプライン定義のパースエラー | JSON構文を確認 |
| - | ResourceNotFoundException | 指定パイプラインが存在しない | パイプラインIDを確認 |
| - | SearchPipelineProcessingException | プロセッサ実行時のエラー | プロセッサ設定を確認 |

### リトライ仕様

クラスタマネージャノードタスクとして実行されるため、タスクスロットリングによるリトライが適用される。

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

パイプラインの作成・更新・削除はAckedClusterStateUpdateTaskとして原子的に実行される。

## パフォーマンス要件

プロセッサチェーンの実行は非同期（ActionListener）で行われ、各プロセッサの実行時間が個別に計測される。

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

- パイプラインの作成・更新・削除にはクラスタ管理権限が必要
- ScriptRequestProcessorなどスクリプト実行を含むプロセッサは追加のセキュリティ考慮が必要

## 備考

Search Pipelineはインジェストパイプラインとは独立した機能であり、検索フェーズに特化したプロセッサチェーンを提供する。3種類のプロセッサ（SearchRequestProcessor、SearchResponseProcessor、SearchPhaseResultsProcessor）をサポートする。

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | SearchRequestProcessor.java | `server/src/main/java/org/opensearch/search/pipeline/SearchRequestProcessor.java` | 検索リクエストプロセッサインターフェース |
| 1-2 | SearchResponseProcessor.java | `server/src/main/java/org/opensearch/search/pipeline/SearchResponseProcessor.java` | 検索レスポンスプロセッサインターフェース |
| 1-3 | SearchPhaseResultsProcessor.java | `server/src/main/java/org/opensearch/search/pipeline/SearchPhaseResultsProcessor.java` | 検索フェーズ結果プロセッサインターフェース |
| 1-4 | PipelineConfiguration.java | `server/src/main/java/org/opensearch/search/pipeline/PipelineConfiguration.java` | パイプライン設定データ構造 |
| 1-5 | SearchPipelineMetadata.java | `server/src/main/java/org/opensearch/search/pipeline/SearchPipelineMetadata.java` | クラスタ状態に保存されるパイプラインメタデータ |

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | SearchPipelineService.java | `server/src/main/java/org/opensearch/search/pipeline/SearchPipelineService.java` | 検索パイプラインのCRUD操作と実行を管理するメインサービス |

**主要処理フロー**:
- **77行目**: ClusterStateApplierを実装し、クラスタ状態変更時にパイプラインを再構築
- **79-83行目**: 定数定義（SEARCH_PIPELINE_ORIGIN, AD_HOC_PIPELINE_ID, NOOP_PIPELINE_ID, MAX_PIPELINE_ID_BYTES）
- **87-98行目**: プロセッサファクトリマップの保持

#### Step 3: パイプライン実行ロジックを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | Pipeline.java | `server/src/main/java/org/opensearch/search/pipeline/Pipeline.java` | パイプラインの実体。プロセッサチェーンの実行ロジック |
| 3-2 | PipelinedRequest.java | `server/src/main/java/org/opensearch/search/pipeline/PipelinedRequest.java` | パイプラインが適用された検索リクエスト |

**主要処理フロー（Pipeline.java）**:
- **31-33行目**: プロセッサキー定数（request_processors, response_processors, phase_results_processors）
- **117-172行目**: transformRequestメソッド。ActionListenerチェーンを逆順に構築してプロセッサを順次実行
- **189-255行目**: transformResponseListenerメソッド。レスポンスプロセッサチェーンを構築
- **257-290行目**: runSearchPhaseResultsTransformerメソッド。フェーズ間プロセッサの実行

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

```
SearchPipelineService
    |
    +-- putPipeline() → AckedClusterStateUpdateTask → ClusterState更新
    |
    +-- deletePipeline() → AckedClusterStateUpdateTask → ClusterState更新
    |
    +-- resolvePipeline() → PipelinedRequest生成
    |
    +-- Pipeline
            |
            +-- transformRequest() → SearchRequestProcessorチェーン実行
            |       +-- SearchRequestProcessor.processRequestAsync()
            |
            +-- transformResponseListener() → SearchResponseProcessorチェーン実行
            |       +-- SearchResponseProcessor.processResponseAsync()
            |
            +-- runSearchPhaseResultsTransformer() → フェーズ結果プロセッサ実行
                    +-- SearchPhaseResultsProcessor.process()
```

### データフロー図

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

SearchRequest ───────▶ Pipeline.transformRequest()      ───▶ Modified SearchRequest
                           |
                           +── RequestProcessor 1
                           +── RequestProcessor 2
                           +── ...
                           ▼
                       Search Execution
                           ▼
SearchResponse ◀──── Pipeline.transformResponseListener() ◀── Raw SearchResponse
                           |
                           +── ResponseProcessor 1
                           +── ResponseProcessor 2
                           +── ...
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| SearchPipelineService.java | `server/src/main/java/org/opensearch/search/pipeline/` | ソース | パイプラインCRUD・実行管理の中核サービス |
| Pipeline.java | 同上 | ソース | プロセッサチェーン実行ロジック |
| PipelineWithMetrics.java | 同上 | ソース | メトリクス付きパイプライン |
| PipelinedRequest.java | 同上 | ソース | パイプライン適用済みリクエスト |
| SearchRequestProcessor.java | 同上 | ソース | リクエストプロセッサインターフェース |
| SearchResponseProcessor.java | 同上 | ソース | レスポンスプロセッサインターフェース |
| SearchPhaseResultsProcessor.java | 同上 | ソース | フェーズ結果プロセッサインターフェース |
| PipelineConfiguration.java | 同上 | ソース | パイプライン設定データ構造 |
| SearchPipelineMetadata.java | 同上 | ソース | クラスタ状態用メタデータ |
| SearchPipelineStats.java | 同上 | ソース | パイプライン統計情報 |
| Processor.java | 同上 | ソース | プロセッサ基底インターフェース |
| AbstractProcessor.java | 同上 | ソース | プロセッサ抽象基底クラス |
