Add support for protobuf serialisation of Arrow Map type (#5359)

* Add protobuf serialisation support for Arrow Map type

* Add protobuf serialisation support for Arrow Map type

Author: ahmedriza

datafusion/proto/proto/datafusion.proto

@@ -743,6 +743,11 @@ message Struct{
   repeated Field sub_field_types = 1;
 }
 
+message Map {
+  Field field_type = 1;
+  bool keys_sorted = 2;
+}
+
 enum UnionMode{
   sparse = 0;
   dense = 1;
@@ -894,6 +899,7 @@ message ArrowType{
     Struct STRUCT = 28;
     Union UNION = 29;
     Dictionary DICTIONARY = 30;
+    Map MAP = 33;
   }
 }
 

datafusion/proto/src/generated/pbjson.rs

@@ -351,6 +351,12 @@ impl TryFrom<&protobuf::arrow_type::ArrowTypeEnum> for DataType {
                 let value_datatype = dict.as_ref().value.as_deref().required("value")?;
                 DataType::Dictionary(Box::new(key_datatype), Box::new(value_datatype))
             }
+            arrow_type::ArrowTypeEnum::Map(map) => {
+                let field: Field =
+                    map.as_ref().field_type.as_deref().required("field_type")?;
+                let keys_sorted = map.keys_sorted;
+                DataType::Map(Box::new(field), keys_sorted)
+            }
         })
     }
 }

datafusion/proto/src/generated/prost.rs

@@ -2212,6 +2212,17 @@ mod roundtrip_tests {
                     4,
                 )),
             ),
+            DataType::Map(
+                new_box_field(
+                    "entries",
+                    DataType::Struct(vec![
+                        Field::new("keys", DataType::Utf8, false),
+                        Field::new("values", DataType::Int32, true),
+                    ]),
+                    true,
+                ),
+                false,
+            ),
         ];
 
         for test_case in test_cases.into_iter() {

datafusion/proto/src/logical_plan/from_proto.rs

@@ -218,9 +218,12 @@ impl TryFrom<&DataType> for protobuf::arrow_type::ArrowTypeEnum {
             DataType::Decimal256(_, _) => {
                 return Err(Error::General("Proto serialization error: The Decimal256 data type is not yet supported".to_owned()))
             }
-            DataType::Map(_, _) => {
-                return Err(Error::General(
-                    "Proto serialization error: The Map data type is not yet supported".to_owned()
+            DataType::Map(field, sorted) => {
+                Self::Map(Box::new(
+                    protobuf::Map {
+                        field_type: Some(Box::new(field.as_ref().try_into()?)),
+                        keys_sorted: *sorted,
+                    }
                 ))
             }
             DataType::RunEndEncoded(_, _) => {