Add BinaryFormatSupport and Row Encoder to arrow-avro Writer (#9171)

# Which issue does this PR close?

- Closes #8701.

# Rationale for this change

`arrow-avro` already supports writing Avro Object Container Files (OCF)
and framed streaming encodings (e.g. Single-Object Encoding / registry
wire formats). However, many systems exchange **raw Avro binary datum
payloads** (i.e. *only* the Avro record body bytes) while supplying the
schema out-of-band (configuration, RPC contract, topic metadata, etc.).

Without first-class support for unframed datum output, users must
either:
- accept framing overhead that downstream systems don’t expect, or
- re-implement datum encoding themselves.

This PR adds the missing unframed write path and exposes a row-by-row
encoding API to make it easy to embed Avro datums into other transport
protocols.

# What changes are included in this PR?

- Added `AvroBinaryFormat` (unframed) as an `AvroFormat` implementation
to emit **raw Avro record body bytes** (no SOE prefix and no OCF header)
and to explicitly reject container-level compression for this format.
- Added `RecordEncoder::encode_rows` to encode a `RecordBatch` into a
single contiguous buffer while tracking per-row boundaries via appended
offsets.
- Introduced a higher-level `Encoder` + `EncodedRows` API for row-by-row
streaming use cases, providing zero-copy access to individual row slices
(via `Bytes`).
- Updated the writer API to provide `build_encoder` for stream formats
(e.g. SOE) and added row-capacity configuration to better support
incremental/streaming workflows.
- Added the `bytes` crate as a dependency to support efficient buffering
and slicing in the row encoder, and adjusted dev-dependencies to support
the new tests/docs.

# Are these changes tested?

Yes.

This PR adds unit tests that cover:
- single- and multi-column row encoding
- nullable columns
- prefix-based vs. unprefixed row encoding behavior
- empty batch encoding
- appending to existing output buffers and validating offset invariants

# Are there any user-facing changes?

Yes, these changes are additive (no breaking public API changes
expected).

- New writer format support for **unframed Avro binary datum** output
(`AvroBinaryFormat`).
- New row-by-row encoding APIs (`RecordEncoder::encode_rows`, `Encoder`,
`EncodedRows`) to support zero-copy access to per-row encoded bytes.
- New `WriterBuilder` functionality (`build_encoder` + row-capacity
configuration) to enable encoder construction without committing to a
specific `Write` sink.

---------

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: jecsand838

arrow-avro/Cargo.toml

@@ -66,6 +66,7 @@ indexmap = "2.10"
 rand = "0.9"
 md5 = { version = "0.8", optional = true }
 sha2 = { version = "0.10", optional = true }
+bytes = "1.11"
 
 [dev-dependencies]
 arrow-data = { workspace = true }
@@ -76,9 +77,8 @@ rand = { version = "0.9.1", default-features = false, features = [
 ] }
 criterion = { workspace = true, default-features = false }
 tempfile = "3.3"
-arrow = { workspace = true }
+arrow = { workspace = true, features = ["prettyprint"] }
 futures = "0.3.31"
-bytes = "1.10.1"
 async-stream = "0.3.6"
 apache-avro = "0.21.0"
 num-bigint = "0.4"
@@ -95,3 +95,7 @@ harness = false
 [[bench]]
 name = "avro_writer"
 harness = false
+
+[[bench]]
+name = "encoder"
+harness = false

arrow-avro/benches/encoder.rs

@@ -0,0 +1,87 @@
+// 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.
+
+//! Benchmarks for the `arrow-avro` Encoder
+
+use arrow_array::{ArrayRef, Int32Array, RecordBatch};
+use arrow_avro::writer::format::AvroSoeFormat;
+use arrow_avro::writer::{EncodedRows, WriterBuilder};
+use arrow_schema::{DataType, Field, Schema};
+use criterion::{BenchmarkId, Criterion, Throughput, criterion_group, criterion_main};
+use once_cell::sync::Lazy;
+use std::hint::black_box;
+use std::sync::Arc;
+use std::time::Duration;
+
+const SIZES: [usize; 4] = [1_000, 10_000, 100_000, 1_000_000];
+
+/// Pre-generate EncodedRows for each size to avoid setup overhead in benchmarks.
+static ENCODED_DATA: Lazy<Vec<EncodedRows>> =
+    Lazy::new(|| SIZES.iter().map(|&n| make_encoded_rows(n)).collect());
+
+/// Create an EncodedRows with `n` rows of Int32 data.
+fn make_encoded_rows(n: usize) -> EncodedRows {
+    let schema = Schema::new(vec![Field::new("x", DataType::Int32, false)]);
+    let values: Vec<i32> = (0..n as i32).collect();
+    let batch = RecordBatch::try_new(
+        Arc::new(schema.clone()),
+        vec![Arc::new(Int32Array::from(values)) as ArrayRef],
+    )
+    .unwrap();
+    let mut encoder = WriterBuilder::new(schema)
+        .build_encoder::<AvroSoeFormat>()
+        .unwrap();
+    encoder.encode(&batch).unwrap();
+    encoder.flush()
+}
+
+fn bench_row_access(c: &mut Criterion) {
+    let mut group = c.benchmark_group("row_access");
+    for (idx, &size) in SIZES.iter().enumerate() {
+        let encoded = &ENCODED_DATA[idx];
+        let num_rows = encoded.len();
+        // Configure sampling based on data size
+        match size {
+            100_000 | 1_000_000 => {
+                group
+                    .sample_size(20)
+                    .measurement_time(Duration::from_secs(10))
+                    .warm_up_time(Duration::from_secs(3));
+            }
+            _ => {
+                group.sample_size(100);
+            }
+        }
+        group.throughput(Throughput::Elements(num_rows as u64));
+        group.bench_function(BenchmarkId::from_parameter(size), |b| {
+            b.iter(|| {
+                for i in 0..num_rows {
+                    black_box(encoded.row(i).unwrap());
+                }
+            })
+        });
+    }
+    group.finish();
+}
+
+criterion_group! {
+    name = encoder;
+    config = Criterion::default().configure_from_args();
+    targets = bench_row_access
+}
+
+criterion_main!(encoder);