feat: add support for serializing and desrializing array type to Prot…

…obuf

pkg/messaging/protobuf.go

@@ -10,26 +10,41 @@ import (
 )
 
 var ErrConvertingToProtoStruct = "Failed converting to Proto struct, unsupported ColumnType: '%v'"
-var ErrConvertingToValueMap = "Failed converting to ValueMap, unsupported ColumnType: '%v'"
+var ErrConvertingToValueMap = "Failed converting to ValueMap, unsupported Value kind: '%v'"
+
+// Helper function to convert ColumnValue to *structpb.Value
+func columnValueToProtoValue(v types.ColumnValue) (*structpb.Value, error) {
+	switch v.Type() {
+	case types.IntType:
+		return structpb.NewNumberValue(float64(v.ToInt())), nil
+	case types.StringType:
+		return structpb.NewStringValue(v.ToString()), nil
+	case types.ArrayType:
+		arrValues := v.ToArray()
+		pbListValues := make([]*structpb.Value, len(arrValues))
+		for i, elem := range arrValues {
+			elemPbValue, err := columnValueToProtoValue(elem)
+			if err != nil {
+				return nil, err
+			}
+			pbListValues[i] = elemPbValue
+		}
+		return structpb.NewListValue(&structpb.ListValue{Values: pbListValues}), nil
+	default:
+		panic(fmt.Errorf(ErrConvertingToProtoStruct, v.Type()))
+	}
+}
 
 // ValueMapToProtoStruct converts data of given record to google.protobuf.Struct
 func ValueMapToProtoStruct(data types.ValueMap) (structpb.Struct, error) {
 	fields := make(map[string]*structpb.Value)
 	for k, v := range data {
-		var pbValue *structpb.Value
-
-		switch v.Type() {
-		case types.IntType:
-			pbValue = structpb.NewNumberValue(float64(v.ToInt()))
-		case types.StringType:
-			pbValue = structpb.NewStringValue(v.ToString())
-		default:
-			return structpb.Struct{}, fmt.Errorf(ErrConvertingToProtoStruct, v.Type())
+		pbValue, err := columnValueToProtoValue(v)
+		if err != nil {
+			return structpb.Struct{}, err
 		}
-
 		fields[k] = pbValue
 	}
-
 	return structpb.Struct{Fields: fields}, nil
 }
 
@@ -48,34 +63,41 @@ func ValueMapToProtocolBuffers(data types.ValueMap) ([]byte, error) {
 	return proto.Marshal(recordMessage)
 }
 
+// Helper function to convert *structpb.Value to ColumnValue
+func protoValueToColumnValue(v *structpb.Value) types.ColumnValue {
+	switch kind := v.GetKind().(type) {
+	case *structpb.Value_NumberValue:
+		return types.Integer{Val: int(kind.NumberValue)}
+	case *structpb.Value_StringValue:
+		return types.String{Val: kind.StringValue}
+	case *structpb.Value_ListValue:
+		pbListValues := kind.ListValue.GetValues()
+		arrValues := make([]types.ColumnValue, len(pbListValues))
+		for i, pbValue := range pbListValues {
+			arrValues[i] = protoValueToColumnValue(pbValue)
+		}
+		return types.Array{Val: arrValues}
+	default:
+		panic(fmt.Errorf(ErrConvertingToValueMap, v.GetKind()))
+	}
+}
+
 // ProtoStructToValueMap converts google.protobuf.Struct to a ValueMap
-// This is opposite of ValueMapToProtoStruct
-func ProtoStructToValueMap(protoStruct *structpb.Struct) (types.ValueMap, error) {
+func ProtoStructToValueMap(protoStruct *structpb.Struct) types.ValueMap {
 	data := types.ValueMap{}
 	for k, v := range protoStruct.Fields {
-		switch v.GetKind().(type) {
-		case *structpb.Value_NumberValue:
-			// FIXME: identify the type correctly
-			data[k] = types.Integer{Val: int(v.GetNumberValue())}
-		case *structpb.Value_StringValue:
-			data[k] = types.String{Val: v.GetStringValue()}
-		default:
-			return types.ValueMap{}, fmt.Errorf(
-				ErrConvertingToValueMap,
-				v.GetKind(),
-			)
-		}
+		columnValue := protoValueToColumnValue(v)
+		data[k] = columnValue
 	}
-
-	return data, nil
+	return data
 }
 
 // ProtocolBuffersToValueMap unmarshals protocol buffers data to a ValueMap
-func ProtocolBuffersToValueMap(data []byte) (types.ValueMap, error) {
+func ProtocolBuffersToValueMap(data []byte) types.ValueMap {
 	recordMessage := RecordData{}
 	err := proto.Unmarshal(data, &recordMessage)
 	if err != nil {
-		return nil, err
+		panic(err)
 	}
 
 	return ProtoStructToValueMap(recordMessage.Data.GetStructValue())
@@ -117,10 +139,7 @@ func ProtocolBuffersToRecord(data []byte) (types.Record, error) {
 	}
 
 	// Convert Data field
-	valueMap, err := ProtoStructToValueMap(recordProto.Data.Data.GetStructValue())
-	if err != nil {
-		return types.Record{}, fmt.Errorf("error converting Data field: %w", err)
-	}
+	valueMap := ProtoStructToValueMap(recordProto.Data.Data.GetStructValue())
 
 	// Convert Metadata field
 	metadata := types.Metadata{
@@ -137,3 +156,6 @@ func ProtocolBuffersToRecord(data []byte) (types.Record, error) {
 
 	return record, nil
 }
+
+// FIXME: Add an issue to change IntegerValue to NumberValue and change the underlying data type to float64
+// FIXME: Add an optimization issue to verify how serialization/deserialization to and from Protobuf is now bad with recursive nature.