This is an automated email from the ASF dual-hosted git repository.
Jefffrey pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new bd4d32bbd7 fix(arrow-avro): decode records with no fields (#10771)
bd4d32bbd7 is described below
commit bd4d32bbd7806dbcbb2b0882fbe409f832d0de2c
Author: Han You <[email protected]>
AuthorDate: Wed Sep 2 10:02:42 2026 +0800
fix(arrow-avro): decode records with no fields (#10771)
An Avro record with no fields is legal, and encodes to zero bytes. The
reader accepts such a schema and decodes its (empty) bytes fine, but
fails at flush:
Invalid argument error: use StructArray::try_new_with_length or
StructArray::new_empty_fields to create a struct array with no
fields so that the length can be set correctly
`Decoder::Record` flushes through `StructArray::try_new`, which infers
its length from the first child array, but a zero-field record has no
child array.
`RecordDecoder::flush` (the top level) already handles this: it passes
its own `row_count` through `RecordBatchOptions::with_row_count`. This
change gives the nested case the same treatment. `Decoder::Record` now
carries a row count, incremented wherever a row is appended and reset at
`flush`.
# Which issue does this PR close?
- Closes #10770.
# Are these changes tested?
See included tests.
# Are there any user-facing changes?
No
---------
Co-authored-by: Han You <[email protected]>
---
arrow-avro/src/reader/mod.rs | 146 ++++++++++++++++++++++++++++++++++++++++
arrow-avro/src/reader/record.rs | 131 +++++++++++++++++++++++++++--------
2 files changed, 248 insertions(+), 29 deletions(-)
diff --git a/arrow-avro/src/reader/mod.rs b/arrow-avro/src/reader/mod.rs
index f05df596a8..fda99c6555 100644
--- a/arrow-avro/src/reader/mod.rs
+++ b/arrow-avro/src/reader/mod.rs
@@ -10060,4 +10060,150 @@ mod test {
.unwrap();
assert_eq!(seconds.value(0), 100);
}
+
+ /// Builds a `Decoder` for a single Confluent-framed writer schema
registered under `id`.
+ fn confluent_decoder(id: u32, writer_schema: AvroSchema) -> Decoder {
+ let mut store = SchemaStore::new_with_type(FingerprintAlgorithm::Id);
+ let _ = store
+ .set(Fingerprint::Id(id), writer_schema.clone())
+ .expect("set id schema");
+ ReaderBuilder::new()
+ .with_batch_size(8)
+ .with_reader_schema(writer_schema)
+ .with_writer_schema_store(store)
+ .with_active_fingerprint(Fingerprint::Id(id))
+ .build_decoder()
+ .expect("decoder")
+ }
+
+ /// An Avro record with no fields is legal: it holds no data and encodes
to zero bytes. It
+ /// still has to decode to a zero-field struct whose length tracks the
rows, which is not a
+ /// length `StructArray::try_new` can infer with no child array to read it
from.
+ #[test]
+ fn test_record_with_no_fields_decodes_as_zero_field_struct() {
+ let id = 11u32;
+ let mut decoder = confluent_decoder(
+ id,
+ AvroSchema::new(
+ r#"{"type":"record","name":"Reading","fields":[
+ {"name":"id","type":"long"},
+
{"name":"heartbeat","type":{"type":"record","name":"Heartbeat","fields":[]}}
+ ]}"#
+ .to_string(),
+ ),
+ );
+ // Two messages; `heartbeat` contributes no bytes to either.
+ let mut input = Vec::new();
+ for id_value in [7i64, 8i64] {
+ input.extend_from_slice(&make_id_prefix(id, 0));
+ input.extend_from_slice(&encode_zigzag(id_value));
+ }
+ assert_eq!(decoder.decode(&input).unwrap(), input.len());
+ let batch = decoder.flush().unwrap().expect("batch");
+
+ assert_eq!(batch.num_rows(), 2);
+ let ids = batch
+ .column(0)
+ .as_any()
+ .downcast_ref::<Int64Array>()
+ .expect("long column");
+ assert_eq!(ids.value(0), 7);
+ assert_eq!(ids.value(1), 8);
+ let heartbeat = batch.column(1).as_struct();
+ assert_eq!(heartbeat.num_columns(), 0);
+ assert_eq!(heartbeat.len(), 2);
+ assert_eq!(heartbeat.null_count(), 0);
+ }
+
+ /// The nullable form: the union index still selects the branch, and the
absent rows have to
+ /// be counted as well so the struct's length covers nulls and values
alike.
+ #[test]
+ fn test_nullable_record_with_no_fields_tracks_nulls() {
+ let id = 12u32;
+ let mut decoder = confluent_decoder(
+ id,
+ AvroSchema::new(
+ r#"{"type":"record","name":"Reading","fields":[
+ {"name":"heartbeat","type":["null",
+ {"type":"record","name":"Heartbeat","fields":[]}]}
+ ]}"#
+ .to_string(),
+ ),
+ );
+ // Branch 1 (the record, zero bytes), then branch 0 (null), then
branch 1 again.
+ let mut input = Vec::new();
+ for branch in [1i64, 0, 1] {
+ input.extend_from_slice(&make_id_prefix(id, 0));
+ input.extend_from_slice(&encode_zigzag(branch));
+ }
+ assert_eq!(decoder.decode(&input).unwrap(), input.len());
+ let batch = decoder.flush().unwrap().expect("batch");
+
+ assert_eq!(batch.num_rows(), 3);
+ let heartbeat = batch.column(0).as_struct();
+ assert_eq!(heartbeat.num_columns(), 0);
+ assert_eq!(heartbeat.len(), 3);
+ assert!(heartbeat.is_valid(0));
+ assert!(heartbeat.is_null(1));
+ assert!(heartbeat.is_valid(2));
+ }
+
+ /// Inside a list the element count comes from the block header alone,
since the elements
+ /// themselves occupy no bytes.
+ #[test]
+ fn test_list_of_records_with_no_fields() {
+ let id = 13u32;
+ let mut decoder = confluent_decoder(
+ id,
+ AvroSchema::new(
+ r#"{"type":"record","name":"Reading","fields":[
+ {"name":"heartbeats","type":{"type":"array","items":
+ {"type":"record","name":"Heartbeat","fields":[]}}}
+ ]}"#
+ .to_string(),
+ ),
+ );
+ // One row holding a block of three elements, then the terminating
zero block.
+ let mut input = make_id_prefix(id, 0);
+ input.extend_from_slice(&encode_zigzag(3));
+ input.extend_from_slice(&encode_zigzag(0));
+ assert_eq!(decoder.decode(&input).unwrap(), input.len());
+ let batch = decoder.flush().unwrap().expect("batch");
+
+ assert_eq!(batch.num_rows(), 1);
+ let heartbeats = batch.column(0).as_list::<i32>();
+ assert_eq!(heartbeats.value_length(0), 3);
+ let elements = heartbeats.values().as_struct();
+ assert_eq!(elements.num_columns(), 0);
+ assert_eq!(elements.len(), 3);
+ }
+
+ /// The same shape written by this crate's own writer and read back.
+ #[test]
+ fn test_ocf_roundtrip_record_with_no_fields() {
+ let schema = Schema::new(vec![
+ Field::new("id", DataType::Int32, false),
+ Field::new("heartbeat", DataType::Struct(Fields::empty()), false),
+ ]);
+ let batch = RecordBatch::try_new(
+ Arc::new(schema.clone()),
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2])) as ArrayRef,
+ Arc::new(StructArray::new_empty_fields(2, None)) as ArrayRef,
+ ],
+ )
+ .unwrap();
+
+ let bytes = write_ocf(&schema, &[batch]);
+ let mut reader = ReaderBuilder::new()
+ .build(Cursor::new(bytes))
+ .expect("reader");
+ let out = reader.next().expect("batch").expect("read");
+
+ assert_eq!(out.num_rows(), 2);
+ assert_eq!(out.column(0).as_primitive::<Int32Type>().values(), &[1,
2]);
+ let heartbeat = out.column(1).as_struct();
+ assert_eq!(heartbeat.num_columns(), 0);
+ assert_eq!(heartbeat.len(), 2);
+ }
}
diff --git a/arrow-avro/src/reader/record.rs b/arrow-avro/src/reader/record.rs
index 7e87c52bf9..63590ab695 100644
--- a/arrow-avro/src/reader/record.rs
+++ b/arrow-avro/src/reader/record.rs
@@ -250,12 +250,16 @@ enum Decoder {
/// String data encoded as UTF-8 bytes, but mapped to Arrow's
StringViewArray
StringView(OffsetBufferBuilder<i32>, Vec<u8>),
Array(FieldRef, OffsetBufferBuilder<i32>, Box<Decoder>),
- Record(
- Fields,
- Vec<Decoder>,
- Vec<Option<AvroLiteral>>,
- Option<Projector>,
- ),
+ Record {
+ fields: Fields,
+ decoders: Vec<Decoder>,
+ defaults: Vec<Option<AvroLiteral>>,
+ projector: Option<Projector>,
+ /// Rows appended since the last flush. A record with no fields
(legal) has no child
+ /// array to take a length from, so this is the only length available
to it; it is
+ /// maintained for every record to keep the arms uniform.
+ row_count: usize,
+ },
Map(
FieldRef,
OffsetBufferBuilder<i32>,
@@ -534,7 +538,13 @@ impl Decoder {
} else {
None
};
- Self::Record(arrow_fields.into(), encodings, field_defaults,
projector)
+ Self::Record {
+ fields: arrow_fields.into(),
+ decoders: encodings,
+ defaults: field_defaults,
+ projector,
+ row_count: 0,
+ }
}
(Codec::Map(child), _) => {
let val_field =
child.field_with_name(ArrowField::MAP_VALUE_FIELD_DEFAULT_NAME);
@@ -737,10 +747,15 @@ impl Decoder {
offsets.push_length(0);
}
}
- Self::Record(_, children, _, _) => {
- for child in children {
+ Self::Record {
+ decoders,
+ row_count,
+ ..
+ } => {
+ for child in decoders.iter_mut() {
child.append_nulls(count)?;
}
+ *row_count += count;
}
Self::Fixed(width, values) => {
values.resize(values.len() + (*width as usize) * count, 0)
@@ -1179,7 +1194,13 @@ impl Decoder {
inner.append_default(lit)
}
Self::Union(u) => u.append_default(lit),
- Self::Record(field_meta, decoders, field_defaults, _) => match lit
{
+ Self::Record {
+ fields: field_meta,
+ decoders,
+ defaults: field_defaults,
+ row_count,
+ ..
+ } => match lit {
AvroLiteral::Map(entries) => {
for (i, dec) in decoders.iter_mut().enumerate() {
let name = field_meta[i].name();
@@ -1191,6 +1212,7 @@ impl Decoder {
dec.append_null()?;
}
}
+ *row_count += 1;
Ok(())
}
AvroLiteral::Null => {
@@ -1201,6 +1223,7 @@ impl Decoder {
dec.append_null()?;
}
}
+ *row_count += 1;
Ok(())
}
_ => Err(AvroError::InvalidArgument(
@@ -1338,13 +1361,25 @@ impl Decoder {
let total_items = read_blocks(buf, |cursor|
encoding.decode(cursor))?;
off.push_length(total_items);
}
- Self::Record(_, encodings, _, None) => {
- for encoding in encodings {
+ Self::Record {
+ decoders,
+ projector: None,
+ row_count,
+ ..
+ } => {
+ for encoding in decoders {
encoding.decode(buf)?;
}
+ *row_count += 1;
}
- Self::Record(_, encodings, _, Some(proj)) => {
- proj.project_record(buf, encodings)?;
+ Self::Record {
+ decoders,
+ projector: Some(proj),
+ row_count,
+ ..
+ } => {
+ proj.project_record(buf, decoders)?;
+ *row_count += 1;
}
Self::Map(_, koff, moff, kdata, valdec) => {
let newly_added = read_blocks(buf, |cur| {
@@ -1517,12 +1552,19 @@ impl Decoder {
Ok(())
}
ResolutionPlan::Record(proj) => {
- let Self::Record(_, encodings, _, _) = self else {
+ let Self::Record {
+ decoders,
+ row_count,
+ ..
+ } = self
+ else {
return Err(AvroError::SchemaError(
"record projection provided for non-record
decoder".into(),
));
};
- proj.project_record(buf, encodings)
+ proj.project_record(buf, decoders)?;
+ *row_count += 1;
+ Ok(())
}
}
}
@@ -1662,12 +1704,24 @@ impl Decoder {
let offsets = flush_offsets(offsets);
Arc::new(ListArray::try_new(field.clone(), offsets, values,
nulls)?)
}
- Self::Record(fields, encodings, _, _) => {
- let arrays = encodings
+ Self::Record {
+ fields,
+ decoders,
+ row_count,
+ ..
+ } => {
+ let arrays = decoders
.iter_mut()
.map(|x| x.flush(None))
.collect::<Result<Vec<_>, _>>()?;
- Arc::new(StructArray::try_new(fields.clone(), arrays, nulls)?)
+ // In case there are no fields (legal, encodes to zero bytes),
manually specify
+ // the length for the output struct to account for this.
+ Arc::new(StructArray::try_new_with_length(
+ fields.clone(),
+ arrays,
+ nulls,
+ std::mem::replace(row_count, 0),
+ )?)
}
Self::Map(map_field, k_off, m_off, kdata, valdec) => {
let moff = flush_offsets(m_off);
@@ -1841,9 +1895,15 @@ impl ResolutionPlan {
(_, ResolutionInfo::EnumMapping(m)) => {
Ok(ResolutionPlan::EnumMapping(EnumResolution::new(m)))
}
- (Decoder::Record(_, _, field_defaults, _),
ResolutionInfo::Record(r)) => Ok(
- ResolutionPlan::Record(ProjectorBuilder::try_new(r,
field_defaults).build()?),
- ),
+ (
+ Decoder::Record {
+ defaults: field_defaults,
+ ..
+ },
+ ResolutionInfo::Record(r),
+ ) => Ok(ResolutionPlan::Record(
+ ProjectorBuilder::try_new(r, field_defaults).build()?,
+ )),
(_, ResolutionInfo::Record(_)) => Err(AvroError::SchemaError(
"record resolution on non-record decoder".into(),
)),
@@ -4284,15 +4344,16 @@ mod tests {
encodings.push(enc);
}
let fields: Fields = field_refs.into();
- Decoder::Record(
+ Decoder::Record {
fields,
- encodings,
- vec![None; reader_fields.len()],
- Some(Projector {
+ decoders: encodings,
+ defaults: vec![None; reader_fields.len()],
+ projector: Some(Projector {
writer_projections,
default_injections: Arc::from(Vec::<(usize,
AvroLiteral)>::new()),
}),
- )
+ row_count: 0,
+ }
}
#[test]
@@ -4674,7 +4735,13 @@ mod tests {
writer_projections: vec![],
default_injections: Arc::from(default_injections),
};
- Decoder::Record(fields, encodings, field_defaults, Some(projector))
+ Decoder::Record {
+ fields,
+ decoders: encodings,
+ defaults: field_defaults,
+ projector: Some(projector),
+ row_count: 0,
+ }
}
#[cfg(feature = "avro_custom_types")]
@@ -5252,7 +5319,13 @@ mod tests {
writer_projections: vec![],
default_injections: Arc::from(Vec::<(usize, AvroLiteral)>::new()),
};
- let mut rec = Decoder::Record(field_refs.into(), encoders,
field_defaults, Some(projector));
+ let mut rec = Decoder::Record {
+ fields: field_refs.into(),
+ decoders: encoders,
+ defaults: field_defaults,
+ projector: Some(projector),
+ row_count: 0,
+ };
let mut map: IndexMap<String, AvroLiteral> = IndexMap::new();
map.insert("a".to_string(), AvroLiteral::Int(9));
rec.append_default(&AvroLiteral::Map(map)).unwrap();
