db: find strict overlap with excise span

`Version.Overlaps()` returns all "indirect" overlaps in L0. This is
not what we want when we look for files to excise.

This change replaces `Overlaps()` with using `LevelSlice.Overlaps()`
on each sublevel/level.

Author: RaduBerinde

compaction.go

@@ -1348,23 +1348,21 @@ func (d *DB) runIngestFlush(c *compaction) (*manifest.VersionEdit, error) {
 	}
 	if ingestFlushable.exciseSpan.Valid() {
 		// Iterate through all levels and find files that intersect with exciseSpan.
-		for l := range c.version.Levels {
-			overlaps := c.version.Overlaps(l, base.UserKeyBoundsEndExclusive(ingestFlushable.exciseSpan.Start, ingestFlushable.exciseSpan.End))
-			for m := range overlaps.All() {
-				newFiles, err := d.excise(context.TODO(), ingestFlushable.exciseSpan.UserKeyBounds(), m, ve, l)
+		for layer, ls := range c.version.AllLevelsAndSublevels() {
+			for m := range ls.Overlaps(d.cmp, ingestFlushable.exciseSpan.UserKeyBounds()).All() {
+				newFiles, err := d.excise(context.TODO(), ingestFlushable.exciseSpan.UserKeyBounds(), m, ve, layer.Level())
 				if err != nil {
 					return nil, err
 				}
 
 				if _, ok := ve.DeletedTables[deletedFileEntry{
-					Level:   l,
+					Level:   layer.Level(),
 					FileNum: m.FileNum,
 				}]; !ok {
-					// We did not excise this file.
-					continue
+					panic("did not excise file that overlaps with the excise span")
 				}
 				replacedFiles[m.FileNum] = newFiles
-				updateLevelMetricsOnExcise(m, l, newFiles)
+				updateLevelMetricsOnExcise(m, layer.Level(), newFiles)
 			}
 		}
 	}

download.go

@@ -248,13 +248,14 @@ func (d *DB) newDownloadSpanTask(vers *version, sp DownloadSpan) (_ *downloadSpa
 	// [sp.StartKey, sp.EndKey). Expand the bounds to the left so that we
 	// include the start keys of any external sstables that overlap with
 	// sp.StartKey.
-	vers.IterAllLevelsAndSublevels(func(iter manifest.LevelIterator, level manifest.Layer) {
+	for _, ls := range vers.AllLevelsAndSublevels() {
+		iter := ls.Iter()
 		if f := iter.SeekGE(d.cmp, sp.StartKey); f != nil &&
 			objstorage.IsExternalTable(d.objProvider, f.FileBacking.DiskFileNum) &&
 			d.cmp(f.Smallest.UserKey, bounds.Start) < 0 {
 			bounds.Start = f.Smallest.UserKey
 		}
-	})
+	}
 	startCursor := downloadCursor{
 		level:  0,
 		key:    bounds.Start,

ingest.go

@@ -2102,22 +2102,21 @@ func (d *DB) ingestApply(
 		// see if any new compactions are conflicting with our chosen target levels
 		// for files, and if they are, we should signal those compactions to error
 		// out.
-		for level := range current.Levels {
-			for m := range current.Overlaps(level, exciseSpan.UserKeyBounds()).All() {
-				newFiles, err := d.excise(ctx, exciseSpan.UserKeyBounds(), m, ve, level)
+		for layer, ls := range current.AllLevelsAndSublevels() {
+			for m := range ls.Overlaps(d.cmp, exciseSpan.UserKeyBounds()).All() {
+				newFiles, err := d.excise(ctx, exciseSpan.UserKeyBounds(), m, ve, layer.Level())
 				if err != nil {
 					return nil, err
 				}
 
 				if _, ok := ve.DeletedTables[deletedFileEntry{
-					Level:   level,
+					Level:   layer.Level(),
 					FileNum: m.FileNum,
 				}]; !ok {
-					// We did not excise this file.
-					continue
+					panic("did not excise file that overlaps with the excise span")
 				}
 				replacedFiles[m.FileNum] = newFiles
-				updateLevelMetricsOnExcise(m, level, newFiles)
+				updateLevelMetricsOnExcise(m, layer.Level(), newFiles)
 			}
 		}
 	}

ingest_test.go

@@ -892,14 +892,15 @@ func TestExcise(t *testing.T) {
 			d.mu.Unlock()
 			current := d.mu.versions.currentVersion()
 
-			current.IterAllLevelsAndSublevels(func(iter manifest.LevelIterator, l manifest.Layer) {
+			for l, ls := range current.AllLevelsAndSublevels() {
+				iter := ls.Iter()
 				for m := iter.SeekGE(d.cmp, exciseSpan.Start); m != nil && d.cmp(m.Smallest.UserKey, exciseSpan.End) < 0; m = iter.Next() {
 					_, err := d.excise(context.Background(), exciseSpan.UserKeyBounds(), m, ve, l.Level())
 					if err != nil {
 						td.Fatalf(t, "error when excising %s: %s", m.FileNum, err.Error())
 					}
 				}
-			})
+			}
 
 			d.mu.Lock()
 			d.mu.versions.logUnlock()

internal/manifest/version.go

@@ -1653,14 +1653,20 @@ func (v *Version) Overlaps(level int, bounds base.UserKeyBounds) LevelSlice {
 	return v.Levels[level].Slice().Overlaps(v.cmp.Compare, bounds)
 }
 
-// IterAllLevelsAndSublevels calls fn with an iterator for each L0 sublevel
-// (from top to bottom), then once for each level below L0.
-func (v *Version) IterAllLevelsAndSublevels(fn func(it LevelIterator, level Layer)) {
-	for sublevel := len(v.L0SublevelFiles) - 1; sublevel >= 0; sublevel-- {
-		fn(v.L0SublevelFiles[sublevel].Iter(), L0Sublevel(sublevel))
-	}
-	for level := 1; level < NumLevels; level++ {
-		fn(v.Levels[level].Iter(), Level(level))
+// AllLevelsAndSublevels returns an iterator that produces a Layer, LevelSlice
+// pair for each L0 sublevel (from top to bottom) and each level below L0.
+func (v *Version) AllLevelsAndSublevels() iter.Seq2[Layer, LevelSlice] {
+	return func(yield func(Layer, LevelSlice) bool) {
+		for sublevel := len(v.L0SublevelFiles) - 1; sublevel >= 0; sublevel-- {
+			if !yield(L0Sublevel(sublevel), v.L0SublevelFiles[sublevel]) {
+				return
+			}
+		}
+		for level := 1; level < NumLevels; level++ {
+			if !yield(Level(level), v.Levels[level].Slice()) {
+				return
+			}
+		}
 	}
 }