Enable arrow-avro to handle writer-only fields during schema resolution.

- Added skipping logic for writer-only fields in `RecordDecoder`.
- Introduced `ResolvedRuntime` for runtime decoding adjustments.
- Updated tests to validate skipping functionality.
- Refactored block-wise processing for optimized performance.

Author: jecsand838

arrow-avro/src/codec.rs

@@ -955,7 +955,7 @@ impl<'a> Maker<'a> {
         // Prepare outputs
         let mut reader_fields: Vec<AvroField> = Vec::with_capacity(reader_record.fields.len());
         let mut writer_to_reader: Vec<Option<usize>> = vec![None; writer_record.fields.len()];
-        //let mut skip_fields: Vec<Option<AvroDataType>> = vec![None; writer_record.fields.len()];
+        let mut skip_fields: Vec<Option<AvroDataType>> = vec![None; writer_record.fields.len()];
         //let mut default_fields: Vec<usize> = Vec::new();
         // Build reader fields and mapping
         for (reader_idx, r_field) in reader_record.fields.iter().enumerate() {
@@ -975,6 +975,14 @@ impl<'a> Maker<'a> {
                 ));
             }
         }
+        // Any writer fields not mapped should be skipped
+        for (writer_idx, writer_field) in writer_record.fields.iter().enumerate() {
+            if writer_to_reader[writer_idx].is_none() {
+                // Parse writer field type to know how to skip data
+                let w_dt = self.parse_type(&writer_field.r#type, writer_ns)?;
+                skip_fields[writer_idx] = Some(w_dt);
+            }
+        }
         // Implement writer-only fields to skip in Follow-up PR here
         // Build resolved record AvroDataType
         let resolved = AvroDataType::new_with_resolution(
@@ -984,7 +992,7 @@ impl<'a> Maker<'a> {
             Some(ResolutionInfo::Record(ResolvedRecord {
                 writer_to_reader: Arc::from(writer_to_reader),
                 default_fields: Arc::default(),
-                skip_fields: Arc::default(),
+                skip_fields: Arc::from(skip_fields),
             })),
         );
         // Register a resolved record by reader name+namespace for potential named type refs

arrow-avro/src/reader/mod.rs

@@ -863,12 +863,39 @@ mod test {
             .with_reader_schema(reader_schema)
             .build(BufReader::new(file))
             .unwrap();
-
         let schema = reader.schema();
         let batches = reader.collect::<Result<Vec<_>, _>>().unwrap();
         arrow::compute::concat_batches(&schema, &batches).unwrap()
     }
 
+    fn make_reader_schema_with_selected_fields_in_order(
+        path: &str,
+        selected: &[&str],
+    ) -> AvroSchema {
+        let mut root = load_writer_schema_json(path);
+        assert_eq!(root["type"], "record", "writer schema must be a record");
+        let writer_fields = root
+            .get("fields")
+            .and_then(|f| f.as_array())
+            .expect("record has fields");
+        let mut field_map: HashMap<String, Value> = HashMap::with_capacity(writer_fields.len());
+        for f in writer_fields {
+            if let Some(name) = f.get("name").and_then(|n| n.as_str()) {
+                field_map.insert(name.to_string(), f.clone());
+            }
+        }
+        let mut new_fields = Vec::with_capacity(selected.len());
+        for name in selected {
+            let f = field_map
+                .get(*name)
+                .unwrap_or_else(|| panic!("field '{name}' not found in writer schema"))
+                .clone();
+            new_fields.push(f);
+        }
+        root["fields"] = Value::Array(new_fields);
+        AvroSchema::new(root.to_string())
+    }
+
     #[test]
     fn test_alltypes_schema_promotion_mixed() {
         let files = [
@@ -1537,6 +1564,57 @@ mod test {
         assert!(batch.column(0).as_any().is::<StringViewArray>());
     }
 
+    #[test]
+    fn test_alltypes_skip_writer_fields_keep_double_only() {
+        let file = arrow_test_data("avro/alltypes_plain.avro");
+        let reader_schema =
+            make_reader_schema_with_selected_fields_in_order(&file, &["double_col"]);
+        let batch = read_alltypes_with_reader_schema(&file, reader_schema);
+        let expected = RecordBatch::try_from_iter_with_nullable([(
+            "double_col",
+            Arc::new(Float64Array::from_iter_values(
+                (0..8).map(|x| (x % 2) as f64 * 10.1),
+            )) as _,
+            true,
+        )])
+        .unwrap();
+        assert_eq!(batch, expected);
+    }
+
+    #[test]
+    fn test_alltypes_skip_writer_fields_reorder_and_skip_many() {
+        let file = arrow_test_data("avro/alltypes_plain.avro");
+        let reader_schema =
+            make_reader_schema_with_selected_fields_in_order(&file, &["timestamp_col", "id"]);
+        let batch = read_alltypes_with_reader_schema(&file, reader_schema);
+        let expected = RecordBatch::try_from_iter_with_nullable([
+            (
+                "timestamp_col",
+                Arc::new(
+                    TimestampMicrosecondArray::from_iter_values([
+                        1235865600000000, // 2009-03-01T00:00:00.000
+                        1235865660000000, // 2009-03-01T00:01:00.000
+                        1238544000000000, // 2009-04-01T00:00:00.000
+                        1238544060000000, // 2009-04-01T00:01:00.000
+                        1233446400000000, // 2009-02-01T00:00:00.000
+                        1233446460000000, // 2009-02-01T00:01:00.000
+                        1230768000000000, // 2009-01-01T00:00:00.000
+                        1230768060000000, // 2009-01-01T00:01:00.000
+                    ])
+                    .with_timezone("+00:00"),
+                ) as _,
+                true,
+            ),
+            (
+                "id",
+                Arc::new(Int32Array::from(vec![4, 5, 6, 7, 2, 3, 0, 1])) as _,
+                true,
+            ),
+        ])
+        .unwrap();
+        assert_eq!(batch, expected);
+    }
+
     #[test]
     fn test_read_zero_byte_avro_file() {
         let batch = read_file("test/data/zero_byte.avro", 3, false);