fix: handle Avro reader schema with no fields

In the degenerate case when the Avro reader schema has no fields,
the RecordDecoder should be able to produce empty record batches with
the number of rows counted from the data.

Author: mzabaluev

arrow-avro/src/reader/async_reader/mod.rs

@@ -1532,6 +1532,38 @@ mod tests {
         assert_eq!(id.values(), &[4, 5, 6, 7, 2, 3, 0, 1]);
     }
 
+    #[tokio::test]
+    async fn test_alltypes_with_empty_schema_large_batch() {
+        // With an empty reader schema -- should count rows but produce no columns
+        let file = arrow_test_data("avro/alltypes_plain.avro");
+        let schema = Arc::new(Schema::new(Vec::<Field>::new()));
+        let batches = read_async_file(&file, 1024, None, Some(schema), None)
+            .await
+            .unwrap();
+        assert_eq!(batches.len(), 1);
+        let batch = &batches[0];
+
+        assert_eq!(batch.num_rows(), 8);
+        assert_eq!(batch.num_columns(), 0);
+    }
+
+    #[tokio::test]
+    async fn test_alltypes_with_empty_schema_small_batch() {
+        // With an empty reader schema -- should count rows but produce no columns
+        let file = arrow_test_data("avro/alltypes_plain.avro");
+        let schema = Arc::new(Schema::new(Vec::<Field>::new()));
+        let batches = read_async_file(&file, 5, None, Some(schema), None)
+            .await
+            .unwrap();
+
+        assert_eq!(batches.len(), 2);
+
+        assert_eq!(batches[0].num_rows(), 5);
+        assert_eq!(batches[0].num_columns(), 0);
+        assert_eq!(batches[1].num_rows(), 3);
+        assert_eq!(batches[1].num_columns(), 0);
+    }
+
     #[tokio::test]
     async fn test_nested_no_schema_no_projection() {
         // No reader schema, no projection
@@ -1597,6 +1629,31 @@ mod tests {
         assert_eq!(batch.schema().field(2).name(), "f1");
     }
 
+    #[tokio::test]
+    async fn test_nested_with_empty_schema() {
+        // With an empty reader schema -- should count rows but produce no columns
+        let file = arrow_test_data("avro/nested_records.avro");
+        let schema = Arc::new(
+            Schema::new(Vec::<Field>::new()).with_metadata(HashMap::from([(
+                SCHEMA_METADATA_KEY.into(),
+                r#"{
+                    "type": "record",
+                    "namespace": "ns1",
+                    "name": "record1",
+                    "fields": []
+                }"#
+                .to_owned(),
+            )])),
+        );
+        let batches = read_async_file(&file, 1024, None, Some(schema), None)
+            .await
+            .unwrap();
+        let batch = &batches[0];
+
+        assert_eq!(batch.num_rows(), 2);
+        assert_eq!(batch.num_columns(), 0);
+    }
+
     #[tokio::test]
     async fn test_projection_error_out_of_bounds() {
         let file = arrow_test_data("avro/alltypes_plain.avro");

arrow-avro/src/reader/record.rs

@@ -94,6 +94,7 @@ pub(crate) struct RecordDecoder {
     schema: SchemaRef,
     fields: Vec<Decoder>,
     projector: Option<Projector>,
+    row_count: usize,
 }
 
 impl RecordDecoder {
@@ -136,6 +137,7 @@ impl RecordDecoder {
                     schema: Arc::new(ArrowSchema::new(arrow_fields)),
                     fields: encodings,
                     projector,
+                    row_count: 0,
                 })
             }
             other => Err(AvroError::ParseError(format!(
@@ -166,6 +168,7 @@ impl RecordDecoder {
                 }
             }
         }
+        self.row_count += count;
         Ok(cursor.position())
     }
 
@@ -176,7 +179,10 @@ impl RecordDecoder {
             .iter_mut()
             .map(|x| x.flush(None))
             .collect::<Result<Vec<_>, _>>()?;
-        RecordBatch::try_new(self.schema.clone(), arrays).map_err(Into::into)
+        let batch_options = RecordBatchOptions::new().with_row_count(Some(self.row_count));
+        self.row_count = 0;
+        RecordBatch::try_new_with_options(self.schema.clone(), arrays, &batch_options)
+            .map_err(Into::into)
     }
 }