Repository: oom-ai/oomstore
Pull request #1085 (2022-01-26 02:29:09)

feat: only support py3.7+ to match pyo3
https://github.com/PyO3/pyo3#usage


Patch:
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/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/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index 3a46ee08..4ef995e2 100644
--- a/internal/database/online/cassandra/get.go
+++ b/internal/database/online/cassandra/get.go
@@ -10,15 +10,16 @@ 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 = ?`,
@@ -51,9 +52,9 @@ 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)`,
diff --git a/internal/database/online/cassandra/import.go b/internal/database/online/cassandra/import.go
index 7da7241f..4004f089 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.RevisionID)
 	} 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.Name, 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..e8cdfff5 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.EntityName, 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..e737c6c8 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)
@@ -50,9 +51,9 @@ 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
diff --git a/internal/database/online/dynamodb/import.go b/internal/database/online/dynamodb/import.go
index 8de93833..c6ceb570 100644
--- a/internal/database/online/dynamodb/import.go
+++ b/internal/database/online/dynamodb/import.go
@@ -11,6 +11,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,9 +24,9 @@ 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.RevisionID)
 	} else {
-		tableName = dbutil.OnlineStreamTableName(opt.Group.ID)
+		tableName = sqlutil.OnlineStreamTableName(opt.Group.ID)
 		_, err := db.DeleteTable(ctx, &dynamodb.DeleteTableInput{
 			TableName: aws.String(tableName),
 		})
@@ -36,10 +37,27 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 		}
 	}
 	// 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 +80,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..be2995c8
--- /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.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),
+		},
+	})
+	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.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/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..23a9cab7 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.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
 }
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index 9b709388..cbeb7490 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)
 
@@ -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.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
 }
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..27b93ca3 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.SqlxPrepareStreamTable(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..4f3e4430 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -16,9 +16,9 @@ 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
@@ -45,9 +45,9 @@ 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
diff --git a/internal/database/online/sqlutil/import.go b/internal/database/online/sqlutil/import.go
index ba2c9ce7..70153ed3 100644
--- a/internal/database/online/sqlutil/import.go
+++ b/internal/database/online/sqlutil/import.go
@@ -18,7 +18,7 @@ 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.RevisionID)
 	}
 	entity := opt.Group.Entity
 	columns := append([]string{entity.Name}, opt.Features.Names()...)
@@ -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/util.go b/internal/database/online/sqlutil/util.go
new file mode 100644
index 00000000..5cd68faf
--- /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, entityName string, backend types.BackendType) (string, error) {
+	var (
+		entityFormat string
+	)
+	switch backend {
+	case types.BackendMySQL:
+		entityFormat = fmt.Sprintf("`%s` VARCHAR(255) PRIMARY KEY", entityName)
+	case types.BackendPostgres, types.BackendCassandra, types.BackendSQLite:
+		entityFormat = fmt.Sprintf(`"%s" TEXT PRIMARY KEY`, entityName)
+	default:
+		return "", errdefs.InvalidAttribute(errdefs.Errorf("backend %s not support", backend))
+	}
+
+	schema := fmt.Sprintf("CREATE TABLE %s ( %s )", tableName, entityFormat)
+	return schema, nil
+}
+
+func SqlxPrepareStreamTable(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.EntityName, 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/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..1997136b 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -34,8 +34,11 @@ type PushOpt struct {
 	FeatureValues []interface{}
 }
 
-type CreateTableOpt struct {
+type PrepareStreamTableOpt struct {
 	EntityName string
-	TableName  string
-	Features   types.FeatureList
+	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/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/oomstore/apply.go b/pkg/oomstore/apply.go
index df20bf7e..709b5197 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{
+					EntityName: entity.Name,
+					GroupID:    id,
+				})
+			}, nil
+		}
 		return nil, nil
 	}
 
@@ -175,7 +181,7 @@ func (s *OomStore) applyFeature(ctx context.Context, tx metadata.DBStore, newFea
 		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 +192,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{
+				return s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
 					EntityName: group.Entity.Name,
-					TableName:  dbutil.OnlineStreamTableName(group.ID),
-					Features:   features,
+					GroupID:    group.ID,
+					Feature:    feature,
 				})
 			}, nil
 		}
diff --git a/pkg/oomstore/feature.go b/pkg/oomstore/feature.go
index 6f1fd0d8..098a8524 100644
--- a/pkg/oomstore/feature.go
+++ b/pkg/oomstore/feature.go
@@ -3,8 +3,6 @@ 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"
@@ -12,12 +10,12 @@ import (
 	"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 +24,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 +56,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,7 +71,7 @@ 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 {
@@ -99,16 +97,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{
+		if err = s.online.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
 			EntityName: group.Entity.Name,
-			TableName:  dbutil.OnlineStreamTableName(group.ID),
-			Features:   features,
+			GroupID:    group.ID,
+			Feature:    feature,
 		}); err != nil {
 			return 0, err
 		}
diff --git a/pkg/oomstore/group.go b/pkg/oomstore/group.go
index 13f04467..a2939379 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{
+			EntityName: entity.Name,
+			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/sdk/python_v2/pyproject.toml b/sdk/python_v2/pyproject.toml
index 99718722..dbf7b424 100644
--- a/sdk/python_v2/pyproject.toml
+++ b/sdk/python_v2/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "oomclient"
-requires-python = ">=3.6"
+requires-python = ">=3.7"
 classifiers = [
     "Programming Language :: Rust",
     "Programming Language :: Python :: Implementation :: CPython",


Test patch:
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
index 70eeaca0..5a157ea3 100644
--- a/internal/database/online/test_impl/create_table.go
+++ b/internal/database/online/test_impl/create_table.go
@@ -7,26 +7,29 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-func TestCreateTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
+func TestPrepareStreamTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
 	t.Cleanup(destroyStore)
 
 	ctx, store := prepareStore(t)
 	defer store.Close()
 
+	group := SampleStream.Group
 	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,
+		err := store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+			EntityName: group.Entity.Name,
+			GroupID:    group.ID,
 		})
 		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,
+
+	for _, feature := range SampleStream.Features {
+		t.Run("stream table add column", func(t *testing.T) {
+			err := store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+				EntityName: feature.Entity().Name,
+				GroupID:    feature.GroupID,
+				Feature:    feature,
+			})
+			assert.NoError(t, err, "stream table add column failed: %v", err)
 		})
-		assert.NoError(t, err, "create batch table failed: %v", err)
-	})
+	}
 }
diff --git a/internal/database/online/test_impl/push.go b/internal/database/online/test_impl/push.go
index 7b7b5105..60930100 100644
--- a/internal/database/online/test_impl/push.go
+++ b/internal/database/online/test_impl/push.go
@@ -3,8 +3,6 @@ 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"
@@ -21,10 +19,15 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 	feature1 := SampleStream.Features[0]
 	feature2 := SampleStream.Features[1]
 
-	assert.NoError(t, store.CreateTable(ctx, online.CreateTableOpt{
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+		EntityName: entity.Name,
+		GroupID:    group.ID,
+	}))
+
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
 		EntityName: entity.Name,
-		TableName:  dbutil.OnlineStreamTableName(group.ID),
-		Features:   SampleStream.Features,
+		GroupID:    group.ID,
+		Feature:    feature1,
 	}))
 
 	assert.NoError(t, store.Push(ctx, online.PushOpt{
@@ -63,6 +66,12 @@ func TestPush(t *testing.T, prepareStore PrepareStoreFn, destoryStore DestroySto
 		feature1.FullName(): int64(2),
 	}, rs)
 
+	assert.NoError(t, store.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+		EntityName: entity.Name,
+		GroupID:    group.ID,
+		Feature:    feature2,
+	}))
+
 	assert.NoError(t, store.Push(ctx, online.PushOpt{
 		EntityName:    entity.Name,
 		EntityKey:     "user1",
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"