ingest-storage: Do not deduplicate metadata by family name when unmarshalling (#12804)

#### What this PR does

This PR builds on the refactor in
#12792.
By default, RW2 deserialization deduplicates metric metadata by family
name. All metadata beyond the first instance for a given family are
discarded. This PR adds a new deserialization option, called
`SkipDeduplicateMetadata`, plus a new `metadataSet` implementation that
avoids the deduplication and preserves all given metadata and ordering.

This is done so that we can preserve metadata to the letter, even if it
conflicts, ensuring we [adhere to prometheus
behavior](https://prometheus.io/docs/prometheus/latest/querying/api/#querying-metadata)
particularly in RW1.0.

RW1 requests don't have the duplication that natural RW2 requests tend
to have. Natural RW2 requests will already be deduplicated in the
distributor before they reach the ingest-storage layer, so this change
does not add undue duplication for clients sending RW2.0 into Mimir.

#### Which issue(s) this PR fixes or relates to

contrib grafana/mimir-squad#2253

#### Checklist

- [x] Tests updated.
- [ ] Documentation added.
- [ ] `CHANGELOG.md` updated - the order of entries should be
`[CHANGE]`, `[FEATURE]`, `[ENHANCEMENT]`, `[BUGFIX]`. If changelog entry
is not needed, please add the `changelog-not-needed` label to the PR.
- [ ]
[`about-versioning.md`](https://github.com/grafana/mimir/blob/main/docs/sources/mimir/configure/about-versioning.md)
updated with experimental features.

---------

Co-authored-by: Marco Pracucci <[email protected]>

Author: alexweav

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.