# 機能設計書 33-パイプライン集計

## 概要

本ドキュメントは、OpenSearchのパイプライン集計（Pipeline Aggregation）機能の設計を記述する。パイプライン集計は、他の集計結果を入力として二次的な計算を行う集計機能である。

### 本機能の処理概要

パイプライン集計は、バケット集計やメトリクス集計の出力結果を入力として使用し、移動平均、累積和、微分、バケット選択、バケットスクリプト等の二次的な計算を行う機能である。他の集計のbuckets_pathを参照して動作する点が特徴である。

**業務上の目的・背景**：時系列データの傾向分析（移動平均）、累積値の計算（累積和）、変化率の算出（微分）、条件に基づくバケットフィルタリング（バケットセレクタ）等、集計結果に対する高度な分析処理を提供する。ダッシュボードでのトレンド可視化やアラート条件の定義に利用される。

**機能の利用シーン**：検索APIのaggregationsパラメータ内で、バケット集計の兄弟または子としてパイプライン集計を定義して利用する。buckets_pathで入力となる集計のパスを指定する。

**主要な処理内容**：
1. 検索リクエストからパイプライン集計定義をパース（PipelineAggregationBuilder）
2. buckets_pathの検証と参照先集計の解決
3. reduce（マージ）フェーズでパイプライン集計を実行
4. 親バケット集計の結果に対してパイプライン処理を適用
5. 最終結果をレスポンスに含めて返却

**関連システム・外部連携**：他の集計結果への依存。buckets_pathによる集計間参照。

**権限による制御**：検索APIの権限に従う。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 21 | 検索 | 補助機能 | 検索結果の集計に対するパイプライン集計処理 |

## 機能種別

計算処理（集計・分析）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| type | String | Yes | パイプライン集計タイプ（avg_bucket, sum_bucket, min_bucket, max_bucket, stats_bucket, extended_stats_bucket, percentiles_bucket, moving_avg, cumulative_sum, derivative, bucket_script, bucket_selector, bucket_sort, serial_diff等） | サポートされるタイプ名 |
| buckets_path | String/Map | Yes | 入力元の集計パス | 有効な集計パス |
| gap_policy | String | No | 欠損バケットの処理方針（skip/insert_zeros）デフォルトskip | skip またはinsert_zeros |
| format | String | No | 出力値のフォーマット | 有効なフォーマット文字列 |
| window | Integer | No | 移動平均のウィンドウサイズ | 正の整数 |
| model | String | No | 移動平均モデル（simple/linear/ewma/holt/holt_winters） | サポートされるモデル名 |
| script | Script | No | バケットスクリプトの処理内容 | 有効なスクリプト |
| lag | Integer | No | Serial Differencing のラグ値 | 正の整数 |

### 入力データソース

検索APIレスポンス内の他の集計結果。buckets_pathで指定された集計のバケット値。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| value | Double | パイプライン集計の算出値 |
| values | Map | パーセンタイルバケット等の複数値 |
| normalized_value | Double | 微分の正規化値 |

### 出力先

検索APIレスポンスのaggregationsフィールド内に、親バケットの一部として返却。

## 処理フロー

### 処理シーケンス

```
1. 検索リクエスト受信
   └─ PipelineAggregationBuilderをパース
2. バリデーション
   └─ buckets_pathの参照先が存在することを検証
3. 通常の集計フェーズ実行
   └─ バケット集計・メトリクス集計が先に実行される
4. reduceフェーズでパイプライン集計実行
   └─ PipelineAggregator.reduce()で二次計算を実行
5. 結果をバケット内に追加
   └─ InternalSimpleValue等として結果を構築
6. レスポンス返却
```

### フローチャート

```mermaid
flowchart TD
    A[検索リクエスト受信] --> B[PipelineAggregationBuilderパース]
    B --> C[buckets_pathバリデーション]
    C --> D[通常集計フェーズ実行]
    D --> E[バケット/メトリクス結果取得]
    E --> F[PipelineAggregator.reduce実行]
    F --> G{gap_policy}
    G -->|skip| H[欠損バケットをスキップ]
    G -->|insert_zeros| I[ゼロ値を挿入]
    H --> J[二次計算実行]
    I --> J
    J --> K[結果をバケットに追加]
    K --> L[レスポンス返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-33-01 | buckets_path必須 | パイプライン集計は必ずbuckets_pathで入力元を指定 | 全パイプライン集計 |
| BR-33-02 | gap_policyデフォルト | 欠損バケットのデフォルト処理はskip | 全パイプライン集計 |
| BR-33-03 | reduceフェーズ実行 | パイプライン集計はreduceフェーズでのみ実行される | 全パイプライン集計 |
| BR-33-04 | 兄弟パイプライン | sibling pipelineは同レベルの集計結果を入力とする | avg_bucket, sum_bucket等 |
| BR-33-05 | 親パイプライン | parent pipelineは親バケットの各バケット結果を入力とする | derivative, cumulative_sum等 |

### 計算ロジック

- **Derivative**: `derivative = current_value - previous_value`（正規化時は間隔で除算）
- **Cumulative Sum**: `cumulative[i] = sum(values[0..i])`
- **Moving Average (Simple)**: `moving_avg[i] = avg(values[max(0, i-window+1)..i])`
- **Moving Average (EWMA)**: `ewma[i] = alpha * value[i] + (1 - alpha) * ewma[i-1]`
- **Bucket Selector**: スクリプト条件がfalseのバケットを除外
- **Bucket Script**: スクリプトで算出した値を各バケットに追加

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| パイプライン集計 | なし（インメモリ） | 計算 | 他の集計結果に対する二次計算 |

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

パイプライン集計はインメモリで他の集計結果を入力として処理するため、直接的なデータベース操作はない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| 400 | illegal_argument_exception | buckets_pathが不正 | 有効な集計パスを指定 |
| 400 | search_phase_execution_exception | 参照先集計が存在しない | 正しい集計名をパスに指定 |
| 400 | illegal_argument_exception | windowサイズが不正 | 正の整数を指定 |

### リトライ仕様

読み取り専用操作のため、クライアント側でのリトライが可能。

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

読み取り専用操作のためトランザクション制御は不要。

## パフォーマンス要件

- パイプライン集計はreduceフェーズでインメモリ処理のため、バケット数に比例した計算量
- Moving Average集計のウィンドウサイズが大きいとメモリ使用量が増加

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

- Bucket Script/Selectorでスクリプトが実行されるため、スクリプトの安全性に注意
- 検索APIの権限に従う

## 備考

- パイプライン集計は集計定義のバリデーション時にbuckets_pathの整合性がチェックされる
- Moving Average集計では複数のモデル（Simple, Linear, EWMA, Holt, Holt-Winters）が選択可能
- AbstractPipelineAggregationBuilderがBUCKETS_PATH_FIELDを共通パースフィールドとして提供

---

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

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

### 推奨読解順序

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

パイプライン集計の基盤となる抽象クラスを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | AbstractPipelineAggregationBuilder.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/AbstractPipelineAggregationBuilder.java` | パイプライン集計ビルダーの基底クラス。BUCKETS_PATH_FIELDの定義 |
| 1-2 | BucketHelpers.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/BucketHelpers.java` | GapPolicy定義、バケット値取得ヘルパー |

**読解のコツ**: パイプライン集計はPipelineAggregationBuilder（ビルド時）とPipelineAggregator（実行時）の2つのクラスで構成される。実行はreduceフェーズで行われる点が通常の集計と異なる。

#### Step 2: 代表的なパイプライン集計（Derivative）を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | DerivativePipelineAggregationBuilder.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/DerivativePipelineAggregationBuilder.java` | 微分パイプライン集計のビルダー |
| 2-2 | DerivativePipelineAggregator.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/DerivativePipelineAggregator.java` | 微分パイプライン集計の実行ロジック |

**主要処理フロー**:
- **50-51行目**: AbstractPipelineAggregationBuilderクラス定義。PipelineAggregationBuilderを継承
- **56行目**: BUCKETS_PATH_FIELD = new ParseField("buckets_path")
- **61行目**: コンストラクタでname, type, bucketsPaths配列を受け取り

#### Step 3: 移動平均モデルを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | MovAvgPipelineAggregationBuilder.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/MovAvgPipelineAggregationBuilder.java` | 移動平均集計ビルダー |
| 3-2 | EwmaModel.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/EwmaModel.java` | EWMA移動平均モデル |
| 3-3 | HoltLinearModel.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/HoltLinearModel.java` | Holt線形モデル |
| 3-4 | HoltWintersModel.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/HoltWintersModel.java` | Holt-Wintersモデル |

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

```
SearchAction (REST API)
    |
    +-- SearchService.executeQueryPhase()
    |     +-- 通常集計フェーズ（バケット/メトリクス）
    |
    +-- SearchPhaseController.reducedQueryPhase()
          +-- InternalAggregations.topLevelReduce()
                +-- PipelineAggregator.reduce()
                |     +-- DerivativePipelineAggregator.reduce()
                |     +-- CumulativeSumPipelineAggregator.reduce()
                |     +-- BucketScriptPipelineAggregator.reduce()
                |     +-- BucketSelectorPipelineAggregator.reduce()
                |
                +-- InternalSimpleValue / InternalDerivative 構築
```

### データフロー図

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

バケット集計結果     +--> buckets_pathで参照先解決
(バケットリスト)     |
                     +--> GapPolicy適用
メトリクス集計結果   |     (skip/insert_zeros)
(各バケットの値)     |
                     +--> PipelineAggregator.reduce()
                     |     +-- 微分/累積和/移動平均等
                     |
                     +--> InternalAggregation構築    +--> 検索レスポンス
                           (パイプライン結果)               (親バケット内)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| AbstractPipelineAggregationBuilder.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/AbstractPipelineAggregationBuilder.java` | ソース | パイプライン集計ビルダー基底 |
| BucketHelpers.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/BucketHelpers.java` | ソース | バケットヘルパー・GapPolicy |
| DerivativePipelineAggregationBuilder.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/DerivativePipelineAggregationBuilder.java` | ソース | 微分集計ビルダー |
| DerivativePipelineAggregator.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/DerivativePipelineAggregator.java` | ソース | 微分集計実行 |
| CumulativeSumPipelineAggregator.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/CumulativeSumPipelineAggregator.java` | ソース | 累積和集計実行 |
| MovAvgPipelineAggregationBuilder.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/MovAvgPipelineAggregationBuilder.java` | ソース | 移動平均集計ビルダー |
| BucketScriptPipelineAggregator.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/BucketScriptPipelineAggregator.java` | ソース | バケットスクリプト実行 |
| BucketSelectorPipelineAggregator.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/BucketSelectorPipelineAggregator.java` | ソース | バケットセレクタ実行 |
| BucketSortPipelineAggregator.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/BucketSortPipelineAggregator.java` | ソース | バケットソート実行 |
| EwmaModel.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/EwmaModel.java` | ソース | EWMA移動平均モデル |
| HoltWintersModel.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/HoltWintersModel.java` | ソース | Holt-Winters季節モデル |
| InternalSimpleValue.java | `server/src/main/java/org/opensearch/search/aggregations/pipeline/InternalSimpleValue.java` | ソース | 単一値パイプライン結果 |
