jordepic commented on code in PR #10713:
URL: https://github.com/apache/arrow-rs/pull/10713#discussion_r3833397754
##########
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:
Implemented in 2a5f2d925. The public decode_datum method is gone;
ReaderBuilder::with_decoder_mode now selects DecoderMode::Framed (default) or
DecoderMode::UnframedDatum at construction, and Decoder::decode dispatches
internally. Updated the crate/module/builder docs, runnable example, README,
and existing unframed tests to use the single decoding entry point.
##########
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:
Addressed in 2a5f2d925. append_nulls now matches Decoder exhaustively and
has direct bulk arms for arrays/maps/strings with reserved offsets, fixed
values, UUIDs, enums, all decimal widths, duration builders, custom
primitive/temporal types, nested records, and run-end encoding. Union handling
remains explicitly per-row where required, and append_null simply delegates to
append_nulls(1).
##########
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:
Addressed in 2a5f2d925 with AvroError::BatchFull for unframed decoding when
capacity is exhausted. A successful zero-width datum still returns Ok(0), and
the new regression test covers both an empty record and a record containing
only a null field, including flushing and decoding another row without
duplication.
##########
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:
Implemented in 2a5f2d925. NullableDecoder::materialize_pending now
bulk-materializes child placeholders and the validity suffix together, while
null branches only increment pending_nulls. Non-null/default paths skip the
bulk dispatcher when no nulls are pending and append validity only after the
value succeeds. Added a regression test that verifies neither bitmap nor child
values are touched during a 64-row null run.
##########
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:
Implemented in 2a5f2d925 with a named NullableDecoder containing plan,
validity, values, and pending_nulls. Its materialize_pending method owns the
shared invariant, and append_null, bulk append, defaults, decoding, and flush
all transition through that representation consistently.
--
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]
