Repository: oom-ai/oomstore
Pull request #1160 (2022-02-09 08:32:32)

feature: oomagent graceful shutdown


Patch:
diff --git a/internal/database/dbutil/database.go b/internal/database/dbutil/database.go
index 0d1f6125..1bc70ec4 100644
--- a/internal/database/dbutil/database.go
+++ b/internal/database/dbutil/database.go
@@ -15,10 +15,6 @@ import (
 	sqlite3 "modernc.org/sqlite/lib"
 )
 
-const (
-	MYSQL_NO_SUCH_TABLE = 1146
-)
-
 func OpenSQLite(dbFile string) (*sqlx.DB, error) {
 	db, err := sqlx.Open("sqlite", dbFile)
 	return db, errdefs.WithStack(err)
@@ -51,36 +47,33 @@ func OpenPostgresDB(host, port, user, password, database string) (*sqlx.DB, erro
 
 var OpenRedshiftDB = OpenPostgresDB
 
-func IsTableNotFoundError(err error, backend types.BackendType) (bool, error) {
-	if err == nil {
-		return false, errdefs.Errorf("expected error, got nil")
-	}
+// TODO: Should return an error when bakcend is not supported ?
+func IsTableNotFoundError(err error, backend types.BackendType) bool {
 	err = errdefs.Cause(err)
 	switch backend {
 	case types.BackendSQLite:
 		if sqliteErr, ok := err.(*sqlite.Error); ok {
-			return sqliteErr.Code() == sqlite3.SQLITE_CORE, nil
+			return sqliteErr.Code() == sqlite3.SQLITE_CORE
 		}
+
 	// https://dev.mysql.com/doc/mysql-errors/5.7/en/server-error-reference.html#error_er_no_such_table
 	case types.BackendMySQL:
 		if e2, ok := err.(*mysql.MySQLError); ok {
-			return e2.Number == MYSQL_NO_SUCH_TABLE, nil
+			return e2.Number == 1146
 		}
 	case types.BackendPostgres, types.BackendRedshift:
 		if e2, ok := err.(*pq.Error); ok {
-			return e2.Code == pgerrcode.UndefinedTable, nil
+			return e2.Code == pgerrcode.UndefinedTable
 		}
 	case types.BackendSnowflake:
 		if e2, ok := err.(*gosnowflake.SnowflakeError); ok {
-			return e2.Number == gosnowflake.ErrObjectNotExistOrAuthorized, nil
+			return e2.Number == gosnowflake.ErrObjectNotExistOrAuthorized
 		}
 	// https://cloud.google.com/bigquery/docs/error-messages
 	case types.BackendBigQuery:
 		if e2, ok := err.(*googleapi.Error); ok {
-			return e2.Code == 404, nil
+			return e2.Code == 404
 		}
-	default:
-		return false, errdefs.Errorf("unsupported backend: %s", backend)
 	}
-	return false, nil
+	return false
 }
diff --git a/internal/database/metadata/sqlutil/feature.go b/internal/database/metadata/sqlutil/feature.go
index a0f7f7e1..aa7c5381 100644
--- a/internal/database/metadata/sqlutil/feature.go
+++ b/internal/database/metadata/sqlutil/feature.go
@@ -114,14 +114,9 @@ func ListFeature(ctx context.Context, sqlxCtx metadata.SqlxContext, opt metadata
 	}
 
 	// filter by entity
-	if opt.EntityIDs != nil {
+	if opt.EntityID != nil {
 		features = features.Filter(func(f *types.Feature) bool {
-			for _, id := range *opt.EntityIDs {
-				if f.Group.EntityID == id {
-					return true
-				}
-			}
-			return false
+			return f.Group.EntityID == *opt.EntityID
 		})
 	}
 	return features, nil
@@ -131,11 +126,8 @@ func buildListFeatureCond(opt metadata.ListFeatureOpt) ([]string, []interface{},
 	in := make(map[string]interface{})
 	and := make(map[string]interface{})
 
-	if opt.GroupIDs != nil {
-		if len(*opt.GroupIDs) == 0 {
-			return []string{"false"}, nil, nil
-		}
-		in["group_id"] = *opt.GroupIDs
+	if opt.GroupID != nil {
+		and["group_id"] = *opt.GroupID
 	}
 
 	if opt.FeatureIDs != nil {
diff --git a/internal/database/metadata/types.go b/internal/database/metadata/types.go
index 3f31a173..5a51cd0c 100644
--- a/internal/database/metadata/types.go
+++ b/internal/database/metadata/types.go
@@ -61,8 +61,8 @@ type UpdateRevisionOpt struct {
 }
 
 type ListFeatureOpt struct {
-	EntityIDs  *[]int
-	GroupIDs   *[]int
+	EntityID   *int
+	GroupID    *int
 	FeatureIDs *[]int
 }
 
diff --git a/internal/database/offline/bigquery/export.go b/internal/database/offline/bigquery/export.go
index 2b9d75b5..992cd8c9 100644
--- a/internal/database/offline/bigquery/export.go
+++ b/internal/database/offline/bigquery/export.go
@@ -5,13 +5,13 @@ import (
 	"strings"
 
 	"cloud.google.com/go/bigquery"
+	"github.com/oom-ai/oomstore/internal/database/dbutil"
+	"github.com/oom-ai/oomstore/internal/database/offline/sqlutil"
+	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/spf13/cast"
 	"google.golang.org/api/iterator"
 
-	"github.com/oom-ai/oomstore/internal/database/dbutil"
 	"github.com/oom-ai/oomstore/internal/database/offline"
-	"github.com/oom-ai/oomstore/internal/database/offline/sqlutil"
-	"github.com/oom-ai/oomstore/pkg/errdefs"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
@@ -30,49 +30,37 @@ func (db *DB) Export(ctx context.Context, opt offline.ExportOpt) (*types.ExportR
 
 func bigqueryQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.ExportOpt, query string, args []interface{}, features types.FeatureList) (*types.ExportResult, 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)
 		if err != nil {
-			select {
-			case stream <- types.ExportRecord{Error: err}:
-				return
-			case <-ctx.Done():
-				return
-			}
+			errs <- err
+			return
 		}
 		for {
 			recordMap := make(map[string]bigquery.Value)
 			err = rows.Next(&recordMap)
 			if err == iterator.Done {
-				return
+				break
 			}
 			if err != nil {
-				select {
-				case stream <- types.ExportRecord{Error: errdefs.Errorf("failed at rows.Next, err=%v", err)}:
-					return
-				case <-ctx.Done():
-					return
-				}
+				errs <- errdefs.Errorf("failed at rows.Next, err=%v", err)
+				return
 			}
 			record := make([]interface{}, 0, len(recordMap))
 			record = append(record, recordMap[opt.EntityName])
 			for _, feature := range features {
 				record = append(record, recordMap[feature.DBFullName(Backend)])
 			}
-			select {
-			case stream <- types.ExportRecord{Record: record}:
-				// nothing to do
-			case <-ctx.Done():
-				return
-			}
+			stream <- record
 		}
 	}()
 	header := append([]string{opt.EntityName}, features.FullNames()...)
-	return types.NewExportResult(header, stream), nil
+	return types.NewExportResult(header, stream, errs), nil
 }
diff --git a/internal/database/offline/bigquery/join.go b/internal/database/offline/bigquery/join.go
index 0e336984..b814e86c 100644
--- a/internal/database/offline/bigquery/join.go
+++ b/internal/database/offline/bigquery/join.go
@@ -54,7 +54,7 @@ func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string,
 			recordMap := make(map[string]bigquery.Value)
 			err = rows.Next(&recordMap)
 			if err == iterator.Done {
-				return
+				break
 			}
 			if err != nil {
 				scanErr = errdefs.WithStack(err)
@@ -70,13 +70,7 @@ func bigqueryQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string,
 				}
 				record = append(record, deserializedValue)
 			}
-
-			select {
-			case data <- record:
-				// nothing to do
-			case <-ctx.Done():
-				return
-			}
+			data <- record
 		}
 	}()
 
diff --git a/internal/database/offline/sqlutil/export.go b/internal/database/offline/sqlutil/export.go
index a5314488..545f3060 100644
--- a/internal/database/offline/sqlutil/export.go
+++ b/internal/database/offline/sqlutil/export.go
@@ -34,6 +34,13 @@ type DoExportOpt struct {
 }
 
 func DoExport(ctx context.Context, dbOpt dbutil.DBOpt, opt DoExportOpt) (*types.ExportResult, error) {
+	var (
+		emptyStream = make(chan types.ExportRecord)
+		errs        = make(chan error, 1) // at most 1 error
+	)
+	defer close(emptyStream)
+	defer close(errs)
+
 	// Step 0: prepare variables
 	var (
 		snapshotTables = make([]string, 0, len(opt.SnapshotTables))
@@ -117,40 +124,28 @@ func DoExport(ctx context.Context, dbOpt dbutil.DBOpt, opt DoExportOpt) (*types.
 
 func sqlxQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.ExportOpt, query string, args []interface{}, features types.FeatureList) (*types.ExportResult, 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))
 		if err != nil {
-			select {
-			case stream <- types.ExportRecord{Error: errdefs.WithStack(err)}:
-				return
-			case <-ctx.Done():
-				return
-			}
+			errs <- errdefs.WithStack(err)
+			return
 		}
 		defer stmt.Close()
-
 		rows, err := stmt.Queryx(args...)
 		if err != nil {
-			select {
-			case stream <- types.ExportRecord{Error: errdefs.WithStack(err)}:
-				return
-			case <-ctx.Done():
-				return
-			}
+			errs <- errdefs.WithStack(err)
+			return
 		}
 		defer rows.Close()
-
 		for rows.Next() {
 			record, err := rows.SliceScan()
 			if err != nil {
-				select {
-				case stream <- types.ExportRecord{Error: errdefs.Errorf("failed at rows.SliceScan, err=%v", err)}:
-					return
-				case <-ctx.Done():
-					return
-				}
+				errs <- errdefs.Errorf("failed at rows.SliceScan, err=%v", err)
+				return
 			}
 			record[0] = cast.ToString(record[0])
 			for i, f := range features {
@@ -159,23 +154,14 @@ func sqlxQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt offline
 				}
 				deserializedValue, err := dbutil.DeserializeByValueType(record[i+1], f.ValueType, dbOpt.Backend)
 				if err != nil {
-					select {
-					case stream <- types.ExportRecord{Error: err}:
-						return
-					case <-ctx.Done():
-						return
-					}
+					errs <- err
+					return
 				}
 				record[i+1] = deserializedValue
 			}
-			select {
-			case stream <- types.ExportRecord{Record: record, Error: nil}:
-				// nothing to do
-			case <-ctx.Done():
-				return
-			}
+			stream <- record
 		}
 	}()
 	header := append([]string{opt.EntityName}, features.FullNames()...)
-	return types.NewExportResult(header, stream), nil
+	return types.NewExportResult(header, stream, errs), nil
 }
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index 1be54be8..d03983bd 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -329,13 +329,7 @@ func sqlxQueryResults(ctx context.Context, dbOpt dbutil.DBOpt, query string, hea
 				}
 				deserializedRecord = append(deserializedRecord, deserializedValue)
 			}
-
-			select {
-			case data <- deserializedRecord:
-				// nothing to do
-			case <-ctx.Done():
-				return
-			}
+			data <- deserializedRecord
 		}
 	}()
 
diff --git a/internal/database/offline/sqlutil/push.go b/internal/database/offline/sqlutil/push.go
index 56f8116a..748d39bc 100644
--- a/internal/database/offline/sqlutil/push.go
+++ b/internal/database/offline/sqlutil/push.go
@@ -21,14 +21,8 @@ func Push(ctx context.Context, dbOpt dbutil.DBOpt, pushOpt offline.PushOpt) erro
 	}
 
 	err := dbutil.InsertRecordsToTable(ctx, dbOpt, tableName, rows, columns)
-	if err != nil {
-		tableNotFound, notFoundErr := dbutil.IsTableNotFoundError(err, dbOpt.Backend)
-		if notFoundErr != nil {
-			return notFoundErr
-		}
-		if tableNotFound {
-			return errdefs.NotFound(errdefs.WithStack(err))
-		}
+	if err != nil && dbutil.IsTableNotFoundError(err, dbOpt.Backend) {
+		return errdefs.NotFound(errdefs.WithStack(err))
 	}
 	return errdefs.WithStack(err)
 }
diff --git a/internal/database/online/cassandra/import.go b/internal/database/online/cassandra/import.go
index b8470d46..7da7241f 100644
--- a/internal/database/online/cassandra/import.go
+++ b/internal/database/online/cassandra/import.go
@@ -39,15 +39,11 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	insertStmt := buildInsertStatement(tableName, columns)
 	batch := db.NewBatch(gocql.LoggedBatch)
 	for record := range opt.ExportStream {
-		if record.Error != nil {
-			return record.Error
-		}
-
-		if len(record.Record) != len(opt.Features)+1 {
-			return errdefs.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record.Record))
+		if len(record) != len(opt.Features)+1 {
+			return errdefs.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
 		}
 		if batch.Size() != BatchSize {
-			batch.Query(insertStmt, record.Record...)
+			batch.Query(insertStmt, record...)
 		} else {
 			if err := db.ExecuteBatch(batch); err != nil {
 				return errdefs.WithStack(err)
diff --git a/internal/database/online/dynamodb/import.go b/internal/database/online/dynamodb/import.go
index 0cc1d70b..b7e2c832 100644
--- a/internal/database/online/dynamodb/import.go
+++ b/internal/database/online/dynamodb/import.go
@@ -46,10 +46,6 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	// Step 2: import items to the table
 	items := make([]types.WriteRequest, 0, BatchWriteItemCapacity)
 	for record := range opt.ExportStream {
-		if record.Error != nil {
-			return record.Error
-		}
-
 		item, err := buildItem(record, opt)
 		if err != nil {
 			return err
@@ -70,6 +66,14 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 		return err
 	}
 
+	if opt.ExportError != nil {
+		select {
+		case err := <-opt.ExportError:
+			return err
+		default:
+			return nil
+		}
+	}
 	return nil
 }
 
diff --git a/internal/database/online/redis/import.go b/internal/database/online/redis/import.go
index 5bfb91de..56e3a8ed 100644
--- a/internal/database/online/redis/import.go
+++ b/internal/database/online/redis/import.go
@@ -16,14 +16,11 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	defer pipe.Close()
 
 	for record := range opt.ExportStream {
-		if record.Error != nil {
-			return record.Error
-		}
-		if len(record.Record) != len(opt.Features)+1 {
-			return fmt.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record.Record))
+		if len(record) != len(opt.Features)+1 {
+			return fmt.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
 		}
 
-		entityKey, values := record.Record[0], record.Record[1:]
+		entityKey, values := record[0], record[1:]
 		var key string
 		var err error
 		if opt.Group.Category == types.CategoryBatch {
@@ -68,5 +65,14 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 			return errdefs.WithStack(err)
 		}
 	}
+
+	if opt.ExportError != nil {
+		select {
+		case err := <-opt.ExportError:
+			return err
+		default:
+			return nil
+		}
+	}
 	return nil
 }
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index b1b78460..1d247163 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -32,11 +32,7 @@ func Get(ctx context.Context, db *sqlx.DB, opt online.GetOpt, backend types.Back
 
 	record, err := db.QueryRowxContext(ctx, db.Rebind(query), opt.EntityKey).SliceScan()
 	if err != nil {
-		tableNotFound, notFoundErr := dbutil.IsTableNotFoundError(err, backend)
-		if notFoundErr != nil {
-			return nil, notFoundErr
-		}
-		if err == sql.ErrNoRows || tableNotFound {
+		if err == sql.ErrNoRows || dbutil.IsTableNotFoundError(err, backend) {
 			return make(dbutil.RowMap), nil
 		}
 		return nil, errdefs.WithStack(err)
@@ -72,11 +68,7 @@ func MultiGet(ctx context.Context, db *sqlx.DB, opt online.MultiGetOpt, backend
 
 	rows, err := db.QueryxContext(ctx, db.Rebind(query), args...)
 	if err != nil {
-		tableNotFound, notFoundErr := dbutil.IsTableNotFoundError(err, backend)
-		if notFoundErr != nil {
-			return nil, notFoundErr
-		}
-		if err == sql.ErrNoRows || tableNotFound {
+		if err == sql.ErrNoRows || dbutil.IsTableNotFoundError(err, backend) {
 			return make(map[string]dbutil.RowMap), nil
 		}
 		return nil, errdefs.WithStack(err)
diff --git a/internal/database/online/sqlutil/import.go b/internal/database/online/sqlutil/import.go
index e011cd16..fe1a9dbf 100644
--- a/internal/database/online/sqlutil/import.go
+++ b/internal/database/online/sqlutil/import.go
@@ -5,7 +5,6 @@ import (
 	"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"
@@ -35,14 +34,10 @@ func Import(ctx context.Context, db *sqlx.DB, opt online.ImportOpt, backend type
 		// populate the data table
 		records := make([]interface{}, 0, importBatchSize)
 		for record := range opt.ExportStream {
-			if record.Error != nil {
-				return record.Error
-			}
-
-			if len(record.Record) != len(opt.Features)+1 {
-				return errdefs.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record.Record))
+			if len(record) != len(opt.Features)+1 {
+				return errdefs.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
 			}
-			records = append(records, record.Record)
+			records = append(records, record)
 
 			if len(records) == importBatchSize {
 				if err = dbutil.InsertRecordsToTableTx(tx, ctx, tableName, records, columns, backend); err != nil {
@@ -54,6 +49,15 @@ func Import(ctx context.Context, db *sqlx.DB, opt online.ImportOpt, backend type
 		if err = dbutil.InsertRecordsToTableTx(tx, ctx, tableName, records, columns, backend); err != nil {
 			return err
 		}
+		if opt.ExportError != nil {
+			select {
+			case err := <-opt.ExportError:
+				if err != nil {
+					return err
+				}
+			default:
+			}
+		}
 
 		if opt.Group.Category == types.CategoryStream {
 			streamTableName := dbutil.OnlineStreamTableName(opt.Group.ID)
diff --git a/internal/database/online/tikv/import.go b/internal/database/online/tikv/import.go
index 7c60c8db..5614d151 100644
--- a/internal/database/online/tikv/import.go
+++ b/internal/database/online/tikv/import.go
@@ -41,15 +41,11 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 	var putVals [][]byte
 
 	for record := range opt.ExportStream {
-		if record.Error != nil {
-			return record.Error
+		if len(record) != len(opt.Features)+1 {
+			return errdefs.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record))
 		}
 
-		if len(record.Record) != len(opt.Features)+1 {
-			return errdefs.Errorf("field count not matched, expected %d, got %d", len(opt.Features)+1, len(record.Record))
-		}
-
-		entityKey, featureValues := record.Record[0], record.Record[1:]
+		entityKey, featureValues := record[0], record[1:]
 
 		serializedEntityKey, err := dbutil.SerializeByValue(entityKey, Backend)
 		if err != nil {
@@ -91,5 +87,14 @@ func (db *DB) Import(ctx context.Context, opt online.ImportOpt) error {
 			return errdefs.WithStack(err)
 		}
 	}
+
+	if opt.ExportError != nil {
+		select {
+		case err := <-opt.ExportError:
+			return err
+		default:
+			return nil
+		}
+	}
 	return nil
 }
diff --git a/internal/database/online/types.go b/internal/database/online/types.go
index 827e1635..9c8d8343 100644
--- a/internal/database/online/types.go
+++ b/internal/database/online/types.go
@@ -41,6 +41,7 @@ type ImportOpt struct {
 	Group        types.Group
 	Features     types.FeatureList
 	ExportStream <-chan types.ExportRecord
+	ExportError  <-chan error
 	RevisionID   *int
 }
 
diff --git a/oomagent/root.go b/oomagent/root.go
index 0120d2ae..65b1b03a 100644
--- a/oomagent/root.go
+++ b/oomagent/root.go
@@ -7,8 +7,10 @@ import (
 	"log"
 	"net"
 	"os"
+	"os/signal"
 	"path/filepath"
 	"strconv"
+	"syscall"
 
 	"github.com/adrg/xdg"
 	"github.com/spf13/cobra"
@@ -34,11 +36,11 @@ var rootCmd = &cobra.Command{
 	Version: version.String(),
 	Run: func(cmd *cobra.Command, args []string) {
 		ctx := context.Background()
-		oomStore, err := oomstore.Open(ctx, oomStoreCfg)
+		oomstore, err := oomstore.Open(ctx, oomStoreCfg)
 		if err != nil {
 			log.Fatal(err)
 		}
-		defer oomStore.Close()
+		defer oomstore.Close()
 
 		lis, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", port))
 		if err != nil {
@@ -55,7 +57,17 @@ var rootCmd = &cobra.Command{
 		}
 
 		grpcServer := grpc.NewServer()
-		codegen.RegisterOomAgentServer(grpcServer, &server{oomstore: oomStore})
+		codegen.RegisterOomAgentServer(grpcServer, &server{oomstore: oomstore})
+
+		exit := make(chan os.Signal, 1)
+		signal.Notify(exit, syscall.SIGTERM, syscall.SIGHUP, syscall.SIGINT, syscall.SIGQUIT)
+		go func() {
+			<-exit
+			grpcServer.GracefulStop()
+			oomstore.Close()
+			os.Exit(0)
+		}()
+
 		if err := grpcServer.Serve(lis); err != nil {
 			panic(err)
 		}
diff --git a/oomagent/server.go b/oomagent/server.go
index 2ef3d00e..0c5a0dba 100644
--- a/oomagent/server.go
+++ b/oomagent/server.go
@@ -242,7 +242,7 @@ func (s *server) ChannelJoin(stream codegen.OomAgent_ChannelJoinServer) error {
 				UnixMilli: req.EntityRow.UnixMilli,
 				Values:    req.EntityRow.Values,
 			}:
-				// nothing to do
+				// nothing todo
 			case <-ctx.Done():
 				return
 			}
@@ -299,9 +299,7 @@ func (s *server) ChannelExport(req *codegen.ChannelExportRequest, stream codegen
 		return nil
 	}
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
+	ctx := context.Background()
 	exportResult, err := s.oomstore.ChannelExport(ctx, types.ChannelExportOpt{
 		FeatureNames: req.Features,
 		UnixMilli:    req.UnixMilli,
@@ -313,11 +311,7 @@ func (s *server) ChannelExport(req *codegen.ChannelExportRequest, stream codegen
 
 	header := exportResult.Header
 	for row := range exportResult.Data {
-		if row.Error != nil {
-			return row.Error
-		}
-
-		valueRow, err := convertToValueSlice(row.Record)
+		valueRow, err := convertToValueSlice(row)
 		if err != nil {
 			return internalError(err.Error())
 		}
@@ -330,6 +324,9 @@ func (s *server) ChannelExport(req *codegen.ChannelExportRequest, stream codegen
 		// Only need to send header upon the first response
 		header = nil
 	}
+	if err := exportResult.CheckStreamError(); err != nil {
+		return internalError(err.Error())
+	}
 	return nil
 }
 
diff --git a/oomcli/cmd/edit_entity.go b/oomcli/cmd/edit_entity.go
index c665ec6e..8784e0c0 100644
--- a/oomcli/cmd/edit_entity.go
+++ b/oomcli/cmd/edit_entity.go
@@ -31,11 +31,7 @@ var editEntityCmd = &cobra.Command{
 		oomStore := mustOpenOomStore(ctx, oomStoreCfg)
 		defer oomStore.Close()
 
-		var listEntityOpt types.ListEntityOpt
-		if editEntityOpt.entityName != nil {
-			listEntityOpt.EntityNames = &[]string{*editEntityOpt.entityName}
-		}
-		entities, err := oomStore.ListEntity(ctx, listEntityOpt)
+		entities, err := queryEntities(ctx, oomStore, editEntityOpt.entityName)
 		if err != nil {
 			exit(err)
 		}
@@ -63,11 +59,7 @@ func writeEntitiesToTempFile(ctx context.Context, oomStore *oomstore.OomStore, e
 	}
 	defer tempFile.Close()
 
-	if err = outputEntity(ctx, entities, outputParams{
-		writer:    tempFile,
-		oomStore:  oomStore,
-		outputOpt: YAML,
-	}); err != nil {
+	if err := serializeEntitiesToWriter(ctx, tempFile, oomStore, entities, YAML); err != nil {
 		return "", err
 	}
 
diff --git a/oomcli/cmd/edit_feature.go b/oomcli/cmd/edit_feature.go
index 1ace89a5..451c09ae 100644
--- a/oomcli/cmd/edit_feature.go
+++ b/oomcli/cmd/edit_feature.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 
+	"github.com/oom-ai/oomstore/pkg/oomstore"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 	"github.com/spf13/cobra"
 )
@@ -14,18 +15,17 @@ type editFeatureOption struct {
 }
 
 var editFeatureOpt editFeatureOption
-var editFeatureEntityName, editFeatureGroupName *string
 
 var editFeatureCmd = &cobra.Command{
 	Use:   "feature [feature_name]",
 	Short: "Edit feature resources",
 	Args:  cobra.MaximumNArgs(1),
 	PreRun: func(cmd *cobra.Command, args []string) {
-		if cmd.Flags().Changed("entity") {
-			editFeatureOpt.EntityNames = &[]string{*editFeatureEntityName}
+		if !cmd.Flags().Changed("entity") {
+			editFeatureOpt.EntityName = nil
 		}
-		if cmd.Flags().Changed("group") {
-			editFeatureOpt.GroupNames = &[]string{*editFeatureGroupName}
+		if !cmd.Flags().Changed("group") {
+			editFeatureOpt.GroupName = nil
 		}
 
 		if len(args) == 1 {
@@ -42,7 +42,7 @@ var editFeatureCmd = &cobra.Command{
 			exit(err)
 		}
 
-		fileName, err := writeFeaturesToTempFile(features)
+		fileName, err := writeFeaturesToTempFile(ctx, oomStore, features)
 		if err != nil {
 			exit(err)
 		}
@@ -58,21 +58,18 @@ func init() {
 	editCmd.AddCommand(editFeatureCmd)
 
 	flags := editFeatureCmd.Flags()
-	editFeatureEntityName = flags.StringP("entity", "e", "", "entity")
-	editFeatureGroupName = flags.StringP("group", "g", "", "feature group")
+	editFeatureOpt.EntityName = flags.StringP("entity", "e", "", "entity")
+	editFeatureOpt.GroupName = flags.StringP("group", "g", "", "feature group")
 }
 
-func writeFeaturesToTempFile(features types.FeatureList) (string, error) {
+func writeFeaturesToTempFile(ctx context.Context, oomStore *oomstore.OomStore, features types.FeatureList) (string, error) {
 	tempFile, err := getTempFile()
 	if err != nil {
 		return "", err
 	}
 	defer tempFile.Close()
 
-	if err = outputFeature(features, outputParams{
-		writer:    tempFile,
-		outputOpt: YAML,
-	}); err != nil {
+	if err = serializeFeatureToWriter(tempFile, features, YAML); err != nil {
 		return "", err
 	}
 	return tempFile.Name(), nil
diff --git a/oomcli/cmd/edit_group.go b/oomcli/cmd/edit_group.go
index bd98ffb6..021d01b2 100644
--- a/oomcli/cmd/edit_group.go
+++ b/oomcli/cmd/edit_group.go
@@ -35,14 +35,7 @@ var editGroupCmd = &cobra.Command{
 		oomStore := mustOpenOomStore(ctx, oomStoreCfg)
 		defer oomStore.Close()
 
-		listGroupOpt := types.ListGroupOpt{}
-		if getMetaGroupOpt.entityName != nil {
-			listGroupOpt.EntityNames = &[]string{*getMetaGroupOpt.entityName}
-		}
-		if getMetaGroupOpt.groupName != nil {
-			listGroupOpt.GroupNames = &[]string{*getMetaGroupOpt.groupName}
-		}
-		groups, err := oomStore.ListGroup(ctx, listGroupOpt)
+		groups, err := queryGroups(ctx, oomStore, editGroupOpt.entityName, editGroupOpt.groupName)
 		if err != nil {
 			exit(err)
 		}
@@ -73,11 +66,7 @@ func writeGroupsToTempFile(ctx context.Context, oomStore *oomstore.OomStore, gro
 	}
 	defer tempFile.Close()
 
-	if err = outputGroup(ctx, groups, outputParams{
-		writer:    tempFile,
-		oomStore:  oomStore,
-		outputOpt: YAML,
-	}); err != nil {
+	if err = serializeGroupToWriter(ctx, tempFile, oomStore, groups, YAML); err != nil {
 		return "", err
 	}
 	return tempFile.Name(), nil
diff --git a/oomcli/cmd/export_helper.go b/oomcli/cmd/export_helper.go
index 2d164de2..31552c2c 100644
--- a/oomcli/cmd/export_helper.go
+++ b/oomcli/cmd/export_helper.go
@@ -15,9 +15,6 @@ import (
 )
 
 func export(ctx context.Context, store *oomstore.OomStore, opt types.ChannelExportOpt, output string) error {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
 	exportResult, err := store.ChannelExport(ctx, opt)
 	if err != nil {
 		return err
@@ -48,15 +45,11 @@ func printExportResultInCSV(exportResult *types.ExportResult) error {
 	}
 
 	for row := range exportResult.Data {
-		if row.Error != nil {
-			return row.Error
-		}
-
-		if err := w.Write(cast.ToStringSlice([]interface{}(row.Record))); err != nil {
+		if err := w.Write(cast.ToStringSlice([]interface{}(row))); err != nil {
 			return err
 		}
 	}
-	return nil
+	return exportResult.CheckStreamError()
 }
 
 func printExportResultInASCIITable(exportResult *types.ExportResult) error {
@@ -65,13 +58,9 @@ func printExportResultInASCIITable(exportResult *types.ExportResult) error {
 	table.SetAutoFormatHeaders(false)
 
 	for row := range exportResult.Data {
-		if row.Error != nil {
-			return row.Error
-		}
-
-		table.Append(cast.ToStringSlice([]interface{}(row.Record)))
+		table.Append(cast.ToStringSlice([]interface{}(row)))
 	}
 
 	table.Render()
-	return nil
+	return exportResult.CheckStreamError()
 }
diff --git a/oomcli/cmd/get_meta_entity.go b/oomcli/cmd/get_meta_entity.go
index 57a050fe..b563738a 100644
--- a/oomcli/cmd/get_meta_entity.go
+++ b/oomcli/cmd/get_meta_entity.go
@@ -32,20 +32,12 @@ var getMetaEntityCmd = &cobra.Command{
 		oomStore := mustOpenOomStore(ctx, oomStoreCfg)
 		defer oomStore.Close()
 
-		var listEntityOpt types.ListEntityOpt
-		if getMetaEntityOpt.entityName != nil {
-			listEntityOpt.EntityNames = &[]string{*getMetaEntityOpt.entityName}
-		}
-		entities, err := oomStore.ListEntity(ctx, listEntityOpt)
+		entities, err := queryEntities(ctx, oomStore, getMetaEntityOpt.entityName)
 		if err != nil {
 			exit(err)
 		}
 
-		if err = outputEntity(ctx, entities, outputParams{
-			writer:    os.Stdout,
-			oomStore:  oomStore,
-			outputOpt: *getMetaOutput,
-		}); err != nil {
+		if err = serializeEntitiesToWriter(ctx, os.Stdout, oomStore, entities, *getMetaOutput); err != nil {
 			exitf("failed printing entities, error: %+v\n", err)
 		}
 	},
@@ -55,32 +47,32 @@ func init() {
 	getMetaCmd.AddCommand(getMetaEntityCmd)
 }
 
-type outputParams struct {
-	writer    io.Writer
-	oomStore  *oomstore.OomStore
-	outputOpt string
+func queryEntities(ctx context.Context, oomStore *oomstore.OomStore, entityName *string) (types.EntityList, error) {
+	if entityName != nil {
+		entity, err := oomStore.GetEntityByName(ctx, *entityName)
+		return types.EntityList{entity}, err
+	}
+	return oomStore.ListEntity(ctx)
 }
 
-func outputEntity(ctx context.Context, entities types.EntityList, params outputParams) error {
-	switch params.outputOpt {
+func serializeEntitiesToWriter(ctx context.Context, w io.Writer, oomStore *oomstore.OomStore,
+	entities types.EntityList, outputOpt string) error {
+
+	switch outputOpt {
 	case YAML:
-		entityNames := entities.Names()
-		groups, err := params.oomStore.ListGroup(ctx, types.ListGroupOpt{
-			EntityNames: &entityNames,
-		})
+		// TODO: Use entity ids to filter, rather than taking them all out
+		groups, err := oomStore.ListGroup(ctx, nil)
 		if err != nil {
 			return err
 		}
-		features, err := params.oomStore.ListFeature(ctx, types.ListFeatureOpt{
-			EntityNames: &entityNames,
-		})
+
+		groupItems, err := groupsToApplyGroupItems(ctx, oomStore, groups)
 		if err != nil {
 			return err
 		}
-		groupItems := apply.BuildGroupItems(groups, features)
-		entityItems := apply.BuildEntityItems(entities, groupItems)
-		return serializeInYaml(params.writer, entityItems)
+
+		return serializeInYaml(w, apply.FromEntityList(entities, groupItems))
 	default:
-		return serializeMetadata(params.writer, entities, params.outputOpt, *getMetaWide)
+		return serializeMetadata(w, entities, outputOpt, *getMetaWide)
 	}
 }
diff --git a/oomcli/cmd/get_meta_feature.go b/oomcli/cmd/get_meta_feature.go
index bf1695bb..07ff6c53 100644
--- a/oomcli/cmd/get_meta_feature.go
+++ b/oomcli/cmd/get_meta_feature.go
@@ -3,6 +3,7 @@ package cmd
 import (
 	"context"
 	"fmt"
+	"io"
 	"os"
 
 	"github.com/oom-ai/oomstore/pkg/errdefs"
@@ -14,18 +15,17 @@ import (
 )
 
 var getMetaFeatureOpt types.ListFeatureOpt
-var getMetaFeatureEntityName, getMetaFeatureGroupName *string
 
 var getMetaFeatureCmd = &cobra.Command{
 	Use:   "feature [feature_name]",
 	Short: "Get existing features given specific conditions",
 	Args:  cobra.MaximumNArgs(1),
 	PreRun: func(cmd *cobra.Command, args []string) {
-		if cmd.Flags().Changed("entity") {
-			getMetaFeatureOpt.EntityNames = &[]string{*getMetaFeatureEntityName}
+		if !cmd.Flags().Changed("entity") {
+			getMetaFeatureOpt.EntityName = nil
 		}
-		if cmd.Flags().Changed("group") {
-			getMetaFeatureOpt.GroupNames = &[]string{*getMetaFeatureGroupName}
+		if !cmd.Flags().Changed("group") {
+			getMetaFeatureOpt.GroupName = nil
 		}
 		if len(args) == 1 {
 			getMetaFeatureOpt.FeatureNames = &[]string{args[0]}
@@ -41,10 +41,7 @@ var getMetaFeatureCmd = &cobra.Command{
 			exit(err)
 		}
 
-		if err := outputFeature(features, outputParams{
-			writer:    os.Stdout,
-			outputOpt: *getMetaOutput,
-		}); err != nil {
+		if err := serializeFeatureToWriter(os.Stdout, features, *getMetaOutput); err != nil {
 			exitf("failed printing features: %+v\n", err)
 		}
 	},
@@ -54,8 +51,8 @@ func init() {
 	getMetaCmd.AddCommand(getMetaFeatureCmd)
 
 	flags := getMetaFeatureCmd.Flags()
-	getMetaFeatureEntityName = flags.StringP("entity", "e", "", "entity")
-	getMetaFeatureGroupName = flags.StringP("group", "g", "", "feature group")
+	getMetaFeatureOpt.EntityName = flags.StringP("entity", "e", "", "entity")
+	getMetaFeatureOpt.GroupName = flags.StringP("group", "g", "", "feature group")
 }
 
 func queryFeatures(ctx context.Context, oomStore *oomstore.OomStore, opt types.ListFeatureOpt) (types.FeatureList, error) {
@@ -64,19 +61,18 @@ func queryFeatures(ctx context.Context, oomStore *oomstore.OomStore, opt types.L
 		return nil, fmt.Errorf("failed getting features, error %v\n", err)
 	}
 
-	if opt.FeatureNames != nil && len(*opt.FeatureNames) != len(features) {
-		missing := features.FindMissingFeatures(*opt.FeatureNames)
-		return nil, errdefs.Errorf("feature '%s' not found", missing)
+	if opt.FeatureNames != nil && len(features) == 0 {
+		return nil, errdefs.Errorf("feature '%s' not found", (*opt.FeatureNames)[0])
 	}
 
 	return features, nil
 }
 
-func outputFeature(features types.FeatureList, params outputParams) error {
-	switch params.outputOpt {
+func serializeFeatureToWriter(w io.Writer, features types.FeatureList, outputOpt string) error {
+	switch outputOpt {
 	case YAML:
-		return serializeInYaml(params.writer, apply.BuildFeatureItems(features))
+		return serializeInYaml(w, apply.FromFeatureList(features))
 	default:
-		return serializeMetadata(params.writer, features, params.outputOpt, *getMetaWide)
+		return serializeMetadata(w, features, outputOpt, *getMetaWide)
 	}
 }
diff --git a/oomcli/cmd/get_meta_group.go b/oomcli/cmd/get_meta_group.go
index 4919ffff..307528a4 100644
--- a/oomcli/cmd/get_meta_group.go
+++ b/oomcli/cmd/get_meta_group.go
@@ -2,10 +2,12 @@ package cmd
 
 import (
 	"context"
+	"fmt"
+	"io"
 	"os"
 
-	"github.com/oom-ai/oomstore/pkg/oomstore/types/apply"
-
+	"github.com/oom-ai/oomstore/pkg/errdefs"
+	"github.com/oom-ai/oomstore/pkg/oomstore"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 	"github.com/spf13/cobra"
 )
@@ -34,23 +36,12 @@ var getMetaGroupCmd = &cobra.Command{
 		oomStore := mustOpenOomStore(ctx, oomStoreCfg)
 		defer oomStore.Close()
 
-		listGroupOpt := types.ListGroupOpt{}
-		if getMetaGroupOpt.entityName != nil {
-			listGroupOpt.EntityNames = &[]string{*getMetaGroupOpt.entityName}
-		}
-		if getMetaGroupOpt.groupName != nil {
-			listGroupOpt.GroupNames = &[]string{*getMetaGroupOpt.groupName}
-		}
-		groups, err := oomStore.ListGroup(ctx, listGroupOpt)
+		groups, err := queryGroups(ctx, oomStore, getMetaGroupOpt.entityName, getMetaGroupOpt.groupName)
 		if err != nil {
 			exitf("%+v", err)
 		}
 
-		if err = outputGroup(ctx, groups, outputParams{
-			writer:    os.Stdout,
-			oomStore:  oomStore,
-			outputOpt: *getMetaOutput,
-		}); err != nil {
+		if err = serializeGroupToWriter(ctx, os.Stdout, oomStore, groups, *getMetaOutput); err != nil {
 			exitf("%+v", err)
 		}
 	},
@@ -64,22 +55,43 @@ func init() {
 	getMetaGroupOpt.entityName = flags.StringP("entity", "", "", "use to filter groups")
 }
 
-func outputGroup(ctx context.Context, groups types.GroupList, params outputParams) error {
-	switch params.outputOpt {
-	case YAML:
-		groupNames := groups.Names()
-		features, err := params.oomStore.ListFeature(ctx, types.ListFeatureOpt{
-			GroupNames: &groupNames,
-		})
+func queryGroups(ctx context.Context, oomStore *oomstore.OomStore, entityName, groupName *string) (types.GroupList, error) {
+	var entityID *int
+
+	if groupName != nil {
+		group, err := oomStore.GetGroupByName(ctx, *groupName)
 		if err != nil {
-			return err
+			return nil, err
+		}
+
+		if entityName != nil && group.Entity.Name != *entityName {
+			return nil, errdefs.Errorf("group '%s' entityName is '%s' not '%s'", *groupName, group.Entity.Name, *entityName)
 		}
-		groupItems := apply.BuildGroupItems(groups, features)
+		return types.GroupList{group}, err
+	}
+
+	if entityName != nil {
+		entity, err := oomStore.GetEntityByName(ctx, *entityName)
 		if err != nil {
+			return nil, fmt.Errorf("failed to get entity name='%s': %v", *entityName, err)
+		}
+		entityID = &entity.ID
+	}
+
+	return oomStore.ListGroup(ctx, entityID)
+}
+
+func serializeGroupToWriter(ctx context.Context, w io.Writer, oomStore *oomstore.OomStore,
+	groups types.GroupList, outputOpt string) error {
+
+	switch outputOpt {
+	case YAML:
+		if items, err := groupsToApplyGroupItems(ctx, oomStore, groups); err != nil {
 			return err
+		} else {
+			return serializeInYaml(w, *items)
 		}
-		return serializeInYaml(params.writer, groupItems)
 	default:
-		return serializeMetadata(params.writer, groups, params.outputOpt, *getMetaWide)
+		return serializeMetadata(w, groups, outputOpt, *getMetaWide)
 	}
 }
diff --git a/oomcli/cmd/util.go b/oomcli/cmd/util.go
index 718eeeb8..3cf94758 100644
--- a/oomcli/cmd/util.go
+++ b/oomcli/cmd/util.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/oom-ai/oomstore/pkg/oomstore"
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
+	"github.com/oom-ai/oomstore/pkg/oomstore/types/apply"
 )
 
 const (
@@ -32,6 +33,15 @@ func stringPtr(s string) *string {
 	return &s
 }
 
+func groupsToApplyGroupItems(ctx context.Context, store *oomstore.OomStore, groups types.GroupList) (*apply.GroupItems, error) {
+	// TODO: Use group ids to filter, rather than taking them all out
+	features, err := store.ListFeature(ctx, types.ListFeatureOpt{})
+	if err != nil {
+		return nil, err
+	}
+	return apply.FromGroupList(groups, features), nil
+}
+
 func exitf(format string, a ...interface{}) {
 	msg := fmt.Sprintf(format, a...)
 	fmt.Fprintf(os.Stderr, "Error: %s\n", msg)
diff --git a/pkg/oomstore/apply.go b/pkg/oomstore/apply.go
index d58dba3a..50bb1d41 100644
--- a/pkg/oomstore/apply.go
+++ b/pkg/oomstore/apply.go
@@ -192,7 +192,7 @@ func (s *OomStore) applyFeature(ctx context.Context, tx metadata.DBStore, newFea
 
 		if group.Category == types.CategoryStream {
 			features, err := tx.ListFeature(ctx, metadata.ListFeatureOpt{
-				GroupIDs: &[]int{group.ID},
+				GroupID: &group.ID,
 			})
 			if err != nil {
 				return nil, err
diff --git a/pkg/oomstore/entity.go b/pkg/oomstore/entity.go
index ab3900f5..f4ef3294 100644
--- a/pkg/oomstore/entity.go
+++ b/pkg/oomstore/entity.go
@@ -17,23 +17,9 @@ func (s *OomStore) GetEntityByName(ctx context.Context, name string) (*types.Ent
 	return s.metadata.GetEntityByName(ctx, name)
 }
 
-// ListEntity lists metadata of entities given entity names.
-func (s *OomStore) ListEntity(ctx context.Context, opt types.ListEntityOpt) (types.EntityList, error) {
-	entities, err := s.metadata.ListEntity(ctx, nil)
-	if err != nil {
-		return nil, err
-	}
-	if opt.EntityNames != nil {
-		entities = entities.Filter(func(e *types.Entity) bool {
-			for _, name := range *opt.EntityNames {
-				if e.Name == name {
-					return true
-				}
-			}
-			return false
-		})
-	}
-	return entities, nil
+// ListEntity lists metadata of all entities.
+func (s *OomStore) ListEntity(ctx context.Context) (types.EntityList, error) {
+	return s.metadata.ListEntity(ctx, nil)
 }
 
 // CreateEntity creates metadata for an entity.
diff --git a/pkg/oomstore/export.go b/pkg/oomstore/export.go
index d5069450..d14d0ca0 100644
--- a/pkg/oomstore/export.go
+++ b/pkg/oomstore/export.go
@@ -41,7 +41,15 @@ func (s *OomStore) ChannelExport(ctx context.Context, opt types.ChannelExportOpt
 	}
 
 	if len(features) != len(opt.FeatureNames) {
-		invalid := features.FindMissingFeatures(opt.FeatureNames)
+		invalid := make([]string, 0)
+		for _, name := range opt.FeatureNames {
+			f := features.Find(func(feature *types.Feature) bool {
+				return feature.FullName() == name
+			})
+			if f == nil {
+				invalid = append(invalid, name)
+			}
+		}
 		return nil, errdefs.Errorf("invalid feature names %s", invalid)
 	}
 
@@ -91,9 +99,6 @@ func (s *OomStore) ChannelExport(ctx context.Context, opt types.ChannelExportOpt
 // Export exports the latest feature value up to the given timestamp, it outputs
 // feature values to the given file path.
 func (s *OomStore) Export(ctx context.Context, opt types.ExportOpt) error {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
 	exportResult, err := s.ChannelExport(ctx, types.ChannelExportOpt{
 		FeatureNames: opt.FeatureNames,
 		UnixMilli:    opt.UnixMilli,
@@ -115,13 +120,9 @@ func (s *OomStore) Export(ctx context.Context, opt types.ExportOpt) error {
 		return err
 	}
 	for row := range exportResult.Data {
-		if row.Error != nil {
-			return row.Error
-		}
-
-		if err := w.Write(cast.ToStringSlice([]interface{}(row.Record))); err != nil {
+		if err := w.Write(cast.ToStringSlice([]interface{}(row))); err != nil {
 			return err
 		}
 	}
-	return nil
+	return exportResult.CheckStreamError()
 }
diff --git a/pkg/oomstore/feature.go b/pkg/oomstore/feature.go
index b89402dd..3ca0d094 100644
--- a/pkg/oomstore/feature.go
+++ b/pkg/oomstore/feature.go
@@ -34,25 +34,19 @@ func (s *OomStore) GetFeatureByName(ctx context.Context, groupName string, featu
 // ListFeature lists metadata of features meeting particular criteria.
 func (s *OomStore) ListFeature(ctx context.Context, opt types.ListFeatureOpt) (types.FeatureList, error) {
 	metadataOpt := metadata.ListFeatureOpt{}
-	if opt.EntityNames != nil {
-		entities, err := s.ListEntity(ctx, types.ListEntityOpt{
-			EntityNames: opt.EntityNames,
-		})
+	if opt.EntityName != nil {
+		entity, err := s.metadata.GetEntityByName(ctx, *opt.EntityName)
 		if err != nil {
 			return nil, err
 		}
-		entityIDs := entities.IDs()
-		metadataOpt.EntityIDs = &entityIDs
+		metadataOpt.EntityID = &entity.ID
 	}
-	if opt.GroupNames != nil {
-		groups, err := s.ListGroup(ctx, types.ListGroupOpt{
-			GroupNames: opt.GroupNames,
-		})
+	if opt.GroupName != nil {
+		group, err := s.metadata.GetGroupByName(ctx, *opt.GroupName)
 		if err != nil {
 			return nil, err
 		}
-		groupIDs := groups.IDs()
-		metadataOpt.GroupIDs = &groupIDs
+		metadataOpt.GroupID = &group.ID
 	}
 	features, err := s.metadata.ListFeature(ctx, metadataOpt)
 	if err != nil {
@@ -106,7 +100,7 @@ func (s *OomStore) CreateFeature(ctx context.Context, opt types.CreateFeatureOpt
 
 	if group.Category == types.CategoryStream {
 		features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
-			GroupIDs: &[]int{group.ID},
+			GroupID: &group.ID,
 		})
 		if err != nil {
 			return 0, err
diff --git a/pkg/oomstore/group.go b/pkg/oomstore/group.go
index 259a62bd..769b791f 100644
--- a/pkg/oomstore/group.go
+++ b/pkg/oomstore/group.go
@@ -38,34 +38,13 @@ func (s *OomStore) GetGroupByName(ctx context.Context, name string) (*types.Grou
 	return s.metadata.GetGroupByName(ctx, name)
 }
 
-// ListGroup lists metadata of feature groups meeting particular criteria.
-func (s *OomStore) ListGroup(ctx context.Context, opt types.ListGroupOpt) (types.GroupList, error) {
-	metadataOpt := metadata.ListGroupOpt{}
-	if opt.EntityNames != nil {
-		entities, err := s.ListEntity(ctx, types.ListEntityOpt{
-			EntityNames: opt.EntityNames,
-		})
-		if err != nil {
-			return nil, err
-		}
-		entityIDs := entities.IDs()
-		metadataOpt.EntityIDs = &entityIDs
+// ListGroup lists metadata of feature groups under the same entity.
+func (s *OomStore) ListGroup(ctx context.Context, entityID *int) (types.GroupList, error) {
+	opt := metadata.ListGroupOpt{}
+	if entityID != nil {
+		opt.EntityIDs = &[]int{*entityID}
 	}
-	groups, err := s.metadata.ListGroup(ctx, metadataOpt)
-	if err != nil {
-		return nil, err
-	}
-	if opt.GroupNames != nil {
-		groups = groups.Filter(func(g *types.Group) bool {
-			for _, name := range *opt.GroupNames {
-				if g.Name == name {
-					return true
-				}
-			}
-			return false
-		})
-	}
-	return groups, nil
+	return s.metadata.ListGroup(ctx, opt)
 }
 
 // UpdateGroup updates metadata of a feature group.
diff --git a/pkg/oomstore/import.go b/pkg/oomstore/import.go
index bb830006..4c01bf0b 100644
--- a/pkg/oomstore/import.go
+++ b/pkg/oomstore/import.go
@@ -67,7 +67,7 @@ func (s *OomStore) parseImportOpt(ctx context.Context, opt types.ImportOpt) (*im
 	}
 
 	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
-		GroupIDs: &[]int{group.ID},
+		GroupID: &group.ID,
 	})
 	if err != nil {
 		return nil, err
diff --git a/pkg/oomstore/revision.go b/pkg/oomstore/revision.go
index 49a9e995..3e651719 100644
--- a/pkg/oomstore/revision.go
+++ b/pkg/oomstore/revision.go
@@ -99,7 +99,7 @@ func (s *OomStore) createSnapshotAndCdcTable(ctx context.Context, revisionID int
 
 	// Create snapshot table in offline store
 	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
-		GroupIDs: &[]int{revision.GroupID},
+		GroupID: &revision.GroupID,
 	})
 	if err != nil {
 		return err
diff --git a/pkg/oomstore/snapshot.go b/pkg/oomstore/snapshot.go
index 07215a70..cd65cd14 100644
--- a/pkg/oomstore/snapshot.go
+++ b/pkg/oomstore/snapshot.go
@@ -19,7 +19,7 @@ func (s *OomStore) Snapshot(ctx context.Context, groupName string) error {
 	}
 
 	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
-		GroupIDs: &[]int{group.ID},
+		GroupID: &group.ID,
 	})
 	if err != nil {
 		return err
diff --git a/pkg/oomstore/sync.go b/pkg/oomstore/sync.go
index a536754c..50925088 100644
--- a/pkg/oomstore/sync.go
+++ b/pkg/oomstore/sync.go
@@ -30,16 +30,13 @@ func (s *OomStore) Sync(ctx context.Context, opt types.SyncOpt) error {
 
 // syncBatch syncs batch feature group from offline store to online store.
 func (s *OomStore) syncBatch(ctx context.Context, opt types.SyncOpt, group *types.Group, revision *types.Revision) error {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
 	prevOnlineRevisionID := group.OnlineRevisionID
 	if prevOnlineRevisionID != nil && *prevOnlineRevisionID == revision.ID {
 		return errdefs.Errorf("the specific revision was synced to the online store, won't do it again this time")
 	}
 
 	features, err := s.ListFeature(ctx, types.ListFeatureOpt{
-		GroupNames: &[]string{group.Name},
+		GroupName: &group.Name,
 	})
 	if err != nil {
 		return err
@@ -59,6 +56,7 @@ func (s *OomStore) syncBatch(ctx context.Context, opt types.SyncOpt, group *type
 		Group:        *group,
 		Features:     features,
 		ExportStream: exportResult.Data,
+		ExportError:  exportResult.GetErrorChannel(),
 		RevisionID:   &revision.ID,
 	}); err != nil {
 		return err
@@ -101,11 +99,8 @@ func (s *OomStore) syncBatch(ctx context.Context, opt types.SyncOpt, group *type
 
 // syncStream syncs stream feature group from offline store to online store.
 func (s *OomStore) syncStream(ctx context.Context, opt types.SyncOpt) error {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
 	features, err := s.ListFeature(ctx, types.ListFeatureOpt{
-		GroupNames: &[]string{opt.GroupName},
+		GroupName: &opt.GroupName,
 	})
 	if err != nil {
 		return err
@@ -123,6 +118,7 @@ func (s *OomStore) syncStream(ctx context.Context, opt types.SyncOpt) error {
 		Group:        *group,
 		Features:     features,
 		ExportStream: exportResult.Data,
+		ExportError:  exportResult.GetErrorChannel(),
 	})
 }
 
diff --git a/pkg/oomstore/types/apply/apply.go b/pkg/oomstore/types/apply/apply.go
index adcc9f17..cb4f616f 100644
--- a/pkg/oomstore/types/apply/apply.go
+++ b/pkg/oomstore/types/apply/apply.go
@@ -44,7 +44,7 @@ func (f FeatureItems) Walk(walkFunc func(Feature) Feature) (rs FeatureItems) {
 	return
 }
 
-func BuildFeatureItems(features types.FeatureList) FeatureItems {
+func FromFeatureList(features types.FeatureList) FeatureItems {
 	items := FeatureItems{
 		Items: make([]Feature, 0, features.Len()),
 	}
@@ -91,8 +91,8 @@ func (g GroupItems) Walk(walkFunc func(Group) Group) (rs GroupItems) {
 	return
 }
 
-func BuildGroupItems(groups types.GroupList, features types.FeatureList) GroupItems {
-	items := GroupItems{
+func FromGroupList(groups types.GroupList, features types.FeatureList) *GroupItems {
+	items := &GroupItems{
 		Items: make([]Group, 0, groups.Len()),
 	}
 
@@ -104,7 +104,7 @@ func BuildGroupItems(groups types.GroupList, features types.FeatureList) GroupIt
 			Category:         group.Category,
 			SnapshotInterval: time.Duration(group.SnapshotInterval) * time.Second,
 			Description:      group.Description,
-			Features: BuildFeatureItems(features.Filter(func(f *types.Feature) bool {
+			Features: FromFeatureList(features.Filter(func(f *types.Feature) bool {
 				return f.Group.Name == group.Name
 			})).Walk(func(f Feature) Feature {
 				f.Kind = ""
@@ -129,7 +129,7 @@ type EntityItems struct {
 	Items []Entity `yaml:"items"`
 }
 
-func BuildEntityItems(entities types.EntityList, groups GroupItems) (items EntityItems) {
+func FromEntityList(entities types.EntityList, groups *GroupItems) (items EntityItems) {
 	for _, entity := range entities {
 		items.Items = append(items.Items, Entity{
 			Kind:        "Entity",
diff --git a/pkg/oomstore/types/apply/validate.go b/pkg/oomstore/types/apply/validate.go
index c45ec300..9646d8f4 100644
--- a/pkg/oomstore/types/apply/validate.go
+++ b/pkg/oomstore/types/apply/validate.go
@@ -12,9 +12,9 @@ import (
 )
 
 var (
-	entityValidator  jsonschema.Schema
-	groupValidator   jsonschema.Schema
-	featureValidator jsonschema.Schema
+	entityValidate  jsonschema.Schema
+	groupValidate   jsonschema.Schema
+	featureValidate jsonschema.Schema
 
 	//go:embed entity.json
 	entitySchema string
@@ -25,15 +25,15 @@ var (
 )
 
 func init() {
-	if err := json.Unmarshal([]byte(entitySchema), &entityValidator); err != nil {
+	if err := json.Unmarshal([]byte(entitySchema), &entityValidate); err != nil {
 		panic(err)
 	}
 
-	if err := json.Unmarshal([]byte(groupSchema), &groupValidator); err != nil {
+	if err := json.Unmarshal([]byte(groupSchema), &groupValidate); err != nil {
 		panic(err)
 	}
 
-	if err := json.Unmarshal([]byte(featureSchema), &featureValidator); err != nil {
+	if err := json.Unmarshal([]byte(featureSchema), &featureValidate); err != nil {
 		panic(err)
 	}
 }
@@ -44,7 +44,7 @@ func (e *Entity) Validate() error {
 		return errdefs.WithStack(err)
 	}
 
-	return validate(fmt.Sprintf("Entity %s", e.Name), data, &entityValidator)
+	return validate(fmt.Sprintf("Entity %s", e.Name), data, &entityValidate)
 }
 
 func (g *Group) Validate() error {
@@ -53,7 +53,7 @@ func (g *Group) Validate() error {
 		return errdefs.WithStack(err)
 	}
 
-	return validate(fmt.Sprintf("Group %s", g.Name), data, &groupValidator)
+	return validate(fmt.Sprintf("Group %s", g.Name), data, &groupValidate)
 }
 
 func (f *Feature) Validate() error {
@@ -62,7 +62,7 @@ func (f *Feature) Validate() error {
 		return err
 	}
 
-	return validate(fmt.Sprintf("Feature %s", f.Name), data, &featureValidator)
+	return validate(fmt.Sprintf("Feature %s", f.Name), data, &featureValidate)
 }
 
 func validate(errPrefix string, data []byte, schema *jsonschema.Schema) error {
diff --git a/pkg/oomstore/types/entity.go b/pkg/oomstore/types/entity.go
index 89f9ef46..81eada28 100644
--- a/pkg/oomstore/types/entity.go
+++ b/pkg/oomstore/types/entity.go
@@ -38,8 +38,8 @@ func (l EntityList) Len() int {
 	return len(l)
 }
 
-func (l EntityList) Find(find func(*Entity) bool) *Entity {
-	for _, e := range l {
+func (l *EntityList) Find(find func(*Entity) bool) *Entity {
+	for _, e := range *l {
 		if find(e) {
 			return e
 		}
@@ -47,25 +47,11 @@ func (l EntityList) Find(find func(*Entity) bool) *Entity {
 	return nil
 }
 
-func (l EntityList) Filter(filter func(*Entity) bool) (rs EntityList) {
-	for _, e := range l {
+func (l *EntityList) Filter(filter func(*Entity) bool) (rs EntityList) {
+	for _, e := range *l {
 		if filter(e) {
 			rs = append(rs, e)
 		}
 	}
 	return
 }
-
-func (l EntityList) IDs() (ids []int) {
-	for _, entity := range l {
-		ids = append(ids, entity.ID)
-	}
-	return
-}
-
-func (l EntityList) Names() (names []string) {
-	for _, entity := range l {
-		names = append(names, entity.Name)
-	}
-	return
-}
diff --git a/pkg/oomstore/types/feature.go b/pkg/oomstore/types/feature.go
index 077e009c..6c182628 100644
--- a/pkg/oomstore/types/feature.go
+++ b/pkg/oomstore/types/feature.go
@@ -171,16 +171,3 @@ func (l FeatureList) GetSharedEntity() (*Entity, error) {
 	}
 	return nil, errdefs.Errorf("expected 1 entity, got 0")
 }
-
-func (l FeatureList) FindMissingFeatures(featureNames []string) []string {
-	missing := make([]string, 0)
-	for _, name := range featureNames {
-		f := l.Find(func(feature *Feature) bool {
-			return feature.FullName() == name
-		})
-		if f == nil {
-			missing = append(missing, name)
-		}
-	}
-	return missing
-}
diff --git a/pkg/oomstore/types/group.go b/pkg/oomstore/types/group.go
index 31138be5..4b9a8d24 100644
--- a/pkg/oomstore/types/group.go
+++ b/pkg/oomstore/types/group.go
@@ -21,11 +21,11 @@ type Group struct {
 	Entity *Entity
 }
 
-func (g *Group) Copy() *Group {
-	if g == nil {
+func (fg *Group) Copy() *Group {
+	if fg == nil {
 		return nil
 	}
-	copied := *g
+	copied := *fg
 
 	if copied.OnlineRevisionID != nil {
 		id := *copied.OnlineRevisionID
@@ -38,8 +38,8 @@ func (g *Group) Copy() *Group {
 
 type GroupList []*Group
 
-func (l GroupList) Len() int {
-	return len(l)
+func (l *GroupList) Len() int {
+	return len(*l)
 }
 
 func (l GroupList) Copy() GroupList {
@@ -82,17 +82,3 @@ func (l GroupList) EntityIDs() []int {
 	}
 	return entityIDs
 }
-
-func (l GroupList) IDs() (ids []int) {
-	for _, group := range l {
-		ids = append(ids, group.ID)
-	}
-	return
-}
-
-func (l GroupList) Names() (names []string) {
-	for _, group := range l {
-		names = append(names, group.Name)
-	}
-	return
-}
diff --git a/pkg/oomstore/types/options.go b/pkg/oomstore/types/options.go
index 59d0206c..ecf16d49 100644
--- a/pkg/oomstore/types/options.go
+++ b/pkg/oomstore/types/options.go
@@ -8,8 +8,8 @@ type CreateFeatureOpt struct {
 }
 
 type ListFeatureOpt struct {
-	EntityNames  *[]string
-	GroupNames   *[]string
+	EntityName   *string
+	GroupName    *string
 	FeatureNames *[]string
 }
 
@@ -23,10 +23,6 @@ type CreateEntityOpt struct {
 	Description string
 }
 
-type ListEntityOpt struct {
-	EntityNames *[]string
-}
-
 type CreateGroupOpt struct {
 	GroupName        string
 	EntityName       string
@@ -35,11 +31,6 @@ type CreateGroupOpt struct {
 	Description      string
 }
 
-type ListGroupOpt struct {
-	EntityNames *[]string
-	GroupNames  *[]string
-}
-
 type ChannelExportOpt struct {
 	FeatureNames []string
 	UnixMilli    int64
diff --git a/pkg/oomstore/types/types.go b/pkg/oomstore/types/types.go
index 0bb9b959..54444c3d 100644
--- a/pkg/oomstore/types/types.go
+++ b/pkg/oomstore/types/types.go
@@ -15,17 +15,14 @@ const (
 	TableStreamCdc      TableType = "stream_cdc"
 )
 
-type ExportRecord struct {
-	Record []interface{}
-	Error  error
-}
+type ExportRecord []interface{}
 
 func (r ExportRecord) EntityKey() string {
-	return r.Record[0].(string)
+	return r[0].(string)
 }
 
 func (r ExportRecord) ValueAt(i int) interface{} {
-	return r.Record[i+1]
+	return r[i+1]
 }
 
 type EntityRow struct {
@@ -43,13 +40,31 @@ type JoinResult struct {
 type ExportResult struct {
 	Header []string
 	Data   <-chan ExportRecord
+	error  <-chan error
 }
 
-func NewExportResult(header []string, data <-chan ExportRecord) *ExportResult {
+func NewExportResult(header []string, data <-chan ExportRecord, error <-chan error) *ExportResult {
 	return &ExportResult{
 		Header: header,
 		Data:   data,
+		error:  error,
+	}
+}
+
+// ATTENTION: call this method only after you consume all elements
+// from Data channel; otherwise, it will block the Data channel.
+func (e *ExportResult) CheckStreamError() error {
+	if e == nil {
+		return nil
 	}
+	if e.error != nil {
+		return <-e.error
+	}
+	return nil
+}
+
+func (e *ExportResult) GetErrorChannel() <-chan error {
+	return e.error
 }
 
 type DataTableTimeRange struct {
diff --git a/pkg/oomstore/util/feature.go b/pkg/oomstore/util/feature.go
index 85183ca8..64e45310 100644
--- a/pkg/oomstore/util/feature.go
+++ b/pkg/oomstore/util/feature.go
@@ -12,19 +12,19 @@ func ComposeFullFeatureName(groupName string, featureName string) string {
 	return groupName + SepFullFeatureName + featureName
 }
 
-func SplitFullFeatureName(fullName string) (string, string, error) {
-	parts := strings.SplitN(fullName, SepFullFeatureName, 2)
+func SplitFullFeatureName(fullname string) (string, string, error) {
+	parts := strings.SplitN(fullname, SepFullFeatureName, 2)
 	if len(parts) != 2 {
-		return "", "", errdefs.Errorf("invalid full feature name: '%s'", fullName)
+		return "", "", errdefs.Errorf("invalid full feature name: '%s'", fullname)
 	}
 	return parts[0], parts[1], nil
 }
 
-func ValidateFullFeatureNames(fullNames ...string) error {
-	for _, fullName := range fullNames {
-		nameSlice := strings.Split(fullName, SepFullFeatureName)
+func ValidateFullFeatureNames(fullnames ...string) error {
+	for _, fullname := range fullnames {
+		nameSlice := strings.Split(fullname, SepFullFeatureName)
 		if len(nameSlice) != 2 {
-			return errdefs.Errorf("invalid full feature name: '%s'", fullName)
+			return errdefs.Errorf("invalid full feature name: '%s'", fullname)
 		}
 	}
 	return nil
diff --git a/sdk/python/Cargo.lock b/sdk/python/Cargo.lock
index 4216d026..2df490ad 100644
--- a/sdk/python/Cargo.lock
+++ b/sdk/python/Cargo.lock
@@ -73,6 +73,12 @@ version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c4872d67bab6358e59559027aa3b9157c53d9358c51423c17554809a8858e0f8"
 
+[[package]]
+name = "cc"
+version = "1.0.72"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "22a9137b95ea06864e018375b72adfb7db6e6f68cfc8df5a04d00288050485ee"
+
 [[package]]
 name = "cfg-if"
 version = "1.0.0"
@@ -461,6 +467,15 @@ version = "2.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "308cc39be01b73d0d18f82a0e7b2a3df85245f84af96fdddc5d202d27e47b86a"
 
+[[package]]
+name = "memoffset"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5aa361d4faea93603064a027415f07bd8e1d5c88c9fbf68bf56a285428fd79ce"
+dependencies = [
+ "autocfg",
+]
+
 [[package]]
 name = "mio"
 version = "0.7.14"
@@ -489,6 +504,19 @@ version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e5ce46fe64a9d73be07dcbe690a38ce1b293be448fd8ce1e6c1b8062c9f72c6a"
 
+[[package]]
+name = "nix"
+version = "0.23.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9f866317acbd3a240710c63f065ffb1e4fd466259045ccb504130b7f668f35c6"
+dependencies = [
+ "bitflags",
+ "cc",
+ "cfg-if",
+ "libc",
+ "memoffset",
+]
+
 [[package]]
 name = "ntapi"
 version = "0.3.6"
@@ -516,13 +544,14 @@ checksum = "da32515d9f6e6e489d7bc9d84c71b060db7247dc035bbe44eac88cf87486d8d5"
 
 [[package]]
 name = "oomclient"
-version = "0.1.0-rc.7"
+version = "0.1.0-rc.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1eaa2a63496b4b8e9f623d900767ee52be101740871eb94fdbf7d04f47f62d6a"
+checksum = "3d3b8230dd2213e41ff44654f5d52246372cbf7923117330eed97cc7c632025c"
 dependencies = [
  "async-stream",
  "futures-core",
  "itertools",
+ "nix",
  "prost",
  "signal-hook",
  "signal-hook-tokio",
diff --git a/sdk/python/Cargo.toml b/sdk/python/Cargo.toml
index 7839fcd9..7305e23a 100644
--- a/sdk/python/Cargo.toml
+++ b/sdk/python/Cargo.toml
@@ -9,7 +9,7 @@ name = "oomclient"
 crate-type = ["cdylib"]
 
 [dependencies]
-oomclient = "0.1.0-rc.7"
+oomclient = "0.1.0-rc.8"
 pyo3-asyncio = { version = "0.15", features = ["tokio-runtime"] }
 pyo3 = { version = "0.15", features = ["extension-module"] }
 tokio = { version = "1.15", features = ["macros", "rt-multi-thread"] }
diff --git a/sdk/python/Makefile.toml b/sdk/python/Makefile.toml
index 11bc20b2..f254f68b 100644
--- a/sdk/python/Makefile.toml
+++ b/sdk/python/Makefile.toml
@@ -1,17 +1,25 @@
 [tasks.maturin]
+dependencies = ["pyenv"]
 script = '''
 source .env/bin/activate
 maturin develop
 '''
 
 [tasks.ipython]
+dependencies = ["pyenv"]
 script = '''
 source .env/bin/activate
 ipython
 '''
 
 [tasks.python]
+dependencies = ["pyenv"]
 script = '''
 source .env/bin/activate
 python "${@}"
 '''
+
+[tasks.pyenv]
+script = '''
+test -d .env || python -m venv .env
+'''
diff --git a/sdk/python/src/lib.rs b/sdk/python/src/lib.rs
index 00f9d647..dd5f386b 100644
--- a/sdk/python/src/lib.rs
+++ b/sdk/python/src/lib.rs
@@ -18,15 +18,7 @@ pub struct Client {
 
 #[pymethods]
 impl Client {
-    /// Connect to an oomagent instance running on the given endpoint.
-    ///
-    /// Args:
-    ///     endpoint (str): The endpoint to connect to.
-    ///
-    /// Returns:
-    ///     The client instance.
     #[classmethod]
-    #[pyo3(text_signature = "(endpoint)")]
     pub fn connect<'p>(_cls: &PyType, py: Python<'p>, endpoint: String) -> PyResult<&'p PyAny> {
         future_into_py(py, async {
             let inner = OomClient::connect(endpoint).await.map_err(err_to_py)?;
@@ -35,16 +27,7 @@ impl Client {
         })
     }
 
-    /// Connect to an embedded oomagent instance.
-    ///
-    /// Args:
-    ///     bin_path (str): The path to the oomagent binary.
-    ///     cfg_path (str): The path to the oomstore configuration file.
-    ///
-    /// Returns:
-    ///    The client instance.
     #[classmethod]
-    #[pyo3(text_signature = "(bin_path, cfg_path)")]
     pub fn with_embedded_oomagent<'p>(
         _cls: &PyType,
         py: Python<'p>,
@@ -60,8 +43,6 @@ impl Client {
         })
     }
 
-    /// Check if oomagent is ready to serve requests.
-    #[pyo3(text_signature = "()")]
     pub fn health_check<'p>(&self, py: Python<'p>) -> PyResult<&'p PyAny> {
         // Don't panic, it's cheap:
         // https://github.com/hyperium/tonic/issues/285#issuecomment-595880400
@@ -69,17 +50,6 @@ impl Client {
         future_into_py(py, async move { inner.health_check().await.map_err(err_to_py) })
     }
 
-    /// Get online features for an entity.
-    ///
-    /// Args:
-    ///     entity_key (str): An entity identifier, could be device ID, user ID, etc.
-    ///     features: A list of feature full names.
-    ///       A feature full name has the following format: &lt;group_name&gt;.&lt;feature_name&gt;,
-    ///       for example, txn_stats.count_7d.
-    ///
-    /// Returns:
-    ///     A dict mapping feature full names to feature values.
-    #[pyo3(text_signature = "(entity_key, features)")]
     pub fn online_get<'p>(&self, py: Python<'p>, entity_key: String, features: Vec<String>) -> PyResult<&'p PyAny> {
         let mut inner = OomClient::clone(&self.inner);
         future_into_py(py, async move {
@@ -91,17 +61,6 @@ impl Client {
         })
     }
 
-    /// Get online features for multiple entities.
-    ///
-    /// Args:
-    ///     entity_keys: A list of entity identifiers, could be device IDs, user IDs, etc.
-    ///     features: A list of feature full names.
-    ///       A feature full name has the following format: &lt;group_name&gt;.&lt;feature_name&gt;,
-    ///       for example, txn_stats.count_7d.
-    ///
-    /// Returns:
-    ///     A dict mapping entity keys to a dict, that maps feature full names to feature values.
-    #[pyo3(text_signature = "(entity_keys, features)")]
     pub fn online_multi_get<'p>(
         &self,
         py: Python<'p>,
@@ -124,22 +83,6 @@ impl Client {
         })
     }
 
-    /// Sync a certain revision of batch features from offline to online store.
-    ///
-    /// Args:
-    ///     group: The group to sync from offline store to online store.
-    ///     revision_id: The revision to sync, it only applies to batch feature.
-    ///       For batch feature: if null, will sync the latest revision;
-    ///       otherwise, sync the designated revision.
-    ///       For streaming feature, revision ID is not required, will always
-    ///       sync the latest values.
-    ///     purge_delay: PurgeDelay represents the seconds to sleep before purging
-    ///       the previous revision in online store.
-    ///       It only applies to batch feature group.
-    ///
-    /// Returns:
-    ///     None
-    #[pyo3(text_signature = "(group, revision_id, purge_delay)")]
     pub fn sync<'p>(
         &mut self,
         py: Python<'p>,
@@ -153,23 +96,7 @@ impl Client {
         })
     }
 
-    /// Import features from external (batch and stream) data sources to offline store through files.
-    ///
-    /// Args:
-    ///     group: The group to be imported from data source to offline store.
-    ///     revision: The revision of the imported data, it only applies to
-    ///       batch feature (not required).
-    ///       For batch features, if revision is null, will use the
-    ///       timestamp when it starts serving in online store; otherwise,
-    ///       use the designated revision.
-    ///     description: Description of this import.
-    ///     input_file: The path of data source.
-    ///     delimiter: Delimiter of data source.
-    ///
-    /// Returns:
-    ///     revision_id: The revision ID of this import, it only applies to batch feature.
     #[pyo3(name = "import_")]
-    #[pyo3(text_signature = "(group, revision, description, input_file, delimiter)")]
     pub fn import<'p>(
         &mut self,
         py: Python<'p>,
@@ -198,16 +125,6 @@ impl Client {
         })
     }
 
-    /// Push stream features from stream data source to both offline and online stores.
-    ///
-    /// Args:
-    ///     group: The group to be pushed from data source to offline store.
-    ///     entity_key: An entity identifier.
-    ///     kv_pairs: Feature values maps feature name to feature value.
-    ///
-    /// Returns:
-    ///     None
-    #[pyo3(text_signature = "(group, entity_key, kv_pairs)")]
     pub fn push<'p>(
         &mut self,
         py: Python<'p>,
@@ -225,17 +142,6 @@ impl Client {
         })
     }
 
-    /// Point-in-Time Join features against labeled entity rows through files.
-    ///
-    /// Args:
-    ///     features: A list of feature full names, their feature values will be
-    ///       joined and fetched from offline store.
-    ///     input_file: File path of entity rows.
-    ///     output_file: File path of joined result.
-    ///
-    /// Returns:
-    ///     None
-    #[pyo3(text_signature = "(features, input_file, output_file)")]
     pub fn join<'p>(
         &mut self,
         py: Python<'p>,
@@ -249,17 +155,6 @@ impl Client {
         })
     }
 
-    /// Export certain features to a file.
-    ///
-    /// Args:
-    ///     features: A list of feature full names.
-    ///     unix_milli: A unix milliseconds, export the feature value before this timestamp.
-    ///     output_file: File path of export result.
-    ///     limit: Limit the size of export data.
-    ///
-    /// Returns:
-    ///     None
-    #[pyo3(text_signature = "(features, unix_milli, output_file, limit)")]
     pub fn export<'p>(
         &mut self,
         py: Python<'p>,
@@ -277,14 +172,6 @@ impl Client {
         })
     }
 
-    /// Take snapshot for a stream feature group in offline store.
-    ///
-    /// Args:
-    ///     group: A streaming feature group.
-    ///
-    /// Returns:
-    ///     None
-    #[pyo3(text_signature = "(group)")]
     pub fn snapshot<'p>(&mut self, py: Python<'p>, group: String) -> PyResult<&'p PyAny> {
         let mut inner = OomClient::clone(&self.inner);
         future_into_py(py, async move { inner.snapshot(group).await.map_err(err_to_py) })
diff --git a/sdk/rust/Cargo.toml b/sdk/rust/Cargo.toml
index 0afd1db2..999bf01b 100644
--- a/sdk/rust/Cargo.toml
+++ b/sdk/rust/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "oomclient"
-version = "0.1.0-rc.7"
+version = "0.1.0-rc.8"
 edition = "2021"
 authors = ["Wenxuan Zhang <wenxuangm@gmail.com>"]
 description = "oomstore client for rust"
@@ -19,6 +19,7 @@ futures-core = "0.3"
 tokio-stream = "0.1"
 signal-hook = "0.3"
 signal-hook-tokio = {version = "0.3", features = ["futures-v0_3"]}
+nix = "0.23"
 
 [build-dependencies]
 tonic-build = "0.6"
diff --git a/sdk/rust/proto/oomagent.proto b/sdk/rust/proto/oomagent.proto
index 316deb3f..e7034248 100644
--- a/sdk/rust/proto/oomagent.proto
+++ b/sdk/rust/proto/oomagent.proto
@@ -3,23 +3,36 @@ package oomagent;
 option go_package = "/codegen";
 
 service OomAgent {
+  // Get online features for an entity.
   rpc OnlineGet(OnlineGetRequest) returns (OnlineGetResponse) {}
+  // Get online features for multiple entities.
   rpc OnlineMultiGet(OnlineMultiGetRequest) returns (OnlineMultiGetResponse) {}
+  // Sync a certain revision of batch features from offline to online store.
   rpc Sync(SyncRequest) returns (SyncResponse) {}
-
+  // Import features from external (batch and stream) data sources to offline store through channels.
   rpc ChannelImport(stream ChannelImportRequest) returns (ImportResponse) {}
+  // Import features from external (batch and stream) data sources to offline store through files.
   rpc Import(ImportRequest) returns (ImportResponse) {}
+  // Push stream features from stream data source to both offline and online stores.
   rpc Push(PushRequest) returns (PushResponse) {}
+  // Take snapshot for a stream feature group in offline store.
   rpc Snapshot(SnapshotRequest) returns (SnapshotResponse) {}
-
+  // Point-in-Time Join features against labeled entity rows through channels.
   rpc ChannelJoin(stream ChannelJoinRequest) returns (stream ChannelJoinResponse) {}
+  // Point-in-Time Join features against labeled entity rows through files.
   rpc Join(JoinRequest) returns (JoinResponse) {}
-
+  // Export certain features to a channel.
   rpc ChannelExport(ChannelExportRequest) returns (stream ChannelExportResponse) {}
+  // Export certain features to a file.
   rpc Export(ExportRequest) returns (ExportResponse) {}
+  // Check if oomagent is ready to serve requests.
   rpc HealthCheck(HealthCheckRequest) returns (HealthCheckResponse) {}
 }
 
+// `Value` represents a dynamically typed value which can be either
+//  a int64, a double, a string, a bool, a unix milliseconds, or a
+// bytes. A producer of value is expected to set one of these
+// variants.
 message Value {
   oneof value {
     // Represents a int64 value.
@@ -37,124 +50,212 @@ message Value {
   }
 }
 
+// `OnlineGetRequest` is a request for `OnlineGet` API.
 message OnlineGetRequest {
+  // An entity identifier, could be device ID, user ID, etc.
   string entity_key = 1;
+  // A list of feature full names.
+  // A feature full name has the following format: <feature_group_name>.<feature_name>,
+  // for example, txn_stats.count_7d
   repeated string features = 2;
 }
 
+// `FeatureValueMap` maps feature full name to feature value.
 message FeatureValueMap {
   map<string, Value> map = 1;
 }
 
+// `OnlineGetRequest` is a response for `OnlineGet` API.
 message OnlineGetResponse {
+  // The result of `OnlineGet`.
   FeatureValueMap result = 1;
 }
 
+// `OnlineMultiGetRequest` is a request for `OnlineMultiGet` API.
 message OnlineMultiGetRequest {
+  // A list of entity identifiers, could be device IDs, user IDs, etc.
   repeated string entity_keys = 1;
+  // A list of feature full names.
+  // A feature full name has the following format: <feature_group_name>.<feature_name>,
+  // for example, txn_stats.count_7d
   repeated string features = 2;
 }
 
+// `OnlineMultiGetResponse` is a response for `OnlineMultiGet` API.
 message OnlineMultiGetResponse {
+  // The result of `OnlineMultiGet`, mapping entity key to FeatureValueMap.
   map<string, FeatureValueMap> result = 1;
 }
 
+// `SyncRequest` is a request for `Sync` API.
 message SyncRequest {
+  // The group to sync from offline store to online store.
   string group = 1;
+  // The revision to sync, it only applies to batch feature.
+  // For batch feature: if null, will sync the latest revision;
+  // otherwise, sync the designated revision.
+  // For streaming feature, revision ID is not required, will always
+  // sync the latest values.
   optional int32 revision_id = 2;
+  // PurgeDelay represents the seconds to sleep before purging
+  // the previous revision in online store.
+  // It only applies to batch feature group.
   int32 purge_delay = 3;
 }
 
+// `SyncResponse` is a response for `Sync` API.
 message SyncResponse {}
 
+// `ChannelImportRequest` is a request for `ChannelImport` API.
 message ChannelImportRequest {
-  // only takes effect (and required) on the first request
+  // The group to be imported from data source to offline store.
+  // It only takes effect (and required) on the first request.
   optional string group = 1;
-  // only takes effect on the first request
+  // The revision of the imported data, it only applies to
+  // batch feature (not required).
+  // For batch features, if revision is null, will use the
+  // timestamp when it starts serving in online store; otherwise,
+  // use the designated revision.
+  // It only takes effect on the first request.
   optional int64 revision = 2;
-  // only takes effect on the first request
+  // Description of this import.
+  // It only takes effect on the first request.
   optional string description = 3;
 
+  // A single row of channel import.
   bytes row = 4;
 }
 
+// `ImportResponse` is a response for `ChannelImport` and `Import` API.
 message ImportResponse {
+  // The revision ID of this import, it only applies to batch feature.
   int32 revision_id = 1;
 }
 
+// `ImportRequest` is a request for `Import` API.
 message ImportRequest {
+  // The group to be imported from data source to offline store.
   string group = 1;
+  // The revision of the imported data, it only applies to
+  // batch feature (not required).
+  // For batch features, if revision is null, will use the
+  // timestamp when it starts serving in online store; otherwise,
+  // use the designated revision.
   optional int64 revision = 2;
+  // Description of this import.
   optional string description = 3;
-
+  // The path of data source.
   string input_file = 4;
+  // Delimiter of data source
   optional string delimiter = 5;
 }
 
-message PushResponse {}
-
+// `PushRequest` is a request for `Push` API.
 message PushRequest {
+  // An entity identifier.
   string entity_key = 1;
+  // A streaming feature group.
   string group = 2;
+  // Feature values maps feature name to feature value.
   map<string, Value> feature_values = 3;
 }
 
-message SnapshotResponse {}
+// `PushResponse` is a response for `Push` API.
+message PushResponse {}
 
+// `SnapshotRequest` is a request for `Snapshot` API.
 message SnapshotRequest {
+  // A streaming feature group.
   string group = 1;
 }
 
+// `SnapshotResponse` is a response for `Snapshot` API.
+message SnapshotResponse {}
+
+// `EntityRow` represents a row in `Join` request.
 message EntityRow {
+  // An entity identifier, could represents device ID, user ID, etc.
   string entity_key = 1;
+  // A unix milliseconds, represents the record timestamp.
   int64 unix_milli = 2;
+  // A list of existing values, could be label or real-time features.
   repeated string values = 3;
 }
 
+// `ChannelJoinRequest` is a request for `ChannelJoin` API.
 message ChannelJoinRequest {
-  // only takes effect (and required) on the first request
+  // A list of feature full names, their feature values will be
+  // joined and fetched from offline store.
+  // It only takes effect (and required) on the first request.
   repeated string join_features = 1;
-  // only takes effect (and required) on the first request
+  // A list of names, could be label name or real-time feature names.
+  // Those feature values will be passed from entity row.
+  // It only takes effect (and required) on the first request.
   repeated string existed_features = 2;
 
+  // An entity row.
   EntityRow entity_row = 3;
 }
 
+// `ChannelJoinResponse` is a response for `ChannelJoin` API.
 message ChannelJoinResponse {
-  // only appears in the first request
+  // Header of the `Join` response.
+  // It only appears in the first request.
   repeated string header = 1;
+  // A single row of joined results.
   repeated Value joined_row = 2;
 }
 
+// `JoinRequest` is a request for `Join` API.
 message JoinRequest {
+  // A list of feature full names, their feature values will be
+  // joined and fetched from offline store.
   repeated string features = 1;
+  // File path of entity rows.
   string input_file = 2;
+  // File path of joined result.
   string output_file = 3;
 }
 
+// `JoinResponse` is a response for `Join` API.
 message JoinResponse {}
 
+// `ChannelExportRequest` is a request for `ChannelExport` API.
 message ChannelExportRequest {
+  // A list of feature full names.
   repeated string features = 1;
+  // A unix milliseconds, export the feature value before this timestamp.
   int64 unix_milli = 2;
+  // Limit the size of export data.
   optional uint64 limit = 3;
 }
 
+// `ChannelExportResponse` is a response for `ChannelExport` API.
 message ChannelExportResponse {
-  // only appears in the first request
+  // Header of export result.
+  // It only appears in the first request.
   repeated string header = 1;
+  // A single row of export result.
   repeated Value row = 2;
 }
 
+// `ExportRequest` is a request for `Export` API.
 message ExportRequest {
+  // A list of feature full names.
   repeated string features = 1;
+  // A unix milliseconds, export the feature value before this timestamp.
   int64 unix_milli = 2;
+  // File path of export result.
   string output_file = 3;
+  // Limit the size of export data.
   optional uint64 limit = 4;
 }
 
+// `ExportResponse` is a response for `Export` API.
 message ExportResponse {}
 
+// `HealthCheckRequesst` is a request for `HealthCheck` API.
 message HealthCheckRequest {}
 
+// `HealthCheckResponse` is a response for `HealthCheck` API.
 message HealthCheckResponse {}
diff --git a/sdk/rust/src/server.rs b/sdk/rust/src/server.rs
index 9c72b127..4840ee4d 100644
--- a/sdk/rust/src/server.rs
+++ b/sdk/rust/src/server.rs
@@ -1,3 +1,7 @@
+use nix::{
+    sys::signal::{self, Signal},
+    unistd::Pid,
+};
 use signal_hook::{consts::signal::*, low_level::emulate_default_handler};
 use signal_hook_tokio::{Handle, Signals};
 use std::{
@@ -14,6 +18,7 @@ use crate::Result;
 pub struct ServerWrapper {
     addr:   SocketAddr,
     handle: Handle,
+    pid:    u32,
 }
 
 // TODO: Add a Builder to create the server wrapper
@@ -35,22 +40,21 @@ impl ServerWrapper {
         if let Some(cfg_path) = cfg_path.clone() {
             oomagent.arg("--config").arg(cfg_path);
         }
-        oomagent.kill_on_drop(true);
 
-        let mut child = oomagent.spawn()?;
-        let pid = child.id();
+        let child = oomagent.spawn()?;
+        let pid = child.id().expect("failed to get child pid");
 
         tokio::spawn({
             async move {
                 while let Some(signal) = signals.next().await {
-                    child.kill().await.unwrap();
-                    emulate_default_handler(signal).unwrap();
+                    graceful_kill(pid, signal);
+                    emulate_default_handler(signal).expect("failed to emulate default signal handler");
                 }
             }
         });
 
-        let addr = get_agent_address(pid.unwrap()).await?;
-        Ok(Self { handle, addr })
+        let addr = get_agent_address(pid).await?;
+        Ok(Self { addr, handle, pid })
     }
 
     pub async fn default() -> Result<Self> {
@@ -73,6 +77,7 @@ impl ServerWrapper {
 impl Drop for ServerWrapper {
     fn drop(&mut self) {
         self.handle.close();
+        graceful_kill(self.pid, SIGTERM);
     }
 }
 
@@ -94,3 +99,14 @@ async fn get_agent_address(pid: u32) -> Result<SocketAddr> {
         time::sleep(Duration::from_millis(200)).await;
     }
 }
+
+fn graceful_kill(pid: u32, signal: i32) {
+    let signal = match signal {
+        SIGHUP => Signal::SIGHUP,
+        SIGTERM => Signal::SIGTERM,
+        SIGINT => Signal::SIGINT,
+        SIGQUIT => Signal::SIGQUIT,
+        _ => panic!("unexpected signal: {}", signal),
+    };
+    signal::kill(Pid::from_raw(pid as i32), signal).expect("failed to send SIGTERM signal to child process");
+}


Test patch:
diff --git a/internal/database/metadata/test_impl/feature.go b/internal/database/metadata/test_impl/feature.go
index 1248f066..7c6a4b8b 100644
--- a/internal/database/metadata/test_impl/feature.go
+++ b/internal/database/metadata/test_impl/feature.go
@@ -233,7 +233,7 @@ func TestListFeature(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 
 	// case 4: list features by EntityID and FeatureIDs
 	features, err = store.ListFeature(ctx, metadata.ListFeatureOpt{
-		EntityIDs:  &[]int{entityID + 1},
+		EntityID:   intPtr(entityID + 1),
 		FeatureIDs: &[]int{featureID},
 	})
 	assert.NoError(t, err)
@@ -241,7 +241,7 @@ func TestListFeature(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 
 	// case 5: list features by GroupID and FeatureIDs
 	features, err = store.ListFeature(ctx, metadata.ListFeatureOpt{
-		GroupIDs:   &[]int{groupID + 1},
+		GroupID:    intPtr(groupID + 1),
 		FeatureIDs: &[]int{featureID},
 	})
 	assert.NoError(t, err)
@@ -249,32 +249,18 @@ func TestListFeature(t *testing.T, prepareStore PrepareStoreFn, destroyStore Des
 
 	// case 6: list features by EntityID and empty FeatureIDs, return no feature
 	features, err = store.ListFeature(ctx, metadata.ListFeatureOpt{
-		EntityIDs:  &[]int{entityID},
+		EntityID:   &entityID,
 		FeatureIDs: &[]int{},
 	})
 	assert.NoError(t, err)
 	assert.Equal(t, 0, len(features))
 
-	// case 7: list features by EntityIDs
+	// case 7: list features by EntityID
 	features, err = store.ListFeature(ctx, metadata.ListFeatureOpt{
-		EntityIDs: &[]int{entityID, entityID + 1},
+		EntityID: &entityID,
 	})
 	assert.NoError(t, err)
 	assert.Equal(t, 1, len(features))
-
-	// case 8: list features by empty EntityIDs
-	features, err = store.ListFeature(ctx, metadata.ListFeatureOpt{
-		EntityIDs: &[]int{},
-	})
-	assert.NoError(t, err)
-	assert.Equal(t, 0, len(features))
-
-	// case 9: list features by empty GroupIDs
-	features, err = store.ListFeature(ctx, metadata.ListFeatureOpt{
-		GroupIDs: &[]int{},
-	})
-	assert.NoError(t, err)
-	assert.Equal(t, 0, len(features))
 }
 
 func TestUpdateFeature(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyStoreFn) {
diff --git a/internal/database/metadata/test_impl/helper.go b/internal/database/metadata/test_impl/helper.go
index 4467438a..feeece08 100644
--- a/internal/database/metadata/test_impl/helper.go
+++ b/internal/database/metadata/test_impl/helper.go
@@ -4,6 +4,10 @@ func boolPtr(i bool) *bool {
 	return &i
 }
 
+func intPtr(i int) *int {
+	return &i
+}
+
 func stringPtr(s string) *string {
 	return &s
 }
diff --git a/internal/database/offline/test_impl/export.go b/internal/database/offline/test_impl/export.go
index 18174bd2..8c8437a8 100644
--- a/internal/database/offline/test_impl/export.go
+++ b/internal/database/offline/test_impl/export.go
@@ -50,9 +50,10 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 	})
 
 	testCases := []struct {
-		description string
-		opt         offline.ExportOpt
-		expected    [][]interface{}
+		description   string
+		opt           offline.ExportOpt
+		expected      [][]interface{}
+		expectedError error
 	}{
 		{
 			description: "one group, batch features",
@@ -93,11 +94,15 @@ func TestExport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 			result, err := store.Export(ctx, tc.opt)
 			values := make([][]interface{}, 0)
 			for row := range result.Data {
-				assert.NoError(t, row.Error)
-				values = append(values, row.Record)
+				values = append(values, row)
+			}
+			if tc.expectedError != nil {
+				assert.EqualError(t, err, tc.expectedError.Error())
+			} else {
+				assert.ElementsMatch(t, tc.expected, values)
+				assert.NoError(t, err)
+				assert.NoError(t, result.CheckStreamError())
 			}
-			assert.ElementsMatch(t, tc.expected, values)
-			assert.NoError(t, err)
 		})
 	}
 }
diff --git a/internal/database/offline/test_impl/import.go b/internal/database/offline/test_impl/import.go
index a01d729e..fa33b3d8 100644
--- a/internal/database/offline/test_impl/import.go
+++ b/internal/database/offline/test_impl/import.go
@@ -68,10 +68,10 @@ func TestImport(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroyS
 		})
 		records := make([][]interface{}, 0)
 		for row := range result.Data {
-			assert.NoError(t, row.Error)
-			records = append(records, row.Record)
+			records = append(records, row)
 		}
 		assert.NoError(t, err)
+		assert.NoError(t, result.CheckStreamError())
 		sort.Slice(records, func(i, j int) bool {
 			return cast.ToString(records[i][0]) < cast.ToString(records[j][0])
 		})
diff --git a/internal/database/offline/test_impl/snapshot.go b/internal/database/offline/test_impl/snapshot.go
index 5b39645d..5b0cddd7 100644
--- a/internal/database/offline/test_impl/snapshot.go
+++ b/internal/database/offline/test_impl/snapshot.go
@@ -61,8 +61,7 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 	})
 	values := make([][]interface{}, 0)
 	for row := range result.Data {
-		assert.NoError(t, err)
-		values = append(values, row.Record)
+		values = append(values, row)
 	}
 	sort.Slice(values, func(i, j int) bool {
 		return cast.ToInt64(values[i][0]) < cast.ToInt64(values[j][0])
@@ -74,5 +73,6 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 		{"1237", "pixel", int64(200), int64(11)},
 	}
 	assert.Equal(t, expected, values)
+	assert.NoError(t, result.CheckStreamError())
 	assert.NoError(t, err)
 }
diff --git a/internal/database/online/test_impl/util.go b/internal/database/online/test_impl/util.go
index 3db208f7..f6d3ab4c 100644
--- a/internal/database/online/test_impl/util.go
+++ b/internal/database/online/test_impl/util.go
@@ -83,22 +83,16 @@ func init() {
 			Group:   &group1,
 		},
 		Data: []types.ExportRecord{
-			{
-				Record: []interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
-			},
-			{
-				Record: []interface{}{"3216", int64(29), nil, 2.0, false, time.Now()},
-			},
-			{
-				Record: []interface{}{"3217", int64(44), "M", 3.1, true, time.Now()},
-			},
+			[]interface{}{"3215", int64(18), "F", 1.1, true, time.Now()},
+			[]interface{}{"3216", int64(29), nil, 2.0, false, time.Now()},
+			[]interface{}{"3217", int64(44), "M", 3.1, true, time.Now()},
 		},
 	}
 
 	var data []types.ExportRecord
 	for i := 0; i < 100; i++ {
 		record := []interface{}{dbutil.RandString(10), rand.Float64()}
-		data = append(data, types.ExportRecord{Record: record, Error: nil})
+		data = append(data, record)
 	}
 	SampleMedium = Sample{
 		Entity: entity,
@@ -140,18 +134,10 @@ func init() {
 			},
 		},
 		Data: []types.ExportRecord{
-			{
-				Record: []interface{}{"3215", int64(1), "1,2,3,4"},
-			},
-			{
-				Record: []interface{}{"3216", int64(2), "2,3,4,5"},
-			},
-			{
-				Record: []interface{}{"3217", int64(3), "3,4,5,6"},
-			},
-			{
-				Record: []interface{}{"3218", int64(4), "4,5,6,7"},
-			},
+			[]interface{}{"3215", int64(1), "1,2,3,4"},
+			[]interface{}{"3216", int64(2), "2,3,4,5"},
+			[]interface{}{"3217", int64(3), "3,4,5,6"},
+			[]interface{}{"3218", int64(4), "4,5,6,7"},
 		},
 	}
 }
diff --git a/oomcli/test/test_online_get.sh b/oomcli/test/test_online_get.sh
index 19e6330c..d3bf0cd0 100755
--- a/oomcli/test/test_online_get.sh
+++ b/oomcli/test/test_online_get.sh
@@ -15,28 +15,6 @@ device,phone.model
 actual=$(oomcli get online --feature phone.model -k 1 -o csv)
 assert_eq "$case" "$expected" "$actual"
 
-case="sync stream feature with no import"
-actual=$(oomcli sync --group-name user-click 2>&1 || true)
-expected=$(cat <<-EOF
-syncing features ...
-Error: failed sync features: group user-click doesn't have any revision
-EOF
-)
-assert_eq "$case" "$expected" "$actual"
-
-import_user_click
-case="sync stream feature"
-oomcli sync --group-name user-click
-
-case="query stream feature"
-actual=$(oomcli get online -k 1,2 --feature user-click.last_5_click_posts,user-click.number_of_user_starred_posts -o csv)
-expected=$(cat <<-EOF
-user,user-click.last_5_click_posts,user-click.number_of_user_starred_posts
-1,"1,2",10
-2,"2,3",10
-EOF
-)
-assert_eq "$case" "$expected" "$actual"
 
 case="query multiple features 1"
 expected='
diff --git a/oomcli/test/util.sh b/oomcli/test/util.sh
index 3a41383c..2e9ba982 100644
--- a/oomcli/test/util.sh
+++ b/oomcli/test/util.sh
@@ -125,14 +125,6 @@ import_student_sample() {
     --description 'test data' | grep -o '[0-9]\+'
 }
 
-import_user_click() {
-    info "import user click sample data to offline store..."
-    oomcli import \
-    --group user-click \
-    --input-file "$(pwd)/user_click.csv" \
-    --description 'test data' > /dev/null
-}
-
 # sync feature values from offline store to online store
 sync() {
     info "sync sample data to online store"
diff --git a/pkg/oomstore/types/apply/validate_test.go b/pkg/oomstore/types/apply/validate_test.go
index 8abdbf78..d99867c2 100644
--- a/pkg/oomstore/types/apply/validate_test.go
+++ b/pkg/oomstore/types/apply/validate_test.go
@@ -18,7 +18,7 @@ func TestEntityValidate(t *testing.T) {
 		wantError   error
 	}{
 		{
-			description: "invalid kind: spelling error",
+			description: "invaid kind: spelling error",
 			entity: &apply.Entity{
 				Kind:        "Entit",
 				Name:        "user",