[wip] Binary IDL With MessagePack Bytes

flyteidl/protos/flyteidl/core/literals.proto

@@ -42,8 +42,8 @@ message BlobMetadata {
 // A simple byte array with a tag to help different parts of the system communicate about what is in the byte array.
 // It's strongly advisable that consumers of this type define a unique tag and validate the tag before parsing the data.
 message Binary {
-    bytes value = 1;
-    string tag = 2;
+    bytes value = 1; // Serialized data (MessagePack) for supported types like Dataclass, Pydantic BaseModel, and dict.
+    string tag = 2; // The serialization format identifier (e.g., MessagePack). Consumers must define unique tags and validate them before deserialization.
 }
 
 // A strongly typed schema that defines the interface of data retrieved from the underlying storage medium.

flytepropeller/go.mod

@@ -21,6 +21,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.16.0
+	github.com/shamaton/msgpack/v2 v2.2.2
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.7.0
 	github.com/spf13/pflag v1.0.5

flytepropeller/go.sum

@@ -372,6 +372,8 @@ github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFR
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=
 github.com/rogpeppe/go-internal v1.12.0/go.mod h1:E+RYuTGaKKdloAfM02xzb0FW3Paa99yedzYV+kq4uf4=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/shamaton/msgpack/v2 v2.2.2 h1:GOIg0c9LV04VwzOOqZSrmsv/JzjNOOMxnS/HvOHGdgs=
+github.com/shamaton/msgpack/v2 v2.2.2/go.mod h1:6khjYnkx73f7VQU7wjcFS9DFjs+59naVWJv1TB7qdOI=
 github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
 github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=

flytepropeller/pkg/compiler/validators/utils.go

@@ -44,7 +44,11 @@
 
 		literalType = &core.LiteralType{Type: &core.LiteralType_Blob{Blob: scalar.GetBlob().GetMetadata().GetType()}}
 	case *core.Scalar_Binary:
-		literalType = &core.LiteralType{Type: &core.LiteralType_Simple{Simple: core.SimpleType_BINARY}}
+		if len(v.Binary.Tag) > 0 {
+			literalType = &core.LiteralType{Type: &core.LiteralType_Simple{Simple: core.SimpleType_STRUCT}}
+		} else {
+			literalType = &core.LiteralType{Type: &core.LiteralType_Simple{Simple: core.SimpleType_BINARY}}
+		}
 	case *core.Scalar_Schema:
 		literalType = &core.LiteralType{
 			Type: &core.LiteralType_Schema{

flytepropeller/pkg/controller/nodes/attr_path_resolver.go

@@ -1,6 +1,7 @@
 package nodes
 
 import (
+	"github.com/shamaton/msgpack/v2"
 	"google.golang.org/protobuf/types/known/structpb"
 
 	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/core"
@@ -37,13 +38,20 @@
 		}
 	}
 
-	// resolve dataclass
-	if currVal.GetScalar() != nil && currVal.GetScalar().GetGeneric() != nil {
-		st := currVal.GetScalar().GetGeneric()
-		// start from index "count"
-		currVal, err = resolveAttrPathInPbStruct(nodeID, st, bindAttrPath[count:])
-		if err != nil {
-			return nil, err
+	// resolve dataclass and Pydantic BaseModel
+	if scalar := currVal.GetScalar(); scalar != nil {
+		if binary := scalar.GetBinary(); binary != nil {
+			// Start from index "count"
+			currVal, err = resolveAttrPathInBinary(nodeID, binary, bindAttrPath[count:])
+			if err != nil {
+				return nil, err
+			}
+		} else if generic := scalar.GetGeneric(); generic != nil {
+			// Start from index "count"
+			currVal, err = resolveAttrPathInPbStruct(nodeID, generic, bindAttrPath[count:])
+			if err != nil {
+				return nil, err
+			}
 		}
 	}
 
@@ -84,6 +92,79 @@
 	return literal, err
 }
 
+// resolveAttrPathInBinary resolves the protobuf binary (e.g. dataclass, pydantic basemodel) with attribute path
+func resolveAttrPathInBinary(nodeID string, binaryIDL *core.Binary, bindAttrPath []*core.PromiseAttribute) (*core.
+	Literal,
+	error) {
+
+	binaryBytes := binaryIDL.GetValue()
+	serializationFormat := binaryIDL.GetTag()
+
+	var currVal interface{}
+	var tmpVal interface{}
+	var exist bool
+
+	if serializationFormat == "msgpack" {
+		err := msgpack.Unmarshal(binaryBytes, &currVal)
+		if err != nil {
+			return nil, err
+
+		}
+	} else {
+		return nil, errors.Errorf(errors.PromiseAttributeResolveError, nodeID,
+			"Unsupported format '%v' found for literal value.\n"+
+				"Please ensure the serialization format is supported.", serializationFormat)
+	}
+
+	// Turn the current value to a map, so it can be resolved more easily
+	for _, attr := range bindAttrPath {
+		switch resolvedVal := currVal.(type) {
+		// map
+		case map[interface{}]interface{}:
+			tmpVal, exist = resolvedVal[attr.GetStringValue()]
+			if !exist {
+				return nil, errors.Errorf(errors.PromiseAttributeResolveError, nodeID, "key [%v] does not exist in literal %v", attr.GetStringValue(), currVal)
+			}
+			currVal = tmpVal
+		// list
+		case []interface{}:
+			if int(attr.GetIntValue()) >= len(resolvedVal) {
+				return nil, errors.Errorf(errors.PromiseAttributeResolveError, nodeID, "index [%v] is out of range of %v", attr.GetIntValue(), currVal)
+			}
+			currVal = resolvedVal[attr.GetIntValue()]
+		}
+	}
+
+	if serializationFormat == "msgpack" {
+		// Marshal the current value to MessagePack bytes
+		resolvedBinaryBytes, err := msgpack.Marshal(currVal)
+		if err != nil {
+			return nil, err
+		}
+		// Construct and return the binary-encoded literal
+		return constructResolvedBinary(resolvedBinaryBytes, serializationFormat), nil
+	}
+	// Unsupported serialization format
+	return nil, errors.Errorf(errors.PromiseAttributeResolveError, nodeID,
+		"Unsupported format '%v' found for literal value.\n"+
+			"Please ensure the serialization format is supported.", serializationFormat)
+}
+
+func constructResolvedBinary(resolvedBinaryBytes []byte, serializationFormat string) *core.Literal {
+	return &core.Literal{
+		Value: &core.Literal_Scalar{
+			Scalar: &core.Scalar{
+				Value: &core.Scalar_Binary{
+					Binary: &core.Binary{
+						Value: resolvedBinaryBytes,
+						Tag:   serializationFormat,
+					},
+				},
+			},
+		},
+	}
+}
+
 // convertInterfaceToLiteral converts the protobuf struct (e.g. dataclass) to literal
 func convertInterfaceToLiteral(nodeID string, obj interface{}) (*core.Literal, error) {
 

flytepropeller/pkg/controller/nodes/branch/evaluator.go

@@ -17,6 +17,8 @@ const ErrorCodeMalformedBranch = "MalformedBranchUserError"
 const ErrorCodeCompilerError = "CompilerError"
 const ErrorCodeFailedFetchOutputs = "FailedFetchOutputs"
 
+// TODO: IF THERE'S BINARY, CONVERT IT TO PRIMITIVE.
+// NEED ERROR HANDLING FOR NON-PRIMITIVE TYPES
 func EvaluateComparison(expr *core.ComparisonExpression, nodeInputs *core.LiteralMap) (bool, error) {
 	var lValue *core.Literal
 	var rValue *core.Literal

go.mod

@@ -165,6 +165,7 @@ require (
 	github.com/robfig/cron/v3 v3.0.0 // indirect
 	github.com/sendgrid/rest v2.6.9+incompatible // indirect
 	github.com/sendgrid/sendgrid-go v3.10.0+incompatible // indirect
+	github.com/shamaton/msgpack/v2 v2.2.2 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/spf13/afero v1.9.2 // indirect
 	github.com/spf13/cast v1.4.1 // indirect