Repository: oom-ai/oomstore
Pull request #1155 (2022-02-09 07:00:37)

test(oomcli): get online add sync stream feature case

YikSanChan: let's also fix the commit msg: onlien -> online

Patch:
diff --git a/internal/database/offline/bigquery/export.go b/internal/database/offline/bigquery/export.go
index 6e8e9dd0..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,15 +30,17 @@ 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 {
-			stream <- types.ExportRecord{Error: err}
+			errs <- err
 			return
 		}
 		for {
@@ -48,7 +50,7 @@ func bigqueryQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt off
 				break
 			}
 			if err != nil {
-				stream <- types.ExportRecord{Error: errdefs.Errorf("failed at rows.Next, err=%v", err)}
+				errs <- errdefs.Errorf("failed at rows.Next, err=%v", err)
 				return
 			}
 			record := make([]interface{}, 0, len(recordMap))
@@ -56,9 +58,9 @@ func bigqueryQueryExportResults(ctx context.Context, dbOpt dbutil.DBOpt, opt off
 			for _, feature := range features {
 				record = append(record, recordMap[feature.DBFullName(Backend)])
 			}
-			stream <- types.ExportRecord{Record: record}
+			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/export.go b/internal/database/offline/sqlutil/export.go
index 9d28bac0..545f3060 100644
--- a/internal/database/offline/sqlutil/export.go
+++ b/internal/database/offline/sqlutil/export.go
@@ -124,27 +124,27 @@ 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 {
-			stream <- types.ExportRecord{Error: errdefs.WithStack(err)}
+			errs <- errdefs.WithStack(err)
 			return
 		}
 		defer stmt.Close()
-
 		rows, err := stmt.Queryx(args...)
 		if err != nil {
-			stream <- types.ExportRecord{Error: errdefs.WithStack(err)}
+			errs <- errdefs.WithStack(err)
 			return
 		}
 		defer rows.Close()
-
 		for rows.Next() {
 			record, err := rows.SliceScan()
 			if err != nil {
-				stream <- types.ExportRecord{Error: errdefs.Errorf("failed at rows.SliceScan, err=%v", err)}
+				errs <- errdefs.Errorf("failed at rows.SliceScan, err=%v", err)
 				return
 			}
 			record[0] = cast.ToString(record[0])
@@ -154,14 +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 {
-					stream <- types.ExportRecord{Error: err}
+					errs <- err
 					return
 				}
 				record[i+1] = deserializedValue
 			}
-			stream <- types.ExportRecord{Record: record, Error: nil}
+			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/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/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/server.go b/oomagent/server.go
index 4dd62c49..0c5a0dba 100644
--- a/oomagent/server.go
+++ b/oomagent/server.go
@@ -311,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())
 		}
@@ -328,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/export_helper.go b/oomcli/cmd/export_helper.go
index dce1215c..31552c2c 100644
--- a/oomcli/cmd/export_helper.go
+++ b/oomcli/cmd/export_helper.go
@@ -45,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 {
@@ -62,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/pkg/oomstore/export.go b/pkg/oomstore/export.go
index 5ffe76bd..d14d0ca0 100644
--- a/pkg/oomstore/export.go
+++ b/pkg/oomstore/export.go
@@ -120,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/sync.go b/pkg/oomstore/sync.go
index 6853ae6b..50925088 100644
--- a/pkg/oomstore/sync.go
+++ b/pkg/oomstore/sync.go
@@ -56,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
@@ -117,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/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 {


Test patch:
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 d3bf0cd0..19e6330c 100755
--- a/oomcli/test/test_online_get.sh
+++ b/oomcli/test/test_online_get.sh
@@ -15,6 +15,28 @@ 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 2e9ba982..3a41383c 100644
--- a/oomcli/test/util.sh
+++ b/oomcli/test/util.sh
@@ -125,6 +125,14 @@ 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"