draft: using arrow-avro

Author: getChan

Cargo.toml

@@ -89,7 +89,7 @@ version = "50.0.0"
 ahash = { version = "0.8", default-features = false, features = [
     "runtime-rng",
 ] }
-apache-avro = { version = "0.20", default-features = false }
+#apache-avro = { version = "0.20", default-features = false }
 arrow = { version = "56.2.0", features = [
     "prettyprint",
     "chrono-tz",
@@ -103,6 +103,7 @@ arrow-ipc = { version = "56.2.0", default-features = false, features = [
 ] }
 arrow-ord = { version = "56.2.0", default-features = false }
 arrow-schema = { version = "56.2.0", default-features = false }
+arrow-avro = { version = "56.2.0", default-features = false }
 async-trait = "0.1.89"
 bigdecimal = "0.4.8"
 bytes = "1.10"
@@ -196,6 +197,12 @@ unexpected_cfgs = { level = "warn", check-cfg = [
 ] }
 unused_qualifications = "deny"
 
+
+# TEMPORARY: override arrow-avro for testing
+[patch.crates-io]
+arrow = { git = "https://github.com/apache/arrow-rs.git", rev = "b444ea7127ebc8136564bc9af036353d0c90991b" }
+arrow-avro = { git = "https://github.com/apache/arrow-rs.git", rev = "b444ea7127ebc8136564bc9af036353d0c90991b" }
+
 # --------------------
 # Compilation Profiles
 # --------------------

datafusion/common/Cargo.toml

@@ -71,7 +71,7 @@ log = { workspace = true }
 object_store = { workspace = true, optional = true }
 parquet = { workspace = true, optional = true, default-features = true }
 paste = "1.0.15"
-pyo3 = { version = "0.25", optional = true }
+pyo3 = { version = "0.26.0", optional = true }
 recursive = { workspace = true, optional = true }
 sqlparser = { workspace = true, optional = true }
 tokio = { workspace = true }

datafusion/datasource-avro/Cargo.toml

@@ -31,8 +31,8 @@ version.workspace = true
 all-features = true
 
 [dependencies]
-apache-avro = { workspace = true }
 arrow = { workspace = true }
+arrow-avro = { workspace = true }
 async-trait = { workspace = true }
 bytes = { workspace = true }
 datafusion-common = { workspace = true, features = ["object_store", "avro"] }

datafusion/datasource-avro/src/avro_to_arrow/mod.rs

@@ -19,21 +19,20 @@
 //!
 //! [Avro]: https://avro.apache.org/docs/1.2.0/
 
-mod arrow_array_reader;
-mod reader;
-mod schema;
+// mod arrow_array_reader;
+// mod reader;
+// mod schema;
 
 use arrow::datatypes::Schema;
-pub use reader::{Reader, ReaderBuilder};
+use arrow_avro::reader::ReaderBuilder;
 
-pub use schema::to_arrow_schema;
+// pub use schema::to_arrow_schema;
 use std::io::Read;
 
 /// Read Avro schema given a reader
 pub fn read_avro_schema_from_reader<R: Read>(
     reader: &mut R,
 ) -> datafusion_common::Result<Schema> {
-    let avro_reader = apache_avro::Reader::new(reader)?;
-    let schema = avro_reader.writer_schema();
-    to_arrow_schema(schema)
+    let avro_reader = ReaderBuilder::new().build(reader)?;
+    Ok(avro_reader.schema().as_ref().clone())
 }

datafusion/datasource-avro/src/mod.rs

@@ -30,5 +30,4 @@ pub mod avro_to_arrow;
 pub mod file_format;
 pub mod source;
 
-pub use apache_avro;
 pub use file_format::*;

datafusion/datasource-avro/src/source.rs

@@ -20,9 +20,9 @@
 use std::any::Any;
 use std::sync::Arc;
 
-use crate::avro_to_arrow::Reader as AvroReader;
-
-use arrow::datatypes::SchemaRef;
+use arrow::datatypes::{Schema, SchemaRef};
+use arrow_avro::reader::{Reader, ReaderBuilder};
+use arrow_avro::schema::AvroSchema;
 use datafusion_common::error::Result;
 use datafusion_common::Statistics;
 use datafusion_datasource::file::FileSource;
@@ -51,13 +51,14 @@ impl AvroSource {
         Self::default()
     }
 
-    fn open<R: std::io::Read>(&self, reader: R) -> Result<AvroReader<'static, R>> {
-        AvroReader::try_new(
-            reader,
-            Arc::clone(self.schema.as_ref().expect("Schema must set before open")),
-            self.batch_size.expect("Batch size must set before open"),
-            self.projection.clone(),
-        )
+    fn open<R: std::io::Read>(&self, reader: R) -> Result<Reader<R>> {
+        let schema: &Schema = self.schema.as_ref().expect("Schema must set before open");
+        let avro_schema = AvroSchema::try_from(schema)?;
+        ReaderBuilder::new()
+            .with_reader_schema(avro_schema)
+            .with_batch_size(self.batch_size.expect("Batch size must set before open"))
+            .build(reader)
+            .map_err(Into::into)
     }
 }