# 機能設計書 116-Arrowストリーミング

## 概要

本ドキュメントは、OpenSearchのApache Arrow形式でのデータストリーミング機能のSPIインターフェース設計書である。Arrow Flight RPCプラグインの基盤となるストリーミングSPIを定義する。

### 本機能の処理概要

**業務上の目的・背景**：OpenSearchが大規模データを効率的にストリーミングするための標準化されたインターフェース（SPI）を提供する必要がある。本機能はserver側に定義されたSPIインターフェース群であり、Arrow Flight RPCプラグイン等の実装によって具体化される。Producer-Consumer パターンに基づき、データの生成と消費を分離・協調させるアーキテクチャを実現する。

**機能の利用シーン**：検索結果のストリーミング転送、大規模集計結果のバッチ転送、ノード間データ転送、外部システムへのデータエクスポートで利用される。

**主要な処理内容**：
1. StreamManagerインターフェースによるストリームのライフサイクル管理
2. StreamProducerインターフェースによるデータ生成定義（スキーマ定義、バッチジョブ）
3. StreamTicketインターフェースによるストリームアクセス制御
4. StreamReaderインターフェースによるデータ消費
5. StreamTicketFactoryインターフェースによるチケット生成
6. BatchedJobとFlushSignalによるProducer-Consumer協調

**関連システム・外部連携**：Arrow Flight RPCプラグイン（No.115）が本SPIの主要な実装を提供する。

**権限による制御**：TaskIdを通じた親タスク追跡。ストリームのアクセスはStreamTicketによる認可。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | SPIインターフェース定義のため、画面に直接対応しない |

## 機能種別

データ連携 / SPI定義

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| producer | StreamProducer | Yes | データ生成ロジック | nullでないこと |
| parentTaskId | TaskId | Yes | 親タスクID | 有効なTaskIdであること |
| ticket | StreamTicket | Yes | ストリームアクセスチケット | 有効なチケットであること |

### 入力データソース

StreamProducer実装による任意のデータソース。

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| StreamTicket | StreamTicket | ストリーム識別用チケット（ticketId + nodeId） |
| VectorSchemaRoot | Generic (VectorRoot) | Apache Arrowスキーマとデータバッチ |
| StreamReader | StreamReader | データ消費用リーダー |

### 出力先

StreamReader経由でのデータ消費。

## 処理フロー

### 処理シーケンス

```
1. ストリーム登録
   └─ StreamProducerをStreamManager.registerStream()に渡す
   └─ StreamTicketが返却される
2. リーダー取得
   └─ StreamTicketを使用してStreamManager.getStreamReader()を呼び出す
   └─ Arrow資源の遅延初期化が行われる
3. データ消費
   └─ StreamReaderからVectorSchemaRootを順次読み出す
4. リソースクリーンアップ
   └─ StreamManager.close()でリソースを解放
```

### フローチャート

```mermaid
flowchart TD
    A[StreamProducer定義] --> B[StreamManager.registerStream]
    B --> C[StreamTicket取得]
    C --> D[StreamManager.getStreamReader]
    D --> E[Arrow資源の遅延初期化]
    E --> F[BatchedJob.run開始]
    F --> G[VectorSchemaRoot書き込み]
    G --> H[FlushSignal.awaitConsumption]
    H --> I{消費完了?}
    I -->|No| G
    I -->|Yes| J[次バッチまたは完了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-116-01 | 遅延初期化 | Arrow資源は最初のconsumer接続時に初期化される | getStreamReader呼び出し時 |
| BR-116-02 | チケットベースアクセス | ストリームへのアクセスはStreamTicketで制御される | 常時 |
| BR-116-03 | バッチ処理 | データはバッチ単位で生成・消費される | BatchedJob実行時 |
| BR-116-04 | ジョブデッドライン | StreamProducer.getJobDeadline()で期限を設定可能 | タイムアウト制御時 |

### 計算ロジック

特になし。データ転送のためのSPIインターフェース定義。

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

SPIインターフェース定義のため、直接のデータベース操作は行わない。

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | IllegalArgumentException | producerがnullまたはparentTaskIdが無効 | 有効な引数を渡す |
| - | IllegalStateException | ストリームがキャンセルまたはクローズ済み | 新しいストリームを登録する |

### リトライ仕様

SPI定義のため、リトライは実装側で制御する。

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

ストリーミングデータ転送のため、トランザクション管理は適用されない。

## パフォーマンス要件

- estimatedRowCount()による事前の行数推定
- FlushSignal.awaitConsumption()によるバックプレッシャー制御

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

- StreamTicketによるストリームアクセス制御
- TaskIdによる親タスク追跡

## 備考

- パッケージ: org.opensearch.arrow.spi
- 全インターフェースに@ExperimentalApiアノテーション
- ジェネリクス型（VectorRoot, Allocator）により特定のArrowバージョンへの依存を回避

---

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

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

### 推奨読解順序

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

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | StreamTicket.java | `server/src/main/java/org/opensearch/arrow/spi/StreamTicket.java` | チケットインターフェース：ticketId, nodeId, toBytes |
| 1-2 | StreamProducer.java | `server/src/main/java/org/opensearch/arrow/spi/StreamProducer.java` | プロデューサーインターフェース：createRoot, createJob, BatchedJob, FlushSignal |

**読解のコツ**: ジェネリクス型パラメータ（VectorRoot, Allocator）はApache Arrowの具体型（VectorSchemaRoot, BufferAllocator）の抽象化。SPIとして型を抽象化することで、server側からArrowライブラリへの直接依存を排除している。

#### Step 2: ストリーム管理を理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | StreamManager.java | `server/src/main/java/org/opensearch/arrow/spi/StreamManager.java` | ストリームマネージャーインターフェース全体 |

**主要処理フロー**:
- **37行目**: registerStream - プロデューサー登録とチケット発行
- **49行目**: getStreamReader - チケットによるリーダー取得
- **56行目**: getStreamTicketFactory - チケットファクトリ取得

#### Step 3: リーダーとファクトリを理解する

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | StreamReader.java | `server/src/main/java/org/opensearch/arrow/spi/StreamReader.java` | リーダーインターフェース |
| 3-2 | StreamTicketFactory.java | `server/src/main/java/org/opensearch/arrow/spi/StreamTicketFactory.java` | チケットファクトリインターフェース |

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

```
StreamManager (SPI)
    |
    +-- registerStream(StreamProducer, TaskId) -> StreamTicket
    |       |
    |       +-- StreamProducer.createRoot(Allocator) -> VectorRoot
    |       +-- StreamProducer.createJob(Allocator) -> BatchedJob
    |               |
    |               +-- BatchedJob.run(VectorRoot, FlushSignal)
    |               +-- BatchedJob.onCancel()
    |
    +-- getStreamReader(StreamTicket) -> StreamReader
    |
    +-- getStreamTicketFactory() -> StreamTicketFactory
```

### データフロー図

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

StreamProducer          --> StreamManager.registerStream()  --> StreamTicket
  (createRoot定義)                                               (ticketId + nodeId)
  (createJob定義)

StreamTicket            --> StreamManager.getStreamReader() --> StreamReader
                             (Arrow資源の遅延初期化)

BatchedJob.run()        --> VectorSchemaRoot書き込み         --> FlushSignal
  (バッチデータ生成)          +-> awaitConsumption()               (Consumer通知)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| StreamManager.java | `server/src/main/java/org/opensearch/arrow/spi/StreamManager.java` | ソース | ストリーム管理SPIインターフェース |
| StreamProducer.java | `server/src/main/java/org/opensearch/arrow/spi/StreamProducer.java` | ソース | ストリームプロデューサーSPIインターフェース |
| StreamTicket.java | `server/src/main/java/org/opensearch/arrow/spi/StreamTicket.java` | ソース | ストリームチケットSPIインターフェース |
| StreamReader.java | `server/src/main/java/org/opensearch/arrow/spi/StreamReader.java` | ソース | ストリームリーダーSPIインターフェース |
| StreamTicketFactory.java | `server/src/main/java/org/opensearch/arrow/spi/StreamTicketFactory.java` | ソース | チケットファクトリSPIインターフェース |
| package-info.java | `server/src/main/java/org/opensearch/arrow/spi/package-info.java` | ソース | パッケージ説明 |
