Fix Parquet Reader's Arrow Schema Inference

parquet/src/arrow/array_reader/list_array.rs

@@ -584,16 +584,25 @@ mod tests {
 
         let mut array_reader = build_array_reader(
             file_reader.metadata().file_metadata().schema_descr_ptr(),
-            Arc::new(arrow_schema.clone()),
+            Arc::new(arrow_schema),
             vec![0usize].into_iter(),
             Box::new(file_reader),
         )
         .unwrap();
 
         let batch = array_reader.next_batch(100).unwrap();
+        assert_eq!(batch.data_type(), array_reader.get_data_type());
         assert_eq!(
             batch.data_type(),
-            &ArrowType::Struct(arrow_schema.fields().clone())
+            &ArrowType::Struct(vec![Field::new(
+                "table_info",
+                ArrowType::List(Box::new(Field::new(
+                    "table_info",
+                    ArrowType::Struct(vec![Field::new("name", ArrowType::Binary, false)]),
+                    false
+                ))),
+                false
+            )])
         );
         assert_eq!(batch.len(), 0);
     }

parquet/src/arrow/arrow_reader.rs

@@ -1050,6 +1050,41 @@ mod tests {
         for batch in record_batch_reader {
             batch.unwrap();
         }
+
+        let projected_reader = arrow_reader
+            .get_record_reader_by_columns(vec![3, 8, 10], 60)
+            .unwrap();
+        let projected_schema = arrow_reader
+            .get_schema_by_columns(vec![3, 8, 10], true)
+            .unwrap();
+
+        let expected_schema = Schema::new(vec![
+            Field::new(
+                "roll_num",
+                ArrowDataType::Struct(vec![Field::new(
+                    "count",
+                    ArrowDataType::UInt64,
+                    false,
+                )]),
+                false,
+            ),
+            Field::new(
+                "PC_CUR",
+                ArrowDataType::Struct(vec![
+                    Field::new("mean", ArrowDataType::Int64, false),
+                    Field::new("sum", ArrowDataType::Int64, false),
+                ]),
+                false,
+            ),
+        ]);
+
+        assert_eq!(projected_reader.schema().as_ref(), &projected_schema);
+        assert_eq!(expected_schema, projected_schema);
+
+        for batch in projected_reader {
+            let batch = batch.unwrap();
+            assert_eq!(batch.schema().as_ref(), &projected_schema);
+        }
     }
 
     #[test]

parquet/src/arrow/arrow_writer.rs

@@ -1058,7 +1058,7 @@ mod tests {
         let stocks_field = Field::new(
             "stocks",
             DataType::Map(
-                Box::new(Field::new("entries", entries_struct_type, false)),
+                Box::new(Field::new("entries", entries_struct_type, true)),
                 false,
             ),
             true,