db: add SkipPoint iterator option

Add a new IterOption that allows a user to configure an arbitrary predicate of
point keys that should be skipped in the form of a function. The Iterator
consults this predicate for every internal key.

Use this new SkipPoint function to remove the complicated logic within the
metamorphic test for ensuring determinisim of RangeKeyChanged() in the presence
of time-bound block-property filters. A SkipPoint predicate may be combined
with a block-property filter to achieve deterministic iteration.

Author: jbowens

iterator.go

@@ -550,6 +550,21 @@ func (i *Iterator) findNextEntry(limit []byte) {
 			return
 		}
 
+		// If the user has configured a SkipPoint function, invoke it to see
+		// whether we should skip over the current user key.
+		if i.opts.SkipPoint != nil && key.Kind() != InternalKeyKindRangeKeySet && i.opts.SkipPoint(i.iterKey.UserKey) {
+			// NB: We could call nextUserKey, but in some cases the SkipPoint
+			// predicate function might be cheaper than nextUserKey's key copy
+			// and key comparison. This should be the case for MVCC suffix
+			// comparisons, for example. In the future, we could expand the
+			// SkipPoint interface to give the implementor more control over
+			// whether we skip over just the internal key, the user key, or even
+			// the key prefix.
+			i.stats.ForwardStepCount[InternalIterCall]++
+			i.iterKey, i.iterValue = i.iter.Next()
+			continue
+		}
+
 		switch key.Kind() {
 		case InternalKeyKindRangeKeySet:
 			// Save the current key.
@@ -619,6 +634,13 @@ func (i *Iterator) findNextEntry(limit []byte) {
 }
 
 func (i *Iterator) nextPointCurrentUserKey() bool {
+	// If the user has configured a SkipPoint function and the current user key
+	// would be skipped by it, there's no need to step forward looking for a
+	// point key. If we were to find one, it should be skipped anyways.
+	if i.opts.SkipPoint != nil && i.opts.SkipPoint(i.key) {
+		return false
+	}
+
 	i.pos = iterPosCurForward
 
 	i.iterKey, i.iterValue = i.iter.Next()
@@ -911,6 +933,26 @@ func (i *Iterator) findPrevEntry(limit []byte) {
 			}
 		}
 
+		// If the user has configured a SkipPoint function, invoke it to see
+		// whether we should skip over the current user key.
+		if i.opts.SkipPoint != nil && key.Kind() != InternalKeyKindRangeKeySet && i.opts.SkipPoint(key.UserKey) {
+			// NB: We could call prevUserKey, but in some cases the SkipPoint
+			// predicate function might be cheaper than prevUserKey's key copy
+			// and key comparison. This should be the case for MVCC suffix
+			// comparisons, for example. In the future, we could expand the
+			// SkipPoint interface to give the implementor more control over
+			// whether we skip over just the internal key, the user key, or even
+			// the key prefix.
+			i.stats.ReverseStepCount[InternalIterCall]++
+			i.iterKey, i.iterValue = i.iter.Prev()
+			if limit != nil && i.iterKey != nil && i.cmp(limit, i.iterKey.UserKey) > 0 && !i.rangeKeyWithinLimit(limit) {
+				i.iterValidityState = IterAtLimit
+				i.pos = iterPosCurReversePaused
+				return
+			}
+			continue
+		}
+
 		switch key.Kind() {
 		case InternalKeyKindRangeKeySet:
 			// Range key start boundary markers are interleaved with the maximum
@@ -948,12 +990,12 @@ func (i *Iterator) findPrevEntry(limit []byte) {
 			// Compare with the limit. We could optimize by only checking when
 			// we step to the previous user key, but detecting that requires a
 			// comparison too. Note that this position may already passed a
-			// number of versions of this user key, but they are all deleted,
-			// so the fact that a subsequent Prev*() call will not see them is
+			// number of versions of this user key, but they are all deleted, so
+			// the fact that a subsequent Prev*() call will not see them is
 			// harmless. Also note that this is the only place in the loop,
-			// other than the firstLoopIter case above, where we could step
-			// to a different user key and start processing it for returning
-			// to the caller.
+			// other than the firstLoopIter and SkipPoint cases above, where we
+			// could step to a different user key and start processing it for
+			// returning to the caller.
 			if limit != nil && i.iterKey != nil && i.cmp(limit, i.iterKey.UserKey) > 0 && !i.rangeKeyWithinLimit(limit) {
 				i.iterValidityState = IterAtLimit
 				i.pos = iterPosCurReversePaused
@@ -2428,7 +2470,8 @@ func (i *Iterator) SetOptions(o *IterOptions) {
 	// If either options specify block property filters for an iterator stack,
 	// reconstruct it.
 	if i.pointIter != nil && (closeBoth || len(o.PointKeyFilters) > 0 || len(i.opts.PointKeyFilters) > 0 ||
-		o.RangeKeyMasking.Filter != nil || i.opts.RangeKeyMasking.Filter != nil) {
+		o.RangeKeyMasking.Filter != nil || i.opts.RangeKeyMasking.Filter != nil || o.SkipPoint != nil ||
+		i.opts.SkipPoint != nil) {
 		i.err = firstError(i.err, i.pointIter.Close())
 		i.pointIter = nil
 	}

iterator_histories_test.go

@@ -309,6 +309,17 @@ func TestIterHistories(t *testing.T) {
 						o.PointKeyFilters = []sstable.BlockPropertyFilter{
 							sstable.NewTestKeysBlockPropertyFilter(min, max),
 						}
+						o.SkipPoint = func(k []byte) bool {
+							i := testkeys.Comparer.Split(k)
+							if i == len(k) {
+								return false
+							}
+							v, err := testkeys.ParseSuffix(k[i:])
+							if err != nil {
+								return false
+							}
+							return uint64(v) < min || uint64(v) >= max
+						}
 					case "snapshot":
 						s, err := strconv.ParseUint(arg.Vals[0], 10, 64)
 						if err != nil {

metamorphic/ops.go

@@ -18,6 +18,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/base"
 	"github.com/cockroachdb/pebble/internal/keyspan"
 	"github.com/cockroachdb/pebble/internal/private"
+	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
@@ -711,7 +712,7 @@ func (o *newIterOp) run(t *test, h historyRecorder) {
 		// close this iter and retry NewIter
 		_ = i.Close()
 	}
-	t.setIter(o.iterID, i, o.filterMin, o.filterMax)
+	t.setIter(o.iterID, i)
 
 	// Trash the bounds to ensure that Pebble doesn't rely on the stability of
 	// the user-provided bounds.
@@ -763,13 +764,7 @@ func (o *newIterUsingCloneOp) run(t *test, h historyRecorder) {
 	if err != nil {
 		panic(err)
 	}
-	filterMin, filterMax := o.filterMin, o.filterMax
-	if cloneOpts.IterOptions == nil {
-		// We're adopting the same block property filters as iter, so we need to
-		// adopt the same run-time filters to ensure determinism.
-		filterMin, filterMax = iter.filterMin, iter.filterMax
-	}
-	t.setIter(o.iterID, i, filterMin, filterMax)
+	t.setIter(o.iterID, i)
 	h.Recordf("%s // %v", o, i.Error())
 }
 
@@ -853,9 +848,6 @@ func (o *iterSetOptionsOp) run(t *test, h historyRecorder) {
 	rand.Read(opts.LowerBound[:])
 	rand.Read(opts.UpperBound[:])
 
-	// Adjust the iterator's filters.
-	i.filterMin, i.filterMax = o.filterMin, o.filterMax
-
 	h.Recordf("%s // %v", o, i.Error())
 }
 
@@ -893,6 +885,22 @@ func iterOptions(o iterOpts) *pebble.IterOptions {
 		opts.PointKeyFilters = []pebble.BlockPropertyFilter{
 			sstable.NewTestKeysBlockPropertyFilter(o.filterMin, o.filterMax),
 		}
+		// Enforce the timestamp bounds in SkipPoint, so that the iterator never
+		// returns a key outside the filterMin, filterMax bounds. This provides
+		// deterministic iteration.
+		opts.SkipPoint = func(k []byte) (skip bool) {
+			n := testkeys.Comparer.Split(k)
+			if n == len(k) {
+				// No suffix, don't skip it.
+				return false
+			}
+			v, err := testkeys.ParseSuffix(k[n:])
+			if err != nil {
+				panic(err)
+			}
+			ts := uint64(v)
+			return ts < o.filterMin || ts >= o.filterMax
+		}
 	}
 	return opts
 }