format_major_version_test.go

// Copyright 2021 The LevelDB-Go and Pebble Authors. All rights reserved. Use
// of this source code is governed by a BSD-style license that can be found in
// the LICENSE file.

package pebble

import (
	"bytes"
	"fmt"
	"strconv"
	"sync"
	"testing"
	"time"

	"github.com/cockroachdb/datadriven"
	"github.com/cockroachdb/pebble/bloom"
	"github.com/cockroachdb/pebble/internal/base"
	"github.com/cockroachdb/pebble/internal/testkeys"
	"github.com/cockroachdb/pebble/sstable"
	"github.com/cockroachdb/pebble/vfs"
	"github.com/cockroachdb/pebble/vfs/atomicfs"
	"github.com/stretchr/testify/require"
)

func TestFormatMajorVersion_MigrationDefined(t *testing.T) {
	for v := FormatMostCompatible; v <= FormatNewest; v++ {
		if _, ok := formatMajorVersionMigrations[v]; !ok {
			t.Errorf("format major version %d has no migration defined", v)
		}
	}
}

func TestRatchetFormat(t *testing.T) {
	fs := vfs.NewMem()
	d, err := Open("", (&Options{FS: fs}).WithFSDefaults())
	require.NoError(t, err)
	require.NoError(t, d.Set([]byte("foo"), []byte("bar"), Sync))
	require.Equal(t, FormatMostCompatible, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatVersioned))
	require.Equal(t, FormatVersioned, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatVersioned))
	require.Equal(t, FormatVersioned, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatSetWithDelete))
	require.Equal(t, FormatSetWithDelete, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatBlockPropertyCollector))
	require.Equal(t, FormatBlockPropertyCollector, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatSplitUserKeysMarked))
	require.Equal(t, FormatSplitUserKeysMarked, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatSplitUserKeysMarkedCompacted))
	require.Equal(t, FormatSplitUserKeysMarkedCompacted, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatRangeKeys))
	require.Equal(t, FormatRangeKeys, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatMinTableFormatPebblev1))
	require.Equal(t, FormatMinTableFormatPebblev1, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatPrePebblev1Marked))
	require.Equal(t, FormatPrePebblev1Marked, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(formatUnusedPrePebblev1MarkedCompacted))
	require.Equal(t, formatUnusedPrePebblev1MarkedCompacted, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatSSTableValueBlocks))
	require.Equal(t, FormatSSTableValueBlocks, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatFlushableIngest))
	require.Equal(t, FormatFlushableIngest, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatPrePebblev1MarkedCompacted))
	require.Equal(t, FormatPrePebblev1MarkedCompacted, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatDeleteSizedAndObsolete))
	require.Equal(t, FormatDeleteSizedAndObsolete, d.FormatMajorVersion())
	require.NoError(t, d.RatchetFormatMajorVersion(FormatVirtualSSTables))
	require.Equal(t, FormatVirtualSSTables, d.FormatMajorVersion())

	require.NoError(t, d.Close())

	// If we Open the database again, leaving the default format, the
	// database should Open using the persisted FormatNewest.
	d, err = Open("", (&Options{FS: fs}).WithFSDefaults())
	require.NoError(t, err)
	require.Equal(t, internalFormatNewest, d.FormatMajorVersion())
	require.NoError(t, d.Close())

	// Move the marker to a version that does not exist.
	m, _, err := atomicfs.LocateMarker(fs, "", formatVersionMarkerName)
	require.NoError(t, err)
	require.NoError(t, m.Move("999999"))
	require.NoError(t, m.Close())

	_, err = Open("", (&Options{
		FS:                 fs,
		FormatMajorVersion: FormatVersioned,
	}).WithFSDefaults())
	require.Error(t, err)
	require.EqualError(t, err, `pebble: database "" written in format major version 999999`)
}

func testBasicDB(d *DB) error {
	key := []byte("a")
	value := []byte("b")
	if err := d.Set(key, value, nil); err != nil {
		return err
	}
	if err := d.Flush(); err != nil {
		return err
	}
	if err := d.Compact(nil, []byte("\xff"), false); err != nil {
		return err
	}

	iter, _ := d.NewIter(nil)
	for valid := iter.First(); valid; valid = iter.Next() {
	}
	if err := iter.Close(); err != nil {
		return err
	}
	return nil
}

func TestFormatMajorVersions(t *testing.T) {
	for vers := FormatMostCompatible; vers <= FormatNewest; vers++ {
		t.Run(fmt.Sprintf("vers=%03d", vers), func(t *testing.T) {
			fs := vfs.NewStrictMem()
			opts := (&Options{
				FS:                 fs,
				FormatMajorVersion: vers,
			}).WithFSDefaults()

			// Create a database at this format major version and perform
			// some very basic operations.
			d, err := Open("", opts)
			require.NoError(t, err)
			require.NoError(t, testBasicDB(d))
			require.NoError(t, d.Close())

			// Re-open the database at this format major version, and again
			// perform some basic operations.
			d, err = Open("", opts)
			require.NoError(t, err)
			require.NoError(t, testBasicDB(d))
			require.NoError(t, d.Close())

			t.Run("upgrade-at-open", func(t *testing.T) {
				for upgradeVers := vers + 1; upgradeVers <= FormatNewest; upgradeVers++ {
					t.Run(fmt.Sprintf("upgrade-vers=%03d", upgradeVers), func(t *testing.T) {
						// We use vfs.MemFS's option to ignore syncs so
						// that we can perform an upgrade on the current
						// database state in fs, and revert it when this
						// subtest is complete.
						fs.SetIgnoreSyncs(true)
						defer fs.ResetToSyncedState()

						// Re-open the database, passing a higher format
						// major version in the Options to automatically
						// ratchet the format major version. Ensure some
						// basic operations pass.
						opts := opts.Clone()
						opts.FormatMajorVersion = upgradeVers
						d, err = Open("", opts)
						require.NoError(t, err)
						require.Equal(t, upgradeVers, d.FormatMajorVersion())
						require.NoError(t, testBasicDB(d))
						require.NoError(t, d.Close())

						// Re-open to ensure the upgrade persisted.
						d, err = Open("", opts)
						require.NoError(t, err)
						require.Equal(t, upgradeVers, d.FormatMajorVersion())
						require.NoError(t, testBasicDB(d))
						require.NoError(t, d.Close())
					})
				}
			})

			t.Run("upgrade-while-open", func(t *testing.T) {
				for upgradeVers := vers + 1; upgradeVers <= FormatNewest; upgradeVers++ {
					t.Run(fmt.Sprintf("upgrade-vers=%03d", upgradeVers), func(t *testing.T) {
						// Ensure the previous tests don't overwrite our
						// options.
						require.Equal(t, vers, opts.FormatMajorVersion)

						// We use vfs.MemFS's option to ignore syncs so
						// that we can perform an upgrade on the current
						// database state in fs, and revert it when this
						// subtest is complete.
						fs.SetIgnoreSyncs(true)
						defer fs.ResetToSyncedState()

						// Re-open the database, still at the current format
						// major version. Perform some basic operations,
						// ratchet the format version up, and perform
						// additional basic operations.
						d, err = Open("", opts)
						require.NoError(t, err)
						require.NoError(t, testBasicDB(d))
						require.Equal(t, vers, d.FormatMajorVersion())
						require.NoError(t, d.RatchetFormatMajorVersion(upgradeVers))
						require.Equal(t, upgradeVers, d.FormatMajorVersion())
						require.NoError(t, testBasicDB(d))
						require.NoError(t, d.Close())

						// Re-open to ensure the upgrade persisted.
						d, err = Open("", opts)
						require.NoError(t, err)
						require.Equal(t, upgradeVers, d.FormatMajorVersion())
						require.NoError(t, testBasicDB(d))
						require.NoError(t, d.Close())
					})
				}
			})
		})
	}
}

func TestFormatMajorVersions_TableFormat(t *testing.T) {
	// NB: This test is intended to validate the mapping between every
	// FormatMajorVersion and sstable.TableFormat exhaustively. This serves as a
	// sanity check that new versions have a corresponding mapping. The test
	// fixture is intentionally verbose.

	m := map[FormatMajorVersion][2]sstable.TableFormat{
		FormatDefault:                          {sstable.TableFormatLevelDB, sstable.TableFormatRocksDBv2},
		FormatMostCompatible:                   {sstable.TableFormatLevelDB, sstable.TableFormatRocksDBv2},
		formatVersionedManifestMarker:          {sstable.TableFormatLevelDB, sstable.TableFormatRocksDBv2},
		FormatVersioned:                        {sstable.TableFormatLevelDB, sstable.TableFormatRocksDBv2},
		FormatSetWithDelete:                    {sstable.TableFormatLevelDB, sstable.TableFormatRocksDBv2},
		FormatBlockPropertyCollector:           {sstable.TableFormatLevelDB, sstable.TableFormatPebblev1},
		FormatSplitUserKeysMarked:              {sstable.TableFormatLevelDB, sstable.TableFormatPebblev1},
		FormatSplitUserKeysMarkedCompacted:     {sstable.TableFormatLevelDB, sstable.TableFormatPebblev1},
		FormatRangeKeys:                        {sstable.TableFormatLevelDB, sstable.TableFormatPebblev2},
		FormatMinTableFormatPebblev1:           {sstable.TableFormatPebblev1, sstable.TableFormatPebblev2},
		FormatPrePebblev1Marked:                {sstable.TableFormatPebblev1, sstable.TableFormatPebblev2},
		formatUnusedPrePebblev1MarkedCompacted: {sstable.TableFormatPebblev1, sstable.TableFormatPebblev2},
		FormatSSTableValueBlocks:               {sstable.TableFormatPebblev1, sstable.TableFormatPebblev3},
		FormatFlushableIngest:                  {sstable.TableFormatPebblev1, sstable.TableFormatPebblev3},
		FormatPrePebblev1MarkedCompacted:       {sstable.TableFormatPebblev1, sstable.TableFormatPebblev3},
		FormatDeleteSizedAndObsolete:           {sstable.TableFormatPebblev1, sstable.TableFormatPebblev4},
		FormatVirtualSSTables:                  {sstable.TableFormatPebblev1, sstable.TableFormatPebblev4},
	}

	// Valid versions.
	for fmv := FormatMostCompatible; fmv <= internalFormatNewest; fmv++ {
		got := [2]sstable.TableFormat{fmv.MinTableFormat(), fmv.MaxTableFormat()}
		require.Equalf(t, m[fmv], got, "got %s; want %s", got, m[fmv])
		require.True(t, got[0] <= got[1] /* min <= max */)
	}

	// Invalid versions.
	fmv := internalFormatNewest + 1
	require.Panics(t, func() { _ = fmv.MaxTableFormat() })
	require.Panics(t, func() { _ = fmv.MinTableFormat() })
}

func TestSplitUserKeyMigration(t *testing.T) {
	var d *DB
	var opts *Options
	var fs vfs.FS
	var buf bytes.Buffer
	defer func() {
		if d != nil {
			require.NoError(t, d.Close())
		}
	}()

	datadriven.RunTest(t, "testdata/format_major_version_split_user_key_migration",
		func(t *testing.T, td *datadriven.TestData) string {
			switch td.Cmd {
			case "define":
				if d != nil {
					if err := d.Close(); err != nil {
						return err.Error()
					}
					buf.Reset()
				}
				opts = (&Options{
					FormatMajorVersion: FormatBlockPropertyCollector,
					EventListener: &EventListener{
						CompactionEnd: func(info CompactionInfo) {
							// Fix the job ID and durations for determinism.
							info.JobID = 100
							info.Duration = time.Second
							info.TotalDuration = 2 * time.Second
							fmt.Fprintln(&buf, info)
						},
					},
					DisableAutomaticCompactions: true,
				}).WithFSDefaults()
				var err error
				if d, err = runDBDefineCmd(td, opts); err != nil {
					return err.Error()
				}

				fs = d.opts.FS
				d.mu.Lock()
				defer d.mu.Unlock()
				return d.mu.versions.currentVersion().DebugString(base.DefaultFormatter)
			case "reopen":
				if d != nil {
					if err := d.Close(); err != nil {
						return err.Error()
					}
					buf.Reset()
				}
				opts.FS = fs
				opts.DisableAutomaticCompactions = true
				var err error
				d, err = Open("", opts)
				if err != nil {
					return err.Error()
				}
				return "OK"
			case "build":
				if err := runBuildCmd(td, d, fs); err != nil {
					return err.Error()
				}
				return ""
			case "force-ingest":
				if err := runForceIngestCmd(td, d); err != nil {
					return err.Error()
				}
				d.mu.Lock()
				defer d.mu.Unlock()
				return d.mu.versions.currentVersion().DebugString(base.DefaultFormatter)
			case "format-major-version":
				return d.FormatMajorVersion().String()
			case "ratchet-format-major-version":
				v, err := strconv.Atoi(td.CmdArgs[0].String())
				if err != nil {
					return err.Error()
				}
				if err := d.RatchetFormatMajorVersion(FormatMajorVersion(v)); err != nil {
					return err.Error()
				}
				return buf.String()
			case "lsm":
				return runLSMCmd(td, d)
			case "marked-file-count":
				m := d.Metrics()
				return fmt.Sprintf("%d files marked for compaction", m.Compact.MarkedFiles)
			case "disable-automatic-compactions":
				d.mu.Lock()
				defer d.mu.Unlock()
				switch v := td.CmdArgs[0].String(); v {
				case "true":
					d.opts.DisableAutomaticCompactions = true
				case "false":
					d.opts.DisableAutomaticCompactions = false
				default:
					return fmt.Sprintf("unknown value %q", v)
				}
				return ""
			default:
				return fmt.Sprintf("unrecognized command %q", td.Cmd)
			}
		})
}

func TestPebblev1Migration(t *testing.T) {
	var d *DB
	defer func() {
		if d != nil {
			require.NoError(t, d.Close())
		}
	}()

	datadriven.RunTest(t, "testdata/format_major_version_pebblev1_migration",
		func(t *testing.T, td *datadriven.TestData) string {
			switch cmd := td.Cmd; cmd {
			case "open":
				var version int
				var err error
				for _, cmdArg := range td.CmdArgs {
					switch cmd := cmdArg.Key; cmd {
					case "version":
						version, err = strconv.Atoi(cmdArg.Vals[0])
						if err != nil {
							return err.Error()
						}
					default:
						return fmt.Sprintf("unknown argument: %s", cmd)
					}
				}
				opts := (&Options{
					FS:                 vfs.NewMem(),
					FormatMajorVersion: FormatMajorVersion(version),
				}).WithFSDefaults()
				d, err = Open("", opts)
				if err != nil {
					return err.Error()
				}
				return ""

			case "format-major-version":
				return d.FormatMajorVersion().String()

			case "min-table-format":
				return d.FormatMajorVersion().MinTableFormat().String()

			case "max-table-format":
				return d.FormatMajorVersion().MaxTableFormat().String()

			case "disable-automatic-compactions":
				d.mu.Lock()
				defer d.mu.Unlock()
				switch v := td.CmdArgs[0].String(); v {
				case "true":
					d.opts.DisableAutomaticCompactions = true
				case "false":
					d.opts.DisableAutomaticCompactions = false
				default:
					return fmt.Sprintf("unknown value %q", v)
				}
				return ""

			case "batch":
				b := d.NewIndexedBatch()
				if err := runBatchDefineCmd(td, b); err != nil {
					return err.Error()
				}
				if err := b.Commit(nil); err != nil {
					return err.Error()
				}
				return ""

			case "flush":
				if err := d.Flush(); err != nil {
					return err.Error()
				}
				return ""

			case "ingest":
				if err := runBuildCmd(td, d, d.opts.FS); err != nil {
					return err.Error()
				}
				// Only the first arg is a filename.
				td.CmdArgs = td.CmdArgs[:1]
				if err := runIngestCmd(td, d, d.opts.FS); err != nil {
					return err.Error()
				}
				return ""

			case "lsm":
				return runLSMCmd(td, d)

			case "tally-table-formats":
				d.mu.Lock()
				defer d.mu.Unlock()
				v := d.mu.versions.currentVersion()
				tally := make([]int, sstable.TableFormatMax+1)
				for _, l := range v.Levels {
					iter := l.Iter()
					for m := iter.First(); m != nil; m = iter.Next() {
						err := d.tableCache.withReader(m.PhysicalMeta(),
							func(r *sstable.Reader) error {
								f, err := r.TableFormat()
								if err != nil {
									return err
								}
								tally[f]++
								return nil
							})
						if err != nil {
							return err.Error()
						}
					}
				}
				var b bytes.Buffer
				for i := 1; i <= int(sstable.TableFormatMax); i++ {
					_, _ = fmt.Fprintf(&b, "%s: %d\n", sstable.TableFormat(i), tally[i])
				}
				return b.String()

			case "ratchet-format-major-version":
				v, err := strconv.Atoi(td.CmdArgs[0].String())
				if err != nil {
					return err.Error()
				}
				if err = d.RatchetFormatMajorVersion(FormatMajorVersion(v)); err != nil {
					return err.Error()
				}
				return ""

			case "marked-file-count":
				m := d.Metrics()
				return fmt.Sprintf("%d files marked for compaction", m.Compact.MarkedFiles)

			default:
				return fmt.Sprintf("unknown command: %s", cmd)
			}
		},
	)
}

// TestPebblev1MigrationRace exercises the race between a PrePebbleV1Marked
// format major version upgrade that needs to open sstables to read their table
// format, and concurrent compactions that may delete the same files from the
// LSM.
//
// Regression test for #2019.
func TestPebblev1MigrationRace(t *testing.T) {
	// Use a smaller table cache size to slow down the PrePebbleV1Marked
	// migration, ensuring each table read needs to re-open the file.
	cache := NewCache(4 << 20)
	defer cache.Unref()
	tableCache := NewTableCache(cache, 1, 5)
	defer tableCache.Unref()
	d, err := Open("", (&Options{
		Cache:              cache,
		FS:                 vfs.NewMem(),
		FormatMajorVersion: FormatMajorVersion(FormatPrePebblev1Marked - 1),
		TableCache:         tableCache,
		Levels:             []LevelOptions{{TargetFileSize: 1}},
	}).WithFSDefaults())
	require.NoError(t, err)
	defer d.Close()

	ks := testkeys.Alpha(3).EveryN(10)
	var key [3]byte
	for i := int64(0); i < ks.Count(); i++ {
		n := testkeys.WriteKey(key[:], ks, i)
		require.NoError(t, d.Set(key[:n], key[:n], nil))
		require.NoError(t, d.Flush())
	}

	// Asynchronously write and flush range deletes that will cause compactions
	// to delete the existing sstables. These deletes will race with the format
	// major version upgrade's migration will attempt to delete the files.
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		for i := ks.Count() - 1; i > 0; i -= 50 {
			endKey := testkeys.Key(ks, i)
			startIndex := i - 50
			if startIndex < 0 {
				startIndex = 0
			}
			startKey := testkeys.Key(ks, startIndex)

			require.NoError(t, d.DeleteRange(startKey, endKey, nil))
			_, err := d.AsyncFlush()
			require.NoError(t, err)
		}
	}()
	require.NoError(t, d.RatchetFormatMajorVersion(FormatPrePebblev1Marked))
	wg.Wait()
}

// Regression test for #2044, where multiple concurrent compactions can lead
// to an indefinite wait on the compaction goroutine in compactMarkedFilesLocked.
func TestPebblev1MigrationConcurrencyRace(t *testing.T) {
	opts := (&Options{
		Comparer:           testkeys.Comparer,
		FS:                 vfs.NewMem(),
		FormatMajorVersion: FormatSplitUserKeysMarked,
		Levels:             []LevelOptions{{FilterPolicy: bloom.FilterPolicy(10)}},
		MaxConcurrentCompactions: func() int {
			return 4
		},
	}).WithFSDefaults()
	func() {
		d, err := Open("", opts)
		require.NoError(t, err)
		defer func() {
			require.NoError(t, d.Close())
		}()

		ks := testkeys.Alpha(3).EveryN(10)
		var key [3]byte
		for i := int64(0); i < ks.Count(); i++ {
			n := testkeys.WriteKey(key[:], ks, i)
			require.NoError(t, d.Set(key[:n], key[:n], nil))
			if i%100 == 0 {
				require.NoError(t, d.Flush())
			}
		}
		require.NoError(t, d.Flush())
	}()

	opts.FormatMajorVersion = formatUnusedPrePebblev1MarkedCompacted
	d, err := Open("", opts)
	require.NoError(t, err)
	require.NoError(t, d.RatchetFormatMajorVersion(formatUnusedPrePebblev1MarkedCompacted))
	require.NoError(t, d.Close())
}