parquet: Remove explicit object_store integration

parquet/Cargo.toml

@@ -89,7 +89,7 @@ zstd = { version = "0.13", default-features = false }
 serde_json = { version = "1.0", features = ["std"], default-features = false }
 arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "json"] }
 arrow-cast = { workspace = true }
-tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs"] }
+tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs", "sync"] }
 rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] }
 object_store = { workspace = true, features = ["azure", "fs"] }
 sysinfo = { version = "0.39.6", default-features = false, features = ["system"] }
@@ -116,6 +116,9 @@ experimental = ["variant_experimental"]
 # Enable async APIs
 async = ["futures", "tokio"]
 # Enable object_store integration
+# Deprecated: implement `AsyncFileReader` directly instead, see the example on
+# the `AsyncFileReader` trait documentation and `parquet/examples/object_store.rs`.
+# This feature will be removed in a future release.
 object_store = ["dep:object_store", "async"]
 # Group Zstd dependencies
 zstd = ["dep:zstd"]
@@ -157,6 +160,11 @@ name = "read_with_rowgroup"
 required-features = ["arrow", "async"]
 path = "./examples/read_with_rowgroup.rs"
 
+[[example]]
+name = "object_store"
+required-features = ["arrow", "async"]
+path = "./examples/object_store.rs"
+
 [[test]]
 name = "arrow_writer_layout"
 required-features = ["arrow"]
@@ -254,7 +262,7 @@ harness = false
 
 [[bench]]
 name = "arrow_reader_clickbench"
-required-features = ["arrow", "async", "object_store"]
+required-features = ["arrow", "async"]
 harness = false
 
 [[bench]]

parquet/benches/arrow_reader_clickbench.rs

@@ -35,16 +35,21 @@ use arrow::compute::{like, nlike, or};
 use arrow_array::types::{Int16Type, Int32Type, Int64Type};
 use arrow_array::{ArrayRef, ArrowPrimitiveType, BooleanArray, PrimitiveArray, StringViewArray};
 use arrow_schema::{ArrowError, DataType, Schema};
+use bytes::Bytes;
 use criterion::{Criterion, criterion_group, criterion_main};
-use futures::StreamExt;
+use futures::future::BoxFuture;
+use futures::{FutureExt, StreamExt, TryFutureExt};
 use object_store::local::LocalFileSystem;
+use object_store::{GetOptions, GetRange, ObjectStore, ObjectStoreExt};
 use parquet::arrow::arrow_reader::{
     ArrowPredicate, ArrowPredicateFn, ArrowReaderMetadata, ArrowReaderOptions,
     ParquetRecordBatchReaderBuilder, RowFilter,
 };
-use parquet::arrow::async_reader::ParquetObjectReader;
+use parquet::arrow::async_reader::{AsyncFileReader, MetadataSuffixFetch};
 use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask};
+use parquet::errors::ParquetError;
 use parquet::file::metadata::PageIndexPolicy;
+use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
 use parquet::schema::types::SchemaDescriptor;
 use std::fmt::{Display, Formatter};
 use std::path::{Path, PathBuf};
@@ -101,6 +106,75 @@ criterion_group!(
 );
 criterion_main!(benches);
 
+fn to_parquet_err(e: object_store::Error) -> ParquetError {
+    ParquetError::External(Box::new(e))
+}
+
+/// An [`AsyncFileReader`] reading via an [`ObjectStore`], mirroring the
+/// example on the [`AsyncFileReader`] trait documentation
+#[derive(Clone)]
+struct ObjectStoreReader {
+    store: Arc<dyn ObjectStore>,
+    path: object_store::path::Path,
+}
+
+impl AsyncFileReader for ObjectStoreReader {
+    fn get_bytes(
+        &mut self,
+        range: std::ops::Range<u64>,
+    ) -> BoxFuture<'_, Result<Bytes, ParquetError>> {
+        self.store
+            .get_range(&self.path, range)
+            .map_err(to_parquet_err)
+            .boxed()
+    }
+
+    fn get_byte_ranges(
+        &mut self,
+        ranges: Vec<std::ops::Range<u64>>,
+    ) -> BoxFuture<'_, Result<Vec<Bytes>, ParquetError>> {
+        async move {
+            self.store
+                .get_ranges(&self.path, &ranges)
+                .await
+                .map_err(to_parquet_err)
+        }
+        .boxed()
+    }
+
+    fn get_metadata<'a>(
+        &'a mut self,
+        options: Option<&'a ArrowReaderOptions>,
+    ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>, ParquetError>> {
+        async move {
+            let metadata = ParquetMetaDataReader::new()
+                .with_arrow_reader_options(options)
+                .load_via_suffix_and_finish(self)
+                .await?;
+            Ok(Arc::new(metadata))
+        }
+        .boxed()
+    }
+}
+
+impl MetadataSuffixFetch for &mut ObjectStoreReader {
+    fn fetch_suffix(&mut self, suffix: usize) -> BoxFuture<'_, Result<Bytes, ParquetError>> {
+        let options = GetOptions {
+            range: Some(GetRange::Suffix(suffix as u64)),
+            ..Default::default()
+        };
+        async move {
+            let resp = self
+                .store
+                .get_opts(&self.path, options)
+                .await
+                .map_err(to_parquet_err)?;
+            resp.bytes().await.map_err(to_parquet_err)
+        }
+        .boxed()
+    }
+}
+
 /// Predicate Function.
 ///
 /// Functions are invoked with the requested array and return a [`BooleanArray`]
@@ -736,15 +810,18 @@ impl ReadTest {
         self.check_row_count(row_count);
     }
 
-    /// Run the filter and projection using the async `ObjectStore` reader
+    /// Like [`Self::run_async`] but reading via an [`ObjectStore`] based reader
     async fn run_async_object_store(&self) {
         let hits_path = hits_1();
         let parent = hits_path.parent().unwrap();
         let file_name = hits_path.file_name().unwrap().to_str().unwrap();
         let store = Arc::new(LocalFileSystem::new_with_prefix(parent).unwrap());
         let location = object_store::path::Path::from(file_name);
 
-        let reader = ParquetObjectReader::new(store, location);
+        let reader = ObjectStoreReader {
+            store,
+            path: location,
+        };
 
         // setup the reader
         let mut stream = ParquetRecordBatchStreamBuilder::new_with_metadata(

parquet/examples/object_store.rs

@@ -0,0 +1,168 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use futures::{FutureExt, TryFutureExt, TryStreamExt};
+use object_store::buffered::BufWriter;
+use object_store::memory::InMemory;
+use object_store::path::Path;
+use object_store::{GetOptions, GetRange, ObjectStore, ObjectStoreExt};
+use parquet::arrow::arrow_reader::ArrowReaderOptions;
+use parquet::arrow::async_reader::{AsyncFileReader, MetadataSuffixFetch, SpawnedReader};
+use parquet::arrow::{AsyncArrowWriter, ParquetRecordBatchStreamBuilder};
+use parquet::errors::{ParquetError, Result};
+use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
+use std::ops::Range;
+use std::sync::Arc;
+
+/// This example demonstrates reading and writing Parquet files on object
+/// storage via the [`object_store`] crate, without the deprecated
+/// `ParquetObjectReader` and `ParquetObjectWriter` types.
+///
+/// # Example Overview
+///
+/// 1. Writes a Parquet file to an [`ObjectStore`] by passing an
+///    [`object_store::buffered::BufWriter`] directly to [`AsyncArrowWriter`],
+///    via the blanket [`AsyncFileWriter`] implementation for types
+///    implementing [`AsyncWrite`] (equivalent of `ParquetObjectWriter`).
+///
+/// 2. Reads it back with [`ObjectStoreReader`], a minimal [`AsyncFileReader`]
+///    implementation on top of an [`ObjectStore`] (equivalent of
+///    `ParquetObjectReader`).
+///
+/// 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 Parquet is not also driving the I/O (equivalent of
+///    `ParquetObjectReader::with_runtime`).
+///
+/// [`AsyncFileWriter`]: parquet::arrow::async_writer::AsyncFileWriter
+/// [`AsyncWrite`]: tokio::io::AsyncWrite
+#[tokio::main]
+async fn main() -> Result<()> {
+    let store: Arc<dyn ObjectStore> = Arc::new(InMemory::new());
+    let path = Path::from("example.parquet");
+
+    // 1. Write a Parquet file: a `BufWriter` implements `AsyncWrite` and can
+    // therefore be passed to `AsyncArrowWriter` directly.
+    let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
+    let batch = RecordBatch::try_from_iter([("col", col)]).unwrap();
+
+    let writer = BufWriter::new(Arc::clone(&store), path.clone());
+    let mut writer = AsyncArrowWriter::try_new(writer, batch.schema(), None)?;
+    writer.write(&batch).await?;
+    writer.close().await?;
+
+    // 2. Read it back with an `AsyncFileReader` implemented on `ObjectStore`.
+    let reader = ObjectStoreReader::new(Arc::clone(&store), path.clone());
+    let builder = ParquetRecordBatchStreamBuilder::new(reader).await?;
+    let read: Vec<RecordBatch> = builder.build()?.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 builder = ParquetRecordBatchStreamBuilder::new(reader).await?;
+    let read: Vec<RecordBatch> = builder.build()?.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(())
+}
+
+fn to_parquet_err(e: object_store::Error) -> ParquetError {
+    ParquetError::External(Box::new(e))
+}
+
+/// 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>> {
+        self.store
+            .get_range(&self.path, range)
+            .map_err(to_parquet_err)
+            .boxed()
+    }
+
+    fn get_byte_ranges(&mut self, ranges: Vec<Range<u64>>) -> BoxFuture<'_, Result<Vec<Bytes>>> {
+        async move {
+            self.store
+                .get_ranges(&self.path, &ranges)
+                .await
+                .map_err(to_parquet_err)
+        }
+        .boxed()
+    }
+
+    /// Loads the metadata, respecting the provided [`ArrowReaderOptions`] via
+    /// [`ParquetMetaDataReader::with_arrow_reader_options`]
+    fn get_metadata<'a>(
+        &'a mut self,
+        options: Option<&'a ArrowReaderOptions>,
+    ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>>> {
+        async move {
+            let metadata = ParquetMetaDataReader::new()
+                .with_arrow_reader_options(options)
+                .load_via_suffix_and_finish(self)
+                .await?;
+            Ok(Arc::new(metadata))
+        }
+        .boxed()
+    }
+}
+
+/// Supports fetching the Parquet footer without knowing the file size upfront,
+/// via suffix range requests
+impl MetadataSuffixFetch for &mut ObjectStoreReader {
+    fn fetch_suffix(&mut self, suffix: usize) -> BoxFuture<'_, Result<Bytes>> {
+        let options = GetOptions {
+            range: Some(GetRange::Suffix(suffix as u64)),
+            ..Default::default()
+        };
+        async move {
+            let resp = self
+                .store
+                .get_opts(&self.path, options)
+                .await
+                .map_err(to_parquet_err)?;
+            resp.bytes().await.map_err(to_parquet_err)
+        }
+        .boxed()
+    }
+}

parquet/src/arrow/arrow_reader/mod.rs

@@ -845,6 +845,44 @@ impl ArrowReaderOptions {
     }
 }
 
+impl ParquetMetaDataReader {
+    /// Applies the metadata related settings from [`ArrowReaderOptions`],
+    /// such as the [`ParquetMetaDataOptions`], decryption properties, and
+    /// [`PageIndexPolicy`] to this reader.
+    ///
+    /// The page index policies are only applied if at least one of them is not
+    /// [`PageIndexPolicy::Skip`], so policies previously configured on this
+    /// reader (e.g. from a preload setting) are preserved when the options do
+    /// not request the page index.
+    ///
+    /// This encodes the canonical way to construct a `ParquetMetaDataReader`
+    /// inside `AsyncFileReader::get_metadata` (available with the `async`
+    /// feature), so implementations outside this crate do not need to
+    /// duplicate it.
+    pub fn with_arrow_reader_options(mut self, options: Option<&ArrowReaderOptions>) -> Self {
+        let Some(options) = options else { return self };
+
+        self = self.with_metadata_options(Some(options.metadata_options().clone()));
+
+        #[cfg(feature = "encryption")]
+        {
+            self = self.with_decryption_properties(
+                options.file_decryption_properties.as_ref().map(Arc::clone),
+            );
+        }
+
+        if options.column_index_policy() != PageIndexPolicy::Skip
+            || options.offset_index_policy() != PageIndexPolicy::Skip
+        {
+            self = self
+                .with_column_index_policy(options.column_index_policy())
+                .with_offset_index_policy(options.offset_index_policy());
+        }
+
+        self
+    }
+}
+
 /// The metadata necessary to construct a [`ArrowReaderBuilder`]
 ///
 /// Note this structure is cheaply clone-able as it consists of several arcs.

parquet/src/arrow/arrow_reader/selection/mod.rs

@@ -363,7 +363,9 @@ impl RowSelection {
     ///
     /// Note: this method does not make any effort to combine consecutive ranges, nor coalesce
     /// ranges that are close together. This is instead delegated to the IO subsystem to optimise,
-    /// e.g. [`ObjectStore::get_ranges`](object_store::ObjectStore::get_ranges)
+    /// e.g. `ObjectStore::get_ranges` in the [`object_store`] crate
+    ///
+    /// [`object_store`]: https://crates.io/crates/object_store
     pub fn scan_ranges(&self, page_locations: &[PageLocation]) -> Vec<Range<u64>> {
         match &self.inner {
             RowSelectionInner::Selectors(selectors) => {