Fix deletes triggering unnecessary compactions

CHANGELOG.md

@@ -59,6 +59,7 @@ The admin UI is removed and unusable in this release. The `[admin]` configuratio
 - [#7425](https://github.com/influxdata/influxdb/issues/7425): Fix compaction aborted log messages
 - [#8123](https://github.com/influxdata/influxdb/issues/8123): TSM compaction does not remove .tmp on error
 - [#8343](https://github.com/influxdata/influxdb/issues/8343): Set the CSV output to an empty string for null values.
+- [#8368](https://github.com/influxdata/influxdb/issues/8368): Compaction exhausting disk resources in InfluxDB
 
 ## v1.2.3 [unreleased]
 

tsdb/engine/tsm1/reader.go

@@ -74,6 +74,12 @@ type TSMIndex interface {
 	// KeyCount returns the count of unique keys in the index.
 	KeyCount() int
 
+	// OverlapsTimeRange returns true if the time range of the file intersect min and max.
+	OverlapsTimeRange(min, max int64) bool
+
+	// OverlapsKeyRange returns true if the min and max keys of the file overlap the arguments min and max.
+	OverlapsKeyRange(min, max string) bool
+
 	// Size returns the size of the current index in bytes.
 	Size() uint32
 
@@ -403,8 +409,24 @@ func (t *TSMReader) ContainsValue(key string, ts int64) bool {
 	return t.index.ContainsValue(key, ts)
 }
 
-// DeleteRange removes the given points for keys between minTime and maxTime.
+// DeleteRange removes the given points for keys between minTime and maxTime.   The series
+// keys passed in must be sorted.
 func (t *TSMReader) DeleteRange(keys []string, minTime, maxTime int64) error {
+	if len(keys) == 0 {
+		return nil
+	}
+
+	// If the keys can't exist in this TSM file, skip it.
+	minKey, maxKey := keys[0], keys[len(keys)-1]
+	if !t.index.OverlapsKeyRange(minKey, maxKey) {
+		return nil
+	}
+
+	// If the timerange can't exist in this TSM file, skip it.
+	if !t.index.OverlapsTimeRange(minTime, maxTime) {
+		return nil
+	}
+
 	if err := t.tombstoner.AddRange(keys, minTime, maxTime); err != nil {
 		return err
 	}
@@ -866,6 +888,16 @@ func (d *indirectIndex) Type(key string) (byte, error) {
 	return 0, fmt.Errorf("key does not exist: %v", key)
 }
 
+// OverlapsTimeRange returns true if the time range of the file intersect min and max.
+func (d *indirectIndex) OverlapsTimeRange(min, max int64) bool {
+	return d.minTime <= max && d.maxTime >= min
+}
+
+// OverlapsKeyRange returns true if the min and max keys of the file overlap the arguments min and max.
+func (d *indirectIndex) OverlapsKeyRange(min, max string) bool {
+	return d.minKey <= max && d.maxKey >= max
+}
+
 // KeyRange returns the min and max keys in the index.
 func (d *indirectIndex) KeyRange() (string, string) {
 	return d.minKey, d.maxKey

tsdb/engine/tsm1/reader_test.go

@@ -404,6 +404,135 @@ func TestTSMReader_MMAP_TombstoneRange(t *testing.T) {
 	}
 }
 
+func TestTSMReader_MMAP_TombstoneOutsideTimeRange(t *testing.T) {
+	dir := MustTempDir()
+	defer os.RemoveAll(dir)
+	f := MustTempFile(dir)
+	defer f.Close()
+
+	w, err := tsm1.NewTSMWriter(f)
+	if err != nil {
+		t.Fatalf("unexpected error creating writer: %v", err)
+	}
+
+	expValues := []tsm1.Value{
+		tsm1.NewValue(1, 1.0),
+		tsm1.NewValue(2, 2.0),
+		tsm1.NewValue(3, 3.0),
+	}
+	if err := w.Write("cpu", expValues); err != nil {
+		t.Fatalf("unexpected error writing: %v", err)
+	}
+
+	if err := w.WriteIndex(); err != nil {
+		t.Fatalf("unexpected error writing index: %v", err)
+	}
+
+	if err := w.Close(); err != nil {
+		t.Fatalf("unexpected error closing: %v", err)
+	}
+
+	f, err = os.Open(f.Name())
+	if err != nil {
+		t.Fatalf("unexpected error open file: %v", err)
+	}
+
+	r, err := tsm1.NewTSMReader(f)
+	if err != nil {
+		t.Fatalf("unexpected error created reader: %v", err)
+	}
+
+	if err := r.DeleteRange([]string{"cpu"}, 0, 0); err != nil {
+		t.Fatalf("unexpected error deleting: %v", err)
+	}
+	defer r.Close()
+
+	if got, exp := r.ContainsValue("cpu", 1), true; got != exp {
+		t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp)
+	}
+
+	if got, exp := r.ContainsValue("cpu", 2), true; got != exp {
+		t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp)
+	}
+
+	if got, exp := r.ContainsValue("cpu", 3), true; got != exp {
+		t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp)
+	}
+
+	if got, exp := r.HasTombstones(), false; got != exp {
+		t.Fatalf("HasTombstones mismatch: got %v, exp %v", got, exp)
+	}
+
+	if got, exp := len(r.TombstoneFiles()), 0; got != exp {
+		t.Fatalf("TombstoneFiles len mismatch: got %v, exp %v", got, exp)
+	}
+}
+
+func TestTSMReader_MMAP_TombstoneOutsideKeyRange(t *testing.T) {
+	dir := MustTempDir()
+	defer os.RemoveAll(dir)
+	f := MustTempFile(dir)
+	defer f.Close()
+
+	w, err := tsm1.NewTSMWriter(f)
+	if err != nil {
+		t.Fatalf("unexpected error creating writer: %v", err)
+	}
+
+	expValues := []tsm1.Value{
+		tsm1.NewValue(1, 1.0),
+		tsm1.NewValue(2, 2.0),
+		tsm1.NewValue(3, 3.0),
+	}
+	if err := w.Write("cpu", expValues); err != nil {
+		t.Fatalf("unexpected error writing: %v", err)
+	}
+
+	if err := w.WriteIndex(); err != nil {
+		t.Fatalf("unexpected error writing index: %v", err)
+	}
+
+	if err := w.Close(); err != nil {
+		t.Fatalf("unexpected error closing: %v", err)
+	}
+
+	f, err = os.Open(f.Name())
+	if err != nil {
+		t.Fatalf("unexpected error open file: %v", err)
+	}
+
+	r, err := tsm1.NewTSMReader(f)
+	if err != nil {
+		t.Fatalf("unexpected error created reader: %v", err)
+	}
+
+	if err := r.DeleteRange([]string{"mem"}, 0, 3); err != nil {
+		t.Fatalf("unexpected error deleting: %v", err)
+	}
+	defer r.Close()
+
+	if got, exp := r.ContainsValue("cpu", 1), true; got != exp {
+		t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp)
+	}
+
+	if got, exp := r.ContainsValue("cpu", 2), true; got != exp {
+		t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp)
+	}
+
+	if got, exp := r.ContainsValue("cpu", 3), true; got != exp {
+		t.Fatalf("ContainsValue mismatch: got %v, exp %v", got, exp)
+	}
+
+	if got, exp := r.HasTombstones(), false; got != exp {
+		t.Fatalf("HasTombstones mismatch: got %v, exp %v", got, exp)
+	}
+
+	if got, exp := len(r.TombstoneFiles()), 0; got != exp {
+		t.Fatalf("TombstoneFiles len mismatch: got %v, exp %v", got, exp)
+
+	}
+}
+
 func TestTSMReader_MMAP_TombstoneFullRange(t *testing.T) {
 	dir := MustTempDir()
 	defer os.RemoveAll(dir)