db: add format major version for range keys

Introduce a new format major version for range keys, with associated
table format `Pebble,v2`. When the DB is opened at this version, it is
free to make use of range key features.

Add various validations and assertions based on the new format major
version. For example, if the DB version does not support range keys,
ingesting tables with range keys will fail, etc.

Related to cockroachdb#1339.

batch.go

@@ -219,9 +219,6 @@ type Batch struct {
 
 	// The count of range key sets, unsets and deletes in the batch. Updated
 	// every time a RANGEKEYSET, RANGEKEYUNSET or RANGEKEYDEL key is added.
-	// TODO(jackson): This likely won't be necessary long-term, but it's useful
-	// for the in-memory only implementation in which these keys require special
-	// handling.
 	countRangeKeys uint64
 
 	// A deferredOp struct, stored in the Batch so that a pointer can be returned

db.go

@@ -723,11 +723,14 @@ func (d *DB) Apply(batch *Batch, opts *WriteOptions) error {
 		if d.opts.Experimental.RangeKeys == nil {
 			panic("pebble: range keys require the Experimental.RangeKeys option")
 		}
-		// TODO(jackson): Assert that all range key operands are suffixless.
+		if d.FormatMajorVersion() < FormatRangeKeys {
+			panic(fmt.Sprintf(
+				"pebble: range keys require at least format major version %d (current: %d)",
+				FormatRangeKeys, d.FormatMajorVersion(),
+			))
+		}
 
-		// TODO(jackson): Once the format major version for range keys is
-		// introduced, error if the batch includes range keys but the active
-		// format major version doesn't enable them.
+		// TODO(jackson): Assert that all range key operands are suffixless.
 	}
 
 	if batch.db == nil {
@@ -871,8 +874,16 @@ func (d *DB) newIterInternal(batch *Batch, s *Snapshot, o *IterOptions) *Iterato
 	if err := d.closed.Load(); err != nil {
 		panic(err)
 	}
-	if o.rangeKeys() && d.opts.Experimental.RangeKeys == nil {
-		panic("pebble: range keys require the Experimental.RangeKeys option")
+	if o.rangeKeys() {
+		if d.opts.Experimental.RangeKeys == nil {
+			panic("pebble: range keys require the Experimental.RangeKeys option")
+		}
+		if d.FormatMajorVersion() < FormatRangeKeys {
+			panic(fmt.Sprintf(
+				"pebble: range keys require at least format major version %d (current: %d)",
+				FormatRangeKeys, d.FormatMajorVersion(),
+			))
+		}
 	}
 	if o != nil && o.RangeKeyMasking.Suffix != nil && o.KeyTypes != IterKeyTypePointsAndRanges {
 		panic("pebble: range key masking requires IterKeyTypePointsAndRanges")

format_major_version.go

@@ -70,11 +70,13 @@ const (
 	// FormatBlockPropertyCollector is a format major version that introduces
 	// BlockPropertyCollectors.
 	FormatBlockPropertyCollector
+	// FormatRangeKeys is a format major version that introduces range keys.
+	FormatRangeKeys
 	// FormatNewest always contains the most recent format major version.
 	// NB: When adding new versions, the MaxTableFormat method should also be
 	// updated to return the maximum allowable version for the new
 	// FormatMajorVersion.
-	FormatNewest FormatMajorVersion = FormatBlockPropertyCollector
+	FormatNewest FormatMajorVersion = FormatRangeKeys
 )
 
 // MaxTableFormat returns the maximum sstable.TableFormat that can be used at
@@ -86,6 +88,8 @@ func (v FormatMajorVersion) MaxTableFormat() sstable.TableFormat {
 		return sstable.TableFormatRocksDBv2
 	case FormatBlockPropertyCollector:
 		return sstable.TableFormatPebblev1
+	case FormatRangeKeys:
+		return sstable.TableFormatPebblev2
 	default:
 		panic(fmt.Sprintf("pebble: unsupported format major version: %s", v))
 	}
@@ -167,6 +171,9 @@ var formatMajorVersionMigrations = map[FormatMajorVersion]func(*DB) error{
 	FormatBlockPropertyCollector: func(d *DB) error {
 		return d.finalizeFormatVersUpgrade(FormatBlockPropertyCollector)
 	},
+	FormatRangeKeys: func(d *DB) error {
+		return d.finalizeFormatVersUpgrade(FormatRangeKeys)
+	},
 }
 
 const formatVersionMarkerName = `format-version`

format_major_version_test.go

@@ -36,6 +36,8 @@ func TestRatchetFormat(t *testing.T) {
 	require.Equal(t, FormatSetWithDelete, d.FormatMajorVersion())
 	require.NoError(t, d.RatchetFormatMajorVersion(FormatBlockPropertyCollector))
 	require.Equal(t, FormatBlockPropertyCollector, d.FormatMajorVersion())
+	require.NoError(t, d.RatchetFormatMajorVersion(FormatRangeKeys))
+	require.Equal(t, FormatRangeKeys, d.FormatMajorVersion())
 	require.NoError(t, d.Close())
 
 	// If we Open the database again, leaving the default format, the
@@ -189,6 +191,7 @@ func TestFormatMajorVersions_TableFormat(t *testing.T) {
 		FormatVersioned:               sstable.TableFormatRocksDBv2,
 		FormatSetWithDelete:           sstable.TableFormatRocksDBv2,
 		FormatBlockPropertyCollector:  sstable.TableFormatPebblev1,
+		FormatRangeKeys:               sstable.TableFormatPebblev2,
 	}
 
 	// Valid versions.

ingest.go

@@ -45,7 +45,11 @@ func ingestValidateKey(opts *Options, key *InternalKey) error {
 }
 
 func ingestLoad1(
-	opts *Options, path string, cacheID uint64, fileNum FileNum,
+	opts *Options,
+	fmv FormatMajorVersion,
+	path string,
+	cacheID uint64,
+	fileNum FileNum,
 ) (*fileMetadata, error) {
 	stat, err := opts.FS.Stat(path)
 	if err != nil {
@@ -64,6 +68,18 @@ func ingestLoad1(
 	}
 	defer r.Close()
 
+	// Avoid ingesting tables with format versions this DB doesn't support.
+	tf, err := r.TableFormat()
+	if err != nil {
+		return nil, err
+	}
+	if tf > fmv.MaxTableFormat() {
+		return nil, errors.Newf(
+			"pebble: table with format %s unsupported at DB format major version %d, %s",
+			tf, fmv, fmv.MaxTableFormat(),
+		)
+	}
+
 	meta := &fileMetadata{}
 	meta.FileNum = fileNum
 	meta.Size = uint64(stat.Size())
@@ -154,12 +170,16 @@ func ingestLoad1(
 }
 
 func ingestLoad(
-	opts *Options, paths []string, cacheID uint64, pending []FileNum,
+	opts *Options,
+	fmv FormatMajorVersion,
+	paths []string,
+	cacheID uint64,
+	pending []FileNum,
 ) ([]*fileMetadata, []string, error) {
 	meta := make([]*fileMetadata, 0, len(paths))
 	newPaths := make([]string, 0, len(paths))
 	for i := range paths {
-		m, err := ingestLoad1(opts, paths[i], cacheID, pending[i])
+		m, err := ingestLoad1(opts, fmv, paths[i], cacheID, pending[i])
 		if err != nil {
 			return nil, nil, err
 		}
@@ -525,7 +545,7 @@ func (d *DB) Ingest(paths []string) error {
 
 	// Load the metadata for all of the files being ingested. This step detects
 	// and elides empty sstables.
-	meta, paths, err := ingestLoad(d.opts, paths, d.cacheID, pendingOutputs)
+	meta, paths, err := ingestLoad(d.opts, d.FormatMajorVersion(), paths, d.cacheID, pendingOutputs)
 	if err != nil {
 		return err
 	}

ingest_test.go

@@ -35,11 +35,28 @@ func TestIngestLoad(t *testing.T) {
 	datadriven.RunTest(t, "testdata/ingest_load", func(td *datadriven.TestData) string {
 		switch td.Cmd {
 		case "load":
+			writerOpts := sstable.WriterOptions{}
+			var dbVersion FormatMajorVersion
+			for _, cmdArgs := range td.CmdArgs {
+				v, err := strconv.Atoi(cmdArgs.Vals[0])
+				if err != nil {
+					return err.Error()
+				}
+				switch k := cmdArgs.Key; k {
+				case "writer-version":
+					fmv := FormatMajorVersion(v)
+					writerOpts.TableFormat = fmv.MaxTableFormat()
+				case "db-version":
+					dbVersion = FormatMajorVersion(v)
+				default:
+					return fmt.Sprintf("unknown cmd %s\n", k)
+				}
+			}
 			f, err := mem.Create("ext")
 			if err != nil {
 				return err.Error()
 			}
-			w := sstable.NewWriter(f, sstable.WriterOptions{})
+			w := sstable.NewWriter(f, writerOpts)
 			for _, data := range strings.Split(td.Input, "\n") {
 				j := strings.Index(data, ":")
 				if j < 0 {
@@ -57,7 +74,7 @@ func TestIngestLoad(t *testing.T) {
 				Comparer: DefaultComparer,
 				FS:       mem,
 			}
-			meta, _, err := ingestLoad(opts, []string{"ext"}, 0, []FileNum{1})
+			meta, _, err := ingestLoad(opts, dbVersion, []string{"ext"}, 0, []FileNum{1})
 			if err != nil {
 				return err.Error()
 			}
@@ -141,7 +158,7 @@ func TestIngestLoadRand(t *testing.T) {
 		Comparer: DefaultComparer,
 		FS:       mem,
 	}
-	meta, _, err := ingestLoad(opts, paths, 0, pending)
+	meta, _, err := ingestLoad(opts, FormatNewest, paths, 0, pending)
 	require.NoError(t, err)
 
 	for _, m := range meta {
@@ -162,7 +179,7 @@ func TestIngestLoadInvalid(t *testing.T) {
 		Comparer: DefaultComparer,
 		FS:       mem,
 	}
-	if _, _, err := ingestLoad(opts, []string{"invalid"}, 0, []FileNum{1}); err == nil {
+	if _, _, err := ingestLoad(opts, FormatNewest, []string{"invalid"}, 0, []FileNum{1}); err == nil {
 		t.Fatalf("expected error, but found success")
 	}
 }

open_test.go

@@ -101,7 +101,7 @@ func TestNewDBFilenames(t *testing.T) {
 			"LOCK",
 			"MANIFEST-000001",
 			"OPTIONS-000003",
-			"marker.format-version.000004.005",
+			"marker.format-version.000005.006",
 			"marker.manifest.000001.MANIFEST-000001",
 		},
 	}

range_keys_test.go

@@ -6,8 +6,10 @@ package pebble
 import (
 	"bytes"
 	"fmt"
+	"strconv"
 	"testing"
 
+	"github.com/cockroachdb/errors"
 	"github.com/cockroachdb/pebble/internal/datadriven"
 	"github.com/cockroachdb/pebble/internal/testkeys"
 	"github.com/cockroachdb/pebble/vfs"
@@ -35,10 +37,24 @@ func TestRangeKeys(t *testing.T) {
 				require.NoError(t, d.Close())
 			}
 			opts := &Options{
-				FS:       vfs.NewMem(),
-				Comparer: testkeys.Comparer,
+				FS:                 vfs.NewMem(),
+				Comparer:           testkeys.Comparer,
+				FormatMajorVersion: FormatRangeKeys,
 			}
 			opts.Experimental.RangeKeys = new(RangeKeysArena)
+
+			for _, cmdArgs := range td.CmdArgs {
+				if cmdArgs.Key != "format-major-version" {
+					return fmt.Sprintf("unknown command %s\n", cmdArgs.Key)
+				}
+				v, err := strconv.Atoi(cmdArgs.Vals[0])
+				if err != nil {
+					return err.Error()
+				}
+				// Override the DB version.
+				opts.FormatMajorVersion = FormatMajorVersion(v)
+			}
+
 			var err error
 			d, err = Open("", opts)
 			require.NoError(t, err)
@@ -52,8 +68,19 @@ func TestRangeKeys(t *testing.T) {
 		case "batch":
 			b := d.NewBatch()
 			require.NoError(t, runBatchDefineCmd(td, b))
+			var err error
+			func() {
+				defer func() {
+					if r := recover(); r != nil {
+						err = errors.New(r.(string))
+					}
+				}()
+				err = b.Commit(nil)
+			}()
+			if err != nil {
+				return err.Error()
+			}
 			count := b.Count()
-			require.NoError(t, b.Commit(nil))
 			return fmt.Sprintf("wrote %d keys\n", count)
 		case "indexed-batch":
 			b = d.NewIndexedBatch()
@@ -76,7 +103,19 @@ func TestRangeKeys(t *testing.T) {
 				}
 				o.RangeKeyMasking.Suffix = []byte(arg.Vals[0])
 			}
-			iter := newIter(o)
+			var iter *Iterator
+			var err error
+			func() {
+				defer func() {
+					if r := recover(); r != nil {
+						err = errors.New(r.(string))
+					}
+				}()
+				iter = newIter(o)
+			}()
+			if err != nil {
+				return err.Error()
+			}
 			return runIterCmd(td, iter, true /* close iter */)
 		case "rangekey-iter":
 			iter := newIter(&IterOptions{KeyTypes: IterKeyTypeRangesOnly})

sstable/reader.go

@@ -2066,6 +2066,7 @@ type Reader struct {
 	mergerOK          bool
 	checksumType      ChecksumType
 	tableFilter       *tableFilterReader
+	tableFormat       TableFormat
 	Properties        Properties
 }
 
@@ -2690,6 +2691,14 @@ func (r *Reader) EstimateDiskUsage(start, end []byte) (uint64, error) {
 	return endBH.Offset + endBH.Length + blockTrailerLen - startBH.Offset, nil
 }
 
+// TableFormat returns the format version for the table.
+func (r *Reader) TableFormat() (TableFormat, error) {
+	if r.err != nil {
+		return TableFormatUnspecified, r.err
+	}
+	return r.tableFormat, nil
+}
+
 // ReadableFile describes subset of vfs.File required for reading SSTs.
 type ReadableFile interface {
 	io.ReaderAt
@@ -2735,6 +2744,7 @@ func NewReader(f ReadableFile, o ReaderOptions, extraOpts ...ReaderOption) (*Rea
 		return nil, r.Close()
 	}
 	r.checksumType = footer.checksum
+	r.tableFormat = footer.format
 	// Read the metaindex.
 	if err := r.readMetaindex(footer.metaindexBH); err != nil {
 		r.err = err

sstable/reader_test.go

@@ -860,6 +860,35 @@ func TestValidateBlockChecksums(t *testing.T) {
 	}
 }
 
+func TestReader_TableFormat(t *testing.T) {
+	test := func(t *testing.T, want TableFormat) {
+		fs := vfs.NewMem()
+		f, err := fs.Create("test")
+		require.NoError(t, err)
+
+		opts := WriterOptions{TableFormat: want}
+		w := NewWriter(f, opts)
+		err = w.Close()
+		require.NoError(t, err)
+
+		f, err = fs.Open("test")
+		require.NoError(t, err)
+		r, err := NewReader(f, ReaderOptions{})
+		require.NoError(t, err)
+		defer r.Close()
+
+		got, err := r.TableFormat()
+		require.NoError(t, err)
+		require.Equal(t, want, got)
+	}
+
+	for tf := TableFormatLevelDB; tf <= TableFormatMax; tf++ {
+		t.Run(tf.String(), func(t *testing.T) {
+			test(t, tf)
+		})
+	}
+}
+
 func buildTestTable(
 	t *testing.T, numEntries uint64, blockSize, indexBlockSize int, compression Compression,
 ) *Reader {