Repository: oom-ai/oomstore
Pull request #863 (2021-12-30 07:19:42)

feature: implement offline push api
TODO:
- [x] implement offline push api
- [x] implement offline create table api
- [x] Add unit tests

codecov[bot]: # [Codecov](https://codecov.io/gh/oom-ai/oomstore/pull/863?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) Report
> Merging [#863](https://codecov.io/gh/oom-ai/oomstore/pull/863?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) (c494938) into [main](https://codecov.io/gh/oom-ai/oomstore/commit/fc0f85d8058ec0544aa937ac875fe5470ca9e5e7?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) (fc0f85d) will **decrease** coverage by `0.68%`.
> The diff coverage is `0.00%`.

[![Impacted file tree graph](https://codecov.io/gh/oom-ai/oomstore/pull/863/graphs/tree.svg?width=650&height=150&src=pr&token=C59L7LTRM4&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai)](https://codecov.io/gh/oom-ai/oomstore/pull/863?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai)

```diff
@@            Coverage Diff             @@
##             main     #863      +/-   ##
==========================================
- Coverage   50.70%   50.01%   -0.69%     
==========================================
  Files          82       82              
  Lines        3282     3327      +45     
==========================================
  Hits         1664     1664              
- Misses       1424     1469      +45     
  Partials      194      194              
```


| [Impacted Files](https://codecov.io/gh/oom-ai/oomstore/pull/863?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) | Coverage Δ | |
|---|---|---|
| [internal/database/dbutil/database.go](https://codecov.io/gh/oom-ai/oomstore/pull/863/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvZGJ1dGlsL2RhdGFiYXNlLmdv) | `0.00% <0.00%> (ø)` | |
| [internal/database/dbutil/db\_opt.go](https://codecov.io/gh/oom-ai/oomstore/pull/863/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvZGJ1dGlsL2RiX29wdC5nbw==) | `0.00% <0.00%> (ø)` | |
| [internal/database/offline/bigquery/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/863/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb2ZmbGluZS9iaWdxdWVyeS9zdG9yZS5nbw==) | `64.00% <0.00%> (-8.73%)` | :arrow_down: |
| [internal/database/offline/mysql/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/863/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb2ZmbGluZS9teXNxbC9zdG9yZS5nbw==) | `68.00% <0.00%> (-21.48%)` | :arrow_down: |
| [internal/database/offline/postgres/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/863/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb2ZmbGluZS9wb3N0Z3Jlcy9zdG9yZS5nbw==) | `48.27% <0.00%> (-12.60%)` | :arrow_down: |
| [internal/database/offline/redshift/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/863/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb2ZmbGluZS9yZWRzaGlmdC9zdG9yZS5nbw==) | `37.93% <0.00%> (-9.90%)` | :arrow_down: |
| [internal/database/offline/snowflake/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/863/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb2ZmbGluZS9zbm93Zmxha2Uvc3RvcmUuZ28=) | `62.16% <0.00%> (-12.04%)` | :arrow_down: |
| [internal/database/offline/sqlite/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/863/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2Uvb2ZmbGluZS9zcWxpdGUvc3RvcmUuZ28=) | `56.00% <0.00%> (-17.69%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/oom-ai/oomstore/pull/863?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/oom-ai/oomstore/pull/863?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai). Last update [fc0f85d...c494938](https://codecov.io/gh/oom-ai/oomstore/pull/863?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai).


Patch:
diff --git a/internal/database/dbutil/data_loader.go b/internal/database/dbutil/data_loader.go
index 0c11f416..0ab18e10 100644
--- a/internal/database/dbutil/data_loader.go
+++ b/internal/database/dbutil/data_loader.go
@@ -12,16 +12,16 @@ import (
 )
 
 // Currying
-func LoadDataFromSource(backendType types.BackendType, batchSize int) func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList) error {
-	return func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList) error {
-		return loadDataFromSource(tx, ctx, source, tableName, header, features, backendType, batchSize)
+func LoadDataFromSource(backendType types.BackendType, batchSize int) func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string) error {
+	return func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string) error {
+		return loadDataFromSource(tx, ctx, source, tableName, header, backendType, batchSize)
 	}
 }
 
-func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList, backendType types.BackendType, batchSize int) error {
+func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, backendType types.BackendType, batchSize int) error {
 	records := make([]interface{}, 0, batchSize)
 	for {
-		record, err := ReadLine(source.Reader, source.Delimiter, features, backendType)
+		record, err := ReadLine(source.Reader, source.Delimiter)
 		if err == io.EOF {
 			break
 		}
@@ -45,36 +45,10 @@ func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSou
 	return nil
 }
 
-func ReadLine(reader *bufio.Reader, delimiter string, features types.FeatureList, backend types.BackendType) ([]interface{}, error) {
+func ReadLine(reader *bufio.Reader, delimiter string) ([]string, error) {
 	row, err := reader.ReadString('\n')
 	if err != nil {
 		return nil, err
 	}
-	rowSlice := strings.Split(strings.Trim(row, "\n"), delimiter)
-	line := make([]interface{}, 0, len(rowSlice))
-	for i, ele := range rowSlice {
-		if i == 0 || len(features) == 0 {
-			// entity_key doesn't need to change type
-			line = append(line, ele)
-		} else {
-			line = append(line, castElement(ele, features[i-1].ValueType, backend))
-		}
-	}
-	return line, nil
-}
-
-func castElement(s string, valueType types.ValueType, backend types.BackendType) interface{} {
-	if backend != types.BackendMySQL {
-		return s
-	}
-	if valueType != types.Bool {
-		return s
-	}
-	if s == "true" || s == "TRUE" {
-		return 1
-	}
-	if s == "false" || s == "FALSE" {
-		return 0
-	}
-	return s
+	return strings.Split(strings.Trim(row, "\n"), delimiter), nil
 }
diff --git a/internal/database/dbutil/database.go b/internal/database/dbutil/database.go
index 8c329af7..7bfd41d3 100644
--- a/internal/database/dbutil/database.go
+++ b/internal/database/dbutil/database.go
@@ -9,6 +9,8 @@ import (
 	"github.com/lib/pq"
 	"github.com/mattn/go-sqlite3"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/snowflakedb/gosnowflake"
+	"google.golang.org/api/googleapi"
 )
 
 func OpenSQLite(dbFile string) (*sqlx.DB, error) {
@@ -49,10 +51,11 @@ func DeserializeString(i interface{}, backend types.BackendType) string {
 	}
 }
 
+// TODO: Should return an error when bakcend is not supported ?
 func IsTableNotFoundError(err error, backend types.BackendType) bool {
 	switch backend {
+	// https://github.com/mattn/go-sqlite3/issues/244
 	case types.BackendSQLite:
-		// https://github.com/mattn/go-sqlite3/issues/244
 		if sqliteErr, ok := err.(sqlite3.Error); ok {
 			return sqliteErr.Code == sqlite3.ErrError
 		}
@@ -62,10 +65,19 @@ func IsTableNotFoundError(err error, backend types.BackendType) bool {
 		if e2, ok := err.(*mysql.MySQLError); ok {
 			return e2.Number == 1146
 		}
-	case types.BackendPostgres:
+	case types.BackendPostgres, types.BackendRedshift:
 		if e2, ok := err.(*pq.Error); ok {
 			return e2.Code == pgerrcode.UndefinedTable
 		}
+	case types.BackendSnowflake:
+		if e2, ok := err.(*gosnowflake.SnowflakeError); ok {
+			return e2.Number == gosnowflake.ErrObjectNotExistOrAuthorized
+		}
+	// https://cloud.google.com/bigquery/docs/error-messages
+	case types.BackendBigQuery:
+		if e2, ok := err.(*googleapi.Error); ok {
+			return e2.Code == 404
+		}
 	}
 	return false
 }
diff --git a/internal/database/dbutil/db_opt.go b/internal/database/dbutil/db_opt.go
index 2f53bbf9..c3124dd3 100644
--- a/internal/database/dbutil/db_opt.go
+++ b/internal/database/dbutil/db_opt.go
@@ -37,12 +37,12 @@ func (d *DBOpt) ExecContext(ctx context.Context, query string, args []interface{
 	}
 }
 
-func (d *DBOpt) BuildInsertQuery(tableName string, records []interface{}, columns []string, backend types.BackendType) (string, []interface{}, error) {
+func (d *DBOpt) BuildInsertQuery(tableName string, records []interface{}, columns []string) (string, []interface{}, error) {
 	if len(records) == 0 {
 		return "", nil, nil
 	}
 
-	qt := QuoteFn(backend)
+	qt := QuoteFn(d.Backend)
 	columnStr := qt(columns...)
 	tableName = qt(tableName)
 
diff --git a/internal/database/dbutil/sql.go b/internal/database/dbutil/sql.go
index 5dd920f3..1b17a736 100644
--- a/internal/database/dbutil/sql.go
+++ b/internal/database/dbutil/sql.go
@@ -27,8 +27,8 @@ func BuildConditions(equal map[string]interface{}, in map[string]interface{}) ([
 	return cond, args, nil
 }
 
-func InsertRecordsToTable(ctx context.Context, dbOpt DBOpt, tableName string, records []interface{}, columns []string, backend types.BackendType) error {
-	query, args, err := dbOpt.BuildInsertQuery(tableName, records, columns, backend)
+func InsertRecordsToTable(ctx context.Context, dbOpt DBOpt, tableName string, records []interface{}, columns []string) error {
+	query, args, err := dbOpt.BuildInsertQuery(tableName, records, columns)
 	if err != nil {
 		return err
 	}
@@ -43,7 +43,7 @@ func InsertRecordsToTableTx(tx *sqlx.Tx, ctx context.Context, tableName string,
 	dbOpt := DBOpt{
 		Backend: backendType,
 	}
-	query, args, err := dbOpt.BuildInsertQuery(tableName, records, columns, backendType)
+	query, args, err := dbOpt.BuildInsertQuery(tableName, records, columns)
 	if err != nil {
 		return err
 	}
diff --git a/internal/database/dbutil/util.go b/internal/database/dbutil/util.go
index c90081a0..d6b0a236 100644
--- a/internal/database/dbutil/util.go
+++ b/internal/database/dbutil/util.go
@@ -38,3 +38,15 @@ func Fill(size int, elem string, sep string) string {
 
 	return strings.Join(r, sep)
 }
+
+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 OfflineStreamCdcTableName(groupID int, revision int64) string {
+	return fmt.Sprintf("offline_stream_cdc_%d_%d", groupID, revision)
+}
diff --git a/internal/database/metadata/mysql/revision.go b/internal/database/metadata/mysql/revision.go
index 06f86c1d..d6bd23d3 100644
--- a/internal/database/metadata/mysql/revision.go
+++ b/internal/database/metadata/mysql/revision.go
@@ -6,8 +6,8 @@ import (
 
 	"github.com/go-sql-driver/mysql"
 
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"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) {
@@ -36,7 +36,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 = sqlutil.OfflineBatchTableName(opt.GroupID, revisionID)
+		snapshotTable = dbutil.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 e6f85905..25abc1b4 100644
--- a/internal/database/metadata/postgres/revision.go
+++ b/internal/database/metadata/postgres/revision.go
@@ -6,8 +6,8 @@ import (
 
 	"github.com/jackc/pgerrcode"
 	"github.com/lib/pq"
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"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 +31,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 = sqlutil.OfflineBatchTableName(opt.GroupID, int64(revisionID))
+		snapshotTable = dbutil.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 b40eb384..5f9e79c2 100644
--- a/internal/database/metadata/sqlite/revision.go
+++ b/internal/database/metadata/sqlite/revision.go
@@ -5,8 +5,8 @@ import (
 	"fmt"
 
 	"github.com/mattn/go-sqlite3"
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"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) {
@@ -35,7 +35,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 = sqlutil.OfflineBatchTableName(opt.GroupID, revisionID)
+		snapshotTable = dbutil.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
deleted file mode 100644
index 19b7b2ec..00000000
--- a/internal/database/metadata/sqlutil/util.go
+++ /dev/null
@@ -1,15 +0,0 @@
-package sqlutil
-
-import "fmt"
-
-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 OfflineStreamCdcTableName(groupID int, revision int64) string {
-	return fmt.Sprintf("offline_stream_cdc_%d_%d", groupID, revision)
-}
diff --git a/internal/database/offline/bigquery/store.go b/internal/database/offline/bigquery/store.go
index d0d59f81..52ac839a 100644
--- a/internal/database/offline/bigquery/store.go
+++ b/internal/database/offline/bigquery/store.go
@@ -4,15 +4,13 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/oom-ai/oomstore/internal/database/offline/sqlutil"
-
-	"github.com/spf13/cast"
-	"google.golang.org/api/iterator"
-
 	"cloud.google.com/go/bigquery"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/offline"
+	"github.com/oom-ai/oomstore/internal/database/offline/sqlutil"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/spf13/cast"
+	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
 )
 
@@ -85,6 +83,14 @@ func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 func (db *DB) CreateTable(ctx context.Context, opt offline.CreateTableOpt) error {
 	schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, opt.IsCDC, opt.Features, nil, Backend)
 	if _, err := db.Client.Query(schema).Run(ctx); err != nil {
+		return nil
+	}
+	return nil
+}
+
+func (db *DB) Push(ctx context.Context, opt offline.PushOpt) error {
+	dbOpt := dbutil.DBOpt{Backend: Backend, BigQueryDB: db.Client, DatasetID: &db.datasetID}
+	if err := sqlutil.Push(ctx, dbOpt, opt); err != nil {
 		return err
 	}
 	return nil
diff --git a/internal/database/offline/mock_offline/store.go b/internal/database/offline/mock_offline/store.go
index ebac294d..cfc256fe 100644
--- a/internal/database/offline/mock_offline/store.go
+++ b/internal/database/offline/mock_offline/store.go
@@ -123,6 +123,20 @@ func (mr *MockStoreMockRecorder) Ping(ctx interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Ping", reflect.TypeOf((*MockStore)(nil).Ping), ctx)
 }
 
+// Push mocks base method.
+func (m *MockStore) Push(ctx context.Context, opt offline.PushOpt) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Push", ctx, opt)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Push indicates an expected call of Push.
+func (mr *MockStoreMockRecorder) Push(ctx, opt interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Push", reflect.TypeOf((*MockStore)(nil).Push), ctx, opt)
+}
+
 // Snapshot mocks base method.
 func (m *MockStore) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 	m.ctrl.T.Helper()
diff --git a/internal/database/offline/mysql/store.go b/internal/database/offline/mysql/store.go
index 2f50dd2f..769db875 100644
--- a/internal/database/offline/mysql/store.go
+++ b/internal/database/offline/mysql/store.go
@@ -58,3 +58,11 @@ func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 func (db *DB) CreateTable(ctx context.Context, opt offline.CreateTableOpt) error {
 	return sqlutil.CreateTable(ctx, db.DB, opt, Backend)
 }
+
+func (db *DB) Push(ctx context.Context, opt offline.PushOpt) error {
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
+	if err := sqlutil.Push(ctx, dbOpt, opt); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/internal/database/offline/postgres/store.go b/internal/database/offline/postgres/store.go
index 06a47f70..75820333 100644
--- a/internal/database/offline/postgres/store.go
+++ b/internal/database/offline/postgres/store.go
@@ -57,3 +57,11 @@ func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 func (db *DB) CreateTable(ctx context.Context, opt offline.CreateTableOpt) error {
 	return sqlutil.CreateTable(ctx, db.DB, opt, Backend)
 }
+
+func (db *DB) Push(ctx context.Context, opt offline.PushOpt) error {
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
+	if err := sqlutil.Push(ctx, dbOpt, opt); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/internal/database/offline/postgres/util.go b/internal/database/offline/postgres/util.go
index ed11894b..cc34149b 100644
--- a/internal/database/offline/postgres/util.go
+++ b/internal/database/offline/postgres/util.go
@@ -4,15 +4,13 @@ import (
 	"context"
 	"io"
 
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-
 	"github.com/jmoiron/sqlx"
 	"github.com/lib/pq"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/offline"
 )
 
-func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList) error {
+func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string) error {
 	stmt, err := tx.PreparexContext(ctx, pq.CopyIn(tableName, header...))
 	if err != nil {
 		return err
@@ -20,7 +18,7 @@ func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSou
 	defer stmt.Close()
 
 	for {
-		record, err := dbutil.ReadLine(source.Reader, source.Delimiter, features, Backend)
+		record, err := dbutil.ReadLine(source.Reader, source.Delimiter)
 		if err == io.EOF {
 			break
 		}
@@ -30,7 +28,11 @@ func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSou
 		if len(record) != len(header) {
 			continue
 		}
-		if _, err := stmt.ExecContext(ctx, record...); err != nil {
+		args := make([]interface{}, 0, len(record))
+		for _, v := range record {
+			args = append(args, v)
+		}
+		if _, err := stmt.ExecContext(ctx, args...); err != nil {
 			return err
 		}
 	}
diff --git a/internal/database/offline/redshift/store.go b/internal/database/offline/redshift/store.go
index 4b959ceb..39c81702 100644
--- a/internal/database/offline/redshift/store.go
+++ b/internal/database/offline/redshift/store.go
@@ -58,3 +58,11 @@ func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 func (db *DB) CreateTable(ctx context.Context, opt offline.CreateTableOpt) error {
 	return sqlutil.CreateTable(ctx, db.DB, opt, Backend)
 }
+
+func (db *DB) Push(ctx context.Context, opt offline.PushOpt) error {
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
+	if err := sqlutil.Push(ctx, dbOpt, opt); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/internal/database/offline/snowflake/store.go b/internal/database/offline/snowflake/store.go
index e3e9fbd6..c2f8b11a 100644
--- a/internal/database/offline/snowflake/store.go
+++ b/internal/database/offline/snowflake/store.go
@@ -73,3 +73,11 @@ func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 func (db *DB) CreateTable(ctx context.Context, opt offline.CreateTableOpt) error {
 	return sqlutil.CreateTable(ctx, db.DB, opt, Backend)
 }
+
+func (db *DB) Push(ctx context.Context, opt offline.PushOpt) error {
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
+	if err := sqlutil.Push(ctx, dbOpt, opt); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/internal/database/offline/sqlite/store.go b/internal/database/offline/sqlite/store.go
index de78990e..62ebedd6 100644
--- a/internal/database/offline/sqlite/store.go
+++ b/internal/database/offline/sqlite/store.go
@@ -67,3 +67,11 @@ func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 func (db *DB) CreateTable(ctx context.Context, opt offline.CreateTableOpt) error {
 	return sqlutil.CreateTable(ctx, db.DB, opt, Backend)
 }
+
+func (db *DB) Push(ctx context.Context, opt offline.PushOpt) error {
+	dbOpt := dbutil.DBOpt{Backend: Backend, SqlxDB: db.DB}
+	if err := sqlutil.Push(ctx, dbOpt, opt); err != nil {
+		return err
+	}
+	return nil
+}
diff --git a/internal/database/offline/sqlutil/create_table.go b/internal/database/offline/sqlutil/create_table.go
index e472a54e..5077fb98 100644
--- a/internal/database/offline/sqlutil/create_table.go
+++ b/internal/database/offline/sqlutil/create_table.go
@@ -12,7 +12,7 @@ import (
 func CreateTable(ctx context.Context, db *sqlx.DB, opt offline.CreateTableOpt, backend types.BackendType) error {
 	if opt.IsCDC {
 		// Create index (entity_key, unix_milli) on cdc table
-		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, false, opt.Features, nil, backend)
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, true, opt.Features, nil, backend)
 		if _, err := db.ExecContext(ctx, schema); err != nil {
 			return err
 		}
diff --git a/internal/database/offline/sqlutil/export.go b/internal/database/offline/sqlutil/export.go
index 49f7009d..79bd35d3 100644
--- a/internal/database/offline/sqlutil/export.go
+++ b/internal/database/offline/sqlutil/export.go
@@ -49,12 +49,19 @@ func Export(ctx context.Context, db *sqlx.DB, opt offline.ExportOpt, backendType
 				if record[i+1] == nil {
 					continue
 				}
-				v, err := DeserializeByValueType(record[i+1], f.ValueType, backendType)
-				if err != nil {
-					errs <- err
-					return
+				if backendType == types.BackendSnowflake {
+
+					v, err := deserializeByTagForSnowflake(record[i+1], f.ValueType)
+					if err != nil {
+						errs <- fmt.Errorf("failed at deserializeByTag, err=%v", err)
+						return
+					}
+					record[i+1] = v
+				} else {
+					if f.ValueType == types.String {
+						record[i+1] = cast.ToString(record[i+1])
+					}
 				}
-				record[i+1] = v
 			}
 			stream <- record
 		}
diff --git a/internal/database/offline/sqlutil/import.go b/internal/database/offline/sqlutil/import.go
index 2ca52dd4..99761f53 100644
--- a/internal/database/offline/sqlutil/import.go
+++ b/internal/database/offline/sqlutil/import.go
@@ -10,7 +10,7 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-type LoadData func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string, features types.FeatureList) error
+type LoadData func(tx *sqlx.Tx, ctx context.Context, source *offline.CSVSource, tableName string, header []string) error
 
 func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData LoadData, backendType types.BackendType) (int64, error) {
 	var revision int64
@@ -27,7 +27,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.SnapshotTableName, opt.Header, opt.Features)
+		err = loadData(tx, ctx, opt.Source, opt.SnapshotTableName, opt.Header)
 		if err != nil {
 			return err
 		}
diff --git a/internal/database/offline/sqlutil/join_helper.go b/internal/database/offline/sqlutil/join_helper.go
index 439dbafb..aa3b00f9 100644
--- a/internal/database/offline/sqlutil/join_helper.go
+++ b/internal/database/offline/sqlutil/join_helper.go
@@ -205,13 +205,13 @@ func insertEntityRows(ctx context.Context,
 		records = append(records, record)
 		if len(records) == InsertBatchSize {
 
-			if err := dbutil.InsertRecordsToTable(ctx, dbOpt, tableName, records, columns, dbOpt.Backend); err != nil {
+			if err := dbutil.InsertRecordsToTable(ctx, dbOpt, tableName, records, columns); err != nil {
 				return err
 			}
 			records = make([]interface{}, 0, InsertBatchSize)
 		}
 	}
-	if err := dbutil.InsertRecordsToTable(ctx, dbOpt, tableName, records, columns, dbOpt.Backend); err != nil {
+	if err := dbutil.InsertRecordsToTable(ctx, dbOpt, tableName, records, columns); err != nil {
 		return err
 	}
 	return nil
diff --git a/internal/database/offline/sqlutil/push.go b/internal/database/offline/sqlutil/push.go
new file mode 100644
index 00000000..727d61d0
--- /dev/null
+++ b/internal/database/offline/sqlutil/push.go
@@ -0,0 +1,18 @@
+package sqlutil
+
+import (
+	"context"
+
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
+	"github.com/oom-ai/oomstore/internal/database/offline"
+	"github.com/oom-ai/oomstore/pkg/errdefs"
+)
+
+func Push(ctx context.Context, dbOpt dbutil.DBOpt, pushOpt offline.PushOpt) error {
+	tableName := dbutil.OfflineStreamCdcTableName(pushOpt.GroupID, pushOpt.Revision)
+	err := dbutil.InsertRecordsToTable(ctx, dbOpt, tableName, pushOpt.FeatureValues, pushOpt.FeatureNames)
+	if err != nil && dbutil.IsTableNotFoundError(err, dbOpt.Backend) {
+		return errdefs.NotFound(err)
+	}
+	return err
+}
diff --git a/internal/database/offline/sqlutil/snapshot.go b/internal/database/offline/sqlutil/snapshot.go
index 8be3fd22..b5ef3579 100644
--- a/internal/database/offline/sqlutil/snapshot.go
+++ b/internal/database/offline/sqlutil/snapshot.go
@@ -10,7 +10,6 @@ import (
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 
-	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
 	"github.com/oom-ai/oomstore/internal/database/offline"
 )
 
@@ -59,9 +58,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 := dbutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.PrevRevision)
+	currSnapshotTableName := dbutil.OfflineStreamSnapshotTableName(opt.Group.ID, opt.Revision)
+	currCdcTableName := dbutil.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/store.go b/internal/database/offline/store.go
index 7ae7c755..db7ae88c 100644
--- a/internal/database/offline/store.go
+++ b/internal/database/offline/store.go
@@ -11,6 +11,7 @@ type Store interface {
 	Join(ctx context.Context, opt JoinOpt) (*types.JoinResult, error)
 	Export(ctx context.Context, opt ExportOpt) (<-chan types.ExportRecord, <-chan error)
 	Import(ctx context.Context, opt ImportOpt) (int64, error)
+	Push(ctx context.Context, opt PushOpt) error
 
 	CreateTable(ctx context.Context, opt CreateTableOpt) error
 	TableSchema(ctx context.Context, tableName string) (*types.DataTableSchema, error)
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index 6f2f4966..6c373a12 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -48,6 +48,15 @@ type ImportOpt struct {
 	NoPK              bool // TODO: to import cdc data temporarily for testing
 }
 
+type PushOpt struct {
+	GroupID      int
+	Revision     int64
+	FeatureNames []string
+
+	// Two-dimension slice, each row represents a record
+	FeatureValues []interface{}
+}
+
 type CSVSource struct {
 	Reader    *bufio.Reader
 	Delimiter string
diff --git a/pkg/oomstore/import.go b/pkg/oomstore/import.go
index 1d35787b..51956571 100644
--- a/pkg/oomstore/import.go
+++ b/pkg/oomstore/import.go
@@ -7,8 +7,6 @@ import (
 	"os"
 	"time"
 
-	"github.com/spf13/cast"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/internal/database/offline"
@@ -46,16 +44,15 @@ func (s *OomStore) Import(ctx context.Context, opt types.ImportOpt) (int, error)
 func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSource *types.CsvReaderDataSource) (int, error) {
 	//make sure csv data source has all defined columns
 	reader := bufio.NewReader(dataSource.Reader)
-	// read header does not need pass down features
-	header, err := dbutil.ReadLine(reader, dataSource.Delimiter, nil, "")
+	header, err := dbutil.ReadLine(reader, dataSource.Delimiter)
 	if err != nil {
 		return 0, err
 	}
-	if hasDup(cast.ToStringSlice(header)) {
+	if hasDup(header) {
 		return 0, fmt.Errorf("csv data source has duplicated columns: %v", header)
 	}
 	columnNames := append([]string{opt.entity.Name}, opt.features.Names()...)
-	if !stringSliceEqual(cast.ToStringSlice(header), columnNames) {
+	if !stringSliceEqual(header, columnNames) {
 		return 0, fmt.Errorf("csv header of the data source %v doesn't match the feature group schema %v", header, columnNames)
 	}
 
@@ -73,7 +70,7 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 	revision, err := s.offline.Import(ctx, offline.ImportOpt{
 		Entity:            opt.entity,
 		Features:          opt.features,
-		Header:            cast.ToStringSlice(header),
+		Header:            header,
 		Revision:          opt.Revision,
 		SnapshotTableName: snapshotTableName,
 		Source: &offline.CSVSource{
diff --git a/pkg/oomstore/revision.go b/pkg/oomstore/revision.go
index 43f26634..6cc57eaa 100644
--- a/pkg/oomstore/revision.go
+++ b/pkg/oomstore/revision.go
@@ -3,12 +3,9 @@ package oomstore
 import (
 	"context"
 
-	"github.com/oom-ai/oomstore/internal/database/offline"
-
-	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
-
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
-
+	"github.com/oom-ai/oomstore/internal/database/offline"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
@@ -28,7 +25,7 @@ func (s *OomStore) GetRevisionBy(ctx context.Context, groupID int, revision int6
 }
 
 func (s *OomStore) createFirstSnapshotTable(ctx context.Context, revision *types.Revision) error {
-	snapshotTable := sqlutil.OfflineStreamSnapshotTableName(revision.GroupID, revision.Revision)
+	snapshotTable := dbutil.OfflineStreamSnapshotTableName(revision.GroupID, revision.Revision)
 
 	// Update snapshot_table in feature_group_revision table
 	err := s.metadata.UpdateRevision(ctx, metadata.UpdateRevisionOpt{
diff --git a/pkg/oomstore/snapshot.go b/pkg/oomstore/snapshot.go
index 25223b11..cb2016a3 100644
--- a/pkg/oomstore/snapshot.go
+++ b/pkg/oomstore/snapshot.go
@@ -4,10 +4,8 @@ import (
 	"context"
 	"sort"
 
-	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
-
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
-
 	"github.com/oom-ai/oomstore/internal/database/offline"
 )
 
@@ -35,7 +33,7 @@ func (s *OomStore) Snapshot(ctx context.Context, groupName string) error {
 		if revision.SnapshotTable != "" {
 			continue
 		}
-		tableName := sqlutil.OfflineStreamSnapshotTableName(group.ID, revision.Revision)
+		tableName := dbutil.OfflineStreamSnapshotTableName(group.ID, revision.Revision)
 		if err = s.offline.Snapshot(ctx, offline.SnapshotOpt{
 			Group:        group,
 			Revision:     revisions[i].Revision,


Test patch:
diff --git a/internal/database/offline/mysql/store_test.go b/internal/database/offline/mysql/store_test.go
index 21207c1a..8ff1834f 100644
--- a/internal/database/offline/mysql/store_test.go
+++ b/internal/database/offline/mysql/store_test.go
@@ -49,6 +49,10 @@ func TestSnapshot(t *testing.T) {
 	test_impl.TestSnapshot(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
 }
 
+func TestPush(t *testing.T) {
+	test_impl.TestPush(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
+}
+
 func TestTableSchema(t *testing.T) {
 	test_impl.TestTableSchema(t, prepareStore, runtime_mysql.DestroyStore(DATABASE), func(ctx context.Context) {
 		opt := runtime_mysql.GetOpt(DATABASE)
diff --git a/internal/database/offline/test_impl/push.go b/internal/database/offline/test_impl/push.go
new file mode 100644
index 00000000..7063cb9e
--- /dev/null
+++ b/internal/database/offline/test_impl/push.go
@@ -0,0 +1,76 @@
+package test_impl
+
+import (
+	"testing"
+
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
+	"github.com/oom-ai/oomstore/internal/database/offline"
+	"github.com/oom-ai/oomstore/pkg/errdefs"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/stretchr/testify/require"
+)
+
+func TestPush(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
+	t.Cleanup(destroyStore)
+
+	ctx, store := prepareStore(t)
+	defer store.Close()
+
+	entity := types.Entity{
+		ID:     18,
+		Name:   "user",
+		Length: 10,
+	}
+	group := types.Group{
+		ID:       1,
+		Name:     "vip",
+		Category: types.CategoryStream,
+		EntityID: entity.ID,
+	}
+	features := types.FeatureList{
+		&types.Feature{
+			ID:        1,
+			Name:      "age",
+			ValueType: types.Int64,
+			GroupID:   group.ID,
+		},
+		&types.Feature{
+			ID:        2,
+			Name:      "gender",
+			ValueType: types.String,
+			GroupID:   group.ID,
+		},
+	}
+	revision := int64(2)
+
+	featureValues := []interface{}{
+		[]interface{}{18, "F"},
+		[]interface{}{21, "M"},
+	}
+
+	pushOpt := offline.PushOpt{
+		GroupID:       group.ID,
+		Revision:      revision,
+		FeatureNames:  features.Names(),
+		FeatureValues: featureValues,
+	}
+
+	t.Run("push when cdc table not exists", func(t *testing.T) {
+		err := store.Push(ctx, pushOpt)
+		require.Error(t, err)
+		require.True(t, errdefs.IsNotFound(err))
+	})
+
+	t.Run("push when cdc table exists", func(t *testing.T) {
+		err := store.CreateTable(ctx, offline.CreateTableOpt{
+			TableName: dbutil.OfflineStreamCdcTableName(group.ID, revision),
+			Entity:    &entity,
+			Features:  features,
+			IsCDC:     true,
+		})
+		require.NoError(t, err)
+
+		err = store.Push(ctx, pushOpt)
+		require.NoError(t, err)
+	})
+}