Repository: oom-ai/oomstore
Pull request #1078 (2022-01-25 08:36:02)

Jinghan/simplify online methods options
This PR simplifies options of online methods: change `Entity` to `EntityName`, `Revision` to `RevisionID`, etc..




Patch:
diff --git a/internal/database/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index 3ff2af86..4ef995e2 100644
--- a/internal/database/online/cassandra/get.go
+++ b/internal/database/online/cassandra/get.go
@@ -25,7 +25,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`,
 		strings.Join(opt.Features.Names(), ","),
 		tableName,
-		opt.Entity.Name,
+		opt.Group.Entity.Name,
 	)
 
 	scan := make(map[string]interface{})
@@ -56,12 +56,12 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	} else {
 		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 	}
-
+	entityName := opt.Group.Entity.Name
 	query := fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s in (%s)`,
-		opt.Entity.Name,
+		entityName,
 		strings.Join(opt.Features.Names(), ","),
 		tableName,
-		opt.Entity.Name,
+		entityName,
 		placeholders,
 	)
 
@@ -77,7 +77,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	}
 
 	for _, s := range scan {
-		entityKey, value := deserializeIntoRowMap(s, opt.Entity.Name, opt.Features)
+		entityKey, value := deserializeIntoRowMap(s, entityName, opt.Features)
 		rs[entityKey] = value
 	}
 	return rs, nil
diff --git a/internal/database/online/cassandra/import.go b/internal/database/online/cassandra/import.go
index 3f40cbd0..4004f089 100644
--- a/internal/database/online/cassandra/import.go
+++ b/internal/database/online/cassandra/import.go
@@ -18,7 +18,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	// Step 0: drop existing table for streaming feature
 	var tableName string
 	if opt.Group.Category == types.CategoryBatch {
-		tableName = sqlutil.OnlineBatchTableName(opt.Revision.ID)
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
 	} else {
 		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 		if err := db.Query(fmt.Sprintf(`DROP TABLE IF EXISTS %s;`, tableName)).Exec(); err != nil {
diff --git a/internal/database/online/cassandra/stream.go b/internal/database/online/cassandra/stream.go
index 3feb201f..e8cdfff5 100644
--- a/internal/database/online/cassandra/stream.go
+++ b/internal/database/online/cassandra/stream.go
@@ -14,7 +14,7 @@ func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTa
 	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
 
 	if opt.Feature == nil {
-		schema, err := sqlutil.CreateStreamTableSchema(ctx, tableName, opt.Entity, Backend)
+		schema, err := sqlutil.CreateStreamTableSchema(ctx, tableName, opt.EntityName, Backend)
 		if err != nil {
 			return err
 		}
diff --git a/internal/database/online/dynamodb/get.go b/internal/database/online/dynamodb/get.go
index 15ebc97e..e737c6c8 100644
--- a/internal/database/online/dynamodb/get.go
+++ b/internal/database/online/dynamodb/get.go
@@ -35,7 +35,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 	result, err := db.GetItem(ctx, &dynamodb.GetItemInput{
 		TableName: aws.String(tableName),
 		Key: map[string]types.AttributeValue{
-			opt.Entity.Name: entityKeyValue,
+			opt.Group.Entity.Name: entityKeyValue,
 		},
 	})
 	if err != nil {
@@ -56,6 +56,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 	}
 
+	entityName := opt.Group.Entity.Name
 	res := make(map[string]dbutil.RowMap)
 	keys := make([]map[string]types.AttributeValue, 0, BatchGetItemCapacity)
 	for _, entityKey := range opt.EntityKeys {
@@ -64,16 +65,16 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 			return nil, errdefs.WithStack(err)
 		}
 		keys = append(keys, map[string]types.AttributeValue{
-			opt.Entity.Name: entityKeyValue,
+			entityName: entityKeyValue,
 		})
 		if len(keys) == BatchGetItemCapacity {
-			if err = batchGetItem(ctx, db, keys, tableName, opt.Entity.Name, opt.Features, res); err != nil {
+			if err = batchGetItem(ctx, db, keys, tableName, entityName, opt.Features, res); err != nil {
 				return nil, err
 			}
 			keys = make([]map[string]types.AttributeValue, 0, BatchGetItemCapacity)
 		}
 	}
-	if err := batchGetItem(ctx, db, keys, tableName, opt.Entity.Name, opt.Features, res); err != nil {
+	if err := batchGetItem(ctx, db, keys, tableName, entityName, opt.Features, res); err != nil {
 		return nil, err
 	}
 	return res, nil
diff --git a/internal/database/online/dynamodb/import.go b/internal/database/online/dynamodb/import.go
index f05392b2..c765c73d 100644
--- a/internal/database/online/dynamodb/import.go
+++ b/internal/database/online/dynamodb/import.go
@@ -23,7 +23,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	// Step 0: clean up existing table for streaming feature
 	var tableName string
 	if opt.Group.Category == oomTypes.CategoryBatch {
-		tableName = sqlutil.OnlineBatchTableName(opt.Revision.ID)
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
 	} else {
 		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 		_, err := db.DeleteTable(ctx, &dynamodb.DeleteTableInput{
diff --git a/internal/database/online/dynamodb/stream.go b/internal/database/online/dynamodb/stream.go
index ea9d90a8..be2995c8 100644
--- a/internal/database/online/dynamodb/stream.go
+++ b/internal/database/online/dynamodb/stream.go
@@ -27,13 +27,13 @@ func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTa
 		TableName: aws.String(tableName),
 		KeySchema: []types.KeySchemaElement{
 			{
-				AttributeName: aws.String(opt.Entity.Name),
+				AttributeName: aws.String(opt.EntityName),
 				KeyType:       types.KeyTypeHash,
 			},
 		},
 		AttributeDefinitions: []types.AttributeDefinition{
 			{
-				AttributeName: aws.String(opt.Entity.Name),
+				AttributeName: aws.String(opt.EntityName),
 				AttributeType: types.ScalarAttributeTypeS,
 			},
 		},
@@ -55,7 +55,7 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	if err != nil {
 		return errdefs.WithStack(err)
 	}
-	item[opt.Entity.Name] = entityKeyValue
+	item[opt.EntityName] = entityKeyValue
 
 	for i, feature := range opt.Features {
 		value, err := dbutil.SerializeByValueType(opt.FeatureValues[i], feature.ValueType, Backend)
diff --git a/internal/database/online/mysql/store.go b/internal/database/online/mysql/store.go
index 6d88c390..23a9cab7 100644
--- a/internal/database/online/mysql/store.go
+++ b/internal/database/online/mysql/store.go
@@ -66,5 +66,5 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
-	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
+	return sqlutil.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index 1c36a73c..cbeb7490 100644
--- a/internal/database/online/postgres/store.go
+++ b/internal/database/online/postgres/store.go
@@ -58,7 +58,7 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 		tableName,
 		cond.Inserts,
 		cond.InsertPlaceholders,
-		opt.Entity.Name,
+		opt.EntityName,
 		cond.UpdatePlaceholders,
 	)
 
@@ -67,5 +67,5 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
-	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
+	return sqlutil.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/redis/get.go b/internal/database/online/redis/get.go
index c6fc357f..0f90ae03 100644
--- a/internal/database/online/redis/get.go
+++ b/internal/database/online/redis/get.go
@@ -58,11 +58,10 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	res := make(map[string]dbutil.RowMap)
 	for _, entityKey := range opt.EntityKeys {
 		rowMap, err := db.Get(ctx, online.GetOpt{
-			Entity:     opt.Entity,
-			RevisionID: opt.RevisionID,
-			Group:      opt.Group,
 			EntityKey:  entityKey,
+			Group:      opt.Group,
 			Features:   opt.Features,
+			RevisionID: opt.RevisionID,
 		})
 		if err != nil {
 			return res, errdefs.WithStack(err)
diff --git a/internal/database/online/redis/import.go b/internal/database/online/redis/import.go
index 44b316b0..b4c9dbf2 100644
--- a/internal/database/online/redis/import.go
+++ b/internal/database/online/redis/import.go
@@ -24,7 +24,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 		var key string
 		var err error
 		if opt.Group.Category == types.CategoryBatch {
-			key, err = serializeRedisKeyForBatchFeature(opt.Revision.ID, entityKey)
+			key, err = serializeRedisKeyForBatchFeature(*opt.RevisionID, entityKey)
 		} else {
 			key, err = serializeRedisKeyForStreamFeature(opt.Group.ID, entityKey)
 		}
diff --git a/internal/database/online/sqlite/store.go b/internal/database/online/sqlite/store.go
index 02640d4f..27b93ca3 100644
--- a/internal/database/online/sqlite/store.go
+++ b/internal/database/online/sqlite/store.go
@@ -64,5 +64,5 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
-	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
+	return sqlutil.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index da9357fb..4f3e4430 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -21,9 +21,10 @@ func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.Back
 		tableName = OnlineStreamTableName(opt.Group.ID)
 	}
 
+	entityName := opt.Group.Entity.Name
 	featureNames := opt.Features.Names()
 	qt := dbutil.QuoteFn(backend)
-	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`, qt(featureNames...), qt(tableName), qt(opt.Entity.Name))
+	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`, qt(featureNames...), qt(tableName), qt(entityName))
 
 	record, err := db.QueryRowxContext(ctx, db.Rebind(query), opt.EntityKey).SliceScan()
 	if err != nil {
@@ -49,9 +50,10 @@ func MultiGet(ctx context.Context, db *sqlx.DB, opt online.MultiGetOpt, backend
 		tableName = OnlineStreamTableName(opt.Group.ID)
 	}
 
+	entityName := opt.Group.Entity.Name
 	featureNames := opt.Features.Names()
 	qt := dbutil.QuoteFn(backend)
-	query := fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s in (?);`, qt(opt.Entity.Name), qt(featureNames...), qt(tableName), qt(opt.Entity.Name))
+	query := fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s in (?);`, qt(entityName), qt(featureNames...), qt(tableName), qt(entityName))
 	sql, args, err := sqlx.In(query, opt.EntityKeys)
 	if err != nil {
 		return nil, errdefs.WithStack(err)
diff --git a/internal/database/online/sqlutil/import.go b/internal/database/online/sqlutil/import.go
index 26dc2010..70153ed3 100644
--- a/internal/database/online/sqlutil/import.go
+++ b/internal/database/online/sqlutil/import.go
@@ -18,7 +18,7 @@ func Import(ctx context.Context, db *sqlx.DB, opt online.ImportOpt, backend type
 	if opt.Group.Category == types.CategoryStream {
 		tableName = dbutil.TempTable("online_stream")
 	} else {
-		tableName = OnlineBatchTableName(opt.Revision.ID)
+		tableName = OnlineBatchTableName(*opt.RevisionID)
 	}
 	entity := opt.Group.Entity
 	columns := append([]string{entity.Name}, opt.Features.Names()...)
diff --git a/internal/database/online/sqlutil/push.go b/internal/database/online/sqlutil/push.go
index 6d2be354..a48b4a7f 100644
--- a/internal/database/online/sqlutil/push.go
+++ b/internal/database/online/sqlutil/push.go
@@ -21,7 +21,7 @@ func BuildPushCondition(opt online.PushOpt, backend types.BackendType) *PushCond
 	qt := dbutil.QuoteFn(backend)
 	cond := PushCondition{}
 
-	cond.Inserts = qt(append([]string{opt.Entity.Name}, opt.Features.Names()...)...)
+	cond.Inserts = qt(append([]string{opt.EntityName}, opt.Features.Names()...)...)
 	cond.InsertValues = append([]interface{}{opt.EntityKey}, opt.FeatureValues...)
 	cond.InsertPlaceholders = dbutil.Fill(len(cond.InsertValues), "?", ",")
 
diff --git a/internal/database/online/sqlutil/util.go b/internal/database/online/sqlutil/util.go
index 2808cadf..5cd68faf 100644
--- a/internal/database/online/sqlutil/util.go
+++ b/internal/database/online/sqlutil/util.go
@@ -20,15 +20,15 @@ func OnlineStreamTableName(groupID int) string {
 	return fmt.Sprintf("online_stream_%d", groupID)
 }
 
-func CreateStreamTableSchema(ctx context.Context, tableName string, entity *types.Entity, backend types.BackendType) (string, error) {
+func CreateStreamTableSchema(ctx context.Context, tableName, entityName string, backend types.BackendType) (string, error) {
 	var (
 		entityFormat string
 	)
 	switch backend {
 	case types.BackendMySQL:
-		entityFormat = fmt.Sprintf("`%s` VARCHAR(255) PRIMARY KEY", entity.Name)
+		entityFormat = fmt.Sprintf("`%s` VARCHAR(255) PRIMARY KEY", entityName)
 	case types.BackendPostgres, types.BackendCassandra, types.BackendSQLite:
-		entityFormat = fmt.Sprintf(`"%s" TEXT PRIMARY KEY`, entity.Name)
+		entityFormat = fmt.Sprintf(`"%s" TEXT PRIMARY KEY`, entityName)
 	default:
 		return "", errdefs.InvalidAttribute(errdefs.Errorf("backend %s not support", backend))
 	}
@@ -37,11 +37,11 @@ func CreateStreamTableSchema(ctx context.Context, tableName string, entity *type
 	return schema, nil
 }
 
-func SqlxPrapareStreamTable(ctx context.Context, db *sqlx.DB, opt online.PrepareStreamTableOpt, backend types.BackendType) error {
+func SqlxPrepareStreamTable(ctx context.Context, db *sqlx.DB, opt online.PrepareStreamTableOpt, backend types.BackendType) error {
 	tableName := OnlineStreamTableName(opt.GroupID)
 
 	if opt.Feature == nil {
-		schema, err := CreateStreamTableSchema(ctx, tableName, opt.Entity, backend)
+		schema, err := CreateStreamTableSchema(ctx, tableName, opt.EntityName, backend)
 		if err != nil {
 			return err
 		}
diff --git a/internal/database/online/tikv/get.go b/internal/database/online/tikv/get.go
index 81dab74c..9e03fa13 100644
--- a/internal/database/online/tikv/get.go
+++ b/internal/database/online/tikv/get.go
@@ -18,11 +18,10 @@ type input struct {
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
 	// Proxy to MultiGet
 	res, err := db.MultiGet(ctx, online.MultiGetOpt{
-		Entity:     opt.Entity,
-		Group:      opt.Group,
-		RevisionID: opt.RevisionID,
 		EntityKeys: []string{opt.EntityKey},
+		Group:      opt.Group,
 		Features:   opt.Features,
+		RevisionID: opt.RevisionID,
 	})
 	if err != nil {
 		return nil, err
@@ -36,14 +35,11 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
 	var (
-		allBatch            bool
 		serializedPrefixKey string
 		err                 error
 	)
 
-	allBatch = (opt.Group.Category == types.CategoryBatch)
-
-	if allBatch {
+	if opt.Group.Category == types.CategoryBatch {
 		serializedPrefixKey, err = dbutil.SerializeByValue(*opt.RevisionID, Backend)
 	} else {
 		serializedPrefixKey, err = dbutil.SerializeByValue(opt.Group.ID, Backend)
@@ -74,7 +70,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	var keys [][]byte
 	for _, serializedEntityKey := range serializedEntityKeys {
 		for _, serializedFeatureID := range serializedFeatureIDs {
-			if allBatch {
+			if opt.Group.Category == types.CategoryBatch {
 				keys = append(keys, getKeyOfBatchFeature(serializedPrefixKey, serializedEntityKey, serializedFeatureID))
 			} else {
 				keys = append(keys, getKeyOfStreamFeature(serializedPrefixKey, serializedEntityKey, serializedFeatureID))
diff --git a/internal/database/online/tikv/import.go b/internal/database/online/tikv/import.go
index 8e7b3f71..30e9a1fa 100644
--- a/internal/database/online/tikv/import.go
+++ b/internal/database/online/tikv/import.go
@@ -14,7 +14,7 @@ const BatchSize = 100
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	var seq int64
 
-	serializedRevisionID, err := dbutil.SerializeByValue(opt.Revision.ID, Backend)
+	serializedRevisionID, err := dbutil.SerializeByValue(*opt.RevisionID, Backend)
 	if err != nil {
 		return err
 	}
diff --git a/internal/database/online/types.go b/internal/database/online/types.go
index 21b2213c..1997136b 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -5,31 +5,29 @@ import (
 )
 
 type GetOpt struct {
-	Entity     *types.Entity
 	EntityKey  string
-	RevisionID *int
-	Group      *types.Group
+	Group      types.Group
 	Features   types.FeatureList
+	RevisionID *int
 }
 
 type MultiGetOpt struct {
-	Entity     *types.Entity
 	EntityKeys []string
-	RevisionID *int
-	Group      *types.Group
+	Group      types.Group
 	Features   types.FeatureList
+	RevisionID *int
 }
 
 type ImportOpt struct {
 	Group        types.Group
 	Features     types.FeatureList
-	Revision     *types.Revision
 	ExportStream <-chan types.ExportRecord
 	ExportError  <-chan error
+	RevisionID   *int
 }
 
 type PushOpt struct {
-	Entity        *types.Entity
+	EntityName    string
 	EntityKey     string
 	GroupID       int
 	Features      types.FeatureList
@@ -37,9 +35,8 @@ type PushOpt struct {
 }
 
 type PrepareStreamTableOpt struct {
-	Entity *types.Entity
-
-	GroupID int
+	EntityName string
+	GroupID    int
 
 	// Feature is not nil to add a new row to the stream table;
 	// otherwise it means the stream table will be created.
diff --git a/pkg/oomstore/apply.go b/pkg/oomstore/apply.go
index 2d749f79..709b5197 100644
--- a/pkg/oomstore/apply.go
+++ b/pkg/oomstore/apply.go
@@ -122,8 +122,8 @@ func (s *OomStore) applyGroup(ctx context.Context, tx metadata.DBStore, newGroup
 		if newGroup.Category == types.CategoryStream {
 			return func() error {
 				return s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-					Entity:  entity,
-					GroupID: id,
+					EntityName: entity.Name,
+					GroupID:    id,
 				})
 			}, nil
 		}
@@ -198,9 +198,9 @@ func (s *OomStore) applyFeature(ctx context.Context, tx metadata.DBStore, newFea
 			}
 			return func() error {
 				return s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-					Entity:  group.Entity,
-					GroupID: group.ID,
-					Feature: feature,
+					EntityName: group.Entity.Name,
+					GroupID:    group.ID,
+					Feature:    feature,
 				})
 			}, nil
 		}
diff --git a/pkg/oomstore/feature.go b/pkg/oomstore/feature.go
index e88d5020..098a8524 100644
--- a/pkg/oomstore/feature.go
+++ b/pkg/oomstore/feature.go
@@ -101,10 +101,10 @@ func (s *OomStore) CreateFeature(ctx context.Context, opt types.CreateFeatureOpt
 		if err != nil {
 			return 0, err
 		}
-		if err := s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-			Entity:  group.Entity,
-			GroupID: group.ID,
-			Feature: feature,
+		if err = s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+			EntityName: group.Entity.Name,
+			GroupID:    group.ID,
+			Feature:    feature,
 		}); err != nil {
 			return 0, err
 		}
diff --git a/pkg/oomstore/group.go b/pkg/oomstore/group.go
index 1b96401d..a2939379 100644
--- a/pkg/oomstore/group.go
+++ b/pkg/oomstore/group.go
@@ -28,8 +28,8 @@ func (s *OomStore) CreateGroup(ctx context.Context, opt types.CreateGroupOpt) (i
 
 	if opt.Category == types.CategoryStream {
 		if err = s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-			Entity:  entity,
-			GroupID: id,
+			EntityName: entity.Name,
+			GroupID:    id,
 		}); err != nil {
 			return 0, err
 		}
diff --git a/pkg/oomstore/online_query.go b/pkg/oomstore/online_query.go
index d38b6860..94f03817 100644
--- a/pkg/oomstore/online_query.go
+++ b/pkg/oomstore/online_query.go
@@ -38,18 +38,15 @@ func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*type
 
 	featureMap := groupFeaturesByGroupID(features)
 
-	for _, features := range featureMap {
-		if len(features) == 0 {
+	for _, featureList := range featureMap {
+		if len(featureList) == 0 {
 			continue
 		}
-		revisionID := features[0].OnlineRevisionID()
-		group := features[0].Group
 		featureValues, err := s.online.Get(ctx, online.GetOpt{
-			Entity:     entity,
-			RevisionID: revisionID,
-			Group:      group,
 			EntityKey:  opt.EntityKey,
-			Features:   features,
+			Group:      *featureList[0].Group,
+			Features:   featureList,
+			RevisionID: featureList[0].OnlineRevisionID(),
 		})
 		if err != nil {
 			return nil, err
@@ -111,14 +108,11 @@ func (s *OomStore) getFeatureValueMap(ctx context.Context, entityKeys []string,
 		if len(features) == 0 {
 			continue
 		}
-		revisionID := features[0].OnlineRevisionID()
-		group := features[0].Group
 		featureValues, err := s.online.MultiGet(ctx, online.MultiGetOpt{
-			Entity:     entity,
-			RevisionID: revisionID,
-			Group:      group,
 			EntityKeys: entityKeys,
+			Group:      *features[0].Group,
 			Features:   features,
+			RevisionID: features[0].OnlineRevisionID(),
 		})
 		if err != nil {
 			return nil, err
diff --git a/pkg/oomstore/push.go b/pkg/oomstore/push.go
index bceb9462..c12a0bb5 100644
--- a/pkg/oomstore/push.go
+++ b/pkg/oomstore/push.go
@@ -33,7 +33,7 @@ func (s *OomStore) Push(ctx context.Context, opt types.PushOpt) error {
 	}
 
 	if err := s.online.Push(ctx, online.PushOpt{
-		Entity:        entity,
+		EntityName:    entity.Name,
 		EntityKey:     opt.EntityKey,
 		GroupID:       group.ID,
 		Features:      features,
diff --git a/pkg/oomstore/sync.go b/pkg/oomstore/sync.go
index c413f1c4..df144f1f 100644
--- a/pkg/oomstore/sync.go
+++ b/pkg/oomstore/sync.go
@@ -50,9 +50,9 @@ func (s *OomStore) Sync(ctx context.Context, opt types.SyncOpt) error {
 	if err = s.online.Import(ctx, online.ImportOpt{
 		Group:        *group,
 		Features:     features,
-		Revision:     revision,
 		ExportStream: exportResult.Data,
 		ExportError:  exportResult.GetErrorChannel(),
+		RevisionID:   &revision.ID,
 	}); err != nil {
 		return err
 	}


Test patch:
diff --git a/internal/database/online/test_impl/get.go b/internal/database/online/test_impl/get.go
index aeb1398f..bb4a9f26 100644
--- a/internal/database/online/test_impl/get.go
+++ b/internal/database/online/test_impl/get.go
@@ -17,11 +17,10 @@ func TestGetExisted(t *testing.T, prepareStore PrepareStoreFn, destroyStore Dest
 
 	for _, target := range s.Data {
 		opt := online.GetOpt{
-			Entity:     &s.Entity,
-			RevisionID: &s.Revision.ID,
-			Group:      s.Revision.Group,
-			Features:   s.Features,
 			EntityKey:  target.EntityKey(),
+			Group:      *s.Revision.Group,
+			Features:   s.Features,
+			RevisionID: &s.Revision.ID,
 		}
 		rs, err := store.Get(ctx, opt)
 		require.NoError(t, err)
@@ -41,11 +40,10 @@ func TestGetNotExistedEntityKey(t *testing.T, prepareStore PrepareStoreFn, destr
 	importSample(t, ctx, store, s)
 
 	rs, err := store.Get(ctx, online.GetOpt{
-		Entity:     &s.Entity,
-		RevisionID: &s.Revision.ID,
-		Group:      s.Revision.Group,
-		Features:   s.Features,
 		EntityKey:  "not-existed-key",
+		Group:      *s.Revision.Group,
+		Features:   s.Features,
+		RevisionID: &s.Revision.ID,
 	})
 	require.NoError(t, err)
 	require.Equal(t, 0, len(rs), "actual: %+v", rs)
@@ -64,11 +62,10 @@ func TestMultiGet(t *testing.T, prepareStore PrepareStoreFn, destroystore Destro
 		keys = append(keys, r.EntityKey())
 	}
 	rs, err := store.MultiGet(ctx, online.MultiGetOpt{
-		Entity:     &s.Entity,
-		RevisionID: &s.Revision.ID,
-		Group:      s.Revision.Group,
-		Features:   s.Features,
 		EntityKeys: keys,
+		Group:      *s.Revision.Group,
+		Features:   s.Features,
+		RevisionID: &s.Revision.ID,
 	})
 	require.NoError(t, err)
 	for _, record := range s.Data {
diff --git a/internal/database/online/test_impl/purge.go b/internal/database/online/test_impl/purge.go
index 88bf4ebc..17117473 100644
--- a/internal/database/online/test_impl/purge.go
+++ b/internal/database/online/test_impl/purge.go
@@ -19,11 +19,10 @@ func TestPurgeRemovesSpecifiedRevision(t *testing.T, prepareStore PrepareStoreFn
 
 	for _, record := range SampleMedium.Data {
 		rs, err := store.Get(ctx, online.GetOpt{
-			Entity:     &SampleMedium.Entity,
-			RevisionID: &SampleMedium.Revision.ID,
-			Group:      SampleMedium.Revision.Group,
 			EntityKey:  record.EntityKey(),
+			Group:      *SampleMedium.Revision.Group,
 			Features:   SampleMedium.Features,
+			RevisionID: &SampleMedium.Revision.ID,
 		})
 		require.NoError(t, err)
 		require.Equal(t, 0, len(rs))
@@ -42,11 +41,10 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T, prepareStore PrepareStoreFn
 
 	for _, record := range SampleSmall.Data {
 		rs, err := store.Get(ctx, online.GetOpt{
-			Entity:     &SampleSmall.Entity,
-			RevisionID: &SampleSmall.Revision.ID,
-			Group:      SampleSmall.Revision.Group,
 			EntityKey:  record.EntityKey(),
+			Group:      *SampleSmall.Revision.Group,
 			Features:   SampleSmall.Features,
+			RevisionID: &SampleSmall.Revision.ID,
 		})
 		require.NoError(t, err)
 		for i, f := range SampleSmall.Features {
diff --git a/internal/database/online/test_impl/stream.go b/internal/database/online/test_impl/stream.go
index 564c5a58..dd79d237 100644
--- a/internal/database/online/test_impl/stream.go
+++ b/internal/database/online/test_impl/stream.go
@@ -77,8 +77,8 @@ func TestPrepareStreamTable(t *testing.T, prepareStore PrepareStoreFn, destroySt
 	for _, group := range simpleStreamData.groups {
 		t.Run("create stream table", func(t *testing.T) {
 			err := store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-				Entity:  group.Entity,
-				GroupID: group.ID,
+				EntityName: group.Entity.Name,
+				GroupID:    group.ID,
 			})
 			assert.NoError(t, err, "create stream table failed: %v", err)
 		})
@@ -87,9 +87,9 @@ func TestPrepareStreamTable(t *testing.T, prepareStore PrepareStoreFn, destroySt
 	for _, feature := range simpleStreamData.features {
 		t.Run("stream table add column", func(t *testing.T) {
 			err := store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-				Entity:  feature.Entity(),
-				GroupID: feature.GroupID,
-				Feature: feature,
+				EntityName: feature.Entity().Name,
+				GroupID:    feature.GroupID,
+				Feature:    feature,
 			})
 			assert.NoError(t, err, "stream table add column failed: %v", err)
 		})
@@ -110,29 +110,28 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 	)
 
 	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-		Entity:  &entity,
-		GroupID: group.ID,
+		EntityName: entity.Name,
+		GroupID:    group.ID,
 	}))
 
 	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-		Entity:  &entity,
-		GroupID: group.ID,
-		Feature: feature1,
+		EntityName: entity.Name,
+		GroupID:    group.ID,
+		Feature:    feature1,
 	}))
 
 	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		Entity:        &entity,
+		EntityName:    entity.Name,
 		EntityKey:     "user1",
 		GroupID:       group.ID,
 		Features:      types.FeatureList{feature1},
 		FeatureValues: []interface{}{"post1"},
 	}))
 	rs, err := store.Get(ctx, online.GetOpt{
-		Entity:     &entity,
 		EntityKey:  "user1",
-		RevisionID: nil,
-		Group:      group,
+		Group:      *group,
 		Features:   types.FeatureList{feature1},
+		RevisionID: nil,
 	})
 	assert.NoError(t, err)
 	assert.Equal(t, map[string]interface{}{
@@ -140,18 +139,17 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 	}, rs)
 
 	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		Entity:        &entity,
+		EntityName:    entity.Name,
 		EntityKey:     "user1",
 		GroupID:       group.ID,
 		Features:      types.FeatureList{feature1},
 		FeatureValues: []interface{}{"post2"},
 	}))
 	rs, err = store.Get(ctx, online.GetOpt{
-		Entity:     &entity,
 		EntityKey:  "user1",
-		RevisionID: nil,
-		Group:      group,
+		Group:      *group,
 		Features:   types.FeatureList{feature1},
+		RevisionID: nil,
 	})
 	assert.NoError(t, err)
 	assert.Equal(t, map[string]interface{}{
@@ -159,24 +157,23 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 	}, rs)
 
 	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-		Entity:  &entity,
-		GroupID: group.ID,
-		Feature: feature2,
+		EntityName: entity.Name,
+		GroupID:    group.ID,
+		Feature:    feature2,
 	}))
 
 	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		Entity:        &entity,
+		EntityName:    entity.Name,
 		EntityKey:     "user1",
 		GroupID:       group.ID,
 		Features:      types.FeatureList{feature1, feature2},
 		FeatureValues: []interface{}{"post1", "post2"},
 	}))
 	rs, err = store.Get(ctx, online.GetOpt{
-		Entity:     &entity,
 		EntityKey:  "user1",
-		RevisionID: nil,
-		Group:      group,
+		Group:      *group,
 		Features:   types.FeatureList{feature1, feature2},
+		RevisionID: nil,
 	})
 	assert.NoError(t, err)
 	assert.Equal(t, map[string]interface{}{
diff --git a/internal/database/online/test_impl/util.go b/internal/database/online/test_impl/util.go
index c5135254..ef49fdc9 100644
--- a/internal/database/online/test_impl/util.go
+++ b/internal/database/online/test_impl/util.go
@@ -32,52 +32,53 @@ var SampleMedium Sample
 func init() {
 	rand.Seed(time.Now().UnixNano())
 	entity := types.Entity{ID: 5, Name: "user"}
-	group := types.Group{ID: 1, Category: types.CategoryBatch, Entity: &entity}
+	group1 := types.Group{ID: 1, Category: types.CategoryBatch, Entity: &entity}
+	group2 := types.Group{ID: 2, Category: types.CategoryBatch, Entity: &entity}
 	{
 		SampleSmall = Sample{
 			Entity: entity,
-			Group:  group,
+			Group:  group1,
 			Features: types.FeatureList{
 				&types.Feature{
 					ID:        1,
 					Name:      "age",
 					GroupID:   1,
-					Group:     &group,
+					Group:     &group1,
 					ValueType: types.Int64,
 				},
 				&types.Feature{
 					ID:        2,
 					Name:      "gender",
 					GroupID:   1,
-					Group:     &group,
+					Group:     &group1,
 					ValueType: types.String,
 				},
 				&types.Feature{
 					ID:        3,
 					Name:      "account",
 					GroupID:   1,
-					Group:     &group,
+					Group:     &group1,
 					ValueType: types.Float64,
 				},
 				&types.Feature{
 					ID:        4,
 					Name:      "is_active",
 					GroupID:   1,
-					Group:     &group,
+					Group:     &group1,
 					ValueType: types.Bool,
 				},
 				&types.Feature{
 					ID:        5,
 					Name:      "register_time",
 					GroupID:   1,
-					Group:     &group,
+					Group:     &group1,
 					ValueType: types.Time,
 				},
 			},
 			Revision: types.Revision{
 				ID:      3,
 				GroupID: 1,
-				Group:   &group,
+				Group:   &group1,
 			},
 			Data: []types.ExportRecord{
 				[]interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
@@ -94,19 +95,19 @@ func init() {
 				ID:        2,
 				Name:      "charge",
 				GroupID:   2,
-				Group:     &types.Group{ID: 2, Category: types.CategoryBatch},
+				Group:     &group2,
 				ValueType: types.Float64,
 			},
 		}
 
-		revision := types.Revision{ID: 9, GroupID: 2, Group: &types.Group{ID: 2, Category: types.CategoryBatch}}
+		revision := types.Revision{ID: 9, GroupID: 2, Group: &group2}
 		var data []types.ExportRecord
 
 		for i := 0; i < 100; i++ {
 			record := []interface{}{dbutil.RandString(10), rand.Float64()}
 			data = append(data, record)
 		}
-		SampleMedium = Sample{entity, group, features, revision, data}
+		SampleMedium = Sample{entity, group1, features, revision, data}
 	}
 }
 
@@ -123,7 +124,7 @@ func importSample(t *testing.T, ctx context.Context, store online.Store, samples
 		err := store.Import(ctx, online.ImportOpt{
 			Group:        sample.Group,
 			Features:     sample.Features,
-			Revision:     &sample.Revision,
+			RevisionID:   &sample.Revision.ID,
 			ExportStream: stream,
 		})
 		require.NoError(t, err)