[RFC] Arrow format as pull-based ingestion source

[zhaih]

Is your feature request related to a problem? Please describe

OpenSearch's pull-based ingestion framework currently supports streaming sources (Kafka, Kinesis) and non-streaming/batch sources (Hive, object-store source plugin in #22556). However, most existing plugins require data to arrive as pre-formatted OpenSearch update messages, or support only a narrow set of file formats (e.g., Parquet). There is no native support for Apache Arrow, a popular columnar in-memory interchange format used across a lot of modern data ecosystems.

This RFC proposes adding Arrow as a supported ingestion format to unlock a wide range of data sources that already speak Arrow natively, including Lance, Parquet (which is already supported), Iceberg, Delta Lake etc.

Describe the solution you'd like

The solution consists of an IngestionConsumerPlugin implementation and a common interface for providing the source:

ArrowPlugin (IngestionConsumerPlugin): Registers ingestion type "ARROW" with OpenSearch's pull-based ingestion engine. It provides an ArrowConsumerFactory that creates per-shard consumers (ArrowShardConsumer). Each shard consumer:

• Accepts an ArrowReader (the standard Apache Arrow Java interface for reading columnar record batches) from a DatasetSource
• Iterates over VectorSchemaRoot batches, converting each row into an OpenSearch index operation (_id, _version, _op_type, _source)
• Handles mapping from Arrow types to opensearch index update message
• Performs shard routing

ArrowSource (extensible interface): Abstracts the data access layer. which essentially creates the ArrowReader and provide required informations such as ingestion lag information which only the source can provide.

This separation is deliberate. The Arrow conversion logic (format handling, type mapping, shard routing) is general-purpose and common across all sources. The data access layer, however, often involves organization-specific concerns: credential management, proprietary storage systems, custom authentication flows, or internal service discovery. By defining ArrowSource as a clean interface, organizations can implement private ArrowSource classes for their specific backends without forking the core plugin.

Related component

Indexing

Describe alternatives you've considered

In our case the source is Lance, so we have two options beyond the proposed plugin:

• Do an offline index build through Spark and port the index into Opensearch by using snapshot recovery. But the later part of this approach is a bit hacky and AFAIK opensearch does not have a good support for build index outside of cluster.
• External data pipeline feeding OpenSearch update API.** Instead of reading data from within the cluster, an external service (e.g., a Spark job, an Airflow DAG, or a custom microservice) could read the Arrow data source, convert rows to bulk index requests, and send them to OpenSearch's _bulk API over HTTP. This is functionally equivalent to the proposed plugin but runs outside the cluster.

I think all 3 approaches (including the proposed plugin) provides different advantages in different cases, but eventually it is better to have options to be able to directly read from a arrow-compatible sources and build index within the cluster.

Additional context

No response

[zhaih]

@varunbharadwaj @yiyuabc @msfroh

[varunbharadwaj]

Tagging @lawofcycles as well as we were discussing this few days ago

[msfroh]

I'm interested in this, though I should disclose that I was involved with @zhaih on getting this going. 😄

One area that could be interesting, and could eventually be combined with a change to the _bulk and core pull-based ingestion APIs, would be integration with Lucene's new columnar indexing APIs. If the data is being ingested in a columnar format anyway, then passing it through to the Lucene IndexWriter as columns could be pretty useful. Definitely not a v1 implementation detail, IMO, but something to consider for later.

I also want to tag @mch2, @Bukhtawar, @mgodwan, and @rishabhmaurya to see if there could be an opportunity to use this as part of the analytics engine. Also, if we have a streaming Arrow search API (especially if it supports resuming, maybe via search_after), combining that with pull-based ingestion could be a nice way of reindexing.