Optimize deletes in tsi

The DropSeries code path ended up creating a MeasurementSeriesIterator
for each dropped series, this was too expensive just to see if a
series exists.

This adds a HasSeries func and fixes and issue where TSI files were
compacted while an iterator was still in use causing a panic.

tsdb/engine/tsm1/engine_test.go

@@ -1109,3 +1109,5 @@ func (itr *seriesIterator) Next() tsdb.SeriesElem {
 	itr.keys = itr.keys[1:]
 	return s
 }
+
+func (itr *seriesIterator) Close() {}

tsdb/index.go

@@ -76,6 +76,7 @@ type SeriesElem interface {
 // SeriesIterator represents a iterator over a list of series.
 type SeriesIterator interface {
 	Next() SeriesElem
+	Close()
 }
 
 // IndexFormat represents the format for an index.

tsdb/index/inmem/inmem.go

@@ -1060,3 +1060,5 @@ func (itr *seriesIterator) Next() tsdb.SeriesElem {
 	itr.keys = itr.keys[1:]
 	return &itr.elem
 }
+
+func (itr *seriesIterator) Close() {}

tsdb/index/tsi1/file_set_test.go

@@ -30,6 +30,7 @@ func TestFileSet_SeriesIterator(t *testing.T) {
 		if itr == nil {
 			t.Fatal("expected iterator")
 		}
+		defer itr.Close()
 
 		if e := itr.Next(); string(e.Name()) != `cpu` || e.Tags().String() != `[{region east}]` {
 			t.Fatalf("unexpected series: %s/%s", e.Name(), e.Tags().String())
@@ -60,6 +61,7 @@ func TestFileSet_SeriesIterator(t *testing.T) {
 		if itr == nil {
 			t.Fatal("expected iterator")
 		}
+		defer itr.Close()
 
 		if e := itr.Next(); string(e.Name()) != `cpu` || e.Tags().String() != `[{region east}]` {
 			t.Fatalf("unexpected series: %s/%s", e.Name(), e.Tags().String())
@@ -100,6 +102,7 @@ func TestFileSet_MeasurementSeriesIterator(t *testing.T) {
 		if itr == nil {
 			t.Fatal("expected iterator")
 		}
+		defer itr.Close()
 
 		if e := itr.Next(); string(e.Name()) != `cpu` || e.Tags().String() != `[{region east}]` {
 			t.Fatalf("unexpected series: %s/%s", e.Name(), e.Tags().String())
@@ -127,6 +130,7 @@ func TestFileSet_MeasurementSeriesIterator(t *testing.T) {
 		if itr == nil {
 			t.Fatalf("expected iterator")
 		}
+		defer itr.Close()
 
 		if e := itr.Next(); string(e.Name()) != `cpu` || e.Tags().String() != `[{region east}]` {
 			t.Fatalf("unexpected series: %s/%s", e.Name(), e.Tags().String())

tsdb/index/tsi1/index.go

@@ -467,6 +467,8 @@ func (i *Index) DropMeasurement(name []byte) error {
 
 	// Delete all series in measurement.
 	if sitr := fs.MeasurementSeriesIterator(name); sitr != nil {
+		defer sitr.Close()
+
 		for s := sitr.Next(); s != nil; s = sitr.Next() {
 			if !s.Deleted() {
 				if err := func() error {
@@ -575,11 +577,8 @@ func (i *Index) DropSeries(key []byte, ts int64) error {
 		fs := i.retainFileSet()
 		defer fs.Release()
 
-		// Check if that was the last series for the measurement in the entire index.
-		itr := fs.MeasurementSeriesIterator(mname)
-		if itr == nil {
-			return nil
-		} else if e := itr.Next(); e != nil {
+		mm := fs.Measurement(mname)
+		if mm == nil || mm.HasSeries() {
 			return nil
 		}
 

tsdb/index/tsi1/index_file.go

@@ -251,7 +251,7 @@ func (f *IndexFile) TagKeySeriesIterator(name, key []byte) tsdb.SeriesIterator {
 	var itrs []tsdb.SeriesIterator
 	for ve := vitr.Next(); ve != nil; ve = vitr.Next() {
 		sitr := &rawSeriesIDIterator{data: ve.(*TagBlockValueElem).series.data}
-		itrs = append(itrs, newSeriesDecodeIterator(&f.sblk, sitr))
+		itrs = append(itrs, newSeriesDecodeIterator(f, &f.sblk, sitr))
 	}
 
 	return MergeSeriesIterators(itrs...)
@@ -273,6 +273,7 @@ func (f *IndexFile) TagValueSeriesIterator(name, key, value []byte) tsdb.SeriesI
 
 	// Create an iterator over value's series.
 	return newSeriesDecodeIterator(
+		f,
 		&f.sblk,
 		&rawSeriesIDIterator{
 			n:    ve.(*TagBlockValueElem).series.n,

tsdb/index/tsi1/log_file.go

@@ -267,7 +267,7 @@ func (f *LogFile) TagKeySeriesIterator(name, key []byte) tsdb.SeriesIterator {
 		if len(tv.series) == 0 {
 			continue
 		}
-		itrs = append(itrs, newLogSeriesIterator(tv.series))
+		itrs = append(itrs, newLogSeriesIterator(f, tv.series))
 	}
 
 	return MergeSeriesIterators(itrs...)
@@ -383,7 +383,7 @@ func (f *LogFile) TagValueSeriesIterator(name, key, value []byte) tsdb.SeriesIte
 		return nil
 	}
 
-	return newLogSeriesIterator(tv.series)
+	return newLogSeriesIterator(f, tv.series)
 }
 
 // MeasurementN returns the total number of measurements.
@@ -753,7 +753,8 @@ func (f *LogFile) SeriesIterator() tsdb.SeriesIterator {
 	if len(series) == 0 {
 		return nil
 	}
-	return &logSeriesIterator{series: series}
+	f.Retain()
+	return &logSeriesIterator{f: f, series: series}
 }
 
 // createMeasurementIfNotExists returns a measurement by name.
@@ -792,7 +793,7 @@ func (f *LogFile) MeasurementSeriesIterator(name []byte) tsdb.SeriesIterator {
 	if mm == nil || len(mm.series) == 0 {
 		return nil
 	}
-	return newLogSeriesIterator(mm.series)
+	return newLogSeriesIterator(f, mm.series)
 }
 
 // CompactTo compacts the log file and writes it to w.
@@ -1290,6 +1291,17 @@ type logMeasurement struct {
 
 func (m *logMeasurement) Name() []byte  { return m.name }
 func (m *logMeasurement) Deleted() bool { return m.deleted }
+func (m *logMeasurement) HasSeries() bool {
+	if m.deleted {
+		return false
+	}
+	for _, v := range m.series {
+		if !v.deleted {
+			return true
+		}
+	}
+	return false
+}
 
 func (m *logMeasurement) createTagSetIfNotExists(key []byte) logTagKey {
 	ts, ok := m.tagSet[string(key)]
@@ -1432,17 +1444,19 @@ func (itr *logTagValueIterator) Next() (e TagValueElem) {
 
 // logSeriesIterator represents an iterator over a slice of series.
 type logSeriesIterator struct {
+	f      *LogFile
 	series logSeries
 }
 
 // newLogSeriesIterator returns a new instance of logSeriesIterator.
 // All series are copied to the iterator.
-func newLogSeriesIterator(m map[string]*logSerie) *logSeriesIterator {
+func newLogSeriesIterator(f *LogFile, m map[string]*logSerie) *logSeriesIterator {
 	if len(m) == 0 {
 		return nil
 	}
 
-	itr := logSeriesIterator{series: make(logSeries, 0, len(m))}
+	f.Retain()
+	itr := logSeriesIterator{f: f, series: make(logSeries, 0, len(m))}
 	for _, s := range m {
 		itr.series = append(itr.series, *s)
 	}
@@ -1460,6 +1474,10 @@ func (itr *logSeriesIterator) Next() (e tsdb.SeriesElem) {
 	return e
 }
 
+func (itr *logSeriesIterator) Close() {
+	itr.f.Release()
+}
+
 // FormatLogFileName generates a log filename for the given index.
 func FormatLogFileName(id int) string {
 	return fmt.Sprintf("L0-%08d%s", id, LogFileExt)

tsdb/index/tsi1/log_file_test.go

@@ -93,6 +93,7 @@ func TestLogFile_SeriesStoredInOrder(t *testing.T) {
 	if itr == nil {
 		t.Fatal("nil iterator")
 	}
+	defer itr.Close()
 
 	mname := []string{"cpu", "mem"}
 	var j int

tsdb/index/tsi1/measurement_block.go

@@ -337,6 +337,8 @@ func (e *MeasurementBlockElem) SeriesID(i int) uint32 {
 	return binary.BigEndian.Uint32(e.series.data[i*SeriesIDSize:])
 }
 
+func (e *MeasurementBlockElem) HasSeries() bool { return e.series.n > 0 }
+
 // SeriesIDs returns a list of decoded series ids.
 //
 // NOTE: This should be used for testing and diagnostics purposes only.

tsdb/index/tsi1/series_block.go

@@ -283,16 +283,20 @@ func (itr *seriesBlockIterator) Next() tsdb.SeriesElem {
 	}
 }
 
+func (itr *seriesBlockIterator) Close() {}
+
 // seriesDecodeIterator decodes a series id iterator into unmarshaled elements.
 type seriesDecodeIterator struct {
+	f    *IndexFile
 	itr  seriesIDIterator
 	sblk *SeriesBlock
 	e    SeriesBlockElem // buffer
 }
 
 // newSeriesDecodeIterator returns a new instance of seriesDecodeIterator.
-func newSeriesDecodeIterator(sblk *SeriesBlock, itr seriesIDIterator) *seriesDecodeIterator {
-	return &seriesDecodeIterator{sblk: sblk, itr: itr}
+func newSeriesDecodeIterator(f *IndexFile, sblk *SeriesBlock, itr seriesIDIterator) *seriesDecodeIterator {
+	f.Retain()
+	return &seriesDecodeIterator{f: f, sblk: sblk, itr: itr}
 }
 
 // Next returns the next series element.
@@ -308,6 +312,10 @@ func (itr *seriesDecodeIterator) Next() tsdb.SeriesElem {
 	return &itr.e
 }
 
+func (itr *seriesDecodeIterator) Close() {
+	itr.f.Release()
+}
+
 // SeriesBlockElem represents a series element in the series list.
 type SeriesBlockElem struct {
 	flag byte

tsdb/index/tsi1/tsi1.go

@@ -21,6 +21,7 @@ const LoadFactor = 80
 type MeasurementElem interface {
 	Name() []byte
 	Deleted() bool
+	HasSeries() bool
 }
 
 // MeasurementElems represents a list of MeasurementElem.
@@ -115,6 +116,15 @@ func (p measurementMergeElem) Deleted() bool {
 	return p[0].Deleted()
 }
 
+func (p measurementMergeElem) HasSeries() bool {
+	for _, v := range p {
+		if v.HasSeries() {
+			return true
+		}
+	}
+	return false
+}
+
 // filterUndeletedMeasurementIterator returns all measurements which are not deleted.
 type filterUndeletedMeasurementIterator struct {
 	itr MeasurementIterator
@@ -455,6 +465,12 @@ func (itr *seriesMergeIterator) Next() tsdb.SeriesElem {
 	return e
 }
 
+func (itr *seriesMergeIterator) Close() {
+	for _, v := range itr.itrs {
+		v.Close()
+	}
+}
+
 // IntersectSeriesIterators returns an iterator that only returns series which
 // occur in both iterators. If both series have associated expressions then
 // they are combined together.
@@ -521,6 +537,12 @@ func (itr *seriesIntersectIterator) Next() (e tsdb.SeriesElem) {
 	}
 }
 
+func (itr *seriesIntersectIterator) Close() {
+	for _, v := range itr.itrs {
+		v.Close()
+	}
+}
+
 // UnionSeriesIterators returns an iterator that returns series from both
 // both iterators. If both series have associated expressions then they are
 // combined together.
@@ -590,6 +612,12 @@ func (itr *seriesUnionIterator) Next() (e tsdb.SeriesElem) {
 	return &itr.e
 }
 
+func (itr *seriesUnionIterator) Close() {
+	for _, v := range itr.itrs {
+		v.Close()
+	}
+}
+
 // DifferenceSeriesIterators returns an iterator that only returns series which
 // occur the first iterator but not the second iterator.
 func DifferenceSeriesIterators(itr0, itr1 tsdb.SeriesIterator) tsdb.SeriesIterator {
@@ -642,6 +670,12 @@ func (itr *seriesDifferenceIterator) Next() (e tsdb.SeriesElem) {
 	}
 }
 
+func (itr *seriesDifferenceIterator) Close() {
+	for _, v := range itr.itrs {
+		v.Close()
+	}
+}
+
 // filterUndeletedSeriesIterator returns all series which are not deleted.
 type filterUndeletedSeriesIterator struct {
 	itr tsdb.SeriesIterator
@@ -667,6 +701,10 @@ func (itr *filterUndeletedSeriesIterator) Next() tsdb.SeriesElem {
 	}
 }
 
+func (itr *filterUndeletedSeriesIterator) Close() {
+	itr.itr.Close()
+}
+
 // seriesExprElem holds a series and its associated filter expression.
 type seriesExprElem struct {
 	tsdb.SeriesElem
@@ -705,6 +743,10 @@ func (itr *seriesExprIterator) Next() tsdb.SeriesElem {
 	return &itr.e
 }
 
+func (itr *seriesExprIterator) Close() {
+	itr.itr.Close()
+}
+
 // seriesIDIterator represents a iterator over a list of series ids.
 type seriesIDIterator interface {
 	next() uint32

tsdb/index/tsi1/tsi1_test.go

@@ -202,12 +202,15 @@ func TestMergeSeriesIterators(t *testing.T) {
 
 // MeasurementElem represents a test implementation of tsi1.MeasurementElem.
 type MeasurementElem struct {
-	name    []byte
-	deleted bool
+	name      []byte
+	deleted   bool
+	hasSeries bool
 }
 
-func (e *MeasurementElem) Name() []byte                        { return e.name }
-func (e *MeasurementElem) Deleted() bool                       { return e.deleted }
+func (e *MeasurementElem) Name() []byte    { return e.name }
+func (e *MeasurementElem) Deleted() bool   { return e.deleted }
+func (e *MeasurementElem) HasSeries() bool { return e.hasSeries }
+
 func (e *MeasurementElem) TagKeyIterator() tsi1.TagKeyIterator { return nil }
 
 // MeasurementIterator represents an iterator over a slice of measurements.
@@ -299,6 +302,8 @@ func (itr *SeriesIterator) Next() (e tsdb.SeriesElem) {
 	return e
 }
 
+func (itr *SeriesIterator) Close() {}
+
 // MustTempDir returns a temporary directory. Panic on error.
 func MustTempDir() string {
 	path, err := ioutil.TempDir("", "tsi-")