From 75d83e429354a6b7351ba4a4cf7337ba1cc7667d Mon Sep 17 00:00:00 2001
From: Ben Ye <yb532204897@gmail.com>
Date: Fri, 5 Mar 2021 14:09:07 -0500
Subject: [PATCH] Tools: rewrite delete can delete series only if it matches
 all matchers (#3886)

* rewrite delete should delete a series only if it matches all matchers in a deletion request

Signed-off-by: yeya24 <yb532204897@gmail.com>

* add test case

Signed-off-by: yeya24 <yb532204897@gmail.com>
---
 pkg/compactv2/compactor_test.go | 32 +++++++++++++++++++++++
 pkg/compactv2/modifiers.go      | 46 +++++++++++++++++----------------
 2 files changed, 56 insertions(+), 22 deletions(-)

diff --git a/pkg/compactv2/compactor_test.go b/pkg/compactv2/compactor_test.go
index cc10a1897a..c1af292d1b 100644
--- a/pkg/compactv2/compactor_test.go
+++ b/pkg/compactv2/compactor_test.go
@@ -274,6 +274,38 @@ func TestCompactor_WriteSeries_e2e(t *testing.T) {
 				NumChunks:  2,
 			},
 		},
+		{
+			name: "1 blocks + delete modifier, deletion request contains multiple matchers, delete second series",
+			input: [][]seriesSamples{
+				{
+					{lset: labels.Labels{{Name: "a", Value: "1"}, {Name: "b", Value: "1"}},
+						chunks: [][]sample{{{0, 0}, {1, 1}, {2, 2}, {10, 10}, {11, 11}, {20, 20}}}},
+					{lset: labels.Labels{{Name: "a", Value: "1"}, {Name: "b", Value: "2"}},
+						chunks: [][]sample{{{0, 0}, {1, 1}, {2, 2}}, {{10, 11}, {11, 11}, {20, 20}}}},
+					{lset: labels.Labels{{Name: "a", Value: "3"}},
+						chunks: [][]sample{{{0, 0}, {1, 1}, {2, 2}, {10, 12}, {11, 11}, {20, 20}}}},
+				},
+			},
+			modifiers: []Modifier{WithDeletionModifier(
+				metadata.DeletionRequest{
+					Matchers: []*labels.Matcher{
+						labels.MustNewMatcher(labels.MatchEqual, "a", "1"),
+						labels.MustNewMatcher(labels.MatchEqual, "b", "2"),
+					},
+				})},
+			expected: []seriesSamples{
+				{lset: labels.Labels{{Name: "a", Value: "1"}, {Name: "b", Value: "1"}},
+					chunks: [][]sample{{{0, 0}, {1, 1}, {2, 2}, {10, 10}, {11, 11}, {20, 20}}}},
+				{lset: labels.Labels{{Name: "a", Value: "3"}},
+					chunks: [][]sample{{{0, 0}, {1, 1}, {2, 2}, {10, 12}, {11, 11}, {20, 20}}}},
+			},
+			expectedChanges: "Deleted {a=\"1\", b=\"2\"} [{0 20}]\n",
+			expectedStats: tsdb.BlockStats{
+				NumSamples: 12,
+				NumSeries:  2,
+				NumChunks:  2,
+			},
+		},
 	} {
 		t.Run(tcase.name, func(t *testing.T) {
 			tmpDir, err := ioutil.TempDir("", "test-series-writer")
diff --git a/pkg/compactv2/modifiers.go b/pkg/compactv2/modifiers.go
index 2139d98d85..9d521c0fd2 100644
--- a/pkg/compactv2/modifiers.go
+++ b/pkg/compactv2/modifiers.go
@@ -58,6 +58,7 @@ SeriesLoop:
 		lbls := s.Labels()
 
 		var intervals tombstones.Intervals
+	DeletionsLoop:
 		for _, deletions := range d.d.deletions {
 			for _, m := range deletions.Matchers {
 				v := lbls.Get(m.Name)
@@ -65,34 +66,35 @@ SeriesLoop:
 					continue
 				}
 
+				// Only if all matchers in the deletion request are matched can we proceed to deletion.
 				if !m.Matches(v) {
-					continue
+					continue DeletionsLoop
 				}
-				if len(deletions.Intervals) > 0 {
-					for _, in := range deletions.Intervals {
-						intervals = intervals.Add(in)
-					}
-					break
+			}
+			if len(deletions.Intervals) > 0 {
+				for _, in := range deletions.Intervals {
+					intervals = intervals.Add(in)
 				}
+				continue
+			}
 
-				// Special case: Delete whole series.
-				chksIter := s.Iterator()
-				var chks []chunks.Meta
-				for chksIter.Next() {
-					chks = append(chks, chksIter.At())
-				}
-				if d.err = chksIter.Err(); d.err != nil {
-					return false
-				}
+			// Special case: Delete whole series.
+			chksIter := s.Iterator()
+			var chks []chunks.Meta
+			for chksIter.Next() {
+				chks = append(chks, chksIter.At())
+			}
+			if d.err = chksIter.Err(); d.err != nil {
+				return false
+			}
 
-				var deleted tombstones.Intervals
-				if len(chks) > 0 {
-					deleted = deleted.Add(tombstones.Interval{Mint: chks[0].MinTime, Maxt: chks[len(chks)-1].MaxTime})
-				}
-				d.log.DeleteSeries(lbls, deleted)
-				d.p.SeriesProcessed()
-				continue SeriesLoop
+			var deleted tombstones.Intervals
+			if len(chks) > 0 {
+				deleted = deleted.Add(tombstones.Interval{Mint: chks[0].MinTime, Maxt: chks[len(chks)-1].MaxTime})
 			}
+			d.log.DeleteSeries(lbls, deleted)
+			d.p.SeriesProcessed()
+			continue SeriesLoop
 		}
 
 		d.curr = &storage.ChunkSeriesEntry{