# 機能設計書 13-バルク操作

## 概要

本ドキュメントは、OpenSearchの「バルク操作」機能の設計を記述する。複数のインデックス・更新・削除操作を一括で実行するBulk APIの処理仕様を定義する。

### 本機能の処理概要

本機能は、複数のドキュメント操作（index/create/update/delete）を単一のAPIリクエストで一括実行する機能である。REST APIの `POST /_bulk` または `POST /{index}/_bulk` エンドポイントを通じて呼び出される。

**業務上の目的・背景**：大量のドキュメント操作を個別のAPIリクエストで行うとネットワークラウンドトリップのオーバーヘッドが大きくなる。バルク操作により複数操作をまとめて送信し、スループットを大幅に向上させることができる。データインポート、ログ収集、ETL処理など高ボリュームのデータ操作において必須の機能である。

**機能の利用シーン**：大量データの初期ロード、ログデータのバッチインジェスト、定期的なデータ同期処理、アプリケーションからの複数ドキュメント一括操作で利用される。

**主要な処理内容**：
1. NDJSONフォーマットのバルクリクエスト解析
2. インジェストパイプラインの適用（該当する場合）
3. 存在しないインデックスの自動作成（auto_create_index設定に基づく）
4. リクエストアイテムのシャード単位グループ化
5. シャード単位でのバルク書き込み実行（プライマリ→レプリカ）
6. 個別操作結果の集約とレスポンス構築

**関連システム・外部連携**：インジェストパイプラインとの連携（ドキュメント登録前の前処理）、インデックス自動作成機能との連携。

**権限による制御**：アクション名 `indices:data/write/bulk` に基づくセキュリティプラグインによるアクセス制御が適用される。個別操作ごとのインデックスレベルの権限チェックも行われる。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 12 | バルク操作 | 主機能 | 複数のindex/update/delete操作を一括実行する主処理 |

## 機能種別

CRUD操作（Create/Update/Delete一括処理）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| requests | List<DocWriteRequest<?>> | Yes | 個別操作リクエストのリスト | 空リスト不可 |
| globalIndex | String | No | 全操作に適用するデフォルトインデックス名 | - |
| globalRouting | String | No | 全操作に適用するデフォルトルーティング | - |
| globalPipeline | String | No | 全操作に適用するデフォルトパイプライン | - |
| globalRequireAlias | Boolean | No | エイリアス必須フラグ | - |
| timeout | TimeValue | No | 書き込みタイムアウト（デフォルト: 1m） | - |
| waitForActiveShards | ActiveShardCount | No | アクティブシャード待機数 | - |
| refreshPolicy | RefreshPolicy | No | リフレッシュポリシー（NONE/IMMEDIATE/WAIT_UNTIL） | バルク全体に適用、個別アイテムでは指定不可 |

### 入力データソース

REST APIリクエスト（HTTP POST、NDJSON形式ボディ）、Java Transport Client

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| took | long | 処理にかかった時間（ミリ秒） |
| errors | boolean | いずれかの操作でエラーが発生したか |
| items | BulkItemResponse[] | 個別操作の結果配列 |
| items[].index/_id/_version/result/status | 各種 | 個別操作の詳細結果 |
| items[].error | Object | エラー発生時のエラー情報 |

### 出力先

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

## 処理フロー

### 処理シーケンス

```
1. リクエスト受信・パース
   └─ NDJSON形式のバルクリクエストボディを解析
2. バリデーション
   └─ 各操作アイテムのバリデーション（空リスト不可、個別アイテム検証）
3. インジェストパイプライン判定
   └─ index/create操作にパイプラインが設定されている場合、インジェスト処理を実行
4. インデックス自動作成
   └─ 存在しないインデックスへの操作がある場合、auto_create_index設定に基づき自動作成
5. シャード単位グループ化
   └─ 各操作をルーティングに基づいて対象シャードごとにグループ化
6. シャード単位バルク実行
   └─ TransportShardBulkActionを呼び出し、各シャードで一括処理
7. 結果集約・レスポンス構築
   └─ 全シャードの処理結果をAtomicArrayに集約しBulkResponse生成
```

### フローチャート

```mermaid
flowchart TD
    A[Bulk リクエスト受信] --> B{バリデーション OK?}
    B -->|No| C[バリデーションエラー返却]
    B -->|Yes| D{インジェストパイプライン必要?}
    D -->|Yes| E[インジェスト処理実行]
    D -->|No| F[インデックス自動作成判定]
    E --> F
    F --> G{未存在インデックスあり?}
    G -->|Yes| H[AutoCreateAction 実行]
    G -->|No| I[シャード単位にグループ化]
    H --> I
    I --> J[TransportShardBulkAction 実行]
    J --> K[プライマリシャードで処理]
    K --> L[レプリカへ複製]
    L --> M[結果集約]
    M --> N[BulkResponse 返却]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-13-01 | 部分的成功 | バルク操作は全体としてアトミックではなく、個別操作が独立して成功/失敗する | 常時 |
| BR-13-02 | リフレッシュポリシー | refreshはバルクリクエスト全体に適用され、個別アイテムごとには指定できない | 常時 |
| BR-13-03 | インジェストパイプライン | index/create操作にのみインジェストパイプラインが適用される。delete操作には適用されない | index/create操作時 |
| BR-13-04 | インデックス自動作成 | auto_create_index設定がtrueの場合、存在しないインデックスへの操作時に自動作成 | auto_create_index=true |
| BR-13-05 | インジェストフォワーディング | インジェストが必要な場合、インジェストノードへリクエストが転送される | パイプライン設定あり |

### 計算ロジック

処理時間（took）はリクエスト受信から全シャードの処理完了までの経過時間。relativeTimeProviderにより計測される。

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| index操作 | Lucene Index | INSERT/UPDATE | ドキュメントの登録/上書き |
| create操作 | Lucene Index | INSERT | 新規ドキュメントの作成（既存ID時は失敗） |
| update操作 | Lucene Index | UPDATE | 既存ドキュメントの部分更新 |
| delete操作 | Lucene Index | DELETE | ドキュメントの論理削除 |

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

#### Lucene Index

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| INDEX/CREATE | _source, _id | リクエストボディのJSON | シャード単位で実行 |
| DELETE | _id | _id = リクエストID | 論理削除 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| 400 | ActionRequestValidationException | リクエストリストが空 | 1件以上の操作を含める |
| 400 | OpenSearchParseException | NDJSONの解析エラー | リクエストフォーマットを修正 |
| 409 | VersionConflictEngineException | 個別操作でバージョン不一致 | 個別アイテムのエラーとして報告 |
| 404 | IndexNotFoundException | 自動作成無効で存在しないインデックスへの操作 | インデックスを事前作成 |
| 429 | OpenSearchRejectedExecutionException | インデキシングプレッシャー超過 | スロットリングして再試行 |

### リトライ仕様

BulkProcessorクラスがリトライ機能を提供する。BackoffPolicyに基づく指数バックオフリトライが可能。Retryクラスで明示的なリトライポリシーを設定できる。

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

バルク操作全体はアトミックではない。個別操作がそれぞれ独立して実行され、一部の操作が失敗しても他の操作は正常に処理される。各操作の結果はレスポンスのitems配列で個別に確認する必要がある。

## パフォーマンス要件

- IndexingPressureServiceによりインデキシング圧力が監視され、過負荷時は429エラーが返却される
- シャード単位でのグループ化によりネットワークラウンドトリップを最小化
- BulkProcessorを利用したクライアント側バッチ処理が推奨される
- REQUEST_OVERHEAD（50バイト）が各リクエストアイテムのメモリ見積もりに加算される

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

- アクション名 `indices:data/write/bulk` に対するセキュリティ権限チェック
- SystemIndicesによるシステムインデックスへの書き込み制御
- 各操作のインデックスごとに個別の権限チェックが行われる

## 備考

- BulkRequestはAccountableインターフェースを実装し、メモリ使用量の追跡が可能
- NDJSON（Newline Delimited JSON）形式でのリクエスト解析はBulkRequestParserが担当
- TransportBulkActionはTracerを利用したテレメトリ情報の収集に対応

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | BulkRequest.java | `server/src/main/java/org/opensearch/action/bulk/BulkRequest.java` | バルクリクエスト構造。List<DocWriteRequest<?>>、グローバルパラメータ（pipeline, routing, index） |
| 1-2 | BulkResponse.java | `server/src/main/java/org/opensearch/action/bulk/BulkResponse.java` | レスポンス構造。BulkItemResponse配列、took、errors |
| 1-3 | BulkItemRequest.java | `server/src/main/java/org/opensearch/action/bulk/BulkItemRequest.java` | 個別操作アイテムのラッパー |
| 1-4 | BulkItemResponse.java | `server/src/main/java/org/opensearch/action/bulk/BulkItemResponse.java` | 個別操作の結果。成功時はDocWriteResponse、失敗時はFailure |

**読解のコツ**: BulkRequestはCompositeIndicesRequest, WriteRequest<BulkRequest>, Accountableを実装。Accountableの実装によりメモリ使用量の追跡が可能。REQUEST_OVERHEAD=50バイトが各アイテムのオーバーヘッドとして加算される。

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | TransportBulkAction.java | `server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java` | バルク操作の中枢。HandledTransportActionを継承。インジェスト判定、自動インデックス作成、シャードグループ化 |

**主要処理フロー**:
1. **128-130行目**: クラス定義。HandledTransportActionとTransportIndicesResolvingActionの実装
2. **134-147行目**: 依存オブジェクト（AutoCreateIndex, IngestService, ClusterService, IndexingPressureService等）
3. **149行目以降**: コンストラクタでの依存注入

#### Step 3: シャードレベル処理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | TransportShardBulkAction.java | `server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java` | シャード単位のバルク実行。プライマリでの処理とレプリカへの複製 |
| 3-2 | BulkPrimaryExecutionContext.java | `server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java` | プライマリシャードでの実行コンテキスト管理 |

#### Step 4: 補助機能を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | BulkRequestParser.java | `server/src/main/java/org/opensearch/action/bulk/BulkRequestParser.java` | NDJSONリクエストの解析 |
| 4-2 | BulkProcessor.java | `server/src/main/java/org/opensearch/action/bulk/BulkProcessor.java` | クライアント側バルク処理ユーティリティ |
| 4-3 | Retry.java | `server/src/main/java/org/opensearch/action/bulk/Retry.java` | リトライロジック |
| 4-4 | BackoffPolicy.java | `server/src/main/java/org/opensearch/action/bulk/BackoffPolicy.java` | バックオフポリシー定義 |

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

```
REST API (POST /_bulk)
    |
    +-- TransportBulkAction
           |
           +-- インジェストパイプライン判定
           |      └─ IngestService.executeBulkRequest()
           |
           +-- インデックス自動作成
           |      └─ AutoCreateAction
           |
           +-- シャード単位グループ化
           |
           +-- TransportShardBulkAction (シャードごと)
                  |
                  +-- BulkPrimaryExecutionContext
                  |      +-- Engine.index() / Engine.delete()
                  |
                  +-- レプリカ複製
```

### データフロー図

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

BulkRequest          -->  TransportBulkAction               -->  BulkResponse
 - requests[]              |                                      - took
 - globalIndex             +-> IngestService (optional)           - errors
 - globalPipeline          +-> AutoCreateAction (optional)        - items[]
 - globalRouting           +-> シャードグループ化                     - status
 - refreshPolicy           +-> TransportShardBulkAction             - result
 - timeout                      +-> Engine operations               - error
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| BulkAction.java | `server/src/main/java/org/opensearch/action/bulk/BulkAction.java` | ソース | アクション型定義 |
| BulkRequest.java | `server/src/main/java/org/opensearch/action/bulk/BulkRequest.java` | ソース | バルクリクエスト構造 |
| BulkResponse.java | `server/src/main/java/org/opensearch/action/bulk/BulkResponse.java` | ソース | バルクレスポンス構造 |
| BulkItemRequest.java | `server/src/main/java/org/opensearch/action/bulk/BulkItemRequest.java` | ソース | 個別アイテムリクエスト |
| BulkItemResponse.java | `server/src/main/java/org/opensearch/action/bulk/BulkItemResponse.java` | ソース | 個別アイテムレスポンス |
| TransportBulkAction.java | `server/src/main/java/org/opensearch/action/bulk/TransportBulkAction.java` | ソース | バルク処理のオーケストレーション |
| TransportShardBulkAction.java | `server/src/main/java/org/opensearch/action/bulk/TransportShardBulkAction.java` | ソース | シャード単位のバルク処理 |
| BulkPrimaryExecutionContext.java | `server/src/main/java/org/opensearch/action/bulk/BulkPrimaryExecutionContext.java` | ソース | プライマリ実行コンテキスト |
| BulkShardRequest.java | `server/src/main/java/org/opensearch/action/bulk/BulkShardRequest.java` | ソース | シャード単位バルクリクエスト |
| BulkShardResponse.java | `server/src/main/java/org/opensearch/action/bulk/BulkShardResponse.java` | ソース | シャード単位バルクレスポンス |
| BulkRequestParser.java | `server/src/main/java/org/opensearch/action/bulk/BulkRequestParser.java` | ソース | NDJSONパーサー |
| BulkProcessor.java | `server/src/main/java/org/opensearch/action/bulk/BulkProcessor.java` | ソース | クライアント側バッチ処理ユーティリティ |
| BulkRequestHandler.java | `server/src/main/java/org/opensearch/action/bulk/BulkRequestHandler.java` | ソース | バルクリクエストハンドラー |
| Retry.java | `server/src/main/java/org/opensearch/action/bulk/Retry.java` | ソース | リトライロジック |
| BackoffPolicy.java | `server/src/main/java/org/opensearch/action/bulk/BackoffPolicy.java` | ソース | バックオフポリシー |
| MappingUpdatePerformer.java | `server/src/main/java/org/opensearch/action/bulk/MappingUpdatePerformer.java` | ソース | マッピング自動更新 |
| TransportSingleItemBulkWriteAction.java | `server/src/main/java/org/opensearch/action/bulk/TransportSingleItemBulkWriteAction.java` | ソース | 単一アイテムバルク書き込み（Delete/Index用） |
