Improve arrow-avro decoding for one-record messages

arrow-avro/README.md

@@ -28,6 +28,7 @@ This crate provides:
 
 - a **reader** that decodes Avro
   - **Object Container Files (OCF)**,
+  - **unframed binary datums**,
   - **Avro Single‑Object Encoding (SOE)**, and
   - **Confluent Schema Registry wire format**  
   into Arrow `RecordBatch`es; and
@@ -103,7 +104,10 @@ fn main() -> anyhow::Result<()> {
 }
 ```
 
-See the crate docs for runnable SOE and Confluent round‑trip examples.
+See the crate docs for runnable unframed-datum, SOE, and Confluent examples. Unframed Kafka
+messages or consecutive raw records can be decoded directly with `Decoder::decode` after
+selecting the known writer schema with `ReaderBuilder::with_active_fingerprint` and configuring
+`ReaderBuilder::with_decoder_mode(DecoderMode::UnframedDatum)`.
 
 ### Async reading (`async` feature)
 
@@ -209,6 +213,7 @@ the example on the `AsyncFileReader` trait documentation.
 ## What formats are supported?
 
 * **OCF (Object Container Files)**: self‑describing Avro files with header, optional compression, sync markers; reader and writer supported.
+* **Unframed binary datums**: bare Avro records with an externally known writer schema; configure `DecoderMode::UnframedDatum` and decode directly with `Decoder::decode`, without adding a synthetic framing prefix.
 * **Confluent Schema Registry wire format**: 1‑byte magic `0x00` + 4‑byte BE schema ID + Avro body; supports decode + encode helpers.
 * **Avro Single‑Object Encoding (SOE)**: 2‑byte magic `0xC3 0x01` + 8‑byte LE CRC‑64‑AVRO fingerprint + Avro body; supports decode + encode helpers.
 

arrow-avro/benches/decoder.rs

@@ -339,6 +339,38 @@ fn gen_nested(sc: &ApacheSchema, n: usize, prefix: &[u8]) -> Vec<u8> {
     )
 }
 
+fn gen_sparse_nested(sc: &ApacheSchema, n: usize, prefix: &[u8]) -> Vec<u8> {
+    encode_records_with_prefix(
+        sc,
+        prefix,
+        (0..n).map(|i| {
+            let active = (i / 100) % 3;
+            let event = |slot| {
+                if slot == active {
+                    Value::Union(
+                        1,
+                        Box::new(Value::Record(vec![
+                            ("id".into(), Value::Long(i as i64)),
+                            ("name".into(), Value::String(format!("event-{i}"))),
+                            (
+                                "timestamp".into(),
+                                Value::Long(1_700_000_000_000 + i as i64),
+                            ),
+                        ])),
+                    )
+                } else {
+                    Value::Union(0, Box::new(Value::Null))
+                }
+            };
+            Value::Record(vec![
+                ("a".into(), event(0)),
+                ("b".into(), event(1)),
+                ("c".into(), event(2)),
+            ])
+        }),
+    )
+}
+
 const LARGE_BATCH: usize = 65_536;
 const SMALL_BATCH: usize = 4096;
 
@@ -410,6 +442,7 @@ const INTERVAL_SCHEMA_ENCODE: &str = r#"{"type":"record","name":"DurRec","fields
 const ENUM_SCHEMA: &str = r#"{"type":"record","name":"EnumRec","fields":[{"name":"field1","type":{"type":"enum","name":"MyEnum","symbols":["A","B","C"]}}]}"#;
 const MIX_SCHEMA: &str = r#"{"type":"record","name":"MixRec","fields":[{"name":"f1","type":"int"},{"name":"f2","type":"long"},{"name":"f3","type":"string"},{"name":"f4","type":"double"}]}"#;
 const NEST_SCHEMA: &str = r#"{"type":"record","name":"NestRec","fields":[{"name":"sub","type":{"type":"record","name":"Sub","fields":[{"name":"x","type":"int"},{"name":"y","type":"string"}]}}]}"#;
+const SPARSE_NEST_SCHEMA: &str = r#"{"type":"record","name":"SparseNestRec","fields":[{"name":"a","type":["null",{"type":"record","name":"Event","fields":[{"name":"id","type":"long"},{"name":"name","type":"string"},{"name":"timestamp","type":"long"}]}]},{"name":"b","type":["null","Event"]},{"name":"c","type":["null","Event"]}]}"#;
 
 macro_rules! dataset {
     ($name:ident, $schema_json:expr, $gen_fn:ident) => {
@@ -468,6 +501,7 @@ dataset!(INTERVAL_DATA, INTERVAL_SCHEMA_ENCODE, gen_interval);
 dataset!(ENUM_DATA, ENUM_SCHEMA, gen_enum);
 dataset!(MIX_DATA, MIX_SCHEMA, gen_mixed);
 dataset!(NEST_DATA, NEST_SCHEMA, gen_nested);
+dataset!(SPARSE_NEST_DATA, SPARSE_NEST_SCHEMA, gen_sparse_nested);
 
 fn bench_with_decoder<F>(
     c: &mut Criterion,
@@ -582,6 +616,9 @@ fn criterion_benches(c: &mut Criterion) {
         bench_with_decoder(c, "Nested(Struct)", &NEST_DATA, &SIZES, || {
             new_decoder(NEST_SCHEMA, batch_size, false)
         });
+        bench_with_decoder(c, "SparseNested(Struct)", &SPARSE_NEST_DATA, &SIZES, || {
+            new_decoder(SPARSE_NEST_SCHEMA, batch_size, false)
+        });
     }
 }
 

arrow-avro/src/errors.rs

@@ -60,6 +60,8 @@ pub enum AvroError {
     /// Returned when a function needs more data to complete properly.
     /// The `Range<u64>` indicates the range of bytes that are needed.
     NeedMoreDataRange(std::ops::Range<u64>),
+    /// Returned when an unframed datum cannot be decoded until the current batch is flushed.
+    BatchFull,
 }
 
 impl std::fmt::Display for AvroError {
@@ -85,6 +87,9 @@ impl std::fmt::Display for AvroError {
             AvroError::NeedMoreDataRange(range) => {
                 write!(fmt, "NeedMoreDataRange: {}..{}", range.start, range.end)
             }
+            AvroError::BatchFull => {
+                write!(fmt, "Batch is full; flush before decoding another datum")
+            }
         }
     }
 }

arrow-avro/src/lib.rs

@@ -18,8 +18,9 @@
 //! Convert data to / from the [Apache Arrow] memory format and [Apache Avro].
 //!
 //! This crate provides:
-//! - a [`reader`] that decodes Avro (Object Container Files, Avro Single‑Object encoding,
-//!   and Confluent Schema Registry wire format) into Arrow `RecordBatch`es,
+//! - a [`reader`] that decodes Avro (Object Container Files, unframed binary datums,
+//!   Avro Single‑Object encoding, and Confluent Schema Registry wire format) into Arrow
+//!   `RecordBatch`es,
 //! - and a [`writer`] that encodes Arrow `RecordBatch`es into Avro (OCF or SOE).
 //!
 //! If you’re new to Arrow or Avro, see:
@@ -64,6 +65,45 @@
 //! # Ok(()) }
 //! ```
 //!
+//! ## Quickstart: unframed Avro datums *(runnable)*
+//!
+//! Kafka messages and other transports can contain bare Avro records without an OCF header,
+//! single-object prefix, or schema-registry framing. When the writer schema is already known,
+//! register it, select its fingerprint and [`reader::DecoderMode::UnframedDatum`], and call
+//! [`reader::Decoder::decode`] once per record. The returned byte count also supports
+//! consecutive datums in one buffer.
+//!
+//! ```
+//! use arrow_array::{Array, Int64Array};
+//! use arrow_avro::reader::{DecoderMode, ReaderBuilder};
+//! use arrow_avro::schema::{AvroSchema, SchemaStore};
+//!
+//! # fn main() -> Result<(), Box<dyn std::error::Error>> {
+//! let schema = AvroSchema::new(
+//!     r#"{"type":"record","name":"Event","fields":[{"name":"id","type":"long"}]}"#
+//!         .to_string(),
+//! );
+//! let mut store = SchemaStore::new();
+//! let fingerprint = store.register(schema)?;
+//! let mut decoder = ReaderBuilder::new()
+//!     .with_writer_schema_store(store)
+//!     .with_active_fingerprint(fingerprint)
+//!     .with_decoder_mode(DecoderMode::UnframedDatum)
+//!     .build_decoder()?;
+//!
+//! // Two consecutive records, {id: 7} and {id: 42}, in Avro zigzag encoding.
+//! let mut remaining: &[u8] = &[0x0e, 0x54];
+//! while !remaining.is_empty() {
+//!     let consumed = decoder.decode(remaining)?;
+//!     remaining = &remaining[consumed..];
+//! }
+//!
+//! let batch = decoder.flush()?.expect("decoded records");
+//! let ids = batch.column(0).as_any().downcast_ref::<Int64Array>().unwrap();
+//! assert_eq!(ids.values(), &[7, 42]);
+//! # Ok(()) }
+//! ```
+//!
 //! ## Quickstart: SOE (Single‑Object Encoding) round‑trip *(runnable)*
 //!
 //! Avro **Single‑Object Encoding (SOE)** wraps an Avro body with a 2‑byte marker
@@ -163,7 +203,7 @@
 //!
 //! ### Modules
 //!
-//! - [`reader`]: read Avro (OCF, SOE, Confluent) into Arrow `RecordBatch`es.
+//! - [`reader`]: read Avro (OCF, unframed datums, SOE, Confluent) into Arrow `RecordBatch`es.
 //!   - With the `async` feature: [`AsyncAvroFileReader`] for async streaming reads,
 //!     from any [`AsyncFileReader`] source including cloud object storage.
 //! - [`writer`]: write Arrow `RecordBatch`es as Avro (OCF, SOE, Confluent, Apicurio).