sstable: include all filter blocks in Reader.Layout

Previously, (*sstable.Reader).Layout only included 1 filter block and only if
it matched the reader's configuration. This commit updates the Layout method to
collect the block handles of all filter blocks contained within the metaindex
block.

Author: jbowens

sstable/copier_test.go

@@ -30,8 +30,35 @@ func TestCopySpan(t *testing.T) {
 	defer cacheHandle.Close()
 	fileNameToNum := make(map[string]base.FileNum)
 	nextFileNum := base.FileNum(1)
-
 	keySchema := colblk.DefaultKeySchema(testkeys.Comparer, 16)
+
+	getReader := func(d *datadriven.TestData) (*Reader, error) {
+		f, err := fs.Open(d.CmdArgs[0].Key)
+		if err != nil {
+			return nil, err
+		}
+		readable, err := NewSimpleReadable(f)
+		if err != nil {
+			return nil, err
+		}
+
+		rOpts := ReaderOptions{
+			ReaderOptions: block.ReaderOptions{
+				CacheOpts: sstableinternal.CacheOptions{
+					CacheHandle: cacheHandle,
+					FileNum:     base.DiskFileNum(fileNameToNum[d.CmdArgs[0].Key]),
+				},
+			},
+			Comparer:   testkeys.Comparer,
+			KeySchemas: KeySchemas{keySchema.Name: &keySchema},
+		}
+		r, err := NewReader(context.TODO(), readable, rOpts)
+		if err != nil {
+			return nil, errors.CombineErrors(err, readable.Close())
+		}
+		return r, nil
+	}
+
 	datadriven.RunTest(t, "testdata/copy_span", func(t *testing.T, d *datadriven.TestData) string {
 		switch d.Cmd {
 		case "build":
@@ -70,14 +97,6 @@ func TestCopySpan(t *testing.T) {
 
 		case "iter":
 			// Iterate over the specified sstable
-			f, err := fs.Open(d.CmdArgs[0].Key)
-			if err != nil {
-				return err.Error()
-			}
-			readable, err := NewSimpleReadable(f)
-			if err != nil {
-				return err.Error()
-			}
 			var start, end []byte
 			for _, arg := range d.CmdArgs[1:] {
 				switch arg.Key {
@@ -87,22 +106,12 @@ func TestCopySpan(t *testing.T) {
 					end = []byte(arg.FirstVal(t))
 				}
 			}
-			rOpts := ReaderOptions{
-				ReaderOptions: block.ReaderOptions{
-					CacheOpts: sstableinternal.CacheOptions{
-						CacheHandle: cacheHandle,
-						FileNum:     base.DiskFileNum(fileNameToNum[d.CmdArgs[0].Key]),
-					},
-				},
-				Comparer:   testkeys.Comparer,
-				KeySchemas: KeySchemas{keySchema.Name: &keySchema},
-			}
 
-			r, err := NewReader(context.TODO(), readable, rOpts)
-			defer r.Close()
+			r, err := getReader(d)
 			if err != nil {
-				return errors.CombineErrors(err, readable.Close()).Error()
+				return err.Error()
 			}
+			defer r.Close()
 			iter, err := r.NewIter(block.NoTransforms, start, end, AssertNoBlobHandles)
 			if err != nil {
 				return err.Error()
@@ -132,14 +141,6 @@ func TestCopySpan(t *testing.T) {
 			fileNameToNum[outputFile] = nextFileNum
 			nextFileNum++
 
-			f, err := fs.Open(inputFile)
-			if err != nil {
-				t.Fatalf("failed to open sstable: %v", err)
-			}
-			readable, err := NewSimpleReadable(f)
-			if err != nil {
-				return err.Error()
-			}
 			rOpts := ReaderOptions{
 				ReaderOptions: block.ReaderOptions{
 					CacheOpts: sstableinternal.CacheOptions{
@@ -150,11 +151,12 @@ func TestCopySpan(t *testing.T) {
 				Comparer:   testkeys.Comparer,
 				KeySchemas: KeySchemas{keySchema.Name: &keySchema},
 			}
-			r, err := NewReader(context.TODO(), readable, rOpts)
+			r, err := getReader(d)
 			if err != nil {
-				return errors.CombineErrors(err, readable.Close()).Error()
+				return err.Error()
 			}
 			defer r.Close()
+
 			wOpts := WriterOptions{
 				Comparer:  testkeys.Comparer,
 				KeySchema: &keySchema,
@@ -175,27 +177,28 @@ func TestCopySpan(t *testing.T) {
 			return fmt.Sprintf("copied %d bytes", size)
 
 		case "describe":
-			f, err := fs.Open(d.CmdArgs[0].Key)
+			r, err := getReader(d)
 			if err != nil {
 				return err.Error()
 			}
-			readable, err := NewSimpleReadable(f)
+			defer r.Close()
+			l, err := r.Layout()
 			if err != nil {
 				return err.Error()
 			}
-			r, err := NewReader(context.TODO(), readable, ReaderOptions{
-				Comparer:   testkeys.Comparer,
-				KeySchemas: KeySchemas{keySchema.Name: &keySchema},
-			})
+			return l.Describe(false /* verbose */, r, nil)
+
+		case "props":
+			r, err := getReader(d)
 			if err != nil {
-				return errors.CombineErrors(err, readable.Close()).Error()
+				return err.Error()
 			}
 			defer r.Close()
-			l, err := r.Layout()
+			props, err := r.ReadPropertiesBlock(context.TODO(), nil)
 			if err != nil {
 				return err.Error()
 			}
-			return l.Describe(true, r, nil)
+			return props.String()
 
 		default:
 			t.Fatalf("unknown command: %s", d.Cmd)

sstable/layout.go

@@ -714,6 +714,8 @@ func forEachIndexEntry(
 	return indexIter.Close()
 }
 
+// decodeMetaindex decodes a row-based meta index block. The returned map owns
+// all its memory and can outlive the provided data slice.
 func decodeMetaindex(
 	data []byte,
 ) (meta map[string]block.Handle, vbih valblk.IndexHandle, err error) {
@@ -723,9 +725,11 @@ func decodeMetaindex(
 	}
 	defer func() { err = firstError(err, i.Close()) }()
 
+	var keysAlloc bytealloc.A
 	meta = map[string]block.Handle{}
 	for valid := i.First(); valid; valid = i.Next() {
 		value := i.Value()
+		var bh block.Handle
 		if bytes.Equal(i.Key().UserKey, []byte(metaValueIndexName)) {
 			var n int
 			vbih, n, err = valblk.DecodeIndexHandle(i.Value())
@@ -735,27 +739,36 @@ func decodeMetaindex(
 			if n == 0 || n != len(value) {
 				return nil, vbih, base.CorruptionErrorf("pebble/table: invalid table (bad value blocks index handle)")
 			}
+			bh = vbih.Handle
 		} else {
-			bh, n := block.DecodeHandle(value)
+			var n int
+			bh, n = block.DecodeHandle(value)
 			if n == 0 || n != len(value) {
 				return nil, vbih, base.CorruptionErrorf("pebble/table: invalid table (bad block handle)")
 			}
-			meta[string(i.Key().UserKey)] = bh
 		}
+		var key []byte
+		keysAlloc, key = keysAlloc.Copy(i.Key().UserKey)
+		keyStr := unsafe.String(unsafe.SliceData(key), len(key))
+		meta[keyStr] = bh
 	}
 	return meta, vbih, nil
 }
 
+// decodeColumnarMetaIndex decodes a columnar meta index block. The returned map
+// owns all its memory and can outlive the provided data slice.
 func decodeColumnarMetaIndex(
 	data []byte,
 ) (meta map[string]block.Handle, vbih valblk.IndexHandle, err error) {
 	var decoder colblk.KeyValueBlockDecoder
 	decoder.Init(data)
+	var keysAlloc bytealloc.A
 	meta = map[string]block.Handle{}
 	for i := 0; i < decoder.BlockDecoder().Rows(); i++ {
 		key := decoder.KeyAt(i)
 		value := decoder.ValueAt(i)
 
+		var bh block.Handle
 		if bytes.Equal(key, []byte(metaValueIndexName)) {
 			var n int
 			vbih, n, err = valblk.DecodeIndexHandle(value)
@@ -765,13 +778,18 @@ func decodeColumnarMetaIndex(
 			if n == 0 || n != len(value) {
 				return nil, vbih, base.CorruptionErrorf("pebble/table: invalid table (bad value blocks index handle)")
 			}
+			bh = vbih.Handle
 		} else {
-			bh, n := block.DecodeHandle(value)
+			var n int
+			bh, n = block.DecodeHandle(value)
 			if n == 0 || n != len(value) {
 				return nil, vbih, base.CorruptionErrorf("pebble/table: invalid table (bad block handle)")
 			}
-			meta[string(key)] = bh
 		}
+		var keyCopy []byte
+		keysAlloc, keyCopy = keysAlloc.Copy(key)
+		keyStr := unsafe.String(unsafe.SliceData(keyCopy), len(keyCopy))
+		meta[keyStr] = bh
 	}
 	return meta, vbih, nil
 }

sstable/reader.go

@@ -11,6 +11,7 @@ import (
 	"fmt"
 	"io"
 	"slices"
+	"strings"
 	"sync"
 
 	"github.com/cockroachdb/errors"
@@ -434,31 +435,41 @@ var metaBufferPools = sync.Pool{
 	},
 }
 
-func (r *Reader) readMetaindex(
-	ctx context.Context,
-	bufferPool *block.BufferPool,
-	readHandle objstorage.ReadHandle,
-	filters map[string]FilterPolicy,
-) error {
+func (r *Reader) readAndDecodeMetaindex(
+	ctx context.Context, bufferPool *block.BufferPool, readHandle objstorage.ReadHandle,
+) (map[string]block.Handle, valblk.IndexHandle, error) {
 	metaEnv := block.ReadEnv{BufferPool: bufferPool}
 	b, err := r.readMetaindexBlock(ctx, metaEnv, readHandle)
 	if err != nil {
-		return err
+		return nil, valblk.IndexHandle{}, err
 	}
 	data := b.BlockData()
 	defer b.Release()
 
 	if uint64(len(data)) != r.metaindexBH.Length {
-		return base.CorruptionErrorf("pebble/table: unexpected metaindex block size: %d vs %d",
+		return nil, valblk.IndexHandle{}, base.CorruptionErrorf("pebble/table: unexpected metaindex block size: %d vs %d",
 			errors.Safe(len(data)), errors.Safe(r.metaindexBH.Length))
 	}
 
 	var meta map[string]block.Handle
+	var valueBIH valblk.IndexHandle
 	if r.tableFormat >= TableFormatPebblev6 {
-		meta, r.valueBIH, err = decodeColumnarMetaIndex(data)
+		meta, valueBIH, err = decodeColumnarMetaIndex(data)
 	} else {
-		meta, r.valueBIH, err = decodeMetaindex(data)
+		meta, valueBIH, err = decodeMetaindex(data)
 	}
+	return meta, valueBIH, err
+}
+
+func (r *Reader) initMetaindexBlocks(
+	ctx context.Context,
+	bufferPool *block.BufferPool,
+	readHandle objstorage.ReadHandle,
+	filters map[string]FilterPolicy,
+) error {
+	var meta map[string]block.Handle
+	var err error
+	meta, r.valueBIH, err = r.readAndDecodeMetaindex(ctx, bufferPool, readHandle)
 	if err != nil {
 		return err
 	}
@@ -577,10 +588,21 @@ func (r *Reader) Layout() (*Layout, error) {
 		Format:             r.tableFormat,
 		BlobReferenceIndex: r.blobRefIndexBH,
 	}
-	if r.filterBH.Length > 0 {
-		l.Filter = []NamedBlockHandle{{Name: "fullfilter." + r.tableFilter.policy.Name(), Handle: r.filterBH}}
-	}
+
+	bufferPool := metaBufferPools.Get().(*block.BufferPool)
+	defer metaBufferPools.Put(bufferPool)
+	defer bufferPool.Release()
+
 	ctx := context.TODO()
+	meta, _, err := r.readAndDecodeMetaindex(ctx, bufferPool, noReadHandle)
+	if err != nil {
+		return nil, err
+	}
+	for name, bh := range meta {
+		if strings.HasPrefix(name, "fullfilter.") {
+			l.Filter = append(l.Filter, NamedBlockHandle{Name: name, Handle: bh})
+		}
+	}
 
 	indexH, err := r.readTopLevelIndexBlock(ctx, block.NoReadEnv, noReadHandle)
 	if err != nil {
@@ -950,7 +972,7 @@ func NewReader(ctx context.Context, f objstorage.Readable, o ReaderOptions) (*Re
 	// allocator.
 	defer bufferPool.Release()
 
-	if err := r.readMetaindex(ctx, bufferPool, rh, o.Filters); err != nil {
+	if err := r.initMetaindexBlocks(ctx, bufferPool, rh, o.Filters); err != nil {
 		r.err = err
 		return nil, err
 	}

sstable/test_utils.go

@@ -261,8 +261,27 @@ func ParseWriterOptions[StringOrStringer any](o *WriterOptions, args ...StringOr
 			o.IndexBlockSize, err = strconv.Atoi(value)
 
 		case "filter":
-			o.FilterPolicy = bloom.FilterPolicy(10)
-
+			fields := strings.FieldsFunc(value, func(r rune) bool {
+				return r == '(' || r == ')'
+			})
+			if len(fields) == 0 {
+				o.FilterPolicy = bloom.FilterPolicy(10)
+				continue
+			} else if len(fields) != 2 {
+				return errors.Errorf("expected filter policy name and parameters, got %q", value)
+			}
+			switch fields[0] {
+			case "bloom":
+				bits, err := strconv.Atoi(fields[1])
+				if err != nil {
+					return errors.Wrapf(err, "parsing bloom filter bits")
+				}
+				o.FilterPolicy = bloom.FilterPolicy(bits)
+			case "none":
+				o.FilterPolicy = nil
+			default:
+				return errors.Errorf("unknown filter policy: %q", value)
+			}
 		case "comparer":
 			o.Comparer, err = comparerFromCmdArg(value)