[flytepropeller] [compiler] Support non-Any Python types as Any input in workflows

flyteidl/protos/flyteidl/core/types.proto

@@ -18,6 +18,7 @@ enum SimpleType {
     BINARY = 7;
     ERROR = 8;
     STRUCT = 9;
+    ANY = 10;
 }
 
 // Defines schema columns and types to strongly type-validate schemas interoperability.

flytepropeller/pkg/compiler/validators/typing.go

@@ -356,7 +356,19 @@ func getTypeChecker(t *flyte.LiteralType) typeChecker {
 	}
 }
 
-func AreTypesCastable(upstreamType, downstreamType *flyte.LiteralType) bool {
+func isTypeAny(t *flyte.LiteralType) bool {
+	if simpleType, ok := t.GetType().(*flyte.LiteralType_Simple); ok {
+		return simpleType.Simple == flyte.SimpleType_ANY
+	}
+
+	return false
+}
+
+func AreTypesCastable(upstreamType *flyte.LiteralType, downstreamType *flyte.LiteralType) bool {
+	if isTypeAny(downstreamType) {
+		return true
+	}
+
 	typeChecker := getTypeChecker(downstreamType)
 
 	// if upstream is a singular union we check if the downstream type is castable from the union variant

flytepropeller/pkg/compiler/validators/typing_test.go

@@ -22,6 +22,18 @@ func TestSimpleLiteralCasting(t *testing.T) {
 		assert.True(t, castable, "Integers should be castable to other integers")
 	})
 
+	t.Run("IntegerToAny", func(t *testing.T) {
+		castable := AreTypesCastable(
+			&core.LiteralType{
+				Type: &core.LiteralType_Simple{Simple: core.SimpleType_INTEGER},
+			},
+			&core.LiteralType{
+				Type: &core.LiteralType_Simple{Simple: core.SimpleType_ANY},
+			},
+		)
+		assert.True(t, castable, "Integers should be castable to Any")
+	})
+
 	t.Run("IntegerToFloat", func(t *testing.T) {
 		castable := AreTypesCastable(
 			&core.LiteralType{