Add feature derivator (#771)

* add feature derivator (WIP)

* wip

* wip add test

* wip: infer column data details

* test done

* clean unused code

* fix CI

Author: typhoonzero

sql/codegen.go

@@ -112,7 +112,12 @@ func newFiller(pr *extendedSelect, ds *trainAndValDataset, fts fieldTypes, db *D
 		},
 	}
 
-	trainResolved, err := resolveTrainClause(&pr.trainClause)
+	var err error
+	r.connectionConfig, err = newConnectionConfig(db)
+	if err != nil {
+		return nil, err
+	}
+	trainResolved, err := resolveTrainClause(&pr.trainClause, &pr.standardSelect, r.connectionConfig)
 	if err != nil {
 		return nil, err
 	}
@@ -225,7 +230,6 @@ func newFiller(pr *extendedSelect, ds *trainAndValDataset, fts fieldTypes, db *D
 		}
 	}
 
-	r.connectionConfig, err = newConnectionConfig(db)
 	if err == nil && r.Driver == "hive" {
 		// remove the last ';' which leads to a (hive)ParseException
 		r.TrainingDatasetSQL = strings.TrimSuffix(r.TrainingDatasetSQL, ";")

sql/codegen_alps.go

@@ -178,7 +178,7 @@ func modelCreatorCode(resolved *resolvedTrainClause, args []string) (string, str
 }
 
 func newALPSTrainFiller(pr *extendedSelect, db *DB, session *pb.Session, ds *trainAndValDataset) (*alpsFiller, error) {
-	resolved, err := resolveTrainClause(&pr.trainClause)
+	resolved, err := resolveTrainClause(&pr.trainClause, &pr.standardSelect, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -675,6 +675,7 @@ func (meta *metadata) getDenseColumnInfo(keys []string, refColumns map[string]*c
 					shape,
 					userSpec.DType,
 					userSpec.Delimiter,
+					nil,
 					*meta.featureMap}
 			} else {
 				output[ct.Name()] = &columns.ColumnSpec{
@@ -683,6 +684,7 @@ func (meta *metadata) getDenseColumnInfo(keys []string, refColumns map[string]*c
 					shape,
 					"float",
 					",",
+					nil,
 					*meta.featureMap}
 			}
 		}
@@ -730,7 +732,7 @@ func (meta *metadata) getSparseColumnInfo() (map[string]*columns.ColumnSpec, err
 		column, present := output[*name]
 		if !present {
 			shape := make([]int, 0, 1000)
-			column := &columns.ColumnSpec{*name, true, shape, "int64", "", *meta.featureMap}
+			column := &columns.ColumnSpec{*name, true, shape, "int64", "", nil, *meta.featureMap}
 			column.DType = "int64"
 			output[*name] = column
 		}

sql/codegen_elasticdl.go

@@ -143,7 +143,7 @@ func newElasticDLDataConversionFiller(pr *extendedSelect, db *DB, recordIODataDi
 }
 
 func newElasticDLTrainFiller(pr *extendedSelect, db *DB, session *pb.Session, ds *trainAndValDataset) (*elasticDLFiller, error) {
-	resolved, err := resolveTrainClause(&pr.trainClause)
+	resolved, err := resolveTrainClause(&pr.trainClause, &pr.standardSelect, nil)
 	if err != nil {
 		return nil, err
 	}

sql/columns/bucket_column.go

@@ -36,6 +36,11 @@ func (bc *BucketColumn) GenerateCode(cs *ColumnSpec) ([]string, error) {
 		strings.Join(strings.Split(fmt.Sprint(bc.Boundaries), " "), ","))}, nil
 }
 
+// GetKey implements the FeatureColumn interface.
+func (bc *BucketColumn) GetKey() string {
+	return bc.SourceColumn.Key
+}
+
 // GetDelimiter implements the FeatureColumn interface.
 func (bc *BucketColumn) GetDelimiter() string {
 	return ""
@@ -46,11 +51,6 @@ func (bc *BucketColumn) GetDtype() string {
 	return ""
 }
 
-// GetKey implements the FeatureColumn interface.
-func (bc *BucketColumn) GetKey() string {
-	return bc.SourceColumn.Key
-}
-
 // GetInputShape implements the FeatureColumn interface.
 func (bc *BucketColumn) GetInputShape() string {
 	return bc.SourceColumn.GetInputShape()

sql/columns/category_id_column.go

@@ -32,7 +32,6 @@ type SequenceCategoryIDColumn struct {
 	BucketSize int
 	Delimiter  string
 	Dtype      string
-	IsSequence bool
 }
 
 // GenerateCode implements the FeatureColumn interface.
@@ -41,6 +40,11 @@ func (cc *CategoryIDColumn) GenerateCode(cs *ColumnSpec) ([]string, error) {
 		cc.Key, cc.BucketSize)}, nil
 }
 
+// GetKey implements the FeatureColumn interface.
+func (cc *CategoryIDColumn) GetKey() string {
+	return cc.Key
+}
+
 // GetDelimiter implements the FeatureColumn interface.
 func (cc *CategoryIDColumn) GetDelimiter() string {
 	return cc.Delimiter
@@ -51,11 +55,6 @@ func (cc *CategoryIDColumn) GetDtype() string {
 	return cc.Dtype
 }
 
-// GetKey implements the FeatureColumn interface.
-func (cc *CategoryIDColumn) GetKey() string {
-	return cc.Key
-}
-
 // GetInputShape implements the FeatureColumn interface.
 func (cc *CategoryIDColumn) GetInputShape() string {
 	return fmt.Sprintf("[%d]", cc.BucketSize)

sql/columns/column_spec.go

@@ -32,6 +32,7 @@ type ColumnSpec struct {
 	Shape      []int
 	DType      string
 	Delimiter  string
+	Vocabulary map[string]string // use a map to generate a list without duplication
 	FeatureMap FeatureMap
 }
 

sql/columns/cross_column.go

@@ -51,6 +51,12 @@ func (cc *CrossColumn) GenerateCode(cs *ColumnSpec) ([]string, error) {
 		strings.Join(keysGenerated, ","), cc.HashBucketSize)}, nil
 }
 
+// GetKey implements the FeatureColumn interface.
+func (cc *CrossColumn) GetKey() string {
+	// NOTE: cross column is a feature on multiple column keys.
+	return ""
+}
+
 // GetDelimiter implements the FeatureColumn interface.
 func (cc *CrossColumn) GetDelimiter() string {
 	return ""
@@ -61,12 +67,6 @@ func (cc *CrossColumn) GetDtype() string {
 	return ""
 }
 
-// GetKey implements the FeatureColumn interface.
-func (cc *CrossColumn) GetKey() string {
-	// NOTE: cross column is a feature on multiple column keys.
-	return ""
-}
-
 // GetInputShape implements the FeatureColumn interface.
 func (cc *CrossColumn) GetInputShape() string {
 	// NOTE: return empty since crossed column input shape is already determined

sql/columns/embedding_column.go

@@ -19,12 +19,21 @@ import (
 
 // EmbeddingColumn is the wrapper of `tf.feature_column.embedding_column`
 type EmbeddingColumn struct {
+	Key            string // only used when CategoryColumn = nil, feature derivation will fill up the details
 	CategoryColumn interface{}
 	Dimension      int
 	Combiner       string
 	Initializer    string
 }
 
+// GetKey implements the FeatureColumn interface.
+func (ec *EmbeddingColumn) GetKey() string {
+	if ec.CategoryColumn != nil {
+		return ec.CategoryColumn.(FeatureColumn).GetKey()
+	}
+	return ec.Key
+}
+
 // GetDelimiter implements the FeatureColumn interface.
 func (ec *EmbeddingColumn) GetDelimiter() string {
 	return ec.CategoryColumn.(FeatureColumn).GetDelimiter()
@@ -35,11 +44,6 @@ func (ec *EmbeddingColumn) GetDtype() string {
 	return ec.CategoryColumn.(FeatureColumn).GetDtype()
 }
 
-// GetKey implements the FeatureColumn interface.
-func (ec *EmbeddingColumn) GetKey() string {
-	return ec.CategoryColumn.(FeatureColumn).GetKey()
-}
-
 // GetInputShape implements the FeatureColumn interface.
 func (ec *EmbeddingColumn) GetInputShape() string {
 	return ec.CategoryColumn.(FeatureColumn).GetInputShape()

sql/columns/feature_column.go

@@ -35,12 +35,13 @@ type FeatureColumn interface {
 	// feature_column code. And we maybe use one compound column's data to generate
 	// multiple feature columns, so return a list of strings.
 	GenerateCode(cs *ColumnSpec) ([]string, error)
+	GetKey() string
+
 	// FIXME(typhoonzero): remove delimiter, dtype shape from feature column
 	// get these from column spec claused or by feature derivation.
 	GetDelimiter() string
 	GetDtype() string
-	GetKey() string
-	// return input shape json string, like "[2,3]"
 	GetInputShape() string
+
 	GetColumnType() int
 }

sql/expression_resolver.go

@@ -67,6 +67,8 @@ type resolvedTrainClause struct {
 	ColumnSpecs                   map[string][]*columns.ColumnSpec
 	EngineParams                  engineSpec
 	CustomModule                  *gitLabModule
+	FeatureColumnInfered          FeatureColumnMap
+	ColumnSpecInfered             ColumnSpecMap
 }
 
 type resolvedPredictClause struct {
@@ -133,7 +135,7 @@ func getStringsAttr(attrs map[string]*attribute, key string, defaultValue []stri
 	return defaultValue
 }
 
-func resolveTrainClause(tc *trainClause) (*resolvedTrainClause, error) {
+func resolveTrainClause(tc *trainClause, slct *standardSelect, connConfig *connectionConfig) (*resolvedTrainClause, error) {
 	modelName := tc.estimator
 	preMadeModel := !strings.ContainsAny(modelName, ".")
 	attrs, err := resolveAttribute(&tc.trainAttrs)
@@ -219,6 +221,11 @@ func resolveTrainClause(tc *trainClause) (*resolvedTrainClause, error) {
 		fcMap[target] = fcs
 		csMap[target] = css
 	}
+	// TODO(typhoonzero): use the derivated maps for codegen, skip checking error
+	// since it's not used by codegen yet.
+	// also, need to clean up what is inside "resolvedTrainClause", keep only
+	// fcInfered, csInfered
+	fcInfered, csInfered, err := InferFeatureColumns(slct, fcMap, csMap, connConfig)
 
 	return &resolvedTrainClause{
 		IsPreMadeModel:                preMadeModel,
@@ -246,7 +253,10 @@ func resolveTrainClause(tc *trainClause) (*resolvedTrainClause, error) {
 		FeatureColumns:                fcMap,
 		ColumnSpecs:                   csMap,
 		EngineParams:                  getEngineSpec(engineParams),
-		CustomModule:                  customModel}, nil
+		CustomModule:                  customModel,
+		FeatureColumnInfered:          fcInfered,
+		ColumnSpecInfered:             csInfered,
+	}, nil
 }
 
 func resolvePredictClause(pc *predictClause) (*resolvedPredictClause, error) {
@@ -284,7 +294,6 @@ func resolveTrainColumns(columnExprs *exprlist) ([]columns.FeatureColumn, []*col
 	for _, expr := range *columnExprs {
 		if expr.typ != 0 {
 			// Column identifier like "COLUMN a1,b1"
-			// FIXME(typhoonzero): infer the column spec here.
 			c := &columns.NumericColumn{
 				Key:   expr.val,
 				Shape: []int{1},
@@ -380,18 +389,6 @@ func expression2string(e interface{}) (string, error) {
 	return "", fmt.Errorf("expression expected to be string, actual: %s", resolved)
 }
 
-// func generateFeatureColumnCode(fcs []columns.FeatureColumn) (string, error) {
-// 	var codes = make([]string, 0, len(fcs))
-// 	for _, fc := range fcs {
-// 		code, err := fc.GenerateCode()
-// 		if err != nil {
-// 			return "", nil
-// 		}
-// 		codes = append(codes, code)
-// 	}
-// 	return fmt.Sprintf("[%s]", strings.Join(codes, ",")), nil
-// }
-
 func resolveDelimiter(delimiter string) (string, error) {
 	if strings.EqualFold(delimiter, comma) {
 		return ",", nil
@@ -476,8 +473,7 @@ func resolveSeqCategoryIDColumn(el *exprlist) (*columns.SequenceCategoryIDColumn
 		BucketSize: bucketSize,
 		Delimiter:  delimiter,
 		// TODO(typhoonzero): support config dtype
-		Dtype:      "int64",
-		IsSequence: true}, cs, nil
+		Dtype: "int64"}, cs, nil
 }
 
 func resolveCategoryIDColumn(el *exprlist) (*columns.CategoryIDColumn, *columns.ColumnSpec, error) {
@@ -551,50 +547,73 @@ func resolveCrossColumn(el *exprlist) (*columns.CrossColumn, error) {
 		HashBucketSize: bucketSize}, nil
 }
 
-func resolveEmbeddingColumn(el *exprlist) (*columns.EmbeddingColumn, error) {
+func resolveEmbeddingColumn(el *exprlist) (*columns.EmbeddingColumn, *columns.ColumnSpec, error) {
 	if len(*el) != 4 && len(*el) != 5 {
-		return nil, fmt.Errorf("bad EMBEDDING expression format: %s", *el)
+		return nil, nil, fmt.Errorf("bad EMBEDDING expression format: %s", *el)
 	}
+
 	sourceExprList := (*el)[1]
 	var source columns.FeatureColumn
+	var cs *columns.ColumnSpec
 	var err error
+	var innerCategoryColumnKey string
+
+	var catColumnResult interface{}
 	if sourceExprList.typ == 0 {
-		source, _, err = resolveColumn(&sourceExprList.sexp)
+		source, cs, err = resolveColumn(&sourceExprList.sexp)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
-	} else {
-		return nil, fmt.Errorf("key of EMBEDDING must be categorical column")
-	}
-	// TODO(uuleon) support other kinds of categorical column in the future
-	var catColumn interface{}
-	catColumn, ok := source.(*columns.CategoryIDColumn)
-	if !ok {
-		catColumn, ok = source.(*columns.SequenceCategoryIDColumn)
-		if !ok {
-			return nil, fmt.Errorf("key of EMBEDDING must be categorical column")
+		// user may write EMBEDDING(SPARSE(...)) or EMBEDDING(DENSE(...))
+		if cs != nil {
+			innerCategoryColumnKey = cs.ColumnName
+			catColumnResult = &columns.CategoryIDColumn{
+				Key:        cs.ColumnName,
+				BucketSize: cs.Shape[0],
+				Delimiter:  cs.Delimiter,
+				Dtype:      cs.DType,
+			}
+		} else {
+			// TODO(uuleon) support other kinds of categorical column in the future
+			var catColumn interface{}
+			catColumn, ok := source.(*columns.CategoryIDColumn)
+			if !ok {
+				catColumn, ok = source.(*columns.SequenceCategoryIDColumn)
+				if !ok {
+					return nil, nil, fmt.Errorf("key of EMBEDDING must be categorical column")
+				}
+			}
+			// NOTE: to avoid golang multiple assignment compiler restrictions
+			catColumnResult = catColumn
+			innerCategoryColumnKey = source.GetKey()
 		}
+	} else {
+		// generate a default CategoryIDColumn for later feature derivation.
+		catColumnResult = nil
+		innerCategoryColumnKey = sourceExprList.val
 	}
+
 	dimension, err := strconv.Atoi((*el)[2].val)
 	if err != nil {
-		return nil, fmt.Errorf("bad EMBEDDING dimension: %s, err: %s", (*el)[2].val, err)
+		return nil, nil, fmt.Errorf("bad EMBEDDING dimension: %s, err: %s", (*el)[2].val, err)
 	}
 	combiner, err := expression2string((*el)[3])
 	if err != nil {
-		return nil, fmt.Errorf("bad EMBEDDING combiner: %s, err: %s", (*el)[3], err)
+		return nil, nil, fmt.Errorf("bad EMBEDDING combiner: %s, err: %s", (*el)[3], err)
 	}
 	initializer := ""
 	if len(*el) == 5 {
 		initializer, err = expression2string((*el)[4])
 		if err != nil {
-			return nil, fmt.Errorf("bad EMBEDDING initializer: %s, err: %s", (*el)[4], err)
+			return nil, nil, fmt.Errorf("bad EMBEDDING initializer: %s, err: %s", (*el)[4], err)
 		}
 	}
 	return &columns.EmbeddingColumn{
-		CategoryColumn: catColumn,
+		Key:            innerCategoryColumnKey,
+		CategoryColumn: catColumnResult,
 		Dimension:      dimension,
 		Combiner:       combiner,
-		Initializer:    initializer}, nil
+		Initializer:    initializer}, cs, nil
 }
 
 func resolveNumericColumn(el *exprlist) (*columns.NumericColumn, error) {
@@ -710,8 +729,7 @@ func resolveColumn(el *exprlist) (columns.FeatureColumn, *columns.ColumnSpec, er
 	case seqCategoryID:
 		return resolveSeqCategoryIDColumn(el)
 	case embedding:
-		fc, err := resolveEmbeddingColumn(el)
-		return fc, nil, err
+		return resolveEmbeddingColumn(el)
 	default:
 		return nil, nil, fmt.Errorf("not supported expr: %s", head)
 	}