ingest-storage: Do not deduplicate metadata by family name when unmarshalling

pkg/mimirpb/compat_rw2_test.go

@@ -797,6 +797,50 @@ func TestRW2Unmarshal(t *testing.T) {
 		require.Equal(t, received.Metadata[0].Help, "It's a cool series, but old description.")
 		require.Equal(t, received.Metadata[0].Unit, "megawatts")
 	})
+
+	t.Run("conflicting metadata, skipDeduplicateMetadata is true, both metadata and their order is preserved", func(t *testing.T) {
+		writeRequest := &WriteRequest{
+			SymbolsRW2: []string{"", "__name__", "my_cool_series", "It's a cool series, but old description.", "It's a cool series, but new description.", "megawatts"},
+			TimeseriesRW2: []TimeSeriesRW2{
+				{
+					LabelsRefs: []uint32{1, 2},
+					Metadata: MetadataRW2{
+						Type:    METRIC_TYPE_COUNTER,
+						HelpRef: 3,
+						UnitRef: 5,
+					},
+				},
+				{
+					LabelsRefs: []uint32{1, 2},
+					Metadata: MetadataRW2{
+						Type:    METRIC_TYPE_COUNTER,
+						HelpRef: 4,
+						UnitRef: 5,
+					},
+				},
+			},
+		}
+		data, err := writeRequest.Marshal()
+		require.NoError(t, err)
+
+		// Unmarshal the data back into Mimir's WriteRequest.
+		received := PreallocWriteRequest{
+			SkipDeduplicateMetadata: true,
+		}
+		received.UnmarshalFromRW2 = true
+		err = received.Unmarshal(data)
+		require.NoError(t, err)
+
+		require.Len(t, received.Metadata, 2)
+		require.Equal(t, received.Metadata[0].MetricFamilyName, "my_cool_series")
+		require.Equal(t, received.Metadata[0].Type, COUNTER)
+		require.Equal(t, received.Metadata[0].Help, "It's a cool series, but old description.")
+		require.Equal(t, received.Metadata[0].Unit, "megawatts")
+		require.Equal(t, received.Metadata[1].MetricFamilyName, "my_cool_series")
+		require.Equal(t, received.Metadata[1].Type, COUNTER)
+		require.Equal(t, received.Metadata[1].Help, "It's a cool series, but new description.")
+		require.Equal(t, received.Metadata[1].Unit, "megawatts")
+	})
 }
 
 func makeTestRW2WriteRequest(syms *test.SymbolTableBuilder) *WriteRequest {

pkg/mimirpb/custom.go

@@ -452,6 +452,13 @@ type MarshalerWithSize interface {
 	MarshalWithSize(size int) ([]byte, error)
 }
 
+func metadataSetFromSettings(skipDeduplicateMetadata bool) metadataSet {
+	if skipDeduplicateMetadata {
+		return newPassthroughMetadataSet()
+	}
+	return newDedupingMetadataSet()
+}
+
 // metadataSet is the collection of metadata within a request.
 // It keeps the order at which metadata is added. Metadata may optionally be deduplicated by family name.
 type metadataSet interface {
@@ -461,6 +468,7 @@ type metadataSet interface {
 }
 
 var _ metadataSet = dedupingMetadataSet{}
+var _ metadataSet = &passthroughMetadataSet{}
 
 // dedupingMetadataSet is a metadataSet that only stores one metadata per metric family.
 // Only the first metadata seen for a given family is kept.
@@ -496,6 +504,28 @@ func (m dedupingMetadataSet) slice() []*MetricMetadata {
 	return result
 }
 
+type passthroughMetadataSet struct {
+	metadata []*MetricMetadata
+}
+
+func newPassthroughMetadataSet() *passthroughMetadataSet {
+	return &passthroughMetadataSet{
+		metadata: make([]*MetricMetadata, 0),
+	}
+}
+
+func (m *passthroughMetadataSet) add(family string, mm MetricMetadata) {
+	m.metadata = append(m.metadata, &mm)
+}
+
+func (m *passthroughMetadataSet) len() int {
+	return len(m.metadata)
+}
+
+func (m *passthroughMetadataSet) slice() []*MetricMetadata {
+	return m.metadata
+}
+
 // orderAwareMetricMetadata is a tuple (index, metadata) that knows its own position in a metadata slice.
 // It's tied to custom logic that unmarshals RW2 metadata into a map, and allows us to
 // remember the order that metadata arrived in when unmarshalling.