Repository: oom-ai/oomstore
Pull request #908 (2022-01-06 03:27:35)

feat(online): error add stack trace
relate #903


Patch:
diff --git a/README.md b/README.md
index a5909324..311a328c 100644
--- a/README.md
+++ b/README.md
@@ -80,6 +80,7 @@ metadata-store: *pg
 ```yaml
 kind: Entity
 name: user
+length: 8
 description: 'user ID'
 groups:
 - name: account
diff --git a/internal/database/dbutil/database.go b/internal/database/dbutil/database.go
index b52e3311..4bd8ffd0 100644
--- a/internal/database/dbutil/database.go
+++ b/internal/database/dbutil/database.go
@@ -57,6 +57,7 @@ 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 {
+	err = errors.Cause(err)
 	switch backend {
 	// https://github.com/mattn/go-sqlite3/issues/244
 	case types.BackendSQLite:
diff --git a/internal/database/dbutil/db_opt.go b/internal/database/dbutil/db_opt.go
index c3124dd3..ba799726 100644
--- a/internal/database/dbutil/db_opt.go
+++ b/internal/database/dbutil/db_opt.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/pkg/errors"
 	"github.com/spf13/cast"
 
 	"cloud.google.com/go/bigquery"
@@ -30,10 +31,10 @@ func (d *DBOpt) ExecContext(ctx context.Context, query string, args []interface{
 			query = strings.Replace(query, "?", cast.ToString(arg), 1)
 		}
 		_, err := d.BigQueryDB.Query(query).Read(ctx)
-		return err
+		return errors.WithStack(err)
 	default:
 		_, err := d.SqlxDB.ExecContext(ctx, d.SqlxDB.Rebind(query), args...)
-		return err
+		return errors.WithStack(err)
 	}
 }
 
@@ -53,7 +54,8 @@ func (d *DBOpt) BuildInsertQuery(tableName string, records []interface{}, column
 	if d.Backend == types.BackendBigQuery {
 		tableName = fmt.Sprintf("%s.%s", *d.DatasetID, tableName)
 	}
-	return sqlx.In(
+	query, args, err := sqlx.In(
 		fmt.Sprintf(`INSERT INTO %s (%s) VALUES %s`, tableName, columnStr, strings.Join(valueFlags, ",")),
 		records...)
+	return query, args, errors.WithStack(err)
 }
diff --git a/internal/database/dbutil/schema.go b/internal/database/dbutil/schema.go
index b23ae56c..82ffe28d 100644
--- a/internal/database/dbutil/schema.go
+++ b/internal/database/dbutil/schema.go
@@ -80,10 +80,10 @@ func parseColumns(entity *types.Entity, isCDC bool, features types.FeatureList,
 	{
 		c := Column{Name: entity.Name, ValueType: types.String}
 		switch backend {
-		case types.BackendCassandra, types.BackendSQLite, types.BackendPostgres, types.BackendRedshift, types.BackendSnowflake:
+		case types.BackendCassandra, types.BackendSQLite:
 			c.DbType = "TEXT"
-		case types.BackendMySQL:
-			c.DbType = "VARCHAR(255)"
+		case types.BackendPostgres, types.BackendRedshift, types.BackendSnowflake, types.BackendMySQL:
+			c.DbType = fmt.Sprintf(`VARCHAR(%d)`, entity.Length)
 		case types.BackendBigQuery:
 			c.DbType = "STRING"
 		default:
diff --git a/internal/database/dbutil/sql.go b/internal/database/dbutil/sql.go
index 4322e786..c376a83a 100644
--- a/internal/database/dbutil/sql.go
+++ b/internal/database/dbutil/sql.go
@@ -53,7 +53,7 @@ func InsertRecordsToTableTx(tx *sqlx.Tx, ctx context.Context, tableName string,
 		return nil
 	}
 	if _, err := tx.ExecContext(ctx, tx.Rebind(query), args...); err != nil {
-		return err
+		return errors.WithStack(err)
 	}
 	return nil
 }
@@ -66,7 +66,7 @@ func GetColumnFormat(backendType types.BackendType) (string, error) {
 	case types.BackendMySQL, types.BackendSQLite, types.BackendBigQuery:
 		columnFormat = "`%s` %s"
 	default:
-		return "", fmt.Errorf("unsupported backend type %s", backendType)
+		return "", errors.Errorf("unsupported backend type %s", backendType)
 	}
 	return columnFormat, nil
 }
diff --git a/internal/database/dbutil/type_mapping.go b/internal/database/dbutil/type_mapping.go
index a5f98848..aaf8e77d 100644
--- a/internal/database/dbutil/type_mapping.go
+++ b/internal/database/dbutil/type_mapping.go
@@ -1,9 +1,10 @@
 package dbutil
 
 import (
-	"fmt"
 	"strings"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -28,12 +29,12 @@ func DBValueType(backend types.BackendType, valueType types.ValueType) (string,
 	case types.BackendBigQuery:
 		mp = valueTypeToBigQueryType
 	default:
-		return "", errdefs.InvalidAttribute(fmt.Errorf("unsupported backend: %s", backend))
+		return "", errdefs.InvalidAttribute(errors.Errorf("unsupported backend: %s", backend))
 	}
 
 	t, ok := mp[valueType]
 	if !ok {
-		return "", errdefs.InvalidAttribute(fmt.Errorf("unsupported value type: %s", valueType))
+		return "", errdefs.InvalidAttribute(errors.Errorf("unsupported value type: %s", valueType))
 	}
 	return t, nil
 }
@@ -55,7 +56,7 @@ func ValueType(backend types.BackendType, dbValueType string) (types.ValueType,
 	case types.BackendRedshift:
 		mp = redshiftTypeToValueType
 	default:
-		return 0, errdefs.InvalidAttribute(fmt.Errorf("unsupported backend: %s", backend))
+		return 0, errdefs.InvalidAttribute(errors.Errorf("unsupported backend: %s", backend))
 	}
 	dbValueType = strings.ToLower(dbValueType)
 	i := strings.Index(dbValueType, "(")
@@ -64,7 +65,7 @@ func ValueType(backend types.BackendType, dbValueType string) (types.ValueType,
 	}
 	t, ok := mp[dbValueType]
 	if !ok {
-		return 0, errdefs.InvalidAttribute(fmt.Errorf("unsupported db value type: %s", dbValueType))
+		return 0, errdefs.InvalidAttribute(errors.Errorf("unsupported db value type: %s", dbValueType))
 	}
 
 	return t, nil
diff --git a/internal/database/metadata/mysql/entity.go b/internal/database/metadata/mysql/entity.go
index b9521a37..eb3066ca 100644
--- a/internal/database/metadata/mysql/entity.go
+++ b/internal/database/metadata/mysql/entity.go
@@ -2,27 +2,27 @@ 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) {
-	query := "INSERT INTO entity(name, description) VALUES(?, ?)"
-	res, err := sqlxCtx.ExecContext(ctx, query, opt.EntityName, opt.Description)
+	query := "INSERT INTO entity(name, length, description) VALUES(?, ?, ?)"
+	res, err := sqlxCtx.ExecContext(ctx, query, opt.EntityName, opt.Length, opt.Description)
 	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/schema.go b/internal/database/metadata/mysql/schema.go
index b50f2a92..56319a8c 100644
--- a/internal/database/metadata/mysql/schema.go
+++ b/internal/database/metadata/mysql/schema.go
@@ -32,6 +32,7 @@ var META_TABLE_SCHEMAS = map[string]string{
 		CREATE TABLE entity (
 			id				INT 			NOT	NULL AUTO_INCREMENT PRIMARY KEY,
 			name        	VARCHAR(32) 	NOT	NULL,
+			length      	SMALLINT    	NOT	NULL COMMENT "feature entity value max length",
 			description 	VARCHAR(64) 	DEFAULT '',
 			create_time 	TIMESTAMP   	NOT	NULL DEFAULT CURRENT_TIMESTAMP,
 			modify_time 	TIMESTAMP   	NOT	NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
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..e7cdf7ce 100644
--- a/internal/database/metadata/postgres/entity.go
+++ b/internal/database/metadata/postgres/entity.go
@@ -2,22 +2,21 @@ 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"
 )
 
 func createEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateEntityOpt) (int, error) {
 	var entityID int
-	query := "INSERT INTO entity(name, description) VALUES($1, $2) returning id"
-	err := sqlxCtx.GetContext(ctx, &entityID, query, opt.EntityName, opt.Description)
+	query := "INSERT INTO entity(name, length, description) VALUES($1, $2, $3) returning id"
+	err := sqlxCtx.GetContext(ctx, &entityID, query, opt.EntityName, opt.Length, 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/schema.go b/internal/database/metadata/postgres/schema.go
index 3ff4a79a..b976b440 100644
--- a/internal/database/metadata/postgres/schema.go
+++ b/internal/database/metadata/postgres/schema.go
@@ -53,11 +53,13 @@ var META_TABLE_SCHEMAS = map[string]string{
 		CREATE TABLE entity (
 			id INT PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
 			name        VARCHAR(32) NOT     NULL,
+			length      SMALLINT    NOT     NULL,
 			description VARCHAR(64) DEFAULT '',
 			create_time TIMESTAMP   NOT     NULL DEFAULT CURRENT_TIMESTAMP,
 			modify_time TIMESTAMP   NOT     NULL DEFAULT CURRENT_TIMESTAMP,
 			UNIQUE (name)
 		);
+		COMMENT ON COLUMN entity.length IS 'feature entity value max length';
 		`,
 	"feature_group_revision": `
 		CREATE TABLE feature_group_revision (
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..ba7cf195 100644
--- a/internal/database/metadata/sqlite/entity.go
+++ b/internal/database/metadata/sqlite/entity.go
@@ -2,20 +2,20 @@ package sqlite
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/mattn/go-sqlite3"
-	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 )
 
 func createEntity(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata.CreateEntityOpt) (int, error) {
-	query := "INSERT INTO entity(name, description) VALUES(?, ?)"
-	res, err := sqlxCtx.ExecContext(ctx, query, opt.EntityName, opt.Description)
+	query := "INSERT INTO entity(name, length, description) VALUES(?, ?, ?)"
+	res, err := sqlxCtx.ExecContext(ctx, query, opt.EntityName, opt.Length, opt.Description)
 	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/schema.go b/internal/database/metadata/sqlite/schema.go
index 29f969bb..65cea474 100644
--- a/internal/database/metadata/sqlite/schema.go
+++ b/internal/database/metadata/sqlite/schema.go
@@ -43,6 +43,7 @@ var META_TABLE_SCHEMAS = map[string]string{
 		CREATE TABLE entity (
 			id				INTEGER 		NOT	NULL PRIMARY KEY AUTOINCREMENT,
 			name        	VARCHAR(32) 	NOT	NULL,
+			length      	SMALLINT    	NOT	NULL,
 			description 	VARCHAR(64) 	DEFAULT '',
 			create_time 	TIMESTAMP   	NOT	NULL DEFAULT CURRENT_TIMESTAMP,
 			modify_time 	TIMESTAMP   	NOT	NULL DEFAULT CURRENT_TIMESTAMP,
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/join_helper.go b/internal/database/offline/sqlutil/join_helper.go
index aadc3be4..aa3b00f9 100644
--- a/internal/database/offline/sqlutil/join_helper.go
+++ b/internal/database/offline/sqlutil/join_helper.go
@@ -166,12 +166,8 @@ func prepareTableSchema(dbOpt dbutil.DBOpt, entity types.Entity, tableName strin
 		entityType = "STRING"
 		valueType = "STRING"
 		qtTableName = fmt.Sprintf("%s.%s", *dbOpt.DatasetID, qt(tableName))
-	case types.BackendMySQL:
-		entityType = "VARCHAR(255)"
-		valueType = "TEXT"
-		qtTableName = qt(tableName)
 	default:
-		entityType = "TEXT"
+		entityType = fmt.Sprintf("VARCHAR(%d)", entity.Length)
 		valueType = "TEXT"
 		qtTableName = qt(tableName)
 	}
diff --git a/internal/database/offline/sqlutil/push.go b/internal/database/offline/sqlutil/push.go
index 7ef4ba0d..727d61d0 100644
--- a/internal/database/offline/sqlutil/push.go
+++ b/internal/database/offline/sqlutil/push.go
@@ -10,17 +10,7 @@ import (
 
 func Push(ctx context.Context, dbOpt dbutil.DBOpt, pushOpt offline.PushOpt) error {
 	tableName := dbutil.OfflineStreamCdcTableName(pushOpt.GroupID, pushOpt.Revision)
-
-	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)
+	err := dbutil.InsertRecordsToTable(ctx, dbOpt, tableName, pushOpt.FeatureValues, pushOpt.FeatureNames)
 	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 fa742375..6c373a12 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -51,9 +51,10 @@ type ImportOpt struct {
 type PushOpt struct {
 	GroupID      int
 	Revision     int64
-	EntityName   string
 	FeatureNames []string
-	Records      []types.StreamRecord
+
+	// Two-dimension slice, each row represents a record
+	FeatureValues []interface{}
 }
 
 type CSVSource struct {
diff --git a/internal/database/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index 61f0c524..f9486cca 100644
--- a/internal/database/online/cassandra/get.go
+++ b/internal/database/online/cassandra/get.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/gocql/gocql"
+	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
@@ -32,7 +33,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 		if err == gocql.ErrNotFound || isTableNotFoundError(err, tableName) {
 			return scan, nil
 		}
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 
 	rs := make(map[string]interface{}, len(scan))
@@ -71,7 +72,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 		if err == gocql.ErrNotFound || isTableNotFoundError(err, tableName) {
 			return rs, nil
 		}
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 
 	for _, s := range scan {
diff --git a/internal/database/online/cassandra/import.go b/internal/database/online/cassandra/import.go
index 1f20861e..878804a1 100644
--- a/internal/database/online/cassandra/import.go
+++ b/internal/database/online/cassandra/import.go
@@ -6,6 +6,7 @@ import (
 	"strings"
 
 	"github.com/gocql/gocql"
+	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
@@ -20,7 +21,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 
 	// create table
 	if err := db.Query(table).Exec(); err != nil {
-		return err
+		return errors.WithStack(err)
 	}
 
 	var (
@@ -29,19 +30,19 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	)
 	for record := range opt.ExportStream {
 		if len(record) != len(opt.Features)+1 {
-			return fmt.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
+			return errors.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
 		}
 
 		if batch.Size() != BatchSize {
 			batch.Query(insertStmt, record...)
 		} else {
 			if err := db.ExecuteBatch(batch); err != nil {
-				return err
+				return errors.WithStack(err)
 			}
 			batch = db.NewBatch(gocql.LoggedBatch)
 		}
 	}
-	return db.ExecuteBatch(batch)
+	return errors.WithStack(db.ExecuteBatch(batch))
 }
 
 func buildInsertStatement(tableName string, columns []string) string {
diff --git a/internal/database/online/cassandra/purge.go b/internal/database/online/cassandra/purge.go
index 72e6cfa4..86aadb95 100644
--- a/internal/database/online/cassandra/purge.go
+++ b/internal/database/online/cassandra/purge.go
@@ -4,10 +4,13 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
 )
 
 func (db *DB) Purge(ctx context.Context, revisionID int) error {
-	return db.Query(fmt.Sprintf("DROP TABLE IF EXISTS %s", sqlutil.OnlineBatchTableName(revisionID))).
+	err := db.Query(fmt.Sprintf("DROP TABLE IF EXISTS %s", sqlutil.OnlineBatchTableName(revisionID))).
 		WithContext(ctx).Exec()
+	return errors.WithStack(err)
 }
diff --git a/internal/database/online/cassandra/store.go b/internal/database/online/cassandra/store.go
index 63521a48..8016d439 100644
--- a/internal/database/online/cassandra/store.go
+++ b/internal/database/online/cassandra/store.go
@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/gocql/gocql"
+	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -45,7 +46,7 @@ func Open(option *types.CassandraOpt) (*DB, error) {
 
 	session, err := cluster.CreateSession()
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 	return &DB{Session: session}, nil
 }
diff --git a/internal/database/online/cassandra/stream.go b/internal/database/online/cassandra/stream.go
index 6e2d3958..01a26d89 100644
--- a/internal/database/online/cassandra/stream.go
+++ b/internal/database/online/cassandra/stream.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/pingcap/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
@@ -28,7 +30,8 @@ func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTa
 
 	sql := fmt.Sprintf("ALTER TABLE %s ADD %s %s", tableName, opt.Feature.Name, dbValueType)
 
-	return db.Query(sql).WithContext(ctx).Exec()
+	err = db.Query(sql).WithContext(ctx).Exec()
+	return errors.WithStack(err)
 }
 
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
@@ -43,5 +46,6 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 		cond.InsertPlaceholders,
 	)
 
-	return db.Query(query, cond.InsertValues...).WithContext(ctx).Exec()
+	err := db.Query(query, cond.InsertValues...).WithContext(ctx).Exec()
+	return errors.WithStack(err)
 }
diff --git a/internal/database/online/dynamodb/get.go b/internal/database/online/dynamodb/get.go
index 697c1cb5..47a8c4cf 100644
--- a/internal/database/online/dynamodb/get.go
+++ b/internal/database/online/dynamodb/get.go
@@ -2,13 +2,13 @@ package dynamodb
 
 import (
 	"context"
-	"errors"
-	"fmt"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
@@ -29,7 +29,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 
 	entityKeyValue, err := attributevalue.Marshal(opt.EntityKey)
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 	result, err := db.GetItem(ctx, &dynamodb.GetItemInput{
 		TableName: aws.String(tableName),
@@ -41,7 +41,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 		if apiErr := new(types.ResourceNotFoundException); errors.As(err, &apiErr) {
 			return make(dbutil.RowMap), nil
 		}
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 	return deserializeFeatureValues(opt.Features, result.Item)
 }
@@ -60,7 +60,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	for _, entityKey := range opt.EntityKeys {
 		entityKeyValue, err := attributevalue.Marshal(entityKey)
 		if err != nil {
-			return nil, err
+			return nil, errors.WithStack(err)
 		}
 		keys = append(keys, map[string]types.AttributeValue{
 			opt.Entity.Name: entityKeyValue,
@@ -88,18 +88,18 @@ func batchGetItem(ctx context.Context, db *DB, keys []map[string]types.Attribute
 	})
 	if err != nil {
 		if apiErr := new(types.ResourceNotFoundException); !errors.As(err, &apiErr) {
-			return err
+			return errors.WithStack(err)
 		}
 	}
 
 	for _, item := range result.Responses[tableName] {
 		entityKeyValue, ok := item[entityName]
 		if !ok {
-			return fmt.Errorf("could not find entity key column %s in table %s", entityName, tableName)
+			return errors.Errorf("could not find entity key column %s in table %s", entityName, tableName)
 		}
 		var entityKey string
 		if err = attributevalue.Unmarshal(entityKeyValue, &entityKey); err != nil {
-			return err
+			return errors.WithStack(err)
 		}
 		rowMap, err := deserializeFeatureValues(features, item)
 		if err != nil {
@@ -119,10 +119,10 @@ func deserializeFeatureValues(features oomTypes.FeatureList, item map[string]typ
 	for _, feature := range features {
 		attributeValue, ok := item[feature.Name]
 		if !ok {
-			return nil, fmt.Errorf("could not find feature %s", feature.Name)
+			return nil, errors.Errorf("could not find feature %s", feature.Name)
 		}
 		if err := attributevalue.Unmarshal(attributeValue, &value); err != nil {
-			return nil, err
+			return nil, errors.WithStack(err)
 		}
 		typedValue, err := deserializeByTag(value, feature.ValueType)
 		if err != nil {
diff --git a/internal/database/online/dynamodb/import.go b/internal/database/online/dynamodb/import.go
index ff375028..ac242080 100644
--- a/internal/database/online/dynamodb/import.go
+++ b/internal/database/online/dynamodb/import.go
@@ -10,6 +10,7 @@ import (
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
 	oomTypes "github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/pkg/errors"
 )
 
 const (
@@ -39,7 +40,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 		},
 	})
 	if err != nil {
-		return err
+		return errors.WithStack(err)
 	}
 
 	// Step 2: import items to the table
@@ -87,7 +88,7 @@ func buildItem(record oomTypes.ExportRecord, opt online.ImportOpt) (map[string]t
 	item := make(map[string]types.AttributeValue)
 	entityKeyValue, err := attributevalue.Marshal(record.EntityKey())
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 	item[opt.Entity.Name] = entityKeyValue
 
@@ -98,7 +99,7 @@ func buildItem(record oomTypes.ExportRecord, opt online.ImportOpt) (map[string]t
 		}
 		attributeValue, err := attributevalue.Marshal(value)
 		if err != nil {
-			return nil, err
+			return nil, errors.WithStack(err)
 		}
 		item[feature.Name] = attributeValue
 	}
diff --git a/internal/database/online/dynamodb/purge.go b/internal/database/online/dynamodb/purge.go
index 6cb3e613..7d257b9d 100644
--- a/internal/database/online/dynamodb/purge.go
+++ b/internal/database/online/dynamodb/purge.go
@@ -3,6 +3,8 @@ package dynamodb
 import (
 	"context"
 
+	"github.com/pkg/errors"
+
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
 	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
@@ -13,5 +15,5 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 	_, err := db.DeleteTable(ctx, &dynamodb.DeleteTableInput{
 		TableName: aws.String(tableName),
 	})
-	return err
+	return errors.WithStack(err)
 }
diff --git a/internal/database/online/dynamodb/store.go b/internal/database/online/dynamodb/store.go
index 1d9e0148..408758be 100644
--- a/internal/database/online/dynamodb/store.go
+++ b/internal/database/online/dynamodb/store.go
@@ -7,6 +7,8 @@ import (
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/credentials"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -38,7 +40,7 @@ func Open(opt *types.DynamoDBOpt) (*DB, error) {
 		}),
 	)
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 
 	return &DB{dynamodb.NewFromConfig(cfg)}, nil
@@ -46,7 +48,7 @@ func Open(opt *types.DynamoDBOpt) (*DB, error) {
 
 func (db *DB) Ping(ctx context.Context) error {
 	_, err := db.ListTables(ctx, nil)
-	return err
+	return errors.WithStack(err)
 }
 
 // dynamodb is serverless so Close won't do anything
diff --git a/internal/database/online/dynamodb/stream.go b/internal/database/online/dynamodb/stream.go
index f8bb7c9a..3a1c97fd 100644
--- a/internal/database/online/dynamodb/stream.go
+++ b/internal/database/online/dynamodb/stream.go
@@ -7,6 +7,8 @@ import (
 	"github.com/aws/aws-sdk-go-v2/feature/dynamodb/attributevalue"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
 )
@@ -38,7 +40,7 @@ func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTa
 			WriteCapacityUnits: aws.Int64(10),
 		},
 	})
-	return err
+	return errors.WithStack(err)
 }
 
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
@@ -49,7 +51,7 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 
 	entityKeyValue, err := attributevalue.Marshal(opt.EntityKey)
 	if err != nil {
-		return err
+		return errors.WithStack(err)
 	}
 	item[opt.Entity.Name] = entityKeyValue
 
@@ -60,7 +62,7 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 		}
 		attributevalue, err := attributevalue.Marshal(value)
 		if err != nil {
-			return err
+			return errors.WithStack(err)
 		}
 		item[feature.Name] = attributevalue
 	}
@@ -69,5 +71,5 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 		TableName: &tableName,
 		Item:      item,
 	})
-	return err
+	return errors.WithStack(err)
 }
diff --git a/internal/database/online/dynamodb/util.go b/internal/database/online/dynamodb/util.go
index 10e6e873..f25a3094 100644
--- a/internal/database/online/dynamodb/util.go
+++ b/internal/database/online/dynamodb/util.go
@@ -1,16 +1,17 @@
 package dynamodb
 
 import (
-	"fmt"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
 func serializeByTag(i interface{}, valueType types.ValueType) (out interface{}, err error) {
 	defer func() {
 		if r := recover(); r != nil {
-			err = fmt.Errorf("failed to serailize by tag: %v", r)
+			err = errors.Errorf("failed to serailize by tag: %v", r)
 		}
 	}()
 
@@ -31,13 +32,13 @@ func deserializeByTag(i interface{}, valueType types.ValueType) (interface{}, er
 	case types.Int64:
 		v, ok := i.(float64)
 		if !ok {
-			return "", fmt.Errorf("not float64 %v", i)
+			return "", errors.Errorf("not float64 %v", i)
 		}
 		return int64(v), nil
 	case types.Time:
 		v, ok := i.(float64)
 		if !ok {
-			return "", fmt.Errorf("not float64 %v", i)
+			return "", errors.Errorf("not float64 %v", i)
 		}
 		return time.UnixMilli(int64(v)), nil
 	default:
diff --git a/internal/database/online/kvutil/serde.go b/internal/database/online/kvutil/serde.go
index 049ea7c9..c580d77b 100644
--- a/internal/database/online/kvutil/serde.go
+++ b/internal/database/online/kvutil/serde.go
@@ -1,10 +1,11 @@
 package kvutil
 
 import (
-	"fmt"
 	"strconv"
 	"time"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
@@ -18,7 +19,7 @@ const (
 func SerializeByValueType(i interface{}, valueType types.ValueType) (s string, err error) {
 	defer func() {
 		if r := recover(); r != nil {
-			err = fmt.Errorf("failed to serailize by tag: %v", r)
+			err = errors.Errorf("failed to serailize by tag: %v", r)
 		}
 	}()
 
@@ -41,7 +42,7 @@ func SerializeByValueType(i interface{}, valueType types.ValueType) (s string, e
 	case types.Bytes:
 		return string(i.([]byte)), nil
 	default:
-		return "", fmt.Errorf("unable to serialize %#v of type %T to string", i, i)
+		return "", errors.Errorf("unable to serialize %#v of type %T to string", i, i)
 	}
 }
 
@@ -89,7 +90,7 @@ func SerializeByValue(i interface{}) (string, error) {
 		}
 
 	default:
-		return "", fmt.Errorf("unable to serialize %#v of type %T to string", i, i)
+		return "", errors.Errorf("unable to serialize %#v of type %T to string", i, i)
 	}
 }
 
@@ -100,7 +101,7 @@ func DeserializeByValueType(i interface{}, valueType types.ValueType) (interface
 
 	s, ok := i.(string)
 	if !ok {
-		return nil, fmt.Errorf("not a string or nil: %v", i)
+		return nil, errors.Errorf("not a string or nil: %v", i)
 	}
 
 	switch valueType {
@@ -121,7 +122,7 @@ func DeserializeByValueType(i interface{}, valueType types.ValueType) (interface
 		} else if s == "0" {
 			return false, nil
 		} else {
-			return nil, fmt.Errorf("invalid bool value: %s", s)
+			return nil, errors.Errorf("invalid bool value: %s", s)
 		}
 	case types.Time:
 		x, err := strconv.ParseInt(s, serializeIntBase, 64)
@@ -130,6 +131,6 @@ func DeserializeByValueType(i interface{}, valueType types.ValueType) (interface
 	case types.Bytes:
 		return []byte(s), nil
 	default:
-		return "", fmt.Errorf("unsupported value type: %s", valueType)
+		return "", errors.Errorf("unsupported value type: %s", valueType)
 	}
 }
diff --git a/internal/database/online/mysql/store.go b/internal/database/online/mysql/store.go
index 359fab36..0b89a363 100644
--- a/internal/database/online/mysql/store.go
+++ b/internal/database/online/mysql/store.go
@@ -6,6 +6,7 @@ import (
 
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
@@ -29,7 +30,7 @@ func Open(opt *types.MySQLOpt) (*DB, error) {
 }
 
 func (db *DB) Ping(ctx context.Context) error {
-	return db.PingContext(ctx)
+	return errors.WithStack(db.PingContext(ctx))
 }
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
@@ -61,7 +62,7 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	)
 
 	_, err := db.ExecContext(ctx, db.Rebind(query), append(cond.InsertValues, cond.UpdateValues...)...)
-	return err
+	return errors.WithStack(err)
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index 1642035d..3c1ac2eb 100644
--- a/internal/database/online/postgres/store.go
+++ b/internal/database/online/postgres/store.go
@@ -6,6 +6,7 @@ import (
 
 	"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/internal/database/online"
@@ -29,7 +30,7 @@ func Open(opt *types.PostgresOpt) (*DB, error) {
 }
 
 func (db *DB) Ping(ctx context.Context) error {
-	return db.PingContext(ctx)
+	return errors.WithStack(db.PingContext(ctx))
 }
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
@@ -62,7 +63,7 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	)
 
 	_, err := db.ExecContext(ctx, db.Rebind(query), append(cond.InsertValues, cond.UpdateValues...)...)
-	return err
+	return errors.WithStack(err)
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
diff --git a/internal/database/online/redis/get.go b/internal/database/online/redis/get.go
index da30f2c8..b85578e3 100644
--- a/internal/database/online/redis/get.go
+++ b/internal/database/online/redis/get.go
@@ -3,6 +3,8 @@ package redis
 import (
 	"context"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
@@ -35,7 +37,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 
 	values, err := db.HMGet(ctx, key, featureIDs...).Result()
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 
 	rowMap := make(dbutil.RowMap)
@@ -64,7 +66,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 			Features:   opt.Features,
 		})
 		if err != nil {
-			return res, err
+			return res, errors.WithStack(err)
 		}
 		if len(rowMap) > 0 {
 			res[entityKey] = rowMap
diff --git a/internal/database/online/redis/import.go b/internal/database/online/redis/import.go
index ddac3450..2844c112 100644
--- a/internal/database/online/redis/import.go
+++ b/internal/database/online/redis/import.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
 )
@@ -47,14 +49,14 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 		seq++
 		if seq%PipelineBatchSize == 0 {
 			if _, err := pipe.Exec(ctx); err != nil {
-				return err
+				return errors.WithStack(err)
 			}
 		}
 	}
 
 	if seq%PipelineBatchSize != 0 {
 		if _, err := pipe.Exec(ctx); err != nil {
-			return err
+			return errors.WithStack(err)
 		}
 	}
 	if opt.ExportError != nil {
diff --git a/internal/database/online/redis/purge.go b/internal/database/online/redis/purge.go
index 83e7478b..2e000828 100644
--- a/internal/database/online/redis/purge.go
+++ b/internal/database/online/redis/purge.go
@@ -3,6 +3,8 @@ package redis
 import (
 	"context"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
 )
 
@@ -18,12 +20,12 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 	for {
 		keys, cursor, err = db.Scan(ctx, cursor, pattern, PipelineBatchSize).Result()
 		if err != nil {
-			return err
+			return errors.WithStack(err)
 		}
 
 		if len(keys) > 0 {
 			if _, err = db.Del(ctx, keys...).Result(); err != nil {
-				return err
+				return errors.WithStack(err)
 			}
 		}
 
diff --git a/internal/database/online/sqlite/store.go b/internal/database/online/sqlite/store.go
index 629ce2d9..723abec9 100644
--- a/internal/database/online/sqlite/store.go
+++ b/internal/database/online/sqlite/store.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 
 	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
@@ -28,7 +29,7 @@ func Open(opt *types.SQLiteOpt) (*DB, error) {
 }
 
 func (db *DB) Ping(ctx context.Context) error {
-	return db.PingContext(ctx)
+	return errors.WithStack(db.PingContext(ctx))
 }
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
@@ -59,7 +60,7 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	)
 
 	_, err := db.ExecContext(ctx, db.Rebind(query), cond.InsertValues...)
-	return err
+	return errors.WithStack(err)
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index 824b3c2e..2ef44fc9 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
 	"github.com/spf13/cast"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
@@ -30,7 +31,7 @@ func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.Back
 		if err == sql.ErrNoRows || dbutil.IsTableNotFoundError(err, backend) {
 			return make(dbutil.RowMap), nil
 		}
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 
 	rs, err := deserializeIntoRowMap(record, opt.Features, backend)
@@ -54,12 +55,12 @@ func MultiGet(ctx context.Context, db *sqlx.DB, opt online.MultiGetOpt, backend
 	query := fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s in (?);`, qt(opt.Entity.Name), qt(featureNames...), qt(tableName), qt(opt.Entity.Name))
 	sql, args, err := sqlx.In(query, opt.EntityKeys)
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 
 	rows, err := db.QueryxContext(ctx, db.Rebind(sql), args...)
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 	defer rows.Close()
 
@@ -71,7 +72,7 @@ func getFeatureValueMapFromRows(rows *sqlx.Rows, features types.FeatureList, bac
 	for rows.Next() {
 		record, err := rows.SliceScan()
 		if err != nil {
-			return nil, err
+			return nil, errors.WithStack(err)
 		}
 		entityKey, values := dbutil.DeserializeString(record[0], backend), record[1:]
 		rowMap, err := deserializeIntoRowMap(values, features, backend)
@@ -115,7 +116,7 @@ func deserializeByTag(i interface{}, valueType types.ValueType, backend types.Ba
 			} else if s == "0" || s == "false" {
 				return false, nil
 			} else {
-				return nil, fmt.Errorf("invalid bool value: %s", i)
+				return nil, errors.Errorf("invalid bool value: %s", i)
 			}
 		}
 		return i, nil
diff --git a/internal/database/online/sqlutil/import.go b/internal/database/online/sqlutil/import.go
index ecbd3b7a..67a8d7eb 100644
--- a/internal/database/online/sqlutil/import.go
+++ b/internal/database/online/sqlutil/import.go
@@ -2,12 +2,12 @@ package sqlutil
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/jmoiron/sqlx"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/pkg/errors"
 )
 
 const importBatchSize = 10
@@ -20,14 +20,14 @@ func Import(ctx context.Context, db *sqlx.DB, opt online.ImportOpt, backend type
 		schema := dbutil.BuildTableSchema(tableName, opt.Entity, false, opt.Features, []string{opt.Entity.Name}, backend)
 		_, err := tx.ExecContext(ctx, schema)
 		if err != nil {
-			return err
+			return errors.WithStack(err)
 		}
 
 		// populate the data table
 		records := make([]interface{}, 0, importBatchSize)
 		for record := range opt.ExportStream {
 			if len(record) != len(opt.Features)+1 {
-				return fmt.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
+				return errors.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
 			}
 			records = append(records, record)
 
diff --git a/internal/database/online/sqlutil/purge.go b/internal/database/online/sqlutil/purge.go
index ba940178..ba78bc71 100644
--- a/internal/database/online/sqlutil/purge.go
+++ b/internal/database/online/sqlutil/purge.go
@@ -5,6 +5,8 @@ import (
 	"fmt"
 
 	"github.com/jmoiron/sqlx"
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -12,8 +14,6 @@ import (
 func Purge(ctx context.Context, db *sqlx.DB, revisionID int, backend types.BackendType) error {
 	qt := dbutil.QuoteFn(backend)
 	query := fmt.Sprintf(`DROP TABLE IF EXISTS %s;`, qt(OnlineBatchTableName(revisionID)))
-	if _, err := db.ExecContext(ctx, query); err != nil {
-		return err
-	}
-	return nil
+	_, err := db.ExecContext(ctx, query)
+	return errors.WithStack(err)
 }
diff --git a/internal/database/online/sqlutil/util.go b/internal/database/online/sqlutil/util.go
index 7b3a89b4..4dc3b3d0 100644
--- a/internal/database/online/sqlutil/util.go
+++ b/internal/database/online/sqlutil/util.go
@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/pkg/errors"
+
 	"github.com/jmoiron/sqlx"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
@@ -25,11 +27,13 @@ func CreateStreamTableSchema(ctx context.Context, tableName string, entity *type
 	)
 	switch backend {
 	case types.BackendMySQL:
-		entityFormat = fmt.Sprintf("`%s` VARCHAR(255) PRIMARY KEY", entity.Name)
-	case types.BackendPostgres, types.BackendCassandra, types.BackendSQLite:
+		entityFormat = fmt.Sprintf("`%s` VARCHAR(%d) PRIMARY KEY", entity.Name, entity.Length)
+	case types.BackendPostgres:
+		entityFormat = fmt.Sprintf(`"%s" VARCHAR(%d) PRIMARY KEY`, entity.Name, entity.Length)
+	case types.BackendCassandra, types.BackendSQLite:
 		entityFormat = fmt.Sprintf(`"%s" TEXT PRIMARY KEY`, entity.Name)
 	default:
-		return "", errdefs.InvalidAttribute(fmt.Errorf("backend %s not support", backend))
+		return "", errdefs.InvalidAttribute(errors.Errorf("backend %s not support", backend))
 	}
 
 	schema := fmt.Sprintf("CREATE TABLE %s ( %s )", tableName, entityFormat)
@@ -45,7 +49,7 @@ func SqlxPrapareStreamTable(ctx context.Context, db *sqlx.DB, opt online.Prepare
 			return err
 		}
 		_, err = db.ExecContext(ctx, schema)
-		return err
+		return errors.WithStack(err)
 	}
 
 	dbValueType, err := dbutil.DBValueType(backend, opt.Feature.ValueType)
@@ -55,5 +59,5 @@ func SqlxPrapareStreamTable(ctx context.Context, db *sqlx.DB, opt online.Prepare
 
 	sql := fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", tableName, opt.Feature.Name, dbValueType)
 	_, err = db.ExecContext(ctx, sql)
-	return err
+	return errors.WithStack(err)
 }
diff --git a/internal/database/online/tikv/get.go b/internal/database/online/tikv/get.go
index 0c54b36d..d41152b9 100644
--- a/internal/database/online/tikv/get.go
+++ b/internal/database/online/tikv/get.go
@@ -3,6 +3,8 @@ package tikv
 import (
 	"context"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
@@ -84,7 +86,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	// Result order is the same as input order
 	batchGetResult, err := db.BatchGet(ctx, keys)
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 
 	// What we need to align with the result
diff --git a/internal/database/online/tikv/import.go b/internal/database/online/tikv/import.go
index 51179700..d26a7507 100644
--- a/internal/database/online/tikv/import.go
+++ b/internal/database/online/tikv/import.go
@@ -2,7 +2,8 @@ package tikv
 
 import (
 	"context"
-	"fmt"
+
+	"github.com/pkg/errors"
 
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
@@ -33,7 +34,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 
 	for record := range opt.ExportStream {
 		if len(record) != len(opt.Features)+1 {
-			return fmt.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
+			return errors.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
 		}
 
 		entityKey, featureValues := record[0], record[1:]
@@ -62,7 +63,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 		if seq%BatchSize == 0 {
 			// We don't expire keys using TTL
 			if err = db.BatchPut(ctx, putKeys, putVals, []uint64{}); err != nil {
-				return err
+				return errors.WithStack(err)
 			}
 			// Reset the slices
 			putKeys, putVals = nil, nil
@@ -72,7 +73,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	if seq%BatchSize != 0 {
 		// We don't expire keys using TTL
 		if err := db.BatchPut(ctx, putKeys, putVals, []uint64{}); err != nil {
-			return err
+			return errors.WithStack(err)
 		}
 	}
 
diff --git a/internal/database/online/tikv/purge.go b/internal/database/online/tikv/purge.go
index 43d9768b..5b849df6 100644
--- a/internal/database/online/tikv/purge.go
+++ b/internal/database/online/tikv/purge.go
@@ -3,6 +3,8 @@ package tikv
 import (
 	"context"
 
+	"github.com/pkg/errors"
+
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
 )
 
@@ -14,5 +16,6 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 	startKey := append([]byte(kvutil.KeyPrefixForBatchFeature+serializedRevisionID), byte(keyDelimiter))
 	endKey := append([]byte(kvutil.KeyPrefixForBatchFeature+serializedRevisionID), byte(keyDelimiter+1))
 
-	return db.DeleteRange(ctx, startKey, endKey)
+	err = db.DeleteRange(ctx, startKey, endKey)
+	return errors.WithStack(err)
 }
diff --git a/internal/database/online/tikv/push.go b/internal/database/online/tikv/push.go
index 31bb3622..d00b523b 100644
--- a/internal/database/online/tikv/push.go
+++ b/internal/database/online/tikv/push.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/internal/database/online/kvutil"
+	"github.com/pkg/errors"
 )
 
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
@@ -42,5 +43,6 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	}
 
 	// We don't expire keys using TTL
-	return db.BatchPut(ctx, putKeys, putVals, []uint64{})
+	err = db.BatchPut(ctx, putKeys, putVals, []uint64{})
+	return errors.WithStack(err)
 }
diff --git a/internal/database/online/tikv/store.go b/internal/database/online/tikv/store.go
index 56ac9030..27c5f70d 100644
--- a/internal/database/online/tikv/store.go
+++ b/internal/database/online/tikv/store.go
@@ -4,6 +4,7 @@ import (
 	"context"
 
 	"github.com/pingcap/log"
+	"github.com/pkg/errors"
 	"github.com/tikv/client-go/v2/config"
 	"github.com/tikv/client-go/v2/rawkv"
 	"go.uber.org/zap/zapcore"
@@ -28,7 +29,7 @@ type DB struct {
 func Open(opt *types.TiKVOpt) (*DB, error) {
 	db, err := rawkv.NewClient(context.Background(), opt.PdAddrs, config.DefaultConfig().Security)
 	if err != nil {
-		return nil, err
+		return nil, errors.WithStack(err)
 	}
 	return &DB{db}, nil
 }
@@ -39,7 +40,7 @@ func (db *DB) Close() error {
 
 func (db *DB) Ping(ctx context.Context) error {
 	_, err := db.Client.Get(ctx, []byte(""))
-	return err
+	return errors.WithStack(err)
 }
 
 func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
diff --git a/oomcli/cmd/register_entity.go b/oomcli/cmd/register_entity.go
index 24da9ba0..4d37e349 100644
--- a/oomcli/cmd/register_entity.go
+++ b/oomcli/cmd/register_entity.go
@@ -33,5 +33,8 @@ func init() {
 
 	flags := registerEntityCmd.Flags()
 
+	flags.IntVarP(&registerEntityOpt.Length, "length", "l", 0, "entity value length")
+	_ = registerEntityCmd.MarkFlagRequired("length")
+
 	flags.StringVarP(&registerEntityOpt.Description, "description", "d", "", "entity description")
 }
diff --git a/oomcli/cmd/types.go b/oomcli/cmd/types.go
index d46438e0..539fcdfa 100644
--- a/oomcli/cmd/types.go
+++ b/oomcli/cmd/types.go
@@ -10,6 +10,7 @@ import (
 type FlattenEntity struct {
 	ID          int    `oomcli:"ID"`
 	Name        string `oomcli:"NAME"`
+	Length      int    `oomcli:"LENGTH"`
 	Description string `oomcli:"DESCRIPTION,truncate"`
 
 	CreateTime time.Time `oomcli:"CREATE-TIME,wide"`
@@ -65,6 +66,7 @@ func parseTokenLists(i interface{}) (headerTokens TokenList, dataTokens []TokenL
 			tokens, err = parseTokens(FlattenEntity{
 				ID:          e.ID,
 				Name:        e.Name,
+				Length:      e.Length,
 				Description: e.Description,
 				CreateTime:  e.CreateTime,
 				ModifyTime:  e.ModifyTime,
diff --git a/pkg/oomstore/apply.go b/pkg/oomstore/apply.go
index 1be400bd..e819bc99 100644
--- a/pkg/oomstore/apply.go
+++ b/pkg/oomstore/apply.go
@@ -60,6 +60,7 @@ func (s *OomStore) applyEntity(ctx context.Context, tx metadata.DBStore, newEnti
 		_, err = tx.CreateEntity(ctx, metadata.CreateEntityOpt{
 			CreateEntityOpt: types.CreateEntityOpt{
 				EntityName:  newEntity.Name,
+				Length:      newEntity.Length,
 				Description: newEntity.Description,
 			},
 		})
@@ -304,6 +305,7 @@ func buildApplyEntity(entity apply.Entity) apply.Entity {
 	return apply.Entity{
 		Kind:        entity.Kind,
 		Name:        entity.Name,
+		Length:      entity.Length,
 		Description: entity.Description,
 	}
 }
diff --git a/pkg/oomstore/types/apply/apply.go b/pkg/oomstore/types/apply/apply.go
index 16637c34..f0f73bf4 100644
--- a/pkg/oomstore/types/apply/apply.go
+++ b/pkg/oomstore/types/apply/apply.go
@@ -136,6 +136,7 @@ func (g *Group) Validate() error {
 type Entity struct {
 	Kind        string `mapstructure:"kind" yaml:"kind"`
 	Name        string `mapstructure:"name" yaml:"name"`
+	Length      int    `mapstructure:"length" yaml:"length"`
 	Description string `mapstructure:"description" yaml:"description"`
 
 	Groups []Group `mapstructure:"groups" yaml:"groups,omitempty"`
@@ -157,6 +158,7 @@ func FromEntityList(entities types.EntityList, groups *GroupItems) (items Entity
 		items.Items = append(items.Items, Entity{
 			Kind:        "Entity",
 			Name:        entity.Name,
+			Length:      entity.Length,
 			Description: entity.Description,
 			Groups: groups.Filter(func(g Group) bool {
 				return g.EntityName == entity.Name
diff --git a/pkg/oomstore/types/entity.go b/pkg/oomstore/types/entity.go
index 81eada28..f893b455 100644
--- a/pkg/oomstore/types/entity.go
+++ b/pkg/oomstore/types/entity.go
@@ -5,8 +5,9 @@ import (
 )
 
 type Entity struct {
-	ID   int    `db:"id"`
-	Name string `db:"name"`
+	ID     int    `db:"id"`
+	Name   string `db:"name"`
+	Length int    `db:"length"`
 
 	Description string    `db:"description"`
 	CreateTime  time.Time `db:"create_time"`
diff --git a/pkg/oomstore/types/feature_values.go b/pkg/oomstore/types/feature_values.go
index 7fdcbe57..41364546 100644
--- a/pkg/oomstore/types/feature_values.go
+++ b/pkg/oomstore/types/feature_values.go
@@ -14,16 +14,3 @@ 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/options.go b/pkg/oomstore/types/options.go
index 732c8037..e05bc4ad 100644
--- a/pkg/oomstore/types/options.go
+++ b/pkg/oomstore/types/options.go
@@ -20,6 +20,7 @@ type UpdateFeatureOpt struct {
 
 type CreateEntityOpt struct {
 	EntityName  string
+	Length      int
 	Description string
 }
 
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/dbutil/schema_test.go b/internal/database/dbutil/schema_test.go
index 36bb6270..37846274 100644
--- a/internal/database/dbutil/schema_test.go
+++ b/internal/database/dbutil/schema_test.go
@@ -35,12 +35,15 @@ func TestBuildSchema(t *testing.T) {
 			description: "postgres schema",
 			backend:     types.BackendPostgres,
 			tableName:   "user",
-			entity:      &types.Entity{Name: "user_id"},
-			features:    features,
+			entity: &types.Entity{
+				Name:   "user_id",
+				Length: 32,
+			},
+			features: features,
 
 			want: `
 CREATE TABLE "user" (
-	"user_id" TEXT,
+	"user_id" VARCHAR(32),
 	"age" bigint,
 	"gender" text
 )`,
@@ -50,12 +53,15 @@ CREATE TABLE "user" (
 			description: "mysql schema",
 			backend:     types.BackendMySQL,
 			tableName:   "user",
-			entity:      &types.Entity{Name: "user_id"},
-			features:    features,
+			entity: &types.Entity{
+				Name:   "user_id",
+				Length: 32,
+			},
+			features: features,
 
 			want: "\n" +
 				"CREATE TABLE `user` (\n" +
-				"	`user_id` VARCHAR(255),\n" +
+				"	`user_id` VARCHAR(32),\n" +
 				"	`age` bigint,\n" +
 				"	`gender` text\n)",
 			wantErr: nil,
diff --git a/internal/database/metadata/informer/informer_test.go b/internal/database/metadata/informer/informer_test.go
index 1dc0b638..9c5b2a4e 100644
--- a/internal/database/metadata/informer/informer_test.go
+++ b/internal/database/metadata/informer/informer_test.go
@@ -14,8 +14,9 @@ import (
 
 func sampleCache() *informer.Cache {
 	entity := types.Entity{
-		ID:   1,
-		Name: "entity",
+		ID:     1,
+		Length: 10,
+		Name:   "entity",
 	}
 	group := types.Group{
 		ID:       100,
diff --git a/internal/database/metadata/test_impl/entity.go b/internal/database/metadata/test_impl/entity.go
index 092f30b7..8da7b038 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"
@@ -19,6 +20,7 @@ func TestCreateEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore De
 	opt := metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  "device",
+			Length:      32,
 			Description: "description",
 		},
 	}
@@ -28,10 +30,11 @@ func TestCreateEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore De
 	_, err = store.CreateEntity(ctx, metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  "device",
+			Length:      32,
 			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) {
@@ -43,6 +46,7 @@ func TestGetEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destr
 	opt := metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  "device",
+			Length:      32,
 			Description: "description",
 		},
 	}
@@ -53,6 +57,7 @@ func TestGetEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destr
 	entity, err := store.GetEntity(ctx, id)
 	require.NoError(t, err)
 	require.Equal(t, opt.EntityName, entity.Name)
+	require.Equal(t, opt.Length, entity.Length)
 	require.Equal(t, opt.Description, entity.Description)
 
 	_, err = store.GetEntity(ctx, 0)
@@ -68,6 +73,7 @@ func TestUpdateEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore De
 	id, err := store.CreateEntity(ctx, metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  "device",
+			Length:      32,
 			Description: "description",
 		},
 	})
@@ -103,6 +109,7 @@ func TestListEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore Dest
 	_, err = store.CreateEntity(ctx, metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  "device",
+			Length:      32,
 			Description: "description",
 		},
 	})
@@ -115,6 +122,7 @@ func TestListEntity(t *testing.T, prepareStore PrepareStoreFn, destroyStore Dest
 	_, err = store.CreateEntity(ctx, metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  "user",
+			Length:      16,
 			Description: "description",
 		},
 	})
diff --git a/internal/database/metadata/test_impl/feature.go b/internal/database/metadata/test_impl/feature.go
index 4c45f396..2a9b7c08 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"
 
@@ -16,6 +17,7 @@ func prepareEntityAndGroup(t *testing.T, ctx context.Context, store metadata.Sto
 	entityID, err := store.CreateEntity(ctx, metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  "device",
+			Length:      32,
 			Description: "description",
 		},
 	})
@@ -67,7 +69,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..493a820d 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"
@@ -14,6 +15,7 @@ func prepareEntity(t *testing.T, ctx context.Context, store metadata.Store, name
 	entityID, err := store.CreateEntity(ctx, metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  name,
+			Length:      32,
 			Description: "description",
 		},
 	})
@@ -145,7 +147,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/metadata/test_impl/revision.go b/internal/database/metadata/test_impl/revision.go
index 6f0568a8..fbea169f 100644
--- a/internal/database/metadata/test_impl/revision.go
+++ b/internal/database/metadata/test_impl/revision.go
@@ -357,6 +357,7 @@ func prepareRevisions(t *testing.T, ctx context.Context, store metadata.Store) (
 	entityID, err := store.CreateEntity(ctx, metadata.CreateEntityOpt{
 		CreateEntityOpt: types.CreateEntityOpt{
 			EntityName:  "device",
+			Length:      32,
 			Description: "description",
 		},
 	})
diff --git a/internal/database/offline/test_impl/create_table.go b/internal/database/offline/test_impl/create_table.go
index 0058a10c..33573366 100644
--- a/internal/database/offline/test_impl/create_table.go
+++ b/internal/database/offline/test_impl/create_table.go
@@ -15,7 +15,10 @@ func TestCreateTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 	defer store.Close()
 
 	tableName := "new_table"
-	entity := types.Entity{Name: "device"}
+	entity := types.Entity{
+		Name:   "device",
+		Length: 16,
+	}
 	features := []*types.Feature{
 		{
 			Name:      "price",
diff --git a/internal/database/offline/test_impl/import.go b/internal/database/offline/test_impl/import.go
index 3d613e29..87a9fc8f 100644
--- a/internal/database/offline/test_impl/import.go
+++ b/internal/database/offline/test_impl/import.go
@@ -19,7 +19,10 @@ func TestImport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 	ctx, store := prepareStore(t)
 	defer store.Close()
 
-	entity := types.Entity{Name: "device"}
+	entity := types.Entity{
+		Name:   "device",
+		Length: 16,
+	}
 	snapshotTable := "offline_1_1"
 
 	opt := offline.ImportOpt{
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index fb550314..9282cfd2 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -23,7 +23,10 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 	ctx, store := prepareStore(t)
 	defer store.Close()
 
-	entity := &types.Entity{Name: "device"}
+	entity := &types.Entity{
+		Name:   "device",
+		Length: 10,
+	}
 	unixMilli := &types.Feature{
 		Name:      "unix_milli",
 		ValueType: types.Int64,
diff --git a/internal/database/offline/test_impl/push.go b/internal/database/offline/test_impl/push.go
index d1c7dcf3..7063cb9e 100644
--- a/internal/database/offline/test_impl/push.go
+++ b/internal/database/offline/test_impl/push.go
@@ -17,8 +17,9 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 	defer store.Close()
 
 	entity := types.Entity{
-		ID:   18,
-		Name: "user",
+		ID:     18,
+		Name:   "user",
+		Length: 10,
 	}
 	group := types.Group{
 		ID:       1,
@@ -41,27 +42,17 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		},
 	}
 	revision := int64(2)
-	records := []types.StreamRecord{
-		{
 
-			EntityKey: "1000",
-			UnixMilli: 8000,
-			Values:    []interface{}{18, "F"},
-		},
-		{
-
-			EntityKey: "1001",
-			UnixMilli: 8001,
-			Values:    []interface{}{21, "M"},
-		},
+	featureValues := []interface{}{
+		[]interface{}{18, "F"},
+		[]interface{}{21, "M"},
 	}
 
 	pushOpt := offline.PushOpt{
-		GroupID:      group.ID,
-		Revision:     revision,
-		EntityName:   entity.Name,
-		FeatureNames: features.Names(),
-		Records:      records,
+		GroupID:       group.ID,
+		Revision:      revision,
+		FeatureNames:  features.Names(),
+		FeatureValues: featureValues,
 	}
 
 	t.Run("push when cdc table not exists", func(t *testing.T) {
diff --git a/internal/database/offline/test_impl/snapshot.go b/internal/database/offline/test_impl/snapshot.go
index 37701b37..21299a94 100644
--- a/internal/database/offline/test_impl/snapshot.go
+++ b/internal/database/offline/test_impl/snapshot.go
@@ -21,8 +21,11 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 	defer store.Close()
 
 	group := &types.Group{
-		ID:     1,
-		Entity: &types.Entity{Name: "device"},
+		ID: 1,
+		Entity: &types.Entity{
+			Name:   "device",
+			Length: 10,
+		},
 	}
 	unixMilli := &types.Feature{
 		Name:      "unix_milli",
diff --git a/internal/database/offline/test_impl/util.go b/internal/database/offline/test_impl/util.go
index 47396ed8..8a18f29f 100644
--- a/internal/database/offline/test_impl/util.go
+++ b/internal/database/offline/test_impl/util.go
@@ -22,7 +22,10 @@ func buildTestSnapshotTable(
 	snapshotTable string,
 	source *offline.CSVSource,
 ) {
-	entity := &types.Entity{Name: "device"}
+	entity := &types.Entity{
+		Name:   "device",
+		Length: 10,
+	}
 	header := []string{"device"}
 	for _, f := range features {
 		header = append(header, f.Name)
diff --git a/internal/database/online/test_impl/stream.go b/internal/database/online/test_impl/stream.go
index 4850deba..7e1c99c2 100644
--- a/internal/database/online/test_impl/stream.go
+++ b/internal/database/online/test_impl/stream.go
@@ -19,8 +19,9 @@ var simpleStreamData StreamData
 
 func init() {
 	entity := types.Entity{
-		ID:   1,
-		Name: "user",
+		ID:     1,
+		Name:   "user",
+		Length: 16,
 	}
 
 	group1 := &types.Group{
diff --git a/internal/database/online/test_impl/util.go b/internal/database/online/test_impl/util.go
index fb3b9840..466e38ad 100644
--- a/internal/database/online/test_impl/util.go
+++ b/internal/database/online/test_impl/util.go
@@ -83,7 +83,7 @@ func init() {
 					Category: types.CategoryBatch,
 				},
 			},
-			Entity: &types.Entity{ID: 5, Name: "user"},
+			Entity: &types.Entity{ID: 5, Name: "user", Length: 4},
 			Data: []types.ExportRecord{
 				[]interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
 				[]interface{}{"3216", int64(29), nil, 2.0, false, time.Now()},
@@ -106,11 +106,11 @@ func init() {
 		}
 
 		revision := &types.Revision{ID: 9, GroupID: 2, Group: &types.Group{ID: 2, Category: types.CategoryBatch}}
-		entity := &types.Entity{ID: 5, Name: "user"}
+		entity := &types.Entity{ID: 5, Name: "user", Length: 5}
 		var data []types.ExportRecord
 
 		for i := 0; i < 100; i++ {
-			record := []interface{}{dbutil.RandString(10), rand.Float64()}
+			record := []interface{}{dbutil.RandString(entity.Length), rand.Float64()}
 			data = append(data, record)
 		}
 		SampleMedium = Sample{features, revision, entity, data}
diff --git a/oomagent/test/data/driver_stats.yaml b/oomagent/test/data/driver_stats.yaml
index 48cfa4a8..3eb8401e 100644
--- a/oomagent/test/data/driver_stats.yaml
+++ b/oomagent/test/data/driver_stats.yaml
@@ -1,6 +1,7 @@
 ---
 kind: Entity
 name: driver
+length: 2
 groups:
   - name: driver_stats
     category: batch
diff --git a/oomagent/test/data/fraud_detection.yaml b/oomagent/test/data/fraud_detection.yaml
index 1244259f..7edcf39b 100644
--- a/oomagent/test/data/fraud_detection.yaml
+++ b/oomagent/test/data/fraud_detection.yaml
@@ -1,5 +1,6 @@
 kind: Entity
 name: user
+length: 12
 description: 'user id'
 groups:
 - name: account
diff --git a/oomcli/test/test_apply_entity.sh b/oomcli/test/test_apply_entity.sh
index 61bb744c..cfe252fb 100755
--- a/oomcli/test/test_apply_entity.sh
+++ b/oomcli/test/test_apply_entity.sh
@@ -8,6 +8,7 @@ apply_single_complex_entity() {
     cat <<EOF | oomcli apply -f /dev/stdin
 kind: Entity
 name: user
+length: 8
 description: 'description'
 groups:
 - name: device
@@ -33,8 +34,8 @@ groups:
 EOF
 
     entity_expected='
-ID,NAME,DESCRIPTION
-1,user,description
+ID,NAME,LENGTH,DESCRIPTION
+1,user,8,description
 '
     entity_actual=$(oomcli get meta entity -o csv)
     assert_eq "apply_single_complex_entity: check entity" "$(sort <<< "$entity_expected")" "$(sort <<< "$entity_actual")"
@@ -65,6 +66,7 @@ apply_multiple_files_of_entity() {
     cat <<EOF | oomcli apply -f /dev/stdin
 kind: Entity
 name: user
+length: 8
 description: 'description'
 groups:
 - name: student
@@ -73,19 +75,21 @@ groups:
 ---
 kind: Entity
 name: device
+length: 16
 description: 'description'
 ---
 kind: Entity
 name: test
+length: 32
 description: 'description'
 EOF
 
 
   entity_expected='
-ID,NAME,DESCRIPTION
-1,user,description
-2,device,description
-3,test,description
+ID,NAME,LENGTH,DESCRIPTION
+1,user,8,description
+2,device,16,description
+3,test,32,description
 '
     entity_actual=$(oomcli get meta entity -o csv)
     assert_eq "apply_multiple_files_of_entity: oomcli get meta entity" "$entity_expected" "$entity_actual"
@@ -106,6 +110,7 @@ apply_entity_items() {
 items:
   - kind: Entity
     name: user
+    length: 8
     description: user ID
     groups:
       - name: account
@@ -133,6 +138,7 @@ items:
             description: transaction_count_30d description
   - kind: Entity
     name: device
+    length: 8
     description: device info
     groups:
       - name: phone
@@ -148,9 +154,9 @@ items:
 EOF
 
       entity_expected='
-ID,NAME,DESCRIPTION
-1,user,user ID
-2,device,device info
+ID,NAME,LENGTH,DESCRIPTION
+1,user,8,user ID
+2,device,8,device info
 '
       entity_actual=$(oomcli get meta entity -o csv)
       assert_eq "apply_entity_items: oomcli apply mutiple entity: check entity" "$(sort <<< "$entity_expected")" "$(sort <<< "$entity_actual")"
diff --git a/oomcli/test/test_apply_feature.sh b/oomcli/test/test_apply_feature.sh
index de5f2494..bdf15553 100755
--- a/oomcli/test/test_apply_feature.sh
+++ b/oomcli/test/test_apply_feature.sh
@@ -10,6 +10,7 @@ apply_multiple_files_of_feature() {
     cat <<EOF | oomcli apply -f /dev/stdin
 kind: Entity
 name: user
+length: 8
 description: 'description'
 ---
 kind: Group
@@ -48,6 +49,7 @@ apply_feature_items() {
     cat <<EOF | oomcli apply -f /dev/stdin
 kind: Entity
 name: user
+length: 8
 description: user ID
 ---
 kind: Group
diff --git a/oomcli/test/test_apply_group.sh b/oomcli/test/test_apply_group.sh
index ca59442f..d73b40f4 100755
--- a/oomcli/test/test_apply_group.sh
+++ b/oomcli/test/test_apply_group.sh
@@ -10,6 +10,7 @@ apply_single_complex_group() {
     cat <<EOF | oomcli apply -f /dev/stdin
 kind: Entity
 name: user
+length: 8
 description: 'description'
 ---
 kind: Group
@@ -53,6 +54,7 @@ apply_multiple_files_of_group() {
     cat <<EOF | oomcli apply -f /dev/stdin
 kind: Entity
 name: user
+length: 8
 description: 'description'
 ---
 kind: Group
@@ -84,10 +86,12 @@ apply_group_items() {
     cat <<EOF | oomcli apply -f /dev/stdin
 kind: Entity
 name: user
+length: 8
 description: 'description'
 ---
 kind: Entity
 name: device
+length: 8
 description: 'description'
 ---
 items:
diff --git a/oomcli/test/test_get_meta_entity.sh b/oomcli/test/test_get_meta_entity.sh
index f4905fa2..ffac5278 100755
--- a/oomcli/test/test_get_meta_entity.sh
+++ b/oomcli/test/test_get_meta_entity.sh
@@ -7,19 +7,19 @@ register_features
 import_sample > /dev/null
 
 case='oomcli get meta entity works'
-expected='ID,NAME,DESCRIPTION,CREATE-TIME,MODIFY-TIME
-1,device,device,2021-10-19T06:56:07Z,2021-10-19T06:56:07Z
-2,user,user,2021-10-19T06:56:07Z,2021-10-19T06:56:07Z
+expected='ID,NAME,LENGTH,DESCRIPTION,CREATE-TIME,MODIFY-TIME
+1,device,32,device,2021-10-19T06:56:07Z,2021-10-19T06:56:07Z
+2,user,64,user,2021-10-19T06:56:07Z,2021-10-19T06:56:07Z
 '
 actual=$(oomcli get meta entity -o csv)
-ignore_time() { cut -d ',' -f 1-3 <<<"$1"; }
+ignore_time() { cut -d ',' -f 1-4 <<<"$1"; }
 assert_eq "$case" "$(ignore_time "$expected" | sort)" "$(ignore_time "$actual" | sort)"
 
 
 case='oomcli get simplified meta entity works'
-expected='ID,NAME,DESCRIPTION
-1,device,device
-2,user,user
+expected='ID,NAME,LENGTH,DESCRIPTION
+1,device,32,device
+2,user,64,user
 '
 actual=$(oomcli get meta entity -o csv)
 assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
@@ -28,6 +28,7 @@ case='oomcli get meta entity -o yaml: one entity'
 expected='
 kind: Entity
 name: device
+length: 32
 description: device
 groups:
     - name: phone
@@ -50,6 +51,7 @@ expected='
 items:
     - kind: Entity
       name: device
+      length: 32
       description: device
       groups:
         - name: phone
@@ -64,6 +66,7 @@ items:
               description: model
     - kind: Entity
       name: user
+      length: 64
       description: user
       groups:
         - name: student
diff --git a/oomcli/test/test_update_entity.sh b/oomcli/test/test_update_entity.sh
index e4ef0699..480f9918 100755
--- a/oomcli/test/test_update_entity.sh
+++ b/oomcli/test/test_update_entity.sh
@@ -9,8 +9,8 @@ import_sample > /dev/null
 case='oomcli update entity works'
 oomcli update entity device --description "new description"
 expected='
-ID,NAME,DESCRIPTION
-1,device,new description
+ID,NAME,LENGTH,DESCRIPTION
+1,device,32,new description
 '
 actual=$(oomcli get meta entity device -o csv)
 assert_eq "$case" "$expected" "$actual"
diff --git a/oomcli/test/util.sh b/oomcli/test/util.sh
index 059a4888..4c2f4b73 100644
--- a/oomcli/test/util.sh
+++ b/oomcli/test/util.sh
@@ -52,8 +52,8 @@ assert_eq() {
 
 # register features for the sample data
 register_features() {
-    oomcli register entity device --description "device"
-    oomcli register entity user   --description "user"
+    oomcli register entity device --length 32     --description "device"
+    oomcli register entity user   --length 64     --description "user"
 
     oomcli register group phone      --entity device --category "batch"  --description "phone"
     oomcli register group student    --entity user   --category "batch"  --description "student"
diff --git a/pkg/oomstore/apply_test.go b/pkg/oomstore/apply_test.go
index 745fe01f..ca696f21 100644
--- a/pkg/oomstore/apply_test.go
+++ b/pkg/oomstore/apply_test.go
@@ -24,6 +24,7 @@ func TestBuildApplyStage(t *testing.T) {
 			opt: apply.ApplyOpt{R: strings.NewReader(`
 # kind: Entity
 name: user
+length: 8
 description: 'User ID'
 `)},
 			wantStage: nil,
@@ -34,6 +35,7 @@ description: 'User ID'
 			opt: apply.ApplyOpt{R: strings.NewReader(`
 kind: Entit
 name: user
+length: 8
 description: 'description'
 `)},
 			wantStage: nil,
@@ -44,6 +46,7 @@ description: 'description'
 			opt: apply.ApplyOpt{R: strings.NewReader(`
 kind: Entity
 name: user
+length: 8
 description: 'description'
 `)},
 			wantStage: &apply.ApplyStage{
@@ -51,6 +54,7 @@ description: 'description'
 					{
 						Kind:        "Entity",
 						Name:        "user",
+						Length:      8,
 						Description: "description",
 					},
 				},
@@ -64,6 +68,7 @@ description: 'description'
 			opt: apply.ApplyOpt{R: strings.NewReader(`
 kind: Entity
 name: user
+length: 8
 description: 'description'
 ---
 kind: Group
@@ -97,6 +102,7 @@ description: 'description'
 					{
 						Kind:        "Entity",
 						Name:        "user",
+						Length:      8,
 						Description: "description",
 					},
 				}, NewGroups: []apply.Group{
@@ -186,6 +192,7 @@ features:
 			opt: apply.ApplyOpt{R: strings.NewReader(`
 kind: Entity
 name: user
+length: 8
 description: 'description'
 groups:
 - name: device
@@ -214,6 +221,7 @@ groups:
 					{
 						Kind:        "Entity",
 						Name:        "user",
+						Length:      8,
 						Description: "description",
 					},
 				},
@@ -439,6 +447,7 @@ items:
 items:
     - kind: Entity
       name: user
+      length: 8
       description: user ID
       groups:
         - name: account
@@ -466,6 +475,7 @@ items:
               description: transaction_count_30d description
     - kind: Entity
       name: device
+      length: 8
       description: device info
       groups:
         - name: phone
@@ -485,11 +495,13 @@ items:
 					{
 						Kind:        "Entity",
 						Name:        "user",
+						Length:      8,
 						Description: "user ID",
 					},
 					{
 						Kind:        "Entity",
 						Name:        "device",
+						Length:      8,
 						Description: "device info",
 					},
 				},
diff --git a/pkg/oomstore/join_test.go b/pkg/oomstore/join_test.go
index 5f505f29..51b6b2bb 100644
--- a/pkg/oomstore/join_test.go
+++ b/pkg/oomstore/join_test.go
@@ -26,7 +26,8 @@ func TestChannelJoin(t *testing.T) {
 	consistentFeatures := prepareFeatures(true, true)
 
 	entity := types.Entity{
-		Name: "device",
+		Name:   "device",
+		Length: 10,
 	}
 	revisions := types.RevisionList{
 		{
diff --git a/sdk/python/src/test/test_client.py b/sdk/python/src/test/test_client.py
index 24ec65ab..1e3d659f 100644
--- a/sdk/python/src/test/test_client.py
+++ b/sdk/python/src/test/test_client.py
@@ -25,6 +25,7 @@ class Data:
             f.write(
                 """kind: Entity
 name: user
+length: 8
 description: "user ID"
 groups:
   - name: account