Repository: oom-ai/oomstore
Pull request #1083 (2022-01-25 10:30:53)

Embedded oomagent


Patch:
diff --git a/go.mod b/go.mod
index 00079c27..b0f4def9 100644
--- a/go.mod
+++ b/go.mod
@@ -20,6 +20,7 @@ require (
 	github.com/lib/pq v1.10.4
 	github.com/mitchellh/mapstructure v1.4.3
 	github.com/olekukonko/tablewriter v0.0.5
+	github.com/pingcap/errors v0.11.5-0.20201126102027-b0a155152ca3
 	github.com/pingcap/log v0.0.0-20211215031037-e024ba4eb0ee
 	github.com/pkg/errors v0.9.1
 	github.com/snowflakedb/gosnowflake v1.6.5
@@ -29,6 +30,7 @@ require (
 	github.com/tikv/client-go/v2 v2.0.0-rc
 	go.uber.org/zap v1.19.1
 	google.golang.org/api v0.63.0
+	google.golang.org/genproto v0.0.0-20211223182754-3ac035c7e7cb
 	google.golang.org/grpc v1.43.0
 	google.golang.org/protobuf v1.27.1
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
@@ -91,7 +93,6 @@ require (
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/opentracing/opentracing-go v1.2.0 // indirect
 	github.com/pierrec/lz4/v4 v4.1.12 // indirect
-	github.com/pingcap/errors v0.11.5-0.20201126102027-b0a155152ca3 // indirect
 	github.com/pingcap/failpoint v0.0.0-20210918120811-547c13e3eb00 // indirect
 	github.com/pingcap/kvproto v0.0.0-20211229082925-7a8280c36daf // indirect
 	github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8 // indirect
@@ -119,7 +120,6 @@ require (
 	golang.org/x/tools v0.1.5 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20211223182754-3ac035c7e7cb // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
 	lukechampine.com/uint128 v1.1.1 // indirect
diff --git a/internal/database/dbutil/data_loader.go b/internal/database/dbutil/data_loader.go
index 7f27b67c..62f752e0 100644
--- a/internal/database/dbutil/data_loader.go
+++ b/internal/database/dbutil/data_loader.go
@@ -13,12 +13,12 @@ import (
 )
 
 type LoadDataFromSourceOpt struct {
-	Source     *offline.CSVSource
-	EntityName string
-	TableName  string
-	Header     []string
-	Features   types.FeatureList
-	Backend    types.BackendType
+	Source    *offline.CSVSource
+	Entity    *types.Entity
+	TableName string
+	Header    []string
+	Features  types.FeatureList
+	Backend   types.BackendType
 }
 
 // Currying
@@ -32,10 +32,10 @@ func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, opt LoadDataFromSource
 	records := make([]interface{}, 0, batchSize)
 	for {
 		record, err := ReadLine(ReadLineOpt{
-			Source:     opt.Source,
-			EntityName: opt.EntityName,
-			Header:     opt.Header,
-			Features:   opt.Features,
+			Source:   opt.Source,
+			Entity:   opt.Entity,
+			Header:   opt.Header,
+			Features: opt.Features,
 		})
 		if errdefs.Cause(err) == io.EOF {
 			break
@@ -61,10 +61,10 @@ func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, opt LoadDataFromSource
 }
 
 type ReadLineOpt struct {
-	Source     *offline.CSVSource
-	EntityName string
-	Header     []string
-	Features   types.FeatureList
+	Source   *offline.CSVSource
+	Entity   *types.Entity
+	Header   []string
+	Features types.FeatureList
 }
 
 func ReadLine(opt ReadLineOpt) ([]interface{}, error) {
@@ -75,7 +75,7 @@ func ReadLine(opt ReadLineOpt) ([]interface{}, error) {
 	rowSlice := strings.Split(strings.Trim(row, "\n"), opt.Source.Delimiter)
 	line := make([]interface{}, 0, len(rowSlice))
 	for i, ele := range rowSlice {
-		if len(opt.Header) == 0 || len(opt.Features) == 0 || opt.Header[i] == opt.EntityName {
+		if len(opt.Header) == 0 || len(opt.Features) == 0 || opt.Header[i] == opt.Entity.Name {
 			// entity_key doesn't need to change type
 			line = append(line, ele)
 		} else if opt.Header[i] == "unix_milli" {
diff --git a/internal/database/dbutil/db_opt.go b/internal/database/dbutil/db_opt.go
index 5fee0cb1..dbc486e5 100644
--- a/internal/database/dbutil/db_opt.go
+++ b/internal/database/dbutil/db_opt.go
@@ -5,8 +5,6 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/gocql/gocql"
-
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/spf13/cast"
 
@@ -24,9 +22,6 @@ type DBOpt struct {
 	// BigQuery
 	BigQueryDB *bigquery.Client
 	DatasetID  *string
-
-	// Cassandra
-	CassandraDB *gocql.Session
 }
 
 func (d *DBOpt) ExecContext(ctx context.Context, query string, args []interface{}) error {
@@ -37,9 +32,6 @@ func (d *DBOpt) ExecContext(ctx context.Context, query string, args []interface{
 		}
 		_, err := d.BigQueryDB.Query(query).Read(ctx)
 		return errdefs.WithStack(err)
-	case types.BackendCassandra:
-		err := d.CassandraDB.Query(query).Exec()
-		return errdefs.WithStack(err)
 	default:
 		_, err := d.SqlxDB.ExecContext(ctx, d.SqlxDB.Rebind(query), args...)
 		return errdefs.WithStack(err)
diff --git a/internal/database/dbutil/schema.go b/internal/database/dbutil/schema.go
index 8ff1bebf..d51c747e 100644
--- a/internal/database/dbutil/schema.go
+++ b/internal/database/dbutil/schema.go
@@ -25,13 +25,13 @@ func (c ColumnList) Names() []string {
 
 func BuildTableSchema(
 	tableName string,
-	entityName string,
+	entity *types.Entity,
 	hasUnixMilli bool,
 	features types.FeatureList,
 	pkFields []string,
 	backend types.BackendType,
 ) string {
-	columns := parseColumns(entityName, hasUnixMilli, features, backend)
+	columns := parseColumns(entity, hasUnixMilli, features, backend)
 	return createTableDDL(tableName, columns, pkFields, backend)
 }
 
@@ -73,10 +73,10 @@ func BuildIndexDDL(tableName string, indexName string, fields []string, backend
 	return fmt.Sprintf("CREATE INDEX %s ON %s (%s)", qt(index), qt(tableName), qt(fields...))
 }
 
-func parseColumns(entityName string, hasUnixMilli bool, features types.FeatureList, backend types.BackendType) (rs []Column) {
+func parseColumns(entity *types.Entity, hasUnixMilli bool, features types.FeatureList, backend types.BackendType) (rs []Column) {
 	// entity column
 	{
-		c := Column{Name: entityName, ValueType: types.String}
+		c := Column{Name: entity.Name, ValueType: types.String}
 		switch backend {
 		case types.BackendCassandra, types.BackendSQLite, types.BackendPostgres, types.BackendRedshift, types.BackendSnowflake:
 			c.DbType = "TEXT"
diff --git a/internal/database/dbutil/sql.go b/internal/database/dbutil/sql.go
index 522f9f39..02bd51b4 100644
--- a/internal/database/dbutil/sql.go
+++ b/internal/database/dbutil/sql.go
@@ -89,9 +89,3 @@ func QuoteFn(backendType types.BackendType) func(...string) string {
 		return strings.Join(rs, ",")
 	}
 }
-
-func DropTable(ctx context.Context, dbOpt DBOpt, tableName string) error {
-	query := fmt.Sprintf(`DROP TABLE IF EXISTS %s;`, tableName)
-	err := dbOpt.ExecContext(ctx, query, nil)
-	return errdefs.WithStack(err)
-}
diff --git a/internal/database/dbutil/util.go b/internal/database/dbutil/util.go
index ea97c2ee..d6b0a236 100644
--- a/internal/database/dbutil/util.go
+++ b/internal/database/dbutil/util.go
@@ -50,11 +50,3 @@ func OfflineStreamSnapshotTableName(groupID int, revision int64) string {
 func OfflineStreamCdcTableName(groupID int, revision int64) string {
 	return fmt.Sprintf("offline_stream_cdc_%d_%d", groupID, revision)
 }
-
-func OnlineBatchTableName(revisionID int) string {
-	return fmt.Sprintf("online_batch_%d", revisionID)
-}
-
-func OnlineStreamTableName(groupID int) string {
-	return fmt.Sprintf("online_stream_%d", groupID)
-}
diff --git a/internal/database/metadata/informer/feature.go b/internal/database/metadata/informer/feature.go
index bdcde282..20c5f15a 100644
--- a/internal/database/metadata/informer/feature.go
+++ b/internal/database/metadata/informer/feature.go
@@ -1,7 +1,6 @@
 package informer
 
 import (
-	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
@@ -17,13 +16,13 @@ func (c *FeatureCache) Enrich(groupCache *GroupCache) {
 	}
 }
 
-func (c *FeatureCache) List(opt metadata.ListCachedFeatureOpt) types.FeatureList {
+func (c *FeatureCache) List(fullNames *[]string) types.FeatureList {
 	features := c.FeatureList
 
-	// filter featureNames
-	if opt.FullNames != nil {
+	// filter names
+	if fullNames != nil {
 		var tmp types.FeatureList
-		for _, fullName := range *opt.FullNames {
+		for _, fullName := range *fullNames {
 			if f := features.Find(func(f *types.Feature) bool {
 				return f.FullName() == fullName
 			}); f != nil {
@@ -32,17 +31,6 @@ func (c *FeatureCache) List(opt metadata.ListCachedFeatureOpt) types.FeatureList
 		}
 		features = tmp
 	}
-	// filter groupName
-	if opt.GroupName != nil {
-		features = features.Filter(func(f *types.Feature) bool {
-			return f.Group.Name == *opt.GroupName
-		})
-	}
-	// filter groupID
-	if opt.GroupID != nil {
-		features = features.Filter(func(f *types.Feature) bool {
-			return f.Group.ID == *opt.GroupID
-		})
-	}
+
 	return features
 }
diff --git a/internal/database/metadata/informer/group.go b/internal/database/metadata/informer/group.go
index f373810b..b97313af 100644
--- a/internal/database/metadata/informer/group.go
+++ b/internal/database/metadata/informer/group.go
@@ -24,13 +24,3 @@ func (c *GroupCache) List(entityID *int) types.GroupList {
 		return g.Entity.ID == *entityID
 	})
 }
-
-func (c *GroupCache) Get(groupID int) *types.Group {
-	groups := c.GroupList.Filter(func(g *types.Group) bool {
-		return g.ID == groupID
-	})
-	if len(groups) == 0 {
-		return nil
-	}
-	return groups[0]
-}
diff --git a/internal/database/metadata/informer/informer.go b/internal/database/metadata/informer/informer.go
index 826ad78f..fbf81c96 100644
--- a/internal/database/metadata/informer/informer.go
+++ b/internal/database/metadata/informer/informer.go
@@ -6,8 +6,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/oom-ai/oomstore/pkg/errdefs"
-
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -90,14 +88,6 @@ func (f *Informer) Cache() *Cache {
 	return f.cache.Load().(*Cache)
 }
 
-func (f *Informer) ListCachedFeature(ctx context.Context, opt metadata.ListCachedFeatureOpt) types.FeatureList {
-	return f.Cache().Features.List(opt).Copy()
-}
-
-func (f *Informer) GetCachedGroup(ctx context.Context, groupID int) (*types.Group, error) {
-	group := f.Cache().Groups.Get(groupID).Copy()
-	if group == nil {
-		return nil, errdefs.NotFound(errdefs.Errorf("group %d not found", groupID))
-	}
-	return group, nil
+func (f *Informer) ListCachedFeature(ctx context.Context, featureNames *[]string) types.FeatureList {
+	return f.Cache().Features.List(featureNames).Copy()
 }
diff --git a/internal/database/metadata/mock_metadata/store.go b/internal/database/metadata/mock_metadata/store.go
index 4f9f81ff..4179e367 100644
--- a/internal/database/metadata/mock_metadata/store.go
+++ b/internal/database/metadata/mock_metadata/store.go
@@ -112,21 +112,6 @@ func (mr *MockStoreMockRecorder) CreateRevision(ctx, opt interface{}) *gomock.Ca
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateRevision", reflect.TypeOf((*MockStore)(nil).CreateRevision), ctx, opt)
 }
 
-// GetCachedGroup mocks base method.
-func (m *MockStore) GetCachedGroup(ctx context.Context, id int) (*types.Group, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetCachedGroup", ctx, id)
-	ret0, _ := ret[0].(*types.Group)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// GetCachedGroup indicates an expected call of GetCachedGroup.
-func (mr *MockStoreMockRecorder) GetCachedGroup(ctx, id interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCachedGroup", reflect.TypeOf((*MockStore)(nil).GetCachedGroup), ctx, id)
-}
-
 // GetEntity mocks base method.
 func (m *MockStore) GetEntity(ctx context.Context, id int) (*types.Entity, error) {
 	m.ctrl.T.Helper()
@@ -248,17 +233,17 @@ func (mr *MockStoreMockRecorder) GetRevisionBy(ctx, groupID, revision interface{
 }
 
 // ListCachedFeature mocks base method.
-func (m *MockStore) ListCachedFeature(ctx context.Context, opt metadata.ListCachedFeatureOpt) types.FeatureList {
+func (m *MockStore) ListCachedFeature(ctx context.Context, fullNames *[]string) types.FeatureList {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "ListCachedFeature", ctx, opt)
+	ret := m.ctrl.Call(m, "ListCachedFeature", ctx, fullNames)
 	ret0, _ := ret[0].(types.FeatureList)
 	return ret0
 }
 
 // ListCachedFeature indicates an expected call of ListCachedFeature.
-func (mr *MockStoreMockRecorder) ListCachedFeature(ctx, opt interface{}) *gomock.Call {
+func (mr *MockStoreMockRecorder) ListCachedFeature(ctx, fullNames interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListCachedFeature", reflect.TypeOf((*MockStore)(nil).ListCachedFeature), ctx, opt)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListCachedFeature", reflect.TypeOf((*MockStore)(nil).ListCachedFeature), ctx, fullNames)
 }
 
 // ListEntity mocks base method.
@@ -776,33 +761,18 @@ func (m *MockCacheStore) EXPECT() *MockCacheStoreMockRecorder {
 	return m.recorder
 }
 
-// GetCachedGroup mocks base method.
-func (m *MockCacheStore) GetCachedGroup(ctx context.Context, id int) (*types.Group, error) {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "GetCachedGroup", ctx, id)
-	ret0, _ := ret[0].(*types.Group)
-	ret1, _ := ret[1].(error)
-	return ret0, ret1
-}
-
-// GetCachedGroup indicates an expected call of GetCachedGroup.
-func (mr *MockCacheStoreMockRecorder) GetCachedGroup(ctx, id interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetCachedGroup", reflect.TypeOf((*MockCacheStore)(nil).GetCachedGroup), ctx, id)
-}
-
 // ListCachedFeature mocks base method.
-func (m *MockCacheStore) ListCachedFeature(ctx context.Context, opt metadata.ListCachedFeatureOpt) types.FeatureList {
+func (m *MockCacheStore) ListCachedFeature(ctx context.Context, fullNames *[]string) types.FeatureList {
 	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "ListCachedFeature", ctx, opt)
+	ret := m.ctrl.Call(m, "ListCachedFeature", ctx, fullNames)
 	ret0, _ := ret[0].(types.FeatureList)
 	return ret0
 }
 
 // ListCachedFeature indicates an expected call of ListCachedFeature.
-func (mr *MockCacheStoreMockRecorder) ListCachedFeature(ctx, opt interface{}) *gomock.Call {
+func (mr *MockCacheStoreMockRecorder) ListCachedFeature(ctx, fullNames interface{}) *gomock.Call {
 	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListCachedFeature", reflect.TypeOf((*MockCacheStore)(nil).ListCachedFeature), ctx, opt)
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListCachedFeature", reflect.TypeOf((*MockCacheStore)(nil).ListCachedFeature), ctx, fullNames)
 }
 
 // Refresh mocks base method.
diff --git a/internal/database/metadata/store.go b/internal/database/metadata/store.go
index d6c73c06..b22badb8 100644
--- a/internal/database/metadata/store.go
+++ b/internal/database/metadata/store.go
@@ -16,7 +16,7 @@ type Store interface {
 	io.Closer
 }
 
-// DBStore defines the methods that a database backend store must implement.
+// The interface defines the methods that a database backend store must implement.
 type DBStore interface {
 	// entity
 	CreateEntity(ctx context.Context, opt CreateEntityOpt) (int, error)
@@ -50,12 +50,11 @@ type DBStore interface {
 	WithTransaction(ctx context.Context, fn func(context.Context, DBStore) error) error
 }
 
-// CacheStore defines methods that a memory backend store must implement.
+// The interface defines methods that a memory backend store must implement.
 type CacheStore interface {
-	ListCachedFeature(ctx context.Context, opt ListCachedFeatureOpt) types.FeatureList
-	GetCachedGroup(ctx context.Context, id int) (*types.Group, error)
+	ListCachedFeature(ctx context.Context, fullNames *[]string) types.FeatureList
 
-	// Refresh pulls data from database and update cache.
+	// Pull data from database and update cache.
 	Refresh() error
 }
 
diff --git a/internal/database/metadata/types.go b/internal/database/metadata/types.go
index c9ab48e7..9bcc0478 100644
--- a/internal/database/metadata/types.go
+++ b/internal/database/metadata/types.go
@@ -65,9 +65,3 @@ type ListFeatureOpt struct {
 	GroupID    *int
 	FeatureIDs *[]int
 }
-
-type ListCachedFeatureOpt struct {
-	FullNames *[]string
-	GroupName *string
-	GroupID   *int
-}
diff --git a/internal/database/offline/bigquery/import.go b/internal/database/offline/bigquery/import.go
index 69324c1e..94847bf7 100644
--- a/internal/database/offline/bigquery/import.go
+++ b/internal/database/offline/bigquery/import.go
@@ -17,9 +17,9 @@ func (db *DB) Import(ctx context.Context, opt offline.ImportOpt) (int64, error)
 	// Step 1: define table schema
 	schema := make(bigquery.Schema, 0, len(opt.Features)+1)
 	for _, h := range opt.Header {
-		if h == opt.EntityName {
+		if h == opt.Entity.Name {
 			schema = append(schema, &bigquery.FieldSchema{
-				Name:        opt.EntityName,
+				Name:        opt.Entity.Name,
 				Type:        bigquery.StringFieldType,
 				Description: "entity key",
 			})
diff --git a/internal/database/offline/postgres/util.go b/internal/database/offline/postgres/util.go
index ecb4c41b..7b1d36bf 100644
--- a/internal/database/offline/postgres/util.go
+++ b/internal/database/offline/postgres/util.go
@@ -20,10 +20,10 @@ func loadDataFromSource(tx *sqlx.Tx, ctx context.Context, opt dbutil.LoadDataFro
 
 	for {
 		record, err := dbutil.ReadLine(dbutil.ReadLineOpt{
-			Source:     opt.Source,
-			EntityName: opt.EntityName,
-			Header:     opt.Header,
-			Features:   opt.Features,
+			Source:   opt.Source,
+			Entity:   opt.Entity,
+			Header:   opt.Header,
+			Features: opt.Features,
 		})
 		if errdefs.Cause(err) == io.EOF {
 			break
diff --git a/internal/database/offline/sqlutil/create_table.go b/internal/database/offline/sqlutil/create_table.go
index 78e272d2..b91f24ed 100644
--- a/internal/database/offline/sqlutil/create_table.go
+++ b/internal/database/offline/sqlutil/create_table.go
@@ -16,25 +16,25 @@ func CreateTable(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.CreateTabl
 	switch opt.TableType {
 	case types.TableBatchSnapshot:
 		// Create primary key (entity_key) on batch snapshot table
-		pkFields := []string{opt.EntityName}
-		schema := dbutil.BuildTableSchema(opt.TableName, opt.EntityName, false, opt.Features, pkFields, dbOpt.Backend)
+		pkFields := []string{opt.Entity.Name}
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, false, opt.Features, pkFields, dbOpt.Backend)
 		if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 			return err
 		}
 	case types.TableStreamSnapshot:
 		// Create primary key (entity_key) on stream snapshot table
-		pkFields := []string{opt.EntityName}
-		schema := dbutil.BuildTableSchema(opt.TableName, opt.EntityName, true, opt.Features, pkFields, dbOpt.Backend)
+		pkFields := []string{opt.Entity.Name}
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, true, opt.Features, pkFields, dbOpt.Backend)
 		if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 			return err
 		}
 	case types.TableStreamCdc:
-		schema := dbutil.BuildTableSchema(opt.TableName, opt.EntityName, true, opt.Features, nil, dbOpt.Backend)
+		schema := dbutil.BuildTableSchema(opt.TableName, opt.Entity, true, opt.Features, nil, dbOpt.Backend)
 		if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 			return err
 		}
 		// Create index (entity_key, unix_milli) on stream cdc table
-		indexFields := []string{opt.EntityName, "unix_milli"}
+		indexFields := []string{opt.Entity.Name, "unix_milli"}
 		indexDDL := dbutil.BuildIndexDDL(opt.TableName, "idx", indexFields, dbOpt.Backend)
 		if err := dbOpt.ExecContext(ctx, indexDDL, nil); err != nil {
 			return err
@@ -60,7 +60,7 @@ func CreateTableNoIndex(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.Cre
 	if dbOpt.Backend == types.BackendBigQuery {
 		tableName = fmt.Sprintf("%s.%s", *dbOpt.DatasetID, tableName)
 	}
-	schema := dbutil.BuildTableSchema(tableName, opt.EntityName, hasUnixMilli, opt.Features, nil, dbOpt.Backend)
+	schema := dbutil.BuildTableSchema(tableName, opt.Entity, hasUnixMilli, opt.Features, nil, dbOpt.Backend)
 	if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 		return err
 	}
diff --git a/internal/database/offline/sqlutil/export.go b/internal/database/offline/sqlutil/export.go
index 545f3060..229ed6dc 100644
--- a/internal/database/offline/sqlutil/export.go
+++ b/internal/database/offline/sqlutil/export.go
@@ -16,6 +16,11 @@ import (
 
 type QueryExportResults func(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.ExportOpt, query string, args []interface{}, features types.FeatureList) (*types.ExportResult, error)
 
+type DoExportOpt struct {
+	offline.ExportOpt
+	QueryResults QueryExportResults
+}
+
 func Export(ctx context.Context, db *sqlx.DB, opt offline.ExportOpt, backend types.BackendType) (*types.ExportResult, error) {
 	dbOpt := dbutil.DBOpt{
 		Backend: backend,
@@ -28,11 +33,6 @@ func Export(ctx context.Context, db *sqlx.DB, opt offline.ExportOpt, backend typ
 	return DoExport(ctx, dbOpt, doJoinOpt)
 }
 
-type DoExportOpt struct {
-	offline.ExportOpt
-	QueryResults QueryExportResults
-}
-
 func DoExport(ctx context.Context, dbOpt dbutil.DBOpt, opt DoExportOpt) (*types.ExportResult, error) {
 	var (
 		emptyStream = make(chan types.ExportRecord)
diff --git a/internal/database/offline/sqlutil/import.go b/internal/database/offline/sqlutil/import.go
index b5f6b915..26a66cd4 100644
--- a/internal/database/offline/sqlutil/import.go
+++ b/internal/database/offline/sqlutil/import.go
@@ -18,11 +18,11 @@ func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData Lo
 	var revision int64
 	err := dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 		// create the data table
-		pkFields := []string{opt.EntityName}
+		pkFields := []string{opt.Entity.Name}
 		if opt.NoPK {
 			pkFields = nil
 		}
-		schema := dbutil.BuildTableSchema(opt.SnapshotTableName, opt.EntityName, false, opt.Features, pkFields, backend)
+		schema := dbutil.BuildTableSchema(opt.SnapshotTableName, opt.Entity, false, opt.Features, pkFields, backend)
 		_, err := tx.ExecContext(ctx, schema)
 		if err != nil {
 			return errdefs.WithStack(err)
@@ -30,12 +30,12 @@ func Import(ctx context.Context, db *sqlx.DB, opt offline.ImportOpt, loadData Lo
 
 		// populate the data table
 		err = loadData(tx, ctx, dbutil.LoadDataFromSourceOpt{
-			Source:     opt.Source,
-			EntityName: opt.EntityName,
-			TableName:  opt.SnapshotTableName,
-			Header:     opt.Header,
-			Features:   opt.Features,
-			Backend:    backend,
+			Source:    opt.Source,
+			Entity:    opt.Entity,
+			TableName: opt.SnapshotTableName,
+			Header:    opt.Header,
+			Features:  opt.Features,
+			Backend:   backend,
 		})
 		if err != nil {
 			return err
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index 8f3a70a3..d5665a83 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -14,6 +14,23 @@ import (
 
 type QueryResults func(ctx context.Context, dbOpt dbutil.DBOpt, query string, header dbutil.ColumnList, dropTableNames []string, backendType types.BackendType) (*types.JoinResult, error)
 
+type ReadJoinedTableOpt struct {
+	EntityRowsTableName string
+	TableNames          []string
+	AllTableNames       []string
+	FeatureMap          map[string]types.FeatureList
+	ValueNames          []string
+	QueryResults        QueryResults
+	ReadJoinResultQuery string
+	BackendType         types.BackendType
+}
+
+type DoJoinOpt struct {
+	offline.JoinOpt
+	QueryResults        QueryResults
+	ReadJoinResultQuery string
+}
+
 func Join(ctx context.Context, db *sqlx.DB, opt offline.JoinOpt, backendType types.BackendType) (*types.JoinResult, error) {
 	dbOpt := dbutil.DBOpt{
 		Backend: backendType,
@@ -27,12 +44,6 @@ func Join(ctx context.Context, db *sqlx.DB, opt offline.JoinOpt, backendType typ
 	return DoJoin(ctx, dbOpt, doJoinOpt)
 }
 
-type DoJoinOpt struct {
-	offline.JoinOpt
-	QueryResults        QueryResults
-	ReadJoinResultQuery string
-}
-
 func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.JoinResult, error) {
 	{
 		empty := true
@@ -51,7 +62,7 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 	}
 
 	// Step 1: prepare temporary table entity_rows
-	entityRowsTableName, err := prepareEntityRowsTable(ctx, dbOpt, opt.EntityRows, opt.ValueNames)
+	entityRowsTableName, err := PrepareEntityRowsTable(ctx, dbOpt, opt.Entity, opt.EntityRows, opt.ValueNames)
 	if err != nil {
 		return nil, err
 	}
@@ -72,10 +83,10 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 		} else {
 			category = types.CategoryStream
 		}
-		joinedTables, err := joinOneGroup(ctx, dbOpt, joinOneGroupOpt{
+		joinedTables, err := JoinOneGroup(ctx, dbOpt, offline.JoinOneGroupOpt{
 			GroupName:           groupName,
 			Category:            category,
-			EntityName:          opt.EntityName,
+			Entity:              opt.Entity,
 			Features:            featureList,
 			RevisionRanges:      revisionRanges,
 			EntityRowsTableName: entityRowsTableName,
@@ -92,7 +103,7 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 	}
 
 	// Step 3: read joined results
-	return readJoinedTable(ctx, dbOpt, readJoinedTableOpt{
+	return ReadJoinedTable(ctx, dbOpt, ReadJoinedTableOpt{
 		EntityRowsTableName: entityRowsTableName,
 		TableNames:          tableNames,
 		AllTableNames:       allTableNames,
@@ -104,23 +115,13 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 	})
 }
 
-type joinOneGroupOpt struct {
-	GroupName           string
-	Category            types.Category
-	Features            types.FeatureList
-	RevisionRanges      []*offline.RevisionRange
-	EntityName          string
-	EntityRowsTableName string
-	ValueNames          []string
-}
-
-func joinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt joinOneGroupOpt) ([]string, error) {
+func JoinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.JoinOneGroupOpt) ([]string, error) {
 	if len(opt.Features) == 0 {
 		return nil, nil
 	}
 
 	// Step 1: create temporary joined table
-	snapshotJoinedTableName, err := prepareJoinedTable(ctx, dbOpt, opt.Features, opt.GroupName, opt.ValueNames)
+	snapshotJoinedTableName, err := PrepareJoinedTable(ctx, dbOpt, opt.Features, opt.Entity, opt.GroupName, opt.ValueNames)
 	if err != nil {
 		return nil, err
 	}
@@ -130,7 +131,7 @@ func joinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt joinOneGroupOpt)
 	for _, r := range opt.RevisionRanges {
 		query, err := buildJoinQuery(joinQueryParams{
 			TableName:           snapshotJoinedTableName,
-			EntityName:          opt.EntityName,
+			EntityName:          opt.Entity.Name,
 			EntityKey:           "entity_key",
 			UnixMilli:           "unix_milli",
 			Columns:             columns,
@@ -151,7 +152,7 @@ func joinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt joinOneGroupOpt)
 	}
 
 	// Step 3: for streaming features, keep joining with cdc_table
-	cdcJoinedTableName, err := prepareJoinedTable(ctx, dbOpt, opt.Features, opt.GroupName, opt.ValueNames)
+	cdcJoinedTableName, err := PrepareJoinedTable(ctx, dbOpt, opt.Features, opt.Entity, opt.GroupName, opt.ValueNames)
 	if err != nil {
 		return nil, err
 	}
@@ -159,7 +160,7 @@ func joinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt joinOneGroupOpt)
 		query, err := buildCdcJoinQuery(cdcJoinQueryParams{
 			TableName:           cdcJoinedTableName,
 			EntityKey:           "entity_key",
-			EntityName:          opt.EntityName,
+			EntityName:          opt.Entity.Name,
 			UnixMilli:           "unix_milli",
 			ValueNames:          opt.ValueNames,
 			FeatureNames:        opt.Features.Names(),
@@ -178,18 +179,7 @@ func joinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt joinOneGroupOpt)
 	return []string{snapshotJoinedTableName, cdcJoinedTableName}, nil
 }
 
-type readJoinedTableOpt struct {
-	EntityRowsTableName string
-	TableNames          []string
-	AllTableNames       []string
-	FeatureMap          map[string]types.FeatureList
-	ValueNames          []string
-	QueryResults        QueryResults
-	ReadJoinResultQuery string
-	BackendType         types.BackendType
-}
-
-func readJoinedTable(ctx context.Context, dbOpt dbutil.DBOpt, opt readJoinedTableOpt) (*types.JoinResult, error) {
+func ReadJoinedTable(ctx context.Context, dbOpt dbutil.DBOpt, opt ReadJoinedTableOpt) (*types.JoinResult, error) {
 	if len(opt.TableNames) == 0 {
 		data := make(chan []interface{})
 		defer close(data)
diff --git a/internal/database/offline/sqlutil/join_helper.go b/internal/database/offline/sqlutil/join_helper.go
index 9d419088..f9daad57 100644
--- a/internal/database/offline/sqlutil/join_helper.go
+++ b/internal/database/offline/sqlutil/join_helper.go
@@ -69,10 +69,11 @@ LEFT JOIN {{ $t2 }}
 ON {{ $t1 }}.{{ qt $.UnixMilli }} = {{ $t2 }}.{{ qt $.UnixMilli }} AND {{ $t1 }}.{{ qt $.EntityKey }} = {{ $t2 }}.{{ qt $.EntityKey }}
 {{end}}`
 
-func prepareJoinedTable(
+func PrepareJoinedTable(
 	ctx context.Context,
 	dbOpt dbutil.DBOpt,
 	features types.FeatureList,
+	entity types.Entity,
 	groupName string,
 	valueNames []string,
 ) (string, error) {
@@ -120,8 +121,9 @@ func prepareJoinedTable(
 	return tableName, nil
 }
 
-func prepareEntityRowsTable(ctx context.Context,
+func PrepareEntityRowsTable(ctx context.Context,
 	dbOpt dbutil.DBOpt,
+	entity types.Entity,
 	entityRows <-chan types.EntityRow,
 	valueNames []string,
 ) (string, error) {
diff --git a/internal/database/offline/sqlutil/snapshot.go b/internal/database/offline/sqlutil/snapshot.go
index d464d981..7f70963b 100644
--- a/internal/database/offline/sqlutil/snapshot.go
+++ b/internal/database/offline/sqlutil/snapshot.go
@@ -71,7 +71,7 @@ func Snapshot(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.SnapshotOpt)
 		currCdcTableName = fmt.Sprintf("%s.%s", *dbOpt.DatasetID, currCdcTableName)
 	}
 
-	schema := dbutil.BuildTableSchema(currSnapshotTableName, opt.Group.Entity.Name, true, opt.Features, []string{opt.Group.Entity.Name}, dbOpt.Backend)
+	schema := dbutil.BuildTableSchema(currSnapshotTableName, opt.Group.Entity, true, opt.Features, []string{opt.Group.Entity.Name}, dbOpt.Backend)
 	if err := dbOpt.ExecContext(ctx, schema, nil); err != nil {
 		return err
 	}
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index 9b04c439..8d378f4a 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -23,7 +23,7 @@ type RevisionRange struct {
 }
 
 type JoinOpt struct {
-	EntityName       string
+	Entity           types.Entity
 	EntityRows       <-chan types.EntityRow
 	GroupNames       []string
 	FeatureMap       map[string]types.FeatureList
@@ -31,12 +31,22 @@ type JoinOpt struct {
 	ValueNames       []string
 }
 
+type JoinOneGroupOpt struct {
+	GroupName           string
+	Category            types.Category
+	Features            types.FeatureList
+	RevisionRanges      []*RevisionRange
+	Entity              types.Entity
+	EntityRowsTableName string
+	ValueNames          []string
+}
+
 type ImportOpt struct {
-	EntityName        string
-	SnapshotTableName string
-	Header            []string
+	Entity            *types.Entity
 	Features          types.FeatureList
+	Header            []string
 	Revision          *int64
+	SnapshotTableName string
 	Source            *CSVSource
 	NoPK              bool // TODO: to import cdc data temporarily for testing
 }
@@ -55,17 +65,17 @@ type CSVSource struct {
 }
 
 type SnapshotOpt struct {
-	Group        types.Group
+	Group        *types.Group
 	Features     types.FeatureList
 	Revision     int64
 	PrevRevision int64
 }
 
 type CreateTableOpt struct {
-	TableName  string
-	EntityName string
-	Features   types.FeatureList
-	TableType  types.TableType
+	TableName string
+	Entity    *types.Entity
+	Features  types.FeatureList
+	TableType types.TableType
 }
 
 type TableSchemaOpt struct {
diff --git a/internal/database/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index 3a46ee08..3ff2af86 100644
--- a/internal/database/online/cassandra/get.go
+++ b/internal/database/online/cassandra/get.go
@@ -10,21 +10,22 @@ import (
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
 	var tableName string
 	if opt.Group.Category == types.CategoryBatch {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 	}
 
 	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`,
 		strings.Join(opt.Features.Names(), ","),
 		tableName,
-		opt.Group.Entity.Name,
+		opt.Entity.Name,
 	)
 
 	scan := make(map[string]interface{})
@@ -51,16 +52,16 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	)
 
 	if opt.Group.Category == types.CategoryBatch {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 	}
-	entityName := opt.Group.Entity.Name
+
 	query := fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s in (%s)`,
-		entityName,
+		opt.Entity.Name,
 		strings.Join(opt.Features.Names(), ","),
 		tableName,
-		entityName,
+		opt.Entity.Name,
 		placeholders,
 	)
 
@@ -76,7 +77,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	}
 
 	for _, s := range scan {
-		entityKey, value := deserializeIntoRowMap(s, entityName, opt.Features)
+		entityKey, value := deserializeIntoRowMap(s, opt.Entity.Name, opt.Features)
 		rs[entityKey] = value
 	}
 	return rs, nil
diff --git a/internal/database/online/cassandra/import.go b/internal/database/online/cassandra/import.go
index 7da7241f..478e9381 100644
--- a/internal/database/online/cassandra/import.go
+++ b/internal/database/online/cassandra/import.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -17,25 +18,23 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	// Step 0: drop existing table for streaming feature
 	var tableName string
 	if opt.Group.Category == types.CategoryBatch {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = sqlutil.OnlineBatchTableName(opt.Revision.ID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 		if err := db.Query(fmt.Sprintf(`DROP TABLE IF EXISTS %s;`, tableName)).Exec(); err != nil {
 			return errdefs.WithStack(err)
 		}
 	}
 
 	// Step 1: create online table
-	if err := db.CreateTable(ctx, online.CreateTableOpt{
-		EntityName: opt.Group.Entity.Name,
-		TableName:  tableName,
-		Features:   opt.Features,
-	}); err != nil {
-		return err
+	entity := opt.Group.Entity
+	columns := append([]string{entity.Name}, opt.Features.Names()...)
+	schema := dbutil.BuildTableSchema(tableName, entity, false, opt.Features, []string{entity.Name}, Backend)
+	if err := db.Query(schema).Exec(); err != nil {
+		return errdefs.WithStack(err)
 	}
 
 	// Step 2: insert records to the online table
-	columns := append([]string{opt.Group.Entity.Name}, opt.Features.Names()...)
 	insertStmt := buildInsertStatement(tableName, columns)
 	batch := db.NewBatch(gocql.LoggedBatch)
 	for record := range opt.ExportStream {
diff --git a/internal/database/online/cassandra/purge.go b/internal/database/online/cassandra/purge.go
index 445d2b17..428d2106 100644
--- a/internal/database/online/cassandra/purge.go
+++ b/internal/database/online/cassandra/purge.go
@@ -4,13 +4,13 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-
 	"github.com/oom-ai/oomstore/pkg/errdefs"
+
+	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
 )
 
 func (db *DB) Purge(ctx context.Context, revisionID int) error {
-	err := db.Query(fmt.Sprintf("DROP TABLE IF EXISTS %s", dbutil.OnlineBatchTableName(revisionID))).
+	err := db.Query(fmt.Sprintf("DROP TABLE IF EXISTS %s", sqlutil.OnlineBatchTableName(revisionID))).
 		WithContext(ctx).Exec()
 	return errdefs.WithStack(err)
 }
diff --git a/internal/database/online/cassandra/store.go b/internal/database/online/cassandra/store.go
index 34f2f132..fa054d0c 100644
--- a/internal/database/online/cassandra/store.go
+++ b/internal/database/online/cassandra/store.go
@@ -4,9 +4,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
-
 	"github.com/gocql/gocql"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
@@ -53,11 +50,3 @@ func Open(option *types.CassandraOpt) (*DB, error) {
 	}
 	return &DB{Session: session}, nil
 }
-
-func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend:     Backend,
-		CassandraDB: db.Session,
-	}
-	return sqlutil.CreateTable(ctx, dbOpt, opt)
-}
diff --git a/internal/database/online/cassandra/push.go b/internal/database/online/cassandra/stream.go
similarity index 52%
rename from internal/database/online/cassandra/push.go
rename to internal/database/online/cassandra/stream.go
index 94824d9e..3feb201f 100644
--- a/internal/database/online/cassandra/push.go
+++ b/internal/database/online/cassandra/stream.go
@@ -10,8 +10,31 @@ import (
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 )
 
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
+
+	if opt.Feature == nil {
+		schema, err := sqlutil.CreateStreamTableSchema(ctx, tableName, opt.Entity, Backend)
+		if err != nil {
+			return err
+		}
+
+		return db.Query(schema).WithContext(ctx).Exec()
+	}
+
+	dbValueType, err := dbutil.DBValueType(Backend, opt.Feature.ValueType)
+	if err != nil {
+		return err
+	}
+
+	sql := fmt.Sprintf("ALTER TABLE %s ADD %s %s", tableName, opt.Feature.Name, dbValueType)
+
+	err = db.Query(sql).WithContext(ctx).Exec()
+	return errdefs.WithStack(err)
+}
+
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
-	tableName := dbutil.OnlineStreamTableName(opt.GroupID)
+	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
 
 	cond := sqlutil.BuildPushCondition(opt, Backend)
 	// cassandra's `insert` is equivalent to `insert_or_update`.
diff --git a/internal/database/online/dynamodb/create_table.go b/internal/database/online/dynamodb/create_table.go
deleted file mode 100644
index ac510d43..00000000
--- a/internal/database/online/dynamodb/create_table.go
+++ /dev/null
@@ -1,37 +0,0 @@
-package dynamodb
-
-import (
-	"context"
-
-	"github.com/aws/aws-sdk-go-v2/aws"
-	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
-	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
-	"github.com/oom-ai/oomstore/internal/database/online"
-	"github.com/oom-ai/oomstore/pkg/errdefs"
-)
-
-func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
-	_, err := db.Client.CreateTable(ctx, &dynamodb.CreateTableInput{
-		TableName: aws.String(opt.TableName),
-		KeySchema: []types.KeySchemaElement{
-			{
-				AttributeName: aws.String(opt.EntityName),
-				KeyType:       types.KeyTypeHash,
-			},
-		},
-		AttributeDefinitions: []types.AttributeDefinition{
-			{
-				AttributeName: aws.String(opt.EntityName),
-				AttributeType: types.ScalarAttributeTypeS,
-			},
-		},
-		ProvisionedThroughput: &types.ProvisionedThroughput{
-			ReadCapacityUnits:  aws.Int64(10),
-			WriteCapacityUnits: aws.Int64(10),
-		},
-	})
-	if err != nil {
-		return errdefs.WithStack(err)
-	}
-	return nil
-}
diff --git a/internal/database/online/dynamodb/get.go b/internal/database/online/dynamodb/get.go
index b827edb6..15ebc97e 100644
--- a/internal/database/online/dynamodb/get.go
+++ b/internal/database/online/dynamodb/get.go
@@ -11,6 +11,7 @@ import (
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	oomTypes "github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -22,9 +23,9 @@ const (
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
 	var tableName string
 	if opt.Group.Category == oomTypes.CategoryBatch {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 	}
 
 	entityKeyValue, err := attributevalue.Marshal(opt.EntityKey)
@@ -34,7 +35,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 	result, err := db.GetItem(ctx, &dynamodb.GetItemInput{
 		TableName: aws.String(tableName),
 		Key: map[string]types.AttributeValue{
-			opt.Group.Entity.Name: entityKeyValue,
+			opt.Entity.Name: entityKeyValue,
 		},
 	})
 	if err != nil {
@@ -50,12 +51,11 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
 	var tableName string
 	if opt.Group.Category == oomTypes.CategoryBatch {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = sqlutil.OnlineBatchTableName(*opt.RevisionID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 	}
 
-	entityName := opt.Group.Entity.Name
 	res := make(map[string]dbutil.RowMap)
 	keys := make([]map[string]types.AttributeValue, 0, BatchGetItemCapacity)
 	for _, entityKey := range opt.EntityKeys {
@@ -64,16 +64,16 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 			return nil, errdefs.WithStack(err)
 		}
 		keys = append(keys, map[string]types.AttributeValue{
-			entityName: entityKeyValue,
+			opt.Entity.Name: entityKeyValue,
 		})
 		if len(keys) == BatchGetItemCapacity {
-			if err = batchGetItem(ctx, db, keys, tableName, entityName, opt.Features, res); err != nil {
+			if err = batchGetItem(ctx, db, keys, tableName, opt.Entity.Name, opt.Features, res); err != nil {
 				return nil, err
 			}
 			keys = make([]map[string]types.AttributeValue, 0, BatchGetItemCapacity)
 		}
 	}
-	if err := batchGetItem(ctx, db, keys, tableName, entityName, opt.Features, res); err != nil {
+	if err := batchGetItem(ctx, db, keys, tableName, opt.Entity.Name, opt.Features, res); err != nil {
 		return nil, err
 	}
 	return res, nil
diff --git a/internal/database/online/dynamodb/import.go b/internal/database/online/dynamodb/import.go
index 8de93833..f05392b2 100644
--- a/internal/database/online/dynamodb/import.go
+++ b/internal/database/online/dynamodb/import.go
@@ -2,7 +2,6 @@ package dynamodb
 
 import (
 	"context"
-	"errors"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 
@@ -11,6 +10,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
 	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 	oomTypes "github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
@@ -23,23 +23,38 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	// Step 0: clean up existing table for streaming feature
 	var tableName string
 	if opt.Group.Category == oomTypes.CategoryBatch {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = sqlutil.OnlineBatchTableName(opt.Revision.ID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 		_, err := db.DeleteTable(ctx, &dynamodb.DeleteTableInput{
 			TableName: aws.String(tableName),
 		})
 		if err != nil {
-			if apiErr := new(types.ResourceNotFoundException); !errors.As(err, &apiErr) {
-				return errdefs.WithStack(err)
-			}
+			return errdefs.WithStack(err)
 		}
 	}
 	// Step 1: create table
-	if err := db.CreateTable(ctx, online.CreateTableOpt{
-		TableName:  tableName,
-		EntityName: opt.Group.Entity.Name,
-	}); err != nil {
+	entity := opt.Group.Entity
+	_, err := db.CreateTable(ctx, &dynamodb.CreateTableInput{
+		TableName: aws.String(tableName),
+		KeySchema: []types.KeySchemaElement{
+			{
+				AttributeName: aws.String(entity.Name),
+				KeyType:       types.KeyTypeHash,
+			},
+		},
+		AttributeDefinitions: []types.AttributeDefinition{
+			{
+				AttributeName: aws.String(entity.Name),
+				AttributeType: types.ScalarAttributeTypeS,
+			},
+		},
+		ProvisionedThroughput: &types.ProvisionedThroughput{
+			ReadCapacityUnits:  aws.Int64(10),
+			WriteCapacityUnits: aws.Int64(10),
+		},
+	})
+	if err != nil {
 		return errdefs.WithStack(err)
 	}
 
@@ -62,7 +77,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 			items = make([]types.WriteRequest, 0, BatchWriteItemCapacity)
 		}
 	}
-	if err := batchWrite(ctx, db, tableName, items); err != nil {
+	if err = batchWrite(ctx, db, tableName, items); err != nil {
 		return err
 	}
 
diff --git a/internal/database/online/dynamodb/purge.go b/internal/database/online/dynamodb/purge.go
index 7f0b5450..77b381bd 100644
--- a/internal/database/online/dynamodb/purge.go
+++ b/internal/database/online/dynamodb/purge.go
@@ -3,16 +3,15 @@ package dynamodb
 import (
 	"context"
 
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
 	"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"
 )
 
 func (db *DB) Purge(ctx context.Context, revisionID int) error {
-	tableName := dbutil.OnlineBatchTableName(revisionID)
+	tableName := sqlutil.OnlineBatchTableName(revisionID)
 	_, err := db.DeleteTable(ctx, &dynamodb.DeleteTableInput{
 		TableName: aws.String(tableName),
 	})
diff --git a/internal/database/online/dynamodb/push.go b/internal/database/online/dynamodb/push.go
deleted file mode 100644
index 55df61c5..00000000
--- a/internal/database/online/dynamodb/push.go
+++ /dev/null
@@ -1,45 +0,0 @@
-package dynamodb
-
-import (
-	"context"
-
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-
-	"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/oom-ai/oomstore/pkg/errdefs"
-
-	"github.com/oom-ai/oomstore/internal/database/online"
-)
-
-func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
-	var (
-		tableName = dbutil.OnlineStreamTableName(opt.GroupID)
-		item      = make(map[string]types.AttributeValue)
-	)
-
-	entityKeyValue, err := attributevalue.Marshal(opt.EntityKey)
-	if err != nil {
-		return errdefs.WithStack(err)
-	}
-	item[opt.EntityName] = entityKeyValue
-
-	for i, feature := range opt.Features {
-		value, err := dbutil.SerializeByValueType(opt.FeatureValues[i], feature.ValueType, Backend)
-		if err != nil {
-			return err
-		}
-		attributeValue, err := attributevalue.Marshal(value)
-		if err != nil {
-			return errdefs.WithStack(err)
-		}
-		item[feature.Name] = attributeValue
-	}
-
-	_, err = db.Client.PutItem(ctx, &dynamodb.PutItemInput{
-		TableName: &tableName,
-		Item:      item,
-	})
-	return errdefs.WithStack(err)
-}
diff --git a/internal/database/online/dynamodb/store.go b/internal/database/online/dynamodb/store.go
index adb08dab..00374843 100644
--- a/internal/database/online/dynamodb/store.go
+++ b/internal/database/online/dynamodb/store.go
@@ -7,9 +7,9 @@ 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/oom-ai/oomstore/pkg/errdefs"
 
 	"github.com/oom-ai/oomstore/internal/database/online"
-	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
diff --git a/internal/database/online/dynamodb/stream.go b/internal/database/online/dynamodb/stream.go
new file mode 100644
index 00000000..ea9d90a8
--- /dev/null
+++ b/internal/database/online/dynamodb/stream.go
@@ -0,0 +1,77 @@
+package dynamodb
+
+import (
+	"context"
+
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
+
+	"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/oom-ai/oomstore/pkg/errdefs"
+
+	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/internal/database/online/sqlutil"
+)
+
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	// dynamodb has no "column", so we do nothing to "add column". see https://stackoverflow.com/a/25610645/16428442
+	if opt.Feature != nil {
+		return nil
+	}
+
+	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
+
+	_, err := db.CreateTable(ctx, &dynamodb.CreateTableInput{
+		TableName: aws.String(tableName),
+		KeySchema: []types.KeySchemaElement{
+			{
+				AttributeName: aws.String(opt.Entity.Name),
+				KeyType:       types.KeyTypeHash,
+			},
+		},
+		AttributeDefinitions: []types.AttributeDefinition{
+			{
+				AttributeName: aws.String(opt.Entity.Name),
+				AttributeType: types.ScalarAttributeTypeS,
+			},
+		},
+		ProvisionedThroughput: &types.ProvisionedThroughput{
+			ReadCapacityUnits:  aws.Int64(10),
+			WriteCapacityUnits: aws.Int64(10),
+		},
+	})
+	return errdefs.WithStack(err)
+}
+
+func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
+	var (
+		tableName = sqlutil.OnlineStreamTableName(opt.GroupID)
+		item      = make(map[string]types.AttributeValue)
+	)
+
+	entityKeyValue, err := attributevalue.Marshal(opt.EntityKey)
+	if err != nil {
+		return errdefs.WithStack(err)
+	}
+	item[opt.Entity.Name] = entityKeyValue
+
+	for i, feature := range opt.Features {
+		value, err := dbutil.SerializeByValueType(opt.FeatureValues[i], feature.ValueType, Backend)
+		if err != nil {
+			return err
+		}
+		attributevalue, err := attributevalue.Marshal(value)
+		if err != nil {
+			return errdefs.WithStack(err)
+		}
+		item[feature.Name] = attributevalue
+	}
+
+	_, err = db.Client.PutItem(ctx, &dynamodb.PutItemInput{
+		TableName: &tableName,
+		Item:      item,
+	})
+	return errdefs.WithStack(err)
+}
diff --git a/internal/database/online/mock_online/store.go b/internal/database/online/mock_online/store.go
index fd6c7a3d..ddc97541 100644
--- a/internal/database/online/mock_online/store.go
+++ b/internal/database/online/mock_online/store.go
@@ -50,20 +50,6 @@ func (mr *MockStoreMockRecorder) Close() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockStore)(nil).Close))
 }
 
-// CreateTable mocks base method.
-func (m *MockStore) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
-	m.ctrl.T.Helper()
-	ret := m.ctrl.Call(m, "CreateTable", ctx, opt)
-	ret0, _ := ret[0].(error)
-	return ret0
-}
-
-// CreateTable indicates an expected call of CreateTable.
-func (mr *MockStoreMockRecorder) CreateTable(ctx, opt interface{}) *gomock.Call {
-	mr.mock.ctrl.T.Helper()
-	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "CreateTable", reflect.TypeOf((*MockStore)(nil).CreateTable), ctx, opt)
-}
-
 // Get mocks base method.
 func (m *MockStore) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
 	m.ctrl.T.Helper()
@@ -122,6 +108,20 @@ func (mr *MockStoreMockRecorder) Ping(ctx interface{}) *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Ping", reflect.TypeOf((*MockStore)(nil).Ping), ctx)
 }
 
+// PrepareStreamTable mocks base method.
+func (m *MockStore) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "PrepareStreamTable", ctx, opt)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// PrepareStreamTable indicates an expected call of PrepareStreamTable.
+func (mr *MockStoreMockRecorder) PrepareStreamTable(ctx, opt interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PrepareStreamTable", reflect.TypeOf((*MockStore)(nil).PrepareStreamTable), ctx, opt)
+}
+
 // Purge mocks base method.
 func (m *MockStore) Purge(ctx context.Context, revisionID int) error {
 	m.ctrl.T.Helper()
diff --git a/internal/database/online/mysql/store.go b/internal/database/online/mysql/store.go
index c28a822f..6d88c390 100644
--- a/internal/database/online/mysql/store.go
+++ b/internal/database/online/mysql/store.go
@@ -51,7 +51,7 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 
 // TODO: refactor with text/template
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
-	tableName := dbutil.OnlineStreamTableName(opt.GroupID)
+	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
 
 	cond := sqlutil.BuildPushCondition(opt, Backend)
 	query := fmt.Sprintf(`INSERT INTO %s (%s) VALUES(%s) ON DUPLICATE KEY UPDATE %s`,
@@ -65,10 +65,6 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	return errdefs.WithStack(err)
 }
 
-func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend: Backend,
-		SqlxDB:  db.DB,
-	}
-	return sqlutil.CreateTable(ctx, dbOpt, opt)
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index 9b709388..1c36a73c 100644
--- a/internal/database/online/postgres/store.go
+++ b/internal/database/online/postgres/store.go
@@ -50,7 +50,7 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 }
 
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
-	tableName := dbutil.OnlineStreamTableName(opt.GroupID)
+	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
 
 	cond := sqlutil.BuildPushCondition(opt, Backend)
 
@@ -58,7 +58,7 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 		tableName,
 		cond.Inserts,
 		cond.InsertPlaceholders,
-		opt.EntityName,
+		opt.Entity.Name,
 		cond.UpdatePlaceholders,
 	)
 
@@ -66,10 +66,6 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	return errdefs.WithStack(err)
 }
 
-func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend: Backend,
-		SqlxDB:  db.DB,
-	}
-	return sqlutil.CreateTable(ctx, dbOpt, opt)
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/redis/get.go b/internal/database/online/redis/get.go
index 0f90ae03..c6fc357f 100644
--- a/internal/database/online/redis/get.go
+++ b/internal/database/online/redis/get.go
@@ -58,10 +58,11 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	res := make(map[string]dbutil.RowMap)
 	for _, entityKey := range opt.EntityKeys {
 		rowMap, err := db.Get(ctx, online.GetOpt{
-			EntityKey:  entityKey,
+			Entity:     opt.Entity,
+			RevisionID: opt.RevisionID,
 			Group:      opt.Group,
+			EntityKey:  entityKey,
 			Features:   opt.Features,
-			RevisionID: opt.RevisionID,
 		})
 		if err != nil {
 			return res, errdefs.WithStack(err)
diff --git a/internal/database/online/redis/import.go b/internal/database/online/redis/import.go
index b4c9dbf2..44b316b0 100644
--- a/internal/database/online/redis/import.go
+++ b/internal/database/online/redis/import.go
@@ -24,7 +24,7 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 		var key string
 		var err error
 		if opt.Group.Category == types.CategoryBatch {
-			key, err = serializeRedisKeyForBatchFeature(*opt.RevisionID, entityKey)
+			key, err = serializeRedisKeyForBatchFeature(opt.Revision.ID, entityKey)
 		} else {
 			key, err = serializeRedisKeyForStreamFeature(opt.Group.ID, entityKey)
 		}
diff --git a/internal/database/online/redis/store.go b/internal/database/online/redis/store.go
index 5a588b9f..9d843943 100644
--- a/internal/database/online/redis/store.go
+++ b/internal/database/online/redis/store.go
@@ -34,6 +34,6 @@ func Open(opt *types.RedisOpt) *DB {
 	return &DB{redis.NewClient(&redisOpt)}
 }
 
-func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
 	return nil
 }
diff --git a/internal/database/online/sqlite/store.go b/internal/database/online/sqlite/store.go
index 7ca8784d..02640d4f 100644
--- a/internal/database/online/sqlite/store.go
+++ b/internal/database/online/sqlite/store.go
@@ -49,7 +49,7 @@ func (db *DB) Purge(ctx context.Context, revisionID int) error {
 }
 
 func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
-	tableName := dbutil.OnlineStreamTableName(opt.GroupID)
+	tableName := sqlutil.OnlineStreamTableName(opt.GroupID)
 
 	cond := sqlutil.BuildPushCondition(opt, Backend)
 
@@ -63,10 +63,6 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	return errdefs.WithStack(err)
 }
 
-func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
-	dbOpt := dbutil.DBOpt{
-		Backend: Backend,
-		SqlxDB:  db.DB,
-	}
-	return sqlutil.CreateTable(ctx, dbOpt, opt)
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	return sqlutil.SqlxPrapareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/sqlutil/create_table.go b/internal/database/online/sqlutil/create_table.go
deleted file mode 100644
index d891495d..00000000
--- a/internal/database/online/sqlutil/create_table.go
+++ /dev/null
@@ -1,24 +0,0 @@
-package sqlutil
-
-import (
-	"context"
-
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-	"github.com/oom-ai/oomstore/pkg/errdefs"
-
-	"github.com/oom-ai/oomstore/internal/database/online"
-)
-
-func CreateTable(ctx context.Context, dbOpt dbutil.DBOpt, opt online.CreateTableOpt) error {
-	// Step 1: drop existing table
-	if err := dbutil.DropTable(ctx, dbOpt, opt.TableName); err != nil {
-		return err
-	}
-	// Step 2: create new table
-	schema := dbutil.BuildTableSchema(opt.TableName, opt.EntityName, false, opt.Features, []string{opt.EntityName}, dbOpt.Backend)
-	err := dbOpt.ExecContext(ctx, schema, nil)
-	if err != nil {
-		return errdefs.WithStack(err)
-	}
-	return nil
-}
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index b199937a..da9357fb 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -16,15 +16,14 @@ import (
 func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.BackendType) (dbutil.RowMap, error) {
 	var tableName string
 	if opt.Group.Category == types.CategoryBatch {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = OnlineBatchTableName(*opt.RevisionID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = OnlineStreamTableName(opt.Group.ID)
 	}
 
-	entityName := opt.Group.Entity.Name
 	featureNames := opt.Features.Names()
 	qt := dbutil.QuoteFn(backend)
-	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`, qt(featureNames...), qt(tableName), qt(entityName))
+	query := fmt.Sprintf(`SELECT %s FROM %s WHERE %s = ?`, qt(featureNames...), qt(tableName), qt(opt.Entity.Name))
 
 	record, err := db.QueryRowxContext(ctx, db.Rebind(query), opt.EntityKey).SliceScan()
 	if err != nil {
@@ -45,15 +44,14 @@ func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.Back
 func MultiGet(ctx context.Context, db *sqlx.DB, opt online.MultiGetOpt, backend types.BackendType) (map[string]dbutil.RowMap, error) {
 	var tableName string
 	if opt.Group.Category == types.CategoryBatch {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = OnlineBatchTableName(*opt.RevisionID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = OnlineStreamTableName(opt.Group.ID)
 	}
 
-	entityName := opt.Group.Entity.Name
 	featureNames := opt.Features.Names()
 	qt := dbutil.QuoteFn(backend)
-	query := fmt.Sprintf(`SELECT %s, %s FROM %s WHERE %s in (?);`, qt(entityName), qt(featureNames...), qt(tableName), qt(entityName))
+	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, errdefs.WithStack(err)
diff --git a/internal/database/online/sqlutil/import.go b/internal/database/online/sqlutil/import.go
index ba2c9ce7..95c56adf 100644
--- a/internal/database/online/sqlutil/import.go
+++ b/internal/database/online/sqlutil/import.go
@@ -18,13 +18,13 @@ func Import(ctx context.Context, db *sqlx.DB, opt online.ImportOpt, backend type
 	if opt.Group.Category == types.CategoryStream {
 		tableName = dbutil.TempTable("online_stream")
 	} else {
-		tableName = dbutil.OnlineBatchTableName(*opt.RevisionID)
+		tableName = OnlineBatchTableName(opt.Revision.ID)
 	}
 	entity := opt.Group.Entity
 	columns := append([]string{entity.Name}, opt.Features.Names()...)
 	err := dbutil.WithTransaction(db, ctx, func(ctx context.Context, tx *sqlx.Tx) error {
 		// create the data table
-		schema := dbutil.BuildTableSchema(tableName, entity.Name, false, opt.Features, []string{entity.Name}, backend)
+		schema := dbutil.BuildTableSchema(tableName, entity, false, opt.Features, []string{entity.Name}, backend)
 		_, err := tx.ExecContext(ctx, schema)
 		if err != nil {
 			return errdefs.WithStack(err)
@@ -52,7 +52,7 @@ func Import(ctx context.Context, db *sqlx.DB, opt online.ImportOpt, backend type
 			return <-opt.ExportError
 		}
 		if opt.Group.Category == types.CategoryStream {
-			streamTableName := dbutil.OnlineStreamTableName(opt.Group.ID)
+			streamTableName := OnlineStreamTableName(opt.Group.ID)
 			if err = PurgeTx(ctx, tx, streamTableName, backend); err != nil {
 				return err
 			}
diff --git a/internal/database/online/sqlutil/purge.go b/internal/database/online/sqlutil/purge.go
index dcae09a5..ca7fcebb 100644
--- a/internal/database/online/sqlutil/purge.go
+++ b/internal/database/online/sqlutil/purge.go
@@ -13,7 +13,7 @@ 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(dbutil.OnlineBatchTableName(revisionID)))
+	query := fmt.Sprintf(`DROP TABLE IF EXISTS %s;`, qt(OnlineBatchTableName(revisionID)))
 	_, err := db.ExecContext(ctx, query)
 	return errdefs.WithStack(err)
 }
diff --git a/internal/database/online/sqlutil/push.go b/internal/database/online/sqlutil/push.go
index a48b4a7f..6d2be354 100644
--- a/internal/database/online/sqlutil/push.go
+++ b/internal/database/online/sqlutil/push.go
@@ -21,7 +21,7 @@ func BuildPushCondition(opt online.PushOpt, backend types.BackendType) *PushCond
 	qt := dbutil.QuoteFn(backend)
 	cond := PushCondition{}
 
-	cond.Inserts = qt(append([]string{opt.EntityName}, opt.Features.Names()...)...)
+	cond.Inserts = qt(append([]string{opt.Entity.Name}, opt.Features.Names()...)...)
 	cond.InsertValues = append([]interface{}{opt.EntityKey}, opt.FeatureValues...)
 	cond.InsertPlaceholders = dbutil.Fill(len(cond.InsertValues), "?", ",")
 
diff --git a/internal/database/online/sqlutil/util.go b/internal/database/online/sqlutil/util.go
new file mode 100644
index 00000000..2808cadf
--- /dev/null
+++ b/internal/database/online/sqlutil/util.go
@@ -0,0 +1,60 @@
+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/errdefs"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+)
+
+func OnlineBatchTableName(revisionID int) string {
+	return fmt.Sprintf("online_batch_%d", revisionID)
+}
+
+func OnlineStreamTableName(groupID int) string {
+	return fmt.Sprintf("online_stream_%d", groupID)
+}
+
+func CreateStreamTableSchema(ctx context.Context, tableName string, entity *types.Entity, backend types.BackendType) (string, error) {
+	var (
+		entityFormat string
+	)
+	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" TEXT PRIMARY KEY`, entity.Name)
+	default:
+		return "", errdefs.InvalidAttribute(errdefs.Errorf("backend %s not support", backend))
+	}
+
+	schema := fmt.Sprintf("CREATE TABLE %s ( %s )", tableName, entityFormat)
+	return schema, nil
+}
+
+func SqlxPrapareStreamTable(ctx context.Context, db *sqlx.DB, opt online.PrepareStreamTableOpt, backend types.BackendType) error {
+	tableName := OnlineStreamTableName(opt.GroupID)
+
+	if opt.Feature == nil {
+		schema, err := CreateStreamTableSchema(ctx, tableName, opt.Entity, backend)
+		if err != nil {
+			return err
+		}
+		_, err = db.ExecContext(ctx, schema)
+		return errdefs.WithStack(err)
+	}
+
+	dbValueType, err := dbutil.DBValueType(backend, opt.Feature.ValueType)
+	if err != nil {
+		return err
+	}
+
+	sql := fmt.Sprintf("ALTER TABLE %s ADD COLUMN %s %s", tableName, opt.Feature.Name, dbValueType)
+	_, err = db.ExecContext(ctx, sql)
+	return errdefs.WithStack(err)
+}
diff --git a/internal/database/online/store.go b/internal/database/online/store.go
index 8e10fee0..4efdbea4 100644
--- a/internal/database/online/store.go
+++ b/internal/database/online/store.go
@@ -11,7 +11,6 @@ type Store interface {
 	Get(ctx context.Context, opt GetOpt) (dbutil.RowMap, error)
 	MultiGet(ctx context.Context, opt MultiGetOpt) (map[string]dbutil.RowMap, error)
 	Purge(ctx context.Context, revisionID int) error
-	CreateTable(ctx context.Context, opt CreateTableOpt) error
 
 	// Import batch / streaming features to online store
 	Import(ctx context.Context, opt ImportOpt) error
@@ -20,6 +19,8 @@ type Store interface {
 	// Note: Make sure that the table corresponding to the stream feature already exists before executing this method
 	Push(ctx context.Context, opt PushOpt) error
 
+	PrepareStreamTable(ctx context.Context, opt PrepareStreamTableOpt) error
+
 	Ping(ctx context.Context) error
 	io.Closer
 }
diff --git a/internal/database/online/tikv/get.go b/internal/database/online/tikv/get.go
index 9e03fa13..81dab74c 100644
--- a/internal/database/online/tikv/get.go
+++ b/internal/database/online/tikv/get.go
@@ -18,10 +18,11 @@ type input struct {
 func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error) {
 	// Proxy to MultiGet
 	res, err := db.MultiGet(ctx, online.MultiGetOpt{
-		EntityKeys: []string{opt.EntityKey},
+		Entity:     opt.Entity,
 		Group:      opt.Group,
-		Features:   opt.Features,
 		RevisionID: opt.RevisionID,
+		EntityKeys: []string{opt.EntityKey},
+		Features:   opt.Features,
 	})
 	if err != nil {
 		return nil, err
@@ -35,11 +36,14 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 
 func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]dbutil.RowMap, error) {
 	var (
+		allBatch            bool
 		serializedPrefixKey string
 		err                 error
 	)
 
-	if opt.Group.Category == types.CategoryBatch {
+	allBatch = (opt.Group.Category == types.CategoryBatch)
+
+	if allBatch {
 		serializedPrefixKey, err = dbutil.SerializeByValue(*opt.RevisionID, Backend)
 	} else {
 		serializedPrefixKey, err = dbutil.SerializeByValue(opt.Group.ID, Backend)
@@ -70,7 +74,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 	var keys [][]byte
 	for _, serializedEntityKey := range serializedEntityKeys {
 		for _, serializedFeatureID := range serializedFeatureIDs {
-			if opt.Group.Category == types.CategoryBatch {
+			if allBatch {
 				keys = append(keys, getKeyOfBatchFeature(serializedPrefixKey, serializedEntityKey, serializedFeatureID))
 			} else {
 				keys = append(keys, getKeyOfStreamFeature(serializedPrefixKey, serializedEntityKey, serializedFeatureID))
diff --git a/internal/database/online/tikv/import.go b/internal/database/online/tikv/import.go
index bba84207..8e7b3f71 100644
--- a/internal/database/online/tikv/import.go
+++ b/internal/database/online/tikv/import.go
@@ -3,8 +3,6 @@ package tikv
 import (
 	"context"
 
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
@@ -15,14 +13,8 @@ const BatchSize = 100
 
 func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	var seq int64
-	var err error
-	var serializedRevisionID, serializedGroupID string
 
-	if opt.Group.Category == types.CategoryBatch {
-		serializedRevisionID, err = dbutil.SerializeByValue(*opt.RevisionID, Backend)
-	} else {
-		serializedGroupID, err = dbutil.SerializeByValue(opt.Group.ID, Backend)
-	}
+	serializedRevisionID, err := dbutil.SerializeByValue(opt.Revision.ID, Backend)
 	if err != nil {
 		return err
 	}
@@ -62,11 +54,8 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 			if err != nil {
 				return err
 			}
-			if opt.Group.Category == types.CategoryBatch {
-				putKeys = append(putKeys, getKeyOfBatchFeature(serializedRevisionID, serializedEntityKey, serializedFeatureIDs[i]))
-			} else {
-				putKeys = append(putKeys, getKeyOfStreamFeature(serializedGroupID, serializedEntityKey, serializedFeatureIDs[i]))
-			}
+
+			putKeys = append(putKeys, getKeyOfBatchFeature(serializedRevisionID, serializedEntityKey, serializedFeatureIDs[i]))
 			putVals = append(putVals, []byte(serializedFeatureValue.(string)))
 		}
 
diff --git a/internal/database/online/tikv/store.go b/internal/database/online/tikv/store.go
index e37a8555..1ed9bfc5 100644
--- a/internal/database/online/tikv/store.go
+++ b/internal/database/online/tikv/store.go
@@ -43,6 +43,6 @@ func (db *DB) Ping(ctx context.Context) error {
 	return errdefs.WithStack(err)
 }
 
-func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
 	return nil
 }
diff --git a/internal/database/online/types.go b/internal/database/online/types.go
index c9c65a25..21b2213c 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -5,37 +5,43 @@ import (
 )
 
 type GetOpt struct {
+	Entity     *types.Entity
 	EntityKey  string
-	Group      types.Group
-	Features   types.FeatureList
 	RevisionID *int
+	Group      *types.Group
+	Features   types.FeatureList
 }
 
 type MultiGetOpt struct {
+	Entity     *types.Entity
 	EntityKeys []string
-	Group      types.Group
-	Features   types.FeatureList
 	RevisionID *int
+	Group      *types.Group
+	Features   types.FeatureList
 }
 
 type ImportOpt struct {
 	Group        types.Group
 	Features     types.FeatureList
+	Revision     *types.Revision
 	ExportStream <-chan types.ExportRecord
 	ExportError  <-chan error
-	RevisionID   *int
 }
 
 type PushOpt struct {
-	EntityName    string
+	Entity        *types.Entity
 	EntityKey     string
 	GroupID       int
 	Features      types.FeatureList
 	FeatureValues []interface{}
 }
 
-type CreateTableOpt struct {
-	EntityName string
-	TableName  string
-	Features   types.FeatureList
+type PrepareStreamTableOpt struct {
+	Entity *types.Entity
+
+	GroupID int
+
+	// Feature is not nil to add a new row to the stream table;
+	// otherwise it means the stream table will be created.
+	Feature *types.Feature
 }
diff --git a/oomagent/root.go b/oomagent/root.go
index 7fe33ec6..0120d2ae 100644
--- a/oomagent/root.go
+++ b/oomagent/root.go
@@ -8,6 +8,7 @@ import (
 	"net"
 	"os"
 	"path/filepath"
+	"strconv"
 
 	"github.com/adrg/xdg"
 	"github.com/spf13/cobra"
@@ -43,6 +44,16 @@ var rootCmd = &cobra.Command{
 		if err != nil {
 			log.Fatalf("failed to listen: %v", err)
 		}
+
+		// write the listening address
+		tmpdir := filepath.Join(os.TempDir(), "oomagent", strconv.Itoa(os.Getpid()))
+		if err := os.MkdirAll(tmpdir, 0755); err != nil {
+			log.Fatalf("failed to create temp directory: %v", err)
+		}
+		if err := os.WriteFile(filepath.Join(tmpdir, "address"), []byte(lis.Addr().String()), 0644); err != nil {
+			log.Fatalf("failed to write listen address: %v", err)
+		}
+
 		grpcServer := grpc.NewServer()
 		codegen.RegisterOomAgentServer(grpcServer, &server{oomstore: oomStore})
 		if err := grpcServer.Serve(lis); err != nil {
diff --git a/oomcli/cmd/util.go b/oomcli/cmd/util.go
index 3cf94758..db938acd 100644
--- a/oomcli/cmd/util.go
+++ b/oomcli/cmd/util.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/oom-ai/oomstore/pkg/oomstore"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -43,6 +44,11 @@ func groupsToApplyGroupItems(ctx context.Context, store *oomstore.OomStore, grou
 }
 
 func exitf(format string, a ...interface{}) {
+	if true {
+		format = strings.ReplaceAll(format, "{err}", "%+v")
+	} else {
+		format = strings.ReplaceAll(format, "{err}", "%v")
+	}
 	msg := fmt.Sprintf(format, a...)
 	fmt.Fprintf(os.Stderr, "Error: %s\n", msg)
 	os.Exit(1)
diff --git a/pkg/errdefs/errdefs.go b/pkg/errdefs/errdefs.go
index 12d3eb3c..55b86b8f 100644
--- a/pkg/errdefs/errdefs.go
+++ b/pkg/errdefs/errdefs.go
@@ -1,19 +1,11 @@
 package errdefs
 
-import (
-	"fmt"
-)
-
 type errNotFound struct{ error }
 
 func (e errNotFound) Unwrap() error {
 	return e.error
 }
 
-func (e errNotFound) Error() string {
-	return fmt.Sprintf("%+v", e.error)
-}
-
 func NotFound(err error) error {
 	if err == nil || IsNotFound(err) {
 		return err
@@ -27,10 +19,6 @@ func (e errInvalidAttribute) Unwrap() error {
 	return e.error
 }
 
-func (e errInvalidAttribute) Error() string {
-	return fmt.Sprintf("%+v", e.error)
-}
-
 func InvalidAttribute(err error) error {
 	if err == nil || IsInvalidAttribute(err) {
 		return err
diff --git a/pkg/errdefs/wrap.go b/pkg/errdefs/wrap.go
index c1b900ed..ac73632b 100644
--- a/pkg/errdefs/wrap.go
+++ b/pkg/errdefs/wrap.go
@@ -26,7 +26,7 @@ func Errorf(format string, args ...interface{}) error {
 }
 
 func WithStack(err error) error {
-	if !debug || hasStack(err) {
+	if !debug {
 		return err
 	}
 
@@ -36,35 +36,3 @@ func WithStack(err error) error {
 func Cause(err error) error {
 	return errors.Cause(err)
 }
-
-// not call callers() on every wrap, see https://github.com/pkg/errors/issues/75#issuecomment-574580408
-func hasStack(err error) bool {
-	if err == nil {
-		return false
-	}
-
-	type causer interface {
-		Cause() error
-	}
-
-	type tracer interface {
-		StackTrace() errors.StackTrace
-	}
-
-	e := err
-	for {
-		if _, ok := e.(tracer); ok {
-			return true
-		}
-
-		cause, ok := e.(causer)
-		if !ok {
-			return false
-		}
-
-		e = cause.Cause()
-		if e == nil {
-			return false
-		}
-	}
-}
diff --git a/pkg/oomstore/apply.go b/pkg/oomstore/apply.go
index df20bf7e..dec4fdaf 100644
--- a/pkg/oomstore/apply.go
+++ b/pkg/oomstore/apply.go
@@ -4,8 +4,6 @@ import (
 	"context"
 	"io"
 
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-
 	"github.com/mitchellh/mapstructure"
 	"github.com/spf13/cast"
 	"gopkg.in/yaml.v3"
@@ -110,7 +108,7 @@ func (s *OomStore) applyGroup(ctx context.Context, tx metadata.DBStore, newGroup
 			return nil, err
 		}
 
-		_, err = tx.CreateGroup(ctx, metadata.CreateGroupOpt{
+		id, err := tx.CreateGroup(ctx, metadata.CreateGroupOpt{
 			GroupName:        newGroup.Name,
 			EntityID:         entity.ID,
 			Category:         newGroup.Category,
@@ -121,6 +119,14 @@ func (s *OomStore) applyGroup(ctx context.Context, tx metadata.DBStore, newGroup
 			return nil, err
 		}
 
+		if newGroup.Category == types.CategoryStream {
+			return func() error {
+				return s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+					Entity:  entity,
+					GroupID: id,
+				})
+			}, nil
+		}
 		return nil, nil
 	}
 
@@ -162,20 +168,11 @@ func (s *OomStore) applyFeature(ctx context.Context, tx metadata.DBStore, newFea
 		if !errdefs.IsNotFound(err) {
 			return nil, err
 		}
-
-		revisions, err := tx.ListRevision(ctx, &group.ID)
-		if err != nil {
-			return nil, err
-		}
-		if len(revisions) > 0 {
-			return nil, errdefs.Errorf("group %s already has data and cannot add features due to the join and export mechanism", group.Name)
-		}
-
 		valueType, err := types.ParseValueType(newFeature.ValueType)
 		if err != nil {
 			return nil, err
 		}
-		_, err = tx.CreateFeature(ctx, metadata.CreateFeatureOpt{
+		id, err := tx.CreateFeature(ctx, metadata.CreateFeatureOpt{
 			FeatureName: newFeature.Name,
 			GroupID:     group.ID,
 			Description: newFeature.Description,
@@ -186,17 +183,15 @@ func (s *OomStore) applyFeature(ctx context.Context, tx metadata.DBStore, newFea
 		}
 
 		if group.Category == types.CategoryStream {
-			features, err := tx.ListFeature(ctx, metadata.ListFeatureOpt{
-				GroupID: &group.ID,
-			})
+			feature, err := tx.GetFeature(ctx, id)
 			if err != nil {
 				return nil, err
 			}
 			return func() error {
-				return s.online.CreateTable(ctx, online.CreateTableOpt{
-					EntityName: group.Entity.Name,
-					TableName:  dbutil.OnlineStreamTableName(group.ID),
-					Features:   features,
+				return s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+					Entity:  group.Entity,
+					GroupID: group.ID,
+					Feature: feature,
 				})
 			}, nil
 		}
diff --git a/pkg/oomstore/feature.go b/pkg/oomstore/feature.go
index 6f1fd0d8..a903350e 100644
--- a/pkg/oomstore/feature.go
+++ b/pkg/oomstore/feature.go
@@ -3,21 +3,18 @@ package oomstore
 import (
 	"context"
 
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-
 	"github.com/oom-ai/oomstore/internal/database/metadata"
 	"github.com/oom-ai/oomstore/internal/database/online"
-	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 	"github.com/oom-ai/oomstore/pkg/oomstore/util"
 )
 
-// GetFeature gets metadata of a feature by ID.
+// Get metadata of a feature by ID.
 func (s *OomStore) GetFeature(ctx context.Context, id int) (*types.Feature, error) {
 	return s.metadata.GetFeature(ctx, id)
 }
 
-// GetFeatureByFullName gets metadata of a feature by full name.
+// Get metadata of a feature by full name.
 func (s *OomStore) GetFeatureByFullName(ctx context.Context, fullName string) (*types.Feature, error) {
 	groupName, featureName, err := util.SplitFullFeatureName(fullName)
 	if err != nil {
@@ -26,12 +23,12 @@ func (s *OomStore) GetFeatureByFullName(ctx context.Context, fullName string) (*
 	return s.GetFeatureByName(ctx, groupName, featureName)
 }
 
-// GetFeatureByName gets metadata of a feature by group name and feature name.
+// Get metadata of a feature by group name and feature name.
 func (s *OomStore) GetFeatureByName(ctx context.Context, groupName string, featureName string) (*types.Feature, error) {
 	return s.metadata.GetFeatureByName(ctx, groupName, featureName)
 }
 
-// ListFeature lists metadata of features meeting particular criteria.
+// List metadata of features meeting particular criteria.
 func (s *OomStore) ListFeature(ctx context.Context, opt types.ListFeatureOpt) (types.FeatureList, error) {
 	metadataOpt := metadata.ListFeatureOpt{}
 	if opt.EntityName != nil {
@@ -58,7 +55,7 @@ func (s *OomStore) ListFeature(ctx context.Context, opt types.ListFeatureOpt) (t
 	return features, nil
 }
 
-// UpdateFeature updates metadata of a feature.
+// Update metadata of a feature.
 func (s *OomStore) UpdateFeature(ctx context.Context, opt types.UpdateFeatureOpt) error {
 	if err := util.ValidateFullFeatureNames(opt.FeatureName); err != nil {
 		return err
@@ -73,21 +70,13 @@ func (s *OomStore) UpdateFeature(ctx context.Context, opt types.UpdateFeatureOpt
 	})
 }
 
-// CreateFeature creates metadata of a feature.
+// Create metadata of a feature.
 func (s *OomStore) CreateFeature(ctx context.Context, opt types.CreateFeatureOpt) (int, error) {
 	group, err := s.metadata.GetGroupByName(ctx, opt.GroupName)
 	if err != nil {
 		return 0, err
 	}
 
-	revisions, err := s.metadata.ListRevision(ctx, &group.ID)
-	if err != nil {
-		return 0, err
-	}
-	if len(revisions) > 0 {
-		return 0, errdefs.Errorf("group %s already has data and cannot add features due to the join and export mechanism", group.Name)
-	}
-
 	id, err := s.metadata.CreateFeature(ctx, metadata.CreateFeatureOpt{
 		FeatureName: opt.FeatureName,
 		GroupID:     group.ID,
@@ -99,16 +88,14 @@ func (s *OomStore) CreateFeature(ctx context.Context, opt types.CreateFeatureOpt
 	}
 
 	if group.Category == types.CategoryStream {
-		features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
-			GroupID: &group.ID,
-		})
+		feature, err := s.metadata.GetFeature(ctx, id)
 		if err != nil {
 			return 0, err
 		}
-		if err = s.online.CreateTable(ctx, online.CreateTableOpt{
-			EntityName: group.Entity.Name,
-			TableName:  dbutil.OnlineStreamTableName(group.ID),
-			Features:   features,
+		if err := s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+			Entity:  group.Entity,
+			GroupID: group.ID,
+			Feature: feature,
 		}); err != nil {
 			return 0, err
 		}
diff --git a/pkg/oomstore/group.go b/pkg/oomstore/group.go
index 13f04467..1b96401d 100644
--- a/pkg/oomstore/group.go
+++ b/pkg/oomstore/group.go
@@ -4,10 +4,11 @@ import (
 	"context"
 
 	"github.com/oom-ai/oomstore/internal/database/metadata"
+	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-// CreateGroup registers a feature group.
+// Create metadata of a feature group.
 func (s *OomStore) CreateGroup(ctx context.Context, opt types.CreateGroupOpt) (int, error) {
 	entity, err := s.metadata.GetEntityByName(ctx, opt.EntityName)
 	if err != nil {
@@ -25,25 +26,33 @@ func (s *OomStore) CreateGroup(ctx context.Context, opt types.CreateGroupOpt) (i
 		return 0, err
 	}
 
+	if opt.Category == types.CategoryStream {
+		if err = s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+			Entity:  entity,
+			GroupID: id,
+		}); err != nil {
+			return 0, err
+		}
+	}
 	return id, nil
 }
 
-// GetGroup gets metadata of a feature group by ID.
+// Get metadata of a feature group by ID.
 func (s *OomStore) GetGroup(ctx context.Context, id int) (*types.Group, error) {
 	return s.metadata.GetGroup(ctx, id)
 }
 
-// GetGroupByName gets metadata of a feature group by name.
+// Get metadata of a feature group by name.
 func (s *OomStore) GetGroupByName(ctx context.Context, name string) (*types.Group, error) {
 	return s.metadata.GetGroupByName(ctx, name)
 }
 
-// ListGroup lists metadata of feature groups under the same entity.
+// List metadata of feature groups under the same entity.
 func (s *OomStore) ListGroup(ctx context.Context, entityID *int) (types.GroupList, error) {
 	return s.metadata.ListGroup(ctx, entityID, nil)
 }
 
-// UpdateGroup updates metadata of a feature group.
+// Update metadata of a feature group.
 func (s *OomStore) UpdateGroup(ctx context.Context, opt types.UpdateGroupOpt) error {
 	group, err := s.metadata.GetGroupByName(ctx, opt.GroupName)
 	if err != nil {
diff --git a/pkg/oomstore/import.go b/pkg/oomstore/import.go
index 38a82312..dda1af7c 100644
--- a/pkg/oomstore/import.go
+++ b/pkg/oomstore/import.go
@@ -51,13 +51,13 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 		Delimiter: dataSource.Delimiter,
 	}
 	// read header
-	columnNames := append([]string{opt.entityName}, opt.features.Names()...)
+	columnNames := append([]string{opt.entity.Name}, opt.features.Names()...)
 	header, err := readHeader(source, columnNames)
 	if err != nil {
 		return 0, err
 	}
 
-	newRevisionID, snapshotTableName, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
+	newRevisionID, snapshotTableName, err := s.CreateRevision(ctx, metadata.CreateRevisionOpt{
 		Revision:      time.Now().UnixMilli(),
 		GroupID:       opt.group.ID,
 		SnapshotTable: nil,
@@ -69,7 +69,7 @@ func (s *OomStore) csvReaderImport(ctx context.Context, opt *importOpt, dataSour
 	}
 
 	revision, err := s.offline.Import(ctx, offline.ImportOpt{
-		EntityName:        opt.entityName,
+		Entity:            opt.entity,
 		Features:          opt.features,
 		Header:            cast.ToStringSlice(header),
 		Revision:          opt.Revision,
@@ -112,7 +112,7 @@ func (s *OomStore) tableLinkImport(ctx context.Context, opt *importOpt, dataSour
 	} else {
 		revision = *opt.Revision
 	}
-	newRevisionID, _, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
+	newRevisionID, _, err := s.CreateRevision(ctx, metadata.CreateRevisionOpt{
 		Revision:      revision,
 		GroupID:       opt.group.ID,
 		SnapshotTable: &dataSource.TableName,
@@ -199,10 +199,10 @@ func (s *OomStore) parseImportOpt(ctx context.Context, opt types.ImportOpt) (*im
 		return nil, err
 	}
 	return &importOpt{
-		ImportOpt:  &opt,
-		entityName: entity.Name,
-		group:      group,
-		features:   features,
+		ImportOpt: &opt,
+		entity:    entity,
+		group:     group,
+		features:  features,
 	}, nil
 }
 
@@ -232,7 +232,7 @@ func (s *OomStore) getGroupInfo(ctx context.Context, groupName string) (*types.E
 
 type importOpt struct {
 	*types.ImportOpt
-	entityName string
-	group      *types.Group
-	features   types.FeatureList
+	entity   *types.Entity
+	group    *types.Group
+	features types.FeatureList
 }
diff --git a/pkg/oomstore/import_stream.go b/pkg/oomstore/import_stream.go
index b6335459..baa9ec2d 100644
--- a/pkg/oomstore/import_stream.go
+++ b/pkg/oomstore/import_stream.go
@@ -66,10 +66,10 @@ func (s *OomStore) csvReaderImportStream(ctx context.Context, opt types.ImportSt
 	records := make([]types.StreamRecord, 0, ImportStreamBatchSize)
 	for {
 		line, err := dbutil.ReadLine(dbutil.ReadLineOpt{
-			Source:     source,
-			EntityName: entity.Name,
-			Header:     header,
-			Features:   features,
+			Source:   source,
+			Entity:   entity,
+			Header:   header,
+			Features: features,
 		})
 		if errdefs.Cause(err) == io.EOF {
 			break
@@ -113,7 +113,7 @@ func (s *OomStore) tableLinkImportStream(ctx context.Context, opt types.ImportSt
 		return err
 	}
 
-	_, _, err = s.createRevision(ctx, metadata.CreateRevisionOpt{
+	_, _, err = s.CreateRevision(ctx, metadata.CreateRevisionOpt{
 		Revision:    tableSchema.TimeRange.MinUnixMilli,
 		GroupID:     group.ID,
 		CdcTable:    &dataSource.TableName,
diff --git a/pkg/oomstore/join.go b/pkg/oomstore/join.go
index acef96bf..db4423a9 100644
--- a/pkg/oomstore/join.go
+++ b/pkg/oomstore/join.go
@@ -23,18 +23,21 @@ func (s *OomStore) ChannelJoin(ctx context.Context, opt types.ChannelJoinOpt) (*
 	if err := util.ValidateFullFeatureNames(opt.JoinFeatureNames...); err != nil {
 		return nil, err
 	}
+	data := make(chan []interface{})
+	defer close(data)
 
+	emptyResult := &types.JoinResult{
+		Data: data,
+	}
 	features, err := s.ListFeature(ctx, types.ListFeatureOpt{
 		FeatureNames: &opt.JoinFeatureNames,
 	})
 	if err != nil {
 		return nil, err
 	}
-	if len(features) == 0 {
-		data := make(chan []interface{})
-		defer close(data)
 
-		return &types.JoinResult{Data: data}, nil
+	if len(features) == 0 {
+		return emptyResult, nil
 	}
 
 	entity, err := getSharedEntity(features)
@@ -62,7 +65,7 @@ func (s *OomStore) ChannelJoin(ctx context.Context, opt types.ChannelJoinOpt) (*
 	}
 
 	return s.offline.Join(ctx, offline.JoinOpt{
-		EntityName:       entity.Name,
+		Entity:           *entity,
 		EntityRows:       opt.EntityRows,
 		GroupNames:       groupNames,
 		FeatureMap:       featureMap,
@@ -144,12 +147,12 @@ func (s *OomStore) buildRevisionRanges(ctx context.Context, group *types.Group)
 func GetEntityRowsFromInputFile(inputFilePath string) (<-chan types.EntityRow, []string, error) {
 	input, err := os.Open(inputFilePath)
 	if err != nil {
-		return nil, nil, errdefs.WithStack(err)
+		return nil, nil, err
 	}
 	reader := csv.NewReader(input)
 	header, err := reader.Read()
 	if err != nil {
-		return nil, nil, errdefs.WithStack(err)
+		return nil, nil, err
 	}
 	entityRows := make(chan types.EntityRow)
 	var readErr error
@@ -163,7 +166,7 @@ func GetEntityRowsFromInputFile(inputFilePath string) (<-chan types.EntityRow, [
 				break
 			}
 			if err != nil {
-				readErr = errdefs.WithStack(err)
+				readErr = err
 				return
 			}
 			if len(line) < 2 {
@@ -172,7 +175,7 @@ func GetEntityRowsFromInputFile(inputFilePath string) (<-chan types.EntityRow, [
 			}
 			unixMilli, err := strconv.Atoi(line[1])
 			if err != nil {
-				readErr = errdefs.WithStack(err)
+				readErr = err
 				return
 			}
 			entityRows <- types.EntityRow{
diff --git a/pkg/oomstore/online_query.go b/pkg/oomstore/online_query.go
index 94f03817..88fd0d63 100644
--- a/pkg/oomstore/online_query.go
+++ b/pkg/oomstore/online_query.go
@@ -3,8 +3,6 @@ package oomstore
 import (
 	"context"
 
-	"github.com/oom-ai/oomstore/internal/database/metadata"
-
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
@@ -23,9 +21,7 @@ func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*type
 		FeatureNames:    opt.FeatureNames,
 		FeatureValueMap: make(map[string]interface{}),
 	}
-	features := s.metadata.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{
-		FullNames: &opt.FeatureNames,
-	})
+	features := s.metadata.ListCachedFeature(ctx, &opt.FeatureNames)
 	if len(features) == 0 {
 		return &rs, nil
 	}
@@ -38,15 +34,18 @@ func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*type
 
 	featureMap := groupFeaturesByGroupID(features)
 
-	for _, featureList := range featureMap {
-		if len(featureList) == 0 {
+	for _, features := range featureMap {
+		if len(features) == 0 {
 			continue
 		}
+		revisionID := features[0].OnlineRevisionID()
+		group := features[0].Group
 		featureValues, err := s.online.Get(ctx, online.GetOpt{
+			Entity:     entity,
+			RevisionID: revisionID,
+			Group:      group,
 			EntityKey:  opt.EntityKey,
-			Group:      *featureList[0].Group,
-			Features:   featureList,
-			RevisionID: featureList[0].OnlineRevisionID(),
+			Features:   features,
 		})
 		if err != nil {
 			return nil, err
@@ -64,9 +63,7 @@ func (s *OomStore) OnlineMultiGet(ctx context.Context, opt types.OnlineMultiGetO
 		return nil, err
 	}
 	result := make(map[string]*types.FeatureValues)
-	features := s.metadata.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{
-		FullNames: &opt.FeatureNames,
-	})
+	features := s.metadata.ListCachedFeature(ctx, &opt.FeatureNames)
 	if len(features) == 0 {
 		return result, nil
 	}
@@ -108,11 +105,14 @@ func (s *OomStore) getFeatureValueMap(ctx context.Context, entityKeys []string,
 		if len(features) == 0 {
 			continue
 		}
+		revisionID := features[0].OnlineRevisionID()
+		group := features[0].Group
 		featureValues, err := s.online.MultiGet(ctx, online.MultiGetOpt{
+			Entity:     entity,
+			RevisionID: revisionID,
+			Group:      group,
 			EntityKeys: entityKeys,
-			Group:      *features[0].Group,
 			Features:   features,
-			RevisionID: features[0].OnlineRevisionID(),
 		})
 		if err != nil {
 			return nil, err
diff --git a/pkg/oomstore/push.go b/pkg/oomstore/push.go
index c12a0bb5..6789758d 100644
--- a/pkg/oomstore/push.go
+++ b/pkg/oomstore/push.go
@@ -4,8 +4,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/oom-ai/oomstore/internal/database/metadata"
-
 	"github.com/oom-ai/oomstore/pkg/errdefs"
 
 	"github.com/oom-ai/oomstore/internal/database/online"
@@ -14,9 +12,12 @@ import (
 
 // Push inserts stream feature values to online store and offline store
 func (s *OomStore) Push(ctx context.Context, opt types.PushOpt) error {
-	features := s.metadata.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{
+	features, err := s.ListFeature(ctx, types.ListFeatureOpt{
 		GroupName: &opt.GroupName,
 	})
+	if err != nil {
+		return err
+	}
 	var featureNames []string
 	for name := range opt.FeatureValues {
 		featureNames = append(featureNames, name)
@@ -32,8 +33,8 @@ func (s *OomStore) Push(ctx context.Context, opt types.PushOpt) error {
 		values = append(values, opt.FeatureValues[f.Name])
 	}
 
-	if err := s.online.Push(ctx, online.PushOpt{
-		EntityName:    entity.Name,
+	if err = s.online.Push(ctx, online.PushOpt{
+		Entity:        entity,
 		EntityKey:     opt.EntityKey,
 		GroupID:       group.ID,
 		Features:      features,
diff --git a/pkg/oomstore/push_processor.go b/pkg/oomstore/push_processor.go
index 6e668ff3..7eff52b3 100644
--- a/pkg/oomstore/push_processor.go
+++ b/pkg/oomstore/push_processor.go
@@ -125,14 +125,17 @@ func (p *PushProcessor) Push(record types.StreamRecord) {
 }
 
 func (p *PushProcessor) pushToOffline(ctx context.Context, s *OomStore, groupID int) error {
-	features := s.metadata.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{
+	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
 		GroupID: &groupID,
 	})
+	if err != nil {
+		return err
+	}
 	entity := features[0].Entity()
 
 	value, _ := p.buffer.Load(groupID)
 	b := value.(GroupBuffer)
-	group, err := s.metadata.GetCachedGroup(ctx, groupID)
+	group, err := s.metadata.GetGroup(ctx, groupID)
 	if err != nil {
 		return err
 	}
@@ -147,23 +150,26 @@ func (p *PushProcessor) pushToOffline(ctx context.Context, s *OomStore, groupID
 }
 
 func (s *OomStore) newRevisionForStream(ctx context.Context, groupID int, revision int64) error {
-	features := s.metadata.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{
+	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
 		GroupID: &groupID,
 	})
+	if err != nil {
+		return err
+	}
 	entity := features[0].Entity()
 
-	if err := s.offline.CreateTable(ctx, offline.CreateTableOpt{
-		TableName:  dbutil.OfflineStreamCdcTableName(groupID, revision),
-		EntityName: entity.Name,
-		Features:   features,
-		TableType:  types.TableStreamCdc,
+	if err = s.offline.CreateTable(ctx, offline.CreateTableOpt{
+		TableName: dbutil.OfflineStreamCdcTableName(groupID, revision),
+		Entity:    entity,
+		Features:  features,
+		TableType: types.TableStreamCdc,
 	}); err != nil {
 		return err
 	}
 
 	snapshotTable := ""
 	cdcTable := dbutil.OfflineStreamCdcTableName(groupID, revision)
-	if _, _, err := s.createRevision(ctx, metadata.CreateRevisionOpt{
+	if _, _, err = s.CreateRevision(ctx, metadata.CreateRevisionOpt{
 		GroupID:       groupID,
 		Revision:      revision,
 		SnapshotTable: &snapshotTable,
diff --git a/pkg/oomstore/revision.go b/pkg/oomstore/revision.go
index 2ee4656a..14f6ebe5 100644
--- a/pkg/oomstore/revision.go
+++ b/pkg/oomstore/revision.go
@@ -25,7 +25,7 @@ func (s *OomStore) GetRevisionBy(ctx context.Context, groupID int, revision int6
 	return s.metadata.GetRevisionBy(ctx, groupID, revision)
 }
 
-func (s *OomStore) createRevision(ctx context.Context, opt metadata.CreateRevisionOpt) (int, string, error) {
+func (s *OomStore) CreateRevision(ctx context.Context, opt metadata.CreateRevisionOpt) (int, string, error) {
 	var revisionID int
 	var snapshotTable string
 	var dummyRevision *types.Revision
@@ -77,10 +77,10 @@ func (s *OomStore) createSnapshotAndCdcTable(ctx context.Context, revision *type
 	}
 
 	if err = s.offline.CreateTable(ctx, offline.CreateTableOpt{
-		TableName:  snapshotTable,
-		EntityName: revision.Group.Entity.Name,
-		Features:   features,
-		TableType:  types.TableStreamSnapshot,
+		TableName: snapshotTable,
+		Entity:    revision.Group.Entity,
+		Features:  features,
+		TableType: types.TableStreamSnapshot,
 	}); err != nil {
 		return err
 	}
@@ -89,10 +89,10 @@ func (s *OomStore) createSnapshotAndCdcTable(ctx context.Context, revision *type
 	if revision.Group.Category == types.CategoryStream {
 		tableName := dbutil.OfflineStreamCdcTableName(revision.GroupID, revision.Revision)
 		if err = s.offline.CreateTable(ctx, offline.CreateTableOpt{
-			TableName:  tableName,
-			EntityName: revision.Group.Entity.Name,
-			Features:   features,
-			TableType:  types.TableStreamCdc,
+			TableName: tableName,
+			Entity:    revision.Group.Entity,
+			Features:  features,
+			TableType: types.TableStreamCdc,
 		}); err != nil {
 			return err
 		}
@@ -127,10 +127,10 @@ func (s *OomStore) createFirstSnapshotTable(ctx context.Context, revision *types
 		return err
 	}
 	if err = s.offline.CreateTable(ctx, offline.CreateTableOpt{
-		TableName:  snapshotTable,
-		EntityName: revision.Group.Entity.Name,
-		Features:   features,
-		TableType:  types.TableStreamSnapshot,
+		TableName: snapshotTable,
+		Entity:    revision.Group.Entity,
+		Features:  features,
+		TableType: types.TableStreamSnapshot,
 	}); err != nil {
 		return err
 	}
diff --git a/pkg/oomstore/snapshot.go b/pkg/oomstore/snapshot.go
index c8b519ab..566a5a50 100644
--- a/pkg/oomstore/snapshot.go
+++ b/pkg/oomstore/snapshot.go
@@ -50,7 +50,7 @@ func (s *OomStore) Snapshot(ctx context.Context, groupName string) error {
 		}
 		tableName := dbutil.OfflineStreamSnapshotTableName(group.ID, revision.Revision)
 		if err = s.offline.Snapshot(ctx, offline.SnapshotOpt{
-			Group:        *group,
+			Group:        group,
 			Features:     features,
 			Revision:     revisions[i].Revision,
 			PrevRevision: revisions[i-1].Revision,
diff --git a/pkg/oomstore/sync.go b/pkg/oomstore/sync.go
index df144f1f..c413f1c4 100644
--- a/pkg/oomstore/sync.go
+++ b/pkg/oomstore/sync.go
@@ -50,9 +50,9 @@ func (s *OomStore) Sync(ctx context.Context, opt types.SyncOpt) error {
 	if err = s.online.Import(ctx, online.ImportOpt{
 		Group:        *group,
 		Features:     features,
+		Revision:     revision,
 		ExportStream: exportResult.Data,
 		ExportError:  exportResult.GetErrorChannel(),
-		RevisionID:   &revision.ID,
 	}); err != nil {
 		return err
 	}
diff --git a/sdk/rust/.gitignore b/sdk/rust/.gitignore
index 4fffb2f8..8eb581d0 100644
--- a/sdk/rust/.gitignore
+++ b/sdk/rust/.gitignore
@@ -1,2 +1,3 @@
 /target
 /Cargo.lock
+/.idea
diff --git a/sdk/rust/Cargo.toml b/sdk/rust/Cargo.toml
index a59a603f..df34e669 100644
--- a/sdk/rust/Cargo.toml
+++ b/sdk/rust/Cargo.toml
@@ -13,10 +13,12 @@ async-stream = "0.3"
 itertools = "0.10"
 prost = "0.9"
 thiserror = "1.0"
-tokio = { version = "1.15", features = ["macros", "rt-multi-thread"] }
+tokio = { version = "1.15", features = ["macros", "rt-multi-thread", "fs", "process"] }
 tonic = "0.6"
 futures-core = "0.3"
 tokio-stream = "0.1"
+signal-hook = "0.3"
+signal-hook-tokio = {version = "0.3", features = ["futures-v0_3"]}
 
 [build-dependencies]
 tonic-build = "0.6"
diff --git a/sdk/rust/examples/embedded_agent.rs b/sdk/rust/examples/embedded_agent.rs
new file mode 100644
index 00000000..f0bef7f2
--- /dev/null
+++ b/sdk/rust/examples/embedded_agent.rs
@@ -0,0 +1,13 @@
+use oomclient::EmbeddedAgent;
+use std::time::Duration;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let agent = EmbeddedAgent::default().await?;
+
+    println!("{:?}", agent.address());
+
+    tokio::time::sleep(Duration::from_secs(10)).await;
+
+    Ok(())
+}
diff --git a/sdk/rust/src/agent.rs b/sdk/rust/src/agent.rs
new file mode 100644
index 00000000..47ae6315
--- /dev/null
+++ b/sdk/rust/src/agent.rs
@@ -0,0 +1,94 @@
+use signal_hook::{consts::signal::*, low_level::emulate_default_handler};
+use signal_hook_tokio::{Handle, Signals};
+use std::{
+    env,
+    net::{IpAddr, SocketAddr},
+    path::Path,
+};
+use tokio::{fs, process::Command, time, time::Duration};
+use tokio_stream::StreamExt;
+
+use crate::Result;
+
+pub struct EmbeddedAgent {
+    addr:   SocketAddr,
+    handle: Handle,
+}
+
+impl EmbeddedAgent {
+    pub async fn new<P1, P2>(cmd_path: Option<P1>, cfg_path: Option<P2>, port: Option<u16>) -> Result<Self>
+    where
+        P1: AsRef<Path>,
+        P2: AsRef<Path>,
+    {
+        let cmd_path = cmd_path.map(|p| p.as_ref().to_owned());
+        let cfg_path = cfg_path.map(|p| p.as_ref().to_owned());
+        let port = port.unwrap_or(0);
+
+        let mut signals = Signals::new(&[SIGHUP, SIGTERM, SIGINT, SIGQUIT])?;
+        let handle = signals.handle();
+
+        let mut oomagent = Command::new(cmd_path.clone().unwrap_or_else(|| "oomagent".into()));
+        oomagent.arg("--port").arg(port.to_string());
+        if let Some(cfg_path) = cfg_path.clone() {
+            oomagent.arg("--config").arg(cfg_path);
+        }
+        oomagent.kill_on_drop(true);
+
+        let mut child = oomagent.spawn()?;
+        let pid = child.id();
+
+        tokio::spawn({
+            async move {
+                while let Some(signal) = signals.next().await {
+                    child.kill().await.unwrap();
+                    emulate_default_handler(signal).unwrap();
+                }
+            }
+        });
+
+        let addr = get_agent_address(pid.unwrap()).await?;
+        Ok(Self { handle, addr })
+    }
+
+    pub async fn default() -> Result<Self> {
+        Self::new(None::<String>, None::<String>, None).await
+    }
+
+    pub fn ip(&self) -> IpAddr {
+        self.addr.ip()
+    }
+
+    pub fn port(&self) -> u16 {
+        self.addr.port()
+    }
+
+    pub fn address(&self) -> SocketAddr {
+        self.addr
+    }
+}
+
+impl Drop for EmbeddedAgent {
+    fn drop(&mut self) {
+        self.handle.close();
+    }
+}
+
+async fn get_agent_address(pid: u32) -> Result<SocketAddr> {
+    let mut path = env::temp_dir();
+    path.push("oomagent");
+    path.push(pid.to_string());
+    path.push("address");
+    let time = time::Instant::now();
+
+    loop {
+        let result = fs::read_to_string(&path)
+            .await
+            .map_err(|e| e.into())
+            .and_then(|addr| Ok(addr.parse()?));
+        if result.is_ok() || time.elapsed() > Duration::from_millis(3000) {
+            return result;
+        }
+        time::sleep(Duration::from_millis(200)).await;
+    }
+}
diff --git a/sdk/rust/src/error.rs b/sdk/rust/src/error.rs
index 193da603..b06272c8 100644
--- a/sdk/rust/src/error.rs
+++ b/sdk/rust/src/error.rs
@@ -1,4 +1,4 @@
-use std::num;
+use std::{io, net::AddrParseError, num};
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -12,6 +12,12 @@ pub enum OomError {
     #[error("a checked integral type conversion fails.")]
     IntConversionError(#[from] num::TryFromIntError),
 
+    #[error(transparent)]
+    IoError(#[from] io::Error),
+
+    #[error(transparent)]
+    AddrParseError(#[from] AddrParseError),
+
     #[error("unknown error")]
     Unknown(String),
 }
diff --git a/sdk/rust/src/lib.rs b/sdk/rust/src/lib.rs
index 7e0a1216..a4e827fe 100644
--- a/sdk/rust/src/lib.rs
+++ b/sdk/rust/src/lib.rs
@@ -1,5 +1,7 @@
+mod agent;
 pub mod error;
 mod util;
+
 mod oomagent {
     tonic::include_proto!("oomagent");
 }
@@ -29,9 +31,10 @@ use std::{collections::HashMap, path::Path};
 use tonic::{codegen::StdError, transport, Request};
 use util::{parse_raw_feature_values, parse_raw_values};
 
+pub use agent::EmbeddedAgent;
 pub use oomagent::{value::Value, EntityRow};
 
-type Result<T> = std::result::Result<T, OomError>;
+pub type Result<T> = std::result::Result<T, OomError>;
 
 #[derive(Debug, Clone)]
 pub struct Client {
@@ -105,7 +108,7 @@ impl Client {
         purge_delay: u32,
     ) -> Result<()> {
         let group = group.into();
-        let revision_id = revision_id.into().map(|x| i32::try_from(x)).transpose()?;
+        let revision_id = revision_id.into().map(i32::try_from).transpose()?;
         let purge_delay = i32::try_from(purge_delay)?;
         self.inner.sync(SyncRequest { revision_id, group, purge_delay }).await?;
         Ok(())


Test patch:
diff --git a/internal/database/dbutil/schema_test.go b/internal/database/dbutil/schema_test.go
index e11ea001..36bb6270 100644
--- a/internal/database/dbutil/schema_test.go
+++ b/internal/database/dbutil/schema_test.go
@@ -81,7 +81,7 @@ CREATE TABLE "user" (
 
 	for _, c := range cases {
 		t.Run(c.description, func(t *testing.T) {
-			schema := dbutil.BuildTableSchema(c.tableName, c.entity.Name, false, c.features, nil, c.backend)
+			schema := dbutil.BuildTableSchema(c.tableName, c.entity, false, c.features, nil, c.backend)
 			require.Equal(t, strings.TrimSpace(c.want), schema)
 		})
 	}
diff --git a/internal/database/metadata/informer/informer_test.go b/internal/database/metadata/informer/informer_test.go
index 46733200..5bf54421 100644
--- a/internal/database/metadata/informer/informer_test.go
+++ b/internal/database/metadata/informer/informer_test.go
@@ -5,8 +5,6 @@ import (
 	"testing"
 	"time"
 
-	"github.com/oom-ai/oomstore/internal/database/metadata"
-
 	"github.com/oom-ai/oomstore/internal/database/metadata/informer"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 	"github.com/oom-ai/oomstore/pkg/oomstore/util"
@@ -52,9 +50,7 @@ func TestInformer(t *testing.T) {
 	defer informer.Close()
 
 	name := util.ComposeFullFeatureName("group", "price")
-	features := informer.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{
-		FullNames: &[]string{name},
-	})
+	features := informer.ListCachedFeature(ctx, &[]string{name})
 	require.Equal(t, 1, len(features))
 	feature := features[0]
 
@@ -66,25 +62,18 @@ func TestInformer(t *testing.T) {
 	assert.Equal(t, 100, feature.Group.ID)
 	assert.Equal(t, 1, feature.Group.Entity.ID)
 	assert.Equal(t, "entity", feature.Group.Entity.Name)
-
-	groupName := "group"
-	featuresByGroup := informer.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{
-		GroupName: &groupName,
-	})
-	assert.Equal(t, 1, len(featuresByGroup))
-	assert.Equal(t, feature, featuresByGroup[0])
 }
 
 func TestInformerDeepCopy(t *testing.T) {
 	ctx, informer := prepareInformer(t)
 	defer informer.Close()
 
-	features := informer.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{})
+	features := informer.ListCachedFeature(ctx, nil)
 	require.Equal(t, 1, len(features))
 	// changing this entity should not change the internal state of the informer
 	features[0].Name = "new_price"
 
-	features = informer.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{})
+	features = informer.ListCachedFeature(ctx, nil)
 	require.Equal(t, 1, len(features))
 	assert.Equal(t, 1, features[0].ID)
 	assert.Equal(t, 100, features[0].GroupID)
diff --git a/internal/database/metadata/test_impl/feature.go b/internal/database/metadata/test_impl/feature.go
index 7c6a4b8b..b1ad9252 100644
--- a/internal/database/metadata/test_impl/feature.go
+++ b/internal/database/metadata/test_impl/feature.go
@@ -176,7 +176,7 @@ func TestListCachedFeature(t *testing.T, prepareStore PrepareStoreFn, destroySto
 	_, groupID := prepareEntityAndGroup(t, ctx, store)
 
 	// case 1: no feature to list
-	features := store.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{})
+	features := store.ListCachedFeature(ctx, nil)
 	assert.Equal(t, 0, features.Len())
 
 	_, err := store.CreateFeature(ctx, metadata.CreateFeatureOpt{
@@ -189,13 +189,11 @@ func TestListCachedFeature(t *testing.T, prepareStore PrepareStoreFn, destroySto
 	require.NoError(t, store.Refresh())
 
 	// case 2: no condition, list all features
-	features = store.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{})
+	features = store.ListCachedFeature(ctx, nil)
 	assert.Equal(t, 1, features.Len())
 
 	// case 8: list features by FeatureNames
-	features = store.ListCachedFeature(ctx, metadata.ListCachedFeatureOpt{
-		FullNames: &[]string{"device_info.phone"},
-	})
+	features = store.ListCachedFeature(ctx, &[]string{"device_info.phone"})
 	assert.Equal(t, 1, len(features))
 }
 
diff --git a/internal/database/offline/test_impl/create_table.go b/internal/database/offline/test_impl/create_table.go
index 2c77b395..77eb4222 100644
--- a/internal/database/offline/test_impl/create_table.go
+++ b/internal/database/offline/test_impl/create_table.go
@@ -35,28 +35,28 @@ func TestCreateTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 		{
 			describtion: "cdc table (with unix_milli)",
 			opt: offline.CreateTableOpt{
-				TableName:  tableName + "_0",
-				EntityName: entity.Name,
-				Features:   features,
-				TableType:  types.TableStreamCdc,
+				TableName: tableName + "_0",
+				Entity:    &entity,
+				Features:  features,
+				TableType: types.TableStreamCdc,
 			},
 		},
 		{
 			describtion: "stream snapshot table (with unix_milli)",
 			opt: offline.CreateTableOpt{
-				TableName:  tableName + "_1",
-				EntityName: entity.Name,
-				Features:   features,
-				TableType:  types.TableStreamSnapshot,
+				TableName: tableName + "_1",
+				Entity:    &entity,
+				Features:  features,
+				TableType: types.TableStreamSnapshot,
 			},
 		},
 		{
 			describtion: "batch snapshot table (without unix_milli)",
 			opt: offline.CreateTableOpt{
-				TableName:  tableName + "_2",
-				EntityName: entity.Name,
-				Features:   features,
-				TableType:  types.TableBatchSnapshot,
+				TableName: tableName + "_2",
+				Entity:    &entity,
+				Features:  features,
+				TableType: types.TableBatchSnapshot,
 			},
 		},
 	}
diff --git a/internal/database/offline/test_impl/import.go b/internal/database/offline/test_impl/import.go
index fa33b3d8..5c171ecb 100644
--- a/internal/database/offline/test_impl/import.go
+++ b/internal/database/offline/test_impl/import.go
@@ -28,7 +28,7 @@ func TestImport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 	snapshotTable := "offline_1_1"
 
 	opt := offline.ImportOpt{
-		EntityName:        entity.Name,
+		Entity:            &entity,
 		SnapshotTableName: snapshotTable,
 		Header:            []string{"device", "model", "price"},
 		Source: &offline.CSVSource{
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index 00b609c6..37a595b1 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -71,6 +71,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		{
 			description: "no features",
 			opt: offline.JoinOpt{
+
 				FeatureMap: make(map[string]types.FeatureList),
 			},
 			expected: prepareEmptyResult(),
@@ -78,7 +79,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		{
 			description: "no entity rows",
 			opt: offline.JoinOpt{
-				EntityName: entity.Name,
+				Entity:     *entity,
 				GroupNames: oneGroupGroupNames,
 				FeatureMap: oneGroupFeatureMap,
 				EntityRows: prepareEntityRows(true, false),
@@ -88,7 +89,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		{
 			description: "one batch feature group",
 			opt: offline.JoinOpt{
-				EntityName:       entity.Name,
+				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, false),
 				GroupNames:       oneGroupGroupNames,
 				FeatureMap:       oneGroupFeatureMap,
@@ -99,7 +100,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		{
 			description: "two batch feature groups",
 			opt: offline.JoinOpt{
-				EntityName:       entity.Name,
+				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, false),
 				GroupNames:       twoGroupGroupNames,
 				FeatureMap:       twoGroupFeatureMap,
@@ -110,7 +111,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		{
 			description: "two batch feature groups, with extra values",
 			opt: offline.JoinOpt{
-				EntityName:       entity.Name,
+				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, true),
 				GroupNames:       twoGroupGroupNames,
 				FeatureMap:       twoGroupFeatureMap,
@@ -122,7 +123,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		{
 			description: "one streaming feature group, one batch group",
 			opt: offline.JoinOpt{
-				EntityName:       entity.Name,
+				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, true),
 				GroupNames:       twoGroupGroupNames,
 				FeatureMap:       twoGroupFeatureMap,
diff --git a/internal/database/offline/test_impl/push.go b/internal/database/offline/test_impl/push.go
index 56b2268e..cc6720e1 100644
--- a/internal/database/offline/test_impl/push.go
+++ b/internal/database/offline/test_impl/push.go
@@ -72,10 +72,10 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 
 	t.Run("push when cdc table exists", func(t *testing.T) {
 		err := store.CreateTable(ctx, offline.CreateTableOpt{
-			TableName:  dbutil.OfflineStreamCdcTableName(group.ID, revision),
-			EntityName: entity.Name,
-			Features:   features,
-			TableType:  types.TableStreamCdc,
+			TableName: dbutil.OfflineStreamCdcTableName(group.ID, revision),
+			Entity:    &entity,
+			Features:  features,
+			TableType: types.TableStreamCdc,
 		})
 		require.NoError(t, err)
 
diff --git a/internal/database/offline/test_impl/snapshot.go b/internal/database/offline/test_impl/snapshot.go
index 5b0cddd7..4d056371 100644
--- a/internal/database/offline/test_impl/snapshot.go
+++ b/internal/database/offline/test_impl/snapshot.go
@@ -47,7 +47,7 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 	})
 
 	err := store.Snapshot(ctx, offline.SnapshotOpt{
-		Group:        *group,
+		Group:        group,
 		Features:     features,
 		Revision:     2,
 		PrevRevision: 1,
diff --git a/internal/database/offline/test_impl/util.go b/internal/database/offline/test_impl/util.go
index d2747613..47396ed8 100644
--- a/internal/database/offline/test_impl/util.go
+++ b/internal/database/offline/test_impl/util.go
@@ -28,7 +28,7 @@ func buildTestSnapshotTable(
 		header = append(header, f.Name)
 	}
 	opt := offline.ImportOpt{
-		EntityName:        entity.Name,
+		Entity:            entity,
 		SnapshotTableName: snapshotTable,
 		Features:          features,
 		Header:            header,
diff --git a/internal/database/online/cassandra/store_test.go b/internal/database/online/cassandra/store_test.go
index 2608fe11..4fdc04a4 100644
--- a/internal/database/online/cassandra/store_test.go
+++ b/internal/database/online/cassandra/store_test.go
@@ -74,6 +74,10 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
 	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, destroyStore(KEYSPACE))
 }
 
+func TestPrepareStreamTable(t *testing.T) {
+	test_impl.TestPrepareStreamTable(t, prepareStore, destroyStore(KEYSPACE))
+}
+
 func TestPush(t *testing.T) {
 	test_impl.TestPush(t, prepareStore, destroyStore(KEYSPACE))
 }
@@ -81,7 +85,3 @@ func TestPush(t *testing.T) {
 func TestPing(t *testing.T) {
 	test_impl.TestPing(t, prepareStore, destroyStore(KEYSPACE))
 }
-
-func TestCreateTable(t *testing.T) {
-	test_impl.TestCreateTable(t, prepareStore, destroyStore(KEYSPACE))
-}
diff --git a/internal/database/online/dynamodb/store_test.go b/internal/database/online/dynamodb/store_test.go
index c1a6d878..f70a2ee5 100644
--- a/internal/database/online/dynamodb/store_test.go
+++ b/internal/database/online/dynamodb/store_test.go
@@ -65,10 +65,10 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
 	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, destroyStore(t))
 }
 
-func TestPush(t *testing.T) {
-	test_impl.TestPush(t, prepareStore, destroyStore(t))
+func TestPrepareStreamTable(t *testing.T) {
+	test_impl.TestPrepareStreamTable(t, prepareStore, destroyStore(t))
 }
 
-func TestCreateTable(t *testing.T) {
-	test_impl.TestCreateTable(t, prepareStore, destroyStore(t))
+func TestPush(t *testing.T) {
+	test_impl.TestPush(t, prepareStore, destroyStore(t))
 }
diff --git a/internal/database/online/mysql/store_test.go b/internal/database/online/mysql/store_test.go
index a6ae64ad..447eea4a 100644
--- a/internal/database/online/mysql/store_test.go
+++ b/internal/database/online/mysql/store_test.go
@@ -53,6 +53,10 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
 	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
 }
 
+func TestPrepareStreamTable(t *testing.T) {
+	test_impl.TestPrepareStreamTable(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
+}
+
 func TestPush(t *testing.T) {
 	test_impl.TestPush(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
 }
@@ -60,7 +64,3 @@ func TestPush(t *testing.T) {
 func TestPing(t *testing.T) {
 	test_impl.TestPing(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
 }
-
-func TestCreateTable(t *testing.T) {
-	test_impl.TestCreateTable(t, prepareStore, runtime_mysql.DestroyStore(DATABASE))
-}
diff --git a/internal/database/online/postgres/store_test.go b/internal/database/online/postgres/store_test.go
index 9789d29c..e7ffbea0 100644
--- a/internal/database/online/postgres/store_test.go
+++ b/internal/database/online/postgres/store_test.go
@@ -53,6 +53,10 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
 	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
 
+func TestPrepareStreamTable(t *testing.T) {
+	test_impl.TestPrepareStreamTable(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
+}
+
 func TestPush(t *testing.T) {
 	test_impl.TestPush(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
@@ -60,7 +64,3 @@ func TestPush(t *testing.T) {
 func TestPing(t *testing.T) {
 	test_impl.TestPing(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
 }
-
-func TestCreateTable(t *testing.T) {
-	test_impl.TestCreateTable(t, prepareStore, runtime_pg.DestroyStore(DATABASE))
-}
diff --git a/internal/database/online/redis/store_test.go b/internal/database/online/redis/store_test.go
index b79d7284..32f8cddd 100644
--- a/internal/database/online/redis/store_test.go
+++ b/internal/database/online/redis/store_test.go
@@ -51,6 +51,10 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
 	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, destroyStore)
 }
 
+func TestPrepareStreamTable(t *testing.T) {
+	test_impl.TestPrepareStreamTable(t, prepareStore, destroyStore)
+}
+
 func TestPush(t *testing.T) {
 	test_impl.TestPush(t, prepareStore, destroyStore)
 }
diff --git a/internal/database/online/sqlite/store_test.go b/internal/database/online/sqlite/store_test.go
index e7ad81f9..98613403 100644
--- a/internal/database/online/sqlite/store_test.go
+++ b/internal/database/online/sqlite/store_test.go
@@ -52,6 +52,10 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
 	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, destroyStore)
 }
 
+func TestPrepareStreamTable(t *testing.T) {
+	test_impl.TestPrepareStreamTable(t, prepareStore, destroyStore)
+}
+
 func TestPush(t *testing.T) {
 	test_impl.TestPush(t, prepareStore, destroyStore)
 }
@@ -59,7 +63,3 @@ func TestPush(t *testing.T) {
 func TestPing(t *testing.T) {
 	test_impl.TestPing(t, prepareStore, destroyStore)
 }
-
-func TestCreateTable(t *testing.T) {
-	test_impl.TestCreateTable(t, prepareStore, destroyStore)
-}
diff --git a/internal/database/online/test_impl/create_table.go b/internal/database/online/test_impl/create_table.go
deleted file mode 100644
index 70eeaca0..00000000
--- a/internal/database/online/test_impl/create_table.go
+++ /dev/null
@@ -1,32 +0,0 @@
-package test_impl
-
-import (
-	"testing"
-
-	"github.com/oom-ai/oomstore/internal/database/online"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestCreateTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
-	t.Cleanup(destroyStore)
-
-	ctx, store := prepareStore(t)
-	defer store.Close()
-
-	t.Run("create stream table", func(t *testing.T) {
-		err := store.CreateTable(ctx, online.CreateTableOpt{
-			EntityName: SampleStream.Entity.Name,
-			TableName:  "stream_online",
-			Features:   SampleStream.Features,
-		})
-		assert.NoError(t, err, "create stream table failed: %v", err)
-	})
-	t.Run("create batch table", func(t *testing.T) {
-		err := store.CreateTable(ctx, online.CreateTableOpt{
-			EntityName: SampleSmall.Entity.Name,
-			TableName:  "batch_online",
-			Features:   SampleSmall.Features,
-		})
-		assert.NoError(t, err, "create batch table failed: %v", err)
-	})
-}
diff --git a/internal/database/online/test_impl/get.go b/internal/database/online/test_impl/get.go
index 39f0e7c7..aeb1398f 100644
--- a/internal/database/online/test_impl/get.go
+++ b/internal/database/online/test_impl/get.go
@@ -3,82 +3,79 @@ package test_impl
 import (
 	"testing"
 
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-
 	"github.com/oom-ai/oomstore/internal/database/online"
 	"github.com/stretchr/testify/require"
 )
 
 func TestGetExisted(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
 	t.Cleanup(destroyStore)
+
+	s := &SampleSmall
 	ctx, store := prepareStore(t)
 	defer store.Close()
+	importSample(t, ctx, store, s)
 
-	for _, s := range []*Sample{&SampleSmall, &SampleStream} {
-		importSample(t, ctx, store, s)
-
-		for _, target := range s.Data {
-			opt := online.GetOpt{
-				EntityKey: target.EntityKey(),
-				Group:     s.Group,
-				Features:  s.Features,
-			}
-			if s.Group.Category == types.CategoryBatch {
-				opt.RevisionID = &s.Revision.ID
-			}
-			rs, err := store.Get(ctx, opt)
-			require.NoError(t, err)
+	for _, target := range s.Data {
+		opt := online.GetOpt{
+			Entity:     &s.Entity,
+			RevisionID: &s.Revision.ID,
+			Group:      s.Revision.Group,
+			Features:   s.Features,
+			EntityKey:  target.EntityKey(),
+		}
+		rs, err := store.Get(ctx, opt)
+		require.NoError(t, err)
 
-			for i, f := range s.Features {
-				compareFeatureValue(t, target.ValueAt(i), rs[f.FullName()], f.ValueType)
-			}
+		for i, f := range s.Features {
+			compareFeatureValue(t, target.ValueAt(i), rs[f.FullName()], f.ValueType)
 		}
 	}
 }
 
 func TestGetNotExistedEntityKey(t *testing.T, prepareStore PrepareStoreFn, destroystore DestroyStoreFn) {
 	t.Cleanup(destroystore)
+
+	s := &SampleSmall
 	ctx, store := prepareStore(t)
 	defer store.Close()
+	importSample(t, ctx, store, s)
 
-	for _, s := range []*Sample{&SampleSmall, &SampleStream} {
-		importSample(t, ctx, store, s)
-
-		rs, err := store.Get(ctx, online.GetOpt{
-			EntityKey:  "not-existed-key",
-			Group:      s.Group,
-			Features:   s.Features,
-			RevisionID: &s.Revision.ID,
-		})
-		require.NoError(t, err)
-		require.Equal(t, 0, len(rs), "actual: %+v", rs)
-	}
+	rs, err := store.Get(ctx, online.GetOpt{
+		Entity:     &s.Entity,
+		RevisionID: &s.Revision.ID,
+		Group:      s.Revision.Group,
+		Features:   s.Features,
+		EntityKey:  "not-existed-key",
+	})
+	require.NoError(t, err)
+	require.Equal(t, 0, len(rs), "actual: %+v", rs)
 }
 
 func TestMultiGet(t *testing.T, prepareStore PrepareStoreFn, destroystore DestroyStoreFn) {
 	t.Cleanup(destroystore)
+
+	s := &SampleSmall
 	ctx, store := prepareStore(t)
 	defer store.Close()
+	importSample(t, ctx, store, s)
 
-	for _, s := range []*Sample{&SampleSmall, &SampleStream} {
-		importSample(t, ctx, store, s)
-
-		keys := []string{"not-existed-key"}
-		for _, r := range s.Data {
-			keys = append(keys, r.EntityKey())
-		}
-		rs, err := store.MultiGet(ctx, online.MultiGetOpt{
-			EntityKeys: keys,
-			Group:      s.Group,
-			Features:   s.Features,
-			RevisionID: &s.Revision.ID,
-		})
-		require.NoError(t, err)
-		for _, record := range s.Data {
-			for i, feature := range s.Features {
-				compareFeatureValue(t, record.ValueAt(i), rs[record.EntityKey()][feature.FullName()], feature.ValueType)
-			}
+	keys := []string{"not-existed-key"}
+	for _, r := range s.Data {
+		keys = append(keys, r.EntityKey())
+	}
+	rs, err := store.MultiGet(ctx, online.MultiGetOpt{
+		Entity:     &s.Entity,
+		RevisionID: &s.Revision.ID,
+		Group:      s.Revision.Group,
+		Features:   s.Features,
+		EntityKeys: keys,
+	})
+	require.NoError(t, err)
+	for _, record := range s.Data {
+		for i, feature := range s.Features {
+			compareFeatureValue(t, record.ValueAt(i), rs[record.EntityKey()][feature.FullName()], feature.ValueType)
 		}
-		require.Equal(t, len(s.Data), len(rs), "actual: %+v", rs)
 	}
+
+	require.Equal(t, len(s.Data), len(rs), "actual: %+v", rs)
 }
diff --git a/internal/database/online/test_impl/purge.go b/internal/database/online/test_impl/purge.go
index 17117473..88bf4ebc 100644
--- a/internal/database/online/test_impl/purge.go
+++ b/internal/database/online/test_impl/purge.go
@@ -19,10 +19,11 @@ func TestPurgeRemovesSpecifiedRevision(t *testing.T, prepareStore PrepareStoreFn
 
 	for _, record := range SampleMedium.Data {
 		rs, err := store.Get(ctx, online.GetOpt{
+			Entity:     &SampleMedium.Entity,
+			RevisionID: &SampleMedium.Revision.ID,
+			Group:      SampleMedium.Revision.Group,
 			EntityKey:  record.EntityKey(),
-			Group:      *SampleMedium.Revision.Group,
 			Features:   SampleMedium.Features,
-			RevisionID: &SampleMedium.Revision.ID,
 		})
 		require.NoError(t, err)
 		require.Equal(t, 0, len(rs))
@@ -41,10 +42,11 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T, prepareStore PrepareStoreFn
 
 	for _, record := range SampleSmall.Data {
 		rs, err := store.Get(ctx, online.GetOpt{
+			Entity:     &SampleSmall.Entity,
+			RevisionID: &SampleSmall.Revision.ID,
+			Group:      SampleSmall.Revision.Group,
 			EntityKey:  record.EntityKey(),
-			Group:      *SampleSmall.Revision.Group,
 			Features:   SampleSmall.Features,
-			RevisionID: &SampleSmall.Revision.ID,
 		})
 		require.NoError(t, err)
 		for i, f := range SampleSmall.Features {
diff --git a/internal/database/online/test_impl/push.go b/internal/database/online/test_impl/push.go
deleted file mode 100644
index 7b7b5105..00000000
--- a/internal/database/online/test_impl/push.go
+++ /dev/null
@@ -1,84 +0,0 @@
-package test_impl
-
-import (
-	"testing"
-
-	"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/stretchr/testify/assert"
-)
-
-func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroyStoreFn) {
-	t.Cleanup(destoryStore)
-
-	ctx, store := prepareStore(t)
-	defer store.Close()
-
-	entity := SampleStream.Entity
-	group := SampleStream.Group
-	feature1 := SampleStream.Features[0]
-	feature2 := SampleStream.Features[1]
-
-	assert.NoError(t, store.CreateTable(ctx, online.CreateTableOpt{
-		EntityName: entity.Name,
-		TableName:  dbutil.OnlineStreamTableName(group.ID),
-		Features:   SampleStream.Features,
-	}))
-
-	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		EntityName:    entity.Name,
-		EntityKey:     "user1",
-		GroupID:       group.ID,
-		Features:      types.FeatureList{feature1},
-		FeatureValues: []interface{}{int64(1)},
-	}))
-	rs, err := store.Get(ctx, online.GetOpt{
-		EntityKey:  "user1",
-		Group:      group,
-		Features:   types.FeatureList{feature1},
-		RevisionID: nil,
-	})
-	assert.NoError(t, err)
-	assert.Equal(t, map[string]interface{}{
-		feature1.FullName(): int64(1),
-	}, rs)
-
-	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		EntityName:    entity.Name,
-		EntityKey:     "user1",
-		GroupID:       group.ID,
-		Features:      types.FeatureList{feature1},
-		FeatureValues: []interface{}{int64(2)},
-	}))
-	rs, err = store.Get(ctx, online.GetOpt{
-		EntityKey:  "user1",
-		Group:      group,
-		Features:   types.FeatureList{feature1},
-		RevisionID: nil,
-	})
-	assert.NoError(t, err)
-	assert.Equal(t, map[string]interface{}{
-		feature1.FullName(): int64(2),
-	}, rs)
-
-	assert.NoError(t, store.Push(ctx, online.PushOpt{
-		EntityName:    entity.Name,
-		EntityKey:     "user1",
-		GroupID:       group.ID,
-		Features:      types.FeatureList{feature1, feature2},
-		FeatureValues: []interface{}{int64(3), "post2"},
-	}))
-	rs, err = store.Get(ctx, online.GetOpt{
-		EntityKey:  "user1",
-		Group:      group,
-		Features:   types.FeatureList{feature1, feature2},
-		RevisionID: nil,
-	})
-	assert.NoError(t, err)
-	assert.Equal(t, map[string]interface{}{
-		feature1.FullName(): int64(3),
-		feature2.FullName(): "post2",
-	}, rs)
-}
diff --git a/internal/database/online/test_impl/stream.go b/internal/database/online/test_impl/stream.go
new file mode 100644
index 00000000..564c5a58
--- /dev/null
+++ b/internal/database/online/test_impl/stream.go
@@ -0,0 +1,186 @@
+package test_impl
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+)
+
+type StreamData struct {
+	entity   types.Entity
+	groups   types.GroupList
+	features types.FeatureList
+}
+
+var simpleStreamData StreamData
+
+func init() {
+	entity := types.Entity{
+		ID:   1,
+		Name: "user",
+	}
+
+	group1 := &types.Group{
+		ID:       1,
+		Name:     "user_clicks",
+		Category: types.CategoryStream,
+		EntityID: 1,
+		Entity:   &entity,
+	}
+
+	group2 := &types.Group{
+		ID:       2,
+		Name:     "user_reads",
+		Category: types.CategoryStream,
+		EntityID: 1,
+		Entity:   &entity,
+	}
+
+	simpleStreamData = StreamData{
+		entity: entity,
+		groups: types.GroupList{group1, group2},
+		features: types.FeatureList{
+			&types.Feature{
+				ID:        1,
+				Name:      "last_5_click_post",
+				ValueType: types.String,
+				GroupID:   1,
+				Group:     group1,
+			},
+			&types.Feature{
+				ID:        2,
+				Name:      "last_10_click_post",
+				ValueType: types.String,
+				GroupID:   1,
+				Group:     group1,
+			},
+			&types.Feature{
+				ID:        3,
+				Name:      "recent_5_read_post",
+				ValueType: types.String,
+				GroupID:   2,
+				Group:     group2,
+			},
+		},
+	}
+}
+
+func TestPrepareStreamTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
+	t.Cleanup(destroyStore)
+
+	ctx, store := prepareStore(t)
+	defer store.Close()
+
+	for _, group := range simpleStreamData.groups {
+		t.Run("create stream table", func(t *testing.T) {
+			err := store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+				Entity:  group.Entity,
+				GroupID: group.ID,
+			})
+			assert.NoError(t, err, "create stream table failed: %v", err)
+		})
+	}
+
+	for _, feature := range simpleStreamData.features {
+		t.Run("stream table add column", func(t *testing.T) {
+			err := store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+				Entity:  feature.Entity(),
+				GroupID: feature.GroupID,
+				Feature: feature,
+			})
+			assert.NoError(t, err, "stream table add column failed: %v", err)
+		})
+	}
+}
+
+func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroyStoreFn) {
+	t.Cleanup(destoryStore)
+
+	ctx, store := prepareStore(t)
+	defer store.Close()
+
+	var (
+		entity   = simpleStreamData.entity
+		group    = simpleStreamData.groups[0]
+		feature1 = simpleStreamData.features[0]
+		feature2 = simpleStreamData.features[1]
+	)
+
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+		Entity:  &entity,
+		GroupID: group.ID,
+	}))
+
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+		Entity:  &entity,
+		GroupID: group.ID,
+		Feature: feature1,
+	}))
+
+	assert.NoError(t, store.Push(ctx, online.PushOpt{
+		Entity:        &entity,
+		EntityKey:     "user1",
+		GroupID:       group.ID,
+		Features:      types.FeatureList{feature1},
+		FeatureValues: []interface{}{"post1"},
+	}))
+	rs, err := store.Get(ctx, online.GetOpt{
+		Entity:     &entity,
+		EntityKey:  "user1",
+		RevisionID: nil,
+		Group:      group,
+		Features:   types.FeatureList{feature1},
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, map[string]interface{}{
+		feature1.FullName(): "post1",
+	}, rs)
+
+	assert.NoError(t, store.Push(ctx, online.PushOpt{
+		Entity:        &entity,
+		EntityKey:     "user1",
+		GroupID:       group.ID,
+		Features:      types.FeatureList{feature1},
+		FeatureValues: []interface{}{"post2"},
+	}))
+	rs, err = store.Get(ctx, online.GetOpt{
+		Entity:     &entity,
+		EntityKey:  "user1",
+		RevisionID: nil,
+		Group:      group,
+		Features:   types.FeatureList{feature1},
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, map[string]interface{}{
+		feature1.FullName(): "post2",
+	}, rs)
+
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+		Entity:  &entity,
+		GroupID: group.ID,
+		Feature: feature2,
+	}))
+
+	assert.NoError(t, store.Push(ctx, online.PushOpt{
+		Entity:        &entity,
+		EntityKey:     "user1",
+		GroupID:       group.ID,
+		Features:      types.FeatureList{feature1, feature2},
+		FeatureValues: []interface{}{"post1", "post2"},
+	}))
+	rs, err = store.Get(ctx, online.GetOpt{
+		Entity:     &entity,
+		EntityKey:  "user1",
+		RevisionID: nil,
+		Group:      group,
+		Features:   types.FeatureList{feature1, feature2},
+	})
+	assert.NoError(t, err)
+	assert.Equal(t, map[string]interface{}{
+		feature1.FullName(): "post1",
+		feature2.FullName(): "post2",
+	}, rs)
+}
diff --git a/internal/database/online/test_impl/util.go b/internal/database/online/test_impl/util.go
index f6d3ab4c..c5135254 100644
--- a/internal/database/online/test_impl/util.go
+++ b/internal/database/online/test_impl/util.go
@@ -28,117 +28,85 @@ type Sample struct {
 
 var SampleSmall Sample
 var SampleMedium Sample
-var SampleStream Sample
 
 func init() {
 	rand.Seed(time.Now().UnixNano())
-	entity := types.Entity{ID: 1, Name: "user"}
-	group1 := types.Group{ID: 1, Category: types.CategoryBatch, Entity: &entity}
-	group2 := types.Group{ID: 2, Category: types.CategoryBatch, Entity: &entity}
-	group3 := types.Group{ID: 3, Name: "user_clicks", Category: types.CategoryStream, Entity: &entity}
-
-	SampleSmall = Sample{
-		Entity: entity,
-		Group:  group1,
-		Features: types.FeatureList{
-			&types.Feature{
-				ID:        1,
-				Name:      "age",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.Int64,
-			},
-			&types.Feature{
-				ID:        2,
-				Name:      "gender",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.String,
-			},
-			&types.Feature{
-				ID:        3,
-				Name:      "account",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.Float64,
+	entity := types.Entity{ID: 5, Name: "user"}
+	group := types.Group{ID: 1, Category: types.CategoryBatch, Entity: &entity}
+	{
+		SampleSmall = Sample{
+			Entity: entity,
+			Group:  group,
+			Features: types.FeatureList{
+				&types.Feature{
+					ID:        1,
+					Name:      "age",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.Int64,
+				},
+				&types.Feature{
+					ID:        2,
+					Name:      "gender",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.String,
+				},
+				&types.Feature{
+					ID:        3,
+					Name:      "account",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.Float64,
+				},
+				&types.Feature{
+					ID:        4,
+					Name:      "is_active",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.Bool,
+				},
+				&types.Feature{
+					ID:        5,
+					Name:      "register_time",
+					GroupID:   1,
+					Group:     &group,
+					ValueType: types.Time,
+				},
 			},
-			&types.Feature{
-				ID:        4,
-				Name:      "is_active",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.Bool,
+			Revision: types.Revision{
+				ID:      3,
+				GroupID: 1,
+				Group:   &group,
 			},
-			&types.Feature{
-				ID:        5,
-				Name:      "register_time",
-				GroupID:   1,
-				Group:     &group1,
-				ValueType: types.Time,
+			Data: []types.ExportRecord{
+				[]interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
+				[]interface{}{"3216", int64(29), nil, 2.0, false, time.Now()},
+				[]interface{}{"3217", int64(44), "M", 3.1, true, time.Now()},
 			},
-		},
-		Revision: types.Revision{
-			ID:      3,
-			GroupID: 1,
-			Group:   &group1,
-		},
-		Data: []types.ExportRecord{
-			[]interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
-			[]interface{}{"3216", int64(29), nil, 2.0, false, time.Now()},
-			[]interface{}{"3217", int64(44), "M", 3.1, true, time.Now()},
-		},
-	}
+		}
 
-	var data []types.ExportRecord
-	for i := 0; i < 100; i++ {
-		record := []interface{}{dbutil.RandString(10), rand.Float64()}
-		data = append(data, record)
 	}
-	SampleMedium = Sample{
-		Entity: entity,
-		Group:  group2,
-		Features: types.FeatureList{
+
+	{
+		features := types.FeatureList{
 			&types.Feature{
-				ID:        6,
+				ID:        2,
 				Name:      "charge",
 				GroupID:   2,
-				Group:     &group2,
+				Group:     &types.Group{ID: 2, Category: types.CategoryBatch},
 				ValueType: types.Float64,
 			},
-		},
-		Revision: types.Revision{
-			ID:      9,
-			GroupID: 2,
-			Group:   &group2,
-		},
-		Data: data,
-	}
+		}
 
-	SampleStream = Sample{
-		Entity: entity,
-		Group:  group3,
-		Features: types.FeatureList{
-			&types.Feature{
-				ID:        7,
-				Name:      "amount",
-				ValueType: types.Int64,
-				GroupID:   3,
-				Group:     &group3,
-			},
-			&types.Feature{
-				ID:        8,
-				Name:      "last_10_click_post",
-				ValueType: types.String,
-				GroupID:   3,
-				Group:     &group3,
-			},
-		},
-		Data: []types.ExportRecord{
-			[]interface{}{"3215", int64(1), "1,2,3,4"},
-			[]interface{}{"3216", int64(2), "2,3,4,5"},
-			[]interface{}{"3217", int64(3), "3,4,5,6"},
-			[]interface{}{"3218", int64(4), "4,5,6,7"},
-		},
+		revision := types.Revision{ID: 9, GroupID: 2, Group: &types.Group{ID: 2, Category: types.CategoryBatch}}
+		var data []types.ExportRecord
+
+		for i := 0; i < 100; i++ {
+			record := []interface{}{dbutil.RandString(10), rand.Float64()}
+			data = append(data, record)
+		}
+		SampleMedium = Sample{entity, group, features, revision, data}
 	}
 }
 
@@ -152,15 +120,12 @@ func importSample(t *testing.T, ctx context.Context, store online.Store, samples
 			}
 		}(sample)
 
-		opt := online.ImportOpt{
+		err := store.Import(ctx, online.ImportOpt{
 			Group:        sample.Group,
 			Features:     sample.Features,
+			Revision:     &sample.Revision,
 			ExportStream: stream,
-		}
-		if sample.Group.Category == types.CategoryBatch {
-			opt.RevisionID = &sample.Revision.ID
-		}
-		err := store.Import(ctx, opt)
+		})
 		require.NoError(t, err)
 	}
 }
diff --git a/internal/database/online/tikv/store_test.go b/internal/database/online/tikv/store_test.go
index 52f22d0f..52338f05 100644
--- a/internal/database/online/tikv/store_test.go
+++ b/internal/database/online/tikv/store_test.go
@@ -57,10 +57,10 @@ func TestPurgeRemovesSpecifiedRevision(t *testing.T) {
 	test_impl.TestPurgeRemovesSpecifiedRevision(t, prepareStore, destroyStore(t))
 }
 
-func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
-	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, destroyStore(t))
+func TestPust(t *testing.T) {
+	test_impl.TestPush(t, prepareStore, destroyStore(t))
 }
 
-func TestPush(t *testing.T) {
-	test_impl.TestPush(t, prepareStore, destroyStore(t))
+func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
+	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, destroyStore(t))
 }
diff --git a/oomagent/test/data/driver_stats.yaml b/oomagent/test/data/driver_stats.yaml
index f3c2aefe..48cfa4a8 100644
--- a/oomagent/test/data/driver_stats.yaml
+++ b/oomagent/test/data/driver_stats.yaml
@@ -1,9 +1,9 @@
+---
 kind: Entity
 name: driver
 groups:
   - name: driver_stats
     category: batch
-    description: ~
     features:
       - name: conv_rate
         value-type: float64
@@ -14,11 +14,5 @@ groups:
       - name: avg_daily_trips
         value-type: int64
         description: ~
-  - name: fake_stream
-    category: stream
-    snapshot-interval: 1s
     description: ~
-    features:
-      - name: f1
-        value-type: int64
 description: ~
diff --git a/oomagent/test/test_channel_join.sh b/oomagent/test/test_channel_join.sh
index bf496ec3..1981c764 100755
--- a/oomagent/test/test_channel_join.sh
+++ b/oomagent/test/test_channel_join.sh
@@ -11,17 +11,12 @@ for i in {1..5}; do
     import_sample driver_stats "./data/driver_stats_revision_$i.csv" "$i"
 done
 
-oomcli push --entity-key 1 --group fake_stream --features f1=10
-t1=${1:-$(perl -MTime::HiRes=time -E 'say int(time * 1000)')}
-oomcli snapshot fake_stream
-
-arg=$(cat <<-EOF
+arg='
 {
   "join_features": [
     "driver_stats.conv_rate",
     "driver_stats.acc_rate",
-    "driver_stats.avg_daily_trips",
-    "fake_stream.f1"
+    "driver_stats.avg_daily_trips"
   ],
   "entity_row": {
     "entity_key": "1",
@@ -40,14 +35,7 @@ arg=$(cat <<-EOF
     "unix_milli": 0
   }
 }
-{
-  "entity_row": {
-    "entity_key": "1",
-    "unix_milli": $t1
-  }
-}
-EOF
-)
+'
 
 actual=$(testgrpc ChannelJoin <<<"$arg")
 
@@ -59,20 +47,16 @@ expected_header='
     "unix_milli",
     "driver_stats.conv_rate",
     "driver_stats.acc_rate",
-    "driver_stats.avg_daily_trips",
-    "fake_stream.f1"
+    "driver_stats.avg_daily_trips"
 ]
 '
 assert_json_eq "$case" "$expected_header" "$actual_header"
 
 case="api returns correct joined rows"
 actual_rows=$(jq -c ".joinedRow" <<< "$actual" | sort)
-expected_rows=$(cat <<-EOF
-[{"string":"1"},{"int64":"3"},{"double":0.556},{"double":0.465},{"int64":"464"},{}]
-[{"string":"7"},{"int64":"1"},{"double":0.758},{"double":0.02},{"int64":"389"},{}]
-[{"string":"7"},{"int64":"0"},{},{},{},{}]
-[{"string":"1"},{"int64":"$t1"},{"double":0.146},{"double":0.031},{"int64":"286"},{"int64": "10"}]
-EOF
-)
-
+expected_rows='
+[{"string":"1"},{"int64":"3"},{"double":0.556},{"double":0.465},{"int64":"464"}]
+[{"string":"7"},{"int64":"1"},{"double":0.758},{"double":0.02},{"int64":"389"}]
+[{"string":"7"},{"int64":"0"},{},{},{}]
+'
 assert_json_eq "$case" "$(sort <<<"$expected_rows")" "$actual_rows"
diff --git a/oomcli/test/test_export.sh b/oomcli/test/test_export.sh
index e540126e..c5633544 100755
--- a/oomcli/test/test_export.sh
+++ b/oomcli/test/test_export.sh
@@ -52,15 +52,6 @@ oomcli_export_batch() {
     assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
 }
 
-oomcli_export_batch_and_stream_before_import() {
-    case="export batch and stream feature before import"
-    expected='user,student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,,,,"2,3,4,5,6",11
-'
-    actual=$(oomcli export --feature student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts --unix-milli $t2 -o csv)
-    assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
-}
-
 oomcli_export_batch_and_stream() {
     case="export batch and stream feature"
     expected='user,student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
@@ -80,7 +71,6 @@ main() {
     oomcli_export_push_feature
     oomcli_export_update_feature
     oomcli_export_batch
-    oomcli_export_batch_and_stream_before_import
     oomcli_export_batch_and_stream
 }
 
diff --git a/oomcli/test/test_join.sh b/oomcli/test/test_join.sh
index 9df1ea2d..2c45fddd 100755
--- a/oomcli/test/test_join.sh
+++ b/oomcli/test/test_join.sh
@@ -8,29 +8,24 @@ register_features
 # clean up the tmp file
 trap 'command rm -rf entity_rows.csv entity_rows_with_values.csv' EXIT INT TERM HUP
 
-# push stream feature
-t0=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
+# import sample data to offline store
+import_student_sample 80
+bt1=50
+bt2=100
 
+# push stream feature
+st0=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 oomcli push --entity-key 1 --group user-click --features last_5_click_posts=0,1,2,3,4 --features number_of_user_starred_posts=10 >> /dev/null 2>&1
-t1=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
-sleep 1
-
-# import sample data to offline store
-import_student_sample
-t2=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
+st1=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 sleep 1
-
 oomcli push --entity-key 1 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_starred_posts=10 >> /dev/null 2>&1
-t3=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
+st2=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 sleep 1
-
 oomcli push --entity-key 2 --group user-click --features last_5_click_posts=2,3,4,5,6 --features number_of_user_starred_posts=11 >> /dev/null 2>&1
-t4=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
+st3=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 sleep 1
-
 oomcli push --entity-key 3 --group user-click --features last_5_click_posts=3,4,5,6,7 --features number_of_user_starred_posts=12 >> /dev/null 2>&1
-t5=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
-
+st4=$(perl -MTime::HiRes -e 'printf("%.0f\n",Time::HiRes::time()*1000)')
 oomcli snapshot user-click
 
 oomcli_join_historical_feature() {
@@ -38,25 +33,17 @@ oomcli_join_historical_feature() {
 
 cat <<-EOF > entity_rows.csv
 entity_key,unix_milli
-1,$t0
-1,$t1
-1,$t2
-1,$t3
-2,$t0
-2,$t1
-2,$t2
-2,$t3
+1,$bt1
+2,$bt1
+1,$bt2
+2,$bt2
 EOF
     expected="
 entity_key,unix_milli,student.name,student.gender,student.age
-1,$t0,,,
-1,$t1,,,
-1,$t2,lian,m,18
-1,$t3,lian,m,18
-2,$t0,,,
-2,$t1,,,
-2,$t2,gao,m,20
-2,$t3,gao,m,20
+1,$bt1,,,
+2,$bt1,,,
+1,$bt2,lian,m,18
+2,$bt2,gao,m,20
     "
 
     actual=$(oomcli join \
@@ -76,26 +63,18 @@ oomcli_join_historical_feature_with_real_time_feature_values() {
 
 cat <<-EOF > entity_rows_with_values.csv
 entity_key,unix_milli,value_1,value_2
-1,$t0,1,2
-1,$t1,3,4
-1,$t2,5,6
-1,$t3,7,8
-2,$t0,1,2
-2,$t1,3,4
-2,$t2,5,6
-2,$t3,7,8
+1,$bt1,1,2
+2,$bt1,3,4
+1,$bt2,5,6
+2,$bt2,7,8
 EOF
 
     expected="
 entity_key,unix_milli,value_1,value_2,student.name,student.gender,student.age
-1,$t0,1,2,,,
-1,$t1,3,4,,,
-1,$t2,5,6,lian,m,18
-1,$t3,7,8,lian,m,18
-2,$t0,1,2,,,
-2,$t1,3,4,,,
-2,$t2,5,6,gao,m,20
-2,$t3,7,8,gao,m,20
+1,$bt1,1,2,,,
+2,$bt1,3,4,,,
+1,$bt2,5,6,lian,m,18
+2,$bt2,7,8,gao,m,20
 "
 
     actual=$(oomcli join \
@@ -116,46 +95,34 @@ oomcli_join_stream_historical_feature() {
 
     cat <<-EOF > entity_rows.csv
 entity_key,unix_milli
-1,$t0
-1,$t1
-1,$t2
-1,$t3
-1,$t4
-1,$t5
-2,$t0
-2,$t1
-2,$t2
-2,$t3
-2,$t4
-2,$t5
-3,$t0
-3,$t1
-3,$t2
-3,$t3
-3,$t4
-3,$t5
+1,$st0
+1,$st1
+1,$st2
+1,$st3
+2,$st1
+2,$st2
+2,$st3
+2,$st4
+3,$st1
+3,$st2
+3,$st3
+3,$st4
 EOF
 
     expected="
 entity_key,unix_milli,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,$t0,,
-1,$t1,\"0,1,2,3,4\",10
-1,$t2,\"0,1,2,3,4\",10
-1,$t3,\"1,2,3,4,5\",10
-1,$t4,\"1,2,3,4,5\",10
-1,$t5,\"1,2,3,4,5\",10
-2,$t0,,
-2,$t1,,
-2,$t2,,
-2,$t3,,
-2,$t4,\"2,3,4,5,6\",11
-2,$t5,\"2,3,4,5,6\",11
-3,$t0,,
-3,$t1,,
-3,$t2,,
-3,$t3,,
-3,$t4,,
-3,$t5,\"3,4,5,6,7\",12"
+1,$st0,,
+1,$st1,\"0,1,2,3,4\",10
+1,$st2,\"1,2,3,4,5\",10
+1,$st3,\"1,2,3,4,5\",10
+2,$st1,,
+2,$st2,,
+2,$st3,\"2,3,4,5,6\",11
+2,$st4,\"2,3,4,5,6\",11
+3,$st1,,
+3,$st2,,
+3,$st3,,
+3,$st4,\"3,4,5,6,7\",12"
 
     actual=$(oomcli join \
         --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts \
@@ -172,46 +139,34 @@ oomcli_join_stream_historical_feature_with_real_time_feature_value() {
 
     cat <<-EOF > entity_rows_with_values.csv
 entity_key,unix_milli,value_1,value_2
-1,$t0,1,2
-1,$t1,3,4
-1,$t2,5,6
-1,$t3,7,8
-1,$t4,9,10
-1,$t5,11,12
-2,$t0,1,2
-2,$t1,3,4
-2,$t2,5,6
-2,$t3,7,8
-2,$t4,9,10
-2,$t5,11,12
-3,$t0,1,2
-3,$t1,3,4
-3,$t2,5,6
-3,$t3,7,8
-3,$t4,9,10
-3,$t5,11,12
+1,$st0,1,2
+1,$st1,2,3
+1,$st2,3,4
+1,$st3,4,5
+2,$st1,1,2
+2,$st2,2,3
+2,$st3,3,4
+2,$st4,1,2
+3,$st1,2,3
+3,$st2,3,4
+3,$st3,4,5
+3,$st4,5,6
 EOF
 
     expected="
 entity_key,unix_milli,value_1,value_2,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,$t0,1,2,,
-1,$t1,3,4,\"0,1,2,3,4\",10
-1,$t2,5,6,\"0,1,2,3,4\",10
-1,$t3,7,8,\"1,2,3,4,5\",10
-1,$t4,9,10,\"1,2,3,4,5\",10
-1,$t5,11,12,\"1,2,3,4,5\",10
-2,$t0,1,2,,
-2,$t1,3,4,,
-2,$t2,5,6,,
-2,$t3,7,8,,
-2,$t4,9,10,\"2,3,4,5,6\",11
-2,$t5,11,12,\"2,3,4,5,6\",11
-3,$t0,1,2,,
-3,$t1,3,4,,
-3,$t2,5,6,,
-3,$t3,7,8,,
-3,$t4,9,10,,
-3,$t5,11,12,\"3,4,5,6,7\",12"
+1,$st0,1,2,,
+1,$st1,2,3,\"0,1,2,3,4\",10
+1,$st2,3,4,\"1,2,3,4,5\",10
+1,$st3,4,5,\"1,2,3,4,5\",10
+2,$st1,1,2,,
+2,$st2,2,3,,
+2,$st3,3,4,\"2,3,4,5,6\",11
+2,$st4,1,2,\"2,3,4,5,6\",11
+3,$st1,2,3,,
+3,$st2,3,4,,
+3,$st3,4,5,,
+3,$st4,5,6,\"3,4,5,6,7\",12"
 
     actual=$(oomcli join \
         --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts \
@@ -229,45 +184,33 @@ oomcli_join_batch_and_stream_historical_feature() {
 
    cat <<-EOF > entity_rows.csv
 entity_key,unix_milli
-1,$t0
-1,$t1
-1,$t2
-1,$t3
-1,$t4
-1,$t5
-2,$t0
-2,$t1
-2,$t2
-2,$t3
-2,$t4
-2,$t5
-3,$t0
-3,$t1
-3,$t2
-3,$t3
-3,$t4
-3,$t5
+1,$st0
+1,$st1
+1,$st2
+1,$st3
+2,$st1
+2,$st2
+2,$st3
+2,$st4
+3,$st1
+3,$st2
+3,$st3
+3,$st4
 EOF
     expected="
 entity_key,unix_milli,student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,$t0,,,,,
-1,$t1,,,,\"0,1,2,3,4\",10
-1,$t2,lian,m,18,\"0,1,2,3,4\",10
-1,$t3,lian,m,18,\"1,2,3,4,5\",10
-1,$t4,lian,m,18,\"1,2,3,4,5\",10
-1,$t5,lian,m,18,\"1,2,3,4,5\",10
-2,$t0,,,,,
-2,$t1,,,,,
-2,$t2,gao,m,20,,
-2,$t3,gao,m,20,,
-2,$t4,gao,m,20,\"2,3,4,5,6\",11
-2,$t5,gao,m,20,\"2,3,4,5,6\",11
-3,$t0,,,,,
-3,$t1,,,,,
-3,$t2,zhang,f,17,,
-3,$t3,zhang,f,17,,
-3,$t4,zhang,f,17,,
-3,$t5,zhang,f,17,\"3,4,5,6,7\",12"
+1,$st0,lian,m,18,,
+1,$st1,lian,m,18,\"0,1,2,3,4\",10
+1,$st2,lian,m,18,\"1,2,3,4,5\",10
+1,$st3,lian,m,18,\"1,2,3,4,5\",10
+2,$st1,gao,m,20,,
+2,$st2,gao,m,20,,
+2,$st3,gao,m,20,\"2,3,4,5,6\",11
+2,$st4,gao,m,20,\"2,3,4,5,6\",11
+3,$st1,zhang,f,17,,
+3,$st2,zhang,f,17,,
+3,$st3,zhang,f,17,,
+3,$st4,zhang,f,17,\"3,4,5,6,7\",12"
 
     actual=$(oomcli join \
         --feature student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts \
@@ -286,45 +229,33 @@ oomcli_join_batch_and_stream_historical_feature_with_real_time_feature_value() {
 
    cat <<-EOF > entity_rows.csv
 entity_key,unix_milli,value1,value2
-1,$t0,1,2
-1,$t1,3,4
-1,$t2,5,6
-1,$t3,7,8
-1,$t4,9,10
-1,$t5,11,12
-2,$t0,1,2
-2,$t1,3,4
-2,$t2,5,6
-2,$t3,7,8
-2,$t4,9,10
-2,$t5,11,12
-3,$t0,1,2
-3,$t1,3,4
-3,$t2,5,6
-3,$t3,7,8
-3,$t4,9,10
-3,$t5,11,12
+1,$st0,1,2
+1,$st1,2,3
+1,$st2,3,4
+1,$st3,4,5
+2,$st1,1,2
+2,$st2,2,3
+2,$st3,3,4
+2,$st4,4,5
+3,$st1,1,2
+3,$st2,2,3
+3,$st3,3,4
+3,$st4,4,5
 EOF
     expected="
 entity_key,unix_milli,value1,value2,student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,$t0,1,2,,,,,
-1,$t1,3,4,,,,\"0,1,2,3,4\",10
-1,$t2,5,6,lian,m,18,\"0,1,2,3,4\",10
-1,$t3,7,8,lian,m,18,\"1,2,3,4,5\",10
-1,$t4,9,10,lian,m,18,\"1,2,3,4,5\",10
-1,$t5,11,12,lian,m,18,\"1,2,3,4,5\",10
-2,$t0,1,2,,,,,
-2,$t1,3,4,,,,,
-2,$t2,5,6,gao,m,20,,
-2,$t3,7,8,gao,m,20,,
-2,$t4,9,10,gao,m,20,\"2,3,4,5,6\",11
-2,$t5,11,12,gao,m,20,\"2,3,4,5,6\",11
-3,$t0,1,2,,,,,
-3,$t1,3,4,,,,,
-3,$t2,5,6,zhang,f,17,,
-3,$t3,7,8,zhang,f,17,,
-3,$t4,9,10,zhang,f,17,,
-3,$t5,11,12,zhang,f,17,\"3,4,5,6,7\",12"
+1,$st0,1,2,lian,m,18,,
+1,$st1,2,3,lian,m,18,\"0,1,2,3,4\",10
+1,$st2,3,4,lian,m,18,\"1,2,3,4,5\",10
+1,$st3,4,5,lian,m,18,\"1,2,3,4,5\",10
+2,$st1,1,2,gao,m,20,,
+2,$st2,2,3,gao,m,20,,
+2,$st3,3,4,gao,m,20,\"2,3,4,5,6\",11
+2,$st4,4,5,gao,m,20,\"2,3,4,5,6\",11
+3,$st1,1,2,zhang,f,17,,
+3,$st2,2,3,zhang,f,17,,
+3,$st3,3,4,zhang,f,17,,
+3,$st4,4,5,zhang,f,17,\"3,4,5,6,7\",12"
 
     actual=$(oomcli join \
         --feature student.name,student.gender,student.age,user-click.last_5_click_posts,user-click.number_of_user_starred_posts \
diff --git a/pkg/oomstore/feature_test.go b/pkg/oomstore/feature_test.go
new file mode 100644
index 00000000..2b4c66f5
--- /dev/null
+++ b/pkg/oomstore/feature_test.go
@@ -0,0 +1,91 @@
+package oomstore_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/golang/mock/gomock"
+	"github.com/stretchr/testify/assert"
+
+	"github.com/oom-ai/oomstore/internal/database/metadata"
+	"github.com/oom-ai/oomstore/internal/database/metadata/mock_metadata"
+	"github.com/oom-ai/oomstore/internal/database/offline/mock_offline"
+	"github.com/oom-ai/oomstore/internal/database/online"
+	"github.com/oom-ai/oomstore/internal/database/online/mock_online"
+	"github.com/oom-ai/oomstore/pkg/oomstore"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+)
+
+func TestCreateFeature(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	defer ctrl.Finish()
+	ctx := context.Background()
+
+	onlineStore := mock_online.NewMockStore(ctrl)
+	offlineStore := mock_offline.NewMockStore(ctrl)
+	metadataStore := mock_metadata.NewMockStore(ctrl)
+	store := oomstore.TEST__New(onlineStore, offlineStore, metadataStore)
+
+	testCases := []struct {
+		description string
+		opt         types.CreateFeatureOpt
+		valueType   types.ValueType
+		group       types.Group
+	}{
+		{
+			description: "create batch feature, succeed",
+			opt: types.CreateFeatureOpt{
+				FeatureName: "model",
+				GroupName:   "device_info",
+				ValueType:   types.String,
+			},
+			valueType: types.String,
+			group: types.Group{
+				ID:       1,
+				Name:     "device_info",
+				Category: types.CategoryBatch,
+			},
+		},
+		{
+			description: "create stream feature, succeed",
+			opt: types.CreateFeatureOpt{
+				FeatureName: "model",
+				GroupName:   "device_info",
+				ValueType:   types.Int64,
+			},
+			valueType: types.Int64,
+			group: types.Group{
+				ID:       1,
+				Name:     "device_info",
+				Category: types.CategoryStream,
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.description, func(t *testing.T) {
+			metadataStore.EXPECT().Refresh().Return(nil).AnyTimes()
+			metadataStore.EXPECT().GetGroupByName(ctx, tc.opt.GroupName).Return(&tc.group, nil)
+
+			metadataOpt := metadata.CreateFeatureOpt{
+				FeatureName: tc.opt.FeatureName,
+				GroupID:     tc.group.ID,
+				ValueType:   tc.valueType,
+				Description: tc.opt.Description,
+			}
+			metadataStore.EXPECT().CreateFeature(ctx, metadataOpt).Return(1, nil)
+
+			if tc.group.Category == types.CategoryStream {
+				var feature types.Feature
+				metadataStore.EXPECT().GetFeature(ctx, 1).Return(&feature, nil)
+				onlineStore.EXPECT().PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+					Entity:  tc.group.Entity,
+					GroupID: tc.group.ID,
+					Feature: &feature,
+				})
+			}
+			_, err := store.CreateFeature(ctx, tc.opt)
+			assert.NoError(t, err)
+		})
+	}
+}
diff --git a/pkg/oomstore/online_query_test.go b/pkg/oomstore/online_query_test.go
index b47c5cf6..534f2e5c 100644
--- a/pkg/oomstore/online_query_test.go
+++ b/pkg/oomstore/online_query_test.go
@@ -5,8 +5,6 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/oom-ai/oomstore/internal/database/metadata"
-
 	"github.com/golang/mock/gomock"
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/metadata/mock_metadata"
@@ -82,9 +80,7 @@ func TestOnlineGet(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.description, func(t *testing.T) {
-			metadataStore.EXPECT().ListCachedFeature(gomock.Any(), metadata.ListCachedFeatureOpt{
-				FullNames: &tc.opt.FeatureNames,
-			}).Return(tc.features)
+			metadataStore.EXPECT().ListCachedFeature(gomock.Any(), &tc.opt.FeatureNames).Return(tc.features)
 			if tc.entityName != nil {
 				onlineStore.EXPECT().Get(gomock.Any(), gomock.Any()).Return(dbutil.RowMap{
 					"price": int64(100),
@@ -178,9 +174,7 @@ func TestOnlineMultiGet(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.description, func(t *testing.T) {
-			metadataStore.EXPECT().ListCachedFeature(gomock.Any(), metadata.ListCachedFeatureOpt{
-				FullNames: &tc.opt.FeatureNames,
-			}).Return(tc.features)
+			metadataStore.EXPECT().ListCachedFeature(gomock.Any(), &tc.opt.FeatureNames).Return(tc.features)
 			if tc.entityName != nil {
 				onlineStore.EXPECT().MultiGet(gomock.Any(), gomock.Any()).Return(map[string]dbutil.RowMap{
 					"1234": {