Repository: oom-ai/oomstore
Pull request #868 (2021-12-30 08:59:14)

refactor: DRY refactor on backend
Define `Backend` once in each `store.go` and use it elsewhere.

wfxr: @YikSanChan You mean adding it to `Store` interface ?

Patch:
diff --git a/internal/database/metadata/sqlutil/util.go b/internal/database/metadata/sqlutil/util.go
index 19b7b2ec..d100f7e1 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, revision int64) string {
-	return fmt.Sprintf("offline_stream_snapshot_%d_%d", groupID, revision)
+func OfflineStreamSnapshotTableName(groupID int, revisionID int) string {
+	return fmt.Sprintf("offline_stream_snapshot_%d_%d", groupID, revisionID)
 }
 
-func OfflineStreamCdcTableName(groupID int, revision int64) string {
-	return fmt.Sprintf("offline_stream_cdc_%d_%d", groupID, revision)
+func OfflineStreamCdcTableName(groupID int, revisionID int) string {
+	return fmt.Sprintf("offline_stream_cdc_%d_%d", groupID, revisionID)
 }
diff --git a/internal/database/offline/bigquery/join.go b/internal/database/offline/bigquery/join.go
index 1243def0..139c24b4 100644
--- a/internal/database/offline/bigquery/join.go
+++ b/internal/database/offline/bigquery/join.go
@@ -17,7 +17,7 @@ import (
 
 func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult, error) {
 	dbOpt := dbutil.DBOpt{
-		Backend:    types.BackendBigQuery,
+		Backend:    Backend,
 		BigQueryDB: db.Client,
 		DatasetID:  &db.datasetID,
 	}
diff --git a/internal/database/offline/bigquery/store.go b/internal/database/offline/bigquery/store.go
index 7b86bf60..d4939576 100644
--- a/internal/database/offline/bigquery/store.go
+++ b/internal/database/offline/bigquery/store.go
@@ -16,7 +16,9 @@ import (
 	"google.golang.org/api/option"
 )
 
-const BackendType = types.BackendBigQuery
+const (
+	Backend = types.BackendBigQuery
+)
 
 var _ offline.Store = &DB{}
 
@@ -58,7 +60,7 @@ func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTab
 		if err != nil {
 			return nil, err
 		}
-		valueType, err := dbutil.ValueType(BackendType, cast.ToString(recordMap["data_type"]))
+		valueType, err := dbutil.ValueType(Backend, cast.ToString(recordMap["data_type"]))
 		if err != nil {
 			return nil, err
 		}
@@ -73,7 +75,7 @@ func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTab
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 	dbOpt := dbutil.DBOpt{
-		Backend:    types.BackendBigQuery,
+		Backend:    Backend,
 		BigQueryDB: db.Client,
 		DatasetID:  &db.datasetID,
 	}
diff --git a/internal/database/offline/mysql/store.go b/internal/database/offline/mysql/store.go
index 824cda4e..b0ed103f 100644
--- a/internal/database/offline/mysql/store.go
+++ b/internal/database/offline/mysql/store.go
@@ -11,7 +11,10 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const MySQLBatchSize = 20
+const (
+	Backend        = types.BackendMySQL
+	MySQLBatchSize = 20
+)
 
 var _ offline.Store = &DB{}
 
@@ -29,15 +32,15 @@ func Open(option *types.MySQLOpt) (*DB, error) {
 }
 
 func (db *DB) Import(ctx context.Context, opt offline.ImportOpt) (int64, error) {
-	return sqlutil.Import(ctx, db.DB, opt, dbutil.LoadDataFromSource(types.BackendMySQL, MySQLBatchSize), types.BackendMySQL)
+	return sqlutil.Import(ctx, db.DB, opt, dbutil.LoadDataFromSource(Backend, MySQLBatchSize), Backend)
 }
 
 func (db *DB) Export(ctx context.Context, opt offline.ExportOpt) (<-chan types.ExportRecord, <-chan error) {
-	return sqlutil.Export(ctx, db.DB, opt, types.BackendMySQL)
+	return sqlutil.Export(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult, error) {
-	return sqlutil.Join(ctx, db.DB, opt, types.BackendMySQL)
+	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
@@ -45,9 +48,6 @@ func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTab
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend: types.BackendMySQL,
-		SqlxDB:  db.DB,
-	}
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
 	return sqlutil.Snapshot(ctx, dbOpt, opt)
 }
diff --git a/internal/database/offline/postgres/store.go b/internal/database/offline/postgres/store.go
index 1104a43c..79c98e89 100644
--- a/internal/database/offline/postgres/store.go
+++ b/internal/database/offline/postgres/store.go
@@ -10,7 +10,9 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const BackendType = types.BackendPostgres
+const (
+	Backend = types.BackendPostgres
+)
 
 var _ offline.Store = &DB{}
 
@@ -28,15 +30,15 @@ func Open(option *types.PostgresOpt) (*DB, error) {
 }
 
 func (db *DB) Import(ctx context.Context, opt offline.ImportOpt) (int64, error) {
-	return sqlutil.Import(ctx, db.DB, opt, loadDataFromSource, types.BackendPostgres)
+	return sqlutil.Import(ctx, db.DB, opt, loadDataFromSource, Backend)
 }
 
 func (db *DB) Export(ctx context.Context, opt offline.ExportOpt) (<-chan types.ExportRecord, <-chan error) {
-	return sqlutil.Export(ctx, db.DB, opt, types.BackendPostgres)
+	return sqlutil.Export(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult, error) {
-	return sqlutil.Join(ctx, db.DB, opt, types.BackendPostgres)
+	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
@@ -44,13 +46,10 @@ func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTab
 	if err != nil {
 		return nil, err
 	}
-	return sqlutil.SqlxTableSchema(ctx, db, BackendType, rows)
+	return sqlutil.SqlxTableSchema(ctx, db, Backend, rows)
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend: types.BackendPostgres,
-		SqlxDB:  db.DB,
-	}
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
 	return sqlutil.Snapshot(ctx, dbOpt, opt)
 }
diff --git a/internal/database/offline/redshift/store.go b/internal/database/offline/redshift/store.go
index ec0581bc..233784e0 100644
--- a/internal/database/offline/redshift/store.go
+++ b/internal/database/offline/redshift/store.go
@@ -10,9 +10,10 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const BackendType = types.BackendRedshift
-
-const RedshiftBatchSize = 20
+const (
+	Backend           = types.BackendRedshift
+	RedshiftBatchSize = 20
+)
 
 var _ offline.Store = &DB{}
 
@@ -30,15 +31,15 @@ func Open(option *types.RedshiftOpt) (*DB, error) {
 }
 
 func (db *DB) Import(ctx context.Context, opt offline.ImportOpt) (int64, error) {
-	return sqlutil.Import(ctx, db.DB, opt, dbutil.LoadDataFromSource(BackendType, RedshiftBatchSize), BackendType)
+	return sqlutil.Import(ctx, db.DB, opt, dbutil.LoadDataFromSource(Backend, RedshiftBatchSize), Backend)
 }
 
 func (db *DB) Export(ctx context.Context, opt offline.ExportOpt) (<-chan types.ExportRecord, <-chan error) {
-	return sqlutil.Export(ctx, db.DB, opt, BackendType)
+	return sqlutil.Export(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult, error) {
-	return sqlutil.Join(ctx, db.DB, opt, BackendType)
+	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
@@ -46,13 +47,10 @@ func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTab
 	if err != nil {
 		return nil, err
 	}
-	return sqlutil.SqlxTableSchema(ctx, db, BackendType, rows)
+	return sqlutil.SqlxTableSchema(ctx, db, Backend, rows)
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend: types.BackendRedshift,
-		SqlxDB:  db.DB,
-	}
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
 	return sqlutil.Snapshot(ctx, dbOpt, opt)
 }
diff --git a/internal/database/offline/snowflake/store.go b/internal/database/offline/snowflake/store.go
index 6b9e78c4..9eb1ed23 100644
--- a/internal/database/offline/snowflake/store.go
+++ b/internal/database/offline/snowflake/store.go
@@ -12,7 +12,10 @@ import (
 	"github.com/snowflakedb/gosnowflake"
 )
 
-const SnowflakeBatchSize = 100
+const (
+	Backend            = types.BackendSnowflake
+	SnowflakeBatchSize = 100
+)
 
 var _ offline.Store = &DB{}
 
@@ -44,15 +47,15 @@ func Open(opt *types.SnowflakeOpt) (*DB, error) {
 }
 
 func (db *DB) Import(ctx context.Context, opt offline.ImportOpt) (int64, error) {
-	return sqlutil.Import(ctx, db.DB, opt, dbutil.LoadDataFromSource(types.BackendSnowflake, SnowflakeBatchSize), types.BackendSnowflake)
+	return sqlutil.Import(ctx, db.DB, opt, dbutil.LoadDataFromSource(Backend, SnowflakeBatchSize), Backend)
 }
 
 func (db *DB) Export(ctx context.Context, opt offline.ExportOpt) (<-chan types.ExportRecord, <-chan error) {
-	return sqlutil.Export(ctx, db.DB, opt, types.BackendSnowflake)
+	return sqlutil.Export(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult, error) {
-	return sqlutil.Join(ctx, db.DB, opt, types.BackendSnowflake)
+	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
@@ -60,9 +63,6 @@ func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTab
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend: types.BackendSnowflake,
-		SqlxDB:  db.DB,
-	}
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
 	return sqlutil.Snapshot(ctx, dbOpt, opt)
 }
diff --git a/internal/database/offline/sqlite/store.go b/internal/database/offline/sqlite/store.go
index 5ba1d326..b3910bf9 100644
--- a/internal/database/offline/sqlite/store.go
+++ b/internal/database/offline/sqlite/store.go
@@ -12,7 +12,10 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const SQLiteBatchSize = 20
+const (
+	Backend         = types.BackendSQLite
+	SQLiteBatchSize = 20
+)
 
 var _ offline.Store = &DB{}
 
@@ -30,15 +33,15 @@ func Open(option *types.SQLiteOpt) (*DB, error) {
 }
 
 func (db *DB) Import(ctx context.Context, opt offline.ImportOpt) (int64, error) {
-	return sqlutil.Import(ctx, db.DB, opt, dbutil.LoadDataFromSource(types.BackendSQLite, SQLiteBatchSize), types.BackendSQLite)
+	return sqlutil.Import(ctx, db.DB, opt, dbutil.LoadDataFromSource(Backend, SQLiteBatchSize), Backend)
 }
 
 func (db *DB) Export(ctx context.Context, opt offline.ExportOpt) (<-chan types.ExportRecord, <-chan error) {
-	return sqlutil.Export(ctx, db.DB, opt, types.BackendSQLite)
+	return sqlutil.Export(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult, error) {
-	return sqlutil.Join(ctx, db.DB, opt, types.BackendSQLite)
+	return sqlutil.Join(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error) {
@@ -46,9 +49,6 @@ func (db *DB) TableSchema(ctx context.Context, tableName string) (*types.DataTab
 }
 
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend: types.BackendSQLite,
-		SqlxDB:  db.DB,
-	}
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
 	return sqlutil.Snapshot(ctx, dbOpt, opt)
 }
diff --git a/internal/database/offline/sqlutil/snapshot.go b/internal/database/offline/sqlutil/snapshot.go
index 9ba53391..12095bef 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.PrevRevision)
-	currSnapshotTableName := sqlutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.Revision)
-	currCdcTableName := sqlutil.OfflineStreamCdcTableName(opt.Group.ID, opt.Revision)
+	prevSnapshotTableName := sqlutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.PrevRevisionID)
+	currSnapshotTableName := sqlutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.RevisionID)
+	currCdcTableName := sqlutil.OfflineStreamCdcTableName(opt.Group.ID, opt.RevisionID)
 
 	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 16bac9e7..15cb333c 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
-	Revision     int64
-	PrevRevision int64
+	Group          *types.Group
+	Features       types.FeatureList
+	RevisionID     int
+	PrevRevisionID int
 }
diff --git a/internal/database/online/cassandra/import.go b/internal/database/online/cassandra/import.go
index 007f2a36..c4905b2e 100644
--- a/internal/database/online/cassandra/import.go
+++ b/internal/database/online/cassandra/import.go
@@ -10,14 +10,13 @@ import (
 	"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"
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	columns := append([]string{opt.Entity.Name}, opt.FeatureList.Names()...)
 	tableName := sqlutil.OnlineBatchTableName(opt.Revision.ID)
 
-	table, err := dbutil.BuildCreateSchema(tableName, opt.Entity, opt.FeatureList, types.BackendCassandra)
+	table, err := dbutil.BuildCreateSchema(tableName, opt.Entity, opt.FeatureList, Backend)
 	if err != nil {
 		return err
 	}
diff --git a/internal/database/online/cassandra/store.go b/internal/database/online/cassandra/store.go
index e1a316d5..7b1bbda8 100644
--- a/internal/database/online/cassandra/store.go
+++ b/internal/database/online/cassandra/store.go
@@ -11,8 +11,8 @@ import (
 )
 
 const (
-	BatchSize   = 1000
-	BackendType = types.BackendCassandra
+	Backend   = types.BackendCassandra
+	BatchSize = 1000
 )
 
 var _ online.Store = &DB{}
diff --git a/internal/database/online/cassandra/stream.go b/internal/database/online/cassandra/stream.go
index 2524cccf..d5ec17c5 100644
--- a/internal/database/online/cassandra/stream.go
+++ b/internal/database/online/cassandra/stream.go
@@ -7,14 +7,13 @@ import (
 	"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"
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
 	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
 
 	if opt.Feature == nil {
-		schema, err := sqlutil.CreateStreamTableSchema(ctx, tableName, opt.Entity, types.BackendCassandra)
+		schema, err := sqlutil.CreateStreamTableSchema(ctx, tableName, opt.Entity, Backend)
 		if err != nil {
 			return err
 		}
@@ -22,7 +21,7 @@ func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTa
 		return db.Query(schema).WithContext(ctx).Exec()
 	}
 
-	dbValueType, err := dbutil.DBValueType(types.BackendCassandra, opt.Feature.ValueType)
+	dbValueType, err := dbutil.DBValueType(Backend, opt.Feature.ValueType)
 	if err != nil {
 		return err
 	}
diff --git a/internal/database/online/dynamodb/store.go b/internal/database/online/dynamodb/store.go
index 1bc8b664..2972e4f0 100644
--- a/internal/database/online/dynamodb/store.go
+++ b/internal/database/online/dynamodb/store.go
@@ -11,7 +11,9 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const BackendType = types.BackendDynamoDB
+const (
+	Backend = types.BackendDynamoDB
+)
 
 var _ online.Store = &DB{}
 
diff --git a/internal/database/online/mysql/store.go b/internal/database/online/mysql/store.go
index e0edf785..f281706b 100644
--- a/internal/database/online/mysql/store.go
+++ b/internal/database/online/mysql/store.go
@@ -13,7 +13,9 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const BackendType = types.BackendMySQL
+const (
+	Backend = types.BackendMySQL
+)
 
 var _ online.Store = &DB{}
 
@@ -31,26 +33,26 @@ func (db *DB) Ping(ctx context.Context) error {
 }
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	return sqlutil.Get(ctx, db.DB, opt, BackendType)
+	return sqlutil.Get(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
-	return sqlutil.MultiGet(ctx, db.DB, opt, BackendType)
+	return sqlutil.MultiGet(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
-	return sqlutil.Import(ctx, db.DB, opt, BackendType)
+	return sqlutil.Import(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Purge(ctx context.Context, revisionID int) error {
-	return sqlutil.Purge(ctx, db.DB, revisionID, BackendType)
+	return sqlutil.Purge(ctx, db.DB, revisionID, Backend)
 }
 
 // TODO: refactor with text/template
 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)
+	inserts, insertPlaceholders, updatePlaceholders, values, err := sqlutil.BuildPushCondition(opt, Backend)
 	if err != nil {
 		return err
 	}
@@ -67,5 +69,5 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
-	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, types.BackendMySQL)
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index ccd90e3e..d569eb88 100644
--- a/internal/database/online/postgres/store.go
+++ b/internal/database/online/postgres/store.go
@@ -13,7 +13,9 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const BackendType = types.BackendPostgres
+const (
+	Backend = types.BackendPostgres
+)
 
 var _ online.Store = &DB{}
 
@@ -31,25 +33,25 @@ func (db *DB) Ping(ctx context.Context) error {
 }
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	return sqlutil.Get(ctx, db.DB, opt, BackendType)
+	return sqlutil.Get(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
-	return sqlutil.MultiGet(ctx, db.DB, opt, BackendType)
+	return sqlutil.MultiGet(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
-	return sqlutil.Import(ctx, db.DB, opt, BackendType)
+	return sqlutil.Import(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Purge(ctx context.Context, revisionID int) error {
-	return sqlutil.Purge(ctx, db.DB, revisionID, BackendType)
+	return sqlutil.Purge(ctx, db.DB, revisionID, Backend)
 }
 
 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)
+	inserts, insertPlaceholders, updatePlaceholders, values, err := sqlutil.BuildPushCondition(opt, Backend)
 	if err != nil {
 		return err
 	}
@@ -67,5 +69,5 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
-	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, types.BackendPostgres)
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/redis/store.go b/internal/database/online/redis/store.go
index 969d3dab..a77107fb 100644
--- a/internal/database/online/redis/store.go
+++ b/internal/database/online/redis/store.go
@@ -9,7 +9,10 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const PipelineBatchSize = 10
+const (
+	Backend           = types.BackendRedis
+	PipelineBatchSize = 10
+)
 
 var _ online.Store = &DB{}
 
diff --git a/internal/database/online/sqlite/store.go b/internal/database/online/sqlite/store.go
index 334a8041..9ea088bb 100644
--- a/internal/database/online/sqlite/store.go
+++ b/internal/database/online/sqlite/store.go
@@ -11,7 +11,9 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const BackendType = types.BackendSQLite
+const (
+	Backend = types.BackendSQLite
+)
 
 var _ online.Store = &DB{}
 
@@ -29,19 +31,19 @@ func (db *DB) Ping(ctx context.Context) error {
 }
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
-	return sqlutil.Get(ctx, db.DB, opt, BackendType)
+	return sqlutil.Get(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
-	return sqlutil.MultiGet(ctx, db.DB, opt, BackendType)
+	return sqlutil.MultiGet(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
-	return sqlutil.Import(ctx, db.DB, opt, BackendType)
+	return sqlutil.Import(ctx, db.DB, opt, Backend)
 }
 
 func (db *DB) Purge(ctx context.Context, revisionID int) error {
-	return sqlutil.Purge(ctx, db.DB, revisionID, BackendType)
+	return sqlutil.Purge(ctx, db.DB, revisionID, Backend)
 }
 
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
@@ -49,5 +51,5 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
-	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, types.BackendSQLite)
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/tikv/store.go b/internal/database/online/tikv/store.go
index d7c75b2d..fdf3e4e2 100644
--- a/internal/database/online/tikv/store.go
+++ b/internal/database/online/tikv/store.go
@@ -10,7 +10,9 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-const BackendType = types.BackendTiKV
+const (
+	Backend = types.BackendTiKV
+)
 
 var _ online.Store = &DB{}
 
diff --git a/pkg/oomstore/snapshot.go b/pkg/oomstore/snapshot.go
index b96c1ed9..04b3dfb5 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.Revision)
+		tableName := sqlutil.OfflineStreamSnapshotTableName(group.ID, revision.ID)
 		if err = s.offline.Snapshot(ctx, offline.SnapshotOpt{
-			Group:        group,
-			Revision:     revisions[i].Revision,
-			PrevRevision: revisions[i-1].Revision,
+			Group:          group,
+			RevisionID:     revisions[i].ID,
+			PrevRevisionID: revisions[i-1].ID,
 		}); err != nil {
 			return err
 		}


Test patch:
diff --git a/internal/database/offline/test_impl/export.go b/internal/database/offline/test_impl/export.go
index 265acfab..cab77571 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, 1, snapshotTable, &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, features, 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 b1203465..b1a0d4af 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, 1, "offline_snapshot_1_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_snapshot_1_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
 1235,apple,200
 `)),
 		Delimiter: ",",
 	})
-	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, 2, "offline_snapshot_1_2", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_snapshot_1_2", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,galaxy,300
 1235,oneplus,240
 `)),
 		Delimiter: ",",
 	})
-	buildTestSnapshotTable(ctx, t, store, twoGroupFeatures[2:], 1, "offline_snapshot_2_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, twoGroupFeatures[2:], "offline_snapshot_2_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,1
 1235,0
 `)),
 		Delimiter: ",",
 	})
 
-	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), 1, "offline_cdc_1_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), "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), 2, "offline_cdc_1_2", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, append(oneGroupFeatures, unixMilli), "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 21299a94..d6e756f8 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, 1, "offline_stream_snapshot_1_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, features, "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), 2, "offline_stream_cdc_1_2", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, append(features, unixMilli), "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,
-		Revision:     2,
-		PrevRevision: 1,
+		Group:          group,
+		Features:       features,
+		RevisionID:     2,
+		PrevRevisionID: 1,
 	})
 	assert.NoError(t, err)
 
diff --git a/internal/database/offline/test_impl/util.go b/internal/database/offline/test_impl/util.go
index bc468741..0b862077 100644
--- a/internal/database/offline/test_impl/util.go
+++ b/internal/database/offline/test_impl/util.go
@@ -13,15 +13,7 @@ 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,
-	revision int64,
-	snapshotTable string,
-	source *offline.CSVSource,
-) {
+func buildTestSnapshotTable(ctx context.Context, t *testing.T, store offline.Store, features []*types.Feature, snapshotTable string, source *offline.CSVSource) {
 	entity := &types.Entity{
 		Name:   "device",
 		Length: 10,
@@ -36,7 +28,6 @@ func buildTestSnapshotTable(
 		Features:          features,
 		Header:            header,
 		Source:            source,
-		Revision:          &revision,
 	}
 	_, err := store.Import(ctx, opt)
 	require.NoError(t, err)