iterv2: introduce a new iterator stack (off by default)

We introduce a new iterator abstraction along with a new merging
iterator, level iterator, interleaving iterator stack. This
abstraction allows presenting the next span boundary at any point,
which allows the merging iterator to (conceptually) divide the key
space into "slabs". The slab approach will make the addition of new
rangedel-type keys much easier. The boundary also naturally allows
"stopping points" in the level iterator without a special contract
with the merging iterator.

The new stack can be used by setting `iterv2.Enabled`. The stack
functions correctly (it passed a few hours of the metamorphic test).
It is still a prototype in that it does not yet have many of the
optimizations the existing stack has (like try-seek-using-next). The
integration of the v2 stack also needs some optimization in terms of
allocations.

Most of the code inside the iterators was written by me. Claude helped
with comments and with limited bits and pieces but it was generating
much more complicated code for the core logic. In all fairness, I also
had to go back and rewrite most of it a few times.

Suggestion on what to review:
 - `iterv2.Iter` semantics;
 - `iterv2.TestIter` implementation. While this is only used for
   testing, it serves as a formal codification of the semantics (with
   a fairly simple implementation). All the new iterators are tested
   against the `TestIter`, so it's less important to review the
   actual iterator implementations;
 - `iterv2.InterleavingIter` struct and `Init()` comments;
 - `levelIterV2` struct comment;
 - `mergingIterV2` struct comment;
 - `slabState` interface (`BuildForward`).

Author: RaduBerinde

checkpoint_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/cockroachdb/crlib/testutils/leaktest"
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/pebble/internal/base"
+	"github.com/cockroachdb/pebble/internal/iterv2"
 	"github.com/cockroachdb/pebble/internal/testutils"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/objstorage/remote"
@@ -281,6 +282,9 @@ func TestCopyCheckpointOptions(t *testing.T) {
 func TestCheckpoint(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	t.Run("shared=false", func(t *testing.T) {
+		if iterv2.Enabled {
+			t.Skipf("file open timing changes with iterv2")
+		}
 		testCheckpointImpl(t, "testdata/checkpoint", false /* createOnShared */)
 	})
 	t.Run("shared=true", func(t *testing.T) {

db.go

@@ -25,6 +25,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/inflight"
 	"github.com/cockroachdb/pebble/internal/invalidating"
 	"github.com/cockroachdb/pebble/internal/invariants"
+	"github.com/cockroachdb/pebble/internal/iterv2"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
 	"github.com/cockroachdb/pebble/internal/manifest"
@@ -1203,7 +1204,8 @@ func finishInitializingIter(ctx context.Context, buf *iterAlloc) *Iterator {
 		useLazyCombinedIteration := dbi.rangeKey == nil &&
 			dbi.opts.KeyTypes == IterKeyTypePointsAndRanges &&
 			(dbi.batch == nil || dbi.batch.batch.countRangeKeys == 0) &&
-			!dbi.opts.disableLazyCombinedIteration
+			!dbi.opts.disableLazyCombinedIteration &&
+			!iterv2.Enabled
 		if useLazyCombinedIteration {
 			// The user requested combined iteration, and there's no indexed
 			// batch currently containing range keys that would prevent lazy
@@ -1309,6 +1311,11 @@ func (i *Iterator) constructPointIter(
 		internalOpts.boundLimitedFilter = &i.rangeKeyMasking
 	}
 
+	if iterv2.Enabled {
+		i.constructPointIterV2(ctx, memtables, buf, internalOpts)
+		return
+	}
+
 	// Merging levels and levels from iterAlloc.
 	mlevels := buf.mlevels[:0]
 	levels := buf.levels[:0]
@@ -1429,6 +1436,102 @@ func (i *Iterator) constructPointIter(
 	i.merging = &buf.merging
 }
 
+// constructPointIterV2 builds the point iterator stack using mergingIterV2 and
+// levelIterV2/InterleavingIter children.
+func (i *Iterator) constructPointIterV2(
+	ctx context.Context, memtables flushableList, buf *iterAlloc, internalOpts internalIterOpts,
+) {
+	var v2iters []iterv2.Iter
+
+	// 1. Batch iterator (if any).
+	if i.batch != nil {
+		if i.batch.batch.index == nil {
+			panic(errors.AssertionFailedf("creating an iterator over an unindexed batch"))
+		}
+		i.batch.batch.initInternalIter(&i.opts, &i.batch.pointIter)
+		i.batch.batch.initRangeDelIter(&i.opts, &i.batch.rangeDelIter, i.batch.batchSeqNum)
+		var rangeDelIter keyspan.FragmentIterator
+		if i.batch.rangeDelIter.Count() > 0 {
+			rangeDelIter = &i.batch.rangeDelIter
+		}
+		iiter := new(iterv2.InterleavingIter)
+		iiter.Init(i.comparer,
+			&iterv2.PointKeyFilter{Iter: &i.batch.pointIter},
+			rangeDelIter,
+			nil, nil, // startKey, endKey: unbounded
+			i.opts.LowerBound, i.opts.UpperBound)
+		v2iters = append(v2iters, iiter)
+	}
+
+	if !i.batchOnlyIter {
+		// 2. Memtable iterators (newest to oldest).
+		for j := len(memtables) - 1; j >= 0; j-- {
+			mem := memtables[j]
+			if fi, ok := mem.flushable.(*ingestedFlushable); ok {
+				flushableLevelIter, flushableRangeDelIter := fi.newItersV2(&i.opts, internalOpts)
+				v2iters = append(v2iters, flushableLevelIter)
+				if flushableRangeDelIter != nil {
+					v2iters = append(v2iters, flushableRangeDelIter)
+				}
+				continue
+			}
+			pointIter := mem.newIter(&i.opts)
+			// Apparently the memtable iterator interleaves range key span boundaries.
+			// Filter them out using a PointKeyFilter.
+			pointIter = &iterv2.PointKeyFilter{Iter: pointIter}
+			rangeDelIter := mem.newRangeDelIter(&i.opts)
+			iiter := new(iterv2.InterleavingIter)
+			iiter.Init(i.comparer, pointIter, rangeDelIter,
+				nil, nil, // startKey, endKey: unbounded
+				i.opts.LowerBound, i.opts.UpperBound)
+			v2iters = append(v2iters, iiter)
+		}
+
+		// 3. Level iterators: L0 sublevels followed by L1+.
+		var current *manifest.Version
+		if i.version != nil {
+			current = i.version
+		} else {
+			current = i.readState.current
+		}
+		i.opts.snapshotForHideObsoletePoints = buf.dbi.seqNum
+
+		addLevelIterV2 := func(files manifest.LevelIterator, layer manifest.Layer) {
+			// Filter to point-key files only; range-key-only files are handled
+			// by the separate range key iterator.
+			files = files.Filter(manifest.KeyTypePoint)
+			li := newLevelIterV2(ctx, i.opts, i.comparer, i.newIters, files, layer, internalOpts)
+			v2iters = append(v2iters, li)
+		}
+
+		// L0 sublevels, newest to oldest.
+		for idx := len(current.L0SublevelFiles) - 1; idx >= 0; idx-- {
+			addLevelIterV2(current.L0SublevelFiles[idx].Iter(), manifest.L0Sublevel(idx))
+		}
+		// L1+.
+		for level := 1; level < len(current.Levels); level++ {
+			if current.Levels[level].Empty() {
+				continue
+			}
+			addLevelIterV2(current.Levels[level].Iter(), manifest.Level(level))
+		}
+	}
+	for j := range v2iters {
+		v2iters[j] = iterv2.MaybeWrapInInvalidating(v2iters[j])
+	}
+
+	m := newMergingIterV2(i.comparer.Compare, i.comparer.Split, i.seqNum, v2iters...)
+	m.lower = i.opts.LowerBound
+	m.upper = i.opts.UpperBound
+	m.logger = i.opts.getLogger()
+	m.stats = &i.stats.InternalStats
+	if i.batch != nil {
+		m.slab.batchSnapshot = i.batch.batchSeqNum
+	}
+	i.pointIter = invalidating.MaybeWrapIfInvariants(m).(topLevelIterator)
+	i.mergingV2 = m
+}
+
 // NewBatch returns a new empty write-only batch. Any reads on the batch will
 // return an error. If the batch is committed it will be applied to the DB.
 func (d *DB) NewBatch(opts ...BatchOption) *Batch {

flushable.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/invariants"
+	"github.com/cockroachdb/pebble/internal/iterv2"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/keyspan/keyspanimpl"
 	"github.com/cockroachdb/pebble/internal/manifest"
@@ -262,6 +263,42 @@ func (s *ingestedFlushable) newIterInternal(o *IterOptions, iio internalIterOpts
 	)
 }
 
+func (s *ingestedFlushable) newItersV2(
+	o *IterOptions, iio internalIterOpts,
+) (levelIter, rangeDelIter iterv2.Iter) {
+	var opts IterOptions
+	if o != nil {
+		opts = *o
+	}
+	files := s.slice.Iter()
+	levelIter = newLevelIterV2(
+		context.Background(),
+		opts,
+		s.comparer,
+		s.newIters,
+		files.Filter(manifest.KeyTypePoint),
+		manifest.FlushableIngestsLayer(),
+		iio,
+	)
+	if s.exciseSpan.Valid() {
+		// We have an excise span; we will set it up as a separate level.
+		rdel := keyspan.Span{
+			Start: s.exciseSpan.Start,
+			End:   s.exciseSpan.End,
+			Keys:  []keyspan.Key{{Trailer: base.MakeTrailer(s.exciseSeqNum, base.InternalKeyKindRangeDelete)}},
+		}
+		iiter := &iterv2.InterleavingIter{}
+		iiter.Init(
+			s.comparer,
+			base.NewFakeIterWithCmp(s.comparer.Compare, nil),
+			keyspan.NewIter(s.comparer.Compare, []keyspan.Span{rdel}),
+			nil, nil, nil, nil,
+		)
+		rangeDelIter = iiter
+	}
+	return levelIter, rangeDelIter
+}
+
 // newFlushIter is part of the flushable interface.
 func (s *ingestedFlushable) newFlushIter(*IterOptions) internalIterator {
 	// newFlushIter is only used for writing memtables to disk as sstables.

ingest_test.go

@@ -2213,9 +2213,6 @@ func TestIngest(t *testing.T) {
 	if runtime.GOARCH == "386" {
 		t.Skip("skipped on 32-bit due to slightly varied output")
 	}
-	defer func() {
-		require.NoError(t, d.Close())
-	}()
 
 	reset := func(split bool) {
 		if d != nil {
@@ -2330,6 +2327,9 @@ func TestIngest(t *testing.T) {
 			return fmt.Sprintf("unknown command: %s", td.Cmd)
 		}
 	})
+	if d != nil {
+		require.NoError(t, d.Close())
+	}
 }
 
 // linkAndRemovePredicate is like errorfs.InjectIndex, and injects only on the

internal/base/comparer.go

@@ -300,6 +300,16 @@ func (c *Comparer) EnsureDefaults() *Comparer {
 	return n
 }
 
+// Prefix returns the prefix of the given key.
+func (c *Comparer) Prefix(key []byte) []byte {
+	return c.Split.Prefix(key)
+}
+
+// HasPrefix returns true if the key's prefix equals the given prefix.
+func (c *Comparer) HasPrefix(key, prefix []byte) bool {
+	return bytes.Equal(c.Split.Prefix(key), prefix)
+}
+
 // DefaultComparer is the default implementation of the Comparer interface.
 // It uses the natural ordering, consistent with bytes.Compare.
 var DefaultComparer = &Comparer{

internal/base/internal.go

@@ -181,6 +181,12 @@ const (
 	// IDs (varint count + varint blob file IDs).
 	InternalKeyKindIngestSSTWithBlobs InternalKeyKind = 26
 
+	// InternalKeyKindSpanBoundary is a synthetic key generated by iterators to
+	// stop at span boundaries. See iterv2.Iter.
+	//
+	// It is important that this key is larger than any other valid point key.
+	InternalKeyKindSpanBoundary InternalKeyKind = 30
+
 	// This maximum value isn't part of the file format. Future extensions may
 	// increase this value.
 	//
@@ -190,7 +196,7 @@ const (
 	// which sorts 'less than or equal to' any other valid internalKeyKind, when
 	// searching for any kind of internal key formed by a certain user key and
 	// seqNum.
-	InternalKeyKindMax InternalKeyKind = 26
+	InternalKeyKindMax InternalKeyKind = 30
 
 	// InternalKeyKindMaxForSSTable is the largest valid key kind that can exist
 	// in an SSTable. This should usually equal InternalKeyKindMax, except
@@ -244,6 +250,7 @@ var internalKeyKindNames = []string{
 	InternalKeyKindExcise:             "EXCISE",
 	InternalKeyKindSyntheticKey:       "SYNTHETIC",
 	InternalKeyKindIngestSSTWithBlobs: "INGESTSSTWITHBLOB",
+	InternalKeyKindSpanBoundary:       "BDRY",
 	InternalKeyKindInvalid:            "INVALID",
 }
 
@@ -362,6 +369,7 @@ var kindsMap = map[string]InternalKeyKind{
 	"EXCISE":            InternalKeyKindExcise,
 	"SYNTHETIC":         InternalKeyKindSyntheticKey,
 	"INGESTSSTWITHBLOB": InternalKeyKindIngestSSTWithBlobs,
+	"BDRY":              InternalKeyKindSpanBoundary,
 }
 
 // ParseSeqNum parses the string representation of a sequence number.