Make schema validation optional

examples/mnist/resources/transformed_columns.yaml

@@ -1,6 +1,6 @@
 - kind: transformed_column
   name: image_pixels
-  transformer: decode_and_normalize
+  transformer_path: implementations/transformers/decode_and_normalize.py
   inputs:
     columns:
       image: image

examples/reviews/resources/max_length.yaml

@@ -1,13 +1,6 @@
-- kind: aggregator
-  name: max_length
-  inputs:
-    columns:
-      col: STRING_COLUMN
-  output_type: INT
-
 - kind: aggregate
   name: max_review_length
-  aggregator: max_length
+  aggregator_path: implementations/aggregators/max_length.py
   inputs:
     columns:
       col: review

examples/reviews/resources/tokenized_columns.yaml

@@ -1,16 +1,6 @@
-- kind: transformer
-  name: tokenize_string_to_int
-  output_type: INT_LIST_COLUMN
-  inputs:
-    columns:
-      col: STRING_COLUMN
-    args:
-      max_len: INT
-      vocab: {STRING: INT}
-
 - kind: transformed_column
   name: embedding_input
-  transformer: tokenize_string_to_int
+  transformer_path: implementations/transformers/tokenize_string_to_int.py
   inputs:
     columns:
       col: review

pkg/consts/consts.go

@@ -36,21 +36,24 @@ var (
 	RequirementsTxt = "requirements.txt"
 	PackageDir      = "packages"
 
-	AppsDir             = "apps"
-	DataDir             = "data"
-	RawDataDir          = "data_raw"
-	TrainingDataDir     = "data_training"
-	AggregatorsDir      = "aggregators"
-	AggregatesDir       = "aggregates"
-	TransformersDir     = "transformers"
-	ModelImplsDir       = "model_implementations"
-	PythonPackagesDir   = "python_packages"
-	ModelsDir           = "models"
-	ConstantsDir        = "constants"
-	ContextsDir         = "contexts"
-	ResourceStatusesDir = "resource_statuses"
-	WorkloadSpecsDir    = "workload_specs"
-	LogPrefixesDir      = "log_prefixes"
+	AppsDir               = "apps"
+	APIsDir               = "apis"
+	DataDir               = "data"
+	RawDataDir            = "data_raw"
+	TrainingDataDir       = "data_training"
+	AggregatorsDir        = "aggregators"
+	AggregatesDir         = "aggregates"
+	TransformersDir       = "transformers"
+	ModelImplsDir         = "model_implementations"
+	PythonPackagesDir     = "python_packages"
+	ModelsDir             = "models"
+	ConstantsDir          = "constants"
+	ContextsDir           = "contexts"
+	ResourceStatusesDir   = "resource_statuses"
+	WorkloadSpecsDir      = "workload_specs"
+	LogPrefixesDir        = "log_prefixes"
+	RawColumnsDir         = "raw_columns"
+	TransformedColumnsDir = "transformed_columns"
 
 	TelemetryURL = "https://telemetry.cortexlabs.dev"
 )

pkg/lib/configreader/string.go

@@ -28,15 +28,16 @@ import (
 )
 
 type StringValidation struct {
-	Required                      bool
-	Default                       string
-	AllowEmpty                    bool
-	AllowedValues                 []string
-	Prefix                        string
-	AlphaNumericDashDotUnderscore bool
-	AlphaNumericDashUnderscore    bool
-	DNS1035                       bool
-	Validator                     func(string) (string, error)
+	Required                             bool
+	Default                              string
+	AllowEmpty                           bool
+	AllowedValues                        []string
+	Prefix                               string
+	AlphaNumericDashDotUnderscoreOrEmpty bool
+	AlphaNumericDashDotUnderscore        bool
+	AlphaNumericDashUnderscore           bool
+	DNS1035                              bool
+	Validator                            func(string) (string, error)
 }
 
 func EnvVar(envVarName string) string {
@@ -190,6 +191,12 @@ func ValidateStringVal(val string, v *StringValidation) error {
 		}
 	}
 
+	if v.AlphaNumericDashDotUnderscoreOrEmpty {
+		if !regex.CheckAlphaNumericDashDotUnderscore(val) && val != "" {
+			return ErrorAlphaNumericDashDotUnderscore(val)
+		}
+	}
+
 	if v.DNS1035 {
 		if err := urls.CheckDNS1035(val); err != nil {
 			return err

pkg/operator/api/context/context.go

@@ -55,6 +55,7 @@ type Resource interface {
 	GetID() string
 	GetIDWithTags() string
 	GetResourceFields() *ResourceFields
+	GetMetadataKey() string
 }
 
 type ComputedResource interface {
@@ -72,6 +73,7 @@ type ResourceFields struct {
 	ID           string        `json:"id"`
 	IDWithTags   string        `json:"id_with_tags"`
 	ResourceType resource.Type `json:"resource_type"`
+	MetadataKey  string        `json:"metadata_key"`
 }
 
 type ComputedResourceFields struct {
@@ -91,6 +93,10 @@ func (r *ResourceFields) GetResourceFields() *ResourceFields {
 	return r
 }
 
+func (r *ResourceFields) GetMetadataKey() string {
+	return r.MetadataKey
+}
+
 func (r *ComputedResourceFields) GetWorkloadID() string {
 	return r.WorkloadID
 }

pkg/operator/api/context/models.go

@@ -38,12 +38,11 @@ type Model struct {
 }
 
 type TrainingDataset struct {
-	userconfig.ResourceConfigFields
+	userconfig.ResourceFields
 	*ComputedResourceFields
-	ModelName   string `json:"model_name"`
-	TrainKey    string `json:"train_key"`
-	EvalKey     string `json:"eval_key"`
-	MetadataKey string `json:"metadata_key"`
+	ModelName string `json:"model_name"`
+	TrainKey  string `json:"train_key"`
+	EvalKey   string `json:"eval_key"`
 }
 
 func (trainingDataset *TrainingDataset) GetResourceType() resource.Type {

pkg/operator/api/context/python_packages.go

@@ -24,7 +24,7 @@ import (
 type PythonPackages map[string]*PythonPackage
 
 type PythonPackage struct {
-	userconfig.ResourceConfigFields
+	userconfig.ResourceFields
 	*ComputedResourceFields
 	SrcKey     string `json:"src_key"`
 	PackageKey string `json:"package_key"`

pkg/operator/api/userconfig/aggregates.go

@@ -26,11 +26,12 @@ import (
 type Aggregates []*Aggregate
 
 type Aggregate struct {
-	ResourceConfigFields
-	Aggregator string        `json:"aggregator" yaml:"aggregator"`
-	Inputs     *Inputs       `json:"inputs" yaml:"inputs"`
-	Compute    *SparkCompute `json:"compute" yaml:"compute"`
-	Tags       Tags          `json:"tags" yaml:"tags"`
+	ResourceFields
+	Aggregator     string        `json:"aggregator" yaml:"aggregator"`
+	AggregatorPath *string       `json:"aggregator_path" yaml:"aggregator_path"`
+	Inputs         *Inputs       `json:"inputs" yaml:"inputs"`
+	Compute        *SparkCompute `json:"compute" yaml:"compute"`
+	Tags           Tags          `json:"tags" yaml:"tags"`
 }
 
 var aggregateValidation = &configreader.StructValidation{
@@ -45,10 +46,14 @@ var aggregateValidation = &configreader.StructValidation{
 		{
 			StructField: "Aggregator",
 			StringValidation: &configreader.StringValidation{
-				Required:                      true,
-				AlphaNumericDashDotUnderscore: true,
+				AllowEmpty:                           true,
+				AlphaNumericDashDotUnderscoreOrEmpty: true,
 			},
 		},
+		{
+			StructField:         "AggregatorPath",
+			StringPtrValidation: &configreader.StringPtrValidation{},
+		},
 		inputValuesFieldValidation,
 		sparkComputeFieldValidation("Compute"),
 		tagsFieldValidation,

pkg/operator/api/userconfig/aggregators.go

@@ -24,7 +24,7 @@ import (
 type Aggregators []*Aggregator
 
 type Aggregator struct {
-	ResourceConfigFields
+	ResourceFields
 	Inputs     *Inputs     `json:"inputs"  yaml:"inputs"`
 	OutputType interface{} `json:"output_type"  yaml:"output_type"`
 	Path       string      `json:"path"  yaml:"path"`

pkg/operator/api/userconfig/apis.go

@@ -24,7 +24,7 @@ import (
 type APIs []*API
 
 type API struct {
-	ResourceConfigFields
+	ResourceFields
 	ModelName string      `json:"model_name" yaml:"model_name"`
 	Compute   *APICompute `json:"compute" yaml:"compute"`
 	Tags      Tags        `json:"tags" yaml:"tags"`

pkg/operator/api/userconfig/config.go

@@ -200,18 +200,38 @@ func (config *Config) Validate(envName string) error {
 		}
 	}
 
-	// Check local aggregators exist
+	// Check local aggregators exist or a path to one is defined
 	aggregatorNames := config.Aggregators.Names()
 	for _, aggregate := range config.Aggregates {
-		if !strings.Contains(aggregate.Aggregator, ".") && !slices.HasString(aggregatorNames, aggregate.Aggregator) {
+		if aggregate.AggregatorPath == nil && aggregate.Aggregator == "" {
+			return errors.Wrap(ErrorSpecifyOnlyOneMissing("aggregator", "aggregator_path"), Identify(aggregate))
+		}
+
+		if aggregate.AggregatorPath != nil && aggregate.Aggregator != "" {
+			return errors.Wrap(ErrorSpecifyOnlyOne("aggregator", "aggregator_path"), Identify(aggregate))
+		}
+
+		if aggregate.Aggregator != "" &&
+			!strings.Contains(aggregate.Aggregator, ".") &&
+			!slices.HasString(aggregatorNames, aggregate.Aggregator) {
 			return errors.Wrap(ErrorUndefinedResource(aggregate.Aggregator, resource.AggregatorType), Identify(aggregate), AggregatorKey)
 		}
 	}
 
-	// Check local transformers exist
+	// Check local transformers exist or a path to one is defined
 	transformerNames := config.Transformers.Names()
 	for _, transformedColumn := range config.TransformedColumns {
-		if !strings.Contains(transformedColumn.Transformer, ".") && !slices.HasString(transformerNames, transformedColumn.Transformer) {
+		if transformedColumn.TransformerPath == nil && transformedColumn.Transformer == "" {
+			return errors.Wrap(ErrorSpecifyOnlyOneMissing("transformer", "transformer_path"), Identify(transformedColumn))
+		}
+
+		if transformedColumn.TransformerPath != nil && transformedColumn.Transformer != "" {
+			return errors.Wrap(ErrorSpecifyOnlyOne("transformer", "transformer_path"), Identify(transformedColumn))
+		}
+
+		if transformedColumn.Transformer != "" &&
+			!strings.Contains(transformedColumn.Transformer, ".") &&
+			!slices.HasString(transformerNames, transformedColumn.Transformer) {
 			return errors.Wrap(ErrorUndefinedResource(transformedColumn.Transformer, resource.TransformerType), Identify(transformedColumn), TransformerKey)
 		}
 	}

pkg/operator/api/userconfig/constants.go

@@ -25,7 +25,7 @@ import (
 type Constants []*Constant
 
 type Constant struct {
-	ResourceConfigFields
+	ResourceFields
 	Type  interface{} `json:"type" yaml:"type"`
 	Value interface{} `json:"value" yaml:"value"`
 	Tags  Tags        `json:"tags" yaml:"tags"`

pkg/operator/api/userconfig/embed.go

@@ -24,7 +24,7 @@ import (
 type Embeds []*Embed
 
 type Embed struct {
-	ResourceConfigFields
+	ResourceFields
 	Template string                 `json:"template" yaml:"template"`
 	Args     map[string]interface{} `json:"args" yaml:"args"`
 }

pkg/operator/api/userconfig/environments.go

@@ -28,7 +28,7 @@ import (
 type Environments []*Environment
 
 type Environment struct {
-	ResourceConfigFields
+	ResourceFields
 	LogLevel *LogLevel `json:"log_level" yaml:"log_level"`
 	Limit    *Limit    `json:"limit" yaml:"limit"`
 	Data     Data      `json:"-" yaml:"-"`

pkg/operator/api/userconfig/errors.go

@@ -58,6 +58,7 @@ const (
 	ErrK8sQuantityMustBeInt
 	ErrRegressionTargetType
 	ErrClassificationTargetType
+	ErrSpecifyOnlyOneMissing
 )
 
 var errorKinds = []string{
@@ -90,9 +91,10 @@ var errorKinds = []string{
 	"err_k8s_quantity_must_be_int",
 	"err_regression_target_type",
 	"err_classification_target_type",
+	"err_specify_only_one_missing",
 }
 
-var _ = [1]int{}[int(ErrClassificationTargetType)-(len(errorKinds)-1)] // Ensure list length matches
+var _ = [1]int{}[int(ErrSpecifyOnlyOneMissing)-(len(errorKinds)-1)] // Ensure list length matches
 
 func (t ErrorKind) String() string {
 	return errorKinds[t]
@@ -376,9 +378,22 @@ func ErrorRegressionTargetType() error {
 		message: "regression models can only predict float target values",
 	}
 }
+
 func ErrorClassificationTargetType() error {
 	return Error{
 		Kind:    ErrClassificationTargetType,
 		message: "classification models can only predict integer target values (i.e. {0, 1, ..., num_classes-1})",
 	}
 }
+
+func ErrorSpecifyOnlyOneMissing(vals ...string) error {
+	message := fmt.Sprintf("please specify one of %s", s.UserStrsOr(vals))
+	if len(vals) == 2 {
+		message = fmt.Sprintf("please specify either %s or %s", s.UserStr(vals[0]), s.UserStr(vals[1]))
+	}
+
+	return Error{
+		Kind:    ErrSpecifyOnlyOneMissing,
+		message: message,
+	}
+}

pkg/operator/api/userconfig/models.go

@@ -29,7 +29,7 @@ import (
 type Models []*Model
 
 type Model struct {
-	ResourceConfigFields
+	ResourceFields
 	Type               ModelType                `json:"type" yaml:"type"`
 	Path               string                   `json:"path" yaml:"path"`
 	TargetColumn       string                   `json:"target_column" yaml:"target_column"`

pkg/operator/api/userconfig/raw_columns.go

@@ -53,7 +53,7 @@ var rawColumnValidation = &cr.InterfaceStructValidation{
 }
 
 type RawIntColumn struct {
-	ResourceConfigFields
+	ResourceFields
 	Type     ColumnType    `json:"type" yaml:"type"`
 	Required bool          `json:"required" yaml:"required"`
 	Min      *int64        `json:"min" yaml:"min"`
@@ -100,7 +100,7 @@ var rawIntColumnFieldValidations = []*cr.StructFieldValidation{
 }
 
 type RawFloatColumn struct {
-	ResourceConfigFields
+	ResourceFields
 	Type     ColumnType    `json:"type" yaml:"type"`
 	Required bool          `json:"required" yaml:"required"`
 	Min      *float32      `json:"min" yaml:"min"`
@@ -147,7 +147,7 @@ var rawFloatColumnFieldValidations = []*cr.StructFieldValidation{
 }
 
 type RawStringColumn struct {
-	ResourceConfigFields
+	ResourceFields
 	Type     ColumnType    `json:"type" yaml:"type"`
 	Required bool          `json:"required" yaml:"required"`
 	Values   []string      `json:"values" yaml:"values"`