feature: online get by group

internal/database/metadata/informer/feature.go

@@ -8,23 +8,35 @@ import (
 type FeatureCache struct {
 	types.FeatureList
 	nameIdx map[string]*types.Feature
+	idIdx   map[int]*types.Feature
 }
 
 func NewFeatureCache(features types.FeatureList) *FeatureCache {
-	return &FeatureCache{FeatureList: features, nameIdx: nil}
+	return &FeatureCache{FeatureList: features, nameIdx: nil, idIdx: nil}
 }
 
 func (c *FeatureCache) Enrich(groupCache *GroupCache) {
 	nameIdx := make(map[string]*types.Feature)
+	idIdx := make(map[int]*types.Feature)
 
 	for _, f := range c.FeatureList {
 		f.Group = groupCache.Find(func(g *types.Group) bool {
 			return g.ID == f.GroupID
 		})
 		nameIdx[f.FullName()] = f
+		idIdx[f.ID] = f
 	}
 
 	c.nameIdx = nameIdx
+	c.idIdx = idIdx
+}
+
+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 {

internal/database/metadata/informer/group.go

@@ -34,3 +34,13 @@ func (c *GroupCache) Get(groupID int) *types.Group {
 	}
 	return groups[0]
 }
+
+func (c *GroupCache) GetByName(groupName string) *types.Group {
+	groups := c.GroupList.Filter(func(g *types.Group) bool {
+		return g.Name == groupName
+	})
+	if len(groups) == 0 {
+		return nil
+	}
+	return groups[0]
+}

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"
 )
 
@@ -98,3 +97,28 @@ func (f *Informer) GetCachedGroup(ctx context.Context, groupID int) (*types.Grou
 	}
 	return group, nil
 }
+
+func (f *Informer) GetCachedGroupByName(ctx context.Context, groupName string) (*types.Group, error) {
+	group := f.Cache().Groups.GetByName(groupName).Copy()
+	if group == nil {
+		return nil, errdefs.NotFound(errdefs.Errorf("group %s not found", groupName))
+	}
+	return group, nil
+}
+
+func (f *Informer) GetCachedFeature(ctx context.Context, featureID int) (*types.Feature, error) {
+	feature := f.Cache().Features.Get(featureID).Copy()
+	if feature == nil {
+		return nil, errdefs.NotFound(errdefs.Errorf("feature %d not found", featureID))
+	}
+	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
+}

internal/database/metadata/store.go

@@ -54,6 +54,9 @@ type DBStore interface {
 type CacheStore interface {
 	ListCachedFeature(ctx context.Context, opt ListCachedFeatureOpt) types.FeatureList
 	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

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) {
@@ -47,6 +48,52 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 	return rs, nil
 }
 
+func (db *DB) GetByGroup(ctx context.Context, opt online.GetByGroupOpt) (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)
+	}
+
+	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
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
 	if err := opt.Validate(); err != nil {

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))
 }

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 (
@@ -50,6 +51,60 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 	return deserializeFeatureValues(opt.Features, result.Item)
 }
 
+func (db *DB) GetByGroup(ctx context.Context, opt online.GetByGroupOpt) (dbutil.RowMap, error) {
+	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
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
 	if err := opt.Validate(); err != nil {

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))
 }