Repository: oom-ai/oomstore
Pull request #842 (2021-12-29 03:17:52)

Jinghan/modify offline method `Join`: handle streaming features
This PR refactors offline method `Join` to handle streaming features.

related to #803 


Patch:
diff --git a/internal/database/dbutil/schema.go b/internal/database/dbutil/schema.go
index d36737c6..51429a91 100644
--- a/internal/database/dbutil/schema.go
+++ b/internal/database/dbutil/schema.go
@@ -21,18 +21,20 @@ CREATE TABLE "{{ .TableName }}" (
 	{{ columnJoin .Columns .Backend }}
 )`
 
+// TODO: Add back `PRIMARY KEY` back when we have functions for
+// creating cdc table
 var (
 	createSchemaFuncs = template.FuncMap{
 		"entity": func(entity types.Entity, backend types.BackendType) string {
 			switch backend {
 			case types.BackendCassandra, types.BackendSQLite:
-				return fmt.Sprintf(`"%s" TEXT PRIMARY KEY`, entity.Name)
+				return fmt.Sprintf(`"%s" TEXT`, entity.Name)
 			case types.BackendPostgres, types.BackendSnowflake:
-				return fmt.Sprintf(`"%s" VARCHAR(%d) PRIMARY KEY`, entity.Name, entity.Length)
+				return fmt.Sprintf(`"%s" VARCHAR(%d)`, entity.Name, entity.Length)
 			case types.BackendMySQL:
-				return fmt.Sprintf("`%s` VARCHAR(%d) PRIMARY KEY", entity.Name, entity.Length)
+				return fmt.Sprintf("`%s` VARCHAR(%d)", entity.Name, entity.Length)
 			default:
-				return fmt.Sprintf("%s VARCHAR(%d) PRIMARY KEY", entity.Name, entity.Length)
+				return fmt.Sprintf("%s VARCHAR(%d)", entity.Name, entity.Length)
 			}
 		},
 		"columnJoin": func(columns []Column, backend types.BackendType) string {
@@ -55,8 +57,19 @@ var (
 )
 
 type Column struct {
-	Name   string
-	DbType string
+	Name      string
+	DbType    string
+	ValueType types.ValueType
+}
+
+type ColumnList []Column
+
+func (c ColumnList) Names() []string {
+	names := make([]string, 0, len(c))
+	for _, column := range c {
+		names = append(names, column.Name)
+	}
+	return names
 }
 
 type CreateSchema struct {
diff --git a/internal/database/metadata/types.go b/internal/database/metadata/types.go
index 2872377d..91cabfcc 100644
--- a/internal/database/metadata/types.go
+++ b/internal/database/metadata/types.go
@@ -6,6 +6,7 @@ type RevisionRange struct {
 	MinRevision   int64
 	MaxRevision   int64
 	SnapshotTable string
+	CdcTable      string
 }
 
 // Create
diff --git a/internal/database/offline/bigquery/join.go b/internal/database/offline/bigquery/join.go
index 9b863764..be671b93 100644
--- a/internal/database/offline/bigquery/join.go
+++ b/internal/database/offline/bigquery/join.go
@@ -29,7 +29,7 @@ func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult,
 	return sqlutil.DoJoin(ctx, dbOpt, doJoinOpt)
 }
 
-func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, header, tableNames []string) (*types.JoinResult, error) {
+func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, header dbutil.ColumnList, dropTableNames []string, backendType types.BackendType) (*types.JoinResult, error) {
 	rows, err := dbOpt.BigQueryDB.Query(query).Read(ctx)
 	if err != nil {
 		return nil, err
@@ -40,7 +40,7 @@ func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string,
 
 	go func() {
 		defer func() {
-			if err = dropTemporaryTables(ctx, dbOpt.BigQueryDB, tableNames); err != nil {
+			if err = dropTemporaryTables(ctx, dbOpt.BigQueryDB, dropTableNames); err != nil {
 				dropErr = err
 			}
 			defer close(data)
@@ -53,11 +53,17 @@ func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string,
 			}
 			if err != nil {
 				scanErr = err
+				continue
 			}
 			record := make([]interface{}, 0, len(recordMap))
-			for _, h := range header {
-				column := strings.Split(h, ".")
-				record = append(record, recordMap[column[len(column)-1]])
+			for i := range header {
+				column := strings.Split(header[i].Name, ".")
+				deserializedValue, err := sqlutil.DeserializeByValueType(recordMap[column[len(column)-1]], header[i].ValueType, backendType)
+				if err != nil {
+					scanErr = err
+					continue
+				}
+				record = append(record, deserializedValue)
 			}
 			data <- record
 		}
@@ -69,7 +75,7 @@ func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string,
 	}
 
 	return &types.JoinResult{
-		Header: header,
+		Header: header.Names(),
 		Data:   data,
 	}, dropErr
 }
diff --git a/internal/database/offline/sqlutil/export.go b/internal/database/offline/sqlutil/export.go
index b9d86201..49aeb347 100644
--- a/internal/database/offline/sqlutil/export.go
+++ b/internal/database/offline/sqlutil/export.go
@@ -3,8 +3,6 @@ package sqlutil
 import (
 	"context"
 	"fmt"
-	"strconv"
-	"time"
 
 	"github.com/jmoiron/sqlx"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
@@ -57,7 +55,7 @@ func Export(ctx context.Context, db *sqlx.DB, opt offline.ExportOpt, backendType
 				}
 				if backendType == types.BackendSnowflake {
 
-					v, err := deserializeByTag(record[i+1], f.ValueType)
+					v, err := deserializeByTagForSnowflake(record[i+1], f.ValueType)
 					if err != nil {
 						errs <- fmt.Errorf("failed at deserializeByTag, err=%v", err)
 						return
@@ -75,49 +73,3 @@ func Export(ctx context.Context, db *sqlx.DB, opt offline.ExportOpt, backendType
 
 	return stream, errs
 }
-
-// gosnowflake Scan always produce string when the destination is interface{}
-// See https://github.com/snowflakedb/gosnowflake/issues/517
-// As a work around, we cast the string to interface{} based on ValueType
-// This method is mostly copied from redis.DeserializeByTag, except we use 10 rather than 36 as the base
-// TODO: we should let the snowflake team fix the gosnowflake converter
-func deserializeByTag(i interface{}, valueType types.ValueType) (interface{}, error) {
-	if i == nil {
-		return nil, nil
-	}
-
-	s, ok := i.(string)
-	if !ok {
-		return nil, fmt.Errorf("not a string or nil: %v", i)
-	}
-
-	switch valueType {
-	case types.String:
-		return s, nil
-
-	case types.Int64:
-		x, err := strconv.ParseInt(s, 10, 64)
-		return x, err
-
-	case types.Float64:
-		x, err := strconv.ParseFloat(s, 64)
-		return x, err
-
-	case types.Bool:
-		if s == "1" {
-			return true, nil
-		} else if s == "0" {
-			return false, nil
-		} else {
-			return nil, fmt.Errorf("invalid bool value: %s", s)
-		}
-	case types.Time:
-		x, err := strconv.ParseInt(s, 10, 64)
-		return time.UnixMilli(x), err
-
-	case types.Bytes:
-		return []byte(s), nil
-	default:
-		return "", fmt.Errorf("unsupported value type: %s", valueType)
-	}
-}
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index 495d53e7..2e4402f9 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -11,15 +11,17 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-type QueryResults func(ctx context.Context, dbOpt dbutil.DBOpt, query string, header, tableNames []string) (*types.JoinResult, error)
+type QueryResults func(ctx context.Context, dbOpt dbutil.DBOpt, query string, header dbutil.ColumnList, dropTableNames []string, backendType types.BackendType) (*types.JoinResult, error)
 
 type ReadJoinedTableOpt struct {
 	EntityRowsTableName string
 	TableNames          []string
+	AllTableNames       []string
 	FeatureMap          map[string]types.FeatureList
 	ValueNames          []string
 	QueryResults        QueryResults
 	ReadJoinResultQuery string
+	BackendType         types.BackendType
 }
 
 type DoJoinOpt struct {
@@ -57,14 +59,22 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 
 	// Step 2: process features by group, insert result to table joined
 	tableNames := make([]string, 0)
+	allTableNames := make([]string, 0)
 	tableToFeatureMap := make(map[string]types.FeatureList)
+	var category types.Category
 	for groupName, featureList := range opt.FeatureMap {
 		revisionRanges, ok := opt.RevisionRangeMap[groupName]
 		if !ok {
 			continue
 		}
-		joinedTableName, err := JoinOneGroup(ctx, dbOpt, offline.JoinOneGroupOpt{
+		if revisionRanges[0].CdcTable == "" {
+			category = types.CategoryBatch
+		} else {
+			category = types.CategoryStream
+		}
+		joinedTables, err := JoinOneGroup(ctx, dbOpt, offline.JoinOneGroupOpt{
 			GroupName:           groupName,
+			Category:            category,
 			Entity:              opt.Entity,
 			Features:            featureList,
 			RevisionRanges:      revisionRanges,
@@ -73,9 +83,11 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 		if err != nil {
 			return nil, err
 		}
-		if joinedTableName != "" {
-			tableNames = append(tableNames, joinedTableName)
-			tableToFeatureMap[joinedTableName] = featureList
+		allTableNames = append(allTableNames, joinedTables...)
+		if len(joinedTables) != 0 {
+			tableName := joinedTables[len(joinedTables)-1]
+			tableNames = append(tableNames, tableName)
+			tableToFeatureMap[tableName] = featureList
 		}
 	}
 
@@ -83,35 +95,37 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 	return ReadJoinedTable(ctx, dbOpt, ReadJoinedTableOpt{
 		EntityRowsTableName: entityRowsTableName,
 		TableNames:          tableNames,
+		AllTableNames:       allTableNames,
 		FeatureMap:          tableToFeatureMap,
 		ValueNames:          opt.ValueNames,
 		QueryResults:        opt.QueryResults,
 		ReadJoinResultQuery: opt.ReadJoinResultQuery,
+		BackendType:         dbOpt.Backend,
 	})
 }
 
-func JoinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.JoinOneGroupOpt) (string, error) {
+func JoinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.JoinOneGroupOpt) ([]string, error) {
 	if len(opt.Features) == 0 {
-		return "", nil
+		return nil, nil
 	}
 	qt, err := dbutil.QuoteFn(dbOpt.Backend)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 	entityKeyStr := qt("entity_key")
 	unixMilliStr := qt("unix_milli")
 
 	// Step 1: create temporary joined table
-	joinedTableName, err := PrepareJoinedTable(ctx, dbOpt, opt.Features, opt.Entity, opt.GroupName, opt.ValueNames)
+	snapshotJoinedTableName, err := PrepareJoinedTable(ctx, dbOpt, opt.Features, opt.Entity, opt.GroupName, opt.ValueNames)
 	if err != nil {
-		return "", err
+		return nil, err
 	}
 
 	// Step 2: iterate each table range, join entity_rows table and each data tables
 	columns := append(opt.ValueNames, opt.Features.Names()...)
 	for _, r := range opt.RevisionRanges {
 		query, err := buildJoinQuery(joinQueryParams{
-			TableName:           joinedTableName,
+			TableName:           snapshotJoinedTableName,
 			EntityKeyStr:        entityKeyStr,
 			EntityName:          opt.Entity.Name,
 			UnixMilliStr:        unixMilliStr,
@@ -122,18 +136,47 @@ func JoinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.JoinOneGr
 			DatasetID:           dbOpt.DatasetID,
 		})
 		if err != nil {
-			return "", err
+			return nil, err
 		}
 		if err = dbOpt.ExecContext(ctx, query, []interface{}{r.MinRevision, r.MaxRevision}); err != nil {
-			return "", err
+			return nil, err
 		}
 	}
+	if opt.Category == types.CategoryBatch {
+		return []string{snapshotJoinedTableName}, nil
+	}
 
-	return joinedTableName, nil
+	// Step 3: for streaming features, keep joining with cdc_table
+	cdcJoinedTableName, err := PrepareJoinedTable(ctx, dbOpt, opt.Features, opt.Entity, opt.GroupName, opt.ValueNames)
+	if err != nil {
+		return nil, err
+	}
+	for _, r := range opt.RevisionRanges {
+		query, err := buildCdcJoinQuery(cdcJoinQueryParams{
+			TableName:           cdcJoinedTableName,
+			EntityKeyStr:        entityKeyStr,
+			EntityName:          opt.Entity.Name,
+			UnixMilliStr:        unixMilliStr,
+			ValueNames:          opt.ValueNames,
+			FeatureNames:        opt.Features.Names(),
+			SnapshotJoinedTable: snapshotJoinedTableName,
+			CdcTable:            r.CdcTable,
+			Backend:             dbOpt.Backend,
+			DatasetID:           dbOpt.DatasetID,
+		})
+		if err != nil {
+			return nil, err
+		}
+		if err = dbOpt.ExecContext(ctx, query, []interface{}{r.MinRevision, r.MaxRevision}); err != nil {
+			return nil, err
+		}
+	}
+	return []string{snapshotJoinedTableName, cdcJoinedTableName}, nil
 }
 
 func ReadJoinedTable(ctx context.Context, dbOpt dbutil.DBOpt, opt ReadJoinedTableOpt) (*types.JoinResult, error) {
 	tableNames := opt.TableNames
+	dropTableNames := append([]string{opt.EntityRowsTableName}, opt.AllTableNames...)
 	if len(tableNames) == 0 {
 		return nil, nil
 	}
@@ -159,18 +202,31 @@ func ReadJoinedTable(ctx context.Context, dbOpt dbutil.DBOpt, opt ReadJoinedTabl
 		ON entity_rows_table.entity_key = joined_table_2.entity_key AND entity_rows_table.unix_milli = joined_table_2.unix_milli;
 	*/
 	var (
-		fields, header []string
+		fields         []string
+		header         []dbutil.Column
 		joinTablePairs []joinTablePair
 	)
-	header = append(header, "entity_key", "unix_milli")
+	header = append(header, dbutil.Column{
+		Name:      "entity_key",
+		ValueType: types.String,
+	}, dbutil.Column{
+		Name:      "unix_milli",
+		ValueType: types.Int64,
+	})
 	for _, name := range opt.ValueNames {
 		fields = append(fields, qt(name))
-		header = append(header, name)
+		header = append(header, dbutil.Column{
+			Name:      name,
+			ValueType: types.String,
+		})
 	}
 	for _, name := range tableNames {
 		for _, f := range opt.FeatureMap[name] {
 			fields = append(fields, fmt.Sprintf("%s.%s", qt(name), qt(f.Name)))
-			header = append(header, f.FullName)
+			header = append(header, dbutil.Column{
+				Name:      f.FullName,
+				ValueType: f.ValueType,
+			})
 		}
 	}
 
@@ -199,11 +255,16 @@ func ReadJoinedTable(ctx context.Context, dbOpt dbutil.DBOpt, opt ReadJoinedTabl
 	}
 
 	// Step 2: read joined results
-	return opt.QueryResults(ctx, dbOpt, query, header, tableNames)
+	return opt.QueryResults(ctx, dbOpt, query, header, dropTableNames, dbOpt.Backend)
 }
 
-func sqlxQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, header, tableNames []string) (*types.JoinResult, error) {
-	rows, err := dbOpt.SqlxDB.QueryxContext(ctx, query)
+func sqlxQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, header dbutil.ColumnList, dropTableNames []string, backendType types.BackendType) (*types.JoinResult, error) {
+	stmt, err := dbOpt.SqlxDB.Preparex(query)
+	if err != nil {
+		return nil, err
+	}
+	defer stmt.Close()
+	rows, err := stmt.Queryx()
 	if err != nil {
 		return nil, err
 	}
@@ -212,7 +273,7 @@ func sqlxQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, hea
 	var scanErr, dropErr error
 	go func() {
 		defer func() {
-			if err := dropTemporaryTables(ctx, dbOpt.SqlxDB, tableNames); err != nil {
+			if err := dropTemporaryTables(ctx, dbOpt.SqlxDB, dropTableNames); err != nil {
 				dropErr = err
 			}
 			defer rows.Close()
@@ -222,8 +283,18 @@ func sqlxQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, hea
 			record, err := rows.SliceScan()
 			if err != nil {
 				scanErr = err
+				continue
+			}
+			serializedRecord := make([]interface{}, 0, len(record))
+			for i, r := range record {
+				deserializedValue, err := DeserializeByValueType(r, header[i].ValueType, backendType)
+				if err != nil {
+					scanErr = err
+					continue
+				}
+				serializedRecord = append(serializedRecord, deserializedValue)
 			}
-			data <- record
+			data <- serializedRecord
 		}
 	}()
 
@@ -232,7 +303,7 @@ func sqlxQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, hea
 		return nil, scanErr
 	}
 	return &types.JoinResult{
-		Header: header,
+		Header: header.Names(),
 		Data:   data,
 	}, dropErr
 }
diff --git a/internal/database/offline/sqlutil/join_helper.go b/internal/database/offline/sqlutil/join_helper.go
index d7621fdc..ac5b49e1 100644
--- a/internal/database/offline/sqlutil/join_helper.go
+++ b/internal/database/offline/sqlutil/join_helper.go
@@ -30,6 +30,31 @@ ON l.{{ .EntityKeyStr }} = r.{{ qt .EntityName }}
 WHERE l.{{ .UnixMilliStr }} >= ? AND l.{{ .UnixMilliStr }} < ?
 `
 
+const CDC_JOIN_QUERY = `
+INSERT INTO {{ qt .TableName }} ( {{ .EntityKeyStr }}, {{ .UnixMilliStr }}, {{ columnJoinWithComma .ValueNames }} {{ columnJoin .FeatureNames }})
+SELECT
+	l.{{ .EntityKeyStr }} AS entity_key,
+	l.{{ .UnixMilliStr }} AS unix_milli,
+	{{ columnJoinWithComma .ValueNames }}
+	{{ featureValue .FeatureNames }}
+FROM
+	{{ qt .SnapshotJoinedTable }} AS l
+LEFT JOIN {{ qt .CdcTable }} AS r
+ON l.{{ .EntityKeyStr }} = r.{{ qt .EntityName }} AND l.{{ .UnixMilliStr }} >= r.{{ .UnixMilliStr }}
+WHERE
+	l.{{ .UnixMilliStr }} >= ? AND l.{{ .UnixMilliStr }} < ? AND
+	(
+		r.{{ .UnixMilliStr }} IS NULL OR
+		r.{{ .UnixMilliStr }} = (
+			SELECT MAX({{ .UnixMilliStr }})
+			FROM {{ qt .CdcTable }} AS r2
+			WHERE
+				l.{{ .EntityKeyStr }} = r2.{{ qt .EntityName }} AND
+				l.{{ .UnixMilliStr }} >= r2.{{ .UnixMilliStr }}
+		)
+	)
+`
+
 const READ_JOIN_RESULT_QUERY = `
 SELECT
 	{{ qt .EntityRowsTableName }}.{{ .EntityKeyStr }},
@@ -291,3 +316,55 @@ func buildJoinQuery(params joinQueryParams) (string, error) {
 	}
 	return buf.String(), nil
 }
+
+type cdcJoinQueryParams struct {
+	TableName           string
+	EntityKeyStr        string
+	EntityName          string
+	UnixMilliStr        string
+	ValueNames          []string
+	FeatureNames        []string
+	SnapshotJoinedTable string
+	CdcTable            string
+	Backend             types.BackendType
+	DatasetID           *string
+}
+
+func buildCdcJoinQuery(params cdcJoinQueryParams) (string, error) {
+	if params.Backend == types.BackendBigQuery {
+		params.TableName = fmt.Sprintf("%s.%s", *params.DatasetID, params.TableName)
+		params.SnapshotJoinedTable = fmt.Sprintf("%s.%s", *params.DatasetID, params.SnapshotJoinedTable)
+		params.CdcTable = fmt.Sprintf("%s.%s", *params.DatasetID, params.CdcTable)
+	}
+
+	qt, err := dbutil.QuoteFn(params.Backend)
+	if err != nil {
+		return "", err
+	}
+
+	t := template.Must(template.New("cdc_join").Funcs(template.FuncMap{
+		"qt": qt,
+		"columnJoin": func(columns []string) string {
+			return qt(columns...)
+		},
+		"columnJoinWithComma": func(columns []string) string {
+			if len(columns) == 0 {
+				return ""
+			}
+			return fmt.Sprintf("%s,", qt(columns...))
+		},
+		"featureValue": func(features []string) string {
+			values := make([]string, 0, len(features))
+			for _, c := range features {
+				values = append(values, fmt.Sprintf("(CASE WHEN r.%s IS NULL THEN l.%s ELSE r.%s END) AS %s", qt("unix_milli"), qt(c), qt(c), c))
+			}
+			return strings.Join(values, ",")
+		},
+	}).Parse(CDC_JOIN_QUERY))
+
+	buf := bytes.NewBuffer(nil)
+	if err := t.Execute(buf, params); err != nil {
+		return "", err
+	}
+	return buf.String(), nil
+}
diff --git a/internal/database/offline/sqlutil/serialize.go b/internal/database/offline/sqlutil/serialize.go
new file mode 100644
index 00000000..dac79f77
--- /dev/null
+++ b/internal/database/offline/sqlutil/serialize.go
@@ -0,0 +1,80 @@
+package sqlutil
+
+import (
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/spf13/cast"
+)
+
+func DeserializeByValueType(i interface{}, valueType types.ValueType, backend types.BackendType) (interface{}, error) {
+	if backend == types.BackendSnowflake {
+		return deserializeByTagForSnowflake(i, valueType)
+	}
+
+	switch valueType {
+	case types.Bool:
+		s := cast.ToString(i)
+		if s == "1" || s == "true" {
+			return true, nil
+		} else if s == "0" || s == "false" {
+			return false, nil
+		}
+		return nil, fmt.Errorf("invalid bool value %v", i)
+	case types.String:
+		return cast.ToString(i), nil
+	default:
+		return i, nil
+	}
+}
+
+// gosnowflake Scan always produce string when the destination is interface{}
+// See https://github.com/snowflakedb/gosnowflake/issues/517
+// As a work around, we cast the string to interface{} based on ValueType
+// This method is mostly copied from redis.DeserializeByTag, except we use 10 rather than 36 as the base
+// TODO: we should let the snowflake team fix the gosnowflake converter
+func deserializeByTagForSnowflake(i interface{}, valueType types.ValueType) (interface{}, error) {
+	if i == nil {
+		return nil, nil
+	}
+	if valueType == types.Bool {
+		return i, nil
+	}
+
+	s, ok := i.(string)
+	if !ok {
+		return nil, fmt.Errorf("not a string or nil: %v", i)
+	}
+
+	switch valueType {
+	case types.String:
+		return s, nil
+
+	case types.Int64:
+		x, err := strconv.ParseInt(s, 10, 64)
+		return x, err
+
+	case types.Float64:
+		x, err := strconv.ParseFloat(s, 64)
+		return x, err
+
+	case types.Bool:
+		if s == "1" {
+			return true, nil
+		} else if s == "0" {
+			return false, nil
+		} else {
+			return nil, fmt.Errorf("invalid bool value: %s", s)
+		}
+	case types.Time:
+		x, err := strconv.ParseInt(s, 10, 64)
+		return time.UnixMilli(x), err
+
+	case types.Bytes:
+		return []byte(s), nil
+	default:
+		return "", fmt.Errorf("unsupported value type: %s", valueType)
+	}
+}
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index 11d3eea4..edab9296 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -24,6 +24,7 @@ type JoinOpt struct {
 
 type JoinOneGroupOpt struct {
 	GroupName           string
+	Category            types.Category
 	Features            types.FeatureList
 	RevisionRanges      []*metadata.RevisionRange
 	Entity              types.Entity


Test patch:
diff --git a/internal/database/dbutil/schema_test.go b/internal/database/dbutil/schema_test.go
index db437a19..a855beb9 100644
--- a/internal/database/dbutil/schema_test.go
+++ b/internal/database/dbutil/schema_test.go
@@ -42,7 +42,7 @@ func TestBuildSchema(t *testing.T) {
 
 			want: `
 CREATE TABLE user (
-	"user_id" VARCHAR(32) PRIMARY KEY,
+	"user_id" VARCHAR(32),
 	"age" bigint,
 	"gender" text
 )`,
@@ -60,7 +60,7 @@ CREATE TABLE user (
 
 			want: "\n" +
 				"CREATE TABLE user (\n" +
-				"	`user_id` VARCHAR(32) PRIMARY KEY,\n" +
+				"	`user_id` VARCHAR(32),\n" +
 				"	`age` bigint,\n" +
 				"	`gender` text\n)",
 			wantErr: nil,
@@ -77,7 +77,7 @@ CREATE TABLE user (
 			wantErr: nil,
 			want: `
 CREATE TABLE user (
-	"user_id" TEXT PRIMARY KEY,
+	"user_id" TEXT,
 	"age" bigint,
 	"gender" text
 )`,
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index 41312658..5dea68b8 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -4,9 +4,12 @@ import (
 	"bufio"
 	"context"
 	"math"
+	"sort"
 	"strings"
 	"testing"
 
+	"github.com/spf13/cast"
+
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
@@ -25,28 +28,45 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 		Name:   "device",
 		Length: 10,
 	}
+	unixMilli := &types.Feature{
+		Name:      "unix_milli",
+		ValueType: types.Int64,
+	}
 	oneGroupFeatures, oneGroupFeatureMap := prepareFeatures(true)
 	twoGroupFeatures, twoGroupFeatureMap := prepareFeatures(false)
 
-	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_1_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_snapshot_1_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
 1235,apple,200
 `)),
 		Delimiter: ",",
 	})
-	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_1_2", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_snapshot_1_2", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,galaxy,300
 1235,oneplus,240
 `)),
 		Delimiter: ",",
 	})
-	buildTestSnapshotTable(ctx, t, store, twoGroupFeatures[2:], "offline_2_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, twoGroupFeatures[2:], "offline_snapshot_2_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,1
 1235,0
 `)),
 		Delimiter: ",",
 	})
 
+	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), "offline_cdc_1_1", &offline.CSVSource{
+		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi-1,120,2
+1235,apple-2,115,14
+1234,xiaomi-1,130,10
+`)),
+		Delimiter: ",",
+	})
+	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), "offline_cdc_1_2", &offline.CSVSource{
+		Reader: bufio.NewReader(strings.NewReader(`1234,galaxy-1,320,18
+`)),
+		Delimiter: ",",
+	})
+
 	testCases := []struct {
 		description string
 		opt         offline.JoinOpt
@@ -69,35 +89,46 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 			expected: nil,
 		},
 		{
-			description: "one feature group",
+			description: "one batch feature group",
 			opt: offline.JoinOpt{
 				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, false),
 				FeatureMap:       oneGroupFeatureMap,
-				RevisionRangeMap: prepareRevisionRanges(true),
+				RevisionRangeMap: prepareRevisionRanges(true, false),
 			},
-			expected: prepareResult(true, false),
+			expected: prepareResult(true, false, prepareBatchResultValues()),
 		},
 		{
-			description: "two feature groups",
+			description: "two batch feature groups",
 			opt: offline.JoinOpt{
 				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, false),
 				FeatureMap:       twoGroupFeatureMap,
-				RevisionRangeMap: prepareRevisionRanges(false),
+				RevisionRangeMap: prepareRevisionRanges(false, false),
+			},
+			expected: prepareResult(false, false, prepareBatchResultValues()),
+		},
+		{
+			description: "two batch feature groups, with extra values",
+			opt: offline.JoinOpt{
+				Entity:           *entity,
+				EntityRows:       prepareEntityRows(false, true),
+				FeatureMap:       twoGroupFeatureMap,
+				RevisionRangeMap: prepareRevisionRanges(false, false),
+				ValueNames:       []string{"value_1", "value_2"},
 			},
-			expected: prepareResult(false, false),
+			expected: prepareResult(false, true, prepareBatchResultValues()),
 		},
 		{
-			description: "two feature groups, with extra values",
+			description: "one streaming feature group, one batch group",
 			opt: offline.JoinOpt{
 				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, true),
 				FeatureMap:       twoGroupFeatureMap,
-				RevisionRangeMap: prepareRevisionRanges(false),
+				RevisionRangeMap: prepareRevisionRanges(false, true),
 				ValueNames:       []string{"value_1", "value_2"},
 			},
-			expected: prepareResult(false, true),
+			expected: prepareResult(false, true, prepareStreamingResultValues()),
 		},
 	}
 
@@ -108,10 +139,12 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 			if tc.expected == nil {
 				assert.Equal(t, tc.expected, actual)
 			} else {
+				assert.ElementsMatch(t, tc.expected.Header, actual.Header)
 				expectedValues := extractValues(tc.expected.Data)
 				actualValues := extractValues(actual.Data)
-				assert.ElementsMatch(t, tc.expected.Header, actual.Header)
-				assert.ObjectsAreEqual(expectedValues, actualValues)
+				for i := range expectedValues {
+					assert.ElementsMatch(t, expectedValues[i], actualValues[i])
+				}
 			}
 		})
 	}
@@ -157,31 +190,36 @@ func prepareFeatures(oneGroup bool) (types.FeatureList, map[string]types.Feature
 	}
 }
 
-func prepareRevisionRanges(oneGroup bool) map[string][]*metadata.RevisionRange {
+func prepareRevisionRanges(oneGroup bool, stream bool) map[string][]*metadata.RevisionRange {
 	basic := []*metadata.RevisionRange{
 		{
 			MinRevision:   1,
 			MaxRevision:   15,
-			SnapshotTable: "offline_1_1",
+			SnapshotTable: "offline_snapshot_1_1",
 		},
 		{
 			MinRevision:   15,
 			MaxRevision:   25,
-			SnapshotTable: "offline_1_2",
+			SnapshotTable: "offline_snapshot_1_2",
 		},
 	}
 	advanced := []*metadata.RevisionRange{
 		{
 			MinRevision:   5,
 			MaxRevision:   math.MaxInt64,
-			SnapshotTable: "offline_2_1",
+			SnapshotTable: "offline_snapshot_2_1",
 		},
 	}
+	if stream {
+		basic[0].CdcTable = "offline_cdc_1_1"
+		basic[1].CdcTable = "offline_cdc_1_2"
+	}
 	if oneGroup {
 		return map[string][]*metadata.RevisionRange{
 			"device_basic": basic,
 		}
 	}
+
 	return map[string][]*metadata.RevisionRange{
 		"device_basic":    basic,
 		"device_advanced": advanced,
@@ -227,12 +265,81 @@ func prepareEntityRows(isEmpty bool, withValue bool) <-chan types.EntityRow {
 	return entityRows
 }
 
-func prepareResult(oneGroup bool, withValue bool) *types.JoinResult {
+func prepareResult(oneGroup bool, withValue bool, values []map[string]interface{}) *types.JoinResult {
 	header := []string{"entity_key", "unix_milli", "device_basic.model", "device_basic.price", "device_advanced.is_active"}
 	if withValue {
 		header = []string{"entity_key", "unix_milli", "value_1", "value_2", "device_basic.model", "device_basic.price", "device_advanced.is_active"}
 	}
-	values := []map[string]interface{}{
+	if oneGroup {
+		header = header[:len(header)-1]
+	}
+	if !withValue {
+		for _, m := range values {
+			delete(m, "value_1")
+			delete(m, "value_2")
+		}
+	}
+
+	data := make(chan []interface{})
+	go func() {
+		defer close(data)
+		for _, item := range values {
+			record := make([]interface{}, 0, len(header))
+			for _, h := range header {
+				record = append(record, item[h])
+			}
+			data <- record
+		}
+	}()
+	return &types.JoinResult{
+		Header: header,
+		Data:   data,
+	}
+}
+
+func prepareStreamingResultValues() []map[string]interface{} {
+	return []map[string]interface{}{
+		{
+			"entity_key":                "1234",
+			"unix_milli":                int64(10),
+			"value_1":                   "1",
+			"value_2":                   "2",
+			"device_basic.model":        "xiaomi-1",
+			"device_basic.price":        int64(130),
+			"device_advanced.is_active": true,
+		},
+		{
+			"entity_key":                "1234",
+			"unix_milli":                int64(20),
+			"value_1":                   "3",
+			"value_2":                   "4",
+			"device_basic.model":        "galaxy-1",
+			"device_basic.price":        int64(320),
+			"device_advanced.is_active": true,
+		},
+		{
+			"entity_key":                "1235",
+			"unix_milli":                int64(5),
+			"value_1":                   "5",
+			"value_2":                   "6",
+			"device_basic.model":        "apple",
+			"device_basic.price":        int64(200),
+			"device_advanced.is_active": false,
+		},
+		{
+			"entity_key":                "1235",
+			"unix_milli":                int64(14),
+			"value_1":                   "7",
+			"value_2":                   "8",
+			"device_basic.model":        "apple-2",
+			"device_basic.price":        int64(115),
+			"device_advanced.is_active": false,
+		},
+	}
+}
+
+func prepareBatchResultValues() []map[string]interface{} {
+	return []map[string]interface{}{
 		{
 			"entity_key":                "1234",
 			"unix_milli":                int64(10),
@@ -270,31 +377,6 @@ func prepareResult(oneGroup bool, withValue bool) *types.JoinResult {
 			"device_advanced.is_active": false,
 		},
 	}
-	if oneGroup {
-		header = header[:len(header)-1]
-	}
-	if !withValue {
-		for _, m := range values {
-			delete(m, "value_1")
-			delete(m, "value_2")
-		}
-	}
-
-	data := make(chan []interface{})
-	go func() {
-		defer close(data)
-		for _, item := range values {
-			record := make([]interface{}, 0, len(header))
-			for _, h := range header {
-				record = append(record, item[h])
-			}
-			data <- record
-		}
-	}()
-	return &types.JoinResult{
-		Header: header,
-		Data:   data,
-	}
 }
 
 func extractValues(stream <-chan []interface{}) [][]interface{} {
@@ -302,5 +384,11 @@ func extractValues(stream <-chan []interface{}) [][]interface{} {
 	for item := range stream {
 		values = append(values, item)
 	}
+	sort.Slice(values, func(i, j int) bool {
+		if cast.ToString(values[i][0]) == cast.ToString(values[j][0]) {
+			return cast.ToInt(values[i][1]) < cast.ToInt(values[j][1])
+		}
+		return cast.ToString(values[i][0]) < cast.ToString(values[j][0])
+	})
 	return values
 }