[flytepropeller][flyteadmin] Compiler unknown literal type error hand…

…ling (#5651)

* [propeller][admin] Compiler literal type error handling

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* support all err msg

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* add execution_validator tests

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* add launch_plan_validator tests

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* add tests for signal_validator

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* change var

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* add tests for compiler_errors

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* support nodes/array/handler.go

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* fix imports

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* add tests for k8s/inputs and catalog/datacatalog/transformer

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* kevin's change

Signed-off-by: Kevin Su <pingsutw@apache.org>

* update all pingsu's interface

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* Trigger Build

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* update pingsu's advice

Signed-off-by: Future-Outlier <eric901201@gmail.com>
Co-authored-by: pingsutw  <pingsutw@apache.org>

* nit

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* use InvalidLiteralTypeErr instead of NewIDLTypeNotFoundErr

Signed-off-by: Future-Outlier <eric901201@gmail.com>

* nit

Signed-off-by: Kevin Su <pingsutw@apache.org>

---------

Signed-off-by: Future-Outlier <eric901201@gmail.com>
Signed-off-by: Kevin Su <pingsutw@apache.org>
Co-authored-by: Kevin Su <pingsutw@apache.org>

flyteadmin/pkg/errors/errors.go

@@ -213,3 +213,8 @@ func NewInactiveProjectError(ctx context.Context, id string) FlyteAdminError {
 	}
 	return statusErr
 }
+
+func NewInvalidLiteralTypeError(name string, err error) FlyteAdminError {
+	return NewFlyteAdminErrorf(codes.InvalidArgument,
+		fmt.Sprintf("Failed to validate literal type for [%s] with err: %s", name, err))
+}

flyteadmin/pkg/manager/impl/validation/execution_validator.go

@@ -107,6 +107,10 @@ func CheckAndFetchInputsForExecution(
 			default:
 				inputType = validators.LiteralTypeForLiteral(executionInputMap[name])
 			}
+			err := validators.ValidateLiteralType(inputType)
+			if err != nil {
+				return nil, errors.NewInvalidLiteralTypeError(name, err)
+			}
 			if !validators.AreTypesCastable(inputType, expectedInput.GetVar().GetType()) {
 				return nil, errors.NewFlyteAdminErrorf(codes.InvalidArgument, "invalid %s input wrong type. Expected %s, but got %s", name, expectedInput.GetVar().GetType(), inputType)
 			}

flyteadmin/pkg/manager/impl/validation/execution_validator_test.go

@@ -17,6 +17,8 @@ import (
 
 var execConfig = testutils.GetApplicationConfigWithDefaultDomains()
 
+const failedToValidateLiteralType = "Failed to validate literal type"
+
 func TestValidateExecEmptyProject(t *testing.T) {
 	request := testutils.GetExecutionRequest()
 	request.Project = ""
@@ -209,6 +211,42 @@ func TestValidateExecEmptyInputs(t *testing.T) {
 	assert.EqualValues(t, expectedMap, actualInputs)
 }
 
+func TestValidateExecUnknownIDLInputs(t *testing.T) {
+	unsupportedLiteral := &core.Literal{
+		Value: &core.Literal_Scalar{
+			Scalar: &core.Scalar{},
+		},
+	}
+	defaultInputs := &core.ParameterMap{
+		Parameters: map[string]*core.Parameter{
+			"foo": {
+				Var: &core.Variable{
+					// 1000 means an unsupported type
+					Type: &core.LiteralType{Type: &core.LiteralType_Simple{Simple: 1000}},
+				},
+				Behavior: &core.Parameter_Default{
+					Default: unsupportedLiteral,
+				},
+			},
+		},
+	}
+	userInputs := &core.LiteralMap{
+		Literals: map[string]*core.Literal{
+			"foo": unsupportedLiteral, // This will lead to a nil inputType
+		},
+	}
+
+	_, err := CheckAndFetchInputsForExecution(
+		userInputs,
+		nil,
+		defaultInputs,
+	)
+	assert.NotNil(t, err)
+
+	// Expected error message
+	assert.Contains(t, err.Error(), failedToValidateLiteralType)
+}
+
 func TestValidExecutionId(t *testing.T) {
 	err := CheckValidExecutionID("abcde123", "a")
 	assert.Nil(t, err)

flyteadmin/pkg/manager/impl/validation/launch_plan_validator.go

@@ -143,6 +143,10 @@ func checkAndFetchExpectedInputForLaunchPlan(
 			return nil, errors.NewFlyteAdminErrorf(codes.InvalidArgument, "unexpected fixed_input %s", name)
 		}
 		inputType := validators.LiteralTypeForLiteral(fixedInput)
+		err := validators.ValidateLiteralType(inputType)
+		if err != nil {
+			return nil, errors.NewInvalidLiteralTypeError(name, err)
+		}
 		if !validators.AreTypesCastable(inputType, value.GetType()) {
 			return nil, errors.NewFlyteAdminErrorf(codes.InvalidArgument,
 				"invalid fixed_input wrong type %s, expected %v, got %v instead", name, value.GetType(), inputType)

flyteadmin/pkg/manager/impl/validation/launch_plan_validator_test.go

@@ -231,6 +231,50 @@ func TestGetLpExpectedInvalidFixedInput(t *testing.T) {
 	assert.Nil(t, actualMap)
 }
 
+func TestGetLpExpectedInvalidFixedInputWithUnknownIDL(t *testing.T) {
+	unsupportedLiteral := &core.Literal{
+		Value: &core.Literal_Scalar{
+			Scalar: &core.Scalar{},
+		},
+	}
+	workflowVariableMap := &core.VariableMap{
+		Variables: map[string]*core.Variable{
+			"foo": {
+				Type: &core.LiteralType{Type: &core.LiteralType_Simple{Simple: 1000}},
+			},
+		},
+	}
+	defaultInputs := &core.ParameterMap{
+		Parameters: map[string]*core.Parameter{
+			"foo": {
+				Var: &core.Variable{
+					// 1000 means an unsupported type
+					Type: &core.LiteralType{Type: &core.LiteralType_Simple{Simple: 1000}},
+				},
+				Behavior: &core.Parameter_Default{
+					Default: unsupportedLiteral,
+				},
+			},
+		},
+	}
+	fixedInputs := &core.LiteralMap{
+		Literals: map[string]*core.Literal{
+			"foo": unsupportedLiteral, // This will lead to a nil inputType
+		},
+	}
+
+	_, err := checkAndFetchExpectedInputForLaunchPlan(
+		workflowVariableMap,
+		fixedInputs,
+		defaultInputs,
+	)
+
+	assert.NotNil(t, err)
+
+	// Expected error message
+	assert.Contains(t, err.Error(), failedToValidateLiteralType)
+}
+
 func TestGetLpExpectedNoFixedInput(t *testing.T) {
 	request := testutils.GetLaunchPlanRequest()
 	actualMap, err := checkAndFetchExpectedInputForLaunchPlan(

flyteadmin/pkg/manager/impl/validation/signal_validator.go

@@ -76,6 +76,10 @@ func ValidateSignalSetRequest(ctx context.Context, db repositoryInterfaces.Repos
 	if err != nil {
 		return err
 	}
+	err = propellervalidators.ValidateLiteralType(valueType)
+	if err != nil {
+		return errors.NewInvalidLiteralTypeError("", err)
+	}
 	if !propellervalidators.AreTypesCastable(lookupSignal.Type, valueType) {
 		return errors.NewFlyteAdminErrorf(codes.InvalidArgument,
 			"requested signal value [%v] is not castable to existing signal type [%v]",

flyteadmin/pkg/manager/impl/validation/signal_validator_test.go

@@ -283,4 +283,52 @@ func TestValidateSignalUpdateRequest(t *testing.T) {
 		utils.AssertEqualWithSanitizedRegex(t,
 			"requested signal value [scalar:{ primitive:{ boolean:false } } ] is not castable to existing signal type [[8 1]]", ValidateSignalSetRequest(ctx, repo, request).Error())
 	})
+
+	t.Run("UnknownIDLType", func(t *testing.T) {
+		ctx := context.TODO()
+
+		// Define an unsupported literal type with a simple type of 1000
+		unsupportedLiteralType := &core.LiteralType{
+			Type: &core.LiteralType_Simple{
+				Simple: 1000, // Using 1000 as an unsupported type
+			},
+		}
+		unsupportedLiteralTypeBytes, _ := proto.Marshal(unsupportedLiteralType)
+
+		// Mock the repository to return a signal with this unsupported type
+		repo := repositoryMocks.NewMockRepository()
+		repo.SignalRepo().(*repositoryMocks.SignalRepoInterface).
+			OnGetMatch(mock.Anything, mock.Anything).Return(
+			models.Signal{
+				Type: unsupportedLiteralTypeBytes, // Set the unsupported type
+			},
+			nil,
+		)
+
+		// Set up the unsupported literal that will trigger the nil valueType condition
+		unsupportedLiteral := &core.Literal{
+			Value: &core.Literal_Scalar{
+				Scalar: &core.Scalar{},
+			},
+		}
+
+		request := admin.SignalSetRequest{
+			Id: &core.SignalIdentifier{
+				ExecutionId: &core.WorkflowExecutionIdentifier{
+					Project: "project",
+					Domain:  "domain",
+					Name:    "name",
+				},
+				SignalId: "signal",
+			},
+			Value: unsupportedLiteral, // This will lead to valueType being nil
+		}
+
+		// Invoke the function and check for the expected error
+		err := ValidateSignalSetRequest(ctx, repo, &request)
+		assert.NotNil(t, err)
+
+		// Expected error message
+		assert.Contains(t, err.Error(), failedToValidateLiteralType)
+	})
 }

flyteadmin/pkg/manager/impl/validation/validation.go

@@ -1,7 +1,6 @@
 package validation
 
 import (
-	"fmt"
 	"net/url"
 	"strconv"
 	"strings"
@@ -283,19 +282,9 @@ func validateParameterMap(inputMap *core.ParameterMap, fieldName string) error {
 			defaultValue := defaultInput.GetDefault()
 			if defaultValue != nil {
 				inputType := validators.LiteralTypeForLiteral(defaultValue)
-
-				if inputType == nil {
-					return errors.NewFlyteAdminErrorf(codes.InvalidArgument,
-						fmt.Sprintf(
-							"Flyte encountered an issue while determining\n"+
-								"the type of the default value for Parameter '%s' in '%s'.\n"+
-								"Registered type: [%s].\n"+
-								"Flyte needs to support the latest FlyteIDL to support this type.\n"+
-								"Suggested solution: Please update all of your Flyte images to the latest version and "+
-								"try again.",
-							name, fieldName, defaultInput.GetVar().GetType().String(),
-						),
-					)
+				err := validators.ValidateLiteralType(inputType)
+				if err != nil {
+					return errors.NewInvalidLiteralTypeError(name, err)
 				}
 
 				if !validators.AreTypesCastable(inputType, defaultInput.GetVar().GetType()) {

flyteadmin/pkg/manager/impl/validation/validation_test.go

@@ -347,16 +347,7 @@ func TestValidateParameterMap(t *testing.T) {
 		err := validateParameterMap(&exampleMap, fieldName)
 		assert.Error(t, err)
 		fmt.Println(err.Error())
-		expectedErrMsg := fmt.Sprintf(
-			"Flyte encountered an issue while determining\n"+
-				"the type of the default value for Parameter '%s' in '%s'.\n"+
-				"Registered type: [%s].\n"+
-				"Flyte needs to support the latest FlyteIDL to support this type.\n"+
-				"Suggested solution: Please update all of your Flyte images to the latest version and "+
-				"try again.",
-			name, fieldName, exampleMap.Parameters[name].GetVar().GetType().String(),
-		)
-		assert.Equal(t, expectedErrMsg, err.Error())
+		assert.Contains(t, err.Error(), failedToValidateLiteralType)
 	})
 }
 

flytepropeller/pkg/compiler/errors/compiler_error_test.go

@@ -33,6 +33,7 @@ func TestErrorCodes(t *testing.T) {
 		UnrecognizedValue:          NewUnrecognizedValueErr("", ""),
 		WorkflowBuildError:         NewWorkflowBuildError(errors.New("")),
 		NoNodesFound:               NewNoNodesFoundErr(""),
+		InvalidLiteralTypeError:    NewInvalidLiteralTypeErr("", "", errors.New("")),
 	}
 
 	for key, value := range testCases {
@@ -48,6 +49,6 @@ func TestIncludeSource(t *testing.T) {
 
 	SetConfig(Config{IncludeSource: true})
 	e = NewCycleDetectedInWorkflowErr("", "")
-	assert.Equal(t, e.source, "compiler_error_test.go:50")
+	assert.Equal(t, e.source, "compiler_error_test.go:51")
 	SetConfig(Config{})
 }

flytepropeller/pkg/compiler/errors/compiler_errors.go

@@ -96,6 +96,9 @@ const (
 
 	// Field not found in the dataclass
 	FieldNotFoundError ErrorCode = "FieldNotFound"
+
+	// IDL not found when variable binding
+	InvalidLiteralTypeError ErrorCode = "InvalidLiteralType"
 )
 
 func NewBranchNodeNotSpecified(branchNodeID string) *CompileError {
@@ -218,6 +221,14 @@ func NewMismatchingVariablesErr(nodeID, fromVar, fromType, toVar, toType string)
 	)
 }
 
+func NewInvalidLiteralTypeErr(nodeID, inputVar string, err error) *CompileError {
+	return newError(
+		InvalidLiteralTypeError,
+		fmt.Sprintf("Failed to validate literal type for [%s] with err: %s", inputVar, err),
+		nodeID,
+	)
+}
+
 func NewMismatchingBindingsErr(nodeID, sinkParam, expectedType, receivedType string) *CompileError {
 	return newError(
 		MismatchingBindings,

flytepropeller/pkg/compiler/transformers/k8s/inputs.go

@@ -42,6 +42,13 @@ func validateInputs(nodeID common.NodeID, iface *core.TypedInterface, inputs cor
 		default:
 			inputType = validators.LiteralTypeForLiteral(inputVal)
 		}
+
+		err := validators.ValidateLiteralType(inputType)
+		if err != nil {
+			errs.Collect(errors.NewInvalidLiteralTypeErr(nodeID, inputVar, err))
+			continue
+		}
+
 		if !validators.AreTypesCastable(inputType, v.Type) {
 			errs.Collect(errors.NewMismatchingTypesErr(nodeID, inputVar, v.Type.String(), inputType.String()))
 			continue

flytepropeller/pkg/compiler/transformers/k8s/inputs_test.go

@@ -1 +1,55 @@
 package k8s
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/flyteorg/flyte/flyteidl/gen/pb-go/flyteidl/core"
+	"github.com/flyteorg/flyte/flytepropeller/pkg/compiler/common"
+	"github.com/flyteorg/flyte/flytepropeller/pkg/compiler/errors"
+)
+
+func TestValidateInputs_InvalidLiteralType(t *testing.T) {
+	nodeID := common.NodeID("test-node")
+
+	iface := &core.TypedInterface{
+		Inputs: &core.VariableMap{
+			Variables: map[string]*core.Variable{
+				"input1": {
+					Type: &core.LiteralType{
+						Type: &core.LiteralType_Simple{
+							Simple: 1000,
+						},
+					},
+				},
+			},
+		},
+	}
+
+	inputs := core.LiteralMap{
+		Literals: map[string]*core.Literal{
+			"input1": nil, // Set this to nil to trigger the nil case
+		},
+	}
+
+	errs := errors.NewCompileErrors()
+	ok := validateInputs(nodeID, iface, inputs, errs)
+
+	assert.False(t, ok)
+	assert.True(t, errs.HasErrors())
+
+	idlNotFound := false
+	var errMsg string
+	for _, err := range errs.Errors().List() {
+		if err.Code() == "InvalidLiteralType" {
+			idlNotFound = true
+			errMsg = err.Error()
+			break
+		}
+	}
+	assert.True(t, idlNotFound, "Expected InvalidLiteralType error was not found in errors")
+
+	expectedContainedErrorMsg := "Failed to validate literal type"
+	assert.Contains(t, errMsg, expectedContainedErrorMsg)
+}

flytepropeller/pkg/compiler/validators/utils.go

@@ -249,6 +249,23 @@ func literalTypeForLiterals(literals []*core.Literal) *core.LiteralType {
 	return buildMultipleTypeUnion(innerType)
 }
 
+// ValidateLiteralType check if the literal type is valid, return error if the literal is invalid.
+func ValidateLiteralType(lt *core.LiteralType) error {
+	if lt == nil {
+		err := fmt.Errorf("got unknown literal type: [%v].\n"+
+			"Suggested solution: Please update all your Flyte deployment images to the latest version and try again", lt)
+		return err
+	}
+	if lt.GetCollectionType() != nil {
+		return ValidateLiteralType(lt.GetCollectionType())
+	}
+	if lt.GetMapValueType() != nil {
+		return ValidateLiteralType(lt.GetMapValueType())
+	}
+
+	return nil
+}
+
 // LiteralTypeForLiteral gets LiteralType for literal, nil if the value of literal is unknown, or type collection/map of
 // type None if the literal is a non-homogeneous type.
 func LiteralTypeForLiteral(l *core.Literal) *core.LiteralType {

flytepropeller/pkg/controller/nodes/array/handler.go

@@ -191,8 +191,15 @@ func (a *arrayNodeHandler) Handle(ctx context.Context, nCtx interfaces.NodeExecu
 		}
 
 		size := -1
-		for _, variable := range literalMap.Literals {
+		for key, variable := range literalMap.Literals {
 			literalType := validators.LiteralTypeForLiteral(variable)
+			err := validators.ValidateLiteralType(literalType)
+			if err != nil {
+				errMsg := fmt.Sprintf("Failed to validate literal type for [%s] with err: %s", key, err)
+				return handler.DoTransition(handler.TransitionTypeEphemeral,
+					handler.PhaseInfoFailure(idlcore.ExecutionError_USER, errors.IDLNotFoundErr, errMsg, nil),
+				), nil
+			}
 			switch literalType.Type.(type) {
 			case *idlcore.LiteralType_CollectionType:
 				collectionLength := len(variable.GetCollection().Literals)