Repository: oom-ai/oomstore
Pull request #895 (2022-01-04 08:28:22)

Feat/online query stream feature
this PR does:
1. modified `GetOpt` and `GetMulOpt`: add field `GroupID` and make `RevisionID` to pointer.
2. online Dbs support gets stream feature.
3. oomstore online query support get stream feature.
4. add unit test for getting stream feature
5. fix Cassandra get/multi-get bug

codecov[bot]: # [Codecov](https://codecov.io/gh/oom-ai/oomstore/pull/895?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) Report
> Merging [#895](https://codecov.io/gh/oom-ai/oomstore/pull/895?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) (415019b) into [main](https://codecov.io/gh/oom-ai/oomstore/commit/ed8e7dc546a53d293c08ba1b9b0af338a90ca3ca?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) (ed8e7dc) will **decrease** coverage by `0.65%`.
> The diff coverage is `59.09%`.

> :exclamation: Current head 415019b differs from pull request most recent head 7b2e48d. Consider uploading reports for the commit 7b2e48d to get more accurate results
[![Impacted file tree graph](https://codecov.io/gh/oom-ai/oomstore/pull/895/graphs/tree.svg?width=650&height=150&src=pr&token=C59L7LTRM4&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai)](https://codecov.io/gh/oom-ai/oomstore/pull/895?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai)

```diff
@@            Coverage Diff             @@
##             main     #895      +/-   ##
==========================================
- Coverage   51.35%   50.69%   -0.66%     
==========================================
  Files          84       84              
  Lines        3441     3507      +66     
==========================================
+ Hits         1767     1778      +11     
- Misses       1461     1513      +52     
- Partials      213      216       +3     
```


| [Impacted Files](https://codecov.io/gh/oom-ai/oomstore/pull/895?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) | Coverage Δ | |
|---|---|---|
| [internal/database/online/cassandra/get.go](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb25saW5lL2Nhc3NhbmRyYS9nZXQuZ28=) | `0.00% <0.00%> (ø)` | |
| [internal/database/online/dynamodb/get.go](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb25saW5lL2R5bmFtb2RiL2dldC5nbw==) | `56.12% <57.14%> (-0.86%)` | :arrow_down: |
| [internal/database/online/tikv/get.go](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb25saW5lL3Rpa3YvZ2V0Lmdv) | `69.23% <66.66%> (-2.20%)` | :arrow_down: |
| [internal/database/online/redis/get.go](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb25saW5lL3JlZGlzL2dldC5nbw==) | `68.75% <77.77%> (+4.46%)` | :arrow_up: |
| [pkg/oomstore/online\_query.go](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-cGtnL29vbXN0b3JlL29ubGluZV9xdWVyeS5nbw==) | `85.21% <100.00%> (+1.18%)` | :arrow_up: |
| [internal/database/dbutil/schema.go](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvZGJ1dGlsL3NjaGVtYS5nbw==) | `44.26% <0.00%> (-15.00%)` | :arrow_down: |
| [internal/database/offline/postgres/util.go](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb2ZmbGluZS9wb3N0Z3Jlcy91dGlsLmdv) | `47.61% <0.00%> (-8.39%)` | :arrow_down: |
| [internal/database/dbutil/data\_loader.go](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvZGJ1dGlsL2RhdGFfbG9hZGVyLmdv) | `0.00% <0.00%> (ø)` | |
| ... and [2 more](https://codecov.io/gh/oom-ai/oomstore/pull/895/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/oom-ai/oomstore/pull/895?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/oom-ai/oomstore/pull/895?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai). Last update [ed8e7dc...7b2e48d](https://codecov.io/gh/oom-ai/oomstore/pull/895?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai).


Patch:
diff --git a/internal/database/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index b4734abf..61f0c524 100644
--- a/internal/database/online/cassandra/get.go
+++ b/internal/database/online/cassandra/get.go
@@ -14,7 +14,12 @@ import (
 )
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	tableName := sqlutil.OnlineBatchTableName(opt.RevisionID)
+	var tableName string
+	if opt.Group.Category == types.CategoryBatch {
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
+	}
 
 	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`,
 		strings.Join(opt.Features.Names(), ","),
@@ -30,7 +35,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 		return nil, err
 	}
 
-	rs := make(map[string]interface{})
+	rs := make(map[string]interface{}, len(scan))
 	for _, feature := range opt.Features {
 		rs[feature.FullName] = deserializeString(scan[feature.Name])
 	}
@@ -40,10 +45,16 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 // response: map[entity_key]map[feature_name]feature_value
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
 	var (
-		tableName    = sqlutil.OnlineBatchTableName(opt.RevisionID)
+		tableName    string
 		placeholders = getPlaceholders(len(opt.EntityKeys))
 	)
 
+	if opt.Group.Category == types.CategoryBatch {
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
+	}
+
 	query := fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s in (%s)`,
 		opt.Entity.Name,
 		strings.Join(opt.Features.Names(), ","),
diff --git a/internal/database/online/dynamodb/get.go b/internal/database/online/dynamodb/get.go
index 3ed91fc4..697c1cb5 100644
--- a/internal/database/online/dynamodb/get.go
+++ b/internal/database/online/dynamodb/get.go
@@ -20,7 +20,13 @@ const (
 )
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	tableName := sqlutil.OnlineBatchTableName(opt.RevisionID)
+	var tableName string
+	if opt.Group.Category == oomTypes.CategoryBatch {
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
+	}
+
 	entityKeyValue, err := attributevalue.Marshal(opt.EntityKey)
 	if err != nil {
 		return nil, err
@@ -42,8 +48,14 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 
 // response: map[entity_key]map[feature_name]feature_value
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
+	var tableName string
+	if opt.Group.Category == oomTypes.CategoryBatch {
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
+	}
+
 	res := make(map[string]dbutil.RowMap)
-	tableName := sqlutil.OnlineBatchTableName(opt.RevisionID)
 	keys := make([]map[string]types.AttributeValue, 0, BatchGetItemCapacity)
 	for _, entityKey := range opt.EntityKeys {
 		entityKeyValue, err := attributevalue.Marshal(entityKey)
diff --git a/internal/database/online/redis/get.go b/internal/database/online/redis/get.go
index 08563d31..da30f2c8 100644
--- a/internal/database/online/redis/get.go
+++ b/internal/database/online/redis/get.go
@@ -6,10 +6,20 @@ import (
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	key, err := serializeRedisKeyForBatchFeature(opt.RevisionID, opt.EntityKey)
+	var (
+		key string
+		err error
+	)
+
+	if opt.Group.Category == types.CategoryBatch {
+		key, err = serializeRedisKeyForBatchFeature(*opt.RevisionID, opt.EntityKey)
+	} else {
+		key, err = serializeRedisKeyForStreamFeature(opt.Group.ID, opt.EntityKey)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -49,6 +59,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 		rowMap, err := db.Get(ctx, online.GetOpt{
 			Entity:     opt.Entity,
 			RevisionID: opt.RevisionID,
+			Group:      opt.Group,
 			EntityKey:  entityKey,
 			Features:   opt.Features,
 		})
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index b8344552..824b3c2e 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -14,8 +14,14 @@ import (
 )
 
 func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.BackendType) (dbutil.RowMap, error) {
+	var tableName string
+	if opt.Group.Category == types.CategoryBatch {
+		tableName = OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = OnlineStreamTableName(opt.Group.ID)
+	}
+
 	featureNames := opt.Features.Names()
-	tableName := OnlineBatchTableName(opt.RevisionID)
 	qt := dbutil.QuoteFn(backend)
 	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`, qt(featureNames...), qt(tableName), qt(opt.Entity.Name))
 
@@ -36,8 +42,14 @@ func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.Back
 
 // response: map[entity_key]map[feature_name]feature_value
 func MultiGet(ctx context.Context, db *sqlx.DB, opt online.MultiGetOpt, backend types.BackendType) (map[string]dbutil.RowMap, error) {
+	var tableName string
+	if opt.Group.Category == types.CategoryBatch {
+		tableName = OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = OnlineStreamTableName(opt.Group.ID)
+	}
+
 	featureNames := opt.Features.Names()
-	tableName := OnlineBatchTableName(opt.RevisionID)
 	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))
 	sql, args, err := sqlx.In(query, opt.EntityKeys)
diff --git a/internal/database/online/tikv/get.go b/internal/database/online/tikv/get.go
index 131767ba..0c54b36d 100644
--- a/internal/database/online/tikv/get.go
+++ b/internal/database/online/tikv/get.go
@@ -18,6 +18,7 @@ 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},
 		Features:   opt.Features,
@@ -33,7 +34,19 @@ 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) {
-	serializedRevisionID, err := kvutil.SerializeByValue(opt.RevisionID)
+	var (
+		allBatch            bool
+		serializedPrefixKey string
+		err                 error
+	)
+
+	allBatch = (opt.Group.Category == types.CategoryBatch)
+
+	if allBatch {
+		serializedPrefixKey, err = kvutil.SerializeByValue(*opt.RevisionID)
+	} else {
+		serializedPrefixKey, err = kvutil.SerializeByValue(opt.Group.ID)
+	}
 	if err != nil {
 		return nil, err
 	}
@@ -60,7 +73,11 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	var keys [][]byte
 	for _, serializedEntityKey := range serializedEntityKeys {
 		for _, serializedFeatureID := range serializedFeatureIDs {
-			keys = append(keys, getKeyOfBatchFeature(serializedRevisionID, serializedEntityKey, serializedFeatureID))
+			if allBatch {
+				keys = append(keys, getKeyOfBatchFeature(serializedPrefixKey, serializedEntityKey, serializedFeatureID))
+			} else {
+				keys = append(keys, getKeyOfStreamFeature(serializedPrefixKey, serializedEntityKey, serializedFeatureID))
+			}
 		}
 	}
 
diff --git a/internal/database/online/types.go b/internal/database/online/types.go
index c65e9f7e..e0a3fada 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -6,15 +6,17 @@ import (
 
 type GetOpt struct {
 	Entity     *types.Entity
-	RevisionID int
 	EntityKey  string
+	RevisionID *int
+	Group      *types.Group
 	Features   types.FeatureList
 }
 
 type MultiGetOpt struct {
 	Entity     *types.Entity
-	RevisionID int
 	EntityKeys []string
+	RevisionID *int
+	Group      *types.Group
 	Features   types.FeatureList
 }
 
diff --git a/pkg/oomstore/online_query.go b/pkg/oomstore/online_query.go
index 62081524..a334e891 100644
--- a/pkg/oomstore/online_query.go
+++ b/pkg/oomstore/online_query.go
@@ -19,8 +19,6 @@ func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*type
 	}
 	features := s.metadata.CacheListFeature(ctx, metadata.ListFeatureOpt{
 		FeatureFullNames: &opt.FeatureFullNames,
-	}).Filter(func(f *types.Feature) bool {
-		return f.Group.OnlineRevisionID != nil
 	})
 	if len(features) == 0 {
 		return &rs, nil
@@ -32,15 +30,18 @@ func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*type
 	}
 	rs.EntityName = entity.Name
 
-	featureMap := groupFeaturesByRevisionID(features)
+	featureMap := groupFeaturesByGroupID(features)
 
-	for onlineRevisionID, features := range featureMap {
+	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: onlineRevisionID,
+			RevisionID: revisionID,
+			Group:      group,
 			EntityKey:  opt.EntityKey,
 			Features:   features,
 		})
@@ -60,10 +61,6 @@ func (s *OomStore) OnlineMultiGet(ctx context.Context, opt types.OnlineMultiGetO
 	features := s.metadata.CacheListFeature(ctx, metadata.ListFeatureOpt{
 		FeatureFullNames: &opt.FeatureFullNames,
 	})
-
-	features = features.Filter(func(f *types.Feature) bool {
-		return f.OnlineRevisionID() != nil
-	})
 	if len(features) == 0 {
 		return result, nil
 	}
@@ -75,7 +72,7 @@ func (s *OomStore) OnlineMultiGet(ctx context.Context, opt types.OnlineMultiGetO
 	if entity == nil {
 		return nil, fmt.Errorf("failed to get shared entity")
 	}
-	featureMap := groupFeaturesByRevisionID(features)
+	featureMap := groupFeaturesByGroupID(features)
 
 	// entity_key -> feature_name -> feature_value
 	featureValueMap, err := s.getFeatureValueMap(ctx, opt.EntityKeys, featureMap, entity)
@@ -101,13 +98,16 @@ func (s *OomStore) getFeatureValueMap(ctx context.Context, entityKeys []string,
 	// entity_key -> types.RecordMap
 	featureValueMap := make(map[string]dbutil.RowMap)
 
-	for onlineRevisionID, features := range featureMap {
+	for _, features := range featureMap {
 		if len(features) == 0 {
 			continue
 		}
+		revisionID := features[0].OnlineRevisionID()
+		group := features[0].Group
 		featureValues, err := s.online.MultiGet(ctx, online.MultiGetOpt{
 			Entity:     entity,
-			RevisionID: onlineRevisionID,
+			RevisionID: revisionID,
+			Group:      group,
 			EntityKeys: entityKeys,
 			Features:   features,
 		})
@@ -126,14 +126,11 @@ func (s *OomStore) getFeatureValueMap(ctx context.Context, entityKeys []string,
 	return featureValueMap, nil
 }
 
-func groupFeaturesByRevisionID(features types.FeatureList) map[int]types.FeatureList {
+func groupFeaturesByGroupID(features types.FeatureList) map[int]types.FeatureList {
 	featureMap := make(map[int]types.FeatureList)
 	for _, f := range features {
-		id := f.OnlineRevisionID()
-		if id == nil {
-			continue
-		}
-		featureMap[*id] = append(featureMap[*id], f)
+		id := f.GroupID
+		featureMap[id] = append(featureMap[id], f)
 	}
 	return featureMap
 }
diff --git a/pkg/oomstore/types/feature.go b/pkg/oomstore/types/feature.go
index cc2f3c34..8d2ef873 100644
--- a/pkg/oomstore/types/feature.go
+++ b/pkg/oomstore/types/feature.go
@@ -35,6 +35,9 @@ func (f *Feature) Entity() *Entity {
 }
 
 func (f *Feature) OnlineRevisionID() *int {
+	if f.Group.Category == CategoryStream {
+		return nil
+	}
 	return f.Group.OnlineRevisionID
 }
 


Test patch:
diff --git a/internal/database/online/test_impl/get.go b/internal/database/online/test_impl/get.go
index 43846ce2..32aa871d 100644
--- a/internal/database/online/test_impl/get.go
+++ b/internal/database/online/test_impl/get.go
@@ -18,7 +18,8 @@ func TestGetExisted(t *testing.T, prepareStore PrepareStoreFn, destroyStore Dest
 	for _, target := range s.Data {
 		opt := online.GetOpt{
 			Entity:     s.Entity,
-			RevisionID: s.Revision.ID,
+			RevisionID: &s.Revision.ID,
+			Group:      s.Revision.Group,
 			Features:   s.Features,
 			EntityKey:  target.EntityKey(),
 		}
@@ -41,7 +42,8 @@ func TestGetNotExistedEntityKey(t *testing.T, prepareStore PrepareStoreFn, destr
 
 	rs, err := store.Get(ctx, online.GetOpt{
 		Entity:     s.Entity,
-		RevisionID: s.Revision.ID,
+		RevisionID: &s.Revision.ID,
+		Group:      s.Revision.Group,
 		Features:   s.Features,
 		EntityKey:  "not-existed-key",
 	})
@@ -63,7 +65,8 @@ func TestMultiGet(t *testing.T, prepareStore PrepareStoreFn, destroystore Destro
 	}
 	rs, err := store.MultiGet(ctx, online.MultiGetOpt{
 		Entity:     s.Entity,
-		RevisionID: s.Revision.ID,
+		RevisionID: &s.Revision.ID,
+		Group:      s.Revision.Group,
 		Features:   s.Features,
 		EntityKeys: keys,
 	})
diff --git a/internal/database/online/test_impl/purge.go b/internal/database/online/test_impl/purge.go
index 9e4c44e9..651bc293 100644
--- a/internal/database/online/test_impl/purge.go
+++ b/internal/database/online/test_impl/purge.go
@@ -20,7 +20,8 @@ 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,
+			RevisionID: &SampleMedium.Revision.ID,
+			Group:      SampleMedium.Revision.Group,
 			EntityKey:  record.EntityKey(),
 			Features:   SampleMedium.Features,
 		})
@@ -42,7 +43,8 @@ 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,
+			RevisionID: &SampleSmall.Revision.ID,
+			Group:      SampleSmall.Revision.Group,
 			EntityKey:  record.EntityKey(),
 			Features:   SampleSmall.Features,
 		})
diff --git a/internal/database/online/test_impl/stream.go b/internal/database/online/test_impl/stream.go
index 5206a25b..7e1c99c2 100644
--- a/internal/database/online/test_impl/stream.go
+++ b/internal/database/online/test_impl/stream.go
@@ -131,6 +131,17 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 		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,
@@ -139,6 +150,17 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 		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,
@@ -153,4 +175,16 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 		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 22ce2d71..466e38ad 100644
--- a/internal/database/online/test_impl/util.go
+++ b/internal/database/online/test_impl/util.go
@@ -39,6 +39,7 @@ func init() {
 					Name:      "age",
 					FullName:  "user.age",
 					GroupID:   1,
+					Group:     &types.Group{ID: 1, Category: types.CategoryBatch},
 					ValueType: types.Int64,
 				},
 				&types.Feature{
@@ -46,6 +47,7 @@ func init() {
 					Name:      "gender",
 					FullName:  "user.gender",
 					GroupID:   1,
+					Group:     &types.Group{ID: 1, Category: types.CategoryBatch},
 					ValueType: types.String,
 				},
 				&types.Feature{
@@ -53,6 +55,7 @@ func init() {
 					Name:      "account",
 					FullName:  "user.account",
 					GroupID:   1,
+					Group:     &types.Group{ID: 1, Category: types.CategoryBatch},
 					ValueType: types.Float64,
 				},
 				&types.Feature{
@@ -60,6 +63,7 @@ func init() {
 					Name:      "is_active",
 					FullName:  "user.is_active",
 					GroupID:   1,
+					Group:     &types.Group{ID: 1, Category: types.CategoryBatch},
 					ValueType: types.Bool,
 				},
 				&types.Feature{
@@ -67,11 +71,19 @@ func init() {
 					Name:      "register_time",
 					FullName:  "user.register_time",
 					GroupID:   1,
+					Group:     &types.Group{ID: 1, Category: types.CategoryBatch},
 					ValueType: types.Time,
 				},
 			},
-			Revision: &types.Revision{ID: 3, GroupID: 1},
-			Entity:   &types.Entity{ID: 5, Name: "user", Length: 4},
+			Revision: &types.Revision{
+				ID:      3,
+				GroupID: 1,
+				Group: &types.Group{
+					ID:       1,
+					Category: types.CategoryBatch,
+				},
+			},
+			Entity: &types.Entity{ID: 5, Name: "user", Length: 4},
 			Data: []types.ExportRecord{
 				[]interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
 				[]interface{}{"3216", int64(29), nil, 2.0, false, time.Now()},
@@ -88,11 +100,12 @@ func init() {
 				Name:      "charge",
 				FullName:  "user.charge",
 				GroupID:   2,
+				Group:     &types.Group{ID: 2, Category: types.CategoryBatch},
 				ValueType: types.Float64,
 			},
 		}
 
-		revision := &types.Revision{ID: 9, GroupID: 2}
+		revision := &types.Revision{ID: 9, GroupID: 2, Group: &types.Group{ID: 2, Category: types.CategoryBatch}}
 		entity := &types.Entity{ID: 5, Name: "user", Length: 5}
 		var data []types.ExportRecord
 
diff --git a/pkg/oomstore/online_query_test.go b/pkg/oomstore/online_query_test.go
index 180c9b8f..6bb3215f 100644
--- a/pkg/oomstore/online_query_test.go
+++ b/pkg/oomstore/online_query_test.go
@@ -25,7 +25,6 @@ func TestOnlineGet(t *testing.T) {
 	entityName := "device"
 	consistentFeatures := prepareFeatures(true, true)
 	inconsistentFeatures := prepareFeatures(false, true)
-	unavailableFeatures := prepareFeatures(true, false)
 
 	testCases := []struct {
 		description   string
@@ -38,14 +37,14 @@ func TestOnlineGet(t *testing.T) {
 		{
 			description: "no available features",
 			opt: types.OnlineGetOpt{
-				FeatureFullNames: unavailableFeatures.FullNames(),
+				FeatureFullNames: []string{"f1", "f2"},
 				EntityKey:        "1234",
 			},
-			features:      unavailableFeatures,
+			features:      nil,
 			expectedError: nil,
 			expected: &types.FeatureValues{
 				EntityKey:        "1234",
-				FeatureFullNames: consistentFeatures.FullNames(),
+				FeatureFullNames: []string{"f1", "f2"},
 				FeatureValueMap:  map[string]interface{}{},
 			},
 		},
@@ -113,7 +112,6 @@ func TestOnlineMultiGet(t *testing.T) {
 	entityName := "device"
 	consistentFeatures := prepareFeatures(true, true)
 	inconsistentFeatures := prepareFeatures(false, true)
-	unavailableFeatures := prepareFeatures(true, false)
 
 	testCases := []struct {
 		description   string
@@ -126,10 +124,10 @@ func TestOnlineMultiGet(t *testing.T) {
 		{
 			description: "no available features, return nil",
 			opt: types.OnlineMultiGetOpt{
-				FeatureFullNames: unavailableFeatures.FullNames(),
+				FeatureFullNames: []string{"f1", "f2"},
 				EntityKeys:       []string{"1234", "1235"},
 			},
-			features:      unavailableFeatures,
+			features:      nil,
 			expectedError: nil,
 			expected:      map[string]*types.FeatureValues{},
 		},