This is an automated email from the ASF dual-hosted git repository.
andygrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-java.git
The following commit(s) were added to refs/heads/main by this push:
new fe4414a feat(avro): expose Avro reader via registerAvro and readAvro
(#60)
fe4414a is described below
commit fe4414a7f92eabca337682f47486f78b9f04d55f
Author: Lantao Jin <[email protected]>
AuthorDate: Tue May 19 12:22:20 2026 +0800
feat(avro): expose Avro reader via registerAvro and readAvro (#60)
---
core/pom.xml | 5 +
.../org/apache/datafusion/AvroReadOptions.java | 60 +++++++
.../java/org/apache/datafusion/SessionContext.java | 70 ++++++++
.../org/apache/datafusion/AvroReadOptionsTest.java | 58 ++++++
.../apache/datafusion/SessionContextAvroTest.java | 197 +++++++++++++++++++++
native/Cargo.lock | 166 +++++++++++++++++
native/Cargo.toml | 2 +-
native/build.rs | 1 +
native/src/avro.rs | 104 +++++++++++
native/src/lib.rs | 1 +
pom.xml | 6 +
native/build.rs => proto/avro_read_options.proto | 34 ++--
12 files changed, 687 insertions(+), 17 deletions(-)
diff --git a/core/pom.xml b/core/pom.xml
index 27c4276..5eddb3b 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -57,6 +57,11 @@ under the License.
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/core/src/main/java/org/apache/datafusion/AvroReadOptions.java
b/core/src/main/java/org/apache/datafusion/AvroReadOptions.java
new file mode 100644
index 0000000..098133a
--- /dev/null
+++ b/core/src/main/java/org/apache/datafusion/AvroReadOptions.java
@@ -0,0 +1,60 @@
+/*
+ * 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.
+ */
+
+package org.apache.datafusion;
+
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.datafusion.protobuf.AvroReadOptionsProto;
+
+/**
+ * Configuration knobs for Avro sources passed to {@link
SessionContext#registerAvro(String, String,
+ * AvroReadOptions)} and {@link SessionContext#readAvro(String,
AvroReadOptions)}.
+ *
+ * <p>Mirrors the subset of DataFusion's {@code AvroReadOptions} that maps
onto the Java surface
+ * today: {@code fileExtension} (default {@code ".avro"}) and an explicit
Arrow {@code schema} that
+ * bypasses on-read schema inference. {@code tablePartitionCols} is
intentionally deferred -- no
+ * other Java reader exposes Hive-style partitioning yet.
+ *
+ * <p>Avro carries its own per-block compression (snappy, deflate, bzip2, xz,
zstandard) inside the
+ * object container itself, negotiated when the file is written, so unlike CSV
/ NDJSON there is no
+ * {@code FileCompressionType} setter.
+ */
+public final class AvroReadOptions {
+
+ private String fileExtension = ".avro";
+ private Schema schema;
+
+ public AvroReadOptions fileExtension(String ext) {
+ this.fileExtension = ext;
+ return this;
+ }
+
+ public AvroReadOptions schema(Schema schema) {
+ this.schema = schema;
+ return this;
+ }
+
+ byte[] toBytes() {
+ return
AvroReadOptionsProto.newBuilder().setFileExtension(fileExtension).build().toByteArray();
+ }
+
+ Schema schema() {
+ return schema;
+ }
+}
diff --git a/core/src/main/java/org/apache/datafusion/SessionContext.java
b/core/src/main/java/org/apache/datafusion/SessionContext.java
index 328eb6d..049761d 100644
--- a/core/src/main/java/org/apache/datafusion/SessionContext.java
+++ b/core/src/main/java/org/apache/datafusion/SessionContext.java
@@ -362,6 +362,70 @@ public final class SessionContext implements AutoCloseable
{
return new DataFrame(dfHandle);
}
+ /** Register an Avro file (or directory of Avro files) as a table. */
+ public void registerAvro(String name, String path) {
+ registerAvro(name, path, new AvroReadOptions());
+ }
+
+ /**
+ * Register an Avro file (or directory of Avro files) as a table with the
supplied {@link
+ * AvroReadOptions}.
+ *
+ * @throws IllegalArgumentException if any of {@code name}, {@code path}, or
{@code options} is
+ * {@code null}.
+ * @throws RuntimeException if registration fails (path not found, schema
mismatch, etc.).
+ */
+ public void registerAvro(String name, String path, AvroReadOptions options) {
+ if (nativeHandle == 0) {
+ throw new IllegalStateException("SessionContext is closed");
+ }
+ if (name == null) {
+ throw new IllegalArgumentException("registerAvro name must be non-null");
+ }
+ if (path == null) {
+ throw new IllegalArgumentException("registerAvro path must be non-null");
+ }
+ if (options == null) {
+ throw new IllegalArgumentException("registerAvro options must be
non-null");
+ }
+ registerAvroWithOptions(
+ nativeHandle,
+ name,
+ path,
+ options.toBytes(),
+ options.schema() != null ? serializeSchemaIpc(options.schema()) :
null);
+ }
+
+ /** Read an Avro file as a {@link DataFrame} without registering it. */
+ public DataFrame readAvro(String path) {
+ return readAvro(path, new AvroReadOptions());
+ }
+
+ /**
+ * Read an Avro file as a {@link DataFrame} with the supplied {@link
AvroReadOptions}.
+ *
+ * @throws IllegalArgumentException if {@code path} or {@code options} is
{@code null}.
+ * @throws RuntimeException if the read fails.
+ */
+ public DataFrame readAvro(String path, AvroReadOptions options) {
+ if (nativeHandle == 0) {
+ throw new IllegalStateException("SessionContext is closed");
+ }
+ if (path == null) {
+ throw new IllegalArgumentException("readAvro path must be non-null");
+ }
+ if (options == null) {
+ throw new IllegalArgumentException("readAvro options must be non-null");
+ }
+ long dfHandle =
+ readAvroWithOptions(
+ nativeHandle,
+ path,
+ options.toBytes(),
+ options.schema() != null ? serializeSchemaIpc(options.schema()) :
null);
+ return new DataFrame(dfHandle);
+ }
+
/**
* Register a Java-implemented scalar UDF. After registration, the function
can be invoked by SQL
* via the UDF's name or referenced in DataFusion plans deserialised with
{@link #fromProto}.
@@ -443,6 +507,12 @@ public final class SessionContext implements AutoCloseable
{
private static native long readArrowWithOptions(
long handle, String path, byte[] optionsBytes, byte[] schemaIpcBytes);
+ private static native void registerAvroWithOptions(
+ long handle, String name, String path, byte[] optionsBytes, byte[]
schemaIpcBytes);
+
+ private static native long readAvroWithOptions(
+ long handle, String path, byte[] optionsBytes, byte[] schemaIpcBytes);
+
private static native void registerJsonWithOptions(
long handle, String name, String path, byte[] optionsBytes, byte[]
schemaIpcBytes);
diff --git a/core/src/test/java/org/apache/datafusion/AvroReadOptionsTest.java
b/core/src/test/java/org/apache/datafusion/AvroReadOptionsTest.java
new file mode 100644
index 0000000..804cd64
--- /dev/null
+++ b/core/src/test/java/org/apache/datafusion/AvroReadOptionsTest.java
@@ -0,0 +1,58 @@
+/*
+ * 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.
+ */
+
+package org.apache.datafusion;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+import java.util.List;
+
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.datafusion.protobuf.AvroReadOptionsProto;
+import org.junit.jupiter.api.Test;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+
+class AvroReadOptionsTest {
+
+ @Test
+ void defaultsRoundTripThroughProto() throws InvalidProtocolBufferException {
+ AvroReadOptionsProto p = AvroReadOptionsProto.parseFrom(new
AvroReadOptions().toBytes());
+ assertEquals(".avro", p.getFileExtension());
+ }
+
+ @Test
+ void fileExtensionRoundTripsThroughProto() throws
InvalidProtocolBufferException {
+ AvroReadOptionsProto p =
+ AvroReadOptionsProto.parseFrom(new
AvroReadOptions().fileExtension(".av").toBytes());
+ assertEquals(".av", p.getFileExtension());
+ }
+
+ @Test
+ void schemaIsHeldByReferenceAndNotInProto() {
+ Schema schema =
+ new Schema(List.of(new Field("x", FieldType.nullable(new
ArrowType.Int(32, true)), null)));
+ AvroReadOptions opts = new AvroReadOptions().schema(schema);
+ assertSame(schema, opts.schema());
+ }
+}
diff --git
a/core/src/test/java/org/apache/datafusion/SessionContextAvroTest.java
b/core/src/test/java/org/apache/datafusion/SessionContextAvroTest.java
new file mode 100644
index 0000000..e4e1bc3
--- /dev/null
+++ b/core/src/test/java/org/apache/datafusion/SessionContextAvroTest.java
@@ -0,0 +1,197 @@
+/*
+ * 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.
+ */
+
+package org.apache.datafusion;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.List;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.VarCharVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.avro.SchemaBuilder;
+import org.apache.avro.file.DataFileWriter;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+class SessionContextAvroTest {
+
+ /**
+ * Write three rows of {@code (id long, name string)} as a single Avro
object container file using
+ * the canonical Apache Avro Java writer. Returns the path the test can hand
to {@code
+ * registerAvro} / {@code readAvro}.
+ *
+ * <p>Avro's logical types are minimal here on purpose: the goal is to pin
that DataFusion's Avro
+ * reader sees the file at all and that the JNI plumbing wires schema
inference and SQL correctly.
+ * Richer Avro type coverage is a job for upstream's own datasource tests.
+ */
+ private static Path writePeopleAvro(Path dir, String name) throws
IOException {
+ org.apache.avro.Schema avroSchema =
+ SchemaBuilder.record("Person")
+ .namespace("org.apache.datafusion.test")
+ .fields()
+ .name("id")
+ .type()
+ .longType()
+ .noDefault()
+ .name("name")
+ .type()
+ .stringType()
+ .noDefault()
+ .endRecord();
+
+ Path file = dir.resolve(name);
+ GenericDatumWriter<GenericRecord> datumWriter = new
GenericDatumWriter<>(avroSchema);
+ try (DataFileWriter<GenericRecord> writer = new
DataFileWriter<>(datumWriter)) {
+ writer.create(avroSchema, file.toFile());
+ for (Object[] row :
+ new Object[][] {
+ {1L, "alice"}, {2L, "bob"}, {3L, "carol"},
+ }) {
+ GenericRecord rec = new GenericData.Record(avroSchema);
+ rec.put("id", row[0]);
+ rec.put("name", row[1]);
+ writer.append(rec);
+ }
+ }
+ return file;
+ }
+
+ @Test
+ void registerAvroInfersSchemaAndCounts(@TempDir Path tempDir) throws
Exception {
+ Path file = writePeopleAvro(tempDir, "people.avro");
+
+ try (BufferAllocator allocator = new RootAllocator();
+ SessionContext ctx = new SessionContext()) {
+ ctx.registerAvro("people", file.toAbsolutePath().toString());
+
+ try (DataFrame df = ctx.sql("SELECT COUNT(*) FROM people");
+ ArrowReader reader = df.collect(allocator)) {
+ assertTrue(reader.loadNextBatch());
+ BigIntVector count = (BigIntVector)
reader.getVectorSchemaRoot().getVector(0);
+ assertEquals(3L, count.get(0));
+ }
+
+ try (DataFrame df = ctx.sql("SELECT name FROM people WHERE id = 2");
+ ArrowReader reader = df.collect(allocator)) {
+ assertTrue(reader.loadNextBatch());
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ assertEquals(1, root.getRowCount());
+ VarCharVector names = (VarCharVector) root.getVector(0);
+ assertEquals("bob", new String(names.get(0)));
+ }
+ }
+ }
+
+ @Test
+ void readAvroYieldsTheStoredRows(@TempDir Path tempDir) throws Exception {
+ Path file = writePeopleAvro(tempDir, "people.avro");
+
+ try (BufferAllocator allocator = new RootAllocator();
+ SessionContext ctx = new SessionContext();
+ DataFrame df = ctx.readAvro(file.toAbsolutePath().toString());
+ ArrowReader reader = df.collect(allocator)) {
+ long total = 0;
+ while (reader.loadNextBatch()) {
+ total += reader.getVectorSchemaRoot().getRowCount();
+ }
+ assertEquals(3L, total);
+ }
+ }
+
+ @Test
+ void registerAvroWithCustomExtension(@TempDir Path tempDir) throws Exception
{
+ Path file = writePeopleAvro(tempDir, "people.av");
+
+ try (BufferAllocator allocator = new RootAllocator();
+ SessionContext ctx = new SessionContext()) {
+ ctx.registerAvro(
+ "t", file.toAbsolutePath().toString(), new
AvroReadOptions().fileExtension(".av"));
+
+ try (DataFrame df = ctx.sql("SELECT SUM(id) FROM t");
+ ArrowReader reader = df.collect(allocator)) {
+ assertTrue(reader.loadNextBatch());
+ BigIntVector sum = (BigIntVector)
reader.getVectorSchemaRoot().getVector(0);
+ assertEquals(6L, sum.get(0));
+ }
+ }
+ }
+
+ @Test
+ void readAvroWithExplicitSchemaIsAccepted(@TempDir Path tempDir) throws
Exception {
+ // Explicit schema overrides on-read inference. We supply the same schema
the file actually
+ // has, so query results stay correct; the test pins that the
explicit-schema code path is
+ // plumbed through and accepted (same as the Arrow / NDJSON readers).
+ Path file = writePeopleAvro(tempDir, "people.avro");
+ Schema schema =
+ new Schema(
+ List.of(
+ new Field("id", FieldType.nullable(new ArrowType.Int(64,
true)), null),
+ new Field("name", FieldType.nullable(new ArrowType.Utf8()),
null)));
+
+ try (BufferAllocator allocator = new RootAllocator();
+ SessionContext ctx = new SessionContext();
+ DataFrame df =
+ ctx.readAvro(file.toAbsolutePath().toString(), new
AvroReadOptions().schema(schema));
+ ArrowReader reader = df.collect(allocator)) {
+ assertTrue(reader.loadNextBatch());
+ VectorSchemaRoot root = reader.getVectorSchemaRoot();
+ assertEquals(3, root.getRowCount());
+ assertEquals("id", root.getSchema().getFields().get(0).getName());
+ assertEquals("name", root.getSchema().getFields().get(1).getName());
+ }
+ }
+
+ @Test
+ void registerAvroRejectsNullArguments() {
+ try (SessionContext ctx = new SessionContext()) {
+ AvroReadOptions opts = new AvroReadOptions();
+ assertThrows(IllegalArgumentException.class, () ->
ctx.registerAvro(null, "/p"));
+ assertThrows(IllegalArgumentException.class, () -> ctx.registerAvro("t",
null));
+ assertThrows(IllegalArgumentException.class, () ->
ctx.registerAvro(null, "/p", opts));
+ assertThrows(IllegalArgumentException.class, () -> ctx.registerAvro("t",
null, opts));
+ assertThrows(IllegalArgumentException.class, () -> ctx.registerAvro("t",
"/p", null));
+ }
+ }
+
+ @Test
+ void readAvroRejectsNullArguments() {
+ try (SessionContext ctx = new SessionContext()) {
+ AvroReadOptions opts = new AvroReadOptions();
+ assertThrows(IllegalArgumentException.class, () -> ctx.readAvro(null));
+ assertThrows(IllegalArgumentException.class, () -> ctx.readAvro(null,
opts));
+ assertThrows(IllegalArgumentException.class, () -> ctx.readAvro("/p",
null));
+ }
+ }
+}
diff --git a/native/Cargo.lock b/native/Cargo.lock
index bb9578f..418c359 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -67,6 +67,35 @@ version = "1.0.102"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
+[[package]]
+name = "apache-avro"
+version = "0.21.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "36fa98bc79671c7981272d91a8753a928ff6a1cd8e4f20a44c45bd5d313840bf"
+dependencies = [
+ "bigdecimal",
+ "bon",
+ "bzip2",
+ "crc32fast",
+ "digest",
+ "liblzma",
+ "log",
+ "miniz_oxide",
+ "num-bigint",
+ "quad-rand",
+ "rand",
+ "regex-lite",
+ "serde",
+ "serde_bytes",
+ "serde_json",
+ "snap",
+ "strum",
+ "strum_macros",
+ "thiserror 2.0.18",
+ "uuid",
+ "zstd",
+]
+
[[package]]
name = "ar_archive_writer"
version = "0.5.1"
@@ -368,6 +397,7 @@ dependencies = [
"num-bigint",
"num-integer",
"num-traits",
+ "serde",
]
[[package]]
@@ -408,6 +438,31 @@ dependencies = [
"generic-array",
]
+[[package]]
+name = "bon"
+version = "3.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "f47dbe92550676ee653353c310dfb9cf6ba17ee70396e1f7cf0a2020ad49b2fe"
+dependencies = [
+ "bon-macros",
+ "rustversion",
+]
+
+[[package]]
+name = "bon-macros"
+version = "3.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "519bd3116aeeb42d5372c29d982d16d0170d3d4a5ed85fc7dd91642ffff3c67c"
+dependencies = [
+ "darling",
+ "ident_case",
+ "prettyplease",
+ "proc-macro2",
+ "quote",
+ "rustversion",
+ "syn",
+]
+
[[package]]
name = "brotli"
version = "8.0.2"
@@ -644,6 +699,40 @@ dependencies = [
"memchr",
]
+[[package]]
+name = "darling"
+version = "0.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d"
+dependencies = [
+ "darling_core",
+ "darling_macro",
+]
+
+[[package]]
+name = "darling_core"
+version = "0.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0"
+dependencies = [
+ "ident_case",
+ "proc-macro2",
+ "quote",
+ "strsim",
+ "syn",
+]
+
+[[package]]
+name = "darling_macro"
+version = "0.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d"
+dependencies = [
+ "darling_core",
+ "quote",
+ "syn",
+]
+
[[package]]
name = "dashmap"
version = "6.1.0"
@@ -676,6 +765,7 @@ dependencies = [
"datafusion-common-runtime",
"datafusion-datasource",
"datafusion-datasource-arrow",
+ "datafusion-datasource-avro",
"datafusion-datasource-csv",
"datafusion-datasource-json",
"datafusion-datasource-parquet",
@@ -768,6 +858,7 @@ source =
"registry+https://github.com/rust-lang/crates.io-index";
checksum = "d72a11ca44a95e1081870d3abb80c717496e8a7acb467a1d3e932bb636af5cc2"
dependencies = [
"ahash",
+ "apache-avro",
"arrow",
"arrow-ipc",
"chrono",
@@ -856,6 +947,26 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "datafusion-datasource-avro"
+version = "53.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "a579c3bd290c66ea4b269493e75e8a3ed42c9c895a651f10210a29538aee50c4"
+dependencies = [
+ "apache-avro",
+ "arrow",
+ "async-trait",
+ "bytes",
+ "datafusion-common",
+ "datafusion-datasource",
+ "datafusion-physical-expr-common",
+ "datafusion-physical-plan",
+ "datafusion-session",
+ "futures",
+ "num-traits",
+ "object_store",
+]
+
[[package]]
name = "datafusion-datasource-csv"
version = "53.1.0"
@@ -1804,6 +1915,12 @@ version = "2.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index";
checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954"
+[[package]]
+name = "ident_case"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
+
[[package]]
name = "idna"
version = "1.1.0"
@@ -2101,6 +2218,7 @@ checksum =
"a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9"
dependencies = [
"num-integer",
"num-traits",
+ "serde",
]
[[package]]
@@ -2456,6 +2574,12 @@ dependencies = [
"cc",
]
+[[package]]
+name = "quad-rand"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "5a651516ddc9168ebd67b24afd085a718be02f8858fe406591b013d101ce2f40"
+
[[package]]
name = "quote"
version = "1.0.45"
@@ -2558,6 +2682,12 @@ dependencies = [
"regex-syntax",
]
+[[package]]
+name = "regex-lite"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "cab834c73d247e67f4fae452806d17d3c7501756d98c8808d7c9c7aa7d18f973"
+
[[package]]
name = "regex-syntax"
version = "0.8.10"
@@ -2632,6 +2762,17 @@ source =
"registry+https://github.com/rust-lang/crates.io-index";
checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e"
dependencies = [
"serde_core",
+ "serde_derive",
+]
+
+[[package]]
+name = "serde_bytes"
+version = "0.11.19"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8"
+dependencies = [
+ "serde",
+ "serde_core",
]
[[package]]
@@ -2761,6 +2902,30 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "strsim"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f"
+
+[[package]]
+name = "strum"
+version = "0.27.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "af23d6f6c1a224baef9d3f61e287d2761385a5b88fdab4eb4c6f11aeb54c4bcf"
+
+[[package]]
+name = "strum_macros"
+version = "0.27.2"
+source = "registry+https://github.com/rust-lang/crates.io-index";
+checksum = "7695ce3845ea4b33927c055a39dc438a45b059f7c1b3d91d38d10355fb8cbca7"
+dependencies = [
+ "heck",
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
[[package]]
name = "subtle"
version = "2.6.1"
@@ -3012,6 +3177,7 @@ checksum =
"ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76"
dependencies = [
"getrandom 0.4.2",
"js-sys",
+ "serde_core",
"wasm-bindgen",
]
diff --git a/native/Cargo.toml b/native/Cargo.toml
index b9fca20..983d7eb 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -26,7 +26,7 @@ crate-type = ["cdylib"]
[dependencies]
arrow = { version = "58", features = ["ffi"] }
-datafusion = "53.1.0"
+datafusion = { version = "53.1.0", features = ["avro"] }
datafusion-proto = "53.1.0"
futures = "0.3"
jni = "0.21"
diff --git a/native/build.rs b/native/build.rs
index 52b1127..78d99c1 100644
--- a/native/build.rs
+++ b/native/build.rs
@@ -20,6 +20,7 @@ fn main() {
"../proto/session_options.proto",
"../proto/file_compression_type.proto",
"../proto/arrow_read_options.proto",
+ "../proto/avro_read_options.proto",
"../proto/csv_read_options.proto",
"../proto/csv_write_options.proto",
"../proto/json_read_options.proto",
diff --git a/native/src/avro.rs b/native/src/avro.rs
new file mode 100644
index 0000000..85d4a07
--- /dev/null
+++ b/native/src/avro.rs
@@ -0,0 +1,104 @@
+// 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 datafusion::error::DataFusionError;
+use datafusion::execution::options::AvroReadOptions;
+use datafusion::prelude::SessionContext;
+use jni::objects::{JByteArray, JClass, JString};
+use jni::sys::jlong;
+use jni::JNIEnv;
+use prost::Message;
+
+use crate::errors::{try_unwrap_or_throw, JniResult};
+use crate::proto_gen::AvroReadOptionsProto;
+use crate::runtime;
+use crate::schema::decode_optional_schema;
+
+fn with_avro_options<R>(
+ env: &mut JNIEnv,
+ options_bytes: JByteArray,
+ schema_ipc_bytes: JByteArray,
+ f: impl FnOnce(AvroReadOptions) -> JniResult<R>,
+) -> JniResult<R> {
+ let bytes: Vec<u8> = env.convert_byte_array(&options_bytes)?;
+ let p = AvroReadOptionsProto::decode(bytes.as_slice())?;
+
+ let schema = decode_optional_schema(env, schema_ipc_bytes)?;
+
+ // AvroReadOptions exposes `file_extension` as a public field (not a
builder
+ // setter); `schema` is the only field with a fluent setter. Build via
+ // struct-update syntax to avoid clippy::field_reassign_with_default.
+ let file_ext = p.file_extension;
+ let mut opts = AvroReadOptions {
+ file_extension: &file_ext,
+ ..AvroReadOptions::default()
+ };
+ if let Some(ref s) = schema {
+ opts = opts.schema(s);
+ }
+
+ f(opts)
+}
+
+#[no_mangle]
+pub extern "system" fn
Java_org_apache_datafusion_SessionContext_registerAvroWithOptions<'local>(
+ mut env: JNIEnv<'local>,
+ _class: JClass<'local>,
+ handle: jlong,
+ name: JString<'local>,
+ path: JString<'local>,
+ options_bytes: JByteArray<'local>,
+ schema_ipc_bytes: JByteArray<'local>,
+) {
+ try_unwrap_or_throw(&mut env, (), |env| -> JniResult<()> {
+ if handle == 0 {
+ return Err("SessionContext handle is null".into());
+ }
+ let ctx = unsafe { &*(handle as *const SessionContext) };
+ let name: String = env.get_string(&name)?.into();
+ let path: String = env.get_string(&path)?.into();
+ with_avro_options(env, options_bytes, schema_ipc_bytes, |opts| {
+ runtime().block_on(async {
+ ctx.register_avro(&name, &path, opts).await?;
+ Ok::<(), DataFusionError>(())
+ })?;
+ Ok(())
+ })
+ })
+}
+
+#[no_mangle]
+pub extern "system" fn
Java_org_apache_datafusion_SessionContext_readAvroWithOptions<'local>(
+ mut env: JNIEnv<'local>,
+ _class: JClass<'local>,
+ handle: jlong,
+ path: JString<'local>,
+ options_bytes: JByteArray<'local>,
+ schema_ipc_bytes: JByteArray<'local>,
+) -> jlong {
+ try_unwrap_or_throw(&mut env, 0, |env| -> JniResult<jlong> {
+ if handle == 0 {
+ return Err("SessionContext handle is null".into());
+ }
+ let ctx = unsafe { &*(handle as *const SessionContext) };
+ let path: String = env.get_string(&path)?.into();
+ with_avro_options(env, options_bytes, schema_ipc_bytes, |opts| {
+ let df = runtime().block_on(ctx.read_avro(path, opts))?;
+ Ok(Box::into_raw(Box::new(df)) as jlong)
+ })
+ })
+}
diff --git a/native/src/lib.rs b/native/src/lib.rs
index 1472628..1d0f36d 100644
--- a/native/src/lib.rs
+++ b/native/src/lib.rs
@@ -16,6 +16,7 @@
// under the License.
mod arrow;
+mod avro;
mod csv;
mod errors;
mod json;
diff --git a/pom.xml b/pom.xml
index ab4bc93..0a92f4b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -43,6 +43,7 @@ under the License.
<datafusion.version>53.1.0</datafusion.version>
<protobuf.version>3.25.5</protobuf.version>
<arrow.version>19.0.0</arrow.version>
+ <avro.version>1.12.0</avro.version>
</properties>
<dependencyManagement>
@@ -77,6 +78,11 @@ under the License.
<artifactId>junit-jupiter</artifactId>
<version>${junit.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.avro</groupId>
+ <artifactId>avro</artifactId>
+ <version>${avro.version}</version>
+ </dependency>
</dependencies>
</dependencyManagement>
diff --git a/native/build.rs b/proto/avro_read_options.proto
similarity index 52%
copy from native/build.rs
copy to proto/avro_read_options.proto
index 52b1127..24c28c7 100644
--- a/native/build.rs
+++ b/proto/avro_read_options.proto
@@ -15,20 +15,22 @@
// specific language governing permissions and limitations
// under the License.
-fn main() {
- const PROTOS: &[&str] = &[
- "../proto/session_options.proto",
- "../proto/file_compression_type.proto",
- "../proto/arrow_read_options.proto",
- "../proto/csv_read_options.proto",
- "../proto/csv_write_options.proto",
- "../proto/json_read_options.proto",
- "../proto/parquet_read_options.proto",
- ];
- for p in PROTOS {
- println!("cargo:rerun-if-changed={p}");
- }
- let protoc = protoc_bin_vendored::protoc_bin_path().expect("vendored
protoc not available");
- std::env::set_var("PROTOC", protoc);
- prost_build::compile_protos(PROTOS, &["../proto"]).expect("failed to
compile protos");
+syntax = "proto3";
+
+package datafusion_java;
+
+option java_package = "org.apache.datafusion.protobuf";
+option java_multiple_files = true;
+
+// Options used to read Avro files. `file_extension` has a non-null Java
+// default and is always sent. The explicit Arrow schema, if present, is
+// transferred separately as Arrow IPC bytes through the JNI layer (mirroring
+// the parquet, csv, ndjson, and arrow paths) and is not encoded in this
+// message.
+//
+// Avro carries its own compression inside the file format (snappy, deflate,
+// bzip2, xz, zstandard) negotiated per object container, so unlike CSV/JSON
+// there is no `FileCompressionType` field here.
+message AvroReadOptionsProto {
+ string file_extension = 1;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
