Repository: oom-ai/oomstore
Pull request #879 (2021-12-31 06:28:06)

Feat/online redis stream push
this PR does:
1. modify Redis batch/stream feature key encode add prefix b/s
2. modify PushOpt: replace `FeatureNames []string` with `FeatureList types.FeatureList`
3. Redis: implement push method

relate #801 


Patch:
diff --git a/internal/database/metadata/mysql/revision.go b/internal/database/metadata/mysql/revision.go
index 06f86c1d..fe597190 100644
--- a/internal/database/metadata/mysql/revision.go
+++ b/internal/database/metadata/mysql/revision.go
@@ -11,16 +11,13 @@ import (
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
-	var snapshotTable, cdcTable string
+	var snapshotTable string
 	if opt.SnapshotTable != nil {
 		snapshotTable = *opt.SnapshotTable
 	}
-	if opt.CdcTable != nil {
-		cdcTable = *opt.CdcTable
-	}
 
-	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, snapshot_table, cdc_table, anchored, description) VALUES (?, ?, ?, ?, ?, ?)"
-	res, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(insertQuery), opt.GroupID, opt.Revision, snapshotTable, cdcTable, opt.Anchored, opt.Description)
+	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, snapshot_table, anchored, description) VALUES (?, ?, ?, ?, ?)"
+	res, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(insertQuery), opt.GroupID, opt.Revision, snapshotTable, opt.Anchored, opt.Description)
 	if err != nil {
 		if er, ok := err.(*mysql.MySQLError); ok {
 			if er.Number == ER_DUP_ENTRY {
diff --git a/internal/database/metadata/postgres/revision.go b/internal/database/metadata/postgres/revision.go
index e6f85905..0ebb2f75 100644
--- a/internal/database/metadata/postgres/revision.go
+++ b/internal/database/metadata/postgres/revision.go
@@ -11,17 +11,14 @@ import (
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
-	var snapshotTable, cdcTable string
+	var snapshotTable string
 	if opt.SnapshotTable != nil {
 		snapshotTable = *opt.SnapshotTable
 	}
-	if opt.CdcTable != nil {
-		cdcTable = *opt.CdcTable
-	}
 
 	var revisionID int
-	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, snapshot_table, cdc_table, anchored, description) VALUES ($1, $2, $3, $4, $5, $6) RETURNING id"
-	if err := sqlxCtx.GetContext(ctx, &revisionID, insertQuery, opt.GroupID, opt.Revision, snapshotTable, cdcTable, opt.Anchored, opt.Description); err != nil {
+	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, snapshot_table, anchored, description) VALUES ($1, $2, $3, $4, $5) RETURNING id"
+	if err := sqlxCtx.GetContext(ctx, &revisionID, insertQuery, opt.GroupID, opt.Revision, snapshotTable, opt.Anchored, opt.Description); err != nil {
 		if e2, ok := err.(*pq.Error); ok {
 			if e2.Code == pgerrcode.UniqueViolation {
 				return 0, "", fmt.Errorf("revision already exists: groupID=%d, revision=%d", opt.GroupID, opt.Revision)
diff --git a/internal/database/metadata/sqlite/revision.go b/internal/database/metadata/sqlite/revision.go
index b40eb384..9a1d39d3 100644
--- a/internal/database/metadata/sqlite/revision.go
+++ b/internal/database/metadata/sqlite/revision.go
@@ -10,16 +10,13 @@ import (
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
-	var snapshotTable, cdcTable string
+	var snapshotTable string
 	if opt.SnapshotTable != nil {
 		snapshotTable = *opt.SnapshotTable
 	}
-	if opt.CdcTable != nil {
-		cdcTable = *opt.CdcTable
-	}
 
-	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, snapshot_table, cdc_table, anchored, description) VALUES (?, ?, ?, ?, ?, ?)"
-	res, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(insertQuery), opt.GroupID, opt.Revision, snapshotTable, cdcTable, opt.Anchored, opt.Description)
+	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, snapshot_table, anchored, description) VALUES (?, ?, ?, ?, ?)"
+	res, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(insertQuery), opt.GroupID, opt.Revision, snapshotTable, opt.Anchored, opt.Description)
 	if err != nil {
 		if sqliteErr, ok := err.(sqlite3.Error); ok {
 			if sqliteErr.ExtendedCode == sqlite3.ErrConstraintUnique {
diff --git a/internal/database/metadata/types.go b/internal/database/metadata/types.go
index b5a88195..657d69d8 100644
--- a/internal/database/metadata/types.go
+++ b/internal/database/metadata/types.go
@@ -26,7 +26,6 @@ type CreateRevisionOpt struct {
 	Revision      int64
 	GroupID       int
 	SnapshotTable *string
-	CdcTable      *string
 	Anchored      bool
 	Description   string
 }
diff --git a/internal/database/online/redis/get.go b/internal/database/online/redis/get.go
index 5b88c552..399efcfc 100644
--- a/internal/database/online/redis/get.go
+++ b/internal/database/online/redis/get.go
@@ -9,7 +9,7 @@ import (
 )
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	key, err := serializeRedisKey(opt.RevisionID, opt.EntityKey)
+	key, err := serializeRedisKeyForBatchFeature(opt.RevisionID, opt.EntityKey)
 	if err != nil {
 		return nil, err
 	}
diff --git a/internal/database/online/redis/import.go b/internal/database/online/redis/import.go
index eda01885..389e4ea9 100644
--- a/internal/database/online/redis/import.go
+++ b/internal/database/online/redis/import.go
@@ -19,8 +19,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 		}
 
 		entityKey, values := record[0], record[1:]
-
-		key, err := serializeRedisKey(opt.Revision.ID, entityKey)
+		key, err := serializeRedisKeyForBatchFeature(opt.Revision.ID, entityKey)
 		if err != nil {
 			return err
 		}
diff --git a/internal/database/online/redis/purge.go b/internal/database/online/redis/purge.go
index df40c9a9..83e7478b 100644
--- a/internal/database/online/redis/purge.go
+++ b/internal/database/online/redis/purge.go
@@ -11,7 +11,7 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 	if err != nil {
 		return nil
 	}
-	pattern := prefix + ":*"
+	pattern := "b" + prefix + ":*"
 
 	var cursor uint64
 	var keys []string
diff --git a/internal/database/online/redis/push.go b/internal/database/online/redis/push.go
new file mode 100644
index 00000000..904e2fff
--- /dev/null
+++ b/internal/database/online/redis/push.go
@@ -0,0 +1,37 @@
+package redis
+
+import (
+	"context"
+
+	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
+)
+
+func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
+	key, err := serializeRedisKeyForStreamFeature(opt.GroupID, opt.EntityKey)
+	if err != nil {
+		return err
+	}
+
+	featureValues := make(map[string]string)
+	for i, value := range opt.FeatureValues {
+		// omit nil feature value
+		if value == nil {
+			continue
+		}
+
+		featureValue, err := kvutil.SerializeByValueType(value, opt.FeatureList[i].ValueType)
+		if err != nil {
+			return err
+		}
+
+		featureID, err := kvutil.SerializeByValue(opt.FeatureList[i].ID)
+		if err != nil {
+			return err
+		}
+		featureValues[featureID] = featureValue
+	}
+
+	db.HSet(ctx, key, featureValues)
+	return nil
+}
diff --git a/internal/database/online/redis/store.go b/internal/database/online/redis/store.go
index a77107fb..9d843943 100644
--- a/internal/database/online/redis/store.go
+++ b/internal/database/online/redis/store.go
@@ -37,7 +37,3 @@ func Open(opt *types.RedisOpt) *DB {
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
 	return nil
 }
-
-func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
-	panic("Implement me!")
-}
diff --git a/internal/database/online/redis/util.go b/internal/database/online/redis/util.go
index eeeec703..06dbd275 100644
--- a/internal/database/online/redis/util.go
+++ b/internal/database/online/redis/util.go
@@ -6,8 +6,18 @@ import (
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
 )
 
-func serializeRedisKey(revisionID int, entityKey interface{}) (string, error) {
-	prefix, err := kvutil.SerializeByValue(revisionID)
+func serializeRedisKeyForBatchFeature(revisionID int, entityKey interface{}) (string, error) {
+	s, err := serializeRediskey(revisionID, entityKey)
+	return "b" + s, err
+}
+
+func serializeRedisKeyForStreamFeature(groupID int, entityKey interface{}) (string, error) {
+	s, err := serializeRediskey(groupID, entityKey)
+	return "s" + s, err
+}
+
+func serializeRediskey(prefixID int, entityKey interface{}) (string, error) {
+	prefix, err := kvutil.SerializeByValue(prefixID)
 	if err != nil {
 		return "", err
 	}
@@ -16,5 +26,6 @@ func serializeRedisKey(revisionID int, entityKey interface{}) (string, error) {
 	if err != nil {
 		return "", err
 	}
+
 	return fmt.Sprintf("%s:%s", prefix, suffix), nil
 }
diff --git a/internal/database/online/sqlutil/push.go b/internal/database/online/sqlutil/push.go
index 006b07b5..08b78aa8 100644
--- a/internal/database/online/sqlutil/push.go
+++ b/internal/database/online/sqlutil/push.go
@@ -21,12 +21,12 @@ func BuildPushCondition(opt online.PushOpt, backend types.BackendType) *PushCond
 	qt := dbutil.QuoteFn(backend)
 	cond := PushCondition{}
 
-	cond.Inserts = qt(append([]string{opt.Entity.Name}, opt.FeatureNames...)...)
+	cond.Inserts = qt(append([]string{opt.Entity.Name}, opt.FeatureList.Names()...)...)
 	cond.InsertValues = append([]interface{}{opt.EntityKey}, opt.FeatureValues...)
 	cond.InsertPlaceholders = dbutil.Fill(len(cond.InsertValues), "?", ",")
 
-	updatePlaceholders := make([]string, 0, len(opt.FeatureNames))
-	for _, name := range opt.FeatureNames {
+	updatePlaceholders := make([]string, 0, opt.FeatureList.Len())
+	for _, name := range opt.FeatureList.Names() {
 		updatePlaceholders = append(updatePlaceholders, fmt.Sprintf("%s=?", qt(name)))
 	}
 	cond.UpdatePlaceholders = strings.Join(updatePlaceholders, ",")
diff --git a/internal/database/online/types.go b/internal/database/online/types.go
index 6d35e63d..3ab84f10 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -30,7 +30,7 @@ type PushOpt struct {
 	Entity        *types.Entity
 	EntityKey     string
 	GroupID       int
-	FeatureNames  []string
+	FeatureList   types.FeatureList
 	FeatureValues []interface{}
 }
 


Test patch:
diff --git a/internal/database/online/redis/store_test.go b/internal/database/online/redis/store_test.go
index 4e6c20c2..32f8cddd 100644
--- a/internal/database/online/redis/store_test.go
+++ b/internal/database/online/redis/store_test.go
@@ -55,6 +55,10 @@ func TestPrepareStreamTable(t *testing.T) {
 	test_impl.TestPrepareStreamTable(t, prepareStore, destroyStore)
 }
 
+func TestPush(t *testing.T) {
+	test_impl.TestPush(t, prepareStore, destroyStore)
+}
+
 func TestPing(t *testing.T) {
 	test_impl.TestPing(t, prepareStore, destroyStore)
 }
diff --git a/internal/database/online/test_impl/stream.go b/internal/database/online/test_impl/stream.go
index 9dedfbf7..f0681249 100644
--- a/internal/database/online/test_impl/stream.go
+++ b/internal/database/online/test_impl/stream.go
@@ -3,9 +3,10 @@ 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"
-	"github.com/stretchr/testify/assert"
 )
 
 type StreamData struct {
@@ -127,7 +128,7 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 		Entity:        &entity,
 		EntityKey:     "user1",
 		GroupID:       group.ID,
-		FeatureNames:  []string{simpleStreamData.features[0].Name},
+		FeatureList:   types.FeatureList{feature1},
 		FeatureValues: []interface{}{"post1"},
 	}))
 
@@ -135,7 +136,7 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 		Entity:        &entity,
 		EntityKey:     "user1",
 		GroupID:       group.ID,
-		FeatureNames:  []string{simpleStreamData.features[0].Name},
+		FeatureList:   types.FeatureList{feature1},
 		FeatureValues: []interface{}{"post2"},
 	}))
 
@@ -149,7 +150,7 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 		Entity:        &entity,
 		EntityKey:     "user1",
 		GroupID:       group.ID,
-		FeatureNames:  []string{feature1.Name, feature2.Name},
+		FeatureList:   types.FeatureList{feature1, feature2},
 		FeatureValues: []interface{}{"post1", "post2"},
 	}))
 }