Repository: jorgecarleitao/arrow2 (rust)

Issue:
Potential bug in reading lists from avro?
Line of code: https://github.com/jorgecarleitao/arrow2/blob/c615095dbfb930a247c23423e36118843f71e1d2/src/io/avro/read/deserialize.rs#L136

Feel free to close this issue if my understanding is incorrect here. Referencing the Java Avro implementation, looks like array values are stored/read as blocks and can be read consecutively until a 0 is encountered. (https://github.com/apache/avro/blob/42822886c28ea74a744abb7e7a80a942c540faa5/lang/java/avro/src/main/java/org/apache/avro/io/Decoder.java#L203)

In the Avro deserialisation logic for reading in lists, looks like we are making a call to `try_push_valid` for each block in a list item, rather than just for the whole item (outside of the `loop`). Intuitively, it looks like this would be incorrect behavior since blocks are an Avro implementation detail and our validity bitmap is tracking Arrow types (e.g. a list item). 

I wasn't able to easily create a list in Avro that is stored as multiple blocks, so as to validate my assumptions, but wanted to open this issue in case it is an obvious bug. Thank you

Pull request description:
Fixed error reading unbounded Avro list
Fixed #1252 - Thanks to @shaeqahmed for the report!

Patch:
diff --git a/src/io/avro/read/deserialize.rs b/src/io/avro/read/deserialize.rs
index 5f53565014..54b7950a77 100644
--- a/src/io/avro/read/deserialize.rs
+++ b/src/io/avro/read/deserialize.rs
@@ -122,19 +122,32 @@ fn deserialize_value<'a>(
                 .as_mut_any()
                 .downcast_mut::<DynMutableListArray<i32>>()
                 .unwrap();
+            // Arrays are encoded as a series of blocks.
             loop {
-                let len = util::zigzag_i64(&mut block)? as usize;
+                // Each block consists of a long count value, followed by that many array items.
+                let len = util::zigzag_i64(&mut block)?;
+                let len = if len < 0 {
+                    // Avro spec: If a block's count is negative, its absolute value is used,
+                    // and the count is followed immediately by a long block size indicating the number of bytes in the block. This block size permits fast skipping through data, e.g., when projecting a record to a subset of its fields.
+                    let _ = util::zigzag_i64(&mut block)?;
+
+                    -len
+                } else {
+                    len
+                };
 
+                // A block with count zero indicates the end of the array.
                 if len == 0 {
                     break;
                 }
 
+                // Each item is encoded per the array’s item schema.
                 let values = array.mut_values();
                 for _ in 0..len {
                     block = deserialize_item(values, is_nullable, avro_inner, block)?;
                 }
-                array.try_push_valid()?;
             }
+            array.try_push_valid()?;
         }
         DataType::Struct(inner_fields) => {
             let fields = match avro_field {
@@ -341,14 +354,35 @@ fn skip_item<'a>(field: &Field, avro_field: &AvroSchema, mut block: &'a [u8]) ->
             };
 
             loop {
-                let len = util::zigzag_i64(&mut block)? as usize;
+                let len = util::zigzag_i64(&mut block)?;
+                let (len, bytes) = if len < 0 {
+                    // Avro spec: If a block's count is negative, its absolute value is used,
+                    // and the count is followed immediately by a long block size indicating the number of bytes in the block. This block size permits fast skipping through data, e.g., when projecting a record to a subset of its fields.
+                    let bytes = util::zigzag_i64(&mut block)?;
+
+                    (-len, Some(bytes))
+                } else {
+                    (len, None)
+                };
+
+                let bytes: Option<usize> = bytes
+                    .map(|bytes| {
+                        bytes
+                            .try_into()
+                            .map_err(|_| Error::oos("Avro block size negative or too large"))
+                    })
+                    .transpose()?;
 
                 if len == 0 {
                     break;
                 }
 
-                for _ in 0..len {
-                    block = skip_item(inner, avro_inner, block)?;
+                if let Some(bytes) = bytes {
+                    block = &block[bytes..];
+                } else {
+                    for _ in 0..len {
+                        block = skip_item(inner, avro_inner, block)?;
+                    }
                 }
             }
         }
@@ -489,7 +523,7 @@ pub fn deserialize(
         arrays
             .iter_mut()
             .zip(projection.iter())
-            .filter_map(|x| if *x.1 { Some(x.0) } else { None })
+            .filter_map(|x| x.1.then(|| x.0))
             .map(|array| array.as_box())
             .collect(),
     )


Test patch:
diff --git a/tests/it/io/avro/read.rs b/tests/it/io/avro/read.rs
index 64ff34c7fe..1deb984a85 100644
--- a/tests/it/io/avro/read.rs
+++ b/tests/it/io/avro/read.rs
@@ -311,3 +311,76 @@ fn test_projected() -> Result<()> {
     }
     Ok(())
 }
+
+fn schema_list() -> (AvroSchema, Schema) {
+    let raw_schema = r#"
+    {
+        "type": "record",
+        "name": "test",
+        "fields": [
+            {"name": "h", "type": {
+                "type": "array",
+                "items": {
+                    "name": "item",
+                    "type": "int"
+                }
+            }}
+        ]
+    }
+"#;
+
+    let schema = Schema::from(vec![Field::new(
+        "h",
+        DataType::List(Box::new(Field::new("item", DataType::Int32, false))),
+        false,
+    )]);
+
+    (AvroSchema::parse_str(raw_schema).unwrap(), schema)
+}
+
+pub(super) fn data_list() -> Chunk<Box<dyn Array>> {
+    let data = [Some(vec![Some(1i32), Some(2), Some(3)]), Some(vec![])];
+
+    let mut array = MutableListArray::<i32, MutablePrimitiveArray<i32>>::new_from(
+        Default::default(),
+        DataType::List(Box::new(Field::new("item", DataType::Int32, false))),
+        0,
+    );
+    array.try_extend(data).unwrap();
+
+    let columns = vec![array.into_box()];
+
+    Chunk::try_new(columns).unwrap()
+}
+
+pub(super) fn write_list(codec: Codec) -> std::result::Result<Vec<u8>, avro_rs::Error> {
+    let (avro, _) = schema_list();
+    // a writer needs a schema and something to write to
+    let mut writer = Writer::with_codec(&avro, Vec::new(), codec);
+
+    // the Record type models our Record schema
+    let mut record = Record::new(writer.schema()).unwrap();
+    record.put(
+        "h",
+        Value::Array(vec![Value::Int(1), Value::Int(2), Value::Int(3)]),
+    );
+    writer.append(record)?;
+
+    let mut record = Record::new(writer.schema()).unwrap();
+    record.put("h", Value::Array(vec![]));
+    writer.append(record)?;
+    Ok(writer.into_inner().unwrap())
+}
+
+#[test]
+fn test_list() -> Result<()> {
+    let avro = write_list(Codec::Null).unwrap();
+    let expected = data_list();
+    let (_, expected_schema) = schema_list();
+
+    let (result, schema) = read_avro(&avro, None)?;
+
+    assert_eq!(schema, expected_schema);
+    assert_eq!(result, expected);
+    Ok(())
+}