sstable: add back support for older formats

Add back support for older sstable formats. These are still not
supported by any non-deprecated Pebble format, but they can be used if
the `sstable` package is used directly. CockroachDB uses it to read
data from backups, which could be old.

This will fix a bunch of "bad magic number" test failures which use
older backup fixtures.

sstable/data_test.go

@@ -27,6 +27,11 @@ import (
 func optsFromArgs(td *datadriven.TestData, writerOpts *WriterOptions) error {
 	for _, arg := range td.CmdArgs {
 		switch arg.Key {
+		case "leveldb":
+			if len(arg.Vals) != 0 {
+				return errors.Errorf("%s: arg %s expects 0 values", td.Cmd, arg.Key)
+			}
+			writerOpts.TableFormat = TableFormatLevelDB
 		case "block-size":
 			if len(arg.Vals) != 1 {
 				return errors.Errorf("%s: arg %s expects 1 value", td.Cmd, arg.Key)

sstable/format.go

@@ -19,15 +19,19 @@ type TableFormat uint32
 // Pebble (i.e. the history is linear).
 const (
 	TableFormatUnspecified TableFormat = iota
-	_                                  // TableFormatLevelDB; deprecated.
-	_                                  // TableFormatRocksDBv2; deprecated.
-	TableFormatPebblev1                // Block properties.
-	TableFormatPebblev2                // Range keys.
-	TableFormatPebblev3                // Value blocks.
-	TableFormatPebblev4                // DELSIZED tombstones.
+	TableFormatLevelDB
+	TableFormatRocksDBv2
+	TableFormatPebblev1 // Block properties.
+	TableFormatPebblev2 // Range keys.
+	TableFormatPebblev3 // Value blocks.
+	TableFormatPebblev4 // DELSIZED tombstones.
 	NumTableFormats
 
-	TableFormatMax          = NumTableFormats - 1
+	TableFormatMax = NumTableFormats - 1
+
+	// TableFormatMinSupported is the minimum format supported by Pebble.  This
+	// package still supports older formats for uses outside of Pebble
+	// (CockroachDB uses it to read data from backups that could be old).
 	TableFormatMinSupported = TableFormatPebblev1
 )
 
@@ -208,6 +212,15 @@ const (
 // corresponding internal TableFormat.
 func ParseTableFormat(magic []byte, version uint32) (TableFormat, error) {
 	switch string(magic) {
+	case levelDBMagic:
+		return TableFormatLevelDB, nil
+	case rocksDBMagic:
+		if version != rocksDBFormatVersion2 {
+			return TableFormatUnspecified, base.CorruptionErrorf(
+				"pebble/table: unsupported rocksdb format version %d", errors.Safe(version),
+			)
+		}
+		return TableFormatRocksDBv2, nil
 	case pebbleDBMagic:
 		switch version {
 		case 1:
@@ -233,6 +246,10 @@ func ParseTableFormat(magic []byte, version uint32) (TableFormat, error) {
 // AsTuple returns the TableFormat's (Magic String, Version) tuple.
 func (f TableFormat) AsTuple() (string, uint32) {
 	switch f {
+	case TableFormatLevelDB:
+		return levelDBMagic, 0
+	case TableFormatRocksDBv2:
+		return rocksDBMagic, 2
 	case TableFormatPebblev1:
 		return pebbleDBMagic, 1
 	case TableFormatPebblev2:
@@ -249,6 +266,10 @@ func (f TableFormat) AsTuple() (string, uint32) {
 // String returns the TableFormat (Magic String,Version) tuple.
 func (f TableFormat) String() string {
 	switch f {
+	case TableFormatLevelDB:
+		return "(LevelDB)"
+	case TableFormatRocksDBv2:
+		return "(RocksDB,v2)"
 	case TableFormatPebblev1:
 		return "(Pebble,v1)"
 	case TableFormatPebblev2:

sstable/format_test.go

@@ -19,6 +19,18 @@ func TestTableFormat_RoundTrip(t *testing.T) {
 		wantErr string
 	}{
 		// Valid cases.
+		{
+			name:    "LevelDB",
+			magic:   levelDBMagic,
+			version: 0,
+			want:    TableFormatLevelDB,
+		},
+		{
+			name:    "RocksDBv2",
+			magic:   rocksDBMagic,
+			version: 2,
+			want:    TableFormatRocksDBv2,
+		},
 		{
 			name:    "PebbleDBv1",
 			magic:   pebbleDBMagic,
@@ -45,9 +57,10 @@ func TestTableFormat_RoundTrip(t *testing.T) {
 		},
 		// Invalid cases.
 		{
-			name:    "Deprecated RocksDB magic",
+			name:    "Invalid RocksDB version",
 			magic:   rocksDBMagic,
-			wantErr: "pebble/table: invalid table (bad magic number: 0xf7cff485b741e288)",
+			version: 1,
+			wantErr: "pebble/table: unsupported rocksdb format version 1",
 		},
 		{
 			name:    "Invalid PebbleDB version",
@@ -60,16 +73,6 @@ func TestTableFormat_RoundTrip(t *testing.T) {
 			magic:   "foo",
 			wantErr: "pebble/table: invalid table (bad magic number: 0x666f6f)",
 		},
-		{
-			name:    "LevelDB",
-			magic:   levelDBMagic,
-			wantErr: "pebble/table: invalid table (bad magic number: 0x57fb808b247547db)",
-		},
-		{
-			name:    "RocksDBv2",
-			magic:   rocksDBMagic,
-			wantErr: "pebble/table: invalid table (bad magic number: 0xf7cff485b741e288)",
-		},
 	}
 
 	for _, tc := range tcs {

sstable/options.go

@@ -295,7 +295,7 @@ func (o WriterOptions) ensureDefaults() WriterOptions {
 		o.Checksum = ChecksumTypeCRC32c
 	}
 	// By default, if the table format is not specified, fall back to using the
-	// most compatible format.
+	// most compatible format that is supported by Pebble.
 	if o.TableFormat == TableFormatUnspecified {
 		o.TableFormat = TableFormatMinSupported
 	}

sstable/random_test.go

@@ -274,8 +274,8 @@ func (cfg *randomTableConfig) readerOpts() ReaderOptions {
 func (cfg *randomTableConfig) randomize() {
 	if cfg.wopts == nil {
 		cfg.wopts = &WriterOptions{
-			// Test all table formats in [TableFormatMinSupported, TableFormatMax].
-			TableFormat:             TableFormatMinSupported + TableFormat(cfg.rng.Intn(int(TableFormatMax-TableFormatMinSupported+1))),
+			// Test all table formats in [TableFormatLevelDB, TableFormatMax].
+			TableFormat:             TableFormat(cfg.rng.Intn(int(TableFormatMax)) + 1),
 			BlockRestartInterval:    (1 << cfg.rng.Intn(6)),             // {1, 2, 4, ..., 32}
 			BlockSizeThreshold:      min(int(100*cfg.rng.Float64()), 1), // 1-100%
 			BlockSize:               (1 << cfg.rng.Intn(18)),            // {1, 2, 4, ..., 128 KiB}

sstable/reader_test.go

@@ -563,7 +563,7 @@ func forEveryTableFormat[I any](
 	t *testing.T, formatTable [NumTableFormats]I, runTest func(*testing.T, TableFormat, I),
 ) {
 	t.Helper()
-	for tf := TableFormatMinSupported; tf <= TableFormatMax; tf++ {
+	for tf := TableFormatUnspecified + 1; tf <= TableFormatMax; tf++ {
 		t.Run(tf.String(), func(t *testing.T) {
 			runTest(t, tf, formatTable[tf])
 		})
@@ -574,6 +574,8 @@ func TestReaderStats(t *testing.T) {
 	forEveryTableFormat[string](t,
 		[NumTableFormats]string{
 			TableFormatUnspecified: "",
+			TableFormatLevelDB:     "testdata/readerstats_LevelDB",
+			TableFormatRocksDBv2:   "testdata/readerstats_LevelDB",
 			TableFormatPebblev1:    "testdata/readerstats_LevelDB",
 			TableFormatPebblev2:    "testdata/readerstats_LevelDB",
 			TableFormatPebblev3:    "testdata/readerstats_Pebblev3",
@@ -604,6 +606,8 @@ func TestReaderWithBlockPropertyFilter(t *testing.T) {
 	forEveryTableFormat[string](t,
 		[NumTableFormats]string{
 			TableFormatUnspecified: "", // Block properties unsupported
+			TableFormatLevelDB:     "", // Block properties unsupported
+			TableFormatRocksDBv2:   "", // Block properties unsupported
 			TableFormatPebblev1:    "", // Block properties unsupported
 			TableFormatPebblev2:    "testdata/reader_bpf/Pebblev2",
 			TableFormatPebblev3:    "testdata/reader_bpf/Pebblev3",
@@ -1385,7 +1389,7 @@ func TestReader_TableFormat(t *testing.T) {
 		require.Equal(t, want, got)
 	}
 
-	for tf := TableFormatMinSupported; tf <= TableFormatMax; tf++ {
+	for tf := TableFormatLevelDB; tf <= TableFormatMax; tf++ {
 		t.Run(tf.String(), func(t *testing.T) {
 			test(t, tf)
 		})

sstable/table.go

@@ -206,6 +206,9 @@ const (
 	minFooterLen = levelDBFooterLen
 	maxFooterLen = rocksDBFooterLen
 
+	levelDBFormatVersion  = 0
+	rocksDBFormatVersion2 = 2
+
 	metaRangeKeyName   = "pebble.range_key"
 	metaValueIndexName = "pebble.value_index"
 	metaPropertiesName = "rocksdb.properties"
@@ -339,7 +342,18 @@ func readFooter(f objstorage.Readable) (footer, error) {
 	}
 
 	switch magic := buf[len(buf)-len(rocksDBMagic):]; string(magic) {
-	case pebbleDBMagic:
+	case levelDBMagic:
+		if len(buf) < levelDBFooterLen {
+			return footer, base.CorruptionErrorf(
+				"pebble/table: invalid table (footer too short): %d", errors.Safe(len(buf)))
+		}
+		footer.footerBH.Offset = uint64(off+int64(len(buf))) - levelDBFooterLen
+		buf = buf[len(buf)-levelDBFooterLen:]
+		footer.footerBH.Length = uint64(len(buf))
+		footer.format = TableFormatLevelDB
+		footer.checksum = ChecksumTypeCRC32c
+
+	case rocksDBMagic, pebbleDBMagic:
 		// NOTE: The Pebble magic string implies the same footer format as that used
 		// by the RocksDBv2 table format.
 		if len(buf) < rocksDBFooterLen {
@@ -390,7 +404,14 @@ func readFooter(f objstorage.Readable) (footer, error) {
 
 func (f footer) encode(buf []byte) []byte {
 	switch magic, version := f.format.AsTuple(); magic {
-	case pebbleDBMagic:
+	case levelDBMagic:
+		buf = buf[:levelDBFooterLen]
+		clear(buf)
+		n := encodeBlockHandle(buf[0:], f.metaindexBH)
+		encodeBlockHandle(buf[n:], f.indexBH)
+		copy(buf[len(buf)-len(levelDBMagic):], levelDBMagic)
+
+	case rocksDBMagic, pebbleDBMagic:
 		buf = buf[:rocksDBFooterLen]
 		clear(buf)
 		switch f.checksum {
@@ -419,5 +440,12 @@ func (f footer) encode(buf []byte) []byte {
 }
 
 func supportsTwoLevelIndex(format TableFormat) bool {
-	return format >= TableFormatMinSupported
+	switch format {
+	case TableFormatLevelDB:
+		return false
+	case TableFormatRocksDBv2, TableFormatPebblev1, TableFormatPebblev2, TableFormatPebblev3, TableFormatPebblev4:
+		return true
+	default:
+		panic("sstable: unspecified table format version")
+	}
 }

sstable/table_test.go

@@ -549,10 +549,10 @@ func TestMetaIndexEntriesSorted(t *testing.T) {
 
 func TestFooterRoundTrip(t *testing.T) {
 	buf := make([]byte, 100+maxFooterLen)
-	for format := TableFormatMinSupported; format < TableFormatMax; format++ {
+	for format := TableFormatLevelDB; format < TableFormatMax; format++ {
 		t.Run(fmt.Sprintf("format=%s", format), func(t *testing.T) {
 			checksums := []ChecksumType{ChecksumTypeCRC32c}
-			if format != TableFormatMinSupported {
+			if format != TableFormatLevelDB {
 				checksums = []ChecksumType{ChecksumTypeCRC32c, ChecksumTypeXXHash64}
 			}
 			for _, checksum := range checksums {
@@ -603,13 +603,25 @@ func TestFooterRoundTrip(t *testing.T) {
 }
 
 func TestReadFooter(t *testing.T) {
+	encode := func(format TableFormat, checksum ChecksumType) string {
+		f := footer{
+			format:   format,
+			checksum: checksum,
+		}
+		return string(f.encode(make([]byte, maxFooterLen)))
+	}
+
 	testCases := []struct {
 		encoded  string
 		expected string
 	}{
 		{strings.Repeat("a", minFooterLen-1), "file size is too small"},
 		{strings.Repeat("a", levelDBFooterLen), "bad magic number"},
 		{strings.Repeat("a", rocksDBFooterLen), "bad magic number"},
+		{encode(TableFormatLevelDB, 0)[1:], "file size is too small"},
+		{encode(TableFormatRocksDBv2, 0)[1:], "footer too short"},
+		{encode(TableFormatRocksDBv2, ChecksumTypeNone), "unsupported checksum type"},
+		{encode(TableFormatRocksDBv2, ChecksumTypeXXHash), "unsupported checksum type"},
 	}
 	for _, c := range testCases {
 		t.Run("", func(t *testing.T) {

sstable/testdata/writer

@@ -343,14 +343,11 @@ layout
          0  data (21)
         26  data (21)
         52  data (21)
-        78  index (22)
-       105  index (22)
-       132  index (22)
-       159  top-index (50)
-       214  properties (580)
-       799  meta-index (33)
-       837  footer (53)
-       890  EOF
+        78  index (47)
+       130  properties (678)
+       813  meta-index (33)
+       851  leveldb-footer (48)
+       899  EOF
 
 # Range keys, if present, are shown in the layout.
 

sstable/testdata/writer_v3

@@ -313,17 +313,14 @@ seqnums:  [1-1]
 
 layout
 ----
-         0  data (22)
-        27  data (22)
-        54  data (22)
-        81  index (22)
-       108  index (22)
-       135  index (22)
-       162  top-index (51)
-       218  properties (580)
-       803  meta-index (33)
-       841  footer (53)
-       894  EOF
+         0  data (21)
+        26  data (21)
+        52  data (21)
+        78  index (47)
+       130  properties (678)
+       813  meta-index (33)
+       851  leveldb-footer (48)
+       899  EOF
 
 # Range keys, if present, are shown in the layout.
 

sstable/writer_test.go

@@ -822,7 +822,7 @@ func TestWriter_TableFormatCompatibility(t *testing.T) {
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
-			for tf := TableFormatMinSupported; tf <= TableFormatMax; tf++ {
+			for tf := TableFormatLevelDB; tf <= TableFormatMax; tf++ {
 				t.Run(tf.String(), func(t *testing.T) {
 					fs := vfs.NewMem()
 					f, err := fs.Create("sst")