From 8dc1ba2487c37c330799051d53c90597b1febc4e Mon Sep 17 00:00:00 2001 From: Raphael Taylor-Davies Date: Tue, 28 Jul 2026 21:35:18 +0100 Subject: [PATCH] Add versioned object_store feature flags --- .github/workflows/arrow.yml | 8 + .github/workflows/parquet.yml | 10 + Cargo.lock | 242 +++++++++++++++++- Cargo.toml | 3 +- arrow-avro/Cargo.toml | 20 +- arrow-avro/README.md | 12 +- arrow-avro/src/lib.rs | 49 +++- .../reader/async_reader/async_file_reader.rs | 6 +- arrow-avro/src/reader/async_reader/mod.rs | 12 +- arrow-avro/src/reader/async_reader/store.rs | 6 +- arrow-avro/src/reader/mod.rs | 2 +- parquet/Cargo.toml | 22 +- parquet/benches/arrow_reader_clickbench.rs | 4 +- .../src/arrow/arrow_reader/selection/mod.rs | 2 +- parquet/src/arrow/async_reader/mod.rs | 10 +- parquet/src/arrow/async_reader/store.rs | 18 +- parquet/src/arrow/async_writer/mod.rs | 6 +- parquet/src/arrow/async_writer/store.rs | 16 +- parquet/src/errors.rs | 6 +- parquet/src/lib.rs | 33 ++- parquet/tests/encryption/encryption_async.rs | 14 +- 21 files changed, 418 insertions(+), 83 deletions(-) diff --git a/.github/workflows/arrow.yml b/.github/workflows/arrow.yml index 336e74ecf995..bffc33e3bd18 100644 --- a/.github/workflows/arrow.yml +++ b/.github/workflows/arrow.yml @@ -85,6 +85,10 @@ jobs: run: cargo test -p arrow-json --all-features - name: Test arrow-avro run: cargo test -p arrow-avro --all-features + # --all-features selects the newest supported object_store version, so test the + # older versions, which are gated behind their own feature flags, explicitly + - name: Test arrow-avro --features object_store + run: cargo test -p arrow-avro --features object_store - name: Test arrow-string run: cargo test -p arrow-string --all-features - name: Test arrow-ord @@ -172,6 +176,10 @@ jobs: run: cargo check -p arrow --no-default-features --all-targets --features ffi - name: Check compilation --no-default-features --all-targets --features chrono-tz run: cargo check -p arrow --no-default-features --all-targets --features chrono-tz + # --all-features selects the newest supported object_store version, so check the + # older versions, which are gated behind their own feature flags, explicitly + - name: Check compilation arrow-avro --all-targets --features object_store + run: cargo check -p arrow-avro --all-targets --features object_store # test the arrow crate builds against wasm32 in nightly rust diff --git a/.github/workflows/parquet.yml b/.github/workflows/parquet.yml index 158f1f08ae80..eefbc2f3611a 100644 --- a/.github/workflows/parquet.yml +++ b/.github/workflows/parquet.yml @@ -65,6 +65,10 @@ jobs: run: cargo test -p parquet - name: Test --all-features run: cargo test -p parquet --all-features + # --all-features selects the newest supported object_store version, so test the + # older versions, which are gated behind their own feature flags, explicitly + - name: Test --features object_store + run: cargo test -p parquet --features object_store - name: Run examples run: | # Test parquet examples @@ -124,6 +128,12 @@ jobs: run: cargo check -p parquet --no-default-features --features flate2 --features flate2-zlib-rs - name: Check compilation --no-default-features --features variant_experimental run: cargo check -p parquet --no-default-features --features variant_experimental + # --all-features selects the newest supported object_store version, so check the + # older versions, which are gated behind their own feature flags, explicitly + - name: Check compilation --no-default-features --features object_store + run: cargo check -p parquet --no-default-features --features object_store + - name: Check compilation --all-targets --no-default-features --features arrow --features object_store + run: cargo check -p parquet --all-targets --no-default-features --features arrow --features object_store # test the parquet crate builds against wasm32 in stable rust diff --git a/Cargo.lock b/Cargo.lock index ecf18221d42a..427c148f1e0e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -252,7 +252,8 @@ dependencies = [ "liblzma", "md5", "num-bigint 0.5.1", - "object_store", + "object_store 0.13.2", + "object_store 0.14.1", "once_cell", "rand 0.9.4", "serde", @@ -956,6 +957,16 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" +[[package]] +name = "combine" +version = "4.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +dependencies = [ + "bytes", + "memchr", +] + [[package]] name = "comfy-table" version = "7.2.2" @@ -1882,12 +1893,70 @@ dependencies = [ "either", ] +[[package]] +name = "itertools" +version = "0.15.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b4baf93f58d4425749ca49a51c50ebab072c5df6994d08fed93541c331481dc" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jni" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498" +dependencies = [ + "cfg-if", + "combine", + "jni-macros", + "jni-sys", + "log", + "simd_cesu8", + "thiserror 2.0.18", + "walkdir", + "windows-link", +] + +[[package]] +name = "jni-macros" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a00109accc170f0bdb141fed3e393c565b6f5e072365c3bd58f5b062591560a3" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "simd_cesu8", + "syn 2.0.119", +] + +[[package]] +name = "jni-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2" +dependencies = [ + "jni-sys-macros", +] + +[[package]] +name = "jni-sys-macros" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" +dependencies = [ + "quote", + "syn 2.0.119", +] + [[package]] name = "jobserver" version = "0.1.34" @@ -2128,6 +2197,18 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" +[[package]] +name = "nix" +version = "0.31.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d" +dependencies = [ + "bitflags", + "cfg-if", + "cfg_aliases", + "libc", +] + [[package]] name = "ntapi" version = "0.4.3" @@ -2258,9 +2339,9 @@ dependencies = [ "itertools 0.14.0", "parking_lot", "percent-encoding", - "quick-xml", + "quick-xml 0.39.4", "rand 0.10.1", - "reqwest", + "reqwest 0.12.28", "ring", "serde", "serde_json", @@ -2274,6 +2355,47 @@ dependencies = [ "web-time", ] +[[package]] +name = "object_store" +version = "0.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d354792e39fa5f0009e47623cf8b15b099bf9a652fa55c6f817fe28ac84fea50" +dependencies = [ + "async-trait", + "aws-lc-rs", + "base64 0.22.1", + "bytes", + "chrono", + "form_urlencoded", + "futures-channel", + "futures-core", + "futures-util", + "http", + "http-body-util", + "httparse", + "humantime", + "hyper", + "itertools 0.15.0", + "nix", + "parking_lot", + "percent-encoding", + "quick-xml 0.41.0", + "rand 0.10.1", + "reqwest 0.13.4", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "thiserror 2.0.18", + "tokio", + "tracing", + "url", + "walkdir", + "wasm-bindgen-futures", + "web-time", + "windows-sys 0.61.2", +] + [[package]] name = "once_cell" version = "1.21.4" @@ -2361,7 +2483,8 @@ dependencies = [ "num-bigint 0.5.1", "num-integer", "num-traits", - "object_store", + "object_store 0.13.2", + "object_store 0.14.1", "parquet-geospatial", "parquet-variant", "parquet-variant-compute", @@ -2714,6 +2837,16 @@ dependencies = [ "serde", ] +[[package]] +name = "quick-xml" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" +dependencies = [ + "memchr", + "serde", +] + [[package]] name = "quinn" version = "0.11.11" @@ -2740,6 +2873,7 @@ version = "0.11.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4fcb935c5bec503c2f0e306bdd3e58bb9029dcb14fa8d9ac76e3a5256ac0763e" dependencies = [ + "aws-lc-rs", "bytes", "getrandom 0.3.4", "lru-slab", @@ -2928,7 +3062,46 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", + "wasm-streams 0.4.2", + "web-sys", +] + +[[package]] +name = "reqwest" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "219c5811de6525e5416c7d5d53bb656d3afdbc6c5af816e0802bcfa42dbdc1c3" +dependencies = [ + "base64 0.22.1", + "bytes", + "futures-core", + "futures-util", + "h2", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "rustls-platform-verifier", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tokio-util", + "tower", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "wasm-streams 0.5.0", "web-sys", ] @@ -3012,6 +3185,33 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-platform-verifier" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0" +dependencies = [ + "core-foundation", + "core-foundation-sys", + "jni", + "log", + "once_cell", + "rustls", + "rustls-native-certs", + "rustls-platform-verifier-android", + "rustls-webpki", + "security-framework", + "security-framework-sys", + "webpki-root-certs", + "windows-sys 0.61.2", +] + +[[package]] +name = "rustls-platform-verifier-android" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f" + [[package]] name = "rustls-webpki" version = "0.103.13" @@ -3192,6 +3392,16 @@ version = "0.3.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214" +[[package]] +name = "simd_cesu8" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11031e251abf8611c80f460e19dbdeb54a66db918e49c65a7065b46ac7aec520" +dependencies = [ + "rustc_version", + "simdutf8", +] + [[package]] name = "simdutf8" version = "0.1.5" @@ -3979,6 +4189,19 @@ dependencies = [ "web-sys", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasmparser" version = "0.244.0" @@ -4011,6 +4234,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-root-certs" +version = "1.0.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b96554aa2acc8ccdb7e1c9a58a7a68dd5d13bccc69cd124cb09406db612a1c9b" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "webpki-roots" version = "1.0.7" diff --git a/Cargo.toml b/Cargo.toml index 4e2048c26f0a..b6b7ecb67d9b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -112,7 +112,8 @@ criterion = { version = "0.8.0", default-features = false } insta = { version = "1.46.3", default-features = false } -object_store = { version = "0.13.2", default-features = false } +# object_store is not a workspace dependency, as crates integrating with it depend on +# multiple versions simultaneously, see e.g. parquet's `object_store_0_14` feature simdutf8 = { version = "0.1.5", default-features = false } diff --git a/arrow-avro/Cargo.toml b/arrow-avro/Cargo.toml index 0cc5af701055..ac150db63801 100644 --- a/arrow-avro/Cargo.toml +++ b/arrow-avro/Cargo.toml @@ -47,8 +47,12 @@ avro_custom_types = ["dep:arrow-select"] # Enable async APIs async = ["futures", "tokio"] -# Enable object_store integration -object_store = ["dep:object_store", "async"] +# Enable integration with object_store 0.13 +# +# Deprecated: prefer `object_store_0_14`, this will be removed in a future major release +object_store = ["dep:object_store_0_13", "async"] +# Enable integration with object_store 0.14 +object_store_0_14 = ["dep:object_store_0_14", "async"] [dependencies] arrow-schema = { workspace = true } @@ -56,7 +60,11 @@ arrow-buffer = { workspace = true } arrow-array = { workspace = true } arrow-select = { workspace = true, optional = true } -object_store = { workspace = true, optional = true } +# Each supported object_store version is a separately renamed dependency, gated behind a +# versioned feature flag. This allows adding support for new object_store releases, which +# are themselves breaking changes, without a breaking change to this crate +object_store_0_13 = { package = "object_store", version = "0.13.2", default-features = false, optional = true } +object_store_0_14 = { package = "object_store", version = "0.14.1", default-features = false, optional = true } bytes = { version = "1.11.0", default-features = false, features = ["std"] } serde_json = { version = "1.0", default-features = false, features = ["std"] } @@ -93,7 +101,11 @@ futures = "0.3.31" async-stream = "0.3.6" apache-avro = "0.21.0" num-bigint = "0.5" -object_store = { workspace = true, features = ["fs"] } +# Enable the store implementations used by tests for every supported object_store version. +# Only the version selected by the enabled feature flag is used, tests access it via the +# `arrow_avro::object_store` re-export rather than by name +object_store_0_13 = { package = "object_store", version = "0.13.2", features = ["fs"] } +object_store_0_14 = { package = "object_store", version = "0.14.1", features = ["fs"] } 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"] } diff --git a/arrow-avro/README.md b/arrow-avro/README.md index dbc1e1760ea3..31ef5c3ecad5 100644 --- a/arrow-avro/README.md +++ b/arrow-avro/README.md @@ -105,15 +105,15 @@ 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 from object stores (`object_store_0_14` feature) ```rust,ignore use std::sync::Arc; use arrow_avro::reader::{AsyncAvroFileReader, AvroObjectReader}; use futures::TryStreamExt; -use object_store::ObjectStore; -use object_store::local::LocalFileSystem; -use object_store::path::Path; +use arrow_avro::object_store::ObjectStore; +use arrow_avro::object_store::local::LocalFileSystem; +use arrow_avro::object_store::path::Path; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -163,7 +163,7 @@ async fn main() -> anyhow::Result<()> { | 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.). | +| `object_store_0_14` | ⬜ | Integration with `object_store` 0.14 (implies `async`) | Enable for reading Avro from cloud storage (S3, GCS, Azure Blob, etc.). New `object_store` releases are breaking changes, so each supported version has its own feature. The unversioned `object_store` feature selects 0.13 and is deprecated. | ### Schema fingerprints & custom logical type helpers @@ -196,7 +196,7 @@ async fn main() -> anyhow::Result<()> { * Async reading from object stores (S3, GCS, etc.): ```toml - arrow-avro = { version = "58", features = ["object_store"] } + arrow-avro = { version = "58", features = ["object_store_0_14"] } ``` * Fingerprint helpers: diff --git a/arrow-avro/src/lib.rs b/arrow-avro/src/lib.rs index da451ea1460e..7288047e4fea 100644 --- a/arrow-avro/src/lib.rs +++ b/arrow-avro/src/lib.rs @@ -129,17 +129,17 @@ //! feature is enabled. //! //! [`AsyncAvroFileReader`] implements `Stream>`, -//! 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. When one of the versioned +//! `object_store` features is enabled, e.g. `object_store_0_14`, [`AvroObjectReader`] +//! provides integration with object storage services such as S3 via the [object_store] crate. //! //! ```ignore //! use std::sync::Arc; //! use arrow_avro::reader::{AsyncAvroFileReader, AvroObjectReader}; //! use futures::TryStreamExt; -//! use object_store::ObjectStore; -//! use object_store::local::LocalFileSystem; -//! use object_store::path::Path; +//! use arrow_avro::object_store::ObjectStore; +//! use arrow_avro::object_store::local::LocalFileSystem; +//! use arrow_avro::object_store::path::Path; //! //! # async fn example() -> Result<(), Box> { //! let store: Arc = Arc::new(LocalFileSystem::new()); @@ -164,7 +164,8 @@ //! //! - [`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 a versioned `object_store` feature, e.g. `object_store_0_14`: +//! [`AvroObjectReader`] for reading from cloud 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). @@ -184,8 +185,12 @@ //! //! **Async & Object Store (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`. +//! - `object_store_0_14` — enable integration with the [`object_store`] crate for reading +//! Avro from cloud storage (S3, GCS, Azure Blob, etc.) via `AvroObjectReader`. Implies +//! `async`. New `object_store` releases are breaking changes, so each supported version +//! has its own feature, and the selected version is re-exported as +//! `arrow_avro::object_store`. The unversioned `object_store` feature selects +//! `object_store` 0.13 and is deprecated. //! //! **Schema fingerprints & helpers (opt‑in)** //! - `md5` — enable MD5 writer‑schema fingerprints. @@ -215,6 +220,32 @@ #![cfg_attr(docsrs, feature(doc_cfg))] #![warn(missing_docs)] +/// Re-export of the [`object_store`] version this crate is integrated with +/// +/// New [`object_store`] releases are breaking changes, and are therefore gated behind +/// versioned feature flags, e.g. `object_store_0_14`. This allows support for new versions +/// to be added, and old versions removed, without a breaking change to this crate. +/// +/// The [`object_store`] types appear in this crate's public API, so downstream crates must +/// use the same version. Using this re-export guarantees they do: +/// +/// ``` +/// use arrow_avro::object_store::ObjectStore; +/// ``` +/// +/// If more than one version is enabled the newest is used, as only one version of these +/// types can be exposed at a time. +/// +/// [`object_store`]: https://docs.rs/object_store/latest/object_store/ +#[cfg(feature = "object_store_0_14")] +pub use object_store_0_14 as object_store; + +/// Re-export of the [`object_store`] version this crate is integrated with +/// +/// [`object_store`]: https://docs.rs/object_store/latest/object_store/ +#[cfg(all(feature = "object_store", not(feature = "object_store_0_14")))] +pub use object_store_0_13 as object_store; + /// Core functionality for reading Avro data into Arrow arrays /// /// Implements the primary reader interface and record decoding logic. diff --git a/arrow-avro/src/reader/async_reader/async_file_reader.rs b/arrow-avro/src/reader/async_reader/async_file_reader.rs index 1257a2f3dd4d..e5a65ea2788d 100644 --- a/arrow-avro/src/reader/async_reader/async_file_reader.rs +++ b/arrow-avro/src/reader/async_reader/async_file_reader.rs @@ -30,10 +30,10 @@ 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. [`super::AvroObjectReader`], available when a versioned `object_store` crate feature +/// is enabled, e.g. `object_store_0_14`, implements this interface for [`ObjectStore`]. /// -/// [`ObjectStore`]: object_store::ObjectStore +/// [`ObjectStore`]: crate::object_store::ObjectStore /// /// [`tokio::fs::File`]: https://docs.rs/tokio/latest/tokio/fs/struct.File.html pub trait AsyncFileReader: Send { diff --git a/arrow-avro/src/reader/async_reader/mod.rs b/arrow-avro/src/reader/async_reader/mod.rs index 43ff931a050d..7d0e9c7c8b0a 100644 --- a/arrow-avro/src/reader/async_reader/mod.rs +++ b/arrow-avro/src/reader/async_reader/mod.rs @@ -39,11 +39,11 @@ mod builder; pub use async_file_reader::AsyncFileReader; pub use builder::{ReaderBuilder, read_header_info}; -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] mod store; use crate::errors::AvroError; -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] pub use store::AvroObjectReader; enum FetchNextBehaviour { @@ -543,10 +543,13 @@ impl Stream for AsyncAvroFileReader { } } -#[cfg(all(test, feature = "object_store"))] +#[cfg(all(test, any(feature = "object_store", feature = "object_store_0_14")))] mod tests { use super::*; use crate::codec::Tz; + use crate::object_store::local::LocalFileSystem; + use crate::object_store::path::Path; + use crate::object_store::{ObjectStore, ObjectStoreExt}; use crate::schema::{ AVRO_NAME_METADATA_KEY, AVRO_NAMESPACE_METADATA_KEY, AvroSchema, SCHEMA_METADATA_KEY, }; @@ -555,9 +558,6 @@ mod tests { use arrow_array::*; use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit}; use futures::{StreamExt, TryStreamExt}; - use object_store::local::LocalFileSystem; - use object_store::path::Path; - use object_store::{ObjectStore, ObjectStoreExt}; use std::collections::HashMap; use std::sync::Arc; diff --git a/arrow-avro/src/reader/async_reader/store.rs b/arrow-avro/src/reader/async_reader/store.rs index 44a4abf1a282..72b02f33de5c 100644 --- a/arrow-avro/src/reader/async_reader/store.rs +++ b/arrow-avro/src/reader/async_reader/store.rs @@ -16,13 +16,13 @@ // under the License. use crate::errors::AvroError; +use crate::object_store::ObjectStore; +use crate::object_store::ObjectStoreExt; +use crate::object_store::path::Path; use crate::reader::async_reader::AsyncFileReader; use bytes::Bytes; use futures::future::BoxFuture; use futures::{FutureExt, TryFutureExt}; -use object_store::ObjectStore; -use object_store::ObjectStoreExt; -use object_store::path::Path; use std::error::Error; use std::ops::Range; use std::sync::Arc; diff --git a/arrow-avro/src/reader/mod.rs b/arrow-avro/src/reader/mod.rs index 6fb75422a4b2..b427ed5a30a6 100644 --- a/arrow-avro/src/reader/mod.rs +++ b/arrow-avro/src/reader/mod.rs @@ -504,7 +504,7 @@ pub mod async_reader; pub use header::{HeaderInfo, read_header_info}; -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] pub use async_reader::AvroObjectReader; #[cfg(feature = "async")] pub use async_reader::{AsyncAvroFileReader, AsyncFileReader}; diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml index 0ae0182263ed..bd17953871aa 100644 --- a/parquet/Cargo.toml +++ b/parquet/Cargo.toml @@ -49,7 +49,11 @@ parquet-variant = { workspace = true, optional = true } parquet-variant-json = { workspace = true, optional = true } parquet-variant-compute = { workspace = true, optional = true } -object_store = { workspace = true, optional = true, features = ["tokio"] } +# Each supported object_store version is a separately renamed dependency, gated behind a +# versioned feature flag. This allows adding support for new object_store releases, which +# are themselves breaking changes, without a breaking change to this crate +object_store_0_13 = { package = "object_store", version = "0.13.2", default-features = false, features = ["tokio"], optional = true } +object_store_0_14 = { package = "object_store", version = "0.14.1", default-features = false, features = ["tokio"], optional = true } bytes = { version = "1.1", default-features = false, features = ["std"] } snap = { version = "1.0", default-features = false, optional = true } @@ -91,7 +95,11 @@ arrow = { workspace = true, features = ["ipc", "test_utils", "prettyprint", "jso arrow-cast = { workspace = true } tokio = { version = "1.0", default-features = false, features = ["macros", "rt-multi-thread", "io-util", "fs"] } rand = { version = "0.9", default-features = false, features = ["std", "std_rng", "thread_rng"] } -object_store = { workspace = true, features = ["azure", "fs"] } +# Enable the store implementations used by tests, doctests and benchmarks for every +# supported object_store version. Only the version selected by the enabled feature flag is +# used, tests access it via the `parquet::object_store` re-export rather than by name +object_store_0_13 = { package = "object_store", version = "0.13.2", features = ["azure", "fs"] } +object_store_0_14 = { package = "object_store", version = "0.14.1", features = ["azure", "fs"] } sysinfo = { version = "0.39.6", default-features = false, features = ["system"] } [package.metadata.docs.rs] @@ -115,8 +123,12 @@ test_common = ["arrow/test_utils"] experimental = ["variant_experimental"] # Enable async APIs async = ["futures", "tokio"] -# Enable object_store integration -object_store = ["dep:object_store", "async"] +# Enable integration with object_store 0.13 +# +# Deprecated: prefer `object_store_0_14`, this will be removed in a future major release +object_store = ["dep:object_store_0_13", "async"] +# Enable integration with object_store 0.14 +object_store_0_14 = ["dep:object_store_0_14", "async"] # Group Zstd dependencies zstd = ["dep:zstd"] # Verify 32-bit CRC checksum when decoding parquet pages @@ -254,7 +266,7 @@ harness = false [[bench]] name = "arrow_reader_clickbench" -required-features = ["arrow", "async", "object_store"] +required-features = ["arrow", "async", "object_store_0_14"] harness = false [[bench]] diff --git a/parquet/benches/arrow_reader_clickbench.rs b/parquet/benches/arrow_reader_clickbench.rs index 039829f1b975..79573963df25 100644 --- a/parquet/benches/arrow_reader_clickbench.rs +++ b/parquet/benches/arrow_reader_clickbench.rs @@ -37,7 +37,6 @@ use arrow_array::{ArrayRef, ArrowPrimitiveType, BooleanArray, PrimitiveArray, St use arrow_schema::{ArrowError, DataType, Schema}; use criterion::{Criterion, criterion_group, criterion_main}; use futures::StreamExt; -use object_store::local::LocalFileSystem; use parquet::arrow::arrow_reader::{ ArrowPredicate, ArrowPredicateFn, ArrowReaderMetadata, ArrowReaderOptions, ParquetRecordBatchReaderBuilder, RowFilter, @@ -45,6 +44,7 @@ use parquet::arrow::arrow_reader::{ use parquet::arrow::async_reader::ParquetObjectReader; use parquet::arrow::{ParquetRecordBatchStreamBuilder, ProjectionMask}; use parquet::file::metadata::PageIndexPolicy; +use parquet::object_store::local::LocalFileSystem; use parquet::schema::types::SchemaDescriptor; use std::fmt::{Display, Formatter}; use std::path::{Path, PathBuf}; @@ -742,7 +742,7 @@ impl ReadTest { 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 location = parquet::object_store::path::Path::from(file_name); let reader = ParquetObjectReader::new(store, location); diff --git a/parquet/src/arrow/arrow_reader/selection/mod.rs b/parquet/src/arrow/arrow_reader/selection/mod.rs index 41a59c048e2b..450201eb519e 100644 --- a/parquet/src/arrow/arrow_reader/selection/mod.rs +++ b/parquet/src/arrow/arrow_reader/selection/mod.rs @@ -363,7 +363,7 @@ 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`](crate::object_store::ObjectStore::get_ranges) pub fn scan_ranges(&self, page_locations: &[PageLocation]) -> Vec> { match &self.inner { RowSelectionInner::Selectors(selectors) => { diff --git a/parquet/src/arrow/async_reader/mod.rs b/parquet/src/arrow/async_reader/mod.rs index 5a0083b7164d..aef0f2357552 100644 --- a/parquet/src/arrow/async_reader/mod.rs +++ b/parquet/src/arrow/async_reader/mod.rs @@ -50,12 +50,12 @@ use crate::file::metadata::{ParquetMetaData, ParquetMetaDataReader}; mod metadata; pub use metadata::*; -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] mod store; use crate::DecodeResult; use crate::arrow::push_decoder::{ParquetPushDecoder, ParquetPushDecoderBuilder, PushDecoderInput}; -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] pub use store::*; /// The asynchronous interface used by [`ParquetRecordBatchStream`] to read parquet files @@ -65,10 +65,10 @@ pub use store::*; /// 1. There is a default implementation for types that implement [`AsyncRead`] /// and [`AsyncSeek`], for example [`tokio::fs::File`]. /// -/// 2. [`ParquetObjectReader`], available when the `object_store` crate feature -/// is enabled, implements this interface for [`ObjectStore`]. +/// 2. [`ParquetObjectReader`], available when a versioned `object_store` crate feature +/// is enabled, e.g. `object_store_0_14`, implements this interface for [`ObjectStore`]. /// -/// [`ObjectStore`]: object_store::ObjectStore +/// [`ObjectStore`]: crate::object_store::ObjectStore /// /// [`tokio::fs::File`]: https://docs.rs/tokio/latest/tokio/fs/struct.File.html pub trait AsyncFileReader: Send { diff --git a/parquet/src/arrow/async_reader/store.rs b/parquet/src/arrow/async_reader/store.rs index d47ca744d8f6..ecc9b91d2560 100644 --- a/parquet/src/arrow/async_reader/store.rs +++ b/parquet/src/arrow/async_reader/store.rs @@ -21,20 +21,20 @@ use crate::arrow::arrow_reader::ArrowReaderOptions; use crate::arrow::async_reader::{AsyncFileReader, MetadataSuffixFetch}; use crate::errors::{ParquetError, Result}; use crate::file::metadata::{PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader}; +use crate::object_store::ObjectStoreExt; +use crate::object_store::{GetOptions, GetRange}; +use crate::object_store::{ObjectStore, path::Path}; use bytes::Bytes; use futures::{FutureExt, TryFutureExt, future::BoxFuture}; -use object_store::ObjectStoreExt; -use object_store::{GetOptions, GetRange}; -use object_store::{ObjectStore, path::Path}; use tokio::runtime::Handle; /// Reads Parquet files in object storage using [`ObjectStore`]. /// /// ```no_run /// # use std::io::stdout; /// # use std::sync::Arc; -/// # use object_store::azure::MicrosoftAzureBuilder; -/// # use object_store::{ObjectStore, ObjectStoreExt}; -/// # use object_store::path::Path; +/// # use parquet::object_store::azure::MicrosoftAzureBuilder; +/// # use parquet::object_store::{ObjectStore, ObjectStoreExt}; +/// # use parquet::object_store::path::Path; /// # use parquet::arrow::async_reader::ParquetObjectReader; /// # use parquet::arrow::ParquetRecordBatchStreamBuilder; /// # use parquet::schema::printer::print_parquet_metadata; @@ -260,11 +260,11 @@ mod tests { use crate::arrow::ParquetRecordBatchStreamBuilder; use crate::arrow::async_reader::{AsyncFileReader, ParquetObjectReader}; use crate::errors::ParquetError; + use crate::object_store::local::LocalFileSystem; + use crate::object_store::path::Path; + use crate::object_store::{ObjectMeta, ObjectStore, ObjectStoreExt}; use arrow::util::test_util::parquet_test_data; use futures::FutureExt; - use object_store::local::LocalFileSystem; - use object_store::path::Path; - use object_store::{ObjectMeta, ObjectStore, ObjectStoreExt}; async fn get_meta_store() -> (ObjectMeta, Arc) { let res = parquet_test_data(); diff --git a/parquet/src/arrow/async_writer/mod.rs b/parquet/src/arrow/async_writer/mod.rs index a050ef77c46c..e717d0975cf5 100644 --- a/parquet/src/arrow/async_writer/mod.rs +++ b/parquet/src/arrow/async_writer/mod.rs @@ -54,10 +54,12 @@ //! ``` //! //! [`object_store`] provides it's native implementation of [`AsyncFileWriter`] by [`ParquetObjectWriter`]. +//! +//! [`object_store`]: crate::object_store -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] mod store; -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] pub use store::*; use crate::{ diff --git a/parquet/src/arrow/async_writer/store.rs b/parquet/src/arrow/async_writer/store.rs index 698248e619db..3327e911cb54 100644 --- a/parquet/src/arrow/async_writer/store.rs +++ b/parquet/src/arrow/async_writer/store.rs @@ -21,18 +21,18 @@ use std::sync::Arc; use crate::arrow::async_writer::AsyncFileWriter; use crate::errors::{ParquetError, Result}; -use object_store::ObjectStore; -use object_store::buffered::BufWriter; -use object_store::path::Path; +use crate::object_store::ObjectStore; +use crate::object_store::buffered::BufWriter; +use crate::object_store::path::Path; use tokio::io::AsyncWriteExt; /// [`ParquetObjectWriter`] for writing to parquet to [`ObjectStore`] /// /// ``` /// # use arrow_array::{ArrayRef, Int64Array, RecordBatch}; -/// # use object_store::memory::InMemory; -/// # use object_store::path::Path; -/// # use object_store::{ObjectStore, ObjectStoreExt}; +/// # use parquet::object_store::memory::InMemory; +/// # use parquet::object_store::path::Path; +/// # use parquet::object_store::{ObjectStore, ObjectStoreExt}; /// # use std::sync::Arc; /// /// # use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; @@ -118,14 +118,14 @@ impl From for ParquetObjectWriter { } #[cfg(test)] mod tests { + use crate::object_store::memory::InMemory; use arrow_array::{ArrayRef, Int64Array, RecordBatch}; - use object_store::memory::InMemory; use std::sync::Arc; use super::*; use crate::arrow::AsyncArrowWriter; use crate::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; - use object_store::ObjectStoreExt; + use crate::object_store::ObjectStoreExt; #[tokio::test] async fn test_async_writer() { diff --git a/parquet/src/errors.rs b/parquet/src/errors.rs index 73b4d76de578..0cf79397ba3f 100644 --- a/parquet/src/errors.rs +++ b/parquet/src/errors.rs @@ -133,9 +133,9 @@ impl From for ParquetError { } } -#[cfg(feature = "object_store")] -impl From for ParquetError { - fn from(e: object_store::Error) -> ParquetError { +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] +impl From for ParquetError { + fn from(e: crate::object_store::Error) -> ParquetError { ParquetError::External(Box::new(e)) } } diff --git a/parquet/src/lib.rs b/parquet/src/lib.rs index 55eebfa3708d..4c2ce748de27 100644 --- a/parquet/src/lib.rs +++ b/parquet/src/lib.rs @@ -101,9 +101,10 @@ //! read and write [`RecordBatch`]es asynchronously. //! //! Most users will use [`AsyncArrowWriter`] for writing and [`ParquetRecordBatchStreamBuilder`] -//! for reading. When the `object_store` feature is enabled, [`ParquetObjectReader`] -//! provides efficient integration with object storage services such as S3 via the [object_store] -//! crate, automatically optimizing IO based on any predicates or projections provided. +//! for reading. When one of the versioned `object_store` features is enabled, e.g. +//! `object_store_0_14`, [`ParquetObjectReader`] provides efficient integration with object +//! storage services such as S3 via the [object_store] crate, automatically optimizing IO +//! based on any predicates or projections provided. //! //! [`async_reader`]: arrow::async_reader //! [`async_writer`]: arrow::async_writer @@ -190,6 +191,32 @@ experimental!(#[macro_use] mod util); pub use util::utf8; +/// Re-export of the [`object_store`] version this crate is integrated with +/// +/// New [`object_store`] releases are breaking changes, and are therefore gated behind +/// versioned feature flags, e.g. `object_store_0_14`. This allows support for new versions +/// to be added, and old versions removed, without a breaking change to this crate. +/// +/// The [`object_store`] types appear in this crate's public API, so downstream crates must +/// use the same version. Using this re-export guarantees they do: +/// +/// ``` +/// use parquet::object_store::ObjectStore; +/// ``` +/// +/// If more than one version is enabled the newest is used, as only one version of these +/// types can be exposed at a time. +/// +/// [`object_store`]: https://docs.rs/object_store/latest/object_store/ +#[cfg(feature = "object_store_0_14")] +pub use object_store_0_14 as object_store; + +/// Re-export of the [`object_store`] version this crate is integrated with +/// +/// [`object_store`]: https://docs.rs/object_store/latest/object_store/ +#[cfg(all(feature = "object_store", not(feature = "object_store_0_14")))] +pub use object_store_0_13 as object_store; + #[cfg(feature = "arrow")] pub mod arrow; pub mod column; diff --git a/parquet/tests/encryption/encryption_async.rs b/parquet/tests/encryption/encryption_async.rs index f86ab59bf755..68c42c3173e5 100644 --- a/parquet/tests/encryption/encryption_async.rs +++ b/parquet/tests/encryption/encryption_async.rs @@ -420,14 +420,14 @@ async fn test_write_non_uniform_encryption() { .await; } -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] async fn get_encrypted_meta_store() -> ( - object_store::ObjectMeta, - std::sync::Arc, + parquet::object_store::ObjectMeta, + std::sync::Arc, ) { - use object_store::local::LocalFileSystem; - use object_store::path::Path; - use object_store::{ObjectStore, ObjectStoreExt}; + use parquet::object_store::local::LocalFileSystem; + use parquet::object_store::path::Path; + use parquet::object_store::{ObjectStore, ObjectStoreExt}; use std::sync::Arc; let test_data = arrow::util::test_util::parquet_test_data(); @@ -442,7 +442,7 @@ async fn get_encrypted_meta_store() -> ( } #[tokio::test] -#[cfg(feature = "object_store")] +#[cfg(any(feature = "object_store", feature = "object_store_0_14"))] async fn test_read_encrypted_file_from_object_store() { use parquet::arrow::async_reader::{AsyncFileReader, ParquetObjectReader}; let (meta, store) = get_encrypted_meta_store().await;