Add _extracted to non-indexed labels with same name as stream labels

pkg/logql/log/labels.go

@@ -1,6 +1,7 @@
 package log
 
 import (
+	"fmt"
 	"sort"
 
 	"github.com/prometheus/prometheus/model/labels"
@@ -244,9 +245,15 @@ func (b *LabelsBuilder) Set(n, v string) *LabelsBuilder {
 	return b
 }
 
+// Add the labels to the builder. If a label with the same name
+// already exists in the base labels, a suffix is added to the name.
 func (b *LabelsBuilder) Add(labels ...labels.Label) *LabelsBuilder {
 	for _, l := range labels {
-		b.Set(l.Name, l.Value)
+		name := l.Name
+		if b.BaseHas(name) {
+			name = fmt.Sprintf("%s%s", name, duplicateSuffix)
+		}
+		b.Set(name, l.Value)
 	}
 	return b
 }

pkg/logql/log/metrics_extraction_test.go

@@ -382,6 +382,25 @@ func TestNewLineSampleExtractorWithNonIndexedLabels(t *testing.T) {
 	require.Equal(t, 1., f)
 	assertLabelResult(t, expectedLabelsResults, l)
 
+	// test duplicated non-indexed labels with stream labels
+	expectedLabelsResults = append(lbs, labels.Label{
+		Name: "foo_extracted", Value: "baz",
+	})
+	expectedLabelsResults = append(expectedLabelsResults, nonIndexedLabels...)
+	f, l, ok = sse.Process(0, []byte(`foo`), append(nonIndexedLabels, labels.Label{
+		Name: "foo", Value: "baz",
+	})...)
+	require.True(t, ok)
+	require.Equal(t, 1., f)
+	assertLabelResult(t, expectedLabelsResults, l)
+
+	f, l, ok = sse.ProcessString(0, `foo`, append(nonIndexedLabels, labels.Label{
+		Name: "foo", Value: "baz",
+	})...)
+	require.True(t, ok)
+	require.Equal(t, 1., f)
+	assertLabelResult(t, expectedLabelsResults, l)
+
 	se, err = NewLineSampleExtractor(BytesExtractor, []Stage{
 		NewStringLabelFilter(labels.MustNewMatcher(labels.MatchEqual, "foo", "bar")),
 		NewStringLabelFilter(labels.MustNewMatcher(labels.MatchEqual, "user", "bob")),

pkg/logql/log/pipeline_test.go

@@ -39,6 +39,18 @@ func TestNoopPipeline(t *testing.T) {
 	require.Equal(t, NewLabelsResult(expectedLabelsResults, expectedLabelsResults.Hash()), lbr)
 	require.Equal(t, true, matches)
 
+	// test duplicated non-indexed labels with stream labels
+	expectedLabelsResults = append(lbs, labels.Label{
+		Name: "foo_extracted", Value: "baz",
+	})
+	expectedLabelsResults = append(expectedLabelsResults, nonIndexedLabels...)
+	l, lbr, matches = pipeline.ForStream(lbs).Process(0, []byte(""), append(nonIndexedLabels, labels.Label{
+		Name: "foo", Value: "baz",
+	})...)
+	require.Equal(t, []byte(""), l)
+	require.Equal(t, NewLabelsResult(expectedLabelsResults, expectedLabelsResults.Hash()), lbr)
+	require.Equal(t, true, matches)
+
 	pipeline.Reset()
 	require.Len(t, pipeline.cache, 0)
 }
@@ -100,6 +112,25 @@ func TestPipelineWithNonIndexedLabels(t *testing.T) {
 	require.Equal(t, NewLabelsResult(expectedLabelsResults, expectedLabelsResults.Hash()), lbr)
 	require.Equal(t, true, matches)
 
+	// test duplicated non-indexed labels with stream labels
+	expectedLabelsResults = append(lbs, labels.Label{
+		Name: "foo_extracted", Value: "baz",
+	})
+	expectedLabelsResults = append(expectedLabelsResults, nonIndexedLabels...)
+	l, lbr, matches = p.ForStream(lbs).Process(0, []byte("line"), append(nonIndexedLabels, labels.Label{
+		Name: "foo", Value: "baz",
+	})...)
+	require.Equal(t, []byte("lbs bar bob"), l)
+	require.Equal(t, NewLabelsResult(expectedLabelsResults, expectedLabelsResults.Hash()), lbr)
+	require.Equal(t, true, matches)
+
+	ls, lbr, matches = p.ForStream(lbs).ProcessString(0, "line", append(nonIndexedLabels, labels.Label{
+		Name: "foo", Value: "baz",
+	})...)
+	require.Equal(t, "lbs bar bob", ls)
+	require.Equal(t, NewLabelsResult(expectedLabelsResults, expectedLabelsResults.Hash()), lbr)
+	require.Equal(t, true, matches)
+
 	l, lbr, matches = p.ForStream(lbs).Process(0, []byte("line"))
 	require.Equal(t, []byte(nil), l)
 	require.Equal(t, nil, lbr)