Repository: oom-ai/oomstore
Pull request #942 (2022-01-12 02:26:52)

Stream push


Patch:
diff --git a/CODE_OF_CONDUCT.md b/CODE_OF_CONDUCT.md
deleted file mode 100644
index e302d863..00000000
--- a/CODE_OF_CONDUCT.md
+++ /dev/null
@@ -1,128 +0,0 @@
-# Contributor Covenant Code of Conduct
-
-## Our Pledge
-
-We as members, contributors, and leaders pledge to make participation in our
-community a harassment-free experience for everyone, regardless of age, body
-size, visible or invisible disability, ethnicity, sex characteristics, gender
-identity and expression, level of experience, education, socio-economic status,
-nationality, personal appearance, race, religion, or sexual identity
-and orientation.
-
-We pledge to act and interact in ways that contribute to an open, welcoming,
-diverse, inclusive, and healthy community.
-
-## Our Standards
-
-Examples of behavior that contributes to a positive environment for our
-community include:
-
-* Demonstrating empathy and kindness toward other people
-* Being respectful of differing opinions, viewpoints, and experiences
-* Giving and gracefully accepting constructive feedback
-* Accepting responsibility and apologizing to those affected by our mistakes,
-  and learning from the experience
-* Focusing on what is best not just for us as individuals, but for the
-  overall community
-
-Examples of unacceptable behavior include:
-
-* The use of sexualized language or imagery, and sexual attention or
-  advances of any kind
-* Trolling, insulting or derogatory comments, and personal or political attacks
-* Public or private harassment
-* Publishing others' private information, such as a physical or email
-  address, without their explicit permission
-* Other conduct which could reasonably be considered inappropriate in a
-  professional setting
-
-## Enforcement Responsibilities
-
-Community leaders are responsible for clarifying and enforcing our standards of
-acceptable behavior and will take appropriate and fair corrective action in
-response to any behavior that they deem inappropriate, threatening, offensive,
-or harmful.
-
-Community leaders have the right and responsibility to remove, edit, or reject
-comments, commits, code, wiki edits, issues, and other contributions that are
-not aligned to this Code of Conduct, and will communicate reasons for moderation
-decisions when appropriate.
-
-## Scope
-
-This Code of Conduct applies within all community spaces, and also applies when
-an individual is officially representing the community in public spaces.
-Examples of representing our community include using an official e-mail address,
-posting via an official social media account, or acting as an appointed
-representative at an online or offline event.
-
-## Enforcement
-
-Instances of abusive, harassing, or otherwise unacceptable behavior may be
-reported to the community leaders responsible for enforcement at
-yiksan@oom.ai.
-All complaints will be reviewed and investigated promptly and fairly.
-
-All community leaders are obligated to respect the privacy and security of the
-reporter of any incident.
-
-## Enforcement Guidelines
-
-Community leaders will follow these Community Impact Guidelines in determining
-the consequences for any action they deem in violation of this Code of Conduct:
-
-### 1. Correction
-
-**Community Impact**: Use of inappropriate language or other behavior deemed
-unprofessional or unwelcome in the community.
-
-**Consequence**: A private, written warning from community leaders, providing
-clarity around the nature of the violation and an explanation of why the
-behavior was inappropriate. A public apology may be requested.
-
-### 2. Warning
-
-**Community Impact**: A violation through a single incident or series
-of actions.
-
-**Consequence**: A warning with consequences for continued behavior. No
-interaction with the people involved, including unsolicited interaction with
-those enforcing the Code of Conduct, for a specified period of time. This
-includes avoiding interactions in community spaces as well as external channels
-like social media. Violating these terms may lead to a temporary or
-permanent ban.
-
-### 3. Temporary Ban
-
-**Community Impact**: A serious violation of community standards, including
-sustained inappropriate behavior.
-
-**Consequence**: A temporary ban from any sort of interaction or public
-communication with the community for a specified period of time. No public or
-private interaction with the people involved, including unsolicited interaction
-with those enforcing the Code of Conduct, is allowed during this period.
-Violating these terms may lead to a permanent ban.
-
-### 4. Permanent Ban
-
-**Community Impact**: Demonstrating a pattern of violation of community
-standards, including sustained inappropriate behavior,  harassment of an
-individual, or aggression toward or disparagement of classes of individuals.
-
-**Consequence**: A permanent ban from any sort of public interaction within
-the community.
-
-## Attribution
-
-This Code of Conduct is adapted from the [Contributor Covenant][homepage],
-version 2.0, available at
-https://www.contributor-covenant.org/version/2/0/code_of_conduct.html.
-
-Community Impact Guidelines were inspired by [Mozilla's code of conduct
-enforcement ladder](https://github.com/mozilla/diversity).
-
-[homepage]: https://www.contributor-covenant.org
-
-For answers to common questions about this code of conduct, see the FAQ at
-https://www.contributor-covenant.org/faq. Translations are available at
-https://www.contributor-covenant.org/translations.
diff --git a/internal/database/offline/bigquery/export.go b/internal/database/offline/bigquery/export.go
index 7bb57e2e..2eb2c955 100644
--- a/internal/database/offline/bigquery/export.go
+++ b/internal/database/offline/bigquery/export.go
@@ -2,11 +2,7 @@ package bigquery
 
 import (
 	"context"
-	"strings"
-
-	"github.com/spf13/cast"
-
-	"github.com/oom-ai/oomstore/internal/database/offline/sqlutil"
+	"fmt"
 
 	"cloud.google.com/go/bigquery"
 	"github.com/oom-ai/oomstore/pkg/errdefs"
@@ -18,29 +14,22 @@ import (
 )
 
 func (db *DB) Export(ctx context.Context, opt offline.ExportOpt) (<-chan types.ExportRecord, <-chan error) {
-	dbOpt := dbutil.DBOpt{
-		Backend:    types.BackendBigQuery,
-		BigQueryDB: db.Client,
-		DatasetID:  &db.datasetID,
-	}
-	doExportOpt := sqlutil.DoExportOpt{
-		ExportOpt:    opt,
-		QueryResults: bigqueryQueryExportResults,
+	qt := dbutil.QuoteFn(Backend)
+	fieldStr := qt(append([]string{opt.EntityName}, opt.Features.Names()...)...)
+	tableName := qt(opt.SnapshotTable)
+
+	q := fmt.Sprintf(`SELECT %s FROM %s.%s`, fieldStr, db.datasetID, tableName)
+	if opt.Limit != nil {
+		q += fmt.Sprintf(" LIMIT %d", *opt.Limit)
 	}
-	return sqlutil.DoExport(ctx, dbOpt, doExportOpt)
-}
+	query := db.Query(q)
 
-func bigqueryQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.ExportOpt, query string, args []interface{}) (<-chan types.ExportRecord, <-chan error) {
 	stream := make(chan types.ExportRecord)
 	errs := make(chan error, 1) // at most 1 error
-	for _, arg := range args {
-		query = strings.Replace(query, "?", cast.ToString(arg), 1)
-	}
-
 	go func() {
 		defer close(stream)
 		defer close(errs)
-		rows, err := dbOpt.BigQueryDB.Query(query).Read(ctx)
+		rows, err := query.Read(ctx)
 		if err != nil {
 			errs <- err
 			return
@@ -65,4 +54,5 @@ func bigqueryQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt off
 	}()
 
 	return stream, errs
+
 }
diff --git a/internal/database/offline/sqlutil/export.go b/internal/database/offline/sqlutil/export.go
index 61894db5..04812aa8 100644
--- a/internal/database/offline/sqlutil/export.go
+++ b/internal/database/offline/sqlutil/export.go
@@ -13,50 +13,24 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-type QueryExportResults func(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.ExportOpt, query string, args []interface{}) (<-chan types.ExportRecord, <-chan error)
-
-type DoExportOpt struct {
-	offline.ExportOpt
-	QueryResults QueryExportResults
-}
-
 func Export(ctx context.Context, db *sqlx.DB, opt offline.ExportOpt, backend types.BackendType) (<-chan types.ExportRecord, <-chan error) {
-	dbOpt := dbutil.DBOpt{
-		Backend: backend,
-		SqlxDB:  db,
-	}
-	doJoinOpt := DoExportOpt{
-		ExportOpt:    opt,
-		QueryResults: sqlxQueryExportResults,
-	}
-	return DoExport(ctx, dbOpt, doJoinOpt)
-}
-
-func DoExport(ctx context.Context, dbOpt dbutil.DBOpt, opt DoExportOpt) (<-chan types.ExportRecord, <-chan error) {
 	var (
-		emptyStream = make(chan types.ExportRecord)
-		errs        = make(chan error, 1) // at most 1 error
+		stream = make(chan types.ExportRecord)
+		errs   = make(chan error, 1) // at most 1 error
 	)
-	defer close(emptyStream)
-	defer close(errs)
 
-	query, args, err := buildExportQuery(dbOpt, opt.ExportOpt)
+	query, args, err := buildExportQuery(opt, backend)
 	if err != nil {
+		defer close(stream)
+		defer close(errs)
 		errs <- errdefs.WithStack(err)
-		return emptyStream, errs
+		return stream, errs
 	}
 
-	return opt.QueryResults(ctx, dbOpt, opt.ExportOpt, query, args)
-}
-
-func sqlxQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.ExportOpt, query string, args []interface{}) (<-chan types.ExportRecord, <-chan error) {
-	stream := make(chan types.ExportRecord)
-	errs := make(chan error, 1) // at most 1 error
-
 	go func() {
 		defer close(stream)
 		defer close(errs)
-		stmt, err := dbOpt.SqlxDB.Preparex(dbOpt.SqlxDB.Rebind(query))
+		stmt, err := db.Preparex(db.Rebind(query))
 		if err != nil {
 			errs <- errdefs.WithStack(err)
 			return
@@ -79,7 +53,7 @@ func sqlxQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt offline
 				if record[i+1] == nil {
 					continue
 				}
-				deserializedValue, err := dbutil.DeserializeByValueType(record[i+1], f.ValueType, dbOpt.Backend)
+				deserializedValue, err := dbutil.DeserializeByValueType(record[i+1], f.ValueType, backend)
 				if err != nil {
 					errs <- err
 					return
@@ -93,40 +67,35 @@ func sqlxQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt offline
 	return stream, errs
 }
 
-func buildExportQuery(dbOpt dbutil.DBOpt, opt offline.ExportOpt) (string, []interface{}, error) {
+func buildExportQuery(opt offline.ExportOpt, backend types.BackendType) (string, []interface{}, error) {
 	if opt.CdcTable == nil && opt.UnixMilli == nil {
-		return buildExportBatchQuery(dbOpt, opt), nil, nil
+		return buildExportBatchQuery(opt, backend), nil, nil
 	}
 	if opt.CdcTable != nil && opt.UnixMilli != nil {
-		return buildExportStreamQuery(dbOpt, opt)
+		return buildExportStreamQuery(opt, backend)
 	}
 	return "", nil, fmt.Errorf("invalid option %+v", opt)
 }
 
-func buildExportBatchQuery(dbOpt dbutil.DBOpt, opt offline.ExportOpt) string {
+func buildExportBatchQuery(opt offline.ExportOpt, backend types.BackendType) string {
 	fields := append([]string{opt.EntityName}, opt.Features.Names()...)
-	qt := dbutil.QuoteFn(dbOpt.Backend)
+	qt := dbutil.QuoteFn(backend)
 
-	tableName := qt(opt.SnapshotTable)
-	if dbOpt.Backend == types.BackendBigQuery {
-		tableName = fmt.Sprintf("%s.%s", *dbOpt.DatasetID, tableName)
-	}
-	query := fmt.Sprintf("SELECT %s FROM %s", qt(fields...), tableName)
+	query := fmt.Sprintf("SELECT %s FROM %s", qt(fields...), qt(opt.SnapshotTable))
 	if opt.Limit != nil {
 		query += fmt.Sprintf(" LIMIT %d", *opt.Limit)
 	}
 	return query
 }
 
-func buildExportStreamQuery(dbOpt dbutil.DBOpt, opt offline.ExportOpt) (string, []interface{}, error) {
+func buildExportStreamQuery(opt offline.ExportOpt, backend types.BackendType) (string, []interface{}, error) {
 	query, err := buildAggregateQuery(aggregateQueryParams{
 		EntityName:            opt.EntityName,
 		UnixMilli:             "unix_milli",
 		FeatureNames:          opt.Features.Names(),
 		PrevSnapshotTableName: opt.SnapshotTable,
 		CurrCdcTableName:      *opt.CdcTable,
-		Backend:               dbOpt.Backend,
-		DatasetID:             dbOpt.DatasetID,
+		Backend:               backend,
 	})
 	if err != nil {
 		return "", nil, errdefs.WithStack(err)
diff --git a/internal/database/offline/sqlutil/export_helper.go b/internal/database/offline/sqlutil/export_helper.go
index 8e06dfa6..7467008c 100644
--- a/internal/database/offline/sqlutil/export_helper.go
+++ b/internal/database/offline/sqlutil/export_helper.go
@@ -72,8 +72,6 @@ type aggregateQueryParams struct {
 func buildAggregateQuery(params aggregateQueryParams) (string, error) {
 	if params.Backend == types.BackendBigQuery {
 		params.Union = "UNION DISTINCT"
-		params.PrevSnapshotTableName = fmt.Sprintf("%s.%s", *params.DatasetID, params.PrevSnapshotTableName)
-		params.CurrCdcTableName = fmt.Sprintf("%s.%s", *params.DatasetID, params.CurrCdcTableName)
 	} else {
 		params.Union = "UNION"
 	}
diff --git a/oomagent/codegen/oomagent.pb.go b/oomagent/codegen/oomagent.pb.go
index 1b209574..7e93f120 100644
--- a/oomagent/codegen/oomagent.pb.go
+++ b/oomagent/codegen/oomagent.pb.go
@@ -7,12 +7,13 @@
 package codegen
 
 import (
+	reflect "reflect"
+	sync "sync"
+
 	status "google.golang.org/genproto/googleapis/rpc/status"
 	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
 	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 	structpb "google.golang.org/protobuf/types/known/structpb"
-	reflect "reflect"
-	sync "sync"
 )
 
 const (
diff --git a/oomagent/codegen/oomagent_grpc.pb.go b/oomagent/codegen/oomagent_grpc.pb.go
index 4a1c4e1f..32a3c5d5 100644
--- a/oomagent/codegen/oomagent_grpc.pb.go
+++ b/oomagent/codegen/oomagent_grpc.pb.go
@@ -4,6 +4,7 @@ package codegen
 
 import (
 	context "context"
+
 	grpc "google.golang.org/grpc"
 	codes "google.golang.org/grpc/codes"
 	status "google.golang.org/grpc/status"
diff --git a/oomagent/server.go b/oomagent/server.go
index e30f5a17..37ad25fa 100644
--- a/oomagent/server.go
+++ b/oomagent/server.go
@@ -294,7 +294,7 @@ func (s *server) Join(ctx context.Context, req *codegen.JoinRequest) (*codegen.J
 
 func (s *server) ChannelExport(req *codegen.ChannelExportRequest, stream codegen.OomAgent_ChannelExportServer) error {
 	ctx := context.Background()
-	exportResult, err := s.oomstore.ChannelExportBatch(ctx, types.ChannelExportBatchOpt{
+	exportResult, err := s.oomstore.ChannelExport(ctx, types.ChannelExportOpt{
 		FeatureNames: req.FeatureNames,
 		RevisionID:   int(req.RevisionId),
 		Limit:        req.Limit,
@@ -331,7 +331,7 @@ func (s *server) ChannelExport(req *codegen.ChannelExportRequest, stream codegen
 }
 
 func (s *server) Export(ctx context.Context, req *codegen.ExportRequest) (*codegen.ExportResponse, error) {
-	err := s.oomstore.ExportBatch(ctx, types.ExportBatchOpt{
+	err := s.oomstore.Export(ctx, types.ExportOpt{
 		FeatureNames:   req.FeatureNames,
 		RevisionID:     int(req.RevisionId),
 		Limit:          req.Limit,
diff --git a/oomcli/cmd/export.go b/oomcli/cmd/export.go
index b1904c59..6cc17fff 100644
--- a/oomcli/cmd/export.go
+++ b/oomcli/cmd/export.go
@@ -8,11 +8,11 @@ import (
 )
 
 var exportOutput *string
-var exportOpt types.ChannelExportBatchOpt
+var exportOpt types.ChannelExportOpt
 
 var exportCmd = &cobra.Command{
 	Use:   "export",
-	Short: "ExportBatch historical features in a group",
+	Short: "Export historical features in a group",
 	PreRun: func(cmd *cobra.Command, args []string) {
 		if !cmd.Flags().Changed("limit") {
 			exportOpt.Limit = nil
diff --git a/oomcli/cmd/export_helper.go b/oomcli/cmd/export_helper.go
index b3fdf9ca..31552c2c 100644
--- a/oomcli/cmd/export_helper.go
+++ b/oomcli/cmd/export_helper.go
@@ -14,8 +14,8 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-func export(ctx context.Context, store *oomstore.OomStore, opt types.ChannelExportBatchOpt, output string) error {
-	exportResult, err := store.ChannelExportBatch(ctx, opt)
+func export(ctx context.Context, store *oomstore.OomStore, opt types.ChannelExportOpt, output string) error {
+	exportResult, err := store.ChannelExport(ctx, opt)
 	if err != nil {
 		return err
 	}
diff --git a/oomcli/cmd/push.go b/oomcli/cmd/push.go
new file mode 100644
index 00000000..91f80723
--- /dev/null
+++ b/oomcli/cmd/push.go
@@ -0,0 +1,64 @@
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"github.com/spf13/cobra"
+
+	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+)
+
+type PushOption struct {
+	EntityKey   string
+	GroupName   string
+	FeaturePair map[string]string
+
+	featureNames  []string
+	featureValues []interface{}
+}
+
+var pushOpt PushOption
+
+var pushCmd = &cobra.Command{
+	Use:   "push",
+	Short: "Push stream feature",
+	PreRun: func(cmd *cobra.Command, args []string) {
+		for feature, value := range pushOpt.FeaturePair {
+			pushOpt.featureNames = append(pushOpt.featureNames, feature)
+			pushOpt.featureValues = append(pushOpt.featureValues, value)
+		}
+	},
+	Run: func(cmd *cobra.Command, args []string) {
+		ctx := context.Background()
+		oomStore := mustOpenOomStore(ctx, oomStoreCfg)
+		defer oomStore.Close()
+
+		if err := oomStore.Push(ctx, types.PushOpt{
+			EntityKey:     pushOpt.EntityKey,
+			GroupName:     pushOpt.GroupName,
+			FeatureNames:  pushOpt.featureNames,
+			FeatureValues: pushOpt.featureValues,
+		}); err != nil {
+			exitf("failed push features: %+v\n", err)
+		}
+
+		fmt.Fprintln(os.Stderr, "succeeded")
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(pushCmd)
+
+	flags := pushCmd.Flags()
+
+	flags.StringVarP(&pushOpt.EntityKey, "entity-key", "k", "", "entity key")
+	_ = getOnlineCmd.MarkFlagRequired("entity-key")
+
+	flags.StringVarP(&pushOpt.GroupName, "group", "g", "", "feature group")
+	_ = pushCmd.MarkFlagRequired("group")
+
+	flags.StringToStringVarP(&pushOpt.FeaturePair, "features", "f", nil, "features")
+	_ = pushCmd.MarkFlagRequired("features")
+}
diff --git a/pkg/oomstore/export.go b/pkg/oomstore/export.go
index 90048fdd..3d2b70c1 100644
--- a/pkg/oomstore/export.go
+++ b/pkg/oomstore/export.go
@@ -15,9 +15,9 @@ import (
 )
 
 /*
-ChannelExportBatch exports batch feature values of a particular revision.
+ChannelExport exports batch feature values of a particular revision.
 Usage Example:
-	exportResult, err := store.ExportBatch(ctx, opt)
+	exportResult, err := store.Export(ctx, opt)
 	if err != nil {
 		return err
 	}
@@ -27,7 +27,7 @@ Usage Example:
 	// Attention: call CheckStreamError after consuming exportResult.Data channel
 	return exportResult.CheckStreamError()
 */
-func (s *OomStore) ChannelExportBatch(ctx context.Context, opt types.ChannelExportBatchOpt) (*types.ExportResult, error) {
+func (s *OomStore) ChannelExport(ctx context.Context, opt types.ChannelExportOpt) (*types.ExportResult, error) {
 	revision, err := s.GetRevision(ctx, opt.RevisionID)
 	if err != nil {
 		return nil, err
@@ -66,8 +66,8 @@ func (s *OomStore) ChannelExportBatch(ctx context.Context, opt types.ChannelExpo
 	return types.NewExportResult(header, stream, exportErr), nil
 }
 
-func (s *OomStore) ExportBatch(ctx context.Context, opt types.ExportBatchOpt) error {
-	exportResult, err := s.ChannelExportBatch(ctx, types.ChannelExportBatchOpt{
+func (s *OomStore) Export(ctx context.Context, opt types.ExportOpt) error {
+	exportResult, err := s.ChannelExport(ctx, types.ChannelExportOpt{
 		RevisionID:   opt.RevisionID,
 		FeatureNames: opt.FeatureNames,
 		Limit:        opt.Limit,
diff --git a/pkg/oomstore/types/options.go b/pkg/oomstore/types/options.go
index c2974a61..6e8d4b1b 100644
--- a/pkg/oomstore/types/options.go
+++ b/pkg/oomstore/types/options.go
@@ -30,7 +30,7 @@ type CreateGroupOpt struct {
 	Description string
 }
 
-type ChannelExportBatchOpt struct {
+type ChannelExportOpt struct {
 	RevisionID   int
 	FeatureNames []string
 	Limit        *uint64
@@ -42,7 +42,7 @@ type ChannelExportStreamOpt struct {
 	Limit            *uint64
 }
 
-type ExportBatchOpt struct {
+type ExportOpt struct {
 	RevisionID     int
 	FeatureNames   []string
 	Limit          *uint64


Test patch:
diff --git a/internal/database/offline/bigquery/store_test.go b/internal/database/offline/bigquery/store_test.go
index 1d5299fb..22f21907 100644
--- a/internal/database/offline/bigquery/store_test.go
+++ b/internal/database/offline/bigquery/store_test.go
@@ -77,6 +77,7 @@ func TestImport(t *testing.T) {
 }
 
 func TestExport(t *testing.T) {
+	t.Skip("skip export test for bigquery until we support exporting stream features")
 	test_impl.TestExport(t, prepareStore, destroyStore(DATASET_ID))
 }
 
diff --git a/oomcli/test/test_get_meta_entity.sh b/oomcli/test/test_get_meta_entity.sh
index f4905fa2..67ccd2a6 100755
--- a/oomcli/test/test_get_meta_entity.sh
+++ b/oomcli/test/test_get_meta_entity.sh
@@ -76,6 +76,13 @@ items:
         - name: user-click
           category: stream
           description: user click post feature
+          features:
+            - name: last_5_click_posts
+              value-type: string
+              description: user last 5 click posts
+            - name: number_of_user_started_posts
+              value-type: int64
+              description: number of posts that users stared today
 '
 
 actual=$(oomcli get meta entity -o yaml)
diff --git a/oomcli/test/test_get_meta_feature.sh b/oomcli/test/test_get_meta_feature.sh
index 5f215439..494ec726 100755
--- a/oomcli/test/test_get_meta_feature.sh
+++ b/oomcli/test/test_get_meta_feature.sh
@@ -12,6 +12,8 @@ ID,NAME,GROUP,ENTITY,CATEGORY,VALUE-TYPE,DESCRIPTION,ONLINE-REVISION-ID
 1,price,phone,device,batch,int64,price,<NULL>
 2,model,phone,device,batch,string,model,<NULL>
 3,age,student,user,batch,int64,age,<NULL>
+4,last_5_click_posts,user-click,user,stream,string,user last 5 click posts,<NULL>
+5,number_of_user_started_posts,user-click,user,stream,int64,number of posts that users stared today,<NULL>
 '
 actual=$(oomcli get meta feature -o csv --wide)
 ignore_time() { cut -d ',' -f 1-8 <<<"$1"; }
@@ -22,6 +24,8 @@ expected='ID,NAME,GROUP,ENTITY,CATEGORY,VALUE-TYPE,DESCRIPTION
 1,price,phone,device,batch,int64,price
 2,model,phone,device,batch,string,model
 3,age,student,user,batch,int64,age
+4,last_5_click_posts,user-click,user,stream,string,user last 5 click posts
+5,number_of_user_started_posts,user-click,user,stream,int64,number of posts that users stared today
 '
 actual=$(oomcli get meta feature -o csv)
 assert_eq "$case" "$(sort <<< "$expected")" "$(sort <<< "$actual")"
@@ -66,6 +70,16 @@ items:
       group-name: student
       value-type: int64
       description: age
+    - kind: Feature
+      name: last_5_click_posts
+      group-name: user-click
+      value-type: string
+      description: user last 5 click posts
+    - kind: Feature
+      name: number_of_user_started_posts
+      group-name: user-click
+      value-type: int64
+      description: number of posts that users stared today
 '
 
 actual=$(oomcli get meta feature -o yaml)
diff --git a/oomcli/test/test_get_meta_group.sh b/oomcli/test/test_get_meta_group.sh
index c4e036c3..4a02b5eb 100755
--- a/oomcli/test/test_get_meta_group.sh
+++ b/oomcli/test/test_get_meta_group.sh
@@ -80,6 +80,13 @@ items:
       entity-name: user
       category: stream
       description: user click post feature
+      features:
+        - name: last_5_click_posts
+          value-type: string
+          description: user last 5 click posts
+        - name: number_of_user_started_posts
+          value-type: int64
+          description: number of posts that users stared today
 '
 actual=$(oomcli get meta group -o yaml)
 assert_eq "$case" "$expected" "$actual"
diff --git a/oomcli/test/test_push.sh b/oomcli/test/test_push.sh
new file mode 100755
index 00000000..2afabe5c
--- /dev/null
+++ b/oomcli/test/test_push.sh
@@ -0,0 +1,16 @@
+#!/usr/bin/env bash
+SDIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd) && cd "$SDIR" || exit 1
+source ./util.sh
+
+init_store
+register_features
+
+case="push stream feature"
+expected='
+user,user-click.last_5_click_posts,user-click.number_of_user_started_posts
+1,"1,2,3,4,5",10
+'
+oomcli push --entity-key 1 --group user-click --features last_5_click_posts=1,2,3,4,5 --features number_of_user_started_posts=10
+
+actual=$(oomcli get online --feature user-click.last_5_click_posts,user-click.number_of_user_started_posts -k 1 -o csv)
+assert_eq "$case" "$expected" "$actual"
diff --git a/oomcli/test/util.sh b/oomcli/test/util.sh
index 059a4888..41537b68 100644
--- a/oomcli/test/util.sh
+++ b/oomcli/test/util.sh
@@ -62,6 +62,9 @@ register_features() {
     oomcli register feature price --group phone   --value-type "int64"  --description "price"
     oomcli register feature model --group phone   --value-type "string" --description "model"
     oomcli register feature age   --group student --value-type "int64"  --description "age"
+
+    oomcli register feature last_5_click_posts           --group user-click --value-type "string" --description "user last 5 click posts"
+    oomcli register feature number_of_user_started_posts --group user-click --value-type "int64"  --description "number of posts that users stared today"
 }
 
 # import sample data
diff --git a/pkg/oomstore/export_test.go b/pkg/oomstore/export_test.go
index 39e3748d..47fa1c4b 100644
--- a/pkg/oomstore/export_test.go
+++ b/pkg/oomstore/export_test.go
@@ -38,7 +38,7 @@ func TestChannelExport(t *testing.T) {
 
 	testCases := []struct {
 		description  string
-		opt          types.ChannelExportBatchOpt
+		opt          types.ChannelExportOpt
 		features     types.FeatureList
 		exportStream <-chan types.ExportRecord
 		exportError  <-chan error
@@ -46,7 +46,7 @@ func TestChannelExport(t *testing.T) {
 	}{
 		{
 			description: "provide no features, should return all feature values",
-			opt: types.ChannelExportBatchOpt{
+			opt: types.ChannelExportOpt{
 				FeatureNames: []string{},
 				RevisionID:   revisionID,
 			},
@@ -56,7 +56,7 @@ func TestChannelExport(t *testing.T) {
 		},
 		{
 			description: "provide one feature, should return one feature values",
-			opt: types.ChannelExportBatchOpt{
+			opt: types.ChannelExportOpt{
 				FeatureNames: []string{"price"},
 				RevisionID:   revisionID,
 			},
@@ -66,7 +66,7 @@ func TestChannelExport(t *testing.T) {
 		},
 		{
 			description: "provide revision and one feature name, should return one feature values",
-			opt: types.ChannelExportBatchOpt{
+			opt: types.ChannelExportOpt{
 				FeatureNames: []string{"price"},
 				RevisionID:   revisionID,
 			},
@@ -76,7 +76,7 @@ func TestChannelExport(t *testing.T) {
 		},
 		{
 			description: "empty stream",
-			opt: types.ChannelExportBatchOpt{
+			opt: types.ChannelExportOpt{
 				FeatureNames: []string{"price"},
 				RevisionID:   revisionID,
 			},
@@ -118,7 +118,7 @@ func TestChannelExport(t *testing.T) {
 			}).Return(tc.exportStream, tc.exportError)
 
 			// execute and compare results
-			actual, err := store.ChannelExportBatch(context.Background(), tc.opt)
+			actual, err := store.ChannelExport(context.Background(), tc.opt)
 			assert.NoError(t, err)
 			values := make([][]interface{}, 0)
 			for row := range actual.Data {