Make raw column kinds optional

cli/cmd/get.go

@@ -306,7 +306,7 @@ func describeRawColumn(name string, resourcesRes *schema.GetResourcesResponse) (
 	}
 	dataStatus := resourcesRes.DataStatuses[rawColumn.GetID()]
 	out := dataStatusSummary(dataStatus)
-	out += resourceStr(rawColumn.GetUserConfig())
+	out += resourceStr(context.GetRawColumnUserConfig(rawColumn))
 	return out, nil
 }
 

examples/reviews/resources/tokenized_columns.yaml → examples/reviews/resources/columns.yaml

@@ -1,3 +1,13 @@
+- kind: environment
+  name: dev
+  data:
+    type: csv
+    path: s3a://cortex-examples/reviews.csv
+    csv_config:
+      header: true
+      escape: "\""
+    schema: ["review", "label"]
+
 - kind: transformed_column
   name: embedding_input
   transformer_path: implementations/transformers/tokenize_string_to_int.py

examples/reviews/resources/vocab.yaml

@@ -1,15 +1,6 @@
-- kind: aggregator
-  name: vocab
-  output_type: {STRING: INT}
-  inputs:
-    columns:
-      col: STRING_COLUMN
-    args:
-      vocab_size: INT
-
 - kind: aggregate
   name: reviews_vocab
-  aggregator: vocab
+  aggregator_path: implementations/aggregators/vocab.py
   inputs:
     columns:
       col: review

pkg/operator/api/context/raw_columns.go

@@ -27,7 +27,6 @@ type RawColumns map[string]RawColumn
 type RawColumn interface {
 	Column
 	GetCompute() *userconfig.SparkCompute
-	GetUserConfig() userconfig.Resource
 }
 
 type RawIntColumn struct {
@@ -45,6 +44,11 @@ type RawStringColumn struct {
 	*ComputedResourceFields
 }
 
+type RawInferredColumn struct {
+	*userconfig.RawInferredColumn
+	*ComputedResourceFields
+}
+
 func (rawColumns RawColumns) OneByID(id string) RawColumn {
 	for _, rawColumn := range rawColumns {
 		if rawColumn.GetID() == id {
@@ -79,6 +83,21 @@ func (rawColumns RawColumns) columnInputsID(columnInputValues map[string]interfa
 	return hash.Any(columnIDMap)
 }
 
+func GetRawColumnUserConfig(rawColumn RawColumn) userconfig.Resource {
+	switch rawColumn.GetType() {
+	case userconfig.IntegerColumnType:
+		return rawColumn.(*RawIntColumn).RawIntColumn
+	case userconfig.FloatColumnType:
+		return rawColumn.(*RawFloatColumn).RawFloatColumn
+	case userconfig.StringColumnType:
+		return rawColumn.(*RawStringColumn).RawStringColumn
+	case userconfig.InferredColumnType:
+		return rawColumn.(*RawInferredColumn).RawInferredColumn
+	}
+
+	return nil
+}
+
 func (rawColumns RawColumns) ColumnInputsID(columnInputValues map[string]interface{}) string {
 	return rawColumns.columnInputsID(columnInputValues, false)
 }
@@ -98,3 +117,7 @@ func (rawColumn *RawFloatColumn) GetInputRawColumnNames() []string {
 func (rawColumn *RawStringColumn) GetInputRawColumnNames() []string {
 	return []string{rawColumn.GetName()}
 }
+
+func (rawColumn *RawInferredColumn) GetInputRawColumnNames() []string {
+	return []string{rawColumn.GetName()}
+}

pkg/operator/api/context/serialize.go

@@ -25,9 +25,10 @@ import (
 )
 
 type RawColumnsTypeSplit struct {
-	RawIntColumns    map[string]*RawIntColumn    `json:"raw_int_columns"`
-	RawStringColumns map[string]*RawStringColumn `json:"raw_string_columns"`
-	RawFloatColumns  map[string]*RawFloatColumn  `json:"raw_float_columns"`
+	RawIntColumns      map[string]*RawIntColumn      `json:"raw_int_columns"`
+	RawStringColumns   map[string]*RawStringColumn   `json:"raw_string_columns"`
+	RawFloatColumns    map[string]*RawFloatColumn    `json:"raw_float_columns"`
+	RawInferredColumns map[string]*RawInferredColumn `json:"raw_inferred_columns"`
 }
 
 type DataSplit struct {
@@ -45,6 +46,7 @@ func (ctx Context) splitRawColumns() *RawColumnsTypeSplit {
 	var rawIntColumns = make(map[string]*RawIntColumn)
 	var rawFloatColumns = make(map[string]*RawFloatColumn)
 	var rawStringColumns = make(map[string]*RawStringColumn)
+	var rawInferredColumns = make(map[string]*RawInferredColumn)
 	for name, rawColumn := range ctx.RawColumns {
 		switch typedRawColumn := rawColumn.(type) {
 		case *RawIntColumn:
@@ -53,13 +55,16 @@ func (ctx Context) splitRawColumns() *RawColumnsTypeSplit {
 			rawFloatColumns[name] = typedRawColumn
 		case *RawStringColumn:
 			rawStringColumns[name] = typedRawColumn
+		case *RawInferredColumn:
+			rawInferredColumns[name] = typedRawColumn
 		}
 	}
 
 	return &RawColumnsTypeSplit{
-		RawIntColumns:    rawIntColumns,
-		RawFloatColumns:  rawFloatColumns,
-		RawStringColumns: rawStringColumns,
+		RawIntColumns:      rawIntColumns,
+		RawFloatColumns:    rawFloatColumns,
+		RawStringColumns:   rawStringColumns,
+		RawInferredColumns: rawInferredColumns,
 	}
 }
 
@@ -75,6 +80,9 @@ func (serial Serial) collectRawColumns() RawColumns {
 	for name, rawColumn := range serial.RawColumnSplit.RawStringColumns {
 		rawColumns[name] = rawColumn
 	}
+	for name, rawColumn := range serial.RawColumnSplit.RawInferredColumns {
+		rawColumns[name] = rawColumn
+	}
 
 	return rawColumns
 }

pkg/operator/api/userconfig/column_type.go

@@ -31,6 +31,7 @@ const (
 	IntegerListColumnType
 	FloatListColumnType
 	StringListColumnType
+	InferredColumnType
 )
 
 var columnTypes = []string{
@@ -41,6 +42,7 @@ var columnTypes = []string{
 	"INT_LIST_COLUMN",
 	"FLOAT_LIST_COLUMN",
 	"STRING_LIST_COLUMN",
+	"INFERRED_COLUMN",
 }
 
 var columnJSONPlaceholders = []string{
@@ -51,6 +53,7 @@ var columnJSONPlaceholders = []string{
 	"[INT]",
 	"[FLOAT]",
 	"[\"STRING\"]",
+	"VALUE",
 }
 
 func ColumnTypeFromString(s string) ColumnType {

pkg/operator/api/userconfig/config.go

@@ -154,9 +154,9 @@ func (config *Config) Validate(envName string) error {
 	rawColumnNames := config.RawColumns.Names()
 	for _, env := range config.Environments {
 		ingestedColumnNames := env.Data.GetIngestedColumns()
-		missingColumns := slices.SubtractStrSlice(rawColumnNames, ingestedColumnNames)
-		if len(missingColumns) > 0 {
-			return errors.Wrap(ErrorRawColumnNotInEnv(env.Name), Identify(config.RawColumns.Get(missingColumns[0])))
+		missingColumnNames := slices.SubtractStrSlice(rawColumnNames, ingestedColumnNames)
+		if len(missingColumnNames) > 0 {
+			return errors.Wrap(ErrorRawColumnNotInEnv(env.Name), Identify(config.RawColumns.Get(missingColumnNames[0])))
 		}
 		extraColumns := slices.SubtractStrSlice(rawColumnNames, ingestedColumnNames)
 		if len(extraColumns) > 0 {
@@ -440,6 +440,22 @@ func New(configs map[string][]byte, envName string) (*Config, error) {
 		}
 	}
 
+	for _, env := range config.Environments {
+		ingestedColumnNames := env.Data.GetIngestedColumns()
+		missingColumnNames := slices.SubtractStrSlice(ingestedColumnNames, config.RawColumns.Names())
+		for _, inferredColumnName := range missingColumnNames {
+			inferredRawColumn := &RawInferredColumn{
+				ResourceFields: ResourceFields{
+					Name: inferredColumnName,
+				},
+				Type:    InferredColumnType,
+				Compute: &SparkCompute{},
+			}
+			cr.Struct(inferredRawColumn.Compute, make(map[string]interface{}), sparkComputeStructValidation)
+			config.RawColumns = append(config.RawColumns, inferredRawColumn)
+		}
+	}
+
 	if err := config.Validate(envName); err != nil {
 		return nil, err
 	}

pkg/operator/api/userconfig/environments.go

@@ -21,6 +21,7 @@ import (
 	cr "github.com/cortexlabs/cortex/pkg/lib/configreader"
 	"github.com/cortexlabs/cortex/pkg/lib/errors"
 	"github.com/cortexlabs/cortex/pkg/lib/pointer"
+	"github.com/cortexlabs/cortex/pkg/lib/sets/strset"
 	"github.com/cortexlabs/cortex/pkg/lib/slices"
 	"github.com/cortexlabs/cortex/pkg/operator/api/resource"
 )
@@ -337,6 +338,13 @@ func (environments Environments) Validate() error {
 		return ErrorDuplicateResourceName(dups...)
 	}
 
+	ingestedColumns := environments[0].Data.GetIngestedColumns()
+	for _, env := range environments[1:] {
+		if !strset.New(ingestedColumns...).IsEqual(strset.New(env.Data.GetIngestedColumns()...)) {
+			return ErrorEnvSchemaMismatch(environments[0], env)
+		}
+	}
+
 	return nil
 }
 

pkg/operator/api/userconfig/errors.go

@@ -59,6 +59,7 @@ const (
 	ErrRegressionTargetType
 	ErrClassificationTargetType
 	ErrSpecifyOnlyOneMissing
+	ErrEnvSchemaMismatch
 )
 
 var errorKinds = []string{
@@ -92,9 +93,10 @@ var errorKinds = []string{
 	"err_regression_target_type",
 	"err_classification_target_type",
 	"err_specify_only_one_missing",
+	"err_env_schema_mismatch",
 }
 
-var _ = [1]int{}[int(ErrSpecifyOnlyOneMissing)-(len(errorKinds)-1)] // Ensure list length matches
+var _ = [1]int{}[int(ErrEnvSchemaMismatch)-(len(errorKinds)-1)] // Ensure list length matches
 
 func (t ErrorKind) String() string {
 	return errorKinds[t]
@@ -397,3 +399,15 @@ func ErrorSpecifyOnlyOneMissing(vals ...string) error {
 		message: message,
 	}
 }
+
+func ErrorEnvSchemaMismatch(env1, env2 *Environment) error {
+	return Error{
+		Kind: ErrEnvSchemaMismatch,
+		message: fmt.Sprintf("schemas diverge between environments (%s lists %s, and %s lists %s)",
+			env1.Name,
+			s.StrsAnd(env1.Data.GetIngestedColumns()),
+			env2.Name,
+			s.StrsAnd(env2.Data.GetIngestedColumns()),
+		),
+	}
+}

pkg/operator/api/userconfig/raw_columns.go

@@ -25,7 +25,6 @@ type RawColumn interface {
 	Column
 	GetType() ColumnType
 	GetCompute() *SparkCompute
-	GetUserConfig() Resource
 }
 
 type RawColumns []RawColumn
@@ -181,6 +180,12 @@ var rawStringColumnFieldValidations = []*cr.StructFieldValidation{
 	typeFieldValidation,
 }
 
+type RawInferredColumn struct {
+	ResourceFields
+	Type    ColumnType    `json:"type" yaml:"type"`
+	Compute *SparkCompute `json:"compute" yaml:"compute"`
+}
+
 func (rawColumns RawColumns) Validate() error {
 	resources := make([]Resource, len(rawColumns))
 	for i, res := range rawColumns {
@@ -224,6 +229,10 @@ func (column *RawStringColumn) GetType() ColumnType {
 	return column.Type
 }
 
+func (column *RawInferredColumn) GetType() ColumnType {
+	return column.Type
+}
+
 func (column *RawIntColumn) GetCompute() *SparkCompute {
 	return column.Compute
 }
@@ -236,6 +245,10 @@ func (column *RawStringColumn) GetCompute() *SparkCompute {
 	return column.Compute
 }
 
+func (column *RawInferredColumn) GetCompute() *SparkCompute {
+	return column.Compute
+}
+
 func (column *RawIntColumn) GetResourceType() resource.Type {
 	return resource.RawColumnType
 }
@@ -248,6 +261,10 @@ func (column *RawStringColumn) GetResourceType() resource.Type {
 	return resource.RawColumnType
 }
 
+func (column *RawInferredColumn) GetResourceType() resource.Type {
+	return resource.RawColumnType
+}
+
 func (column *RawIntColumn) IsRaw() bool {
 	return true
 }
@@ -260,14 +277,6 @@ func (column *RawStringColumn) IsRaw() bool {
 	return true
 }
 
-func (column *RawIntColumn) GetUserConfig() Resource {
-	return column
-}
-
-func (column *RawFloatColumn) GetUserConfig() Resource {
-	return column
-}
-
-func (column *RawStringColumn) GetUserConfig() Resource {
-	return column
+func (column *RawInferredColumn) IsRaw() bool {
+	return true
 }

pkg/operator/api/userconfig/validators.go

@@ -139,6 +139,11 @@ func CheckColumnRuntimeTypesMatch(columnRuntimeTypes map[string]interface{}, col
 			if !ok {
 				return errors.Wrap(ErrorUnsupportedColumnType(columnRuntimeTypeInter, validTypes), columnInputName)
 			}
+
+			if columnRuntimeType == InferredColumnType {
+				continue
+			}
+
 			if !slices.HasString(validTypes, columnRuntimeType.String()) {
 				return errors.Wrap(ErrorUnsupportedColumnType(columnRuntimeType, validTypes), columnInputName)
 			}

pkg/operator/context/raw_columns.go

@@ -93,8 +93,20 @@ func getRawColumns(
 				},
 				RawStringColumn: typedColumnConfig,
 			}
+		case *userconfig.RawInferredColumn:
+			buf.WriteString(typedColumnConfig.Name)
+			id := hash.Bytes(buf.Bytes())
+			rawColumn = &context.RawInferredColumn{
+				ComputedResourceFields: &context.ComputedResourceFields{
+					ResourceFields: &context.ResourceFields{
+						ID:           id,
+						ResourceType: resource.RawColumnType,
+					},
+				},
+				RawInferredColumn: typedColumnConfig,
+			}
 		default:
-			return nil, errors.Wrap(configreader.ErrorInvalidStr(userconfig.TypeKey, userconfig.IntegerColumnType.String(), userconfig.FloatColumnType.String(), userconfig.StringColumnType.String()), userconfig.Identify(columnConfig)) // unexpected error
+			return nil, errors.Wrap(configreader.ErrorInvalidStr(typedColumnConfig.GetType().String(), userconfig.IntegerColumnType.String(), userconfig.FloatColumnType.String(), userconfig.StringColumnType.String()), userconfig.Identify(columnConfig)) // unexpected error
 		}
 
 		rawColumns[columnConfig.GetName()] = rawColumn

pkg/operator/context/transformers.go

@@ -67,7 +67,8 @@ func loadUserTransformers(
 			ResourceFields: userconfig.ResourceFields{
 				Name: implHash,
 			},
-			Path: *transColConfig.TransformerPath,
+			OutputType: userconfig.InferredColumnType,
+			Path:       *transColConfig.TransformerPath,
 		}
 		transformer, err := newTransformer(*anonTransformerConfig, impl, nil, pythonPackages)
 		if err != nil {