jecsand838 commented on code in PR #10713:
URL: https://github.com/apache/arrow-rs/pull/10713#discussion_r3833146137
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -736,6 +741,31 @@ impl Decoder {
Ok(total_consumed)
}
+ /// Decode exactly one unframed Avro datum with the active writer schema.
+ ///
+ /// This is intended for transports such as Kafka where the message
boundary is external to
+ /// Avro. It returns the number of datum bytes consumed, allowing the
caller to ignore transport
+ /// payload bytes after the first datum when its format contract requires
that behavior.
+ /// Consecutive unframed datums can be decoded by repeatedly passing the
unconsumed suffix.
+ /// If the current batch is full, this method returns `Ok(0)` until
[`Self::flush`] is called.
+ ///
+ /// The decoder must already have the desired active fingerprint, and this
method does not
+ /// inspect or switch framing fingerprints.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if the datum is incomplete, malformed, or
incompatible with the active
+ /// writer schema.
+ pub fn decode_datum(&mut self, data: &[u8]) -> Result<usize, AvroError> {
+ if self.remaining_capacity == 0 {
+ return Ok(0);
+ }
Review Comment:
The new unframed mode may make `Ok(0)` too ambiguous. An empty record or a
record containing only `null` fields is a valid datum that consumes zero bytes
but still appends one row. The same result currently means the batch was
already full and no row was appended. A caller that interprets zero as flush
and retry can therefore duplicate a successfully decoded row.
I'd recommending giving batch-full a distinct outcome such as
`AvroError::BatchFull` or something along those lines:
```rust
if self.capacity == 0 {
return Err(AvroError::BatchFull);
}
// A successful zero-width datum remains unambiguously Ok(0)
let consumed = self.active_decoder.decode(data, 1)?;
self.capacity -= 1;
Ok(consumed)
```
##########
arrow-avro/src/reader/record.rs:
##########
@@ -717,9 +719,61 @@ impl Decoder {
inner.append_null()?;
}
Self::Union(u) => u.append_null()?,
- Self::Nullable(_, null_buffer, inner) => {
+ Self::Nullable(_, null_buffer, _, pending_nulls) => {
Review Comment:
This defers child placeholders, but each null still calls
`NullBufferBuilder::append(false)`, so long null runs retain one bitmap write
per row. The non-null path also calls the large `append_nulls` dispatcher when
`pending_nulls == 0`.
Could we defer both the validity suffix and child placeholders behind one
helper?
```rust
#[inline]
fn materialize_pending(
validity: &mut NullBufferBuilder,
values: &mut Decoder,
pending: &mut usize,
) -> Result<(), AvroError> {
let count = *pending;
if count == 0 {
return Ok(());
}
values.append_nulls(count)?;
validity.append_n_nulls(count);
*pending = 0;
Ok(())
}
```
Null branches would only increment `pending_nulls`. Before a
non-null/default value or flush, call this helper; after successfully decoding
a value, call `validity.append_non_null()`. This makes long and all-null runs
bulk operations and avoids zero-count dispatch on dense nullable data.
##########
arrow-avro/src/reader/record.rs:
##########
@@ -717,9 +719,61 @@ impl Decoder {
inner.append_null()?;
}
Self::Union(u) => u.append_null()?,
- Self::Nullable(_, null_buffer, inner) => {
+ Self::Nullable(_, null_buffer, _, pending_nulls) => {
null_buffer.append(false);
- inner.append_null()?;
+ *pending_nulls += 1;
+ }
+ }
+ Ok(())
+ }
+
+ /// Append a run of null placeholders, deferring nullable children until
their next value or
+ /// flush so sparse record subtrees can be materialized in bulk.
+ fn append_nulls(&mut self, count: usize) -> Result<(), AvroError> {
+ if count == 0 {
+ return Ok(());
+ }
+ match self {
+ Self::Null(size) => *size += count,
+ Self::Boolean(values) => values.append_n(count, false),
+ Self::Int32(values) | Self::Date32(values) |
Self::TimeMillis(values) => {
+ values.resize(values.len() + count, 0)
+ }
+ Self::Int64(values)
+ | Self::Int32ToInt64(values)
+ | Self::TimeMicros(values)
+ | Self::TimestampMillis(_, values)
+ | Self::TimestampMicros(_, values)
+ | Self::TimestampNanos(_, values) => values.resize(values.len() +
count, 0),
+ Self::Float32(values) | Self::Int32ToFloat32(values) |
Self::Int64ToFloat32(values) => {
+ values.resize(values.len() + count, 0.0)
+ }
+ Self::Float64(values)
+ | Self::Int32ToFloat64(values)
+ | Self::Int64ToFloat64(values)
+ | Self::Float32ToFloat64(values) => values.resize(values.len() +
count, 0.0),
+ Self::Binary(offsets, _)
+ | Self::String(offsets, _)
+ | Self::StringView(offsets, _)
+ | Self::BytesToString(offsets, _)
+ | Self::StringToBytes(offsets, _) => {
+ for _ in 0..count {
+ offsets.push_length(0);
+ }
+ }
+ Self::Record(_, children, _, _) => {
+ for child in children {
+ child.append_nulls(count)?;
+ }
+ }
+ Self::Nullable(_, null_buffer, _, pending_nulls) => {
+ null_buffer.append_n_nulls(count);
+ *pending_nulls += count;
+ }
+ other => {
+ for _ in 0..count {
+ other.append_null()?;
+ }
Review Comment:
The catch-all calls `append_null()` `count` times, re-running the full
`Decoder` match for every placeholder. Consequently, nullable arrays, maps,
fixed values, UUIDs, enums, decimals, custom primitives, and REE do not receive
the intended bulk optimization.
Perhaps add direct bulk arms where possible? for example:
```rust
Self::Uuid(values) => {
values.resize(values.len() + 16 * count, 0);
}
Self::Fixed(width, values) => {
values.resize(values.len() + (*width as usize) * count, 0);
}
Self::Enum(values, _, _) => {
values.resize(values.len() + count, 0);
}
Self::Decimal128(_, _, _, builder) => {
builder.append_value_n(0, count);
}
Self::Decimal256(_, _, _, builder) => {
builder.append_value_n(i256::ZERO, count);
}
Self::RunEndEncoded(_, len, inner) => {
inner.append_nulls(count)?;
*len += count;
}
```
For offset-backed types, we could reserve `count` offsets before pushing
repeated zero lengths. I would also make the match exhaustive and keep any
necessarily per-row union handling explicit, so future variants cannot silently
fall back to a slower implementation. This could always be looked into as a
follow-up.
##########
arrow-avro/src/reader/mod.rs:
##########
@@ -736,6 +741,31 @@ impl Decoder {
Ok(total_consumed)
}
+ /// Decode exactly one unframed Avro datum with the active writer schema.
+ ///
+ /// This is intended for transports such as Kafka where the message
boundary is external to
+ /// Avro. It returns the number of datum bytes consumed, allowing the
caller to ignore transport
+ /// payload bytes after the first datum when its format contract requires
that behavior.
+ /// Consecutive unframed datums can be decoded by repeatedly passing the
unconsumed suffix.
+ /// If the current batch is full, this method returns `Ok(0)` until
[`Self::flush`] is called.
+ ///
+ /// The decoder must already have the desired active fingerprint, and this
method does not
+ /// inspect or switch framing fingerprints.
+ ///
+ /// # Errors
+ ///
+ /// Returns an error if the datum is incomplete, malformed, or
incompatible with the active
+ /// writer schema.
+ pub fn decode_datum(&mut self, data: &[u8]) -> Result<usize, AvroError> {
+ if self.remaining_capacity == 0 {
+ return Ok(0);
+ }
+ let consumed = self.active_decoder.decode(data, 1)?;
+ self.remaining_capacity -= 1;
+ self.awaiting_body = false;
+ Ok(consumed)
+ }
Review Comment:
I'd recommend we avoid adding a second public decoding method and instead
select the input grammar when building the decoder.
A decoder should generally consume one stable wire format for its lifetime,
and this follows existing arrow-rs patterns such as `arrow_json::StructMode`,
IPC’s `DictionaryHandling`, and the Avro writer’s construction-time choice
between `AvroSoeFormat` and `AvroBinaryFormat`.
I suggest replacing `decode_datum` with:
```rust
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub enum DecoderMode {
#[default]
Framed,
UnframedDatum,
}
```
Then expose:
```rust
pub fn with_decoder_mode(mut self, mode: DecoderMode) -> Self {
self.decoder_mode = mode;
self
}
```
`Decoder::decode` can dispatch internally:
```rust
pub fn decode(&mut self, data: &[u8]) -> Result<usize, AvroError> {
match self.mode {
DecoderMode::Framed => self.decode_framed(data),
DecoderMode::UnframedDatum => self.decode_unframed(data),
}
}
```
##########
arrow-avro/src/reader/record.rs:
##########
@@ -276,7 +276,8 @@ enum Decoder {
#[cfg(feature = "avro_custom_types")]
RunEndEncoded(u8, usize, Box<Decoder>),
Union(UnionDecoder),
- Nullable(NullablePlan, NullBufferBuilder, Box<Decoder>),
+ /// Nullable value plus trailing null placeholders not yet materialized in
the child decoder.
+ Nullable(NullablePlan, NullBufferBuilder, Box<Decoder>, usize),
Review Comment:
Adding `pending_nulls` as a fourth positional field makes the nullable state
transitions difficult to audit across `append_null`, `append_nulls`,
`append_default`, `decode`, and `flush`.
Could this state be represented by a named structure?
```rust
struct NullableDecoder {
plan: NullablePlan,
validity: NullBufferBuilder,
values: Box<Decoder>,
pending_nulls: usize,
}
```
A `materialize_pending()` method on this structure would centralize the
invariant and ensure every transition updates the child, validity bitmap, and
pending count consistently. If `append_nulls` is made exhaustive,
`append_null()` could also delegate to `append_nulls(1)`, removing the
duplicated per-variant implementation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
