Repository: oom-ai/oomstore
Pull request #867 (2021-12-30 08:33:07)

refactor: rename offline snapshot and cdc table


Patch:
diff --git a/internal/database/database.go b/internal/database/database.go
index 0a9db934..8ebbe938 100644
--- a/internal/database/database.go
+++ b/internal/database/database.go
@@ -17,7 +17,6 @@ import (
 	offlinePG "github.com/oom-ai/oomstore/internal/database/offline/postgres"
 	offlineRedshift "github.com/oom-ai/oomstore/internal/database/offline/redshift"
 	offlineSnowflake "github.com/oom-ai/oomstore/internal/database/offline/snowflake"
-	offlineSQLite "github.com/oom-ai/oomstore/internal/database/offline/sqlite"
 
 	"github.com/oom-ai/oomstore/internal/database/online"
 	onlineCassandra "github.com/oom-ai/oomstore/internal/database/online/cassandra"
@@ -25,8 +24,6 @@ import (
 	onlineMySQL "github.com/oom-ai/oomstore/internal/database/online/mysql"
 	onlinePG "github.com/oom-ai/oomstore/internal/database/online/postgres"
 	onlineRedis "github.com/oom-ai/oomstore/internal/database/online/redis"
-	onlineSQLite "github.com/oom-ai/oomstore/internal/database/online/sqlite"
-	onlineTiKV "github.com/oom-ai/oomstore/internal/database/online/tikv"
 )
 
 func OpenOnlineStore(opt types.OnlineStoreConfig) (online.Store, error) {
@@ -41,10 +38,6 @@ func OpenOnlineStore(opt types.OnlineStoreConfig) (online.Store, error) {
 		return onlineDynamoDB.Open(opt.DynamoDB)
 	case types.BackendCassandra:
 		return onlineCassandra.Open(opt.Cassandra)
-	case types.BackendSQLite:
-		return onlineSQLite.Open(opt.SQLite)
-	case types.BackendTiKV:
-		return onlineTiKV.Open(opt.TiKV)
 	default:
 		return nil, fmt.Errorf("unsupported backend: %s", opt.Backend)
 	}
@@ -88,8 +81,6 @@ func OpenOfflineStore(ctx context.Context, opt types.OfflineStoreConfig) (offlin
 		return offlineBigQuery.Open(ctx, opt.BigQuery)
 	case types.BackendRedshift:
 		return offlineRedshift.Open(opt.Redshift)
-	case types.BackendSQLite:
-		return offlineSQLite.Open(opt.SQLite)
 	default:
 		return nil, fmt.Errorf("unsupported backend: %s", opt.Backend)
 	}
diff --git a/internal/database/dbutil/util.go b/internal/database/dbutil/util.go
index c90081a0..0a7d24cb 100644
--- a/internal/database/dbutil/util.go
+++ b/internal/database/dbutil/util.go
@@ -29,7 +29,7 @@ func RandString(n int) string {
 	return string(b)
 }
 
-func Fill(size int, elem string, sep string) string {
+func Placeholders(size int, elem string, sep string) string {
 	r := make([]string, 0, size)
 
 	for i := 0; i < size; i++ {
diff --git a/internal/database/metadata/sqlutil/util.go b/internal/database/metadata/sqlutil/util.go
index d100f7e1..19b7b2ec 100644
--- a/internal/database/metadata/sqlutil/util.go
+++ b/internal/database/metadata/sqlutil/util.go
@@ -6,10 +6,10 @@ func OfflineBatchTableName(groupID int, revisionID int64) string {
 	return fmt.Sprintf("offline_batch_%d_%d", groupID, revisionID)
 }
 
-func OfflineStreamSnapshotTableName(groupID int, revisionID int) string {
-	return fmt.Sprintf("offline_stream_snapshot_%d_%d", groupID, revisionID)
+func OfflineStreamSnapshotTableName(groupID int, revision int64) string {
+	return fmt.Sprintf("offline_stream_snapshot_%d_%d", groupID, revision)
 }
 
-func OfflineStreamCdcTableName(groupID int, revisionID int) string {
-	return fmt.Sprintf("offline_stream_cdc_%d_%d", groupID, revisionID)
+func OfflineStreamCdcTableName(groupID int, revision int64) string {
+	return fmt.Sprintf("offline_stream_cdc_%d_%d", groupID, revision)
 }
diff --git a/internal/database/offline/sqlutil/snapshot.go b/internal/database/offline/sqlutil/snapshot.go
index 12095bef..9ba53391 100644
--- a/internal/database/offline/sqlutil/snapshot.go
+++ b/internal/database/offline/sqlutil/snapshot.go
@@ -59,9 +59,9 @@ ON l.{{ qt .EntityName }} = r.{{ qt .EntityName }}
 `
 
 func Snapshot(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.SnapshotOpt) error {
-	prevSnapshotTableName := sqlutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.PrevRevisionID)
-	currSnapshotTableName := sqlutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.RevisionID)
-	currCdcTableName := sqlutil.OfflineStreamCdcTableName(opt.Group.ID, opt.RevisionID)
+	prevSnapshotTableName := sqlutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.PrevRevision)
+	currSnapshotTableName := sqlutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.Revision)
+	currCdcTableName := sqlutil.OfflineStreamCdcTableName(opt.Group.ID, opt.Revision)
 
 	if dbOpt.Backend == types.BackendBigQuery {
 		currSnapshotTableName = fmt.Sprintf("%s.%s", *dbOpt.DatasetID, currSnapshotTableName)
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index 15cb333c..16bac9e7 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -53,8 +53,8 @@ type CSVSource struct {
 }
 
 type SnapshotOpt struct {
-	Group          *types.Group
-	Features       types.FeatureList
-	RevisionID     int
-	PrevRevisionID int
+	Group        *types.Group
+	Features     types.FeatureList
+	Revision     int64
+	PrevRevision int64
 }
diff --git a/internal/database/online/mysql/store.go b/internal/database/online/mysql/store.go
index e0edf785..bf45b636 100644
--- a/internal/database/online/mysql/store.go
+++ b/internal/database/online/mysql/store.go
@@ -3,10 +3,10 @@ package mysql
 import (
 	"context"
 	"fmt"
+	"strings"
 
 	_ "github.com/go-sql-driver/mysql"
 	"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/internal/database/online/sqlutil"
@@ -50,19 +50,22 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
 
-	inserts, insertPlaceholders, updatePlaceholders, values, err := sqlutil.BuildPushCondition(opt, types.BackendMySQL)
-	if err != nil {
-		return err
+	insertColumns := append([]string{opt.Entity.Name}, opt.FeatureNames...)
+	insertValues := append([]interface{}{opt.EntityKey}, opt.FeatureValues...)
+
+	updateValues := opt.FeatureValues
+	updatePlaceholders := make([]string, 0, len(opt.FeatureNames))
+	for _, name := range opt.FeatureNames {
+		updatePlaceholders = append(updatePlaceholders, fmt.Sprintf("%s=?", name))
 	}
 
 	query := fmt.Sprintf(`INSERT INTO %s (%s) VALUES(%s) ON DUPLICATE KEY UPDATE %s`,
 		tableName,
-		inserts,
-		insertPlaceholders,
-		updatePlaceholders,
+		strings.Join(insertColumns, ","),
+		dbutil.Placeholders(len(insertColumns), "?", ","),
+		strings.Join(updatePlaceholders, ","),
 	)
-
-	_, err = db.ExecContext(ctx, db.Rebind(query), values...)
+	_, err := db.ExecContext(ctx, query, append(insertValues, updateValues...)...)
 	return err
 }
 
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index ccd90e3e..434676af 100644
--- a/internal/database/online/postgres/store.go
+++ b/internal/database/online/postgres/store.go
@@ -2,7 +2,6 @@ package postgres
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/jmoiron/sqlx"
 	_ "github.com/lib/pq"
@@ -47,23 +46,7 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 }
 
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
-	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
-
-	inserts, insertPlaceholders, updatePlaceholders, values, err := sqlutil.BuildPushCondition(opt, types.BackendPostgres)
-	if err != nil {
-		return err
-	}
-
-	query := fmt.Sprintf(`INSERT INTO %s(%s) VALUES(%s) ON CONFLICT ("%s") DO UPDATE SET %s`,
-		tableName,
-		inserts,
-		insertPlaceholders,
-		opt.Entity.Name,
-		updatePlaceholders,
-	)
-
-	_, err = db.ExecContext(ctx, db.Rebind(query), values...)
-	return err
+	panic("Implement me!")
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
diff --git a/internal/database/online/sqlutil/push.go b/internal/database/online/sqlutil/push.go
deleted file mode 100644
index 9cc063ba..00000000
--- a/internal/database/online/sqlutil/push.go
+++ /dev/null
@@ -1,32 +0,0 @@
-package sqlutil
-
-import (
-	"fmt"
-	"strings"
-
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-	"github.com/oom-ai/oomstore/internal/database/online"
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-)
-
-func BuildPushCondition(opt online.PushOpt, backend types.BackendType) (string, string, string, []interface{}, error) {
-	qt, err := dbutil.QuoteFn(backend)
-	if err != nil {
-		return "", "", "", nil, err
-	}
-
-	insertColumns := append([]string{opt.Entity.Name}, opt.FeatureNames...)
-	insertValues := append([]interface{}{opt.EntityKey}, opt.FeatureValues...)
-
-	updatePlaceholders := make([]string, 0, len(opt.FeatureNames))
-	for _, name := range opt.FeatureNames {
-		updatePlaceholders = append(updatePlaceholders, fmt.Sprintf("%s=?", qt(name)))
-	}
-
-	return qt(insertColumns...),
-		dbutil.Fill(len(insertColumns), "?", ","),
-		strings.Join(updatePlaceholders, ","),
-		append(insertValues, opt.FeatureValues...),
-		nil
-
-}
diff --git a/pkg/oomstore/snapshot.go b/pkg/oomstore/snapshot.go
index 04b3dfb5..b96c1ed9 100644
--- a/pkg/oomstore/snapshot.go
+++ b/pkg/oomstore/snapshot.go
@@ -27,11 +27,11 @@ func (s *OomStore) Snapshot(ctx context.Context, groupName string) error {
 		if revision.SnapshotTable != "" {
 			continue
 		}
-		tableName := sqlutil.OfflineStreamSnapshotTableName(group.ID, revision.ID)
+		tableName := sqlutil.OfflineStreamSnapshotTableName(group.ID, revision.Revision)
 		if err = s.offline.Snapshot(ctx, offline.SnapshotOpt{
-			Group:          group,
-			RevisionID:     revisions[i].ID,
-			PrevRevisionID: revisions[i-1].ID,
+			Group:        group,
+			Revision:     revisions[i].Revision,
+			PrevRevision: revisions[i-1].Revision,
 		}); err != nil {
 			return err
 		}
diff --git a/pkg/oomstore/types/config.go b/pkg/oomstore/types/config.go
index 67f8059f..660080b8 100644
--- a/pkg/oomstore/types/config.go
+++ b/pkg/oomstore/types/config.go
@@ -33,9 +33,7 @@ type OnlineStoreConfig struct {
 	MySQL     *MySQLOpt     `yaml:"mysql"`
 	DynamoDB  *DynamoDBOpt  `yaml:"dynamodb"`
 	Cassandra *CassandraOpt `yaml:"cassandra"`
-	SQLite    *SQLiteOpt    `yaml:"sqlite"`
 	TiDB      *MySQLOpt     `yaml:"tidb"`
-	TiKV      *TiKVOpt      `yaml:"tikv"`
 }
 
 type OfflineStoreConfig struct {
@@ -45,7 +43,6 @@ type OfflineStoreConfig struct {
 	Snowflake *SnowflakeOpt `yaml:"snowflake"`
 	BigQuery  *BigQueryOpt  `yaml:"bigquery"`
 	Redshift  *RedshiftOpt  `yaml:"redshift"`
-	SQLite    *SQLiteOpt    `yaml:"sqlite"`
 	TiDB      *MySQLOpt     `yaml:"tidb"`
 }
 
@@ -143,14 +140,6 @@ func (cfg *MetadataStoreConfig) Validate() error {
 		cfg.Backend = BackendMySQL
 		n++
 	}
-	if cfg.SQLite != nil {
-		cfg.Backend = BackendSQLite
-		n++
-	}
-	if cfg.TiDB != nil {
-		cfg.Backend = BackendMySQL
-		n++
-	}
 	if n != 1 {
 		return fmt.Errorf("require exactly one metadata store backend")
 	}
@@ -175,22 +164,6 @@ func (cfg *OnlineStoreConfig) Validate() error {
 		cfg.Backend = BackendDynamoDB
 		n++
 	}
-	if cfg.Cassandra != nil {
-		cfg.Backend = BackendCassandra
-		n++
-	}
-	if cfg.SQLite != nil {
-		cfg.Backend = BackendSQLite
-		n++
-	}
-	if cfg.TiDB != nil {
-		cfg.Backend = BackendMySQL
-		n++
-	}
-	if cfg.TiKV != nil {
-		cfg.Backend = BackendTiKV
-		n++
-	}
 	if n != 1 {
 		return fmt.Errorf("require exactly one online store backend")
 	}
@@ -211,22 +184,6 @@ func (cfg *OfflineStoreConfig) Validate() error {
 		cfg.Backend = BackendSnowflake
 		n++
 	}
-	if cfg.BigQuery != nil {
-		cfg.Backend = BackendBigQuery
-		n++
-	}
-	if cfg.Redshift != nil {
-		cfg.Backend = BackendRedshift
-		n++
-	}
-	if cfg.SQLite != nil {
-		cfg.Backend = BackendSQLite
-		n++
-	}
-	if cfg.TiDB != nil {
-		cfg.Backend = BackendMySQL
-		n++
-	}
 	if n != 1 {
 		return fmt.Errorf("require exactly one offline store backend")
 	}


Test patch:
diff --git a/internal/database/dbutil/util_test.go b/internal/database/dbutil/util_test.go
index 2dd35dea..930891fb 100644
--- a/internal/database/dbutil/util_test.go
+++ b/internal/database/dbutil/util_test.go
@@ -52,7 +52,7 @@ func TestPlaceholders(t *testing.T) {
 
 	for _, c := range cases {
 		t.Run(c.describe, func(t *testing.T) {
-			got := dbutil.Fill(c.size, c.elem, c.sep)
+			got := dbutil.Placeholders(c.size, c.elem, c.sep)
 			assert.Equal(t, c.want, got)
 		})
 	}
diff --git a/internal/database/offline/test_impl/export.go b/internal/database/offline/test_impl/export.go
index cab77571..265acfab 100644
--- a/internal/database/offline/test_impl/export.go
+++ b/internal/database/offline/test_impl/export.go
@@ -27,7 +27,7 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 			ValueType: types.Int64,
 		},
 	}
-	buildTestSnapshotTable(ctx, t, store, features, snapshotTable, &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, features, 1, snapshotTable, &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
 1235,apple,200
 1236,huawei,300
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index b1a0d4af..b1203465 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -34,33 +34,33 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 	oneGroupFeatures, oneGroupFeatureMap := prepareFeatures(true)
 	twoGroupFeatures, twoGroupFeatureMap := prepareFeatures(false)
 
-	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_snapshot_1_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, 1, "offline_snapshot_1_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
 1235,apple,200
 `)),
 		Delimiter: ",",
 	})
-	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_snapshot_1_2", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, 2, "offline_snapshot_1_2", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,galaxy,300
 1235,oneplus,240
 `)),
 		Delimiter: ",",
 	})
-	buildTestSnapshotTable(ctx, t, store, twoGroupFeatures[2:], "offline_snapshot_2_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, twoGroupFeatures[2:], 1, "offline_snapshot_2_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,1
 1235,0
 `)),
 		Delimiter: ",",
 	})
 
-	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), "offline_cdc_1_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), 1, "offline_cdc_1_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi-1,120,2
 1235,apple-2,115,14
 1234,xiaomi-1,130,10
 `)),
 		Delimiter: ",",
 	})
-	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), "offline_cdc_1_2", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), 2, "offline_cdc_1_2", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,galaxy-1,320,18
 `)),
 		Delimiter: ",",
diff --git a/internal/database/offline/test_impl/snapshot.go b/internal/database/offline/test_impl/snapshot.go
index d6e756f8..21299a94 100644
--- a/internal/database/offline/test_impl/snapshot.go
+++ b/internal/database/offline/test_impl/snapshot.go
@@ -33,7 +33,7 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 	}
 	features, _ := prepareFeatures(true)
 
-	buildTestSnapshotTable(ctx, t, store, features, "offline_stream_snapshot_1_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, features, 1, "offline_stream_snapshot_1_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
 1235,apple,200
 1236,oneplus,155
@@ -41,7 +41,7 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 		Delimiter: ",",
 	})
 
-	buildTestSnapshotTable(ctx, t, store, append(features, unixMilli), "offline_stream_cdc_1_2", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, append(features, unixMilli), 2, "offline_stream_cdc_1_2", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi-1,120,2
 1235,apple-2,115,14
 1234,xiaomi-1,130,10
@@ -52,10 +52,10 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 	})
 
 	err := store.Snapshot(ctx, offline.SnapshotOpt{
-		Group:          group,
-		Features:       features,
-		RevisionID:     2,
-		PrevRevisionID: 1,
+		Group:        group,
+		Features:     features,
+		Revision:     2,
+		PrevRevision: 1,
 	})
 	assert.NoError(t, err)
 
diff --git a/internal/database/offline/test_impl/util.go b/internal/database/offline/test_impl/util.go
index 0b862077..bc468741 100644
--- a/internal/database/offline/test_impl/util.go
+++ b/internal/database/offline/test_impl/util.go
@@ -13,7 +13,15 @@ type PrepareStoreFn func(t *testing.T) (context.Context, offline.Store)
 
 type DestroyStoreFn func()
 
-func buildTestSnapshotTable(ctx context.Context, t *testing.T, store offline.Store, features []*types.Feature, snapshotTable string, source *offline.CSVSource) {
+func buildTestSnapshotTable(
+	ctx context.Context,
+	t *testing.T,
+	store offline.Store,
+	features []*types.Feature,
+	revision int64,
+	snapshotTable string,
+	source *offline.CSVSource,
+) {
 	entity := &types.Entity{
 		Name:   "device",
 		Length: 10,
@@ -28,6 +36,7 @@ func buildTestSnapshotTable(ctx context.Context, t *testing.T, store offline.Sto
 		Features:          features,
 		Header:            header,
 		Source:            source,
+		Revision:          &revision,
 	}
 	_, err := store.Import(ctx, opt)
 	require.NoError(t, err)
diff --git a/internal/database/online/postgres/store_test.go b/internal/database/online/postgres/store_test.go
index e7ffbea0..48627621 100644
--- a/internal/database/online/postgres/store_test.go
+++ b/internal/database/online/postgres/store_test.go
@@ -57,10 +57,6 @@ func TestPrepareStreamTable(t *testing.T) {
 	test_impl.TestPrepareStreamTable(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
 
-func TestPush(t *testing.T) {
-	test_impl.TestPush(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
-}
-
 func TestPing(t *testing.T) {
 	test_impl.TestPing(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }