db: prohibit ingestion or external iteration of sstables with blob references

This commit adds validation to NewExternalIter and the ingestion pathways
prohibiting the external iteration or ingestion of sstables containing blob
references.

This could conceivably be relaxed in the future if we provide a mechanism to
provide the referenced blob files too, but for now this is an error condition.

Additionally the most recent experimental format major version that introduces
the TableFormatPebblev6 is renamed to capture the fact that it most
significantly will enable use of blob value separation (in addition to the
introduction of a checksum covering sstable footer data).

Author: jbowens

compaction_test.go

@@ -1225,8 +1225,8 @@ func TestManualCompaction(t *testing.T) {
 		},
 		{
 			testData:   "testdata/manual_compaction_set_with_del_sstable_Pebblev6",
-			minVersion: formatChecksumFooter,
-			maxVersion: formatChecksumFooter,
+			minVersion: formatTableFormatV6,
+			maxVersion: formatTableFormatV6,
 		},
 	}
 

data_test.go

@@ -18,7 +18,9 @@ import (
 	"strings"
 	"testing"
 	"time"
+	"unicode"
 
+	"github.com/cockroachdb/crlib/crstrings"
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/bloom"
@@ -28,6 +30,7 @@ import (
 	"github.com/cockroachdb/pebble/internal/rangekey"
 	"github.com/cockroachdb/pebble/internal/sstableinternal"
 	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/internal/testutils"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/objstorage/remote"
 	"github.com/cockroachdb/pebble/sstable"
@@ -428,26 +431,51 @@ func parseValue(s string) []byte {
 	return []byte(s)
 }
 
+func splitFields(line string, n int) ([]string, error) {
+	return splitFieldsRange(line, n, n)
+}
+
+func splitFieldsRange(line string, minmum, maximum int) ([]string, error) {
+	fields := strings.Fields(line)
+	if len(fields) < minmum {
+		return nil, errors.Errorf("require at least %d fields, got %d", minmum, len(fields))
+	}
+	if len(fields) > maximum {
+		fields[maximum-1] = strings.Join(fields[maximum-1:], " ")
+		fields = fields[:maximum]
+	}
+	for i := range fields {
+		if fields[i] == `<nil>` {
+			fields[i] = ""
+		}
+	}
+	return fields, nil
+}
+
 func runBatchDefineCmd(d *datadriven.TestData, b *Batch) error {
-	for _, line := range strings.Split(d.Input, "\n") {
-		parts := strings.Fields(line)
-		if len(parts) == 0 {
+	for _, line := range crstrings.Lines(d.Input) {
+		i := strings.IndexFunc(line, unicode.IsSpace)
+		cmd := line
+		if i > 0 {
+			cmd = line[:i]
+		} else if cmd == "" {
 			continue
 		}
-		if parts[1] == `<nil>` {
-			parts[1] = ""
-		}
+
+		var parts []string
 		var err error
-		switch parts[0] {
+		switch cmd {
 		case "set":
-			if len(parts) != 3 {
-				return errors.Errorf("%s expects 2 arguments", parts[0])
+			parts, err = splitFields(line, 3)
+			if err != nil {
+				return err
 			}
 			err = b.Set([]byte(parts[1]), parseValue(parts[2]), nil)
 
 		case "set-multiple":
-			if len(parts) != 3 {
-				return errors.Errorf("%s expects 2 arguments (n and prefix)", parts[0])
+			parts, err = splitFields(line, 3)
+			if err != nil {
+				return err
 			}
 			n, err := strconv.ParseUint(parts[1], 10, 32)
 			if err != nil {
@@ -462,13 +490,15 @@ func runBatchDefineCmd(d *datadriven.TestData, b *Batch) error {
 			}
 
 		case "del":
-			if len(parts) != 2 {
-				return errors.Errorf("%s expects 1 argument", parts[0])
+			parts, err = splitFields(line, 2)
+			if err != nil {
+				return err
 			}
 			err = b.Delete([]byte(parts[1]), nil)
 		case "del-sized":
-			if len(parts) != 3 {
-				return errors.Errorf("%s expects 2 arguments", parts[0])
+			parts, err = splitFields(line, 3)
+			if err != nil {
+				return err
 			}
 			var valSize uint64
 			valSize, err = strconv.ParseUint(parts[2], 10, 32)
@@ -477,23 +507,27 @@ func runBatchDefineCmd(d *datadriven.TestData, b *Batch) error {
 			}
 			err = b.DeleteSized([]byte(parts[1]), uint32(valSize), nil)
 		case "singledel":
-			if len(parts) != 2 {
-				return errors.Errorf("%s expects 1 argument", parts[0])
+			parts, err = splitFields(line, 2)
+			if err != nil {
+				return err
 			}
 			err = b.SingleDelete([]byte(parts[1]), nil)
 		case "del-range":
-			if len(parts) != 3 {
-				return errors.Errorf("%s expects 2 arguments", parts[0])
+			parts, err = splitFields(line, 3)
+			if err != nil {
+				return err
 			}
 			err = b.DeleteRange([]byte(parts[1]), []byte(parts[2]), nil)
 		case "merge":
-			if len(parts) != 3 {
-				return errors.Errorf("%s expects 2 arguments", parts[0])
+			parts, err = splitFields(line, 3)
+			if err != nil {
+				return err
 			}
 			err = b.Merge([]byte(parts[1]), parseValue(parts[2]), nil)
 		case "range-key-set":
-			if len(parts) < 4 || len(parts) > 5 {
-				return errors.Errorf("%s expects 3 or 4 arguments", parts[0])
+			parts, err = splitFieldsRange(line, 4, 5)
+			if err != nil {
+				return err
 			}
 			var val []byte
 			if len(parts) == 5 {
@@ -506,24 +540,26 @@ func runBatchDefineCmd(d *datadriven.TestData, b *Batch) error {
 				val,
 				nil)
 		case "range-key-unset":
-			if len(parts) != 4 {
-				return errors.Errorf("%s expects 3 arguments", parts[0])
+			parts, err = splitFields(line, 4)
+			if err != nil {
+				return err
 			}
 			err = b.RangeKeyUnset(
 				[]byte(parts[1]),
 				[]byte(parts[2]),
 				[]byte(parts[3]),
 				nil)
 		case "range-key-del":
-			if len(parts) != 3 {
-				return errors.Errorf("%s expects 2 arguments", parts[0])
+			parts, err = splitFields(line, 3)
+			if err != nil {
+				return err
 			}
 			err = b.RangeKeyDelete(
 				[]byte(parts[1]),
 				[]byte(parts[2]),
 				nil)
 		default:
-			return errors.Errorf("unknown op: %s", parts[0])
+			return errors.Errorf("unknown op: %s", cmd)
 		}
 		if err != nil {
 			return err
@@ -635,7 +671,27 @@ func runBuildRemoteCmd(td *datadriven.TestData, d *DB, storage remote.Storage) e
 	return w.Close()
 }
 
-func runBuildCmd(td *datadriven.TestData, d *DB, fs vfs.FS) error {
+type dataDrivenCmdOptions struct {
+	blobValues *testutils.BlobValues
+}
+
+func withBlobValues(bv *testutils.BlobValues) func(*dataDrivenCmdOptions) {
+	return func(o *dataDrivenCmdOptions) { o.blobValues = bv }
+}
+
+func combineDataDrivenOpts(opts ...func(*dataDrivenCmdOptions)) dataDrivenCmdOptions {
+	combined := dataDrivenCmdOptions{}
+	for _, opt := range opts {
+		opt(&combined)
+	}
+	return combined
+}
+
+func runBuildCmd(
+	td *datadriven.TestData, d *DB, fs vfs.FS, opts ...func(*dataDrivenCmdOptions),
+) error {
+	ddOpts := combineDataDrivenOpts(opts...)
+
 	b := newIndexedBatch(nil, d.opts.Comparer)
 	if err := runBatchDefineCmd(td, b); err != nil {
 		return err
@@ -660,14 +716,16 @@ func runBuildCmd(td *datadriven.TestData, d *DB, fs vfs.FS) error {
 				tableFormat = sstable.TableFormatPebblev3
 			case "pebblev4":
 				tableFormat = sstable.TableFormatPebblev4
+			case "pebblev5":
+				tableFormat = sstable.TableFormatPebblev5
 			default:
 				return errors.Errorf("unknown format string %s", cmdArg.Vals[0])
 			}
 		}
 	}
 
 	writeOpts := d.opts.MakeWriterOptions(0 /* level */, tableFormat)
-
+	var blobReferences testutils.BlobReferences
 	f, err := fs.Create(path, vfs.WriteCategoryUnspecified)
 	if err != nil {
 		return err
@@ -677,7 +735,22 @@ func runBuildCmd(td *datadriven.TestData, d *DB, fs vfs.FS) error {
 	for kv := iter.First(); kv != nil; kv = iter.Next() {
 		tmp := kv.K
 		tmp.SetSeqNum(0)
-		if err := w.Raw().Add(tmp, kv.InPlaceValue(), false); err != nil {
+
+		v := kv.InPlaceValue()
+		// If the value looks like it's a debug blob handle, parse it and add it
+		// to the sstable as a blob handle.
+		if ddOpts.blobValues != nil && ddOpts.blobValues.IsBlobHandle(string(v)) {
+			handle, err := ddOpts.blobValues.ParseInlineHandle(string(v), &blobReferences)
+			if err != nil {
+				return err
+			}
+			if err := w.Raw().AddWithBlobHandle(tmp, handle, base.ShortAttribute(0), false); err != nil {
+				return err
+			}
+			continue
+		}
+		// Otherwise add it as an ordinary value.
+		if err := w.Raw().Add(tmp, v, false); err != nil {
 			return err
 		}
 	}

external_iterator.go

@@ -26,7 +26,8 @@ import (
 // sorted in internal key order, where lower index files contain keys that sort
 // left of files with higher indexes.
 //
-// Input sstables must only contain keys with the zero sequence number.
+// Input sstables must only contain keys with the zero sequence number and must
+// not contain references to values in external blob files.
 //
 // Iterators constructed through NewExternalIter do not support all iterator
 // options, including block-property and table filters. NewExternalIter errors
@@ -302,6 +303,9 @@ func openExternalTables(
 		if err != nil {
 			return readers, err
 		}
+		if r.Properties.NumValuesInBlobFiles > 0 {
+			return readers, errors.Newf("pebble: NewExternalIter does not support blob references")
+		}
 		readers = append(readers, r)
 	}
 	return readers, err

external_iterator_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/cockroachdb/datadriven"
 	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/testkeys"
+	"github.com/cockroachdb/pebble/internal/testutils"
 	"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
 	"github.com/cockroachdb/pebble/sstable"
 	"github.com/cockroachdb/pebble/vfs"
@@ -33,39 +34,49 @@ func TestExternalIterator(t *testing.T) {
 	d, err := Open("", o)
 	require.NoError(t, err)
 	defer func() { require.NoError(t, d.Close()) }()
+	var bv testutils.BlobValues
+
+	getOptsAndFiles := func(td *datadriven.TestData) (opts IterOptions, files [][]sstable.ReadableFile) {
+		opts = IterOptions{KeyTypes: IterKeyTypePointsAndRanges}
+		for _, arg := range td.CmdArgs {
+			switch arg.Key {
+			case "mask-suffix":
+				opts.RangeKeyMasking.Suffix = []byte(arg.Vals[0])
+			case "lower":
+				opts.LowerBound = []byte(arg.Vals[0])
+			case "upper":
+				opts.UpperBound = []byte(arg.Vals[0])
+			case "files":
+				for _, v := range arg.Vals {
+					f, err := mem.Open(v)
+					require.NoError(t, err)
+					files = append(files, []sstable.ReadableFile{f})
+				}
+			}
+		}
+		return opts, files
+	}
 
 	datadriven.RunTest(t, "testdata/external_iterator", func(t *testing.T, td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "reset":
 			mem = vfs.NewMem()
 			return ""
 		case "build":
-			if err := runBuildCmd(td, d, mem); err != nil {
+			if err := runBuildCmd(td, d, mem, withBlobValues(&bv)); err != nil {
 				return err.Error()
 			}
 			return ""
-		case "iter":
-			opts := IterOptions{KeyTypes: IterKeyTypePointsAndRanges}
-			var files [][]sstable.ReadableFile
-			for _, arg := range td.CmdArgs {
-				switch arg.Key {
-				case "mask-suffix":
-					opts.RangeKeyMasking.Suffix = []byte(arg.Vals[0])
-				case "lower":
-					opts.LowerBound = []byte(arg.Vals[0])
-				case "upper":
-					opts.UpperBound = []byte(arg.Vals[0])
-				case "files":
-					for _, v := range arg.Vals {
-						f, err := mem.Open(v)
-						require.NoError(t, err)
-						files = append(files, []sstable.ReadableFile{f})
-					}
-				}
-			}
+		case "iter-init-error":
+			opts, files := getOptsAndFiles(td)
 			testExternalIteratorInitError(t, o, &opts, files)
+			return ""
+		case "iter":
+			opts, files := getOptsAndFiles(td)
 			it, err := NewExternalIter(o, &opts, files)
-			require.NoError(t, err)
+			if err != nil {
+				return fmt.Sprintf("error: %s", err.Error())
+			}
 			return runIterCmd(td, it, true /* close iter */)
 		default:
 			return fmt.Sprintf("unknown command: %s", td.Cmd)

format_major_version.go

@@ -214,10 +214,15 @@ const (
 
 	// -- Add experimental versions here --
 
-	// formatChecksumFooter is a format major version enabling use of the
-	// TableFormatPebblev6 table format. It is a format allowing for the checksum
-	// of sstable footers.
-	formatChecksumFooter
+	// formatTableFormatV6 is a format major version enabling the sstable table
+	// format TableFormatPebblev6.
+	//
+	// The TableFormatPebblev6 sstable format introduces a checksum within the
+	// sstable footer, and allows inclusion of blob handle references within the
+	// value column of a sstable block.
+	//
+	// This format major version does not yet enable use of value separation.
+	formatTableFormatV6
 
 	// internalFormatNewest is the most recent, possibly experimental format major
 	// version.
@@ -249,7 +254,7 @@ func (v FormatMajorVersion) MaxTableFormat() sstable.TableFormat {
 		return sstable.TableFormatPebblev4
 	case FormatColumnarBlocks, FormatWALSyncChunks:
 		return sstable.TableFormatPebblev5
-	case formatChecksumFooter:
+	case formatTableFormatV6:
 		return sstable.TableFormatPebblev6
 	default:
 		panic(fmt.Sprintf("pebble: unsupported format major version: %s", v))
@@ -263,7 +268,7 @@ func (v FormatMajorVersion) MinTableFormat() sstable.TableFormat {
 	case FormatDefault, FormatFlushableIngest, FormatPrePebblev1MarkedCompacted,
 		FormatDeleteSizedAndObsolete, FormatVirtualSSTables, FormatSyntheticPrefixSuffix,
 		FormatFlushableIngestExcises, FormatColumnarBlocks, FormatWALSyncChunks,
-		formatChecksumFooter:
+		formatTableFormatV6:
 		return sstable.TableFormatPebblev1
 	default:
 		panic(fmt.Sprintf("pebble: unsupported format major version: %s", v))
@@ -309,8 +314,8 @@ var formatMajorVersionMigrations = map[FormatMajorVersion]func(*DB) error{
 	FormatWALSyncChunks: func(d *DB) error {
 		return d.finalizeFormatVersUpgrade(FormatWALSyncChunks)
 	},
-	formatChecksumFooter: func(d *DB) error {
-		return d.finalizeFormatVersUpgrade(formatChecksumFooter)
+	formatTableFormatV6: func(d *DB) error {
+		return d.finalizeFormatVersUpgrade(formatTableFormatV6)
 	},
 }