Repository: oom-ai/oomstore
Pull request #1047 (2022-01-20 09:26:13)

Fix/join
resolve: #1046 


Patch:
diff --git a/internal/database/offline/sqlutil/join.go b/internal/database/offline/sqlutil/join.go
index 8d3fc981..9abf8ca9 100644
--- a/internal/database/offline/sqlutil/join.go
+++ b/internal/database/offline/sqlutil/join.go
@@ -45,19 +45,23 @@ 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,
+	{
+		empty := true
+		for _, features := range opt.FeatureMap {
+			if len(features) > 0 {
+				empty = false
+				break
+			}
+		}
+		if empty {
+			data := make(chan []interface{})
+			defer close(data)
+			return &types.JoinResult{Data: data}, nil
+		}
+
 	}
+
 	// Step 1: prepare temporary table entity_rows
-	features := types.FeatureList{}
-	for _, featureList := range opt.FeatureMap {
-		features = append(features, featureList...)
-	}
-	if len(features) == 0 {
-		return emptyResult, nil
-	}
 	entityRowsTableName, err := PrepareEntityRowsTable(ctx, dbOpt, opt.Entity, opt.EntityRows, opt.ValueNames)
 	if err != nil {
 		return nil, err
@@ -68,7 +72,8 @@ func DoJoin(ctx context.Context, dbOpt dbutil.DBOpt, opt DoJoinOpt) (*types.Join
 	allTableNames := make([]string, 0)
 	tableToFeatureMap := make(map[string]types.FeatureList)
 	var category types.Category
-	for groupName, featureList := range opt.FeatureMap {
+	for _, groupName := range opt.GroupNames {
+		featureList := opt.FeatureMap[groupName]
 		revisionRanges, ok := opt.RevisionRangeMap[groupName]
 		if !ok {
 			continue
diff --git a/internal/database/offline/types.go b/internal/database/offline/types.go
index 730026b7..dc213622 100644
--- a/internal/database/offline/types.go
+++ b/internal/database/offline/types.go
@@ -25,6 +25,7 @@ type RevisionRange struct {
 type JoinOpt struct {
 	Entity           types.Entity
 	EntityRows       <-chan types.EntityRow
+	GroupNames       []string
 	FeatureMap       map[string]types.FeatureList
 	RevisionRangeMap map[string][]*RevisionRange
 	ValueNames       []string
diff --git a/pkg/oomstore/join.go b/pkg/oomstore/join.go
index 821b4bed..f888ca3f 100644
--- a/pkg/oomstore/join.go
+++ b/pkg/oomstore/join.go
@@ -48,7 +48,7 @@ func (s *OomStore) ChannelJoin(ctx context.Context, opt types.ChannelJoinOpt) (*
 		return nil, errdefs.Errorf("failed to get shared entity")
 	}
 
-	featureMap := buildGroupToFeaturesMap(features)
+	groupNames, featureMap := buildGroupToFeaturesMap(features)
 	revisionRangeMap := make(map[string][]*offline.RevisionRange)
 	for groupName, featureList := range featureMap {
 		if len(featureList) == 0 {
@@ -64,6 +64,7 @@ func (s *OomStore) ChannelJoin(ctx context.Context, opt types.ChannelJoinOpt) (*
 	return s.offline.Join(ctx, offline.JoinOpt{
 		Entity:           *entity,
 		EntityRows:       opt.EntityRows,
+		GroupNames:       groupNames,
 		FeatureMap:       featureMap,
 		RevisionRangeMap: revisionRangeMap,
 		ValueNames:       opt.ExistedFeatureNames,
@@ -95,15 +96,18 @@ func (s *OomStore) Join(ctx context.Context, opt types.JoinOpt) error {
 }
 
 // key: group_name, value: slice of features
-func buildGroupToFeaturesMap(features types.FeatureList) map[string]types.FeatureList {
-	groups := make(map[string]types.FeatureList)
+func buildGroupToFeaturesMap(features types.FeatureList) ([]string, map[string]types.FeatureList) {
+	groupNames := make([]string, 0, features.Len())
+	featureMap := make(map[string]types.FeatureList)
+
 	for _, f := range features {
-		if _, ok := groups[f.Group.Name]; !ok {
-			groups[f.Group.Name] = types.FeatureList{}
+		if _, ok := featureMap[f.Group.Name]; !ok {
+			groupNames = append(groupNames, f.Group.Name)
+			featureMap[f.Group.Name] = types.FeatureList{}
 		}
-		groups[f.Group.Name] = append(groups[f.Group.Name], f)
+		featureMap[f.Group.Name] = append(featureMap[f.Group.Name], f)
 	}
-	return groups
+	return groupNames, featureMap
 }
 
 func (s *OomStore) buildRevisionRanges(ctx context.Context, group *types.Group) ([]*offline.RevisionRange, error) {


Test patch:
diff --git a/internal/database/offline/test_impl/join.go b/internal/database/offline/test_impl/join.go
index c03c98f7..37a595b1 100644
--- a/internal/database/offline/test_impl/join.go
+++ b/internal/database/offline/test_impl/join.go
@@ -28,8 +28,8 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		Name:      "unix_milli",
 		ValueType: types.Int64,
 	}
-	oneGroupFeatures, oneGroupFeatureMap := prepareFeatures(true)
-	twoGroupFeatures, twoGroupFeatureMap := prepareFeatures(false)
+	oneGroupFeatures, oneGroupFeatureMap, oneGroupGroupNames := prepareFeatures(true)
+	twoGroupFeatures, twoGroupFeatureMap, twoGroupGroupNames := prepareFeatures(false)
 
 	buildTestSnapshotTable(ctx, t, store, oneGroupFeatures, 1, "offline_snapshot_1_1", &offline.CSVSource{
 		Reader: bufio.NewReader(strings.NewReader(`1234,xiaomi,100
@@ -71,6 +71,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 		{
 			description: "no features",
 			opt: offline.JoinOpt{
+
 				FeatureMap: make(map[string]types.FeatureList),
 			},
 			expected: prepareEmptyResult(),
@@ -79,6 +80,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 			description: "no entity rows",
 			opt: offline.JoinOpt{
 				Entity:     *entity,
+				GroupNames: oneGroupGroupNames,
 				FeatureMap: oneGroupFeatureMap,
 				EntityRows: prepareEntityRows(true, false),
 			},
@@ -89,6 +91,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 			opt: offline.JoinOpt{
 				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, false),
+				GroupNames:       oneGroupGroupNames,
 				FeatureMap:       oneGroupFeatureMap,
 				RevisionRangeMap: prepareRevisionRanges(true, false),
 			},
@@ -99,6 +102,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 			opt: offline.JoinOpt{
 				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, false),
+				GroupNames:       twoGroupGroupNames,
 				FeatureMap:       twoGroupFeatureMap,
 				RevisionRangeMap: prepareRevisionRanges(false, false),
 			},
@@ -109,6 +113,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 			opt: offline.JoinOpt{
 				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, true),
+				GroupNames:       twoGroupGroupNames,
 				FeatureMap:       twoGroupFeatureMap,
 				RevisionRangeMap: prepareRevisionRanges(false, false),
 				ValueNames:       []string{"value_1", "value_2"},
@@ -120,6 +125,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 			opt: offline.JoinOpt{
 				Entity:           *entity,
 				EntityRows:       prepareEntityRows(false, true),
+				GroupNames:       twoGroupGroupNames,
 				FeatureMap:       twoGroupFeatureMap,
 				RevisionRangeMap: prepareRevisionRanges(false, true),
 				ValueNames:       []string{"value_1", "value_2"},
@@ -143,7 +149,7 @@ func TestJoin(t *testing.T, prepareStore PrepareStoreFn, destroyStore DestroySto
 	}
 }
 
-func prepareFeatures(oneGroup bool) (types.FeatureList, map[string]types.FeatureList) {
+func prepareFeatures(oneGroup bool) (types.FeatureList, map[string]types.FeatureList, []string) {
 	groupBasic := &types.Group{
 		ID:       1,
 		Name:     "device_basic",
@@ -180,14 +186,14 @@ func prepareFeatures(oneGroup bool) (types.FeatureList, map[string]types.Feature
 		featureMap := map[string]types.FeatureList{
 			groupBasic.Name: {model, price},
 		}
-		return features, featureMap
+		return features, featureMap, []string{groupBasic.Name}
 	} else {
 		features := types.FeatureList{model, price, isActive}
 		featureMap := map[string]types.FeatureList{
 			groupBasic.Name:     {model, price},
 			isActive.Group.Name: {isActive},
 		}
-		return features, featureMap
+		return features, featureMap, []string{groupBasic.Name, groupAdvanced.Name}
 	}
 }
 
@@ -310,6 +316,7 @@ func prepareEmptyResult() *types.JoinResult {
 		Data: data,
 	}
 }
+
 func prepareStreamingResultValues() []map[string]interface{} {
 	return []map[string]interface{}{
 		{
diff --git a/internal/database/offline/test_impl/snapshot.go b/internal/database/offline/test_impl/snapshot.go
index fe15e0b3..4d056371 100644
--- a/internal/database/offline/test_impl/snapshot.go
+++ b/internal/database/offline/test_impl/snapshot.go
@@ -20,7 +20,7 @@ func TestSnapshot(t *testing.T, prepareStore PrepareStoreFn, destroyStore Destro
 	ctx, store := prepareStore(t)
 	defer store.Close()
 
-	features, _ := prepareFeatures(true)
+	features, _, _ := prepareFeatures(true)
 	group := features[0].Group
 	unixMilli := &types.Feature{
 		Name:      "unix_milli",