Repository: oom-ai/oomstore
Pull request #1208 (2022-02-21 09:07:46)

Feat/online get all other db
this PR does:
1. refactor online/GetByGroupOpt: modify field `GetFeature(id int)` to `GetFeature(id *int, fullName *string)`
2. mysql,sqlite,pg,cassandra,dynamodb implement method `GetByGroup`


Patch:
diff --git a/internal/database/metadata/informer/feature.go b/internal/database/metadata/informer/feature.go
index b9d0d1c4..2b60ed6b 100644
--- a/internal/database/metadata/informer/feature.go
+++ b/internal/database/metadata/informer/feature.go
@@ -35,6 +35,10 @@ func (c *FeatureCache) Get(featureID int) *types.Feature {
 	return c.idIdx[featureID]
 }
 
+func (c *FeatureCache) GetByName(fullName string) *types.Feature {
+	return c.nameIdx[fullName]
+}
+
 func (c *FeatureCache) List(opt metadata.ListCachedFeatureOpt) types.FeatureList {
 	features := c.FeatureList
 
diff --git a/internal/database/metadata/informer/informer.go b/internal/database/metadata/informer/informer.go
index 5e0b25bb..7c67ae72 100644
--- a/internal/database/metadata/informer/informer.go
+++ b/internal/database/metadata/informer/informer.go
@@ -6,9 +6,8 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/oom-ai/oomstore/pkg/errdefs"
-
 	"github.com/oom-ai/oomstore/internal/database/metadata"
+	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
@@ -114,3 +113,12 @@ func (f *Informer) GetCachedFeature(ctx context.Context, featureID int) (*types.
 	}
 	return feature, nil
 }
+
+func (f *Informer) GetCachedFeatureByName(ctx context.Context, fullName string) (*types.Feature, error) {
+	feature := f.Cache().Features.GetByName(fullName).Copy()
+
+	if feature == nil {
+		return nil, errdefs.NotFound(errdefs.Errorf("feature %s not found", fullName))
+	}
+	return feature, nil
+}
diff --git a/internal/database/metadata/mock_metadata/store.go b/internal/database/metadata/mock_metadata/store.go
index 2894e39e..a48e0d35 100644
--- a/internal/database/metadata/mock_metadata/store.go
+++ b/internal/database/metadata/mock_metadata/store.go
@@ -126,6 +126,21 @@ func (mr *MockStoreMockRecorder) GetCachedFeature(ctx, id interface{}) *gomock.C
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCachedFeature", reflect.TypeOf((*MockStore)(nil).GetCachedFeature), ctx, id)
 }
 
+// GetCachedFeatureByName mocks base method.
+func (m *MockStore) GetCachedFeatureByName(ctx context.Context, name string) (*types.Feature, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetCachedFeatureByName", ctx, name)
+	ret0, _ := ret[0].(*types.Feature)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetCachedFeatureByName indicates an expected call of GetCachedFeatureByName.
+func (mr *MockStoreMockRecorder) GetCachedFeatureByName(ctx, name interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCachedFeatureByName", reflect.TypeOf((*MockStore)(nil).GetCachedFeatureByName), ctx, name)
+}
+
 // GetCachedGroup mocks base method.
 func (m *MockStore) GetCachedGroup(ctx context.Context, id int) (*types.Group, error) {
 	m.ctrl.T.Helper()
@@ -819,6 +834,21 @@ func (mr *MockCacheStoreMockRecorder) GetCachedFeature(ctx, id interface{}) *gom
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCachedFeature", reflect.TypeOf((*MockCacheStore)(nil).GetCachedFeature), ctx, id)
 }
 
+// GetCachedFeatureByName mocks base method.
+func (m *MockCacheStore) GetCachedFeatureByName(ctx context.Context, name string) (*types.Feature, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetCachedFeatureByName", ctx, name)
+	ret0, _ := ret[0].(*types.Feature)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetCachedFeatureByName indicates an expected call of GetCachedFeatureByName.
+func (mr *MockCacheStoreMockRecorder) GetCachedFeatureByName(ctx, name interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCachedFeatureByName", reflect.TypeOf((*MockCacheStore)(nil).GetCachedFeatureByName), ctx, name)
+}
+
 // GetCachedGroup mocks base method.
 func (m *MockCacheStore) GetCachedGroup(ctx context.Context, id int) (*types.Group, error) {
 	m.ctrl.T.Helper()
diff --git a/internal/database/metadata/store.go b/internal/database/metadata/store.go
index d54a338d..0dda0e7b 100644
--- a/internal/database/metadata/store.go
+++ b/internal/database/metadata/store.go
@@ -56,6 +56,7 @@ type CacheStore interface {
 	GetCachedGroup(ctx context.Context, id int) (*types.Group, error)
 	GetCachedGroupByName(ctx context.Context, name string) (*types.Group, error)
 	GetCachedFeature(ctx context.Context, id int) (*types.Feature, error)
+	GetCachedFeatureByName(ctx context.Context, name string) (*types.Feature, error)
 
 	// Refresh pulls data from database and update cache.
 	Refresh() error
diff --git a/internal/database/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index b1cb7b46..3f30a101 100644
--- a/internal/database/online/cassandra/get.go
+++ b/internal/database/online/cassandra/get.go
@@ -11,6 +11,7 @@ import (
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/oom-ai/oomstore/pkg/oomstore/util"
 )
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
@@ -48,7 +49,49 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 }
 
 func (db *DB) GetByGroup(ctx context.Context, opt online.GetByGroupOpt) (dbutil.RowMap, error) {
-	panic("TODO: implement me")
+	if err := opt.Validate(); err != nil {
+		return nil, err
+	}
+
+	var tableName string
+	if opt.Group.Category == types.CategoryBatch {
+		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+	}
+
+	query := fmt.Sprintf(`SELECT * FROM %s WHERE %s = ?`,
+		tableName,
+		opt.Group.Entity.Name,
+	)
+
+	scan := make(map[string]interface{})
+	if err := db.Query(query, opt.EntityKey).WithContext(ctx).MapScan(scan); err != nil {
+		if err == gocql.ErrNotFound || isTableNotFoundError(err, tableName) {
+			return scan, nil
+		}
+		return nil, errdefs.WithStack(err)
+	}
+
+	rs := map[string]interface{}{}
+	for k, v := range scan {
+		if k == opt.Group.Entity.Name {
+			continue
+		}
+
+		featureFullName := util.ComposeFullFeatureName(opt.Group.Name, k)
+		feature, err := opt.GetFeature(nil, &featureFullName)
+		if err != nil {
+			return nil, err
+		}
+		deserializedValue, err := dbutil.DeserializeByValueType(v, feature.ValueType, Backend)
+		if err != nil {
+			return nil, err
+		}
+
+		rs[featureFullName] = deserializedValue
+	}
+	return rs, nil
 }
 
 // response: map[entity_key]map[feature_name]feature_value
diff --git a/internal/database/online/dynamodb/get.go b/internal/database/online/dynamodb/get.go
index 38718699..0d5e74d6 100644
--- a/internal/database/online/dynamodb/get.go
+++ b/internal/database/online/dynamodb/get.go
@@ -13,6 +13,7 @@ import (
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	oomTypes "github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/oom-ai/oomstore/pkg/oomstore/util"
 )
 
 const (
@@ -51,7 +52,57 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 }
 
 func (db *DB) GetByGroup(ctx context.Context, opt online.GetByGroupOpt) (dbutil.RowMap, error) {
-	panic("TODO: implement me")
+	if err := opt.Validate(); err != nil {
+		return nil, err
+	}
+
+	var tableName string
+	if opt.Group.Category == oomTypes.CategoryBatch {
+		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+	}
+	entityKeyValue, err := attributevalue.Marshal(opt.EntityKey)
+	if err != nil {
+		return nil, errdefs.WithStack(err)
+	}
+
+	result, err := db.GetItem(ctx, &dynamodb.GetItemInput{
+		TableName: aws.String(tableName),
+		Key: map[string]types.AttributeValue{
+			opt.Group.Entity.Name: entityKeyValue,
+		},
+	})
+	if err != nil {
+		if apiErr := new(types.ResourceNotFoundException); errors.As(err, &apiErr) {
+			return make(dbutil.RowMap), nil
+		}
+		return nil, errdefs.WithStack(err)
+	}
+
+	rs := map[string]interface{}{}
+	var value interface{}
+	for k, v := range result.Item {
+		if k == opt.Group.Entity.Name {
+			continue
+		}
+
+		if err := attributevalue.Unmarshal(v, &value); err != nil {
+			return nil, errdefs.WithStack(err)
+		}
+
+		featureFullName := util.ComposeFullFeatureName(opt.Group.Name, k)
+		feature, err := opt.GetFeature(nil, &featureFullName)
+		if err != nil {
+			return nil, err
+		}
+		deserializedValue, err := dbutil.DeserializeByValueType(value, feature.ValueType, oomTypes.BackendDynamoDB)
+		if err != nil {
+			return nil, err
+		}
+		rs[featureFullName] = deserializedValue
+	}
+	return rs, nil
 }
 
 // response: map[entity_key]map[feature_name]feature_value
diff --git a/internal/database/online/mysql/store.go b/internal/database/online/mysql/store.go
index e0f55bea..938e8a22 100644
--- a/internal/database/online/mysql/store.go
+++ b/internal/database/online/mysql/store.go
@@ -37,7 +37,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 }
 
 func (db *DB) GetByGroup(ctx context.Context, opt online.GetByGroupOpt) (dbutil.RowMap, error) {
-	panic("TODO: implement me")
+	return sqlutil.GetByGroup(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index a7065f8f..9135e62a 100644
--- a/internal/database/online/postgres/store.go
+++ b/internal/database/online/postgres/store.go
@@ -37,7 +37,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 }
 
 func (db *DB) GetByGroup(ctx context.Context, opt online.GetByGroupOpt) (dbutil.RowMap, error) {
-	panic("TODO: implement me")
+	return sqlutil.GetByGroup(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
diff --git a/internal/database/online/redis/get.go b/internal/database/online/redis/get.go
index 8c90e0d8..89466b5d 100644
--- a/internal/database/online/redis/get.go
+++ b/internal/database/online/redis/get.go
@@ -70,7 +70,8 @@ func (db *DB) GetByGroup(ctx context.Context, opt online.GetByGroupOpt) (dbutil.
 		if err != nil {
 			return nil, err
 		}
-		feature, err := opt.GetFeature(int(featureID.(int64)))
+		featureIDInt := int(featureID.(int64))
+		feature, err := opt.GetFeature(&featureIDInt, nil)
 		if err != nil {
 			return nil, err
 		}
diff --git a/internal/database/online/sqlite/store.go b/internal/database/online/sqlite/store.go
index a9ebaaed..96ee9b5d 100644
--- a/internal/database/online/sqlite/store.go
+++ b/internal/database/online/sqlite/store.go
@@ -36,7 +36,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 }
 
 func (db *DB) GetByGroup(ctx context.Context, opt online.GetByGroupOpt) (dbutil.RowMap, error) {
-	panic("TODO: implement me")
+	return sqlutil.GetByGroup(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index b1b78460..109c02b1 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -6,13 +6,62 @@ import (
 	"fmt"
 
 	"github.com/jmoiron/sqlx"
-	"github.com/oom-ai/oomstore/pkg/errdefs"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/oom-ai/oomstore/pkg/oomstore/util"
 )
 
+func GetByGroup(ctx context.Context, db *sqlx.DB, opt online.GetByGroupOpt, backend types.BackendType) (dbutil.RowMap, error) {
+	if err := opt.Validate(); err != nil {
+		return nil, err
+	}
+
+	var tableName string
+	if opt.Group.Category == types.CategoryBatch {
+		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+	} else {
+		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+	}
+
+	qt := dbutil.QuoteFn(backend)
+	query := fmt.Sprintf(`SELECT * FROM %s WHERE %s = ?`, qt(tableName), qt(opt.Group.Entity.Name))
+
+	result := make(map[string]interface{})
+	if err := db.QueryRowxContext(ctx, db.Rebind(query), opt.EntityKey).MapScan(result); err != nil {
+		tableNotFound, notFoundErr := dbutil.IsTableNotFoundError(err, backend)
+		if notFoundErr != nil {
+			return nil, notFoundErr
+		}
+		if err == sql.ErrNoRows || tableNotFound {
+			return make(dbutil.RowMap), nil
+		}
+		return nil, errdefs.WithStack(err)
+	}
+
+	rs := map[string]interface{}{}
+	for k, v := range result {
+		if k == opt.Group.Entity.Name {
+			continue
+		}
+
+		featureFullName := util.ComposeFullFeatureName(opt.Group.Name, k)
+		feature, err := opt.GetFeature(nil, &featureFullName)
+		if err != nil {
+			return nil, err
+		}
+		deserializedValue, err := dbutil.DeserializeByValueType(v, feature.ValueType, backend)
+		if err != nil {
+			return nil, err
+		}
+
+		rs[featureFullName] = deserializedValue
+	}
+	return rs, nil
+}
+
 func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.BackendType) (dbutil.RowMap, error) {
 	if err := opt.Validate(); err != nil {
 		return nil, err
diff --git a/internal/database/online/types.go b/internal/database/online/types.go
index ba1cc7f7..972784b5 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -22,9 +22,13 @@ func (g *GetOpt) Validate() error {
 }
 
 type GetByGroupOpt struct {
-	EntityKey  string
-	Group      types.Group
-	GetFeature func(featureID int) (*types.Feature, error)
+	EntityKey string
+	Group     types.Group
+
+	// One of featureID and featureFullName must be nil and one must not be nil.
+	// If featureID is not nil, then use featureID to query the feature
+	// If featureFullName is not nil, then featureFullName is used to query feature
+	GetFeature func(featureID *int, featureFullName *string) (*types.Feature, error)
 
 	// Only works when get batch features, it should be nil when get stream features
 	RevisionID *int
diff --git a/pkg/oomstore/online_query.go b/pkg/oomstore/online_query.go
index 376af22c..9af81f2e 100644
--- a/pkg/oomstore/online_query.go
+++ b/pkg/oomstore/online_query.go
@@ -33,8 +33,14 @@ func (s *OomStore) OnlineGetByGroup(ctx context.Context, groupName string, entit
 	featureValues, err := s.online.GetByGroup(ctx, online.GetByGroupOpt{
 		EntityKey: entityKey,
 		Group:     *group,
-		GetFeature: func(id int) (*types.Feature, error) {
-			return s.metadata.GetCachedFeature(ctx, id)
+		GetFeature: func(id *int, fullName *string) (*types.Feature, error) {
+			if id != nil {
+				return s.metadata.GetCachedFeature(ctx, *id)
+			}
+			if fullName != nil {
+				return s.metadata.GetCachedFeatureByName(ctx, *fullName)
+			}
+			return nil, errdefs.Errorf("FeatureID and featureFullName must have one that is not null")
 		},
 		RevisionID: group.OnlineRevisionID,
 	})


Test patch:
diff --git a/internal/database/online/cassandra/store_test.go b/internal/database/online/cassandra/store_test.go
index b172546e..a35803e3 100644
--- a/internal/database/online/cassandra/store_test.go
+++ b/internal/database/online/cassandra/store_test.go
@@ -70,6 +70,10 @@ func TestMultiGet(t *testing.T) {
 	test_impl.TestMultiGet(t, prepareStore, destroyStore(KEYSPACE))
 }
 
+func TestGetByGroup(t *testing.T) {
+	test_impl.TestGetByGroup(t, prepareStore, destroyStore(KEYSPACE))
+}
+
 func TestPurgeRemovesSpecifiedRevision(t *testing.T) {
 	test_impl.TestPurgeRemovesSpecifiedRevision(t, prepareStore, destroyStore(KEYSPACE))
 }
diff --git a/internal/database/online/dynamodb/store_test.go b/internal/database/online/dynamodb/store_test.go
index 5e8c854f..afc6d601 100644
--- a/internal/database/online/dynamodb/store_test.go
+++ b/internal/database/online/dynamodb/store_test.go
@@ -61,6 +61,10 @@ func TestMultiGet(t *testing.T) {
 	test_impl.TestMultiGet(t, prepareStore, destroyStore(t))
 }
 
+func TestGetByGroup(t *testing.T) {
+	test_impl.TestGetByGroup(t, prepareStore, destroyStore(t))
+}
+
 func TestPurgeRemovesSpecifiedRevision(t *testing.T) {
 	test_impl.TestPurgeRemovesSpecifiedRevision(t, prepareStore, destroyStore(t))
 }
diff --git a/internal/database/online/mysql/store_test.go b/internal/database/online/mysql/store_test.go
index 954c1023..20d1b038 100644
--- a/internal/database/online/mysql/store_test.go
+++ b/internal/database/online/mysql/store_test.go
@@ -49,6 +49,10 @@ func TestMultiGet(t *testing.T) {
 	test_impl.TestMultiGet(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
 }
 
+func TestGetByGroup(t *testing.T) {
+	test_impl.TestGetByGroup(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
+}
+
 func TestPurgeRemovesSpecifiedRevision(t *testing.T) {
 	test_impl.TestPurgeRemovesSpecifiedRevision(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
 }
diff --git a/internal/database/online/postgres/store_test.go b/internal/database/online/postgres/store_test.go
index f1da7d0e..bb449d8b 100644
--- a/internal/database/online/postgres/store_test.go
+++ b/internal/database/online/postgres/store_test.go
@@ -49,6 +49,10 @@ func TestMultiGet(t *testing.T) {
 	test_impl.TestMultiGet(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
 
+func TestGetByGroup(t *testing.T) {
+	test_impl.TestGetByGroup(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
+}
+
 func TestPurgeRemovesSpecifiedRevision(t *testing.T) {
 	test_impl.TestPurgeRemovesSpecifiedRevision(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
diff --git a/internal/database/online/sqlite/store_test.go b/internal/database/online/sqlite/store_test.go
index 039f9d57..4b6a9da0 100644
--- a/internal/database/online/sqlite/store_test.go
+++ b/internal/database/online/sqlite/store_test.go
@@ -48,6 +48,10 @@ func TestMultiGet(t *testing.T) {
 	test_impl.TestMultiGet(t, prepareStore, destroyStore)
 }
 
+func TestGetByGroup(t *testing.T) {
+	test_impl.TestGetByGroup(t, prepareStore, destroyStore)
+}
+
 func TestPurgeRemovesSpecifiedRevision(t *testing.T) {
 	test_impl.TestPurgeRemovesSpecifiedRevision(t, prepareStore, destroyStore)
 }
diff --git a/internal/database/online/test_impl/get.go b/internal/database/online/test_impl/get.go
index 4df5fee1..31a71fea 100644
--- a/internal/database/online/test_impl/get.go
+++ b/internal/database/online/test_impl/get.go
@@ -130,9 +130,12 @@ func TestGetByGroup(t *testing.T, prepareStore PrepareStoreFn, destroyStore Dest
 			actual, err := store.GetByGroup(ctx, online.GetByGroupOpt{
 				EntityKey: r.EntityKey(),
 				Group:     s.Group,
-				GetFeature: func(featureID int) (*types.Feature, error) {
+				GetFeature: func(featureID *int, featureFullName *string) (*types.Feature, error) {
 					for _, feature := range s.Features {
-						if feature.ID == featureID {
+						if featureID != nil && feature.ID == *featureID {
+							return feature, nil
+						}
+						if featureFullName != nil && feature.FullName() == *featureFullName {
 							return feature, nil
 						}
 					}