This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 4cd8be954f docs(parquet): add OpenDAL read and write example (#11014)
4cd8be954f is described below
commit 4cd8be954f6bc6b6dd265140207365b59a9900ec
Author: Xuanwo <[email protected]>
AuthorDate: Sat Sep 12 04:47:41 2026 +0800
docs(parquet): add OpenDAL read and write example (#11014)
# Which issue does this PR close?
Related to #10308.
# Rationale for this change
Demonstrate how to use OpenDAL with Parquet's generic async I/O
interfaces, alongside the existing object_store example.
# What changes are included in this PR?
Add an in-memory read/write example with AsyncFileReader and
AsyncFileWriter adapters and a SpawnedReader demonstration. Point
application developers to parquet_opendal for ready-made adapters.
OpenDAL 0.59 is a development dependency with only the memory service
enabled.
# Are these changes tested?
The example asserts that both ordinary and dedicated-runtime reads
reproduce the written RecordBatch. Formatting and whitespace checks
pass.
# Are there any user-facing changes?
A new example, runnable with `cargo run -p parquet --example opendal
--features async`. No public API changes. OpenDAL 0.59 requires Rust
1.91, above this workspace's declared Rust 1.88 MSRV; this affects
development targets that resolve/build the new dependency.
---
Cargo.lock | 174 +++++++++++++++++++++++++++++++++++++++++---
parquet/Cargo.toml | 6 ++
parquet/examples/opendal.rs | 155 +++++++++++++++++++++++++++++++++++++++
3 files changed, 326 insertions(+), 9 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index 4573d81078..358062a672 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -512,7 +512,7 @@ dependencies = [
name = "arrow-schema"
version = "60.0.0"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"criterion",
"insta",
"postcard",
@@ -598,6 +598,16 @@ dependencies = [
"syn 3.0.5",
]
+[[package]]
+name = "asyncband"
+version = "0.6.7"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "94a214ba60d6231afd0e805e3c27c45a1626d9debaa5a5061c45a1ea1b2f1ed0"
+dependencies = [
+ "hashbrown 0.17.1",
+ "slab",
+]
+
[[package]]
name = "atoi"
version = "3.1.0"
@@ -711,6 +721,12 @@ dependencies = [
"serde",
]
+[[package]]
+name = "bitflags"
+version = "1.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
+
[[package]]
name = "bitflags"
version = "2.13.1"
@@ -1176,6 +1192,37 @@ dependencies = [
"syn 2.0.119",
]
+[[package]]
+name = "defmt"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "e2953bfe4f93bbd20cc71198842756f77d161884c99ebbabc41d80231ded88d1"
+dependencies = [
+ "bitflags 1.3.2",
+ "defmt-macros",
+]
+
+[[package]]
+name = "defmt-macros"
+version = "1.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "bad9c72e7ca2137e0dc3813245a0d282fd6daad32fd800af018306a9169b5fe8"
+dependencies = [
+ "defmt-parser",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "defmt-parser"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "10d60334b3b2e7c9d91ef8150abfb6fa4c1c39ebbcf4a81c2e346aad939fee3e"
+dependencies = [
+ "thiserror 2.0.19",
+]
+
[[package]]
name = "derive_arbitrary"
version = "1.4.2"
@@ -1285,7 +1332,7 @@ version = "25.12.19"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "35f6839d7b3b98adde531effaf34f0c2badc6f4735d26fe74709d8e513a96ef3"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"rustc_version",
]
@@ -1880,6 +1927,61 @@ version = "1.0.18"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
+[[package]]
+name = "jiff"
+version = "0.2.35"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "668b7183bd07af9a4885f5c35b0cc5c83c4607a913c16b7e17291832910d2dcc"
+dependencies = [
+ "defmt",
+ "jiff-core",
+ "jiff-static",
+ "jiff-tzdb-platform",
+ "js-sys",
+ "log",
+ "portable-atomic",
+ "portable-atomic-util",
+ "serde_core",
+ "wasm-bindgen",
+ "windows-link",
+]
+
+[[package]]
+name = "jiff-core"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "7feca88439efe53da3754500c1851dedf3cb36c524dd5cf8225cc0794de95d09"
+dependencies = [
+ "defmt",
+]
+
+[[package]]
+name = "jiff-static"
+version = "0.2.35"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "3a69dcb3a21cfb32ce1cd056169337ca284af0766dd766e7878819b251a49204"
+dependencies = [
+ "jiff-core",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
+[[package]]
+name = "jiff-tzdb"
+version = "0.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "142bd39932ad231f10513df9ab62661fead8719872150b7ad02a2df79f4e141e"
+
+[[package]]
+name = "jiff-tzdb-platform"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "875a5a69ac2bab1a891711cf5eccbec1ce0341ea805560dcd90b7a2e925132e8"
+dependencies = [
+ "jiff-tzdb",
+]
+
[[package]]
name = "jni"
version = "0.22.4"
@@ -2108,6 +2210,16 @@ version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
+[[package]]
+name = "md-5"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98"
+dependencies = [
+ "cfg-if",
+ "digest",
+]
+
[[package]]
name = "md5"
version = "0.8.1"
@@ -2168,7 +2280,7 @@ version = "0.31.3"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "cf20d2fde8ff38632c426f1165ed7436270b44f199fc55284c38276f9db47c3d"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"cfg-if",
"cfg_aliases",
"libc",
@@ -2269,7 +2381,7 @@ version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
]
[[package]]
@@ -2341,6 +2453,40 @@ version = "11.1.5"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
+[[package]]
+name = "opendal"
+version = "0.59.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "f950151f9587a51a7bed70a15fa0cff464eae96e41ae7499f97067bdafdf43eb"
+dependencies = [
+ "opendal-core",
+]
+
+[[package]]
+name = "opendal-core"
+version = "0.59.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "a43405d217dfdfb543f58847336d3af672897dd1939bb7dcf314b63cf364f1c9"
+dependencies = [
+ "anyhow",
+ "asyncband",
+ "base64 0.23.1",
+ "bytes",
+ "futures",
+ "http",
+ "jiff",
+ "log",
+ "md-5",
+ "percent-encoding",
+ "quick-xml",
+ "serde",
+ "serde_json",
+ "tokio",
+ "url",
+ "uuid",
+ "web-time",
+]
+
[[package]]
name = "openssl-probe"
version = "0.2.1"
@@ -2435,6 +2581,7 @@ dependencies = [
"num-integer",
"num-traits",
"object_store",
+ "opendal",
"parquet-geospatial",
"parquet-variant",
"parquet-variant-compute",
@@ -2602,6 +2749,15 @@ version = "1.14.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "3d20d5497ef88037a52ff98267d066e7f11fcc5e99bbfbd58a42336193aacec3"
+[[package]]
+name = "portable-atomic-util"
+version = "0.2.8"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "10ab3eb7f3becc3a1cbc4f2c6f20267996cfc1a6467a873763411b136a122715"
+dependencies = [
+ "portable-atomic",
+]
+
[[package]]
name = "postcard"
version = "1.1.3"
@@ -2706,7 +2862,7 @@ version = "1.11.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "4b45fcc2344c680f5025fe57779faef368840d0bd1f42f216291f0dc4ace4744"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"num-traits",
"rand 0.9.5",
"rand_chacha",
@@ -2987,7 +3143,7 @@ version = "0.5.18"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
]
[[package]]
@@ -3099,7 +3255,7 @@ version = "1.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"errno",
"libc",
"linux-raw-sys",
@@ -3225,7 +3381,7 @@ version = "3.7.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"core-foundation",
"core-foundation-sys",
"libc",
@@ -3825,7 +3981,7 @@ version = "0.6.11"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840"
dependencies = [
- "bitflags",
+ "bitflags 2.13.1",
"bytes",
"futures-util",
"http",
diff --git a/parquet/Cargo.toml b/parquet/Cargo.toml
index e4d4855d99..d918be65d8 100644
--- a/parquet/Cargo.toml
+++ b/parquet/Cargo.toml
@@ -91,6 +91,7 @@ arrow-cast = { workspace = true }
tokio = { version = "1.0", default-features = false, features = ["macros",
"rt-multi-thread", "io-util", "fs", "sync"] }
rand = { version = "0.10", default-features = false, features = ["std",
"std_rng", "thread_rng"] }
object_store = { workspace = true, features = ["azure", "fs"] }
+opendal = { version = "0.59.1", default-features = false, features =
["services-memory"] }
sysinfo = { version = "0.39.6", default-features = false, features =
["system"] }
[package.metadata.docs.rs]
@@ -163,6 +164,11 @@ name = "object_store"
required-features = ["arrow", "async"]
path = "./examples/object_store.rs"
+[[example]]
+name = "opendal"
+required-features = ["arrow", "async"]
+path = "./examples/opendal.rs"
+
[[example]]
name = "custom_page_index"
required-features = ["arrow"]
diff --git a/parquet/examples/opendal.rs b/parquet/examples/opendal.rs
new file mode 100644
index 0000000000..0006c1d2ee
--- /dev/null
+++ b/parquet/examples/opendal.rs
@@ -0,0 +1,155 @@
+// 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.
+
+//! Read and write Parquet with Apache OpenDAL using an in-memory service.
+//!
+//! Run with `cargo run -p parquet --example opendal --features async`.
+//! Replace the Memory service with another OpenDAL service to use remote
storage.
+//! The adapters use OpenDAL's native buffer APIs for range reads and writes.
+//!
+//! Applications can also use the [`parquet_opendal`] crate, which provides
+//! ready-made `AsyncReader` and `AsyncWriter` adapters to avoid writing this
glue code.
+//!
+//! [`parquet_opendal`]: https://docs.rs/parquet_opendal
+
+use arrow_array::{ArrayRef, Int64Array, RecordBatch};
+use bytes::Bytes;
+use futures::future::BoxFuture;
+use futures::{FutureExt, TryStreamExt};
+use opendal::{Operator, Reader, Writer, services::Memory};
+use parquet::arrow::arrow_reader::ArrowReaderOptions;
+use parquet::arrow::async_reader::{AsyncFileReader, SpawnedReader};
+use parquet::arrow::async_writer::AsyncFileWriter;
+use parquet::arrow::{AsyncArrowWriter, ParquetRecordBatchStreamBuilder};
+use parquet::errors::{ParquetError, Result};
+use parquet::file::metadata::{ParquetMetaData, ParquetMetaDataReader};
+use std::ops::Range;
+use std::sync::Arc;
+
+#[tokio::main]
+async fn main() -> Result<()> {
+ let operator = Operator::new(Memory::default()).map_err(to_parquet_err)?;
+ let path = "example.parquet";
+ let col = Arc::new(Int64Array::from_iter_values([1, 2, 3])) as ArrayRef;
+ let batch = RecordBatch::try_from_iter([("col", col)]).unwrap();
+
+ // Closing the Parquet writer also closes the OpenDAL writer, committing
the object.
+ let writer =
OpenDalWriter(operator.writer(path).await.map_err(to_parquet_err)?);
+ let mut writer = AsyncArrowWriter::try_new(writer, batch.schema(), None)?;
+ writer.write(&batch).await?;
+ writer.close().await?;
+
+ let reader = OpenDalReader::new(&operator, path).await?;
+ let builder = ParquetRecordBatchStreamBuilder::new(reader.clone()).await?;
+ let read: Vec<RecordBatch> = builder.build()?.try_collect().await?;
+ assert_eq!(read, vec![batch.clone()]);
+ println!("read {} rows", read[0].num_rows());
+
+ // The same adapter can perform I/O on a runtime separate from Parquet
decoding.
+ let io_runtime = tokio::runtime::Builder::new_multi_thread()
+ .worker_threads(1)
+ .enable_all()
+ .build()
+ .expect("failed to build I/O runtime");
+ 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(error: opendal::Error) -> ParquetError {
+ ParquetError::External(Box::new(error))
+}
+
+/// Reads byte ranges through OpenDAL and uses the object size to locate the
footer.
+#[derive(Clone)]
+struct OpenDalReader {
+ reader: Reader,
+ file_size: u64,
+}
+
+impl OpenDalReader {
+ async fn new(operator: &Operator, path: &str) -> Result<Self> {
+ let file_size = operator
+ .stat(path)
+ .await
+ .map_err(to_parquet_err)?
+ .content_length();
+ let reader = operator.reader(path).await.map_err(to_parquet_err)?;
+ Ok(Self { reader, file_size })
+ }
+}
+
+impl AsyncFileReader for OpenDalReader {
+ fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, Result<Bytes>>
{
+ async move {
+ self.reader
+ .read(range)
+ .await
+ .map(|buffer| buffer.to_bytes())
+ .map_err(to_parquet_err)
+ }
+ .boxed()
+ }
+
+ fn get_byte_ranges(&mut self, ranges: Vec<Range<u64>>) -> BoxFuture<'_,
Result<Vec<Bytes>>> {
+ async move {
+ self.reader
+ .fetch(ranges)
+ .await
+ .map(|buffers| {
+ buffers
+ .into_iter()
+ .map(|buffer| buffer.to_bytes())
+ .collect()
+ })
+ .map_err(to_parquet_err)
+ }
+ .boxed()
+ }
+
+ fn get_metadata<'a>(
+ &'a mut self,
+ options: Option<&'a ArrowReaderOptions>,
+ ) -> BoxFuture<'a, Result<Arc<ParquetMetaData>>> {
+ async move {
+ let file_size = self.file_size;
+ let metadata = ParquetMetaDataReader::new()
+ .with_arrow_reader_options(options)
+ .load_and_finish(self, file_size)
+ .await?;
+ Ok(Arc::new(metadata))
+ }
+ .boxed()
+ }
+}
+
+/// Passes owned bytes directly to OpenDAL without an AsyncWrite compatibility
layer.
+struct OpenDalWriter(Writer);
+
+impl AsyncFileWriter for OpenDalWriter {
+ fn write(&mut self, bytes: Bytes) -> BoxFuture<'_, Result<()>> {
+ async move { self.0.write(bytes).await.map_err(to_parquet_err)
}.boxed()
+ }
+
+ fn complete(&mut self) -> BoxFuture<'_, Result<()>> {
+ async move { self.0.close().await.map(|_| ()).map_err(to_parquet_err)
}.boxed()
+ }
+}
