Repository: kythe/kythe
Pull request #4825 (2021-02-12 16:19:07)

Merge FileDecorations in Beam pipeline.
This will solve a problem if pipeline gets run using multiple instances.


Patch:
diff --git a/kythe/go/serving/pipeline/BUILD b/kythe/go/serving/pipeline/BUILD
index 11812b0d5..7f23d22bc 100644
--- a/kythe/go/serving/pipeline/BUILD
+++ b/kythe/go/serving/pipeline/BUILD
@@ -58,6 +58,8 @@ go_test(
         "@com_github_apache_beam//sdks/go/pkg/beam/testing/passert:go_default_library",
         "@com_github_apache_beam//sdks/go/pkg/beam/testing/ptest:go_default_library",
         "@com_github_apache_beam//sdks/go/pkg/beam/x/debug:go_default_library",
+        "@com_github_google_go_cmp//cmp:go_default_library",
+        "@com_github_google_go_cmp//cmp/cmpopts:go_default_library",
     ],
 )
 
diff --git a/kythe/go/serving/pipeline/beam.go b/kythe/go/serving/pipeline/beam.go
index ffc980b26..072e7a7bc 100644
--- a/kythe/go/serving/pipeline/beam.go
+++ b/kythe/go/serving/pipeline/beam.go
@@ -539,6 +539,13 @@ func (c *combineDecorPieces) CreateAccumulator() *srvpb.FileDecorations {
 }
 
 func (c *combineDecorPieces) MergeAccumulators(accum, n *srvpb.FileDecorations) *srvpb.FileDecorations {
+	accum.Decoration = append(accum.Decoration, n.Decoration...)
+	if accum.File == nil {
+		accum.File = n.File
+	}
+	accum.Target = append(accum.Target, n.Target...)
+	accum.TargetDefinitions = append(accum.TargetDefinitions, n.TargetDefinitions...)
+	accum.Diagnostic = append(accum.Diagnostic, n.Diagnostic...)
 	return accum
 }
 


Test patch:
diff --git a/kythe/go/serving/pipeline/beam_test.go b/kythe/go/serving/pipeline/beam_test.go
index de03e36b9..cc63a63bb 100644
--- a/kythe/go/serving/pipeline/beam_test.go
+++ b/kythe/go/serving/pipeline/beam_test.go
@@ -25,6 +25,8 @@ import (
 	"github.com/apache/beam/sdks/go/pkg/beam/testing/passert"
 	"github.com/apache/beam/sdks/go/pkg/beam/testing/ptest"
 	"github.com/apache/beam/sdks/go/pkg/beam/x/debug"
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
 	"google.golang.org/protobuf/proto"
 
 	cpb "kythe.io/kythe/proto/common_go_proto"
@@ -958,3 +960,69 @@ func TestDocuments_registrations(t *testing.T) {
 	FromNodes(s, nodes).Documents()
 	beamtest.CheckRegistrations(t, p)
 }
+
+func TestCombineDecorPieces_mergeAccumulators(t *testing.T) {
+	accum := &srvpb.FileDecorations{
+		Decoration: []*srvpb.FileDecorations_Decoration{{
+			Target: "target1",
+		}},
+		Target: []*srvpb.Node{{
+			Ticket: "ticket1",
+		}},
+		TargetDefinitions: []*srvpb.ExpandedAnchor{{
+			Ticket: "ticket1",
+		}},
+		Diagnostic: []*cpb.Diagnostic{{
+			Message: "msg1",
+		}},
+	}
+
+	other := &srvpb.FileDecorations{
+		Decoration: []*srvpb.FileDecorations_Decoration{{
+			Target: "target2",
+		}},
+		Target: []*srvpb.Node{{
+			Ticket: "ticket2",
+		}},
+		TargetDefinitions: []*srvpb.ExpandedAnchor{{
+			Ticket: "ticket2",
+		}},
+		Diagnostic: []*cpb.Diagnostic{{
+			Message: "msg2",
+		}},
+		File: &srvpb.File{
+			Text: []byte("some text\n"),
+		},
+	}
+
+	expected := &srvpb.FileDecorations{
+		Decoration: []*srvpb.FileDecorations_Decoration{{
+			Target: "target1",
+		}, {
+			Target: "target2",
+		}},
+		Target: []*srvpb.Node{{
+			Ticket: "ticket1",
+		}, {
+			Ticket: "ticket2",
+		}},
+		TargetDefinitions: []*srvpb.ExpandedAnchor{{
+			Ticket: "ticket1",
+		}, {
+			Ticket: "ticket2",
+		}},
+		Diagnostic: []*cpb.Diagnostic{{
+			Message: "msg1",
+		}, {
+			Message: "msg2",
+		}},
+		File: &srvpb.File{
+			Text: []byte("some text\n"),
+		},
+	}
+
+	actual := (&combineDecorPieces{}).MergeAccumulators(accum, other)
+	if d := cmp.Diff(actual, expected, cmpopts.IgnoreUnexported(cpb.Diagnostic{}, srvpb.ExpandedAnchor{}, srvpb.Node{}, srvpb.FileDecorations_Decoration{}, srvpb.File{}, srvpb.FileDecorations{})); d != "" {
+		t.Errorf("Expected %v but was %v. Diff: %v", expected, actual, d)
+	}
+}