Repository: oom-ai/oomstore
Pull request #889 (2022-01-04 03:39:11)

Jinghan/API return empty result instead of nil
This PR refactors API `Join` and `OnlineGet`: if there is no valid return result, we should return empty result instead of nil.


Patch:
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index 5ebaa666..9500eba0 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -44,13 +44,18 @@ func Join(ctx context.Context, db *sqlx.DB, opt offline.JoinOpt, backendType typ
 }
 
 func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.JoinResult, error) {
+	data := make(chan []interface{})
+	defer close(data)
+	emptyResult := &types.JoinResult{
+		Data: data,
+	}
 	// Step 1: prepare temporary table entity_rows
 	features := types.FeatureList{}
 	for _, featureList := range opt.FeatureMap {
 		features = append(features, featureList...)
 	}
 	if len(features) == 0 {
-		return nil, nil
+		return emptyResult, nil
 	}
 	entityRowsTableName, err := PrepareEntityRowsTable(ctx, dbOpt, opt.Entity, opt.EntityRows, opt.ValueNames)
 	if err != nil {
@@ -169,10 +174,15 @@ func JoinOneGroup(ctx context.Context, dbOpt dbutil.DBOpt, opt offline.JoinOneGr
 }
 
 func ReadJoinedTable(ctx context.Context, dbOpt dbutil.DBOpt, opt ReadJoinedTableOpt) (*types.JoinResult, error) {
+	data := make(chan []interface{})
+	defer close(data)
+	emptyResult := &types.JoinResult{
+		Data: data,
+	}
 	tableNames := opt.TableNames
 	dropTableNames := append([]string{opt.EntityRowsTableName}, opt.AllTableNames...)
 	if len(tableNames) == 0 {
-		return nil, nil
+		return emptyResult, nil
 	}
 	qt := dbutil.QuoteFn(dbOpt.Backend)
 
diff --git a/internal/database/online/cassandra/get.go b/internal/database/online/cassandra/get.go
index 58da07ae..6bcedd21 100644
--- a/internal/database/online/cassandra/get.go
+++ b/internal/database/online/cassandra/get.go
@@ -25,7 +25,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 	rs := make(map[string]interface{})
 	if err := db.Query(query, opt.EntityKey).WithContext(ctx).MapScan(rs); err != nil {
 		if err == gocql.ErrNotFound || isTableNotFoundError(err, tableName) {
-			return nil, nil
+			return rs, nil
 		}
 		return nil, err
 	}
@@ -57,7 +57,7 @@ func (db *DB) MultiGet(ctx context.Context, opt online.MultiGetOpt) (map[string]
 		Iter().SliceMap()
 	if err != nil {
 		if err == gocql.ErrNotFound || isTableNotFoundError(err, tableName) {
-			return nil, nil
+			return rs, nil
 		}
 		return nil, err
 	}
diff --git a/internal/database/online/dynamodb/get.go b/internal/database/online/dynamodb/get.go
index c2e06059..3ed91fc4 100644
--- a/internal/database/online/dynamodb/get.go
+++ b/internal/database/online/dynamodb/get.go
@@ -33,7 +33,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 	})
 	if err != nil {
 		if apiErr := new(types.ResourceNotFoundException); errors.As(err, &apiErr) {
-			return nil, nil
+			return make(dbutil.RowMap), nil
 		}
 		return nil, err
 	}
@@ -100,7 +100,7 @@ func batchGetItem(ctx context.Context, db *DB, keys []map[string]types.Attribute
 
 func deserializeFeatureValues(features oomTypes.FeatureList, item map[string]types.AttributeValue) (dbutil.RowMap, error) {
 	if item == nil {
-		return nil, nil
+		return make(dbutil.RowMap), nil
 	}
 	rowMap := make(dbutil.RowMap)
 	var value interface{}
diff --git a/internal/database/online/sqlutil/get.go b/internal/database/online/sqlutil/get.go
index 009bdff9..b8344552 100644
--- a/internal/database/online/sqlutil/get.go
+++ b/internal/database/online/sqlutil/get.go
@@ -22,7 +22,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 {
 		if err == sql.ErrNoRows || dbutil.IsTableNotFoundError(err, backend) {
-			return nil, nil
+			return make(dbutil.RowMap), nil
 		}
 		return nil, err
 	}
diff --git a/internal/database/online/tikv/get.go b/internal/database/online/tikv/get.go
index 45343a3c..131767ba 100644
--- a/internal/database/online/tikv/get.go
+++ b/internal/database/online/tikv/get.go
@@ -26,7 +26,7 @@ func (db *DB) Get(ctx context.Context, opt online.GetOpt) (dbutil.RowMap, error)
 		return nil, err
 	}
 	if rowMap, ok := res[opt.EntityKey]; !ok {
-		return nil, nil
+		return make(dbutil.RowMap), nil
 	} else {
 		return rowMap, nil
 	}
diff --git a/pkg/oomstore/join.go b/pkg/oomstore/join.go
index b804d677..718ad838 100644
--- a/pkg/oomstore/join.go
+++ b/pkg/oomstore/join.go
@@ -19,6 +19,11 @@ import (
 // ChannelJoin gets point-in-time correct feature values for each entity row.
 // Currently, this API only supports batch features.
 func (s *OomStore) ChannelJoin(ctx context.Context, opt types.ChannelJoinOpt) (*types.JoinResult, error) {
+	data := make(chan []interface{})
+	defer close(data)
+	emptyResult := &types.JoinResult{
+		Data: data,
+	}
 	features, err := s.metadata.ListFeature(ctx, metadata.ListFeatureOpt{
 		FeatureFullNames: &opt.FeatureFullNames,
 	})
@@ -30,7 +35,7 @@ func (s *OomStore) ChannelJoin(ctx context.Context, opt types.ChannelJoinOpt) (*
 		return f.Group.Category == types.CategoryBatch
 	})
 	if len(features) == 0 {
-		return nil, nil
+		return emptyResult, nil
 	}
 
 	entity, err := getSharedEntity(features)
diff --git a/pkg/oomstore/online_query.go b/pkg/oomstore/online_query.go
index 354efd2a..62081524 100644
--- a/pkg/oomstore/online_query.go
+++ b/pkg/oomstore/online_query.go
@@ -10,28 +10,27 @@ import (
 	"github.com/oom-ai/oomstore/pkg/oomstore/types"
 )
 
-// Get online features of a particular entity instance.
+// OnlineGet gets online features of a particular entity instance.
 func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*types.FeatureValues, error) {
+	rs := types.FeatureValues{
+		EntityKey:        opt.EntityKey,
+		FeatureFullNames: opt.FeatureFullNames,
+		FeatureValueMap:  make(map[string]interface{}),
+	}
 	features := s.metadata.CacheListFeature(ctx, metadata.ListFeatureOpt{
 		FeatureFullNames: &opt.FeatureFullNames,
 	}).Filter(func(f *types.Feature) bool {
 		return f.Group.OnlineRevisionID != nil
 	})
 	if len(features) == 0 {
-		return nil, nil
+		return &rs, nil
 	}
 
 	entity, err := getSharedEntity(features)
 	if err != nil {
 		return nil, err
 	}
-
-	rs := types.FeatureValues{
-		EntityName:       entity.Name,
-		EntityKey:        opt.EntityKey,
-		FeatureFullNames: opt.FeatureFullNames,
-		FeatureValueMap:  make(map[string]interface{}),
-	}
+	rs.EntityName = entity.Name
 
 	featureMap := groupFeaturesByRevisionID(features)
 
@@ -55,8 +54,9 @@ func (s *OomStore) OnlineGet(ctx context.Context, opt types.OnlineGetOpt) (*type
 	return &rs, nil
 }
 
-// Get online features of multiple entity instances.
+// OnlineMultiGet gets online features of multiple entity instances.
 func (s *OomStore) OnlineMultiGet(ctx context.Context, opt types.OnlineMultiGetOpt) (map[string]*types.FeatureValues, error) {
+	result := make(map[string]*types.FeatureValues)
 	features := s.metadata.CacheListFeature(ctx, metadata.ListFeatureOpt{
 		FeatureFullNames: &opt.FeatureFullNames,
 	})
@@ -65,7 +65,7 @@ func (s *OomStore) OnlineMultiGet(ctx context.Context, opt types.OnlineMultiGetO
 		return f.OnlineRevisionID() != nil
 	})
 	if len(features) == 0 {
-		return nil, nil
+		return result, nil
 	}
 
 	entity, err := getSharedEntity(features)
@@ -83,7 +83,6 @@ func (s *OomStore) OnlineMultiGet(ctx context.Context, opt types.OnlineMultiGetO
 		return nil, err
 	}
 
-	result := make(map[string]*types.FeatureValues)
 	for _, entityKey := range opt.EntityKeys {
 		result[entityKey] = &types.FeatureValues{
 			EntityName:       entity.Name,


Test patch:
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index b1203465..9282cfd2 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -76,7 +76,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 			opt: offline.JoinOpt{
 				FeatureMap: make(map[string]types.FeatureList),
 			},
-			expected: nil,
+			expected: prepareEmptyResult(),
 		},
 		{
 			description: "no entity rows",
@@ -85,7 +85,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 				FeatureMap: oneGroupFeatureMap,
 				EntityRows: prepareEntityRows(true, false),
 			},
-			expected: nil,
+			expected: prepareEmptyResult(),
 		},
 		{
 			description: "one batch feature group",
@@ -135,15 +135,12 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		t.Run(tc.description, func(t *testing.T) {
 			actual, err := store.Join(context.Background(), tc.opt)
 			require.NoError(t, err)
-			if tc.expected == nil {
-				assert.Equal(t, tc.expected, actual)
-			} else {
-				assert.ElementsMatch(t, tc.expected.Header, actual.Header)
-				expectedValues := extractValues(tc.expected.Data)
-				actualValues := extractValues(actual.Data)
-				for i := range expectedValues {
-					assert.ElementsMatch(t, expectedValues[i], actualValues[i])
-				}
+
+			assert.ElementsMatch(t, tc.expected.Header, actual.Header)
+			expectedValues := extractValues(tc.expected.Data)
+			actualValues := extractValues(actual.Data)
+			for i := range expectedValues {
+				assert.ElementsMatch(t, expectedValues[i], actualValues[i])
 			}
 		})
 	}
@@ -296,6 +293,13 @@ func prepareResult(oneGroup bool, withValue bool, values []map[string]interface{
 	}
 }
 
+func prepareEmptyResult() *types.JoinResult {
+	data := make(chan []interface{})
+	defer close(data)
+	return &types.JoinResult{
+		Data: data,
+	}
+}
 func prepareStreamingResultValues() []map[string]interface{} {
 	return []map[string]interface{}{
 		{
diff --git a/pkg/oomstore/join_test.go b/pkg/oomstore/join_test.go
index bfb88a94..51b6b2bb 100644
--- a/pkg/oomstore/join_test.go
+++ b/pkg/oomstore/join_test.go
@@ -40,7 +40,6 @@ func TestChannelJoin(t *testing.T) {
 		},
 	}
 
-	var emptyResult *types.JoinResult
 	validResult := prepareResult()
 	entityRows := make(chan types.EntityRow)
 
@@ -62,7 +61,7 @@ func TestChannelJoin(t *testing.T) {
 			},
 			features:      streamFeatures,
 			expectedError: nil,
-			expected:      emptyResult,
+			expected:      prepareEmptyResult(),
 		},
 		{
 			description: "inconsistent features, return nil",
@@ -86,9 +85,9 @@ func TestChannelJoin(t *testing.T) {
 				"device_basic":    {consistentFeatures[0]},
 				"device_advanced": {consistentFeatures[1]},
 			},
-			joined:        nil,
+			joined:        prepareEmptyResult(),
 			expectedError: nil,
-			expected:      emptyResult,
+			expected:      prepareEmptyResult(),
 		},
 		{
 			description: "consistent entity type, succeed",
@@ -122,8 +121,6 @@ func TestChannelJoin(t *testing.T) {
 			actual, err := store.ChannelJoin(context.Background(), tc.opt)
 			if tc.expectedError != nil {
 				assert.Error(t, err, tc.expectedError)
-			} else if tc.expected == nil {
-				assert.Equal(t, tc.expected, actual)
 			} else {
 				assert.NoError(t, err)
 				assert.Equal(t, tc.expected.Header, actual.Header)
@@ -151,8 +148,19 @@ func prepareResult() *types.JoinResult {
 
 func extractValues(stream <-chan []interface{}) [][]interface{} {
 	values := make([][]interface{}, 0)
+	if stream == nil {
+		return values
+	}
 	for item := range stream {
 		values = append(values, item)
 	}
 	return values
 }
+
+func prepareEmptyResult() *types.JoinResult {
+	data := make(chan []interface{})
+	defer close(data)
+	return &types.JoinResult{
+		Data: data,
+	}
+}
diff --git a/pkg/oomstore/online_query_test.go b/pkg/oomstore/online_query_test.go
index 21a12332..180c9b8f 100644
--- a/pkg/oomstore/online_query_test.go
+++ b/pkg/oomstore/online_query_test.go
@@ -43,7 +43,11 @@ func TestOnlineGet(t *testing.T) {
 			},
 			features:      unavailableFeatures,
 			expectedError: nil,
-			expected:      nil,
+			expected: &types.FeatureValues{
+				EntityKey:        "1234",
+				FeatureFullNames: consistentFeatures.FullNames(),
+				FeatureValueMap:  map[string]interface{}{},
+			},
 		},
 		{
 			description: "inconsistent entity type, fail",
@@ -127,7 +131,7 @@ func TestOnlineMultiGet(t *testing.T) {
 			},
 			features:      unavailableFeatures,
 			expectedError: nil,
-			expected:      nil,
+			expected:      map[string]*types.FeatureValues{},
 		},
 		{
 			description: "inconsistent entity type, fail",