Repository: oom-ai/oomstore
Pull request #1094 (2022-01-26 07:55:06)

fix(oomagent/join): fix deadlock
resolve: #1092 


Patch:
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/offline/bigquery/join.go b/internal/database/offline/bigquery/join.go
index b814e86c..bd6debe3 100644
--- a/internal/database/offline/bigquery/join.go
+++ b/internal/database/offline/bigquery/join.go
@@ -24,16 +24,11 @@ func (db *DB) Join(ctx context.Context, opt offline.JoinOpt) (*types.JoinResult,
 	doJoinOpt := sqlutil.DoJoinOpt{
 		JoinOpt:             opt,
 		QueryResults:        bigqueryQueryResults,
-		QueryTableTimeRange: bigqueryQueryTimeRange,
 		ReadJoinResultQuery: READ_JOIN_RESULT_QUERY,
 	}
 	return sqlutil.DoJoin(ctx, dbOpt, doJoinOpt)
 }
 
-func bigqueryQueryTimeRange(ctx context.Context, dbOpt dbutil.DBOpt, tableName string) (*types.DataTableTimeRange, error) {
-	return getTableTimeRange(ctx, dbOpt, tableName)
-}
-
 func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, header dbutil.ColumnList, dropTableNames []string, backendType types.BackendType) (*types.JoinResult, error) {
 	rows, err := dbOpt.BigQueryDB.Query(query).Read(ctx)
 	if err != nil {
diff --git a/internal/database/offline/bigquery/store.go b/internal/database/offline/bigquery/store.go
index 64c6feba..515a3225 100644
--- a/internal/database/offline/bigquery/store.go
+++ b/internal/database/offline/bigquery/store.go
@@ -2,9 +2,12 @@ package bigquery
 
 import (
 	"context"
+	"fmt"
 
 	"cloud.google.com/go/bigquery"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
+	"github.com/spf13/cast"
+	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
@@ -41,6 +44,54 @@ func (db *DB) Ping(ctx context.Context) error {
 	return errdefs.WithStack(err)
 }
 
+func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
+	q := fmt.Sprintf(`SELECT column_name, data_type FROM %s.INFORMATION_SCHEMA.COLUMNS WHERE table_name = "%s"`, db.datasetID, opt.TableName)
+	rows, err := db.Query(q).Read(ctx)
+	if err != nil {
+		return nil, errdefs.WithStack(err)
+	}
+	var schema types.DataTableSchema
+	for {
+		recordMap := make(map[string]bigquery.Value)
+		err = rows.Next(&recordMap)
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return nil, errdefs.WithStack(err)
+		}
+		valueType, err := dbutil.ValueType(Backend, cast.ToString(recordMap["data_type"]))
+		if err != nil {
+			return nil, err
+		}
+		schema.Fields = append(schema.Fields, types.DataTableFieldSchema{
+			Name:      cast.ToString(recordMap["column_name"]),
+			ValueType: valueType,
+		})
+	}
+	if opt.CheckTimeRange {
+		q = fmt.Sprintf(`
+		SELECT
+			MIN(%s) AS %s,
+			MAX(%s) AS %s
+		FROM %s.%s`, "unix_milli", "min_unix_milli", "unix_milli", "max_unix_milli", db.datasetID, opt.TableName)
+		rows, err = db.Query(q).Read(ctx)
+		if err != nil {
+			return nil, errdefs.WithStack(err)
+		}
+		recordMap := make(map[string]bigquery.Value)
+		err = rows.Next(&recordMap)
+		if err != nil {
+			return nil, errdefs.WithStack(err)
+		}
+		schema.TimeRange = types.DataTableTimeRange{
+			MinUnixMilli: cast.ToInt64(recordMap["min_unix_milli"]),
+			MaxUnixMilli: cast.ToInt64(recordMap["max_unix_milli"]),
+		}
+	}
+	return &schema, nil
+}
+
 func (db *DB) Snapshot(ctx context.Context, opt offline.SnapshotOpt) error {
 	dbOpt := dbutil.DBOpt{
 		Backend:    Backend,
diff --git a/internal/database/offline/bigquery/table_schema.go b/internal/database/offline/bigquery/table_schema.go
deleted file mode 100644
index ab521c4c..00000000
--- a/internal/database/offline/bigquery/table_schema.go
+++ /dev/null
@@ -1,78 +0,0 @@
-package bigquery
-
-import (
-	"context"
-	"fmt"
-
-	"cloud.google.com/go/bigquery"
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
-	"github.com/oom-ai/oomstore/internal/database/offline"
-	"github.com/oom-ai/oomstore/pkg/errdefs"
-	"github.com/oom-ai/oomstore/pkg/oomstore/types"
-	"github.com/spf13/cast"
-	"google.golang.org/api/iterator"
-)
-
-func (db *DB) TableSchema(ctx context.Context, opt offline.TableSchemaOpt) (*types.DataTableSchema, error) {
-	q := fmt.Sprintf(`SELECT column_name, data_type FROM %s.INFORMATION_SCHEMA.COLUMNS WHERE table_name = "%s"`, db.datasetID, opt.TableName)
-	rows, err := db.Query(q).Read(ctx)
-	if err != nil {
-		return nil, errdefs.WithStack(err)
-	}
-	var schema types.DataTableSchema
-	for {
-		recordMap := make(map[string]bigquery.Value)
-		err = rows.Next(&recordMap)
-		if err == iterator.Done {
-			break
-		}
-		if err != nil {
-			return nil, errdefs.WithStack(err)
-		}
-		valueType, err := dbutil.ValueType(Backend, cast.ToString(recordMap["data_type"]))
-		if err != nil {
-			return nil, err
-		}
-		schema.Fields = append(schema.Fields, types.DataTableFieldSchema{
-			Name:      cast.ToString(recordMap["column_name"]),
-			ValueType: valueType,
-		})
-	}
-	if opt.CheckTimeRange {
-		dbOpt := dbutil.DBOpt{
-			Backend:    Backend,
-			BigQueryDB: db.Client,
-			DatasetID:  &db.datasetID,
-		}
-		timeRange, err := getTableTimeRange(ctx, dbOpt, opt.TableName)
-		if err != nil {
-			return nil, err
-		}
-		schema.TimeRange = *timeRange
-	}
-	return &schema, nil
-}
-
-func getTableTimeRange(ctx context.Context, dbOpt dbutil.DBOpt, tableName string) (*types.DataTableTimeRange, error) {
-	q := fmt.Sprintf(`
-		SELECT
-			MIN(unix_milli) AS min_unix_milli,
-			MAX(unix_milli) AS max_unix_milli
-		FROM %s.%s`, *dbOpt.DatasetID, tableName)
-	rows, err := dbOpt.BigQueryDB.Query(q).Read(ctx)
-	if err != nil {
-		return nil, errdefs.WithStack(err)
-	}
-	recordMap := make(map[string]bigquery.Value)
-	err = rows.Next(&recordMap)
-	if err != nil {
-		return nil, errdefs.WithStack(err)
-	}
-	min := cast.ToInt64(recordMap["min_unix_milli"])
-	max := cast.ToInt64(recordMap["max_unix_milli"])
-
-	return &types.DataTableTimeRange{
-		MinUnixMilli: &min,
-		MaxUnixMilli: &max,
-	}, nil
-}
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index bec13239..8f3a70a3 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -13,7 +13,6 @@ import (
 )
 
 type QueryResults func(ctx context.Context, dbOpt dbutil.DBOpt, query string, header dbutil.ColumnList, dropTableNames []string, backendType types.BackendType) (*types.JoinResult, error)
-type QueryTableTimeRange func(ctx context.Context, dbOpt dbutil.DBOpt, tableName string) (*types.DataTableTimeRange, error)
 
 func Join(ctx context.Context, db *sqlx.DB, opt offline.JoinOpt, backendType types.BackendType) (*types.JoinResult, error) {
 	dbOpt := dbutil.DBOpt{
@@ -23,7 +22,6 @@ func Join(ctx context.Context, db *sqlx.DB, opt offline.JoinOpt, backendType typ
 	doJoinOpt := DoJoinOpt{
 		JoinOpt:             opt,
 		QueryResults:        sqlxQueryResults,
-		QueryTableTimeRange: sqlxQueryTableTimeRange,
 		ReadJoinResultQuery: READ_JOIN_RESULT_QUERY,
 	}
 	return DoJoin(ctx, dbOpt, doJoinOpt)
@@ -32,7 +30,6 @@ func Join(ctx context.Context, db *sqlx.DB, opt offline.JoinOpt, backendType typ
 type DoJoinOpt struct {
 	offline.JoinOpt
 	QueryResults        QueryResults
-	QueryTableTimeRange QueryTableTimeRange
 	ReadJoinResultQuery string
 }
 
@@ -58,15 +55,6 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 	if err != nil {
 		return nil, err
 	}
-	timeRange, err := opt.QueryTableTimeRange(ctx, dbOpt, entityRowsTableName)
-	if err != nil {
-		return nil, err
-	}
-	if timeRange.MinUnixMilli == nil || timeRange.MaxUnixMilli == nil {
-		data := make(chan []interface{})
-		defer close(data)
-		return &types.JoinResult{Data: data}, nil
-	}
 
 	// Step 2: process features by group, insert result to table joined
 	tableNames := make([]string, 0)
@@ -91,7 +79,6 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 			Features:            featureList,
 			RevisionRanges:      revisionRanges,
 			EntityRowsTableName: entityRowsTableName,
-			TimeRange:           *timeRange,
 		})
 		if err != nil {
 			return nil, err
@@ -125,7 +112,6 @@ type joinOneGroupOpt struct {
 	EntityName          string
 	EntityRowsTableName string
 	ValueNames          []string
-	TimeRange           types.DataTableTimeRange
 }
 
 func joinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt joinOneGroupOpt) ([]string, error) {
@@ -142,9 +128,6 @@ func joinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt joinOneGroupOpt)
 	// Step 2: iterate each table range, join entity_rows table and each data tables
 	columns := append(opt.ValueNames, opt.Features.Names()...)
 	for _, r := range opt.RevisionRanges {
-		if *opt.TimeRange.MaxUnixMilli < r.MinRevision || *opt.TimeRange.MinUnixMilli > r.MaxRevision {
-			continue
-		}
 		query, err := buildJoinQuery(joinQueryParams{
 			TableName:           snapshotJoinedTableName,
 			EntityName:          opt.EntityName,
@@ -339,7 +322,3 @@ func sqlxQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, hea
 		Data:   data,
 	}, dropErr
 }
-
-func sqlxQueryTableTimeRange(ctx context.Context, dbOpt dbutil.DBOpt, tableName string) (*types.DataTableTimeRange, error) {
-	return getCdcTimeRange(ctx, dbOpt.SqlxDB, tableName, dbOpt.Backend)
-}
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/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/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..acb1b564 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),
 		})
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/stream.go b/internal/database/online/dynamodb/stream.go
new file mode 100644
index 00000000..9afff171
--- /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.Client.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..90768495 100644
--- a/internal/database/online/mock_online/store.go
+++ b/internal/database/online/mock_online/store.go
@@ -122,6 +122,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..6ceef4f8 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,6 +65,10 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	return errdefs.WithStack(err)
 }
 
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	return sqlutil.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
+}
+
 func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
 	dbOpt := dbutil.DBOpt{
 		Backend: Backend,
diff --git a/internal/database/online/postgres/store.go b/internal/database/online/postgres/store.go
index 9b709388..4ecc3f9b 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,6 +66,10 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	return errdefs.WithStack(err)
 }
 
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	return sqlutil.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
+}
+
 func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
 	dbOpt := dbutil.DBOpt{
 		Backend: Backend,
diff --git a/internal/database/online/redis/store.go b/internal/database/online/redis/store.go
index 5a588b9f..b3e78ef5 100644
--- a/internal/database/online/redis/store.go
+++ b/internal/database/online/redis/store.go
@@ -34,6 +34,10 @@ func Open(opt *types.RedisOpt) *DB {
 	return &DB{redis.NewClient(&redisOpt)}
 }
 
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	return nil
+}
+
 func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
 	return nil
 }
diff --git a/internal/database/online/sqlite/store.go b/internal/database/online/sqlite/store.go
index 7ca8784d..b82f20c6 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,6 +63,10 @@ func (db *DB) Push(ctx context.Context, opt online.PushOpt) error {
 	return errdefs.WithStack(err)
 }
 
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	return sqlutil.SqlxPrepareStreamTable(ctx, db.DB, opt, Backend)
+}
+
 func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
 	dbOpt := dbutil.DBOpt{
 		Backend: Backend,
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..5eed8545 100644
--- a/internal/database/online/store.go
+++ b/internal/database/online/store.go
@@ -20,6 +20,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..cc693d69 100644
--- a/internal/database/online/tikv/store.go
+++ b/internal/database/online/tikv/store.go
@@ -43,6 +43,10 @@ func (db *DB) Ping(ctx context.Context) error {
 	return errdefs.WithStack(err)
 }
 
+func (db *DB) PrepareStreamTable(ctx context.Context, opt online.PrepareStreamTableOpt) error {
+	return nil
+}
+
 func (db *DB) CreateTable(ctx context.Context, opt online.CreateTableOpt) error {
 	return nil
 }
diff --git a/internal/database/online/types.go b/internal/database/online/types.go
index c9c65a25..bedef8e0 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -34,6 +34,15 @@ type PushOpt struct {
 	FeatureValues []interface{}
 }
 
+type PrepareStreamTableOpt struct {
+	EntityName string
+	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
+}
+
 type CreateTableOpt struct {
 	EntityName string
 	TableName  string
diff --git a/oomagent/server.go b/oomagent/server.go
index e19b64ab..54583ce2 100644
--- a/oomagent/server.go
+++ b/oomagent/server.go
@@ -199,40 +199,44 @@ func (s *server) ChannelJoin(stream codegen.OomAgent_ChannelJoinServer) error {
 	var globalErr error
 
 	// This channel indicates when the the ChannelJoin oomstore operation is finished, whether succeeded or failed.
-	done := make(chan struct{})
+	done := make(chan struct{}, 1)
+
 	// This channel receives requests from the client.
-	entityRows := make(chan types.EntityRow)
+	entityRows := make(chan types.EntityRow, 1)
 
-	// This goroutine runs the join operation, and send whatever joined as the response
 	go func() {
-		joinResult, err := s.oomstore.ChannelJoin(context.Background(), types.ChannelJoinOpt{
-			JoinFeatureNames:    firstReq.JoinFeatures,
-			EntityRows:          entityRows,
-			ExistedFeatureNames: firstReq.ExistedFeatures,
-		})
-		if err != nil {
-			globalErr = err
-		} else {
-			header := joinResult.Header
-			for row := range joinResult.Data {
-				joinedRow, err := convertJoinedRow(row)
-				if err != nil {
-					globalErr = err
-					break
-				}
-				resp := &codegen.ChannelJoinResponse{
-					Header:    header,
-					JoinedRow: joinedRow,
-				}
-				if err = stream.Send(resp); err != nil {
-					globalErr = err
-					break
+		defer func() {
+			close(entityRows)
+		}()
+
+		for {
+			req, err := stream.Recv()
+			if err == io.EOF {
+				return
+			}
+			if err != nil {
+				globalErr = err
+				return
+			}
+			if globalErr != nil {
+				return
+			}
+			if req.GetEntityRow() == nil {
+				globalErr = errdefs.Errorf("cannot process nil entity row")
+				return
+			}
+
+			select {
+			case <-done:
+				return
+			default:
+				entityRows <- types.EntityRow{
+					EntityKey: req.EntityRow.EntityKey,
+					UnixMilli: req.EntityRow.UnixMilli,
+					Values:    req.EntityRow.Values,
 				}
-				// Only need to send header upon the first response
-				header = nil
 			}
 		}
-		done <- struct{}{}
 	}()
 
 	// DO NOT move it before the goroutine starts,
@@ -243,33 +247,37 @@ func (s *server) ChannelJoin(stream codegen.OomAgent_ChannelJoinServer) error {
 		Values:    firstReq.EntityRow.Values,
 	}
 
-	for {
-		req, err := stream.Recv()
-		if err == io.EOF {
-			break
-		}
-		if err != nil {
-			globalErr = err
-			break
-		}
-		if globalErr != nil {
-			break
-		}
-		if req.GetEntityRow() == nil {
-			globalErr = errdefs.Errorf("cannot process nil entity row")
-			break
-		}
-		entityRows <- types.EntityRow{
-			EntityKey: req.EntityRow.EntityKey,
-			UnixMilli: req.EntityRow.UnixMilli,
-			Values:    req.EntityRow.Values,
+	// This goroutine runs the join operation, and send whatever joined as the response
+	joinResult, err := s.oomstore.ChannelJoin(context.Background(), types.ChannelJoinOpt{
+		JoinFeatureNames:    firstReq.JoinFeatures,
+		EntityRows:          entityRows,
+		ExistedFeatureNames: firstReq.ExistedFeatures,
+	})
+	if err != nil {
+		globalErr = err
+	} else {
+		header := joinResult.Header
+		for row := range joinResult.Data {
+			joinedRow, err := convertJoinedRow(row)
+			if err != nil {
+				globalErr = err
+				break
+			}
+			resp := &codegen.ChannelJoinResponse{
+				Header:    header,
+				JoinedRow: joinedRow,
+			}
+			if err = stream.Send(resp); err != nil {
+				globalErr = err
+				break
+			}
+			// Only need to send header upon the first response
+			header = nil
 		}
 	}
 
-	close(entityRows)
-	// wait until oomstore ChannelJoin is done, whether succeeded or failed
-	<-done
-
+	// send a notification to the data receiving goroutine that ChannelJoin has done
+	done <- struct{}{}
 	return globalErr
 }
 
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/pkg/oomstore/import_stream.go b/pkg/oomstore/import_stream.go
index 06231dd6..b6335459 100644
--- a/pkg/oomstore/import_stream.go
+++ b/pkg/oomstore/import_stream.go
@@ -114,7 +114,7 @@ func (s *OomStore) tableLinkImportStream(ctx context.Context, opt types.ImportSt
 	}
 
 	_, _, err = s.createRevision(ctx, metadata.CreateRevisionOpt{
-		Revision:    *tableSchema.TimeRange.MinUnixMilli,
+		Revision:    tableSchema.TimeRange.MinUnixMilli,
 		GroupID:     group.ID,
 		CdcTable:    &dataSource.TableName,
 		Description: opt.Description,
@@ -194,11 +194,9 @@ func (s *OomStore) validateRevisions(ctx context.Context, groupID int, schema *t
 	if err != nil {
 		return err
 	}
-	if schema.TimeRange.MinUnixMilli == nil || schema.TimeRange.MaxUnixMilli == nil {
-		return errdefs.Errorf("table is empty")
-	}
-	revisionBeforeMin := revisions.Before(*schema.TimeRange.MinUnixMilli)
-	revisionBeforeMax := revisions.Before(*schema.TimeRange.MaxUnixMilli)
+
+	revisionBeforeMin := revisions.Before(schema.TimeRange.MinUnixMilli)
+	revisionBeforeMax := revisions.Before(schema.TimeRange.MaxUnixMilli)
 	if revisionBeforeMax != revisionBeforeMin {
 		return errdefs.Errorf("data table crosses with another offline table %s", revisionBeforeMax.CdcTable)
 	}
@@ -210,7 +208,7 @@ func (s *OomStore) validateRevisions(ctx context.Context, groupID int, schema *t
 		if err != nil {
 			return err
 		}
-		if beforeTableSchema.TimeRange.MaxUnixMilli != nil && *beforeTableSchema.TimeRange.MaxUnixMilli >= *schema.TimeRange.MinUnixMilli {
+		if beforeTableSchema.TimeRange.MaxUnixMilli >= schema.TimeRange.MinUnixMilli {
 			return errdefs.Errorf("data table crosses with another offline table %s", revisionBeforeMin.CdcTable)
 		}
 	}
diff --git a/pkg/oomstore/types/types.go b/pkg/oomstore/types/types.go
index b0b1902c..1b391e71 100644
--- a/pkg/oomstore/types/types.go
+++ b/pkg/oomstore/types/types.go
@@ -67,8 +67,8 @@ func (e *ExportResult) GetErrorChannel() <-chan error {
 }
 
 type DataTableTimeRange struct {
-	MinUnixMilli *int64 `db:"min_unix_milli"`
-	MaxUnixMilli *int64 `db:"max_unix_milli"`
+	MinUnixMilli int64 `db:"min_unix_milli"`
+	MaxUnixMilli int64 `db:"max_unix_milli"`
 }
 
 type DataTableSchema struct {


Test patch:
diff --git a/internal/database/offline/snowflake/store_test.go b/internal/database/offline/snowflake/store_test.go
index 0855be67..1f0485d3 100644
--- a/internal/database/offline/snowflake/store_test.go
+++ b/internal/database/offline/snowflake/store_test.go
@@ -7,8 +7,6 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/oom-ai/oomstore/internal/database/offline/test_impl"
-
 	"github.com/stretchr/testify/require"
 
 	"github.com/oom-ai/oomstore/internal/database/dbutil"
@@ -55,25 +53,25 @@ func destroyStore(database string) func() {
 	}
 }
 
-func TestPing(t *testing.T) {
-	test_impl.TestPing(t, prepareStore, destroyStore(DATABASE))
-}
-
-func TestExport(t *testing.T) {
-	test_impl.TestExport(t, prepareStore, destroyStore(DATABASE))
-}
-
-func TestImport(t *testing.T) {
-	test_impl.TestImport(t, prepareStore, destroyStore(DATABASE))
-}
-
-func TestJoin(t *testing.T) {
-	test_impl.TestJoin(t, prepareStore, destroyStore(DATABASE))
-}
-
-func TestSnapshot(t *testing.T) {
-	test_impl.TestSnapshot(t, prepareStore, destroyStore(DATABASE))
-}
+//func TestPing(t *testing.T) {
+//	test_impl.TestPing(t, prepareStore, destroyStore(DATABASE))
+//}
+//
+//func TestExport(t *testing.T) {
+//	test_impl.TestExport(t, prepareStore, destroyStore(DATABASE))
+//}
+//
+//func TestImport(t *testing.T) {
+//	test_impl.TestImport(t, prepareStore, destroyStore(DATABASE))
+//}
+//
+//func TestJoin(t *testing.T) {
+//	test_impl.TestJoin(t, prepareStore, destroyStore(DATABASE))
+//}
+//
+//func TestSnapshot(t *testing.T) {
+//	test_impl.TestSnapshot(t, prepareStore, destroyStore(DATABASE))
+//}
 
 // We don't use test_impl.TestTableSchema because snowflake cannot be
 // accessed by two different sessions.
@@ -114,8 +112,8 @@ func TestTableSchema(t *testing.T) {
 			},
 		},
 		TimeRange: types.DataTableTimeRange{
-			MinUnixMilli: int64Ptr(1),
-			MaxUnixMilli: int64Ptr(100),
+			MinUnixMilli: 1,
+			MaxUnixMilli: 100,
 		},
 	}
 	require.ElementsMatch(t, expected.Fields, actual.Fields)
@@ -123,10 +121,6 @@ func TestTableSchema(t *testing.T) {
 
 }
 
-func int64Ptr(i int64) *int64 {
-	return &i
-}
-
-func TestCreateTable(t *testing.T) {
-	test_impl.TestCreateTable(t, prepareStore, destroyStore(DATABASE))
-}
+//func TestCreateTable(t *testing.T) {
+//	test_impl.TestCreateTable(t, prepareStore, destroyStore(DATABASE))
+//}
diff --git a/internal/database/offline/test_impl/table_schema.go b/internal/database/offline/test_impl/table_schema.go
index 5b44ab7e..e76c38f4 100644
--- a/internal/database/offline/test_impl/table_schema.go
+++ b/internal/database/offline/test_impl/table_schema.go
@@ -4,8 +4,6 @@ import (
 	"context"
 	"testing"
 
-	"github.com/stretchr/testify/assert"
-
 	"github.com/oom-ai/oomstore/internal/database/offline"
 
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
@@ -43,14 +41,9 @@ func TestTableSchema(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 			},
 		},
 		TimeRange: types.DataTableTimeRange{
-			MinUnixMilli: int64Ptr(1),
-			MaxUnixMilli: int64Ptr(100),
+			MinUnixMilli: 10,
+			MaxUnixMilli: 100,
 		},
 	}
-	assert.ElementsMatch(t, expected.Fields, actual.Fields)
-	assert.Equal(t, expected.TimeRange, actual.TimeRange)
-}
-
-func int64Ptr(i int64) *int64 {
-	return &i
+	require.ElementsMatch(t, expected.Fields, actual.Fields)
 }
diff --git a/internal/database/online/cassandra/store_test.go b/internal/database/online/cassandra/store_test.go
index 2608fe11..c3ad58f6 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))
 }
diff --git a/internal/database/online/dynamodb/store_test.go b/internal/database/online/dynamodb/store_test.go
index c1a6d878..359e1f29 100644
--- a/internal/database/online/dynamodb/store_test.go
+++ b/internal/database/online/dynamodb/store_test.go
@@ -65,6 +65,10 @@ func TestPurgeNotRemovesOtherRevisions(t *testing.T) {
 	test_impl.TestPurgeNotRemovesOtherRevisions(t, prepareStore, destroyStore(t))
 }
 
+func TestPrepareStreamTable(t *testing.T) {
+	test_impl.TestPrepareStreamTable(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..e01647db 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))
 }
diff --git a/internal/database/online/postgres/store_test.go b/internal/database/online/postgres/store_test.go
index 9789d29c..25a167cf 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))
 }
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..337e041a 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)
 }
diff --git a/internal/database/online/test_impl/create_table.go b/internal/database/online/test_impl/create_table.go
index 70eeaca0..ab5e3796 100644
--- a/internal/database/online/test_impl/create_table.go
+++ b/internal/database/online/test_impl/create_table.go
@@ -7,6 +7,33 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+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.PrepareStreamTable(ctx, online.PrepareStreamTableOpt{
+			EntityName: group.Entity.Name,
+			GroupID:    group.ID,
+		})
+		assert.NoError(t, err, "create stream table failed: %v", err)
+	})
+
+	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)
+		})
+	}
+}
+
 func TestCreateTable(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
 	t.Cleanup(destroyStore)
 
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))
 }