Repository: oom-ai/oomstore
Pull request #833 (2021-12-28 02:48:35)

Jinghan/refactor `feature_group_revision` schema
This PR refactors `feature_group_revision` schema: 
- rename `data_table` to `snapshot_table`
- add new field `cdc_table`

related to #803 


Patch:
diff --git a/internal/database/metadata/mysql/db.go b/internal/database/metadata/mysql/db.go
index 5faeae9c..c92043e3 100644
--- a/internal/database/metadata/mysql/db.go
+++ b/internal/database/metadata/mysql/db.go
@@ -106,15 +106,15 @@ func (db *DB) ListGroup(ctx context.Context, entityID *int, groupIDs *[]int) (ty
 
 func (db *DB) CreateRevision(ctx context.Context, opt metadata.CreateRevisionOpt) (int, string, error) {
 	var (
-		revisionID int
-		dataTable  string
-		err        error
+		revisionID    int
+		snapshotTable string
+		err           error
 	)
 	err = db.WithTransaction(ctx, func(c context.Context, tx metadata.DBStore) error {
-		revisionID, dataTable, err = tx.CreateRevision(c, opt)
+		revisionID, snapshotTable, err = tx.CreateRevision(c, opt)
 		return err
 	})
-	return revisionID, dataTable, err
+	return revisionID, snapshotTable, err
 }
 
 func (db *DB) UpdateRevision(ctx context.Context, opt metadata.UpdateRevisionOpt) error {
diff --git a/internal/database/metadata/mysql/revision.go b/internal/database/metadata/mysql/revision.go
index 18874b0b..6656aef8 100644
--- a/internal/database/metadata/mysql/revision.go
+++ b/internal/database/metadata/mysql/revision.go
@@ -9,13 +9,13 @@ import (
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
-	var dataTable string
-	if opt.DataTable != nil {
-		dataTable = *opt.DataTable
+	var snapshotTable string
+	if opt.SnapshotTable != nil {
+		snapshotTable = *opt.SnapshotTable
 	}
 
-	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, data_table, anchored, description) VALUES (?, ?, ?, ?, ?)"
-	res, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(insertQuery), opt.GroupID, opt.Revision, dataTable, 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 {
@@ -29,10 +29,10 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		return 0, "", err
 	}
 
-	if opt.DataTable == nil {
-		updateQuery := "UPDATE feature_group_revision SET data_table = ? WHERE id = ?"
-		dataTable = fmt.Sprintf("offline_%d_%d", opt.GroupID, revisionID)
-		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), dataTable, revisionID)
+	if opt.SnapshotTable == nil {
+		updateQuery := "UPDATE feature_group_revision SET snapshot_table = ? WHERE id = ?"
+		snapshotTable = fmt.Sprintf("offline_%d_%d", opt.GroupID, revisionID)
+		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), snapshotTable, revisionID)
 		if err != nil {
 			return 0, "", err
 		}
@@ -44,5 +44,5 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 			return 0, "", fmt.Errorf("failed to update revision %d: revision not found", revisionID)
 		}
 	}
-	return int(revisionID), dataTable, nil
+	return int(revisionID), snapshotTable, nil
 }
diff --git a/internal/database/metadata/mysql/schema.go b/internal/database/metadata/mysql/schema.go
index bf764e33..ca0aa0e7 100644
--- a/internal/database/metadata/mysql/schema.go
+++ b/internal/database/metadata/mysql/schema.go
@@ -44,7 +44,8 @@ var META_TABLE_SCHEMAS = map[string]string{
 			id				INT 		NOT	NULL AUTO_INCREMENT PRIMARY KEY,
 			group_id    	INT         NOT	NULL,
 			revision    	BIGINT      NOT	NULL COMMENT "group data point-in-time epoch seconds",
-			data_table  	VARCHAR(64) NOT	NULL COMMENT "feature data table name",
+			snapshot_table  VARCHAR(64) NOT	NULL COMMENT "batch & streaming feature snapshot table name",
+			cdc_table    	VARCHAR(64) NOT	NULL DEFAULT '' COMMENT "streaming feature cdc table name",
 			anchored    	BOOLEAN     NOT	NULL,
 			description 	VARCHAR(64) DEFAULT '',
 			create_time 	TIMESTAMP   NOT	NULL DEFAULT CURRENT_TIMESTAMP,
diff --git a/internal/database/metadata/postgres/db.go b/internal/database/metadata/postgres/db.go
index 3e408653..307ff2e1 100644
--- a/internal/database/metadata/postgres/db.go
+++ b/internal/database/metadata/postgres/db.go
@@ -95,15 +95,15 @@ func (db *DB) GetFeatureByName(ctx context.Context, name string) (*types.Feature
 
 func (db *DB) CreateRevision(ctx context.Context, opt metadata.CreateRevisionOpt) (int, string, error) {
 	var (
-		revisionID int
-		dataTable  string
-		err        error
+		revisionID    int
+		snapshotTable string
+		err           error
 	)
 	err = db.WithTransaction(ctx, func(c context.Context, tx metadata.DBStore) error {
-		revisionID, dataTable, err = tx.CreateRevision(ctx, opt)
+		revisionID, snapshotTable, err = tx.CreateRevision(ctx, opt)
 		return err
 	})
-	return revisionID, dataTable, err
+	return revisionID, snapshotTable, err
 }
 
 func (db *DB) UpdateRevision(ctx context.Context, opt metadata.UpdateRevisionOpt) error {
diff --git a/internal/database/metadata/postgres/revision.go b/internal/database/metadata/postgres/revision.go
index cb7013ec..14b10361 100644
--- a/internal/database/metadata/postgres/revision.go
+++ b/internal/database/metadata/postgres/revision.go
@@ -10,14 +10,14 @@ import (
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
-	var dataTable string
-	if opt.DataTable != nil {
-		dataTable = *opt.DataTable
+	var snapshotTable string
+	if opt.SnapshotTable != nil {
+		snapshotTable = *opt.SnapshotTable
 	}
 
 	var revisionID int
-	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, data_table, anchored, description) VALUES ($1, $2, $3, $4, $5) RETURNING id"
-	if err := sqlxCtx.GetContext(ctx, &revisionID, insertQuery, opt.GroupID, opt.Revision, dataTable, 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)
@@ -25,10 +25,10 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		}
 		return 0, "", err
 	}
-	if opt.DataTable == nil {
-		updateQuery := "UPDATE feature_group_revision SET data_table = $1 WHERE id = $2"
-		dataTable = fmt.Sprintf("offline_%d_%d", opt.GroupID, revisionID)
-		result, err := sqlxCtx.ExecContext(ctx, updateQuery, dataTable, revisionID)
+	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)
+		result, err := sqlxCtx.ExecContext(ctx, updateQuery, snapshotTable, revisionID)
 		if err != nil {
 			return 0, "", err
 		}
@@ -41,5 +41,5 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		}
 	}
 
-	return revisionID, dataTable, nil
+	return revisionID, snapshotTable, nil
 }
diff --git a/internal/database/metadata/postgres/schema.go b/internal/database/metadata/postgres/schema.go
index 0f0afc37..b976b440 100644
--- a/internal/database/metadata/postgres/schema.go
+++ b/internal/database/metadata/postgres/schema.go
@@ -63,18 +63,20 @@ var META_TABLE_SCHEMAS = map[string]string{
 		`,
 	"feature_group_revision": `
 		CREATE TABLE feature_group_revision (
-			id INT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
-			group_id    INT         NOT     NULL,
-			revision    BIGINT      NOT     NULL,
-			data_table  VARCHAR(64) NOT     NULL,
-			anchored    BOOLEAN     NOT     NULL,
-			description VARCHAR(64) DEFAULT '',
-			create_time TIMESTAMP   NOT     NULL DEFAULT CURRENT_TIMESTAMP,
-			modify_time TIMESTAMP   NOT     NULL DEFAULT CURRENT_TIMESTAMP,
+			id    	    	INT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+			group_id    	INT         NOT NULL,
+			revision    	BIGINT      NOT NULL,
+			snapshot_table  VARCHAR(64) NOT	NULL,
+			cdc_table    	VARCHAR(64) NOT	NULL DEFAULT '',
+			anchored    	BOOLEAN     NOT NULL,
+			description 	VARCHAR(64) DEFAULT '',
+			create_time 	TIMESTAMP   NOT NULL DEFAULT CURRENT_TIMESTAMP,
+			modify_time 	TIMESTAMP   NOT NULL DEFAULT CURRENT_TIMESTAMP,
 			UNIQUE (group_id, revision)
 		);
 		COMMENT ON COLUMN feature_group_revision.revision   IS 'group data point-in-time epoch seconds';
-		COMMENT ON COLUMN feature_group_revision.data_table IS 'feature data table name';
+		COMMENT ON COLUMN feature_group_revision.snapshot_table IS 'batch & streaming feature snapshot table name';
+		COMMENT ON COLUMN feature_group_revision.cdc_table IS 'streaming feature cdc table name';
 		`,
 }
 
diff --git a/internal/database/metadata/sqlite/db.go b/internal/database/metadata/sqlite/db.go
index e32a8420..24fcc9fa 100644
--- a/internal/database/metadata/sqlite/db.go
+++ b/internal/database/metadata/sqlite/db.go
@@ -96,15 +96,15 @@ func (db *DB) ListGroup(ctx context.Context, entityID *int, groupIDs *[]int) (ty
 
 func (db *DB) CreateRevision(ctx context.Context, opt metadata.CreateRevisionOpt) (int, string, error) {
 	var (
-		revisionID int
-		dataTable  string
-		err        error
+		revisionID    int
+		snapshotTable string
+		err           error
 	)
 	err = db.WithTransaction(ctx, func(c context.Context, tx metadata.DBStore) error {
-		revisionID, dataTable, err = tx.CreateRevision(c, opt)
+		revisionID, snapshotTable, err = tx.CreateRevision(c, opt)
 		return err
 	})
-	return revisionID, dataTable, err
+	return revisionID, snapshotTable, err
 }
 
 func (db *DB) UpdateRevision(ctx context.Context, opt metadata.UpdateRevisionOpt) error {
diff --git a/internal/database/metadata/sqlite/revision.go b/internal/database/metadata/sqlite/revision.go
index b94ef813..ae327f92 100644
--- a/internal/database/metadata/sqlite/revision.go
+++ b/internal/database/metadata/sqlite/revision.go
@@ -9,13 +9,13 @@ import (
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
-	var dataTable string
-	if opt.DataTable != nil {
-		dataTable = *opt.DataTable
+	var snapshotTable string
+	if opt.SnapshotTable != nil {
+		snapshotTable = *opt.SnapshotTable
 	}
 
-	insertQuery := "INSERT INTO feature_group_revision(group_id, revision, data_table, anchored, description) VALUES (?, ?, ?, ?, ?)"
-	res, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(insertQuery), opt.GroupID, opt.Revision, dataTable, 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 {
@@ -29,10 +29,10 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		return 0, "", err
 	}
 
-	if opt.DataTable == nil {
-		updateQuery := "UPDATE feature_group_revision SET data_table = ? WHERE id = ?"
-		dataTable = fmt.Sprintf("offline_%d_%d", opt.GroupID, revisionID)
-		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), dataTable, revisionID)
+	if opt.SnapshotTable == nil {
+		updateQuery := "UPDATE feature_group_revision SET snapshot_table = ? WHERE id = ?"
+		snapshotTable = fmt.Sprintf("offline_%d_%d", opt.GroupID, revisionID)
+		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), snapshotTable, revisionID)
 		if err != nil {
 			return 0, "", err
 		}
@@ -44,5 +44,5 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 			return 0, "", fmt.Errorf("failed to update revision %d: revision not found", revisionID)
 		}
 	}
-	return int(revisionID), dataTable, nil
+	return int(revisionID), snapshotTable, nil
 }
diff --git a/internal/database/metadata/sqlite/schema.go b/internal/database/metadata/sqlite/schema.go
index bb3c495c..65cea474 100644
--- a/internal/database/metadata/sqlite/schema.go
+++ b/internal/database/metadata/sqlite/schema.go
@@ -55,7 +55,8 @@ var META_TABLE_SCHEMAS = map[string]string{
 			id				INTEGER 	NOT	NULL PRIMARY KEY AUTOINCREMENT,
 			group_id    	INT         NOT	NULL,
 			revision    	BIGINT      NOT	NULL,
-			data_table  	VARCHAR(64) NOT	NULL,
+			snapshot_table  VARCHAR(64) NOT	NULL,
+			cdc_table    	VARCHAR(64) NOT	NULL DEFAULT '',
 			anchored    	BOOLEAN     NOT	NULL,
 			description 	VARCHAR(64) DEFAULT '',
 			create_time 	TIMESTAMP   NOT	NULL DEFAULT CURRENT_TIMESTAMP,
diff --git a/internal/database/metadata/types.go b/internal/database/metadata/types.go
index 0a2c863a..2872377d 100644
--- a/internal/database/metadata/types.go
+++ b/internal/database/metadata/types.go
@@ -3,9 +3,9 @@ package metadata
 import "github.com/oom-ai/oomstore/pkg/oomstore/types"
 
 type RevisionRange struct {
-	MinRevision int64  `db:"min_revision"`
-	MaxRevision int64  `db:"max_revision"`
-	DataTable   string `db:"data_table"`
+	MinRevision   int64
+	MaxRevision   int64
+	SnapshotTable string
 }
 
 // Create
@@ -29,11 +29,11 @@ type CreateGroupOpt struct {
 }
 
 type CreateRevisionOpt struct {
-	Revision    int64
-	GroupID     int
-	DataTable   *string
-	Anchored    bool
-	Description string
+	Revision      int64
+	GroupID       int
+	SnapshotTable *string
+	Anchored      bool
+	Description   string
 }
 
 // Update
diff --git a/internal/database/offline/bigquery/export.go b/internal/database/offline/bigquery/export.go
index b78e6691..72348d22 100644
--- a/internal/database/offline/bigquery/export.go
+++ b/internal/database/offline/bigquery/export.go
@@ -13,7 +13,7 @@ import (
 
 func (db *DB) Export(ctx context.Context, opt offline.ExportOpt) (<-chan types.ExportRecord, <-chan error) {
 	fieldStr := dbutil.Quote("`", append([]string{opt.EntityName}, opt.Features.Names()...)...)
-	tableName := dbutil.Quote("`", opt.DataTable)
+	tableName := dbutil.Quote("`", opt.SnapshotTable)
 
 	q := fmt.Sprintf(`SELECT %s FROM %s.%s`, fieldStr, db.datasetID, tableName)
 	if opt.Limit != nil {
diff --git a/internal/database/offline/bigquery/import.go b/internal/database/offline/bigquery/import.go
index 4e7f1055..d22dd9f0 100644
--- a/internal/database/offline/bigquery/import.go
+++ b/internal/database/offline/bigquery/import.go
@@ -12,7 +12,7 @@ import (
 )
 
 func (db *DB) Import(ctx context.Context, opt offline.ImportOpt) (int64, error) {
-	tableID := opt.DataTableName
+	tableID := opt.SnapshotTableName
 
 	// Step 1: define table schema
 	schema := make(bigquery.Schema, 0, len(opt.Features)+1)
@@ -44,7 +44,7 @@ func (db *DB) Import(ctx context.Context, opt offline.ImportOpt) (int64, error)
 
 	// Step 2: create offline table
 	metaData := &bigquery.TableMetadata{
-		Name:   opt.DataTableName,
+		Name:   opt.SnapshotTableName,
 		Schema: schema,
 	}
 	tableRef := db.Dataset(db.datasetID).Table(tableID)
diff --git a/internal/database/offline/sqlutil/export.go b/internal/database/offline/sqlutil/export.go
index 34ad3cf1..b9d86201 100644
--- a/internal/database/offline/sqlutil/export.go
+++ b/internal/database/offline/sqlutil/export.go
@@ -24,7 +24,7 @@ func Export(ctx context.Context, db *sqlx.DB, opt offline.ExportOpt, backendType
 		errs <- err
 		return stream, errs
 	}
-	query := fmt.Sprintf("SELECT %s FROM %s", qt(fields...), qt(opt.DataTable))
+	query := fmt.Sprintf("SELECT %s FROM %s", qt(fields...), qt(opt.SnapshotTable))
 	if opt.Limit != nil {
 		query += fmt.Sprintf(" LIMIT %d", *opt.Limit)
 	}
diff --git a/internal/database/offline/sqlutil/import.go b/internal/database/offline/sqlutil/import.go
index a7424931..16718bd1 100644
--- a/internal/database/offline/sqlutil/import.go
+++ b/internal/database/offline/sqlutil/import.go
@@ -16,7 +16,7 @@ func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData Lo
 	var revision int64
 	err := dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 		// create the data table
-		schema, err := dbutil.BuildCreateSchema(opt.DataTableName, opt.Entity, opt.Features, backendType)
+		schema, err := dbutil.BuildCreateSchema(opt.SnapshotTableName, opt.Entity, opt.Features, backendType)
 		if err != nil {
 			return err
 		}
@@ -26,7 +26,7 @@ func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData Lo
 		}
 
 		// populate the data table
-		err = loadData(tx, ctx, opt.Source, opt.DataTableName, opt.Header)
+		err = loadData(tx, ctx, opt.Source, opt.SnapshotTableName, opt.Header)
 		if err != nil {
 			return err
 		}
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index 6dc1954d..495d53e7 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -117,7 +117,7 @@ func JoinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.JoinOneGr
 			UnixMilliStr:        unixMilliStr,
 			Columns:             columns,
 			EntityRowsTableName: opt.EntityRowsTableName,
-			DataTable:           r.DataTable,
+			SnapshotTable:       r.SnapshotTable,
 			Backend:             dbOpt.Backend,
 			DatasetID:           dbOpt.DatasetID,
 		})
diff --git a/internal/database/offline/sqlutil/join_helper.go b/internal/database/offline/sqlutil/join_helper.go
index 6eec3eed..d7621fdc 100644
--- a/internal/database/offline/sqlutil/join_helper.go
+++ b/internal/database/offline/sqlutil/join_helper.go
@@ -25,7 +25,7 @@ SELECT
 	{{ columnJoin .Columns }}
 FROM
 	{{ qt .EntityRowsTableName }} AS l
-LEFT JOIN {{ qt .DataTable }} AS r
+LEFT JOIN {{ qt .SnapshotTable }} AS r
 ON l.{{ .EntityKeyStr }} = r.{{ qt .EntityName }}
 WHERE l.{{ .UnixMilliStr }} >= ? AND l.{{ .UnixMilliStr }} < ?
 `
@@ -261,7 +261,7 @@ type joinQueryParams struct {
 	UnixMilliStr        string
 	Columns             []string
 	EntityRowsTableName string
-	DataTable           string
+	SnapshotTable       string
 	Backend             types.BackendType
 	DatasetID           *string
 }
@@ -270,7 +270,7 @@ func buildJoinQuery(params joinQueryParams) (string, error) {
 	if params.Backend == types.BackendBigQuery {
 		params.TableName = fmt.Sprintf("%s.%s", *params.DatasetID, params.TableName)
 		params.EntityRowsTableName = fmt.Sprintf("%s.%s", *params.DatasetID, params.EntityRowsTableName)
-		params.DataTable = fmt.Sprintf("%s.%s", *params.DatasetID, params.DataTable)
+		params.SnapshotTable = fmt.Sprintf("%s.%s", *params.DatasetID, params.SnapshotTable)
 	}
 
 	qt, err := dbutil.QuoteFn(params.Backend)
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index cc0ae008..11d3eea4 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -8,10 +8,10 @@ import (
 )
 
 type ExportOpt struct {
-	DataTable  string
-	EntityName string
-	Features   types.FeatureList
-	Limit      *uint64
+	SnapshotTable string
+	EntityName    string
+	Features      types.FeatureList
+	Limit         *uint64
 }
 
 type JoinOpt struct {
@@ -32,12 +32,12 @@ type JoinOneGroupOpt struct {
 }
 
 type ImportOpt struct {
-	Entity        *types.Entity
-	Features      types.FeatureList
-	Header        []string
-	Revision      *int64
-	DataTableName string
-	Source        *CSVSource
+	Entity            *types.Entity
+	Features          types.FeatureList
+	Header            []string
+	Revision          *int64
+	SnapshotTableName string
+	Source            *CSVSource
 }
 
 type CSVSource struct {
diff --git a/oomcli/cmd/types.go b/oomcli/cmd/types.go
index 45779bd7..539fcdfa 100644
--- a/oomcli/cmd/types.go
+++ b/oomcli/cmd/types.go
@@ -44,11 +44,12 @@ type FlattenFeature struct {
 }
 
 type FlattenRevision struct {
-	ID          int    `oomcli:"ID"`
-	Revision    int64  `oomcli:"REVISION"`
-	Group       string `oomcli:"GROUP"`
-	DataTable   string `oomcli:"DATA-TABLE"`
-	Description string `oomcli:"DESCRIPTION,truncate"`
+	ID            int    `oomcli:"ID"`
+	Revision      int64  `oomcli:"REVISION"`
+	Group         string `oomcli:"GROUP"`
+	SnapshotTable string `oomcli:"SNAPSHOT-TABLE"`
+	CdcTable      string `oomcli:"CDC-TABLE"`
+	Description   string `oomcli:"DESCRIPTION,truncate"`
 
 	Anchored   bool      `oomcli:"ANCHORED,wide"`
 	CreateTime time.Time `oomcli:"CREATE-TIME,wide"`
@@ -117,14 +118,14 @@ func parseTokenLists(i interface{}) (headerTokens TokenList, dataTokens []TokenL
 		element = FlattenRevision{}
 		for _, e := range s {
 			tokens, err = parseTokens(FlattenRevision{
-				ID:          e.ID,
-				Revision:    e.Revision,
-				Group:       e.Group.Name,
-				DataTable:   e.DataTable,
-				Description: e.Description,
-				Anchored:    e.Anchored,
-				CreateTime:  e.CreateTime,
-				ModifyTime:  e.ModifyTime,
+				ID:            e.ID,
+				Revision:      e.Revision,
+				Group:         e.Group.Name,
+				SnapshotTable: e.SnapshotTable,
+				Description:   e.Description,
+				Anchored:      e.Anchored,
+				CreateTime:    e.CreateTime,
+				ModifyTime:    e.ModifyTime,
 			})
 			if err != nil {
 				return
diff --git a/pkg/oomstore/export.go b/pkg/oomstore/export.go
index 3f8807bf..333a207a 100644
--- a/pkg/oomstore/export.go
+++ b/pkg/oomstore/export.go
@@ -54,10 +54,10 @@ func (s *OomStore) ChannelExport(ctx context.Context, opt types.ChannelExportOpt
 	}
 
 	stream, exportErr := s.offline.Export(ctx, offline.ExportOpt{
-		DataTable:  revision.DataTable,
-		EntityName: entity.Name,
-		Features:   features,
-		Limit:      opt.Limit,
+		SnapshotTable: revision.SnapshotTable,
+		EntityName:    entity.Name,
+		Features:      features,
+		Limit:         opt.Limit,
 	})
 	header := append([]string{entity.Name}, features.Names()...)
 	return types.NewExportResult(header, stream, exportErr), nil
diff --git a/pkg/oomstore/import.go b/pkg/oomstore/import.go
index 5072bafd..51956571 100644
--- a/pkg/oomstore/import.go
+++ b/pkg/oomstore/import.go
@@ -56,23 +56,23 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 		return 0, fmt.Errorf("csv header of the data source %v doesn't match the feature group schema %v", header, columnNames)
 	}
 
-	newRevisionID, dataTableName, err := s.metadata.CreateRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:    0,
-		GroupID:     opt.group.ID,
-		DataTable:   nil,
-		Description: opt.Description,
-		Anchored:    opt.Revision != nil,
+	newRevisionID, snapshotTableName, err := s.metadata.CreateRevision(ctx, metadata.CreateRevisionOpt{
+		Revision:      0,
+		GroupID:       opt.group.ID,
+		SnapshotTable: nil,
+		Description:   opt.Description,
+		Anchored:      opt.Revision != nil,
 	})
 	if err != nil {
 		return 0, err
 	}
 
 	revision, err := s.offline.Import(ctx, offline.ImportOpt{
-		Entity:        opt.entity,
-		Features:      opt.features,
-		Header:        header,
-		Revision:      opt.Revision,
-		DataTableName: dataTableName,
+		Entity:            opt.entity,
+		Features:          opt.features,
+		Header:            header,
+		Revision:          opt.Revision,
+		SnapshotTableName: snapshotTableName,
 		Source: &offline.CSVSource{
 			Reader:    reader,
 			Delimiter: dataSource.Delimiter,
@@ -127,11 +127,11 @@ func (s *OomStore) tableLinkImport(ctx context.Context, opt *importOpt, dataSour
 		revision = *opt.Revision
 	}
 	newRevisionID, _, err := s.metadata.CreateRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:    revision,
-		GroupID:     opt.group.ID,
-		DataTable:   &dataSource.TableName,
-		Description: opt.Description,
-		Anchored:    opt.Revision != nil,
+		Revision:      revision,
+		GroupID:       opt.group.ID,
+		SnapshotTable: &dataSource.TableName,
+		Description:   opt.Description,
+		Anchored:      opt.Revision != nil,
 	})
 	if err != nil {
 		return 0, err
diff --git a/pkg/oomstore/join.go b/pkg/oomstore/join.go
index 17514a35..ea22c76e 100644
--- a/pkg/oomstore/join.go
+++ b/pkg/oomstore/join.go
@@ -112,16 +112,16 @@ func (s *OomStore) buildRevisionRanges(ctx context.Context, groupID int) ([]*met
 	var ranges []*metadata.RevisionRange
 	for i := 1; i < len(revisions); i++ {
 		ranges = append(ranges, &metadata.RevisionRange{
-			MinRevision: revisions[i-1].Revision,
-			MaxRevision: revisions[i].Revision,
-			DataTable:   revisions[i-1].DataTable,
+			MinRevision:   revisions[i-1].Revision,
+			MaxRevision:   revisions[i].Revision,
+			SnapshotTable: revisions[i-1].SnapshotTable,
 		})
 	}
 
 	return append(ranges, &metadata.RevisionRange{
-		MinRevision: revisions[len(revisions)-1].Revision,
-		MaxRevision: math.MaxInt64,
-		DataTable:   revisions[len(revisions)-1].DataTable,
+		MinRevision:   revisions[len(revisions)-1].Revision,
+		MaxRevision:   math.MaxInt64,
+		SnapshotTable: revisions[len(revisions)-1].SnapshotTable,
 	}), nil
 }
 
diff --git a/pkg/oomstore/sync.go b/pkg/oomstore/sync.go
index 358271c0..86066584 100644
--- a/pkg/oomstore/sync.go
+++ b/pkg/oomstore/sync.go
@@ -39,9 +39,9 @@ func (s *OomStore) Sync(ctx context.Context, opt types.SyncOpt) error {
 
 	// Move data from offline to online store
 	exportStream, exportError := s.offline.Export(ctx, offline.ExportOpt{
-		DataTable:  revision.DataTable,
-		EntityName: group.Entity.Name,
-		Features:   features,
+		SnapshotTable: revision.SnapshotTable,
+		EntityName:    group.Entity.Name,
+		Features:      features,
 	})
 
 	if err = s.online.Import(ctx, online.ImportOpt{
diff --git a/pkg/oomstore/types/revision.go b/pkg/oomstore/types/revision.go
index 484898a1..66af0ecc 100644
--- a/pkg/oomstore/types/revision.go
+++ b/pkg/oomstore/types/revision.go
@@ -5,10 +5,11 @@ import (
 )
 
 type Revision struct {
-	ID        int    `db:"id"`
-	Revision  int64  `db:"revision"`
-	DataTable string `db:"data_table"`
-	Anchored  bool   `db:"anchored"`
+	ID            int    `db:"id"`
+	Revision      int64  `db:"revision"`
+	SnapshotTable string `db:"snapshot_table"`
+	CdcTable      string `db:"cdc_table"`
+	Anchored      bool   `db:"anchored"`
 
 	Description string    `db:"description"`
 	CreateTime  time.Time `db:"create_time"`


Test patch:
diff --git a/internal/database/metadata/test_impl/revision.go b/internal/database/metadata/test_impl/revision.go
index d88929f2..f2b45183 100644
--- a/internal/database/metadata/test_impl/revision.go
+++ b/internal/database/metadata/test_impl/revision.go
@@ -22,10 +22,10 @@ func TestCreateRevision(t *testing.T, prepareStore PrepareStoreFn, destroyStore
 	group, err := store.GetGroup(ctx, groupID)
 	require.NoError(t, err)
 	opt := metadata.CreateRevisionOpt{
-		GroupID:     groupID,
-		Revision:    1000,
-		DataTable:   stringPtr("device_info_20211028"),
-		Description: "description",
+		GroupID:       groupID,
+		Revision:      1000,
+		SnapshotTable: stringPtr("device_info_20211028"),
+		Description:   "description",
 	}
 
 	testCases := []struct {
@@ -41,13 +41,13 @@ func TestCreateRevision(t *testing.T, prepareStore PrepareStoreFn, destroyStore
 			expectedError: nil,
 			expected:      1,
 			expectedRevision: &types.Revision{
-				ID:          1,
-				Revision:    1000,
-				DataTable:   "device_info_20211028",
-				Anchored:    false,
-				Description: "description",
-				GroupID:     groupID,
-				Group:       group,
+				ID:            1,
+				Revision:      1000,
+				SnapshotTable: "device_info_20211028",
+				Anchored:      false,
+				Description:   "description",
+				GroupID:       groupID,
+				Group:         group,
 			},
 		},
 		{
@@ -60,13 +60,13 @@ func TestCreateRevision(t *testing.T, prepareStore PrepareStoreFn, destroyStore
 			expectedError: nil,
 			expected:      2,
 			expectedRevision: &types.Revision{
-				ID:          2,
-				Revision:    2000,
-				DataTable:   "offline_1_2",
-				Anchored:    false,
-				Description: "description",
-				GroupID:     groupID,
-				Group:       group,
+				ID:            2,
+				Revision:      2000,
+				SnapshotTable: "offline_1_2",
+				Anchored:      false,
+				Description:   "description",
+				GroupID:       groupID,
+				Group:         group,
 			},
 		},
 		{
@@ -103,10 +103,10 @@ func TestUpdateRevision(t *testing.T, prepareStore PrepareStoreFn, destroyStore
 
 	_, groupID := prepareEntityAndGroup(t, ctx, store)
 	revisionID, _, err := store.CreateRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:  1000,
-		GroupID:   groupID,
-		DataTable: stringPtr("device_info_1000"),
-		Anchored:  false,
+		Revision:      1000,
+		GroupID:       groupID,
+		SnapshotTable: stringPtr("device_info_1000"),
+		Anchored:      false,
 	})
 	require.NoError(t, err)
 
@@ -153,10 +153,10 @@ func TestGetRevision(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 
 	_, groupID := prepareEntityAndGroup(t, ctx, store)
 	revisionID, _, err := store.CreateRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:  1000,
-		GroupID:   groupID,
-		DataTable: stringPtr("device_info_1000"),
-		Anchored:  false,
+		Revision:      1000,
+		GroupID:       groupID,
+		SnapshotTable: stringPtr("device_info_1000"),
+		Anchored:      false,
 	})
 	require.NoError(t, err)
 
@@ -164,12 +164,12 @@ func TestGetRevision(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 	require.NoError(t, err)
 
 	revision := types.Revision{
-		ID:        revisionID,
-		Revision:  1000,
-		GroupID:   groupID,
-		DataTable: "device_info_1000",
-		Anchored:  false,
-		Group:     group,
+		ID:            revisionID,
+		Revision:      1000,
+		GroupID:       groupID,
+		SnapshotTable: "device_info_1000",
+		Anchored:      false,
+		Group:         group,
 	}
 
 	testCases := []struct {
@@ -216,10 +216,10 @@ func TestGetRevisionBy(t *testing.T, prepareStore PrepareStoreFn, destroyStore D
 
 	_, groupID := prepareEntityAndGroup(t, ctx, store)
 	revisionID, _, err := store.CreateRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:  1000,
-		GroupID:   groupID,
-		DataTable: stringPtr("device_info_1000"),
-		Anchored:  false,
+		Revision:      1000,
+		GroupID:       groupID,
+		SnapshotTable: stringPtr("device_info_1000"),
+		Anchored:      false,
 	})
 	require.NoError(t, err)
 
@@ -227,12 +227,12 @@ func TestGetRevisionBy(t *testing.T, prepareStore PrepareStoreFn, destroyStore D
 	require.NoError(t, err)
 
 	revision := types.Revision{
-		ID:        revisionID,
-		Revision:  1000,
-		GroupID:   groupID,
-		DataTable: "device_info_1000",
-		Anchored:  false,
-		Group:     group,
+		ID:            revisionID,
+		Revision:      1000,
+		GroupID:       groupID,
+		SnapshotTable: "device_info_1000",
+		Anchored:      false,
+		Group:         group,
 	}
 
 	testCases := []struct {
@@ -341,18 +341,18 @@ func prepareRevisions(t *testing.T, ctx context.Context, store metadata.Store) (
 	require.NoError(t, err)
 	require.NoError(t, store.Refresh())
 	revisionID1, _, err := store.CreateRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:  1000,
-		GroupID:   groupID,
-		DataTable: stringPtr("device_info_1000"),
-		Anchored:  false,
+		Revision:      1000,
+		GroupID:       groupID,
+		SnapshotTable: stringPtr("device_info_1000"),
+		Anchored:      false,
 	})
 	require.NoError(t, err)
 
 	revisionID2, _, err := store.CreateRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:  2000,
-		GroupID:   groupID,
-		DataTable: stringPtr("device_info_2000"),
-		Anchored:  false,
+		Revision:      2000,
+		GroupID:       groupID,
+		SnapshotTable: stringPtr("device_info_2000"),
+		Anchored:      false,
 	})
 	require.NoError(t, err)
 
@@ -361,21 +361,21 @@ func prepareRevisions(t *testing.T, ctx context.Context, store metadata.Store) (
 	require.NoError(t, err)
 
 	revision1 := &types.Revision{
-		ID:        revisionID1,
-		Revision:  1000,
-		GroupID:   groupID,
-		DataTable: "device_info_1000",
-		Anchored:  false,
-		Group:     group,
+		ID:            revisionID1,
+		Revision:      1000,
+		GroupID:       groupID,
+		SnapshotTable: "device_info_1000",
+		Anchored:      false,
+		Group:         group,
 	}
 
 	revision2 := &types.Revision{
-		ID:        revisionID2,
-		Revision:  2000,
-		GroupID:   groupID,
-		DataTable: "device_info_2000",
-		Anchored:  false,
-		Group:     group,
+		ID:            revisionID2,
+		Revision:      2000,
+		GroupID:       groupID,
+		SnapshotTable: "device_info_2000",
+		Anchored:      false,
+		Group:         group,
 	}
 
 	return entityID, groupID, []int{revisionID1, revisionID2}, types.RevisionList{revision1, revision2}
diff --git a/internal/database/offline/test_impl/export.go b/internal/database/offline/test_impl/export.go
index 504b7250..cab77571 100644
--- a/internal/database/offline/test_impl/export.go
+++ b/internal/database/offline/test_impl/export.go
@@ -16,7 +16,7 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 	ctx, store := prepareStore(t)
 	defer store.Close()
 
-	dataTable := "offline_1_1"
+	snapshotTable := "offline_1_1"
 	features := []*types.Feature{
 		{
 			Name:      "model",
@@ -27,7 +27,7 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 			ValueType: types.Int64,
 		},
 	}
-	buildTestDataTable(ctx, t, store, features, dataTable, &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, features, snapshotTable, &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
 1235,apple,200
 1236,huawei,300
@@ -44,18 +44,18 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 		{
 			description: "no features",
 			opt: offline.ExportOpt{
-				DataTable:  dataTable,
-				EntityName: "device",
-				Features:   types.FeatureList{},
+				SnapshotTable: snapshotTable,
+				EntityName:    "device",
+				Features:      types.FeatureList{},
 			},
 			expected: [][]interface{}{{"1234"}, {"1235"}, {"1236"}, {"1237"}},
 		},
 		{
 			description: "valid features and valid entity rows",
 			opt: offline.ExportOpt{
-				DataTable:  dataTable,
-				EntityName: "device",
-				Features:   features,
+				SnapshotTable: snapshotTable,
+				EntityName:    "device",
+				Features:      features,
 			},
 			expected: [][]interface{}{{"1234", "xiaomi", int64(100)}, {"1235", "apple", int64(200)}, {"1236", "huawei", int64(300)}, {"1237", "oneplus", int64(240)}},
 		},
diff --git a/internal/database/offline/test_impl/import.go b/internal/database/offline/test_impl/import.go
index 1e458740..dab66df6 100644
--- a/internal/database/offline/test_impl/import.go
+++ b/internal/database/offline/test_impl/import.go
@@ -23,11 +23,11 @@ func TestImport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 		Name:   "device",
 		Length: 16,
 	}
-	dataTable := "offline_1_1"
+	snapshotTable := "offline_1_1"
 
 	opt := offline.ImportOpt{
-		Entity:        &entity,
-		DataTableName: dataTable,
+		Entity:            &entity,
+		SnapshotTableName: snapshotTable,
 		Features: []*types.Feature{
 			{
 				Name:      "price",
@@ -71,8 +71,8 @@ func TestImport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 		assert.NoError(t, err)
 
 		stream, errch := store.Export(ctx, offline.ExportOpt{
-			DataTable:  dataTable,
-			EntityName: entity.Name,
+			SnapshotTable: snapshotTable,
+			EntityName:    entity.Name,
 			Features: []*types.Feature{
 				{Name: "model", ValueType: types.String},
 				{Name: "price", ValueType: types.Int64},
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index eb137327..41312658 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -28,19 +28,19 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 	oneGroupFeatures, oneGroupFeatureMap := prepareFeatures(true)
 	twoGroupFeatures, twoGroupFeatureMap := prepareFeatures(false)
 
-	buildTestDataTable(ctx, t, store, oneGroupFeatures, "offline_1_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_1_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
 1235,apple,200
 `)),
 		Delimiter: ",",
 	})
-	buildTestDataTable(ctx, t, store, oneGroupFeatures, "offline_1_2", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, "offline_1_2", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,galaxy,300
 1235,oneplus,240
 `)),
 		Delimiter: ",",
 	})
-	buildTestDataTable(ctx, t, store, twoGroupFeatures[2:], "offline_2_1", &offline.CSVSource{
+	buildTestSnapshotTable(ctx, t, store, twoGroupFeatures[2:], "offline_2_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,1
 1235,0
 `)),
@@ -160,21 +160,21 @@ func prepareFeatures(oneGroup bool) (types.FeatureList, map[string]types.Feature
 func prepareRevisionRanges(oneGroup bool) map[string][]*metadata.RevisionRange {
 	basic := []*metadata.RevisionRange{
 		{
-			MinRevision: 1,
-			MaxRevision: 15,
-			DataTable:   "offline_1_1",
+			MinRevision:   1,
+			MaxRevision:   15,
+			SnapshotTable: "offline_1_1",
 		},
 		{
-			MinRevision: 15,
-			MaxRevision: 25,
-			DataTable:   "offline_1_2",
+			MinRevision:   15,
+			MaxRevision:   25,
+			SnapshotTable: "offline_1_2",
 		},
 	}
 	advanced := []*metadata.RevisionRange{
 		{
-			MinRevision: 5,
-			MaxRevision: math.MaxInt64,
-			DataTable:   "offline_2_1",
+			MinRevision:   5,
+			MaxRevision:   math.MaxInt64,
+			SnapshotTable: "offline_2_1",
 		},
 	}
 	if oneGroup {
diff --git a/internal/database/offline/test_impl/util.go b/internal/database/offline/test_impl/util.go
index 965c3c8a..0b862077 100644
--- a/internal/database/offline/test_impl/util.go
+++ b/internal/database/offline/test_impl/util.go
@@ -13,7 +13,7 @@ type PrepareStoreFn func(t *testing.T) (context.Context, offline.Store)
 
 type DestroyStoreFn func()
 
-func buildTestDataTable(ctx context.Context, t *testing.T, store offline.Store, features []*types.Feature, dataTable 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,
@@ -23,11 +23,11 @@ func buildTestDataTable(ctx context.Context, t *testing.T, store offline.Store,
 		header = append(header, f.Name)
 	}
 	opt := offline.ImportOpt{
-		Entity:        entity,
-		DataTableName: dataTable,
-		Features:      features,
-		Header:        header,
-		Source:        source,
+		Entity:            entity,
+		SnapshotTableName: snapshotTable,
+		Features:          features,
+		Header:            header,
+		Source:            source,
 	}
 	_, err := store.Import(ctx, opt)
 	require.NoError(t, err)
diff --git a/oomcli/test/test_get_meta_revision.sh b/oomcli/test/test_get_meta_revision.sh
index 7e5f4ad5..742ba79c 100755
--- a/oomcli/test/test_get_meta_revision.sh
+++ b/oomcli/test/test_get_meta_revision.sh
@@ -8,8 +8,8 @@ import_sample > /dev/null
 
 case='oomcli get meta revision works'
 expected='
-ID,REVISION,GROUP,DATA-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
+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
 '
 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 da78f00f..bbcbb2e6 100755
--- a/oomcli/test/test_import_table_link.sh
+++ b/oomcli/test/test_import_table_link.sh
@@ -12,9 +12,9 @@ oomcli import \
     --table-link "offline_1_1" \
     --description 'linked table' > /dev/null
 expected='
-ID,REVISION,GROUP,DATA-TABLE,DESCRIPTION
-1,1639047117470,phone,offline_1_1,test data
-2,1639047117552,phone,offline_1_1,linked table
+ID,REVISION,GROUP,SNAPSHOT-TABLE,CDC-TABLE,DESCRIPTION
+1,1639047117470,phone,offline_1_1,,test data
+2,1639047117552,phone,offline_1_1,,linked table
 '
 filter() { cut -d, -f 1,3,4,5; }
 actual=$(oomcli get meta revision -o csv)
diff --git a/pkg/oomstore/export_test.go b/pkg/oomstore/export_test.go
index f43288c8..47fa1c4b 100644
--- a/pkg/oomstore/export_test.go
+++ b/pkg/oomstore/export_test.go
@@ -90,9 +90,9 @@ func TestChannelExport(t *testing.T) {
 		t.Run(tc.description, func(t *testing.T) {
 			metadataStore.EXPECT().Refresh().Return(nil).AnyTimes()
 			revision := types.Revision{
-				ID:        1,
-				GroupID:   1,
-				DataTable: "device_info_10",
+				ID:            1,
+				GroupID:       1,
+				SnapshotTable: "device_info_10",
 				Group: &types.Group{
 					Name:     "device_info",
 					ID:       1,
@@ -111,10 +111,10 @@ func TestChannelExport(t *testing.T) {
 			metadataStore.EXPECT().ListFeature(gomock.Any(), gomock.Any()).Return(features, nil)
 
 			offlineStore.EXPECT().Export(gomock.Any(), offline.ExportOpt{
-				DataTable:  "device_info_10",
-				EntityName: "device",
-				Features:   features,
-				Limit:      tc.opt.Limit,
+				SnapshotTable: "device_info_10",
+				EntityName:    "device",
+				Features:      features,
+				Limit:         tc.opt.Limit,
 			}).Return(tc.exportStream, tc.exportError)
 
 			// execute and compare results
diff --git a/pkg/oomstore/join_test.go b/pkg/oomstore/join_test.go
index f0dc8083..bfb88a94 100644
--- a/pkg/oomstore/join_test.go
+++ b/pkg/oomstore/join_test.go
@@ -31,12 +31,12 @@ func TestChannelJoin(t *testing.T) {
 	}
 	revisions := types.RevisionList{
 		{
-			Revision:  1,
-			DataTable: "device_basic_1",
+			Revision:      1,
+			SnapshotTable: "device_basic_1",
 		},
 		{
-			Revision:  20,
-			DataTable: "device_basic_20",
+			Revision:      20,
+			SnapshotTable: "device_basic_20",
 		},
 	}
 
diff --git a/pkg/oomstore/sync_test.go b/pkg/oomstore/sync_test.go
index 2f0bf13f..d414b072 100644
--- a/pkg/oomstore/sync_test.go
+++ b/pkg/oomstore/sync_test.go
@@ -75,9 +75,9 @@ func TestSync(t *testing.T) {
 
 				stream := make(chan types.ExportRecord)
 				offlineStore.EXPECT().Export(ctx, offline.ExportOpt{
-					DataTable:  "data-table-name",
-					EntityName: "device",
-					Features:   features,
+					SnapshotTable: "snapshot-table-name",
+					EntityName:    "device",
+					Features:      features,
 				}).Return(stream, nil)
 
 				onlineStore.EXPECT().Import(ctx, online.ImportOpt{
@@ -110,9 +110,9 @@ func TestSync(t *testing.T) {
 
 				stream := make(chan types.ExportRecord)
 				offlineStore.EXPECT().Export(ctx, offline.ExportOpt{
-					DataTable:  "data-table-name",
-					EntityName: "device",
-					Features:   features,
+					SnapshotTable: "snapshot-table-name",
+					EntityName:    "device",
+					Features:      features,
 				}).Return(stream, nil)
 
 				onlineStore.EXPECT().Import(ctx, online.ImportOpt{
@@ -160,6 +160,6 @@ func buildRevision() *types.Revision {
 			},
 			OnlineRevisionID: nil,
 		},
-		DataTable: "data-table-name",
+		SnapshotTable: "snapshot-table-name",
 	}
 }