Repository: oom-ai/oomstore
Pull request #834 (2021-12-28 05:34:10)

Refactor/online offline table name
this PR does:
1. rename offline batch table: `offline_<group_id>_<revision_id>` to `offline_batch_<group_id>_<revision>`
2. rename online batch table: `online_<revision_id>` to `online_batch_<revision_id>`
3. add online streamtable `online_stream_<group_id>`

relate #801 


Patch:
diff --git a/internal/database/metadata/mysql/revision.go b/internal/database/metadata/mysql/revision.go
index 6656aef8..fe597190 100644
--- a/internal/database/metadata/mysql/revision.go
+++ b/internal/database/metadata/mysql/revision.go
@@ -5,7 +5,9 @@ import (
 	"fmt"
 
 	"github.com/go-sql-driver/mysql"
+
 	"github.com/oom-ai/oomstore/internal/database/metadata"
+	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
@@ -31,7 +33,7 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 
 	if opt.SnapshotTable == nil {
 		updateQuery := "UPDATE feature_group_revision SET snapshot_table = ? WHERE id = ?"
-		snapshotTable = fmt.Sprintf("offline_%d_%d", opt.GroupID, revisionID)
+		snapshotTable = sqlutil.OfflineBatchTableName(opt.GroupID, revisionID)
 		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), snapshotTable, revisionID)
 		if err != nil {
 			return 0, "", err
diff --git a/internal/database/metadata/postgres/revision.go b/internal/database/metadata/postgres/revision.go
index 14b10361..0ebb2f75 100644
--- a/internal/database/metadata/postgres/revision.go
+++ b/internal/database/metadata/postgres/revision.go
@@ -7,6 +7,7 @@ import (
 	"github.com/jackc/pgerrcode"
 	"github.com/lib/pq"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
+	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
@@ -27,7 +28,7 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 	}
 	if opt.SnapshotTable == nil {
 		updateQuery := "UPDATE feature_group_revision SET snapshot_table = $1 WHERE id = $2"
-		snapshotTable = fmt.Sprintf("offline_%d_%d", opt.GroupID, revisionID)
+		snapshotTable = sqlutil.OfflineBatchTableName(opt.GroupID, int64(revisionID))
 		result, err := sqlxCtx.ExecContext(ctx, updateQuery, snapshotTable, revisionID)
 		if err != nil {
 			return 0, "", err
diff --git a/internal/database/metadata/sqlite/revision.go b/internal/database/metadata/sqlite/revision.go
index ae327f92..9a1d39d3 100644
--- a/internal/database/metadata/sqlite/revision.go
+++ b/internal/database/metadata/sqlite/revision.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/mattn/go-sqlite3"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
+	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
@@ -31,7 +32,7 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 
 	if opt.SnapshotTable == nil {
 		updateQuery := "UPDATE feature_group_revision SET snapshot_table = ? WHERE id = ?"
-		snapshotTable = fmt.Sprintf("offline_%d_%d", opt.GroupID, revisionID)
+		snapshotTable = sqlutil.OfflineBatchTableName(opt.GroupID, revisionID)
 		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), snapshotTable, revisionID)
 		if err != nil {
 			return 0, "", err
diff --git a/internal/database/metadata/sqlutil/util.go b/internal/database/metadata/sqlutil/util.go
new file mode 100644
index 00000000..609858c5
--- /dev/null
+++ b/internal/database/metadata/sqlutil/util.go
@@ -0,0 +1,7 @@
+package sqlutil
+
+import "fmt"
+
+func OfflineBatchTableName(groupID int, revisionID int64) string {
+	return fmt.Sprintf("offline_batch_%d_%d", groupID, revisionID)
+}
diff --git a/internal/database/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index 6838b653..5edb080e 100644
--- a/internal/database/online/cassandra/get.go
+++ b/internal/database/online/cassandra/get.go
@@ -14,7 +14,7 @@ import (
 )
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	tableName := sqlutil.OnlineTableName(opt.RevisionID)
+	tableName := sqlutil.OnlineBatchTableName(opt.RevisionID)
 
 	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`,
 		strings.Join(opt.FeatureList.Names(), ","),
@@ -39,7 +39,7 @@ 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.OnlineTableName(opt.RevisionID)
+		tableName    = sqlutil.OnlineBatchTableName(opt.RevisionID)
 		placeholders = getPlaceholders(len(opt.EntityKeys))
 	)
 
diff --git a/internal/database/online/cassandra/import.go b/internal/database/online/cassandra/import.go
index cc62b648..007f2a36 100644
--- a/internal/database/online/cassandra/import.go
+++ b/internal/database/online/cassandra/import.go
@@ -15,7 +15,7 @@ import (
 
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	columns := append([]string{opt.Entity.Name}, opt.FeatureList.Names()...)
-	tableName := sqlutil.OnlineTableName(opt.Revision.ID)
+	tableName := sqlutil.OnlineBatchTableName(opt.Revision.ID)
 
 	table, err := dbutil.BuildCreateSchema(tableName, opt.Entity, opt.FeatureList, types.BackendCassandra)
 	if err != nil {
diff --git a/internal/database/online/cassandra/purge.go b/internal/database/online/cassandra/purge.go
index 89d9aac7..72e6cfa4 100644
--- a/internal/database/online/cassandra/purge.go
+++ b/internal/database/online/cassandra/purge.go
@@ -8,6 +8,6 @@ import (
 )
 
 func (db *DB) Purge(ctx context.Context, revisionID int) error {
-	return db.Query(fmt.Sprintf("DROP TABLE IF EXISTS %s", sqlutil.OnlineTableName(revisionID))).
+	return db.Query(fmt.Sprintf("DROP TABLE IF EXISTS %s", sqlutil.OnlineBatchTableName(revisionID))).
 		WithContext(ctx).Exec()
 }
diff --git a/internal/database/online/dynamodb/get.go b/internal/database/online/dynamodb/get.go
index 41e91ef0..72665b26 100644
--- a/internal/database/online/dynamodb/get.go
+++ b/internal/database/online/dynamodb/get.go
@@ -20,7 +20,7 @@ const (
 )
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	tableName := sqlutil.OnlineTableName(opt.RevisionID)
+	tableName := sqlutil.OnlineBatchTableName(opt.RevisionID)
 	entityKeyValue, err := attributevalue.Marshal(opt.EntityKey)
 	if err != nil {
 		return nil, err
@@ -43,7 +43,7 @@ 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) {
 	res := make(map[string]dbutil.RowMap)
-	tableName := sqlutil.OnlineTableName(opt.RevisionID)
+	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/dynamodb/import.go b/internal/database/online/dynamodb/import.go
index 0732e0da..4649f55e 100644
--- a/internal/database/online/dynamodb/import.go
+++ b/internal/database/online/dynamodb/import.go
@@ -18,7 +18,7 @@ const (
 
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	// Step 1: create table
-	tableName := sqlutil.OnlineTableName(opt.Revision.ID)
+	tableName := sqlutil.OnlineBatchTableName(opt.Revision.ID)
 	_, err := db.CreateTable(ctx, &dynamodb.CreateTableInput{
 		TableName: aws.String(tableName),
 		KeySchema: []types.KeySchemaElement{
diff --git a/internal/database/online/dynamodb/purge.go b/internal/database/online/dynamodb/purge.go
index 38a1746d..6cb3e613 100644
--- a/internal/database/online/dynamodb/purge.go
+++ b/internal/database/online/dynamodb/purge.go
@@ -9,7 +9,7 @@ import (
 )
 
 func (db *DB) Purge(ctx context.Context, revisionID int) error {
-	tableName := sqlutil.OnlineTableName(revisionID)
+	tableName := sqlutil.OnlineBatchTableName(revisionID)
 	_, err := db.DeleteTable(ctx, &dynamodb.DeleteTableInput{
 		TableName: aws.String(tableName),
 	})
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index 4357cf55..b4b74065 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/jmoiron/sqlx"
+
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -13,7 +14,7 @@ import (
 
 func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.BackendType) (dbutil.RowMap, error) {
 	featureNames := opt.FeatureList.Names()
-	tableName := OnlineTableName(opt.RevisionID)
+	tableName := OnlineBatchTableName(opt.RevisionID)
 	qt, err := dbutil.QuoteFn(backend)
 	if err != nil {
 		return nil, err
@@ -38,7 +39,7 @@ 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) {
 	featureNames := opt.FeatureList.Names()
-	tableName := OnlineTableName(opt.RevisionID)
+	tableName := OnlineBatchTableName(opt.RevisionID)
 	qt, err := dbutil.QuoteFn(backend)
 	if err != nil {
 		return nil, err
diff --git a/internal/database/online/sqlutil/import.go b/internal/database/online/sqlutil/import.go
index 76fddf89..8f559f4a 100644
--- a/internal/database/online/sqlutil/import.go
+++ b/internal/database/online/sqlutil/import.go
@@ -14,7 +14,7 @@ func Import(ctx context.Context, db *sqlx.DB, opt online.ImportOpt, backend type
 	columns := append([]string{opt.Entity.Name}, opt.FeatureList.Names()...)
 	err := dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 		// create the data table
-		tableName := OnlineTableName(opt.Revision.ID)
+		tableName := OnlineBatchTableName(opt.Revision.ID)
 		schema, err := dbutil.BuildCreateSchema(tableName, opt.Entity, opt.FeatureList, backend)
 		if err != nil {
 			return err
diff --git a/internal/database/online/sqlutil/purge.go b/internal/database/online/sqlutil/purge.go
index 7986645a..0cf8df70 100644
--- a/internal/database/online/sqlutil/purge.go
+++ b/internal/database/online/sqlutil/purge.go
@@ -14,7 +14,7 @@ func Purge(ctx context.Context, db *sqlx.DB, revisionID int, backend types.Backe
 	if err != nil {
 		return err
 	}
-	query := fmt.Sprintf(`DROP TABLE IF EXISTS %s;`, qt(OnlineTableName(revisionID)))
+	query := fmt.Sprintf(`DROP TABLE IF EXISTS %s;`, qt(OnlineBatchTableName(revisionID)))
 	if _, err := db.ExecContext(ctx, query); err != nil {
 		return err
 	}
diff --git a/internal/database/online/sqlutil/util.go b/internal/database/online/sqlutil/util.go
index 91c44919..9b9bbf3f 100644
--- a/internal/database/online/sqlutil/util.go
+++ b/internal/database/online/sqlutil/util.go
@@ -8,8 +8,12 @@ import (
 
 const BatchSize = 10
 
-func OnlineTableName(revisionID int) string {
-	return fmt.Sprintf("online_%d", revisionID)
+func OnlineBatchTableName(revisionID int) string {
+	return fmt.Sprintf("online_batch_%d", revisionID)
+}
+
+func OnlineStreamTableName(groupID int) string {
+	return fmt.Sprintf("online_stream_%d", groupID)
 }
 
 func deserializeByTag(i interface{}, valueType types.ValueType, backend types.BackendType) (interface{}, error) {


Test patch:
diff --git a/internal/database/metadata/test_impl/revision.go b/internal/database/metadata/test_impl/revision.go
index f2b45183..8d977020 100644
--- a/internal/database/metadata/test_impl/revision.go
+++ b/internal/database/metadata/test_impl/revision.go
@@ -62,7 +62,7 @@ func TestCreateRevision(t *testing.T, prepareStore PrepareStoreFn, destroyStore
 			expectedRevision: &types.Revision{
 				ID:            2,
 				Revision:      2000,
-				SnapshotTable: "offline_1_2",
+				SnapshotTable: "offline_batch_1_2",
 				Anchored:      false,
 				Description:   "description",
 				GroupID:       groupID,
diff --git a/oomcli/test/test_get_meta_revision.sh b/oomcli/test/test_get_meta_revision.sh
index 742ba79c..29f2fe30 100755
--- a/oomcli/test/test_get_meta_revision.sh
+++ b/oomcli/test/test_get_meta_revision.sh
@@ -9,7 +9,7 @@ import_sample > /dev/null
 case='oomcli get meta revision works'
 expected='
 ID,REVISION,GROUP,SNAPSHOT-TABLE,CDC-TABLE,DESCRIPTION,ANCHORED,CREATE-TIME,MODIFY-TIME
-1,1638519905556,phone,offline_1_1,,test data,true,2021-12-03T08:25:05Z,2021-12-03T08:25:05Z
+1,1638519905556,phone,offline_batch_1_1,,test data,true,2021-12-03T08:25:05Z,2021-12-03T08:25:05Z
 '
 actual=$(oomcli get meta revision --group phone -o csv --wide)
 filter() { cut -d ',' -f 1,3-6 <<<"$1"; }
diff --git a/oomcli/test/test_import_table_link.sh b/oomcli/test/test_import_table_link.sh
index bbcbb2e6..9ba80805 100755
--- a/oomcli/test/test_import_table_link.sh
+++ b/oomcli/test/test_import_table_link.sh
@@ -9,12 +9,12 @@ import_sample
 case='oomcli import using table link'
 oomcli import \
     --group phone \
-    --table-link "offline_1_1" \
+    --table-link "offline_batch_1_1" \
     --description 'linked table' > /dev/null
 expected='
 ID,REVISION,GROUP,SNAPSHOT-TABLE,CDC-TABLE,DESCRIPTION
-1,1639047117470,phone,offline_1_1,,test data
-2,1639047117552,phone,offline_1_1,,linked table
+1,1639047117470,phone,offline_batch_1_1,,test data
+2,1639047117552,phone,offline_batch_1_1,,linked table
 '
 filter() { cut -d, -f 1,3,4,5; }
 actual=$(oomcli get meta revision -o csv)