Repository: oom-ai/oomstore
Pull request #907 (2022-01-06 02:25:02)

fix(offline/push): add missing entity key and unix_milli
Close #907 

codecov[bot]: # [Codecov](https://codecov.io/gh/oom-ai/oomstore/pull/907?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) Report
> Merging [#907](https://codecov.io/gh/oom-ai/oomstore/pull/907?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) (2d22a3c) into [main](https://codecov.io/gh/oom-ai/oomstore/commit/da2b2eb560ac941c70d13937fc919cbc63ee7e6b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) (da2b2eb) will **decrease** coverage by `0.04%`.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/oom-ai/oomstore/pull/907/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/907?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     #907      +/-   ##
==========================================
- Coverage   51.16%   51.12%   -0.05%     
==========================================
  Files          84       84              
  Lines        3563     3566       +3     
==========================================
  Hits         1823     1823              
- Misses       1523     1526       +3     
  Partials      217      217              
```


| [Impacted Files](https://codecov.io/gh/oom-ai/oomstore/pull/907?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/sql.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvZGJ1dGlsL3NxbC5nbw==) | `20.31% <0.00%> (ø)` | |
| [internal/database/dbutil/database.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/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/metadata/mysql/group.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvbWV0YWRhdGEvbXlzcWwvZ3JvdXAuZ28=) | `76.47% <0.00%> (ø)` | |
| [internal/database/metadata/mysql/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvbWV0YWRhdGEvbXlzcWwvc3RvcmUuZ28=) | `46.66% <0.00%> (ø)` | |
| [internal/database/metadata/sqlite/group.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvbWV0YWRhdGEvc3FsaXRlL2dyb3VwLmdv) | `76.47% <0.00%> (ø)` | |
| [internal/database/metadata/sqlite/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvbWV0YWRhdGEvc3FsaXRlL3N0b3JlLmdv) | `42.30% <0.00%> (ø)` | |
| [internal/database/metadata/mysql/feature.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvbWV0YWRhdGEvbXlzcWwvZmVhdHVyZS5nbw==) | `76.47% <0.00%> (ø)` | |
| [internal/database/metadata/postgres/group.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvbWV0YWRhdGEvcG9zdGdyZXMvZ3JvdXAuZ28=) | `100.00% <0.00%> (ø)` | |
| [internal/database/metadata/postgres/store.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvbWV0YWRhdGEvcG9zdGdyZXMvc3RvcmUuZ28=) | `42.30% <0.00%> (ø)` | |
| [internal/database/metadata/postgres/entity.go](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai#diff-aW50ZXJuYWwvZGF0YWJhc2UvbWV0YWRhdGEvcG9zdGdyZXMvZW50aXR5Lmdv) | `100.00% <0.00%> (ø)` | |
| ... and [3 more](https://codecov.io/gh/oom-ai/oomstore/pull/907/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/oom-ai/oomstore/pull/907?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/907?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=oom-ai). Last update [da2b2eb...2d22a3c](https://codecov.io/gh/oom-ai/oomstore/pull/907?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/database.go b/internal/database/dbutil/database.go
index b52e3311..7bfd41d3 100644
--- a/internal/database/dbutil/database.go
+++ b/internal/database/dbutil/database.go
@@ -9,14 +9,12 @@ import (
 	"github.com/lib/pq"
 	"github.com/mattn/go-sqlite3"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-	"github.com/pkg/errors"
 	"github.com/snowflakedb/gosnowflake"
 	"google.golang.org/api/googleapi"
 )
 
 func OpenSQLite(dbFile string) (*sqlx.DB, error) {
-	db, err := sqlx.Open("sqlite3", dbFile)
-	return db, errors.WithStack(err)
+	return sqlx.Open("sqlite3", dbFile)
 }
 
 func OpenMysqlDB(host, port, user, password, database string) (*sqlx.DB, error) {
@@ -27,12 +25,11 @@ func OpenMysqlDB(host, port, user, password, database string) (*sqlx.DB, error)
 	cfg.DBName = database
 	cfg.ParseTime = true
 
-	db, err := sqlx.Open("mysql", cfg.FormatDSN())
-	return db, errors.WithStack(err)
+	return sqlx.Open("mysql", cfg.FormatDSN())
 }
 
 func OpenPostgresDB(host, port, user, password, database string) (*sqlx.DB, error) {
-	db, err := sqlx.Open(
+	return sqlx.Open(
 		"postgres",
 		fmt.Sprintf("postgres://%s:%s@%s:%s/%s?sslmode=disable",
 			user,
@@ -41,7 +38,6 @@ func OpenPostgresDB(host, port, user, password, database string) (*sqlx.DB, erro
 			port,
 			database),
 	)
-	return db, errors.WithStack(err)
 }
 
 var OpenRedshiftDB = OpenPostgresDB
diff --git a/internal/database/dbutil/sql.go b/internal/database/dbutil/sql.go
index 4322e786..1b17a736 100644
--- a/internal/database/dbutil/sql.go
+++ b/internal/database/dbutil/sql.go
@@ -6,8 +6,6 @@ import (
 	"strings"
 
 	"github.com/jmoiron/sqlx"
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
@@ -21,7 +19,7 @@ func BuildConditions(equal map[string]interface{}, in map[string]interface{}) ([
 	for key, value := range in {
 		s, inArgs, err := sqlx.In(fmt.Sprintf("%s IN (?)", key), value)
 		if err != nil {
-			return nil, nil, errors.WithStack(err)
+			return nil, nil, err
 		}
 		cond = append(cond, s)
 		args = append(args, inArgs...)
diff --git a/internal/database/metadata/mysql/entity.go b/internal/database/metadata/mysql/entity.go
index b9521a37..80db89fe 100644
--- a/internal/database/metadata/mysql/entity.go
+++ b/internal/database/metadata/mysql/entity.go
@@ -2,10 +2,10 @@ package mysql
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/go-sql-driver/mysql"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
-	"github.com/pkg/errors"
 )
 
 func createEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateEntityOpt) (int, error) {
@@ -14,15 +14,15 @@ func createEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadat
 	if err != nil {
 		if er, ok := err.(*mysql.MySQLError); ok {
 			if er.Number == ER_DUP_ENTRY {
-				return 0, errors.Errorf("entity %s already exists", opt.EntityName)
+				return 0, fmt.Errorf("entity %s already exists", opt.EntityName)
 			}
 		}
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
 
 	entityID, err := res.LastInsertId()
 	if err != nil {
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
-	return int(entityID), nil
+	return int(entityID), err
 }
diff --git a/internal/database/metadata/mysql/feature.go b/internal/database/metadata/mysql/feature.go
index bb739a86..a9d2d213 100644
--- a/internal/database/metadata/mysql/feature.go
+++ b/internal/database/metadata/mysql/feature.go
@@ -2,8 +2,7 @@ package mysql
 
 import (
 	"context"
-
-	"github.com/pkg/errors"
+	"fmt"
 
 	"github.com/go-sql-driver/mysql"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
@@ -11,14 +10,14 @@ import (
 
 func createFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateFeatureOpt) (int, error) {
 	if err := opt.ValueType.Validate(); err != nil {
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
 	query := "INSERT INTO feature(name, full_name, group_id, value_type, description) VALUES (?, ?, ?, ?, ?)"
 	res, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(query), opt.FeatureName, opt.FullName, opt.GroupID, opt.ValueType, opt.Description)
 	if err != nil {
 		if er, ok := err.(*mysql.MySQLError); ok {
 			if er.Number == ER_DUP_ENTRY {
-				return 0, errors.Errorf("feature %s already exists", opt.FeatureName)
+				return 0, fmt.Errorf("feature %s already exists", opt.FeatureName)
 			}
 		}
 		return 0, err
@@ -26,7 +25,7 @@ func createFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metada
 
 	featureID, err := res.LastInsertId()
 	if err != nil {
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
-	return int(featureID), nil
+	return int(featureID), err
 }
diff --git a/internal/database/metadata/mysql/group.go b/internal/database/metadata/mysql/group.go
index 9c035ddb..4d255059 100644
--- a/internal/database/metadata/mysql/group.go
+++ b/internal/database/metadata/mysql/group.go
@@ -2,8 +2,7 @@ package mysql
 
 import (
 	"context"
-
-	"github.com/pkg/errors"
+	"fmt"
 
 	"github.com/go-sql-driver/mysql"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
@@ -13,7 +12,7 @@ import (
 
 func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateGroupOpt) (int, error) {
 	if opt.Category != types.CategoryBatch && opt.Category != types.CategoryStream {
-		return 0, errdefs.InvalidAttribute(errors.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
+		return 0, errdefs.InvalidAttribute(fmt.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
 	}
 
 	query := "INSERT INTO feature_group(name, entity_id, category, description) VALUES(?, ?, ?, ?)"
@@ -21,16 +20,16 @@ func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	if err != nil {
 		if er, ok := err.(*mysql.MySQLError); ok {
 			if er.Number == ER_DUP_ENTRY {
-				return 0, errors.Errorf("feature group %s already exists", opt.GroupName)
+				return 0, fmt.Errorf("feature group %s already exists", opt.GroupName)
 			}
 		}
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
 
 	groupID, err := res.LastInsertId()
 	if err != nil {
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
 
-	return int(groupID), nil
+	return int(groupID), err
 }
diff --git a/internal/database/metadata/mysql/revision.go b/internal/database/metadata/mysql/revision.go
index 63a11c9e..d6bd23d3 100644
--- a/internal/database/metadata/mysql/revision.go
+++ b/internal/database/metadata/mysql/revision.go
@@ -2,9 +2,9 @@ package mysql
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/go-sql-driver/mysql"
-	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
@@ -24,14 +24,14 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 	if err != nil {
 		if er, ok := err.(*mysql.MySQLError); ok {
 			if er.Number == ER_DUP_ENTRY {
-				return 0, "", errors.Errorf("revision already exists: groupID=%d, revision=%d", opt.GroupID, opt.Revision)
+				return 0, "", fmt.Errorf("revision already exists: groupID=%d, revision=%d", opt.GroupID, opt.Revision)
 			}
 		}
-		return 0, "", errors.WithStack(err)
+		return 0, "", err
 	}
 	revisionID, err := res.LastInsertId()
 	if err != nil {
-		return 0, "", errors.WithStack(err)
+		return 0, "", err
 	}
 
 	if opt.SnapshotTable == nil {
@@ -39,14 +39,14 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		snapshotTable = dbutil.OfflineBatchTableName(opt.GroupID, revisionID)
 		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), snapshotTable, revisionID)
 		if err != nil {
-			return 0, "", errors.WithStack(err)
+			return 0, "", err
 		}
 		rowsAffected, err := result.RowsAffected()
 		if err != nil {
-			return 0, "", errors.WithStack(err)
+			return 0, "", err
 		}
 		if rowsAffected != 1 {
-			return 0, "", errors.Errorf("failed to update revision %d: revision not found", revisionID)
+			return 0, "", fmt.Errorf("failed to update revision %d: revision not found", revisionID)
 		}
 	}
 	return int(revisionID), snapshotTable, nil
diff --git a/internal/database/metadata/mysql/store.go b/internal/database/metadata/mysql/store.go
index 7edb9bc5..3d58f5f1 100644
--- a/internal/database/metadata/mysql/store.go
+++ b/internal/database/metadata/mysql/store.go
@@ -11,7 +11,6 @@ import (
 	"github.com/oom-ai/oomstore/internal/database/metadata/informer"
 	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-	"github.com/pkg/errors"
 )
 
 var _ metadata.Store = &DB{}
@@ -32,10 +31,10 @@ func (db *DB) Ping(ctx context.Context) error {
 
 func (db *DB) Close() error {
 	if err := db.Informer.Close(); err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	if err := db.DB.Close(); err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	return nil
 }
@@ -60,20 +59,20 @@ func Open(ctx context.Context, option *types.MySQLOpt) (*DB, error) {
 	}, nil
 }
 
-func CreateDatabase(ctx context.Context, opt *types.MySQLOpt) error {
+func CreateDatabase(ctx context.Context, opt *types.MySQLOpt) (err error) {
 	defaultDB, err := dbutil.OpenMysqlDB(opt.Host, opt.Port, opt.User, opt.Password, "")
 	if err != nil {
-		return errors.WithStack(err)
+		return
 	}
 	defer defaultDB.Close()
 
 	if _, err = defaultDB.ExecContext(ctx, fmt.Sprintf("CREATE DATABASE IF NOT EXISTS %s", opt.Database)); err != nil {
-		return errors.WithStack(err)
+		return
 	}
 
 	db, err := dbutil.OpenMysqlDB(opt.Host, opt.Port, opt.User, opt.Password, opt.Database)
 	if err != nil {
-		return errors.WithStack(err)
+		return
 	}
 	defer db.Close()
 	return createMetaSchemas(ctx, db)
@@ -86,21 +85,21 @@ func createMetaSchemas(ctx context.Context, db *sqlx.DB) (err error) {
 		// create meta tables
 		for _, schema := range META_TABLE_SCHEMAS {
 			if _, err = tx.ExecContext(ctx, schema); err != nil {
-				return errors.Errorf("failed to create table, err=%+v", err)
+				return fmt.Errorf("failed to create table, err=%+v", err)
 			}
 		}
 
 		// create foreign keys
 		for _, stmt := range META_TABLE_FOREIGN_KEYS {
 			if _, err = tx.ExecContext(ctx, stmt); err != nil {
-				return errors.Errorf("failed to add foreign key, err=%+v", err)
+				return fmt.Errorf("failed to add foreign key, err=%+v", err)
 			}
 		}
 
 		// create meta views
 		for _, schema := range META_VIEW_SCHEMAS {
 			if _, err = tx.ExecContext(ctx, schema); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
diff --git a/internal/database/metadata/postgres/create_database.go b/internal/database/metadata/postgres/create_database.go
index f6e0e4bb..93aea8cb 100644
--- a/internal/database/metadata/postgres/create_database.go
+++ b/internal/database/metadata/postgres/create_database.go
@@ -8,8 +8,6 @@ import (
 	"github.com/jackc/pgerrcode"
 	"github.com/jmoiron/sqlx"
 	"github.com/lib/pq"
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -23,7 +21,7 @@ func CreateDatabase(ctx context.Context, opt *types.PostgresOpt) (err error) {
 
 	if _, err = defaultDB.ExecContext(ctx, fmt.Sprintf("CREATE DATABASE %s", opt.Database)); err != nil {
 		if e2, ok := err.(*pq.Error); ok && e2.Code != pgerrcode.DuplicateDatabase {
-			return errors.WithStack(err)
+			return err
 		}
 	}
 
@@ -42,28 +40,28 @@ func createMetaSchemas(ctx context.Context, db *sqlx.DB) (err error) {
 		// create database functions
 		for _, fn := range DB_FUNCTIONS {
 			if _, err = tx.ExecContext(ctx, fn); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
 		// create meta tables
 		for _, schema := range META_TABLE_SCHEMAS {
 			if _, err = tx.ExecContext(ctx, schema); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
 		// create foreign keys
 		for _, stmt := range META_TABLE_FOREIGN_KEYS {
 			if _, err = tx.ExecContext(ctx, stmt); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
 		// create meta views
 		for _, schema := range META_VIEW_SCHEMAS {
 			if _, err = tx.ExecContext(ctx, schema); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
@@ -71,7 +69,7 @@ func createMetaSchemas(ctx context.Context, db *sqlx.DB) (err error) {
 		for table := range META_TABLE_SCHEMAS {
 			trigger := strings.ReplaceAll(TRIGGER_TEMPLATE, `{{TABLE_NAME}}`, table)
 			if _, err = tx.ExecContext(ctx, trigger); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
diff --git a/internal/database/metadata/postgres/entity.go b/internal/database/metadata/postgres/entity.go
index 399fe3cb..a67d2a67 100644
--- a/internal/database/metadata/postgres/entity.go
+++ b/internal/database/metadata/postgres/entity.go
@@ -2,11 +2,10 @@ package postgres
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/jackc/pgerrcode"
 	"github.com/lib/pq"
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 )
 
@@ -16,8 +15,8 @@ func createEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadat
 	err := sqlxCtx.GetContext(ctx, &entityID, query, opt.EntityName, opt.Description)
 	if er, ok := err.(*pq.Error); ok {
 		if er.Code == pgerrcode.UniqueViolation {
-			return 0, errors.Errorf("entity %s already exists", opt.EntityName)
+			return 0, fmt.Errorf("entity %s already exists", opt.EntityName)
 		}
 	}
-	return entityID, errors.WithStack(err)
+	return entityID, err
 }
diff --git a/internal/database/metadata/postgres/feature.go b/internal/database/metadata/postgres/feature.go
index b32bcbd5..dfc45ef0 100644
--- a/internal/database/metadata/postgres/feature.go
+++ b/internal/database/metadata/postgres/feature.go
@@ -2,11 +2,11 @@ package postgres
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/jackc/pgerrcode"
 	"github.com/lib/pq"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
-	"github.com/pkg/errors"
 )
 
 func createFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateFeatureOpt) (int, error) {
@@ -19,9 +19,9 @@ func createFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metada
 	if err != nil {
 		if e2, ok := err.(*pq.Error); ok {
 			if e2.Code == pgerrcode.UniqueViolation {
-				return 0, errors.Errorf("feature %s already exists", opt.FeatureName)
+				return 0, fmt.Errorf("feature %s already exists", opt.FeatureName)
 			}
 		}
 	}
-	return featureID, errors.WithStack(err)
+	return featureID, err
 }
diff --git a/internal/database/metadata/postgres/group.go b/internal/database/metadata/postgres/group.go
index ebcc446c..a1e5ccda 100644
--- a/internal/database/metadata/postgres/group.go
+++ b/internal/database/metadata/postgres/group.go
@@ -2,11 +2,10 @@ package postgres
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/jackc/pgerrcode"
 	"github.com/lib/pq"
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -14,7 +13,7 @@ import (
 
 func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateGroupOpt) (int, error) {
 	if opt.Category != types.CategoryBatch && opt.Category != types.CategoryStream {
-		return 0, errdefs.InvalidAttribute(errors.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
+		return 0, errdefs.InvalidAttribute(fmt.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
 	}
 	var groupID int
 	query := "insert into feature_group(name, entity_id, category, description) values($1, $2, $3, $4) returning id"
@@ -22,9 +21,9 @@ func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	if err != nil {
 		if e2, ok := err.(*pq.Error); ok {
 			if e2.Code == pgerrcode.UniqueViolation {
-				return 0, errors.Errorf("feature group %s already exists", opt.GroupName)
+				return 0, fmt.Errorf("feature group %s already exists", opt.GroupName)
 			}
 		}
 	}
-	return groupID, errors.WithStack(err)
+	return groupID, err
 }
diff --git a/internal/database/metadata/postgres/revision.go b/internal/database/metadata/postgres/revision.go
index 7592bf1b..25abc1b4 100644
--- a/internal/database/metadata/postgres/revision.go
+++ b/internal/database/metadata/postgres/revision.go
@@ -2,11 +2,10 @@ package postgres
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/jackc/pgerrcode"
 	"github.com/lib/pq"
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 )
@@ -25,24 +24,24 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 	if err := sqlxCtx.GetContext(ctx, &revisionID, insertQuery, opt.GroupID, opt.Revision, snapshotTable, cdcTable, opt.Anchored, opt.Description); err != nil {
 		if e2, ok := err.(*pq.Error); ok {
 			if e2.Code == pgerrcode.UniqueViolation {
-				return 0, "", errors.Errorf("revision already exists: groupID=%d, revision=%d", opt.GroupID, opt.Revision)
+				return 0, "", fmt.Errorf("revision already exists: groupID=%d, revision=%d", opt.GroupID, opt.Revision)
 			}
 		}
-		return 0, "", errors.WithStack(err)
+		return 0, "", err
 	}
 	if opt.SnapshotTable == nil {
 		updateQuery := "UPDATE feature_group_revision SET snapshot_table = $1 WHERE id = $2"
 		snapshotTable = dbutil.OfflineBatchTableName(opt.GroupID, int64(revisionID))
 		result, err := sqlxCtx.ExecContext(ctx, updateQuery, snapshotTable, revisionID)
 		if err != nil {
-			return 0, "", errors.WithStack(err)
+			return 0, "", err
 		}
 		rowsAffected, err := result.RowsAffected()
 		if err != nil {
-			return 0, "", errors.WithStack(err)
+			return 0, "", err
 		}
 		if rowsAffected != 1 {
-			return 0, "", errors.Errorf("failed to update revision %d: revision not found", revisionID)
+			return 0, "", fmt.Errorf("failed to update revision %d: revision not found", revisionID)
 		}
 	}
 
diff --git a/internal/database/metadata/postgres/store.go b/internal/database/metadata/postgres/store.go
index da391477..4edca74f 100644
--- a/internal/database/metadata/postgres/store.go
+++ b/internal/database/metadata/postgres/store.go
@@ -8,7 +8,6 @@ import (
 	_ "github.com/lib/pq"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
-	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/internal/database/metadata/informer"
@@ -56,7 +55,7 @@ func (db *DB) Close() error {
 		return err
 	}
 	if err := db.DB.Close(); err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	return nil
 }
diff --git a/internal/database/metadata/sqlite/create_database.go b/internal/database/metadata/sqlite/create_database.go
index 260a3e2f..98ff4445 100644
--- a/internal/database/metadata/sqlite/create_database.go
+++ b/internal/database/metadata/sqlite/create_database.go
@@ -5,8 +5,6 @@ import (
 	"strings"
 
 	"github.com/jmoiron/sqlx"
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -24,20 +22,20 @@ func createMetaSchemas(ctx context.Context, db *sqlx.DB) (err error) {
 	return dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 		for _, schema := range META_TABLE_SCHEMAS {
 			if _, err = tx.ExecContext(ctx, schema); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
 		for _, schema := range META_VIEW_SCHEMAS {
 			if _, err = tx.ExecContext(ctx, schema); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
 		for table := range META_TABLE_SCHEMAS {
 			trigger := strings.ReplaceAll(TRIGGER_TEMPLATE, `{{TABLE_NAME}}`, table)
 			if _, err = tx.ExecContext(ctx, trigger); err != nil {
-				return errors.WithStack(err)
+				return err
 			}
 		}
 
diff --git a/internal/database/metadata/sqlite/entity.go b/internal/database/metadata/sqlite/entity.go
index cbed176d..4f719da9 100644
--- a/internal/database/metadata/sqlite/entity.go
+++ b/internal/database/metadata/sqlite/entity.go
@@ -2,9 +2,9 @@ package sqlite
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/mattn/go-sqlite3"
-	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 )
@@ -15,7 +15,7 @@ func createEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadat
 	if err != nil {
 		if er, ok := err.(sqlite3.Error); ok {
 			if er.ExtendedCode == sqlite3.ErrConstraintUnique {
-				return 0, errors.Errorf("entity %s already exists", opt.EntityName)
+				return 0, fmt.Errorf("entity %s already exists", opt.EntityName)
 			}
 		}
 		return 0, err
@@ -23,7 +23,7 @@ func createEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadat
 
 	entityID, err := res.LastInsertId()
 	if err != nil {
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
-	return int(entityID), nil
+	return int(entityID), err
 }
diff --git a/internal/database/metadata/sqlite/feature.go b/internal/database/metadata/sqlite/feature.go
index fd35068d..17b7264b 100644
--- a/internal/database/metadata/sqlite/feature.go
+++ b/internal/database/metadata/sqlite/feature.go
@@ -2,9 +2,9 @@ package sqlite
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/mattn/go-sqlite3"
-	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 )
@@ -18,15 +18,15 @@ func createFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metada
 	if err != nil {
 		if sqliteErr, ok := err.(sqlite3.Error); ok {
 			if sqliteErr.ExtendedCode == sqlite3.ErrConstraintUnique {
-				return 0, errors.Errorf("feature %s already exists", opt.FeatureName)
+				return 0, fmt.Errorf("feature %s already exists", opt.FeatureName)
 			}
 		}
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
 
 	featureID, err := res.LastInsertId()
 	if err != nil {
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
-	return int(featureID), nil
+	return int(featureID), err
 }
diff --git a/internal/database/metadata/sqlite/group.go b/internal/database/metadata/sqlite/group.go
index 329d4fe5..57e6603d 100644
--- a/internal/database/metadata/sqlite/group.go
+++ b/internal/database/metadata/sqlite/group.go
@@ -2,8 +2,7 @@ package sqlite
 
 import (
 	"context"
-
-	"github.com/pkg/errors"
+	"fmt"
 
 	"github.com/mattn/go-sqlite3"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
@@ -13,7 +12,7 @@ import (
 
 func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateGroupOpt) (int, error) {
 	if opt.Category != types.CategoryBatch && opt.Category != types.CategoryStream {
-		return 0, errdefs.InvalidAttribute(errors.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
+		return 0, errdefs.InvalidAttribute(fmt.Errorf("illegal category '%s', should be either 'stream' or 'batch'", opt.Category))
 	}
 
 	query := "INSERT INTO feature_group(name, entity_id, category, description) VALUES(?, ?, ?, ?)"
@@ -21,16 +20,16 @@ func createGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	if err != nil {
 		if sqliteErr, ok := err.(sqlite3.Error); ok {
 			if sqliteErr.ExtendedCode == sqlite3.ErrConstraintUnique {
-				return 0, errors.Errorf("feature group %s already exists", opt.GroupName)
+				return 0, fmt.Errorf("feature group %s already exists", opt.GroupName)
 			}
 		}
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
 
 	groupID, err := res.LastInsertId()
 	if err != nil {
-		return 0, errors.WithStack(err)
+		return 0, err
 	}
 
-	return int(groupID), nil
+	return int(groupID), err
 }
diff --git a/internal/database/metadata/sqlite/revision.go b/internal/database/metadata/sqlite/revision.go
index 04cd0137..5f9e79c2 100644
--- a/internal/database/metadata/sqlite/revision.go
+++ b/internal/database/metadata/sqlite/revision.go
@@ -2,11 +2,11 @@ package sqlite
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/mattn/go-sqlite3"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
-	"github.com/pkg/errors"
 )
 
 func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateRevisionOpt) (int, string, error) {
@@ -23,14 +23,14 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 	if err != nil {
 		if sqliteErr, ok := err.(sqlite3.Error); ok {
 			if sqliteErr.ExtendedCode == sqlite3.ErrConstraintUnique {
-				return 0, "", errors.Errorf("revision already exists: groupID=%d, revision=%d", opt.GroupID, opt.Revision)
+				return 0, "", fmt.Errorf("revision already exists: groupID=%d, revision=%d", opt.GroupID, opt.Revision)
 			}
 		}
-		return 0, "", errors.WithStack(err)
+		return 0, "", err
 	}
 	revisionID, err := res.LastInsertId()
 	if err != nil {
-		return 0, "", errors.WithStack(err)
+		return 0, "", err
 	}
 
 	if opt.SnapshotTable == nil {
@@ -38,14 +38,14 @@ func createRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		snapshotTable = dbutil.OfflineBatchTableName(opt.GroupID, revisionID)
 		result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(updateQuery), snapshotTable, revisionID)
 		if err != nil {
-			return 0, "", errors.WithStack(err)
+			return 0, "", err
 		}
 		rowsAffected, err := result.RowsAffected()
 		if err != nil {
-			return 0, "", errors.WithStack(err)
+			return 0, "", err
 		}
 		if rowsAffected != 1 {
-			return 0, "", errors.Errorf("failed to update revision %d: revision not found", revisionID)
+			return 0, "", fmt.Errorf("failed to update revision %d: revision not found", revisionID)
 		}
 	}
 	return int(revisionID), snapshotTable, nil
diff --git a/internal/database/metadata/sqlite/store.go b/internal/database/metadata/sqlite/store.go
index e19376d4..fa2880e9 100644
--- a/internal/database/metadata/sqlite/store.go
+++ b/internal/database/metadata/sqlite/store.go
@@ -6,7 +6,6 @@ import (
 
 	"github.com/oom-ai/oomstore/internal/database/metadata/informer"
 	"github.com/oom-ai/oomstore/internal/database/metadata/sqlutil"
-	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 
@@ -55,7 +54,7 @@ func (db *DB) Close() error {
 		return err
 	}
 	if err := db.DB.Close(); err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	return nil
 }
diff --git a/internal/database/metadata/sqlutil/db.go b/internal/database/metadata/sqlutil/db.go
index ee8e8f40..b4a5169f 100644
--- a/internal/database/metadata/sqlutil/db.go
+++ b/internal/database/metadata/sqlutil/db.go
@@ -4,8 +4,6 @@ import (
 	"context"
 
 	"github.com/jmoiron/sqlx"
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata/informer"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -16,17 +14,17 @@ func ListMetadata(ctx context.Context, db *sqlx.DB) (*informer.Cache, error) {
 	err := dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 		entities := types.EntityList{}
 		if err := tx.SelectContext(ctx, &entities, `SELECT * FROM entity`); err != nil {
-			return errors.WithStack(err)
+			return err
 		}
 
 		features := types.FeatureList{}
 		if err := tx.SelectContext(ctx, &features, `SELECT * FROM feature`); err != nil {
-			return errors.WithStack(err)
+			return err
 		}
 
 		groups := types.GroupList{}
 		if err := tx.SelectContext(ctx, &groups, `SELECT * FROM feature_group`); err != nil {
-			return errors.WithStack(err)
+			return err
 		}
 
 		cache = informer.NewCache(entities, features, groups)
diff --git a/internal/database/metadata/sqlutil/entity.go b/internal/database/metadata/sqlutil/entity.go
index 93a89420..0be8e7f6 100644
--- a/internal/database/metadata/sqlutil/entity.go
+++ b/internal/database/metadata/sqlutil/entity.go
@@ -5,8 +5,6 @@ import (
 	"database/sql"
 	"fmt"
 
-	"github.com/pkg/errors"
-
 	"github.com/jmoiron/sqlx"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
@@ -15,22 +13,22 @@ import (
 
 func UpdateEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.UpdateEntityOpt) error {
 	if opt.NewDescription == nil {
-		return errors.Errorf("invalid option: nothing to update")
+		return fmt.Errorf("invalid option: nothing to update")
 	}
 
 	query := "UPDATE entity SET description = ? WHERE id = ?"
 	result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(query), opt.NewDescription, opt.EntityID)
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 
 	rowsAffected, err := result.RowsAffected()
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 
 	if rowsAffected != 1 {
-		return errors.Errorf("failed to update entity %d: entity not found", opt.EntityID)
+		return fmt.Errorf("failed to update entity %d: entity not found", opt.EntityID)
 	}
 	return nil
 }
@@ -40,9 +38,9 @@ func GetEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, id int) (*type
 	query := "SELECT * FROM entity WHERE id = ?"
 	if err := sqlxCtx.GetContext(ctx, &entity, sqlxCtx.Rebind(query), id); err != nil {
 		if err == sql.ErrNoRows {
-			return nil, errdefs.NotFound(errors.Errorf("feature entity %d not found", id))
+			return nil, errdefs.NotFound(fmt.Errorf("feature entity %d not found", id))
 		}
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	return &entity, nil
@@ -53,12 +51,13 @@ func GetEntityByName(ctx context.Context, sqlxCtx metadata.SqlxContext, name str
 	query := "SELECT * FROM entity WHERE name = ?"
 	if err := sqlxCtx.GetContext(ctx, &entity, sqlxCtx.Rebind(query), name); err != nil {
 		if err == sql.ErrNoRows {
-			return nil, errdefs.NotFound(errors.Errorf("feature entity %s not found", name))
+			return nil, errdefs.NotFound(fmt.Errorf("feature entity %s not found", name))
 		}
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	return &entity, nil
+
 }
 
 func ListEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, entityIDs *[]int) (types.EntityList, error) {
@@ -71,12 +70,12 @@ func ListEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, entityIDs *[]
 		}
 		query, args, err = sqlx.In(fmt.Sprintf("%s WHERE id IN (?)", query), *entityIDs)
 		if err != nil {
-			return nil, errors.WithStack(err)
+			return nil, err
 		}
 	}
 	entities := types.EntityList{}
 	if err := sqlxCtx.SelectContext(ctx, &entities, sqlxCtx.Rebind(query), args...); err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 	return entities, nil
 }
diff --git a/internal/database/metadata/sqlutil/feature.go b/internal/database/metadata/sqlutil/feature.go
index 23b02ef3..05cb9d43 100644
--- a/internal/database/metadata/sqlutil/feature.go
+++ b/internal/database/metadata/sqlutil/feature.go
@@ -6,8 +6,6 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
@@ -16,20 +14,20 @@ import (
 
 func UpdateFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.UpdateFeatureOpt) error {
 	if opt.NewDescription == nil {
-		return errors.Errorf("invalid option: nothing to update")
+		return fmt.Errorf("invalid option: nothing to update")
 	}
 
 	query := "UPDATE feature SET description = ? WHERE id = ?"
 	result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(query), opt.NewDescription, opt.FeatureID)
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	rowsAffected, err := result.RowsAffected()
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	if rowsAffected != 1 {
-		return errors.Errorf("failed to update feature %d: feature not found", opt.FeatureID)
+		return fmt.Errorf("failed to update feature %d: feature not found", opt.FeatureID)
 	}
 	return nil
 }
@@ -44,13 +42,13 @@ func GetFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, id int) (*typ
 	query := `SELECT * FROM feature WHERE id = ?`
 	if err := sqlxCtx.GetContext(ctx, &feature, sqlxCtx.Rebind(query), id); err != nil {
 		if err == sql.ErrNoRows {
-			return nil, errdefs.NotFound(errors.Errorf("feature %d not found", id))
+			return nil, errdefs.NotFound(fmt.Errorf("feature %d not found", id))
 		}
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	if group, err = GetGroup(ctx, sqlxCtx, feature.GroupID); err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 	feature.Group = group
 
@@ -67,13 +65,13 @@ func GetFeatureByName(ctx context.Context, sqlxCtx metadata.SqlxContext, fullNam
 	query := `SELECT * FROM feature WHERE full_name = ?`
 	if err := sqlxCtx.GetContext(ctx, &feature, sqlxCtx.Rebind(query), fullName); err != nil {
 		if err == sql.ErrNoRows {
-			return nil, errdefs.NotFound(errors.Errorf("feature %s not found", fullName))
+			return nil, errdefs.NotFound(fmt.Errorf("feature %s not found", fullName))
 		}
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	if group, err = GetGroup(ctx, sqlxCtx, feature.GroupID); err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 	feature.Group = group
 
@@ -96,7 +94,7 @@ func ListFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	}
 	query = fmt.Sprintf("%s ORDER BY id ASC", query)
 	if err := sqlxCtx.SelectContext(ctx, &features, sqlxCtx.Rebind(query), args...); err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	// enrich group
@@ -108,7 +106,7 @@ func ListFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	for _, f := range features {
 		group := groups.Find(func(g *types.Group) bool { return g.ID == f.GroupID })
 		if group == nil {
-			return nil, errdefs.InvalidAttribute(errors.Errorf("no group found for feature %s", f.Name))
+			return nil, errdefs.InvalidAttribute(fmt.Errorf("no group found for feature %s", f.Name))
 		}
 		f.Group = group
 	}
diff --git a/internal/database/metadata/sqlutil/group.go b/internal/database/metadata/sqlutil/group.go
index 3991e970..c7b789a7 100644
--- a/internal/database/metadata/sqlutil/group.go
+++ b/internal/database/metadata/sqlutil/group.go
@@ -6,8 +6,6 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/pkg/errors"
-
 	"github.com/jmoiron/sqlx"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
@@ -30,20 +28,20 @@ func UpdateGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	args = append(args, opt.GroupID)
 
 	if len(cond) == 0 {
-		return errors.Errorf("invalid option: nothing to update")
+		return fmt.Errorf("invalid option: nothing to update")
 	}
 
 	query := fmt.Sprintf("UPDATE feature_group SET %s WHERE id = ?", strings.Join(cond, ","))
 	result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(query), args...)
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	rowsAffected, err := result.RowsAffected()
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	if rowsAffected != 1 {
-		return errors.Errorf("failed to update feature group %d: feature group not found", opt.GroupID)
+		return fmt.Errorf("failed to update feature group %d: feature group not found", opt.GroupID)
 	}
 	return nil
 }
@@ -53,9 +51,9 @@ func GetGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, id int) (*types
 	query := `SELECT * FROM feature_group WHERE id = ?`
 	if err := sqlxCtx.GetContext(ctx, &group, sqlxCtx.Rebind(query), id); err != nil {
 		if err == sql.ErrNoRows {
-			return nil, errdefs.NotFound(errors.Errorf("feature group %d not found", id))
+			return nil, errdefs.NotFound(fmt.Errorf("feature group %d not found", id))
 		}
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	entity, err := GetEntity(ctx, sqlxCtx, group.EntityID)
@@ -71,14 +69,14 @@ func GetGroupByName(ctx context.Context, sqlxCtx metadata.SqlxContext, name stri
 	query := `SELECT * FROM feature_group WHERE name = ?`
 	if err := sqlxCtx.GetContext(ctx, &group, sqlxCtx.Rebind(query), name); err != nil {
 		if err == sql.ErrNoRows {
-			return nil, errdefs.NotFound(errors.Errorf("feature group %s not found", name))
+			return nil, errdefs.NotFound(fmt.Errorf("feature group %s not found", name))
 		}
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	entity, err := GetEntity(ctx, sqlxCtx, group.EntityID)
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 	group.Entity = entity
 	return &group, nil
@@ -87,7 +85,7 @@ func GetGroupByName(ctx context.Context, sqlxCtx metadata.SqlxContext, name stri
 func ListGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, entityID *int, groupIDs *[]int) (types.GroupList, error) {
 	cond, args, err := buildListGroupCond(entityID, groupIDs)
 	if err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	query := `SELECT * FROM feature_group`
@@ -97,7 +95,7 @@ func ListGroup(ctx context.Context, sqlxCtx metadata.SqlxContext, entityID *int,
 	query = fmt.Sprintf("%s ORDER BY id ASC", query)
 	var groups types.GroupList
 	if err := sqlxCtx.SelectContext(ctx, &groups, sqlxCtx.Rebind(query), args...); err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	if err := enrichGroups(ctx, sqlxCtx, groups); err != nil {
@@ -120,7 +118,7 @@ func buildListGroupCond(entityID *int, groupIDs *[]int) (string, []interface{},
 		}
 		s, inArgs, err := sqlx.In("id IN (?)", *groupIDs)
 		if err != nil {
-			return "", nil, errors.WithStack(err)
+			return "", nil, err
 		}
 		cond = append(cond, s)
 		args = append(args, inArgs...)
@@ -132,14 +130,14 @@ func enrichGroups(ctx context.Context, sqlxCtx metadata.SqlxContext, groups type
 	entityIDs := groups.EntityIDs()
 	entities, err := ListEntity(ctx, sqlxCtx, &entityIDs)
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	for _, group := range groups {
 		entity := entities.Find(func(e *types.Entity) bool {
 			return group.EntityID == e.ID
 		})
 		if entity == nil {
-			return errors.Errorf("cannot find entity %d for group %d", group.EntityID, group.ID)
+			return fmt.Errorf("cannot find entity %d for group %d", group.EntityID, group.ID)
 		}
 		group.Entity = entity
 	}
diff --git a/internal/database/metadata/sqlutil/revision.go b/internal/database/metadata/sqlutil/revision.go
index f8688738..0c81658c 100644
--- a/internal/database/metadata/sqlutil/revision.go
+++ b/internal/database/metadata/sqlutil/revision.go
@@ -6,8 +6,6 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
@@ -32,21 +30,21 @@ func UpdateRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metad
 		return err
 	}
 	if len(cond) == 0 {
-		return errors.Errorf("invliad option: nothing to update")
+		return fmt.Errorf("invliad option: nothing to update")
 	}
 	args = append(args, opt.RevisionID)
 
 	query := fmt.Sprintf("UPDATE feature_group_revision SET %s WHERE id = ?", strings.Join(cond, ","))
 	result, err := sqlxCtx.ExecContext(ctx, sqlxCtx.Rebind(query), args...)
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	rowsAffected, err := result.RowsAffected()
 	if err != nil {
-		return errors.WithStack(err)
+		return err
 	}
 	if rowsAffected != 1 {
-		return errors.Errorf("failed to update revision %d: revision not found", opt.RevisionID)
+		return fmt.Errorf("failed to update revision %d: revision not found", opt.RevisionID)
 	}
 	return nil
 }
@@ -56,9 +54,9 @@ func GetRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, id int) (*ty
 	query := `SELECT * FROM feature_group_revision WHERE id = ?`
 	if err := sqlxCtx.GetContext(ctx, &revision, sqlxCtx.Rebind(query), id); err != nil {
 		if err == sql.ErrNoRows {
-			return nil, errdefs.NotFound(errors.Errorf("revision %d not found", id))
+			return nil, errdefs.NotFound(fmt.Errorf("revision %d not found", id))
 		}
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	group, err := GetGroup(ctx, sqlxCtx, revision.GroupID)
@@ -74,9 +72,9 @@ func GetRevisionBy(ctx context.Context, sqlxCtx metadata.SqlxContext, groupID in
 	query := `SELECT * FROM feature_group_revision WHERE group_id = ? AND revision = ?`
 	if err := sqlxCtx.GetContext(ctx, &r, sqlxCtx.Rebind(query), groupID, revision); err != nil {
 		if err == sql.ErrNoRows {
-			return nil, errdefs.NotFound(errors.Errorf("revision not found by group %d and revision %d", groupID, revision))
+			return nil, errdefs.NotFound(fmt.Errorf("revision not found by group %d and revision %d", groupID, revision))
 		}
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	group, err := GetGroup(ctx, sqlxCtx, r.GroupID)
@@ -97,7 +95,7 @@ func ListRevision(ctx context.Context, sqlxCtx metadata.SqlxContext, groupID *in
 	query = fmt.Sprintf("%s ORDER BY id ASC", query)
 	var revisions types.RevisionList
 	if err := sqlxCtx.SelectContext(ctx, &revisions, sqlxCtx.Rebind(query), args...); err != nil {
-		return nil, errors.WithStack(err)
+		return nil, err
 	}
 
 	if err := enrichRevisions(ctx, sqlxCtx, revisions); err != nil {
@@ -117,7 +115,7 @@ func enrichRevisions(ctx context.Context, sqlxCtx metadata.SqlxContext, revision
 			return revision.GroupID == g.ID
 		})
 		if group == nil {
-			return errors.Errorf("cannot find group %d for revision %d", revision.GroupID, revision.ID)
+			return fmt.Errorf("cannot find group %d for revision %d", revision.GroupID, revision.ID)
 		}
 		revision.Group = group
 	}
diff --git a/internal/database/offline/sqlutil/push.go b/internal/database/offline/sqlutil/push.go
index 727d61d0..7ef4ba0d 100644
--- a/internal/database/offline/sqlutil/push.go
+++ b/internal/database/offline/sqlutil/push.go
@@ -10,7 +10,17 @@ import (
 
 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)
+
+	columns := make([]string, 0, len(pushOpt.FeatureNames)+2)
+	columns = append(columns, pushOpt.EntityName, "unix_milli")
+	columns = append(columns, pushOpt.FeatureNames...)
+
+	rows := make([]interface{}, 0, len(pushOpt.Records))
+	for _, record := range pushOpt.Records {
+		rows = append(rows, record.ToRow())
+	}
+
+	err := dbutil.InsertRecordsToTable(ctx, dbOpt, tableName, rows, columns)
 	if err != nil && dbutil.IsTableNotFoundError(err, dbOpt.Backend) {
 		return errdefs.NotFound(err)
 	}
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index 6c373a12..fa742375 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -51,10 +51,9 @@ type ImportOpt struct {
 type PushOpt struct {
 	GroupID      int
 	Revision     int64
+	EntityName   string
 	FeatureNames []string
-
-	// Two-dimension slice, each row represents a record
-	FeatureValues []interface{}
+	Records      []types.StreamRecord
 }
 
 type CSVSource struct {
diff --git a/pkg/oomstore/types/feature_values.go b/pkg/oomstore/types/feature_values.go
index 41364546..7fdcbe57 100644
--- a/pkg/oomstore/types/feature_values.go
+++ b/pkg/oomstore/types/feature_values.go
@@ -14,3 +14,16 @@ func (fv *FeatureValues) FeatureValueSlice() []interface{} {
 	}
 	return values
 }
+
+type StreamRecord struct {
+	GroupID   int
+	EntityKey string
+	UnixMilli int64
+	Values    []interface{}
+}
+
+func (r *StreamRecord) ToRow() []interface{} {
+	row := make([]interface{}, 0, len(r.Values)+2)
+	row = append(row, r.EntityKey, r.UnixMilli)
+	return append(row, r.Values...)
+}
diff --git a/pkg/oomstore/types/value_type.go b/pkg/oomstore/types/value_type.go
index ea8de622..73d8e738 100644
--- a/pkg/oomstore/types/value_type.go
+++ b/pkg/oomstore/types/value_type.go
@@ -1,10 +1,9 @@
 package types
 
 import (
+	"fmt"
 	"strconv"
 
-	"github.com/pkg/errors"
-
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 )
 
@@ -50,7 +49,7 @@ func (v ValueType) Validate() error {
 	if _, ok := allValueTypes[v]; ok {
 		return nil
 	} else {
-		return errdefs.InvalidAttribute(errors.Errorf("Invalid value type: %d", v))
+		return errdefs.InvalidAttribute(fmt.Errorf("Invalid value type: %d", v))
 	}
 }
 
@@ -58,6 +57,6 @@ func ParseValueType(s string) (ValueType, error) {
 	if v, ok := allValueTypeStrings[s]; ok {
 		return v, nil
 	} else {
-		return Invalid, errdefs.InvalidAttribute(errors.Errorf("Unknown value type: %s", s))
+		return Invalid, errdefs.InvalidAttribute(fmt.Errorf("Unknown value type: %s", s))
 	}
 }


Test patch:
diff --git a/internal/database/metadata/test_impl/entity.go b/internal/database/metadata/test_impl/entity.go
index 092f30b7..cd23c45f 100644
--- a/internal/database/metadata/test_impl/entity.go
+++ b/internal/database/metadata/test_impl/entity.go
@@ -1,6 +1,7 @@
 package test_impl
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -31,7 +32,7 @@ func TestCreateEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore De
 			Description: "description",
 		},
 	})
-	require.Equal(t, "entity device already exists", err.Error())
+	require.Equal(t, fmt.Errorf("entity device already exists"), err)
 }
 
 func TestGetEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
diff --git a/internal/database/metadata/test_impl/feature.go b/internal/database/metadata/test_impl/feature.go
index 4c45f396..9f3b8129 100644
--- a/internal/database/metadata/test_impl/feature.go
+++ b/internal/database/metadata/test_impl/feature.go
@@ -2,6 +2,7 @@ package test_impl
 
 import (
 	"context"
+	"fmt"
 	"testing"
 	"time"
 
@@ -67,7 +68,7 @@ func TestCreateFeatureWithSameName(t *testing.T, prepareStore PrepareStoreFn, de
 	require.NoError(t, err)
 
 	_, err = store.CreateFeature(ctx, opt)
-	assert.Equal(t, "feature phone already exists", err.Error())
+	assert.Equal(t, err, fmt.Errorf("feature phone already exists"))
 }
 
 func TestCreateFeatureWithSQLKeyword(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
diff --git a/internal/database/metadata/test_impl/group.go b/internal/database/metadata/test_impl/group.go
index d9bfa307..085c5846 100644
--- a/internal/database/metadata/test_impl/group.go
+++ b/internal/database/metadata/test_impl/group.go
@@ -2,6 +2,7 @@ package test_impl
 
 import (
 	"context"
+	"fmt"
 	"testing"
 
 	"github.com/oom-ai/oomstore/internal/database/metadata"
@@ -145,7 +146,7 @@ func TestCreateGroup(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 
 	// cannot create feature group with same name
 	_, err = store.CreateGroup(ctx, opt)
-	require.Equal(t, "feature group device_info already exists", err.Error())
+	require.Equal(t, fmt.Errorf("feature group device_info already exists"), err)
 
 	// cannot create feature group with invalid category
 	opt.Category = "invalid-category"
diff --git a/internal/database/offline/test_impl/push.go b/internal/database/offline/test_impl/push.go
index c2b142b2..d1c7dcf3 100644
--- a/internal/database/offline/test_impl/push.go
+++ b/internal/database/offline/test_impl/push.go
@@ -41,17 +41,27 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		},
 	}
 	revision := int64(2)
+	records := []types.StreamRecord{
+		{
 
-	featureValues := []interface{}{
-		[]interface{}{18, "F"},
-		[]interface{}{21, "M"},
+			EntityKey: "1000",
+			UnixMilli: 8000,
+			Values:    []interface{}{18, "F"},
+		},
+		{
+
+			EntityKey: "1001",
+			UnixMilli: 8001,
+			Values:    []interface{}{21, "M"},
+		},
 	}
 
 	pushOpt := offline.PushOpt{
-		GroupID:       group.ID,
-		Revision:      revision,
-		FeatureNames:  features.Names(),
-		FeatureValues: featureValues,
+		GroupID:      group.ID,
+		Revision:     revision,
+		EntityName:   entity.Name,
+		FeatureNames: features.Names(),
+		Records:      records,
 	}
 
 	t.Run("push when cdc table not exists", func(t *testing.T) {