This is an automated email from the ASF dual-hosted git repository.
Jefffrey 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 6faa52777b docs(avro): add OpenDAL read and write example (#11080)
6faa52777b is described below
commit 6faa52777bfe9ee112603fc717aaaf014d2dda96
Author: Xuanwo <[email protected]>
AuthorDate: Thu Sep 17 08:37:35 2026 +0800
docs(avro): add OpenDAL read and write example (#11080)
# Which issue does this PR close?
No associated issue. Follows the Parquet example in #11014.
# Rationale for this change
Demonstrate how to read Avro through OpenDAL using the generic async I/O
interfaces, alongside the existing object_store example.
# What changes are included in this PR?
Add an in-memory round-trip example with an `AsyncFileReader` adapter
using OpenDAL range reads and a `SpawnedReader` demonstration. The
synchronous Avro writer encodes a small OCF in memory before uploading
it. OpenDAL uses the same development dependency version and memory
service as the Parquet example.
# Are these changes tested?
The example asserts that ordinary and dedicated-runtime reads reproduce
the written `RecordBatch`, including nulls. It runs successfully with
`--features async` and with `--no-default-features --features async`.
Example Clippy with `-D warnings`, workspace formatting, and whitespace
checks pass.
# Are there any user-facing changes?
A new example, runnable with `cargo run -p arrow-avro --example opendal
--features async`. No public API changes. OpenDAL 0.59 requires Rust
1.91, above the workspace's declared Rust 1.88 MSRV; this affects
development targets that build the new dependency.
Co-authored-by: Andrew Lamb <[email protected]>
---
Cargo.lock | 1 +
arrow-avro/Cargo.toml | 6 +++
arrow-avro/examples/opendal.rs | 113 +++++++++++++++++++++++++++++++++++++++++
3 files changed, 120 insertions(+)
diff --git a/Cargo.lock b/Cargo.lock
index 6eea94409d..c695aad893 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -258,6 +258,7 @@ dependencies = [
"liblzma",
"md5",
"object_store",
+ "opendal",
"rand 0.10.2",
"serde",
"serde_json",
diff --git a/arrow-avro/Cargo.toml b/arrow-avro/Cargo.toml
index bfdeb74903..9e6581f199 100644
--- a/arrow-avro/Cargo.toml
+++ b/arrow-avro/Cargo.toml
@@ -95,6 +95,7 @@ futures = "0.3.31"
async-stream = "0.3.6"
apache-avro = "0.22.0"
object_store = { workspace = true, features = ["fs"] }
+opendal = { version = "0.59.1", default-features = false, features =
["services-memory"] }
half = { version = "2.1", default-features = false }
tokio = { version = "1.0", default-features = false, features = ["macros",
"rt-multi-thread", "io-util", "fs"] }
@@ -103,6 +104,11 @@ name = "object_store"
required-features = ["async"]
path = "./examples/object_store.rs"
+[[example]]
+name = "opendal"
+required-features = ["async"]
+path = "./examples/opendal.rs"
+
[[bench]]
name = "avro_reader"
harness = false
diff --git a/arrow-avro/examples/opendal.rs b/arrow-avro/examples/opendal.rs
new file mode 100644
index 0000000000..cd3403cc45
--- /dev/null
+++ b/arrow-avro/examples/opendal.rs
@@ -0,0 +1,113 @@
+// 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 Avro with Apache OpenDAL using an in-memory service.
+//!
+//! Run with `cargo run -p arrow-avro --example opendal --features async`.
+//! Replace the Memory service and enable the corresponding OpenDAL service
feature
+//! to use remote storage. The adapter uses OpenDAL's native buffer APIs for
range reads.
+//! The synchronous Avro writer buffers this small file in memory before
uploading it.
+
+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 opendal::{Operator, Reader, services::Memory};
+use std::ops::Range;
+use std::sync::Arc;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+ let operator = Operator::new(Memory::default())?;
+ let path = "example.avro";
+ let col = Arc::new(Int64Array::from(vec![Some(1), None, Some(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)?;
+ // Flush the final Avro block before uploading the complete file.
+ writer.finish()?;
+ operator.write(path, writer.into_inner()).await?;
+
+ let file_size = operator.stat(path).await?.content_length();
+ let reader = OpenDalReader(operator.reader(path).await?);
+ let stream = AsyncAvroFileReader::builder(reader.clone(), 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());
+
+ // The same adapter can perform I/O on a runtime separate from Avro
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 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(())
+}
+
+fn to_avro_err(error: opendal::Error) -> AvroError {
+ AvroError::External(Box::new(error))
+}
+
+/// Reads byte ranges through OpenDAL without an AsyncRead compatibility layer.
+#[derive(Clone)]
+struct OpenDalReader(Reader);
+
+impl AsyncFileReader for OpenDalReader {
+ fn get_bytes(&mut self, range: Range<u64>) -> BoxFuture<'_, Result<Bytes,
AvroError>> {
+ async move {
+ self.0
+ .read(range)
+ .await
+ .map(|buffer| buffer.to_bytes())
+ .map_err(to_avro_err)
+ }
+ .boxed()
+ }
+
+ fn get_byte_ranges(
+ &mut self,
+ ranges: Vec<Range<u64>>,
+ ) -> BoxFuture<'_, Result<Vec<Bytes>, AvroError>> {
+ async move {
+ self.0
+ .fetch(ranges)
+ .await
+ .map(|buffers| {
+ buffers
+ .into_iter()
+ .map(|buffer| buffer.to_bytes())
+ .collect()
+ })
+ .map_err(to_avro_err)
+ }
+ .boxed()
+ }
+}
