Repository: oom-ai/oomstore (go)

Issue:
bug: errdefs.ErrNotFound(errdefs.WithStack(err)) not print call stack
this is because `errdefs.errNotFound` not implement interface `fmt.Format`

Pull request:
fix(oomstore/errdefs): fix wrapped error does not print out the call stack
resolve #1082 

relate [stack overflow](https://stackoverflow.com/questions/70851429/golang-pkg-errors-how-to-print-custom-wrapped-errors)

Patch:
diff --git a/internal/database/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index 4ef995e2..3ff2af86 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.Group.Entity.Name,
+		opt.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)`,
-		entityName,
+		opt.Entity.Name,
 		strings.Join(opt.Features.Names(), ","),
 		tableName,
-		entityName,
+		opt.Entity.Name,
 		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, entityName, opt.Features)
+		entityKey, value := deserializeIntoRowMap(s, opt.Entity.Name, 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 4004f089..3f40cbd0 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.RevisionID)
+		tableName = sqlutil.OnlineBatchTableName(opt.Revision.ID)
 	} 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 e8cdfff5..3feb201f 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.EntityName, Backend)
+		schema, err := sqlutil.CreateStreamTableSchema(ctx, tableName, opt.Entity, Backend)
 		if err != nil {
 			return err
 		}
diff --git a/internal/database/online/dynamodb/get.go b/internal/database/online/dynamodb/get.go
index e737c6c8..15ebc97e 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.Group.Entity.Name: entityKeyValue,
+			opt.Entity.Name: entityKeyValue,
 		},
 	})
 	if err != nil {
@@ -56,7 +56,6 @@ 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 {
@@ -65,16 +64,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{
-			entityName: entityKeyValue,
+			opt.Entity.Name: entityKeyValue,
 		})
 		if len(keys) == BatchGetItemCapacity {
-			if err = batchGetItem(ctx, db, keys, tableName, entityName, opt.Features, res); err != nil {
+			if err = batchGetItem(ctx, db, keys, tableName, opt.Entity.Name, opt.Features, res); err != nil {
 				return nil, err
 			}
 			keys = make([]map[string]types.AttributeValue, 0, BatchGetItemCapacity)
 		}
 	}
-	if err := batchGetItem(ctx, db, keys, tableName, entityName, opt.Features, res); err != nil {
+	if err := batchGetItem(ctx, db, keys, tableName, opt.Entity.Name, 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 c6ceb570..f05392b2 100644
--- a/internal/database/online/dynamodb/import.go
+++ b/internal/database/online/dynamodb/import.go
@@ -2,7 +2,6 @@ package dynamodb
 
 import (
 	"context"
-	"errors"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 
@@ -24,16 +23,14 @@ 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.RevisionID)
+		tableName = sqlutil.OnlineBatchTableName(opt.Revision.ID)
 	} else {
 		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 		_, err := db.DeleteTable(ctx, &dynamodb.DeleteTableInput{
 			TableName: aws.String(tableName),
 		})
 		if err != nil {
-			if apiErr := new(types.ResourceNotFoundException); !errors.As(err, &apiErr) {
-				return errdefs.WithStack(err)
-			}
+			return errdefs.WithStack(err)
 		}
 	}
 	// Step 1: create table
diff --git a/internal/database/online/dynamodb/stream.go b/internal/database/online/dynamodb/stream.go
index be2995c8..ea9d90a8 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.EntityName),
+				AttributeName: aws.String(opt.Entity.Name),
 				KeyType:       types.KeyTypeHash,
 			},
 		},
 		AttributeDefinitions: []types.AttributeDefinition{
 			{
-				AttributeName: aws.String(opt.EntityName),
+				AttributeName: aws.String(opt.Entity.Name),
 				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.EntityName] = entityKeyValue
+	item[opt.Entity.Name] = 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 23a9cab7..6d88c390 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.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index cbeb7490..1c36a73c 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.EntityName,
+		opt.Entity.Name,
 		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.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/redis/get.go b/internal/database/online/redis/get.go
index 0f90ae03..c6fc357f 100644
--- a/internal/database/online/redis/get.go
+++ b/internal/database/online/redis/get.go
@@ -58,10 +58,11 @@ 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{
-			EntityKey:  entityKey,
+			Entity:     opt.Entity,
+			RevisionID: opt.RevisionID,
 			Group:      opt.Group,
+			EntityKey:  entityKey,
 			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 b4c9dbf2..44b316b0 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.RevisionID, entityKey)
+			key, err = serializeRedisKeyForBatchFeature(opt.Revision.ID, 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 27b93ca3..02640d4f 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.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index 4f3e4430..da9357fb 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -21,10 +21,9 @@ 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(entityName))
+	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`, qt(featureNames...), qt(tableName), qt(opt.Entity.Name))
 
 	record, err := db.QueryRowxContext(ctx, db.Rebind(query), opt.EntityKey).SliceScan()
 	if err != nil {
@@ -50,10 +49,9 @@ 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(entityName), qt(featureNames...), qt(tableName), qt(entityName))
+	query := fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s in (?);`, qt(opt.Entity.Name), qt(featureNames...), qt(tableName), qt(opt.Entity.Name))
 	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 70153ed3..26dc2010 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.RevisionID)
+		tableName = OnlineBatchTableName(opt.Revision.ID)
 	}
 	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 a48b4a7f..6d2be354 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.EntityName}, opt.Features.Names()...)...)
+	cond.Inserts = qt(append([]string{opt.Entity.Name}, 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 5cd68faf..2808cadf 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, entityName string, backend types.BackendType) (string, error) {
+func CreateStreamTableSchema(ctx context.Context, tableName string, entity *types.Entity, backend types.BackendType) (string, error) {
 	var (
 		entityFormat string
 	)
 	switch backend {
 	case types.BackendMySQL:
-		entityFormat = fmt.Sprintf("`%s` VARCHAR(255) PRIMARY KEY", entityName)
+		entityFormat = fmt.Sprintf("`%s` VARCHAR(255) PRIMARY KEY", entity.Name)
 	case types.BackendPostgres, types.BackendCassandra, types.BackendSQLite:
-		entityFormat = fmt.Sprintf(`"%s" TEXT PRIMARY KEY`, entityName)
+		entityFormat = fmt.Sprintf(`"%s" TEXT PRIMARY KEY`, entity.Name)
 	default:
 		return "", errdefs.InvalidAttribute(errdefs.Errorf("backend %s not support", backend))
 	}
@@ -37,11 +37,11 @@ func CreateStreamTableSchema(ctx context.Context, tableName, entityName string,
 	return schema, nil
 }
 
-func SqlxPrepareStreamTable(ctx context.Context, db *sqlx.DB, opt online.PrepareStreamTableOpt, backend types.BackendType) error {
+func SqlxPrapareStreamTable(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.EntityName, backend)
+		schema, err := CreateStreamTableSchema(ctx, tableName, opt.Entity, backend)
 		if err != nil {
 			return err
 		}
diff --git a/internal/database/online/tikv/get.go b/internal/database/online/tikv/get.go
index 9e03fa13..81dab74c 100644
--- a/internal/database/online/tikv/get.go
+++ b/internal/database/online/tikv/get.go
@@ -18,10 +18,11 @@ 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{
-		EntityKeys: []string{opt.EntityKey},
+		Entity:     opt.Entity,
 		Group:      opt.Group,
-		Features:   opt.Features,
 		RevisionID: opt.RevisionID,
+		EntityKeys: []string{opt.EntityKey},
+		Features:   opt.Features,
 	})
 	if err != nil {
 		return nil, err
@@ -35,11 +36,14 @@ 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
 	)
 
-	if opt.Group.Category == types.CategoryBatch {
+	allBatch = (opt.Group.Category == types.CategoryBatch)
+
+	if allBatch {
 		serializedPrefixKey, err = dbutil.SerializeByValue(*opt.RevisionID, Backend)
 	} else {
 		serializedPrefixKey, err = dbutil.SerializeByValue(opt.Group.ID, Backend)
@@ -70,7 +74,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 opt.Group.Category == types.CategoryBatch {
+			if allBatch {
 				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 bba84207..8e7b3f71 100644
--- a/internal/database/online/tikv/import.go
+++ b/internal/database/online/tikv/import.go
@@ -3,8 +3,6 @@ package tikv
 import (
 	"context"
 
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
@@ -15,14 +13,8 @@ const BatchSize = 100
 
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	var seq int64
-	var err error
-	var serializedRevisionID, serializedGroupID string
 
-	if opt.Group.Category == types.CategoryBatch {
-		serializedRevisionID, err = dbutil.SerializeByValue(*opt.RevisionID, Backend)
-	} else {
-		serializedGroupID, err = dbutil.SerializeByValue(opt.Group.ID, Backend)
-	}
+	serializedRevisionID, err := dbutil.SerializeByValue(opt.Revision.ID, Backend)
 	if err != nil {
 		return err
 	}
@@ -62,11 +54,8 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 			if err != nil {
 				return err
 			}
-			if opt.Group.Category == types.CategoryBatch {
-				putKeys = append(putKeys, getKeyOfBatchFeature(serializedRevisionID, serializedEntityKey, serializedFeatureIDs[i]))
-			} else {
-				putKeys = append(putKeys, getKeyOfStreamFeature(serializedGroupID, serializedEntityKey, serializedFeatureIDs[i]))
-			}
+
+			putKeys = append(putKeys, getKeyOfBatchFeature(serializedRevisionID, serializedEntityKey, serializedFeatureIDs[i]))
 			putVals = append(putVals, []byte(serializedFeatureValue.(string)))
 		}
 
diff --git a/internal/database/online/types.go b/internal/database/online/types.go
index 1997136b..21b2213c 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -5,29 +5,31 @@ import (
 )
 
 type GetOpt struct {
+	Entity     *types.Entity
 	EntityKey  string
-	Group      types.Group
-	Features   types.FeatureList
 	RevisionID *int
+	Group      *types.Group
+	Features   types.FeatureList
 }
 
 type MultiGetOpt struct {
+	Entity     *types.Entity
 	EntityKeys []string
-	Group      types.Group
-	Features   types.FeatureList
 	RevisionID *int
+	Group      *types.Group
+	Features   types.FeatureList
 }
 
 type ImportOpt struct {
 	Group        types.Group
 	Features     types.FeatureList
+	Revision     *types.Revision
 	ExportStream <-chan types.ExportRecord
 	ExportError  <-chan error
-	RevisionID   *int
 }
 
 type PushOpt struct {
-	EntityName    string
+	Entity        *types.Entity
 	EntityKey     string
 	GroupID       int
 	Features      types.FeatureList
@@ -35,8 +37,9 @@ type PushOpt struct {
 }
 
 type PrepareStreamTableOpt struct {
-	EntityName string
-	GroupID    int
+	Entity *types.Entity
+
+	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/oomcli/cmd/util.go b/oomcli/cmd/util.go
index 3cf94758..db938acd 100644
--- a/oomcli/cmd/util.go
+++ b/oomcli/cmd/util.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/oom-ai/oomstore/pkg/oomstore"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -43,6 +44,11 @@ func groupsToApplyGroupItems(ctx context.Context, store *oomstore.OomStore, grou
 }
 
 func exitf(format string, a ...interface{}) {
+	if true {
+		format = strings.ReplaceAll(format, "{err}", "%+v")
+	} else {
+		format = strings.ReplaceAll(format, "{err}", "%v")
+	}
 	msg := fmt.Sprintf(format, a...)
 	fmt.Fprintf(os.Stderr, "Error: %s\n", msg)
 	os.Exit(1)
diff --git a/pkg/errdefs/errdefs.go b/pkg/errdefs/errdefs.go
index 55b86b8f..12d3eb3c 100644
--- a/pkg/errdefs/errdefs.go
+++ b/pkg/errdefs/errdefs.go
@@ -1,11 +1,19 @@
 package errdefs
 
+import (
+	"fmt"
+)
+
 type errNotFound struct{ error }
 
 func (e errNotFound) Unwrap() error {
 	return e.error
 }
 
+func (e errNotFound) Error() string {
+	return fmt.Sprintf("%+v", e.error)
+}
+
 func NotFound(err error) error {
 	if err == nil || IsNotFound(err) {
 		return err
@@ -19,6 +27,10 @@ func (e errInvalidAttribute) Unwrap() error {
 	return e.error
 }
 
+func (e errInvalidAttribute) Error() string {
+	return fmt.Sprintf("%+v", e.error)
+}
+
 func InvalidAttribute(err error) error {
 	if err == nil || IsInvalidAttribute(err) {
 		return err
diff --git a/pkg/errdefs/wrap.go b/pkg/errdefs/wrap.go
index c1b900ed..ac73632b 100644
--- a/pkg/errdefs/wrap.go
+++ b/pkg/errdefs/wrap.go
@@ -26,7 +26,7 @@ func Errorf(format string, args ...interface{}) error {
 }
 
 func WithStack(err error) error {
-	if !debug || hasStack(err) {
+	if !debug {
 		return err
 	}
 
@@ -36,35 +36,3 @@ func WithStack(err error) error {
 func Cause(err error) error {
 	return errors.Cause(err)
 }
-
-// not call callers() on every wrap, see https://github.com/pkg/errors/issues/75#issuecomment-574580408
-func hasStack(err error) bool {
-	if err == nil {
-		return false
-	}
-
-	type causer interface {
-		Cause() error
-	}
-
-	type tracer interface {
-		StackTrace() errors.StackTrace
-	}
-
-	e := err
-	for {
-		if _, ok := e.(tracer); ok {
-			return true
-		}
-
-		cause, ok := e.(causer)
-		if !ok {
-			return false
-		}
-
-		e = cause.Cause()
-		if e == nil {
-			return false
-		}
-	}
-}
diff --git a/pkg/oomstore/apply.go b/pkg/oomstore/apply.go
index 709b5197..2d749f79 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{
-					EntityName: entity.Name,
-					GroupID:    id,
+					Entity:  entity,
+					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{
-					EntityName: group.Entity.Name,
-					GroupID:    group.ID,
-					Feature:    feature,
+					Entity:  group.Entity,
+					GroupID: group.ID,
+					Feature: feature,
 				})
 			}, nil
 		}
diff --git a/pkg/oomstore/feature.go b/pkg/oomstore/feature.go
index 098a8524..e88d5020 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{
-			EntityName: group.Entity.Name,
-			GroupID:    group.ID,
-			Feature:    feature,
+		if err := s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+			Entity:  group.Entity,
+			GroupID: group.ID,
+			Feature: feature,
 		}); err != nil {
 			return 0, err
 		}
diff --git a/pkg/oomstore/group.go b/pkg/oomstore/group.go
index a2939379..1b96401d 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{
-			EntityName: entity.Name,
-			GroupID:    id,
+			Entity:  entity,
+			GroupID: id,
 		}); err != nil {
 			return 0, err
 		}
diff --git a/pkg/oomstore/online_query.go b/pkg/oomstore/online_query.go
index 94f03817..d38b6860 100644
--- a/pkg/oomstore/online_query.go
+++ b/pkg/oomstore/online_query.go
@@ -38,15 +38,18 @@ func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*type
 
 	featureMap := groupFeaturesByGroupID(features)
 
-	for _, featureList := range featureMap {
-		if len(featureList) == 0 {
+	for _, features := range featureMap {
+		if len(features) == 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,
-			Group:      *featureList[0].Group,
-			Features:   featureList,
-			RevisionID: featureList[0].OnlineRevisionID(),
+			Features:   features,
 		})
 		if err != nil {
 			return nil, err
@@ -108,11 +111,14 @@ 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 c12a0bb5..bceb9462 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{
-		EntityName:    entity.Name,
+		Entity:        entity,
 		EntityKey:     opt.EntityKey,
 		GroupID:       group.ID,
 		Features:      features,
diff --git a/pkg/oomstore/sync.go b/pkg/oomstore/sync.go
index df144f1f..c413f1c4 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/create_table.go b/internal/database/online/test_impl/create_table.go
deleted file mode 100644
index 5a157ea3..00000000
--- a/internal/database/online/test_impl/create_table.go
+++ /dev/null
@@ -1,35 +0,0 @@
-package test_impl
-
-import (
-	"testing"
-
-	"github.com/oom-ai/oomstore/internal/database/online"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestPrepareStreamTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
-	t.Cleanup(destroyStore)
-
-	ctx, store := prepareStore(t)
-	defer store.Close()
-
-	group := SampleStream.Group
-	t.Run("create stream table", func(t *testing.T) {
-		err := store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-			EntityName: group.Entity.Name,
-			GroupID:    group.ID,
-		})
-		assert.NoError(t, err, "create stream table failed: %v", err)
-	})
-
-	for _, feature := range SampleStream.Features {
-		t.Run("stream table add column", func(t *testing.T) {
-			err := store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-				EntityName: feature.Entity().Name,
-				GroupID:    feature.GroupID,
-				Feature:    feature,
-			})
-			assert.NoError(t, err, "stream table add column failed: %v", err)
-		})
-	}
-}
diff --git a/internal/database/online/test_impl/get.go b/internal/database/online/test_impl/get.go
index 39f0e7c7..aeb1398f 100644
--- a/internal/database/online/test_impl/get.go
+++ b/internal/database/online/test_impl/get.go
@@ -3,82 +3,79 @@ package test_impl
 import (
 	"testing"
 
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/stretchr/testify/require"
 )
 
 func TestGetExisted(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
 	t.Cleanup(destroyStore)
+
+	s := &SampleSmall
 	ctx, store := prepareStore(t)
 	defer store.Close()
+	importSample(t, ctx, store, s)
 
-	for _, s := range []*Sample{&SampleSmall, &SampleStream} {
-		importSample(t, ctx, store, s)
-
-		for _, target := range s.Data {
-			opt := online.GetOpt{
-				EntityKey: target.EntityKey(),
-				Group:     s.Group,
-				Features:  s.Features,
-			}
-			if s.Group.Category == types.CategoryBatch {
-				opt.RevisionID = &s.Revision.ID
-			}
-			rs, err := store.Get(ctx, opt)
-			require.NoError(t, err)
+	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(),
+		}
+		rs, err := store.Get(ctx, opt)
+		require.NoError(t, err)
 
-			for i, f := range s.Features {
-				compareFeatureValue(t, target.ValueAt(i), rs[f.FullName()], f.ValueType)
-			}
+		for i, f := range s.Features {
+			compareFeatureValue(t, target.ValueAt(i), rs[f.FullName()], f.ValueType)
 		}
 	}
 }
 
 func TestGetNotExistedEntityKey(t *testing.T, prepareStore PrepareStoreFn, destroystore DestroyStoreFn) {
 	t.Cleanup(destroystore)
+
+	s := &SampleSmall
 	ctx, store := prepareStore(t)
 	defer store.Close()
+	importSample(t, ctx, store, s)
 
-	for _, s := range []*Sample{&SampleSmall, &SampleStream} {
-		importSample(t, ctx, store, s)
-
-		rs, err := store.Get(ctx, online.GetOpt{
-			EntityKey:  "not-existed-key",
-			Group:      s.Group,
-			Features:   s.Features,
-			RevisionID: &s.Revision.ID,
-		})
-		require.NoError(t, err)
-		require.Equal(t, 0, len(rs), "actual: %+v", rs)
-	}
+	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",
+	})
+	require.NoError(t, err)
+	require.Equal(t, 0, len(rs), "actual: %+v", rs)
 }
 
 func TestMultiGet(t *testing.T, prepareStore PrepareStoreFn, destroystore DestroyStoreFn) {
 	t.Cleanup(destroystore)
+
+	s := &SampleSmall
 	ctx, store := prepareStore(t)
 	defer store.Close()
+	importSample(t, ctx, store, s)
 
-	for _, s := range []*Sample{&SampleSmall, &SampleStream} {
-		importSample(t, ctx, store, s)
-
-		keys := []string{"not-existed-key"}
-		for _, r := range s.Data {
-			keys = append(keys, r.EntityKey())
-		}
-		rs, err := store.MultiGet(ctx, online.MultiGetOpt{
-			EntityKeys: keys,
-			Group:      s.Group,
-			Features:   s.Features,
-			RevisionID: &s.Revision.ID,
-		})
-		require.NoError(t, err)
-		for _, record := range s.Data {
-			for i, feature := range s.Features {
-				compareFeatureValue(t, record.ValueAt(i), rs[record.EntityKey()][feature.FullName()], feature.ValueType)
-			}
+	keys := []string{"not-existed-key"}
+	for _, r := range s.Data {
+		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,
+	})
+	require.NoError(t, err)
+	for _, record := range s.Data {
+		for i, feature := range s.Features {
+			compareFeatureValue(t, record.ValueAt(i), rs[record.EntityKey()][feature.FullName()], feature.ValueType)
 		}
-		require.Equal(t, len(s.Data), len(rs), "actual: %+v", rs)
 	}
+
+	require.Equal(t, len(s.Data), len(rs), "actual: %+v", rs)
 }
diff --git a/internal/database/online/test_impl/purge.go b/internal/database/online/test_impl/purge.go
index 17117473..88bf4ebc 100644
--- a/internal/database/online/test_impl/purge.go
+++ b/internal/database/online/test_impl/purge.go
@@ -19,10 +19,11 @@ 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))
@@ -41,10 +42,11 @@ 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/push.go b/internal/database/online/test_impl/push.go
deleted file mode 100644
index 60930100..00000000
--- a/internal/database/online/test_impl/push.go
+++ /dev/null
@@ -1,93 +0,0 @@
-package test_impl
-
-import (
-	"testing"
-
-	"github.com/oom-ai/oomstore/internal/database/online"
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroyStoreFn) {
-	t.Cleanup(destoryStore)
-
-	ctx, store := prepareStore(t)
-	defer store.Close()
-
-	entity := SampleStream.Entity
-	group := SampleStream.Group
-	feature1 := SampleStream.Features[0]
-	feature2 := SampleStream.Features[1]
-
-	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-		EntityName: entity.Name,
-		GroupID:    group.ID,
-	}))
-
-	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-		EntityName: entity.Name,
-		GroupID:    group.ID,
-		Feature:    feature1,
-	}))
-
-	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		EntityName:    entity.Name,
-		EntityKey:     "user1",
-		GroupID:       group.ID,
-		Features:      types.FeatureList{feature1},
-		FeatureValues: []interface{}{int64(1)},
-	}))
-	rs, err := store.Get(ctx, online.GetOpt{
-		EntityKey:  "user1",
-		Group:      group,
-		Features:   types.FeatureList{feature1},
-		RevisionID: nil,
-	})
-	assert.NoError(t, err)
-	assert.Equal(t, map[string]interface{}{
-		feature1.FullName(): int64(1),
-	}, rs)
-
-	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		EntityName:    entity.Name,
-		EntityKey:     "user1",
-		GroupID:       group.ID,
-		Features:      types.FeatureList{feature1},
-		FeatureValues: []interface{}{int64(2)},
-	}))
-	rs, err = store.Get(ctx, online.GetOpt{
-		EntityKey:  "user1",
-		Group:      group,
-		Features:   types.FeatureList{feature1},
-		RevisionID: nil,
-	})
-	assert.NoError(t, err)
-	assert.Equal(t, map[string]interface{}{
-		feature1.FullName(): int64(2),
-	}, rs)
-
-	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
-		EntityName: entity.Name,
-		GroupID:    group.ID,
-		Feature:    feature2,
-	}))
-
-	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		EntityName:    entity.Name,
-		EntityKey:     "user1",
-		GroupID:       group.ID,
-		Features:      types.FeatureList{feature1, feature2},
-		FeatureValues: []interface{}{int64(3), "post2"},
-	}))
-	rs, err = store.Get(ctx, online.GetOpt{
-		EntityKey:  "user1",
-		Group:      group,
-		Features:   types.FeatureList{feature1, feature2},
-		RevisionID: nil,
-	})
-	assert.NoError(t, err)
-	assert.Equal(t, map[string]interface{}{
-		feature1.FullName(): int64(3),
-		feature2.FullName(): "post2",
-	}, rs)
-}
diff --git a/internal/database/online/test_impl/stream.go b/internal/database/online/test_impl/stream.go
new file mode 100644
index 00000000..564c5a58
--- /dev/null
+++ b/internal/database/online/test_impl/stream.go
@@ -0,0 +1,186 @@
+package test_impl
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+)
+
+type StreamData struct {
+	entity   types.Entity
+	groups   types.GroupList
+	features types.FeatureList
+}
+
+var simpleStreamData StreamData
+
+func init() {
+	entity := types.Entity{
+		ID:   1,
+		Name: "user",
+	}
+
+	group1 := &types.Group{
+		ID:       1,
+		Name:     "user_clicks",
+		Category: types.CategoryStream,
+		EntityID: 1,
+		Entity:   &entity,
+	}
+
+	group2 := &types.Group{
+		ID:       2,
+		Name:     "user_reads",
+		Category: types.CategoryStream,
+		EntityID: 1,
+		Entity:   &entity,
+	}
+
+	simpleStreamData = StreamData{
+		entity: entity,
+		groups: types.GroupList{group1, group2},
+		features: types.FeatureList{
+			&types.Feature{
+				ID:        1,
+				Name:      "last_5_click_post",
+				ValueType: types.String,
+				GroupID:   1,
+				Group:     group1,
+			},
+			&types.Feature{
+				ID:        2,
+				Name:      "last_10_click_post",
+				ValueType: types.String,
+				GroupID:   1,
+				Group:     group1,
+			},
+			&types.Feature{
+				ID:        3,
+				Name:      "recent_5_read_post",
+				ValueType: types.String,
+				GroupID:   2,
+				Group:     group2,
+			},
+		},
+	}
+}
+
+func TestPrepareStreamTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
+	t.Cleanup(destroyStore)
+
+	ctx, store := prepareStore(t)
+	defer store.Close()
+
+	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,
+			})
+			assert.NoError(t, err, "create stream table failed: %v", err)
+		})
+	}
+
+	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,
+			})
+			assert.NoError(t, err, "stream table add column failed: %v", err)
+		})
+	}
+}
+
+func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroyStoreFn) {
+	t.Cleanup(destoryStore)
+
+	ctx, store := prepareStore(t)
+	defer store.Close()
+
+	var (
+		entity   = simpleStreamData.entity
+		group    = simpleStreamData.groups[0]
+		feature1 = simpleStreamData.features[0]
+		feature2 = simpleStreamData.features[1]
+	)
+
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+		Entity:  &entity,
+		GroupID: group.ID,
+	}))
+
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+		Entity:  &entity,
+		GroupID: group.ID,
+		Feature: feature1,
+	}))
+
+	assert.NoError(t, store.Push(ctx, online.PushOpt{
+		Entity:        &entity,
+		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,
+		Features:   types.FeatureList{feature1},
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, map[string]interface{}{
+		feature1.FullName(): "post1",
+	}, rs)
+
+	assert.NoError(t, store.Push(ctx, online.PushOpt{
+		Entity:        &entity,
+		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,
+		Features:   types.FeatureList{feature1},
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, map[string]interface{}{
+		feature1.FullName(): "post2",
+	}, rs)
+
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+		Entity:  &entity,
+		GroupID: group.ID,
+		Feature: feature2,
+	}))
+
+	assert.NoError(t, store.Push(ctx, online.PushOpt{
+		Entity:        &entity,
+		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,
+		Features:   types.FeatureList{feature1, feature2},
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, map[string]interface{}{
+		feature1.FullName(): "post1",
+		feature2.FullName(): "post2",
+	}, rs)
+}
diff --git a/internal/database/online/test_impl/util.go b/internal/database/online/test_impl/util.go
index f6d3ab4c..c5135254 100644
--- a/internal/database/online/test_impl/util.go
+++ b/internal/database/online/test_impl/util.go
@@ -28,117 +28,85 @@ type Sample struct {
 
 var SampleSmall Sample
 var SampleMedium Sample
-var SampleStream Sample
 
 func init() {
 	rand.Seed(time.Now().UnixNano())
-	entity := types.Entity{ID: 1, Name: "user"}
-	group1 := types.Group{ID: 1, Category: types.CategoryBatch, Entity: &entity}
-	group2 := types.Group{ID: 2, Category: types.CategoryBatch, Entity: &entity}
-	group3 := types.Group{ID: 3, Name: "user_clicks", Category: types.CategoryStream, Entity: &entity}
-
-	SampleSmall = Sample{
-		Entity: entity,
-		Group:  group1,
-		Features: types.FeatureList{
-			&types.Feature{
-				ID:        1,
-				Name:      "age",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.Int64,
-			},
-			&types.Feature{
-				ID:        2,
-				Name:      "gender",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.String,
-			},
-			&types.Feature{
-				ID:        3,
-				Name:      "account",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.Float64,
+	entity := types.Entity{ID: 5, Name: "user"}
+	group := types.Group{ID: 1, Category: types.CategoryBatch, Entity: &entity}
+	{
+		SampleSmall = Sample{
+			Entity: entity,
+			Group:  group,
+			Features: types.FeatureList{
+				&types.Feature{
+					ID:        1,
+					Name:      "age",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.Int64,
+				},
+				&types.Feature{
+					ID:        2,
+					Name:      "gender",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.String,
+				},
+				&types.Feature{
+					ID:        3,
+					Name:      "account",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.Float64,
+				},
+				&types.Feature{
+					ID:        4,
+					Name:      "is_active",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.Bool,
+				},
+				&types.Feature{
+					ID:        5,
+					Name:      "register_time",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.Time,
+				},
 			},
-			&types.Feature{
-				ID:        4,
-				Name:      "is_active",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.Bool,
+			Revision: types.Revision{
+				ID:      3,
+				GroupID: 1,
+				Group:   &group,
 			},
-			&types.Feature{
-				ID:        5,
-				Name:      "register_time",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.Time,
+			Data: []types.ExportRecord{
+				[]interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
+				[]interface{}{"3216", int64(29), nil, 2.0, false, time.Now()},
+				[]interface{}{"3217", int64(44), "M", 3.1, true, time.Now()},
 			},
-		},
-		Revision: types.Revision{
-			ID:      3,
-			GroupID: 1,
-			Group:   &group1,
-		},
-		Data: []types.ExportRecord{
-			[]interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
-			[]interface{}{"3216", int64(29), nil, 2.0, false, time.Now()},
-			[]interface{}{"3217", int64(44), "M", 3.1, true, time.Now()},
-		},
-	}
+		}
 
-	var data []types.ExportRecord
-	for i := 0; i < 100; i++ {
-		record := []interface{}{dbutil.RandString(10), rand.Float64()}
-		data = append(data, record)
 	}
-	SampleMedium = Sample{
-		Entity: entity,
-		Group:  group2,
-		Features: types.FeatureList{
+
+	{
+		features := types.FeatureList{
 			&types.Feature{
-				ID:        6,
+				ID:        2,
 				Name:      "charge",
 				GroupID:   2,
-				Group:     &group2,
+				Group:     &types.Group{ID: 2, Category: types.CategoryBatch},
 				ValueType: types.Float64,
 			},
-		},
-		Revision: types.Revision{
-			ID:      9,
-			GroupID: 2,
-			Group:   &group2,
-		},
-		Data: data,
-	}
+		}
 
-	SampleStream = Sample{
-		Entity: entity,
-		Group:  group3,
-		Features: types.FeatureList{
-			&types.Feature{
-				ID:        7,
-				Name:      "amount",
-				ValueType: types.Int64,
-				GroupID:   3,
-				Group:     &group3,
-			},
-			&types.Feature{
-				ID:        8,
-				Name:      "last_10_click_post",
-				ValueType: types.String,
-				GroupID:   3,
-				Group:     &group3,
-			},
-		},
-		Data: []types.ExportRecord{
-			[]interface{}{"3215", int64(1), "1,2,3,4"},
-			[]interface{}{"3216", int64(2), "2,3,4,5"},
-			[]interface{}{"3217", int64(3), "3,4,5,6"},
-			[]interface{}{"3218", int64(4), "4,5,6,7"},
-		},
+		revision := types.Revision{ID: 9, GroupID: 2, Group: &types.Group{ID: 2, Category: types.CategoryBatch}}
+		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}
 	}
 }
 
@@ -152,15 +120,12 @@ func importSample(t *testing.T, ctx context.Context, store online.Store, samples
 			}
 		}(sample)
 
-		opt := online.ImportOpt{
+		err := store.Import(ctx, online.ImportOpt{
 			Group:        sample.Group,
 			Features:     sample.Features,
+			Revision:     &sample.Revision,
 			ExportStream: stream,
-		}
-		if sample.Group.Category == types.CategoryBatch {
-			opt.RevisionID = &sample.Revision.ID
-		}
-		err := store.Import(ctx, opt)
+		})
 		require.NoError(t, err)
 	}
 }