Fixes log deduplication when mutating Labels using LogQL

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## Main
 
+* [5289](https://github.com/grafana/loki/pull/5289) **ctovena**: Fix deduplication bug in queries when mutating labels.
 * [5302](https://github.com/grafana/loki/pull/5302) **MasslessParticle** Update azure blobstore client to use new sdk.
 * [5243](https://github.com/grafana/loki/pull/5290) **ssncferreira**: Update Promtail to support duration string formats.
 * [5266](https://github.com/grafana/loki/pull/5266) **jeschkies**: Write Promtail position file atomically on Unix.

pkg/chunkenc/dumb_chunk.go

@@ -153,6 +153,10 @@ func (i *dumbChunkIterator) Labels() string {
 	return ""
 }
 
+func (i *dumbChunkIterator) StreamHash() uint64 {
+	return 0
+}
+
 func (i *dumbChunkIterator) Error() error {
 	return nil
 }

pkg/chunkenc/memchunk.go

@@ -816,7 +816,7 @@ func (c *MemChunk) Iterator(ctx context.Context, mintT, maxtT time.Time, directi
 
 		var it iter.EntryIterator
 		if ordered {
-			it = iter.NewNonOverlappingIterator(blockItrs, "")
+			it = iter.NewNonOverlappingIterator(blockItrs)
 		} else {
 			it = iter.NewSortEntryIterator(blockItrs, direction)
 		}
@@ -849,7 +849,7 @@ func (c *MemChunk) Iterator(ctx context.Context, mintT, maxtT time.Time, directi
 	}
 
 	if ordered {
-		return iter.NewNonOverlappingIterator(blockItrs, ""), nil
+		return iter.NewNonOverlappingIterator(blockItrs), nil
 	}
 	return iter.NewSortEntryIterator(blockItrs, direction), nil
 }
@@ -884,7 +884,7 @@ func (c *MemChunk) SampleIterator(ctx context.Context, from, through time.Time,
 
 	var it iter.SampleIterator
 	if ordered {
-		it = iter.NewNonOverlappingSampleIterator(its, "")
+		it = iter.NewNonOverlappingSampleIterator(its)
 	} else {
 		it = iter.NewSortSampleIterator(its)
 	}
@@ -1252,6 +1252,8 @@ func (e *entryBufferedIterator) Entry() logproto.Entry {
 
 func (e *entryBufferedIterator) Labels() string { return e.currLabels.String() }
 
+func (e *entryBufferedIterator) StreamHash() uint64 { return e.pipeline.BaseLabels().Hash() }
+
 func (e *entryBufferedIterator) Next() bool {
 	for e.bufferedIterator.Next() {
 		newLine, lbs, ok := e.pipeline.Process(e.currLine)
@@ -1299,6 +1301,8 @@ func (e *sampleBufferedIterator) Next() bool {
 }
 func (e *sampleBufferedIterator) Labels() string { return e.currLabels.String() }
 
+func (e *sampleBufferedIterator) StreamHash() uint64 { return e.extractor.BaseLabels().Hash() }
+
 func (e *sampleBufferedIterator) Sample() logproto.Sample {
 	return e.cur
 }

pkg/chunkenc/memchunk_test.go

@@ -677,6 +677,7 @@ func BenchmarkWrite(b *testing.B) {
 
 type nomatchPipeline struct{}
 
+func (nomatchPipeline) BaseLabels() log.LabelsResult                         { return log.EmptyLabelsResult }
 func (nomatchPipeline) Process(line []byte) ([]byte, log.LabelsResult, bool) { return line, nil, false }
 func (nomatchPipeline) ProcessString(line string) (string, log.LabelsResult, bool) {
 	return line, nil, false

pkg/ingester/stream.go

@@ -468,7 +468,7 @@ func (s *stream) Iterator(ctx context.Context, statsCtx *stats.Context, from, th
 	}
 
 	if ordered {
-		return iter.NewNonOverlappingIterator(iterators, ""), nil
+		return iter.NewNonOverlappingIterator(iterators), nil
 	}
 	return iter.NewSortEntryIterator(iterators, direction), nil
 }
@@ -505,7 +505,7 @@ func (s *stream) SampleIterator(ctx context.Context, statsCtx *stats.Context, fr
 	}
 
 	if ordered {
-		return iter.NewNonOverlappingSampleIterator(iterators, ""), nil
+		return iter.NewNonOverlappingSampleIterator(iterators), nil
 	}
 	return iter.NewSortSampleIterator(iterators), nil
 }

pkg/iter/cache.go

@@ -53,7 +53,7 @@ func (it *cachedIterator) consumeWrapped() bool {
 		return false
 	}
 	// we're caching entries
-	it.cache = append(it.cache, entryWithLabels{entry: it.Wrapped().Entry(), labels: it.Wrapped().Labels()})
+	it.cache = append(it.cache, entryWithLabels{entry: it.Wrapped().Entry(), labels: it.Wrapped().Labels(), streamHash: it.Wrapped().StreamHash()})
 	it.curr++
 	return true
 }
@@ -87,6 +87,13 @@ func (it *cachedIterator) Labels() string {
 	return it.cache[it.curr].labels
 }
 
+func (it *cachedIterator) StreamHash() uint64 {
+	if len(it.cache) == 0 || it.curr < 0 || it.curr >= len(it.cache) {
+		return 0
+	}
+	return it.cache[it.curr].streamHash
+}
+
 func (it *cachedIterator) Error() error { return it.iterErr }
 
 func (it *cachedIterator) Close() error {
@@ -143,7 +150,7 @@ func (it *cachedSampleIterator) consumeWrapped() bool {
 		return false
 	}
 	// we're caching entries
-	it.cache = append(it.cache, sampleWithLabels{Sample: it.Wrapped().Sample(), labels: it.Wrapped().Labels()})
+	it.cache = append(it.cache, sampleWithLabels{Sample: it.Wrapped().Sample(), labels: it.Wrapped().Labels(), streamHash: it.Wrapped().StreamHash()})
 	it.curr++
 	return true
 }
@@ -176,6 +183,13 @@ func (it *cachedSampleIterator) Labels() string {
 	return it.cache[it.curr].labels
 }
 
+func (it *cachedSampleIterator) StreamHash() uint64 {
+	if len(it.cache) == 0 || it.curr < 0 || it.curr >= len(it.cache) {
+		return 0
+	}
+	return it.cache[it.curr].streamHash
+}
+
 func (it *cachedSampleIterator) Error() error { return it.iterErr }
 
 func (it *cachedSampleIterator) Close() error {

pkg/iter/cache_test.go

@@ -251,6 +251,7 @@ type errorIter struct{}
 func (errorIter) Next() bool              { return false }
 func (errorIter) Error() error            { return errors.New("error") }
 func (errorIter) Labels() string          { return "" }
+func (errorIter) StreamHash() uint64      { return 0 }
 func (errorIter) Entry() logproto.Entry   { return logproto.Entry{} }
 func (errorIter) Sample() logproto.Sample { return logproto.Sample{} }
 func (errorIter) Close() error            { return errors.New("close") }