sstable: handle synthetic prefix when checking bloom filter

Fix the sstable iterators to strip the synthetic prefix before
checking the bloom filter.

Fix the `TestRandomizedPrefixSuffixRewriter` to check `SeekPrefixGE`
and randomly enable filters.

Author: RaduBerinde

options.go

@@ -788,7 +788,8 @@ type Options struct {
 	// Filters is a map from filter policy name to filter policy. It is used for
 	// debugging tools which may be used on multiple databases configured with
 	// different filter policies. It is not necessary to populate this filters
-	// map during normal usage of a DB.
+	// map during normal usage of a DB (it will be done automatically by
+	// EnsureDefaults).
 	Filters map[string]FilterPolicy
 
 	// FlushDelayDeleteRange configures how long the database should wait before

sstable/options.go

@@ -98,10 +98,8 @@ type ReaderOptions struct {
 	// Merge defines the Merge function in use for this keyspace.
 	Merge base.Merge
 
-	// Filters is a map from filter policy name to filter policy. It is used for
-	// debugging tools which may be used on multiple databases configured with
-	// different filter policies. It is not necessary to populate this filters
-	// map during normal usage of a DB.
+	// Filters is a map from filter policy name to filter policy. Filters with
+	// policies that are not in this map will be ignored.
 	Filters map[string]FilterPolicy
 
 	// Merger defines the associative merge operation to use for merging values

sstable/reader_iter_single_lvl.go

@@ -786,19 +786,13 @@ func (i *singleLevelIterator) seekPrefixGE(
 		}
 		i.lastBloomFilterMatched = false
 		// Check prefix bloom filter.
-		var dataH bufferHandle
-		dataH, i.err = i.reader.readFilter(i.ctx, i.indexFilterRH, i.stats, &i.iterStats)
-		if i.err != nil {
-			i.data.invalidate()
-			return nil
-		}
-		mayContain := i.reader.tableFilter.mayContain(dataH.Get(), prefix)
-		dataH.Release()
-		if !mayContain {
-			// This invalidation may not be necessary for correctness, and may
-			// be a place to optimize later by reusing the already loaded
-			// block. It was necessary in earlier versions of the code since
-			// the caller was allowed to call Next when SeekPrefixGE returned
+		var mayContain bool
+		mayContain, i.err = i.bloomFilterMayContain(prefix)
+		if i.err != nil || !mayContain {
+			// In the i.err == nil case, this invalidation may not be necessary for
+			// correctness, and may be a place to optimize later by reusing the
+			// already loaded block. It was necessary in earlier versions of the code
+			// since the caller was allowed to call Next when SeekPrefixGE returned
 			// nil. This is no longer allowed.
 			i.data.invalidate()
 			return nil
@@ -833,6 +827,27 @@ func (i *singleLevelIterator) seekPrefixGE(
 	return i.maybeVerifyKey(i.seekGEHelper(key, boundsCmp, flags))
 }
 
+func (i *singleLevelIterator) bloomFilterMayContain(prefix []byte) (bool, error) {
+	// Check prefix bloom filter.
+	prefixToCheck := prefix
+	if i.transforms.SyntheticPrefix.IsSet() {
+		// We have to remove the synthetic prefix.
+		var ok bool
+		prefixToCheck, ok = bytes.CutPrefix(prefix, i.transforms.SyntheticPrefix)
+		if !ok {
+			// This prefix will not be found inside this table.
+			return false, nil
+		}
+	}
+
+	dataH, err := i.reader.readFilter(i.ctx, i.indexFilterRH, i.stats, &i.iterStats)
+	if err != nil {
+		return false, err
+	}
+	defer dataH.Release()
+	return i.reader.tableFilter.mayContain(dataH.Get(), prefixToCheck), nil
+}
+
 // virtualLast should only be called if i.vReader != nil.
 func (i *singleLevelIterator) virtualLast() *base.InternalKV {
 	if i.vState == nil {

sstable/reader_iter_two_lvl.go

@@ -390,19 +390,13 @@ func (i *twoLevelIterator) SeekPrefixGE(
 			flags = flags.DisableTrySeekUsingNext()
 		}
 		i.lastBloomFilterMatched = false
-		var dataH bufferHandle
-		dataH, i.err = i.reader.readFilter(i.ctx, i.indexFilterRH, i.stats, &i.iterStats)
-		if i.err != nil {
-			i.data.invalidate()
-			return nil
-		}
-		mayContain := i.reader.tableFilter.mayContain(dataH.Get(), prefix)
-		dataH.Release()
-		if !mayContain {
-			// This invalidation may not be necessary for correctness, and may
-			// be a place to optimize later by reusing the already loaded
-			// block. It was necessary in earlier versions of the code since
-			// the caller was allowed to call Next when SeekPrefixGE returned
+		var mayContain bool
+		mayContain, i.err = i.bloomFilterMayContain(prefix)
+		if i.err != nil || !mayContain {
+			// In the i.err == nil case, this invalidation may not be necessary for
+			// correctness, and may be a place to optimize later by reusing the
+			// already loaded block. It was necessary in earlier versions of the code
+			// since the caller was allowed to call Next when SeekPrefixGE returned
 			// nil. This is no longer allowed.
 			i.data.invalidate()
 			return nil

sstable/reader_test.go

@@ -1029,6 +1029,7 @@ type readCallType int
 
 const (
 	SeekGE readCallType = iota
+	SeekPrefixGE
 	Next
 	Prev
 	SeekLT
@@ -1040,6 +1041,8 @@ func (rct readCallType) String() string {
 	switch rct {
 	case SeekGE:
 		return "SeekGE"
+	case SeekPrefixGE:
+		return "SeekPrefixGE"
 	case Next:
 		return "Next"
 	case Prev:
@@ -1057,7 +1060,7 @@ func (rct readCallType) String() string {
 
 func (rct readCallType) Opposite() readCallType {
 	switch rct {
-	case SeekGE:
+	case SeekGE, SeekPrefixGE:
 		return SeekLT
 	case Next:
 		return Prev
@@ -1136,13 +1139,18 @@ func (rw *readerWorkload) setCallAfterInvalid() {
 }
 
 func (rw *readerWorkload) handleInvalid(callType readCallType, iter Iterator) *base.InternalKV {
-	switch {
-	case (SeekGE == callType || Next == callType || Last == callType):
+	switch callType {
+	case SeekGE, Next, Last:
 		if len(rw.seekKeyAfterInvalid) == 0 {
 			return iter.Prev()
 		}
 		return iter.SeekLT(rw.seekKeyAfterInvalid, base.SeekLTFlagsNone)
-	case (SeekLT == callType || Prev == callType || First == callType):
+	case SeekPrefixGE:
+		if len(rw.seekKeyAfterInvalid) == 0 {
+			return iter.First()
+		}
+		return iter.SeekLT(rw.seekKeyAfterInvalid, base.SeekLTFlagsNone)
+	case SeekLT, Prev, First:
 		if len(rw.seekKeyAfterInvalid) == 0 {
 			return iter.Next()
 		}
@@ -1157,6 +1165,17 @@ func (rw *readerWorkload) read(call readCall, iter Iterator) *base.InternalKV {
 	switch call.callType {
 	case SeekGE:
 		return iter.SeekGE(call.seekKey, base.SeekGEFlagsNone)
+	case SeekPrefixGE:
+		cmp := testkeys.Comparer
+		prefix := call.seekKey[:cmp.Split(call.seekKey)]
+		kv := iter.SeekPrefixGE(prefix, call.seekKey, base.SeekGEFlagsNone)
+		// If there is no key with this prefix to return, SeekPrefixGE might return
+		// nil or it might return a larger key. Make it always return nil so we can
+		// cross-check results.
+		if kv != nil && !cmp.Equal(prefix, kv.K.UserKey[:cmp.Split(kv.K.UserKey)]) {
+			return nil
+		}
+		return kv
 	case Next:
 		return rw.repeatRead(call, iter)
 	case SeekLT:
@@ -1207,8 +1226,7 @@ func createReadWorkload(
 			// Sqrt the likelihood of calling First and Last as they're not very interesting.
 			callType = readCallType(rng.Intn(int(Last + 1)))
 		}
-		if callType == SeekLT || callType == SeekGE {
-
+		if callType == SeekLT || callType == SeekGE || callType == SeekPrefixGE {
 			idx := rng.Int63n(int64(ks.MaxLen()))
 			ts := rng.Int63n(maxTS) + 1
 			key := testkeys.KeyAt(ks, idx, ts)
@@ -1239,11 +1257,10 @@ func createReadWorkload(
 // the same sequence of keys as another iterator initialized with a suffix
 // replacement rule to that fixed timestamp which reads an sst with the same
 // keys and randomized suffixes. The iterator with the suffix replacement rule
-// may also be initialized with a prefix synthenthsis rule while the control file
+// may also be initialized with a prefix synthesis rule while the control file
 // will contain all keys with that prefix. In other words, this is a randomized
 // version of TestBlockSyntheticSuffix and TestBlockSyntheticPrefix.
 func TestRandomizedPrefixSuffixRewriter(t *testing.T) {
-
 	ks := testkeys.Alpha(3)
 
 	callCount := 500
@@ -1252,29 +1269,34 @@ func TestRandomizedPrefixSuffixRewriter(t *testing.T) {
 	syntheticSuffix := []byte("@" + strconv.Itoa(int(suffix)))
 	var syntheticPrefix []byte
 
-	potentialBlockSize := []int{32, 64, 128, 256}
-	potentialRestartInterval := []int{1, 4, 8, 16}
+	blockSizeCandidates := []int{32, 64, 128, 256}
+	restartIntervalCandidates := []int{1, 4, 8, 16}
+	filterPolicyCandidates := []FilterPolicy{nil, bloom.FilterPolicy(1), bloom.FilterPolicy(10)}
 
 	seed := uint64(time.Now().UnixNano())
 	rng := rand.New(rand.NewSource(seed))
 	mem := vfs.NewMem()
 
-	blockSize := potentialBlockSize[rng.Intn(len(potentialBlockSize))]
-	restartInterval := potentialRestartInterval[rng.Intn(len(potentialRestartInterval))]
-	if rng.Intn(1) == 0 {
+	blockSize := blockSizeCandidates[rng.Intn(len(blockSizeCandidates))]
+	restartInterval := restartIntervalCandidates[rng.Intn(len(restartIntervalCandidates))]
+	filterPolicy := filterPolicyCandidates[rng.Intn(len(filterPolicyCandidates))]
+	if rng.Intn(10) < 9 {
 		randKey := testkeys.Key(ks, rng.Int63n(ks.Count()))
 		// Choose from 3 prefix candidates: "_" sorts before all keys, randKey sorts
 		// somewhere between all keys, and "~" sorts after all keys
 		prefixCandidates := []string{"~", string(randKey) + "_", "_"}
 		syntheticPrefix = []byte(prefixCandidates[rng.Intn(len(prefixCandidates))])
-
 	}
-	t.Logf("Configured Block Size %d, Restart Interval %d, Seed %d, Prefix %s", blockSize, restartInterval, seed, string(syntheticPrefix))
+	t.Logf("Configured Block Size %d, Restart Interval %d, Seed %d, Prefix %s, Filter policy %v", blockSize, restartInterval, seed, string(syntheticPrefix), filterPolicy)
 
 	createIter := func(fileName string, syntheticSuffix SyntheticSuffix, syntheticPrefix SyntheticPrefix) (Iterator, func()) {
 		f, err := mem.Open(fileName)
 		require.NoError(t, err)
-		eReader, err := newReader(f, ReaderOptions{Comparer: testkeys.Comparer})
+		opts := ReaderOptions{Comparer: testkeys.Comparer}
+		if filterPolicy != nil {
+			opts.Filters = map[string]FilterPolicy{filterPolicy.Name(): filterPolicy}
+		}
+		eReader, err := newReader(f, opts)
 		require.NoError(t, err)
 		iter, err := eReader.newIterWithBlockPropertyFiltersAndContext(
 			context.Background(),
@@ -1299,7 +1321,7 @@ func TestRandomizedPrefixSuffixRewriter(t *testing.T) {
 			testCaseName = "two-level"
 		}
 		t.Run(testCaseName, func(t *testing.T) {
-			indexBlockSize := 4096
+			indexBlockSize := 10 * 1024 * 1024
 			if twoLevelIndex {
 				indexBlockSize = 1
 			}
@@ -1321,6 +1343,7 @@ func TestRandomizedPrefixSuffixRewriter(t *testing.T) {
 					BlockSize:            blockSize,
 					IndexBlockSize:       indexBlockSize,
 					Comparer:             testkeys.Comparer,
+					FilterPolicy:         filterPolicy,
 				})
 
 				keyIdx := int64(0)
@@ -1366,7 +1389,6 @@ func TestRandomizedPrefixSuffixRewriter(t *testing.T) {
 				alsoCheck: func() {
 					require.Nil(t, eIter.Error())
 					require.Nil(t, iter.Error())
-
 				},
 			}
 			for _, call := range w.calls {