arrow-avro: Deprecate object_store integration

arrow-avro/Cargo.toml

@@ -48,6 +48,9 @@ avro_custom_types = ["dep:arrow-select"]
 # Enable async APIs
 async = ["futures", "tokio"]
 # Enable object_store integration
+# Deprecated: implement `AsyncFileReader` directly instead, see the example on
+# the `AsyncFileReader` trait documentation and `arrow-avro/examples/object_store.rs`.
+# This feature will be removed in a future release.
 object_store = ["dep:object_store", "async"]
 
 [dependencies]
@@ -98,6 +101,11 @@ once_cell = "1.21.3"
 half = { version = "2.1", default-features = false }
 tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs"] }
 
+[[example]]
+name = "object_store"
+required-features = ["async"]
+path = "./examples/object_store.rs"
+
 [[bench]]
 name = "avro_reader"
 harness = false

arrow-avro/README.md

@@ -105,25 +105,18 @@ fn main() -> anyhow::Result<()> {
 
 See the crate docs for runnable SOE and Confluent round‑trip examples.
 
-### Async reading from object stores (`object_store` feature)
+### Async reading (`async` feature)
 
 ```rust,ignore
-use std::sync::Arc;
-use arrow_avro::reader::{AsyncAvroFileReader, AvroObjectReader};
+use arrow_avro::reader::AsyncAvroFileReader;
 use futures::TryStreamExt;
-use object_store::ObjectStore;
-use object_store::local::LocalFileSystem;
-use object_store::path::Path;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
-    let store: Arc<dyn ObjectStore> = Arc::new(LocalFileSystem::new());
-    let path = Path::from("data/example.avro");
-
-    let meta = store.head(&path).await?;
-    let reader = AvroObjectReader::new(store, path);
+    let file = tokio::fs::File::open("data/example.avro").await?;
+    let file_size = file.metadata().await?.len();
 
-    let stream = AsyncAvroFileReader::builder(reader, meta.size, 1024)
+    let stream = AsyncAvroFileReader::builder(file, file_size, 1024)
         .try_build()
         .await?;
 
@@ -135,6 +128,13 @@ async fn main() -> anyhow::Result<()> {
 }
 ```
 
+Any `AsyncFileReader` implementation can be used as the source, so object
+storage services (S3, GCS, Azure Blob, etc.) can be integrated by implementing
+`AsyncFileReader` on top of a client such as the [`object_store`] crate. See
+the example on the `AsyncFileReader` trait documentation.
+
+[`object_store`]: https://docs.rs/object_store/latest/object_store/
+
 ---
 
 ## Feature Flags (what they do and when to use them)
@@ -158,12 +158,12 @@ async fn main() -> anyhow::Result<()> {
 * Only **OCF** uses these codecs (they compress per‑block). They do **not** apply to raw Avro frames used by Confluent wire format or SOE. The crate’s `compression` module is specifically for **OCF blocks**.
 * `deflate` uses `flate2` with the `rust_backend` (no system zlib required).
 
-### Async & Object Store
+### Async
 
 | Feature        | Default | What it enables                                                             | When to use                                                                   |
 |----------------|--------:|-----------------------------------------------------------------------------|-------------------------------------------------------------------------------|
-| `async`        |       ⬜ | Async APIs for reading Avro via `futures` and `tokio`                       | Enable for non-blocking async Avro reading with `AsyncAvroFileReader`.        |
-| `object_store` |       ⬜ | Integration with `object_store` crate (implies `async`)                     | Enable for reading Avro from cloud storage (S3, GCS, Azure Blob, etc.).       |
+| `async`        |       ⬜ | Async APIs for reading Avro via `futures` and `tokio`                       | Enable for non-blocking async Avro reading with `AsyncAvroFileReader`, including from cloud storage via a custom `AsyncFileReader`. |
+| `object_store` |       ⬜ | **Deprecated**: the deprecated `AvroObjectReader` (implies `async`)         | Do not enable in new code; implement `AsyncFileReader` directly instead. Will be removed in a future release. |
 
 ### Schema fingerprints & custom logical type helpers
 
@@ -193,10 +193,10 @@ async fn main() -> anyhow::Result<()> {
   ```toml
   arrow-avro = { version = "58", default-features = false, features = ["deflate", "snappy", "zstd"] }
   ```
-* Async reading from object stores (S3, GCS, etc.):
+* Async reading (including from object stores such as S3, GCS, etc.):
 
   ```toml
-  arrow-avro = { version = "58", features = ["object_store"] }
+  arrow-avro = { version = "58", features = ["async"] }
   ```
 * Fingerprint helpers:
 

arrow-avro/examples/object_store.rs

@@ -0,0 +1,133 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use arrow_array::{ArrayRef, Int64Array, RecordBatch};
+use arrow_avro::errors::AvroError;
+use arrow_avro::reader::{AsyncAvroFileReader, AsyncFileReader, SpawnedReader};
+use arrow_avro::writer::AvroWriter;
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use futures::{FutureExt, TryStreamExt};
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{ObjectStore, ObjectStoreExt};
+use std::error::Error;
+use std::ops::Range;
+use std::sync::Arc;
+
+/// This example demonstrates reading Avro files from object storage via the
+/// [`object_store`] crate, without the deprecated `AvroObjectReader` type.
+///
+/// # Example Overview
+///
+/// 1. Writes an Avro Object Container File to an [`ObjectStore`]
+///
+/// 2. Reads it back with [`ObjectStoreReader`], a minimal [`AsyncFileReader`]
+///    implementation on top of an [`ObjectStore`] (equivalent of
+///    `AvroObjectReader`)
+///
+/// 3. Reads it again with the reader wrapped in a [`SpawnedReader`], which
+///    performs all I/O on a separate tokio runtime so that the runtime
+///    decoding Avro is not also driving the I/O (equivalent of
+///    `AvroObjectReader::with_runtime`)
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+    let path = Path::from("example.avro");
+
+    // 1. Write an Avro Object Container File to the store
+    let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
+    let batch = RecordBatch::try_from_iter([("col", col)])?;
+
+    let mut writer = AvroWriter::new(Vec::new(), batch.schema().as_ref().clone())?;
+    writer.write(&batch)?;
+    writer.finish()?;
+    store
+        .put(&path, Bytes::from(writer.into_inner()).into())
+        .await?;
+
+    // 2. Read it back with an `AsyncFileReader` implemented on `ObjectStore`.
+    // The builder requires the file size, which can be obtained via `head`
+    let file_size = store.head(&path).await?.size;
+
+    let reader = ObjectStoreReader::new(Arc::clone(&store), path.clone());
+    let stream = AsyncAvroFileReader::builder(reader, file_size, 1024)
+        .try_build()
+        .await?;
+    let read: Vec<RecordBatch> = stream.try_collect().await?;
+    assert_eq!(read, vec![batch.clone()]);
+    println!("read {} rows", read[0].num_rows());
+
+    // 3. Read again, performing the I/O on a dedicated runtime
+    let io_runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(1)
+        .enable_all()
+        .build()
+        .expect("failed to build I/O runtime");
+
+    let reader = ObjectStoreReader::new(Arc::clone(&store), path);
+    let reader = SpawnedReader::new(reader, io_runtime.handle().clone());
+    let stream = AsyncAvroFileReader::builder(reader, file_size, 1024)
+        .try_build()
+        .await?;
+    let read: Vec<RecordBatch> = stream.try_collect().await?;
+    assert_eq!(read, vec![batch]);
+    println!("read {} rows via dedicated I/O runtime", read[0].num_rows());
+
+    io_runtime.shutdown_background();
+    Ok(())
+}
+
+/// An [`AsyncFileReader`] for a location in an [`ObjectStore`]
+///
+/// This mirrors the example on the [`AsyncFileReader`] trait documentation.
+#[derive(Clone, Debug)]
+struct ObjectStoreReader {
+    store: Arc<dyn ObjectStore>,
+    path: Path,
+}
+
+impl ObjectStoreReader {
+    fn new(store: Arc<dyn ObjectStore>, path: Path) -> Self {
+        Self { store, path }
+    }
+}
+
+impl AsyncFileReader for ObjectStoreReader {
+    fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, Result<Bytes, AvroError>> {
+        async move {
+            self.store
+                .get_range(&self.path, range)
+                .await
+                .map_err(|e| AvroError::General(e.to_string()))
+        }
+        .boxed()
+    }
+
+    fn get_byte_ranges(
+        &mut self,
+        ranges: Vec<Range<u64>>,
+    ) -> BoxFuture<'_, Result<Vec<Bytes>, AvroError>> {
+        async move {
+            self.store
+                .get_ranges(&self.path, &ranges)
+                .await
+                .map_err(|e| AvroError::General(e.to_string()))
+        }
+        .boxed()
+    }
+}

arrow-avro/src/lib.rs

@@ -129,25 +129,22 @@
 //! feature is enabled.
 //!
 //! [`AsyncAvroFileReader`] implements `Stream<Item = Result<RecordBatch, ArrowError>>`,
-//! allowing efficient async streaming of record batches. When the `object_store` feature
-//! is enabled, [`AvroObjectReader`] provides integration with object storage services
-//! such as S3 via the [object_store] crate.
+//! allowing efficient async streaming of record batches. Any [`AsyncFileReader`]
+//! can be used as the source; there is a built-in implementation for types
+//! implementing `AsyncRead + AsyncSeek` (such as `tokio::fs::File`), and object
+//! storage services such as S3 can be integrated by implementing
+//! [`AsyncFileReader`] on top of a client such as the [object_store] crate
+//! (see the example on the trait documentation).
 //!
 //! ```ignore
-//! use std::sync::Arc;
-//! use arrow_avro::reader::{AsyncAvroFileReader, AvroObjectReader};
+//! use arrow_avro::reader::AsyncAvroFileReader;
 //! use futures::TryStreamExt;
-//! use object_store::ObjectStore;
-//! use object_store::local::LocalFileSystem;
-//! use object_store::path::Path;
 //!
 //! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
-//! let store: Arc<dyn ObjectStore> = Arc::new(LocalFileSystem::new());
-//! let path = Path::from("data/example.avro");
-//! let meta = store.head(&path).await?;
+//! let file = tokio::fs::File::open("data/example.avro").await?;
+//! let file_size = file.metadata().await?.len();
 //!
-//! let reader = AvroObjectReader::new(store, path);
-//! let stream = AsyncAvroFileReader::builder(reader, meta.size, 1024)
+//! let stream = AsyncAvroFileReader::builder(file, file_size, 1024)
 //!     .try_build()
 //!     .await?;
 //!
@@ -163,15 +160,15 @@
 //! ### Modules
 //!
 //! - [`reader`]: read Avro (OCF, SOE, Confluent) into Arrow `RecordBatch`es.
-//!   - With the `async` feature: [`AsyncAvroFileReader`] for async streaming reads.
-//!   - With the `object_store` feature: [`AvroObjectReader`] for reading from cloud storage.
+//!   - With the `async` feature: [`AsyncAvroFileReader`] for async streaming reads,
+//!     from any [`AsyncFileReader`] source including cloud object storage.
 //! - [`writer`]: write Arrow `RecordBatch`es as Avro (OCF, SOE, Confluent, Apicurio).
 //! - [`schema`]: Avro schema parsing / fingerprints / registries.
 //! - [`compression`]: codecs used for **OCF block compression** (i.e., Deflate, Snappy, Zstandard, BZip2, and XZ).
 //! - [`codec`]: internal Avro-Arrow type conversion and row decode/encode plans.
 //!
 //! [`AsyncAvroFileReader`]: reader::AsyncAvroFileReader
-//! [`AvroObjectReader`]: reader::AvroObjectReader
+//! [`AsyncFileReader`]: reader::AsyncFileReader
 //!
 //! ### Features
 //!
@@ -182,10 +179,13 @@
 //! - `bzip2` — enable BZip2 block compression.
 //! - `xz` — enable XZ/LZMA block compression.
 //!
-//! **Async & Object Store (opt‑in)**
+//! **Async (opt‑in)**
 //! - `async` — enable async APIs for reading Avro (`AsyncAvroFileReader`, `AsyncFileReader` trait).
-//! - `object_store` — enable integration with the [`object_store`] crate for reading Avro
-//!   from cloud storage (S3, GCS, Azure Blob, etc.) via `AvroObjectReader`. Implies `async`.
+//!   Cloud storage (S3, GCS, Azure Blob, etc.) can be integrated by implementing
+//!   `AsyncFileReader` on top of a client such as the [`object_store`] crate.
+//! - `object_store` (**deprecated**): enables the deprecated `AvroObjectReader`.
+//!   Implement `AsyncFileReader` directly instead (see above). Implies `async`.
+//!   This feature will be removed in a future release.
 //!
 //! **Schema fingerprints & helpers (opt‑in)**
 //! - `md5` — enable MD5 writer‑schema fingerprints.

arrow-avro/src/reader/async_reader/async_file_reader.rs

@@ -30,10 +30,56 @@ use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeek, AsyncSeekExt};
 /// 1. There is a default implementation for types that implement [`AsyncRead`]
 ///    and [`AsyncSeek`], for example [`tokio::fs::File`].
 ///
-/// 2. [`super::AvroObjectReader`], available when the `object_store` crate feature
-///    is enabled, implements this interface for [`ObjectStore`].
+/// 2. Implementations for remote storage, such as the `object_store` crate,
+///    can implement this interface directly, typically by pairing a store
+///    handle with an object path and delegating [`Self::get_bytes`] and
+///    [`Self::get_byte_ranges`] to ranged reads. [`super::SpawnedReader`] can
+///    wrap such a reader to perform its I/O on a dedicated tokio runtime.
 ///
-/// [`ObjectStore`]: object_store::ObjectStore
+/// # Example: implementing `AsyncFileReader` for the `object_store` crate
+///
+/// ```no_run
+/// # use std::ops::Range;
+/// # use std::sync::Arc;
+/// use arrow_avro::errors::AvroError;
+/// use arrow_avro::reader::AsyncFileReader;
+/// use bytes::Bytes;
+/// use futures::FutureExt;
+/// use futures::future::BoxFuture;
+/// use object_store::path::Path;
+/// use object_store::{ObjectStore, ObjectStoreExt};
+///
+/// #[derive(Clone, Debug)]
+/// struct ObjectStoreReader {
+///     store: Arc<dyn ObjectStore>,
+///     path: Path,
+/// }
+///
+/// impl AsyncFileReader for ObjectStoreReader {
+///     fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, Result<Bytes, AvroError>> {
+///         async move {
+///             self.store
+///                 .get_range(&self.path, range)
+///                 .await
+///                 .map_err(|e| AvroError::General(e.to_string()))
+///         }
+///         .boxed()
+///     }
+///
+///     fn get_byte_ranges(
+///         &mut self,
+///         ranges: Vec<Range<u64>>,
+///     ) -> BoxFuture<'_, Result<Vec<Bytes>, AvroError>> {
+///         async move {
+///             self.store
+///                 .get_ranges(&self.path, &ranges)
+///                 .await
+///                 .map_err(|e| AvroError::General(e.to_string()))
+///         }
+///         .boxed()
+///     }
+/// }
+/// ```
 ///
 /// [`tokio::fs::File`]: https://docs.rs/tokio/latest/tokio/fs/struct.File.html
 pub trait AsyncFileReader: Send {