smeb_y/ceremonyclient
1
0
Fork 0
mirror of https://source.quilibrium.com/quilibrium/ceremonyclient.git synced 2024-12-26 16:45:18 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
b5d82ba1b7
ceremonyclient/pebble/ingest_test.go
Cassandra Heart 2e2a1e4789
v1.2.0 (#31)
2024-01-03 01:31:42 -06:00

3517 lines
95 KiB
Go
Raw Blame History

// Copyright 2018 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"
"context"
"fmt"
"io"
"math"
"os"
"path/filepath"
"slices"
"sort"
"strconv"
"strings"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/cockroachdb/datadriven"
"github.com/cockroachdb/errors"
"github.com/cockroachdb/errors/oserror"
"github.com/cockroachdb/pebble/internal/base"
"github.com/cockroachdb/pebble/internal/keyspan"
"github.com/cockroachdb/pebble/internal/manifest"
"github.com/cockroachdb/pebble/internal/rangekey"
"github.com/cockroachdb/pebble/internal/testkeys"
"github.com/cockroachdb/pebble/objstorage"
"github.com/cockroachdb/pebble/objstorage/objstorageprovider"
"github.com/cockroachdb/pebble/objstorage/remote"
"github.com/cockroachdb/pebble/record"
"github.com/cockroachdb/pebble/sstable"
"github.com/cockroachdb/pebble/vfs"
"github.com/cockroachdb/pebble/vfs/errorfs"
"github.com/kr/pretty"
"github.com/stretchr/testify/require"
"golang.org/x/exp/rand"
)
func TestSSTableKeyCompare(t *testing.T) {
var buf bytes.Buffer
datadriven.RunTest(t, "testdata/sstable_key_compare", func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "cmp":
buf.Reset()
for _, line := range strings.Split(td.Input, "\n") {
fields := strings.Fields(line)
a := base.ParseInternalKey(fields[0])
b := base.ParseInternalKey(fields[1])
got := sstableKeyCompare(testkeys.Comparer.Compare, a, b)
fmt.Fprintf(&buf, "%38s", fmt.Sprint(a.Pretty(base.DefaultFormatter)))
switch got {
case -1:
fmt.Fprint(&buf, " < ")
case +1:
fmt.Fprint(&buf, " > ")
case 0:
fmt.Fprint(&buf, " = ")
}
fmt.Fprintf(&buf, "%s\n", fmt.Sprint(b.Pretty(base.DefaultFormatter)))
}
return buf.String()
default:
return fmt.Sprintf("unrecognized command %q", td.Cmd)
}
})
}
func TestIngestLoad(t *testing.T) {
mem := vfs.NewMem()
datadriven.RunTest(t, "testdata/ingest_load", func(t *testing.T, 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(objstorageprovider.NewFileWritable(f), writerOpts)
for _, data := range strings.Split(td.Input, "\n") {
if strings.HasPrefix(data, "rangekey: ") {
data = strings.TrimPrefix(data, "rangekey: ")
s := keyspan.ParseSpan(data)
err := rangekey.Encode(&s, w.AddRangeKey)
if err != nil {
return err.Error()
}
continue
}
j := strings.Index(data, ":")
if j < 0 {
return fmt.Sprintf("malformed input: %s\n", data)
}
key := base.ParseInternalKey(data[:j])
value := []byte(data[j+1:])
if err := w.Add(key, value); err != nil {
return err.Error()
}
}
if err := w.Close(); err != nil {
return err.Error()
}
opts := (&Options{
Comparer: DefaultComparer,
FS: mem,
}).WithFSDefaults()
lr, err := ingestLoad(opts, dbVersion, []string{"ext"}, nil, nil, 0, []base.DiskFileNum{base.FileNum(1).DiskFileNum()}, nil, 0)
if err != nil {
return err.Error()
}
var buf bytes.Buffer
for _, m := range lr.localMeta {
fmt.Fprintf(&buf, "%d: %s-%s\n", m.FileNum, m.Smallest, m.Largest)
fmt.Fprintf(&buf, " points: %s-%s\n", m.SmallestPointKey, m.LargestPointKey)
fmt.Fprintf(&buf, " ranges: %s-%s\n", m.SmallestRangeKey, m.LargestRangeKey)
}
return buf.String()
default:
return fmt.Sprintf("unknown command: %s", td.Cmd)
}
})
}
func TestIngestLoadRand(t *testing.T) {
mem := vfs.NewMem()
rng := rand.New(rand.NewSource(uint64(time.Now().UnixNano())))
cmp := DefaultComparer.Compare
version := internalFormatNewest
randBytes := func(size int) []byte {
data := make([]byte, size)
for i := range data {
data[i] = byte(rng.Int() & 0xff)
}
return data
}
paths := make([]string, 1+rng.Intn(10))
pending := make([]base.DiskFileNum, len(paths))
expected := make([]*fileMetadata, len(paths))
for i := range paths {
paths[i] = fmt.Sprint(i)
pending[i] = base.FileNum(rng.Uint64()).DiskFileNum()
expected[i] = &fileMetadata{
FileNum: pending[i].FileNum(),
}
expected[i].StatsMarkValid()
func() {
f, err := mem.Create(paths[i])
require.NoError(t, err)
keys := make([]InternalKey, 1+rng.Intn(100))
for i := range keys {
keys[i] = base.MakeInternalKey(
randBytes(1+rng.Intn(10)),
0,
InternalKeyKindSet)
}
slices.SortFunc(keys, func(a, b base.InternalKey) int {
return base.InternalCompare(cmp, a, b)
})
expected[i].ExtendPointKeyBounds(cmp, keys[0], keys[len(keys)-1])
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
TableFormat: version.MaxTableFormat(),
})
var count uint64
for i := range keys {
if i > 0 && base.InternalCompare(cmp, keys[i-1], keys[i]) == 0 {
// Duplicate key, ignore.
continue
}
w.Add(keys[i], nil)
count++
}
expected[i].Stats.NumEntries = count
require.NoError(t, w.Close())
meta, err := w.Metadata()
require.NoError(t, err)
expected[i].Size = meta.Size
expected[i].InitPhysicalBacking()
}()
}
opts := (&Options{
Comparer: DefaultComparer,
FS: mem,
}).WithFSDefaults()
lr, err := ingestLoad(opts, version, paths, nil, nil, 0, pending, nil, 0)
require.NoError(t, err)
for _, m := range lr.localMeta {
m.CreationTime = 0
}
t.Log(strings.Join(pretty.Diff(expected, lr.localMeta), "\n"))
require.Equal(t, expected, lr.localMeta)
}
func TestIngestLoadInvalid(t *testing.T) {
mem := vfs.NewMem()
f, err := mem.Create("invalid")
require.NoError(t, err)
require.NoError(t, f.Close())
opts := (&Options{
Comparer: DefaultComparer,
FS: mem,
}).WithFSDefaults()
if _, err := ingestLoad(opts, internalFormatNewest, []string{"invalid"}, nil, nil, 0, []base.DiskFileNum{base.FileNum(1).DiskFileNum()}, nil, 0); err == nil {
t.Fatalf("expected error, but found success")
}
}
func TestIngestSortAndVerify(t *testing.T) {
comparers := map[string]Compare{
"default": DefaultComparer.Compare,
"reverse": func(a, b []byte) int {
return DefaultComparer.Compare(b, a)
},
}
t.Run("", func(t *testing.T) {
datadriven.RunTest(t, "testdata/ingest_sort_and_verify", func(t *testing.T, d *datadriven.TestData) string {
switch d.Cmd {
case "ingest":
var buf bytes.Buffer
var meta []*fileMetadata
var paths []string
var cmpName string
d.ScanArgs(t, "cmp", &cmpName)
cmp := comparers[cmpName]
if cmp == nil {
return fmt.Sprintf("%s unknown comparer: %s", d.Cmd, cmpName)
}
for i, data := range strings.Split(d.Input, "\n") {
parts := strings.Split(data, "-")
if len(parts) != 2 {
return fmt.Sprintf("malformed test case: %s", d.Input)
}
smallest := base.ParseInternalKey(parts[0])
largest := base.ParseInternalKey(parts[1])
if cmp(smallest.UserKey, largest.UserKey) > 0 {
return fmt.Sprintf("range %v-%v is not valid", smallest, largest)
}
m := (&fileMetadata{}).ExtendPointKeyBounds(cmp, smallest, largest)
m.InitPhysicalBacking()
meta = append(meta, m)
paths = append(paths, strconv.Itoa(i))
}
lr := ingestLoadResult{localPaths: paths, localMeta: meta}
err := ingestSortAndVerify(cmp, lr, KeyRange{})
if err != nil {
return fmt.Sprintf("%v\n", err)
}
for i := range meta {
fmt.Fprintf(&buf, "%s: %v-%v\n", paths[i], meta[i].Smallest, meta[i].Largest)
}
return buf.String()
default:
return fmt.Sprintf("unknown command: %s", d.Cmd)
}
})
})
}
func TestIngestLink(t *testing.T) {
// Test linking of tables into the DB directory. Test cleanup when one of the
// tables cannot be linked.
const dir = "db"
const count = 10
for i := 0; i <= count; i++ {
t.Run("", func(t *testing.T) {
opts := &Options{FS: vfs.NewMem()}
opts.EnsureDefaults().WithFSDefaults()
require.NoError(t, opts.FS.MkdirAll(dir, 0755))
objProvider, err := objstorageprovider.Open(objstorageprovider.DefaultSettings(opts.FS, dir))
require.NoError(t, err)
defer objProvider.Close()
paths := make([]string, 10)
meta := make([]*fileMetadata, len(paths))
contents := make([][]byte, len(paths))
for j := range paths {
paths[j] = fmt.Sprintf("external%d", j)
meta[j] = &fileMetadata{}
meta[j].FileNum = FileNum(j)
meta[j].InitPhysicalBacking()
f, err := opts.FS.Create(paths[j])
require.NoError(t, err)
contents[j] = []byte(fmt.Sprintf("data%d", j))
// memFile.Write will modify the supplied buffer when invariants are
// enabled, so provide a throw-away copy.
_, err = f.Write(append([]byte(nil), contents[j]...))
require.NoError(t, err)
require.NoError(t, f.Close())
}
if i < count {
opts.FS.Remove(paths[i])
}
lr := ingestLoadResult{localMeta: meta, localPaths: paths}
err = ingestLink(0 /* jobID */, opts, objProvider, lr, nil /* shared */)
if i < count {
if err == nil {
t.Fatalf("expected error, but found success")
}
} else {
require.NoError(t, err)
}
files, err := opts.FS.List(dir)
require.NoError(t, err)
sort.Strings(files)
if i < count {
if len(files) > 0 {
t.Fatalf("expected all of the files to be cleaned up, but found:\n%s",
strings.Join(files, "\n"))
}
} else {
if len(files) != count {
t.Fatalf("expected %d files, but found:\n%s", count, strings.Join(files, "\n"))
}
for j := range files {
ftype, fileNum, ok := base.ParseFilename(opts.FS, files[j])
if !ok {
t.Fatalf("unable to parse filename: %s", files[j])
}
if fileTypeTable != ftype {
t.Fatalf("expected table, but found %d", ftype)
}
if j != int(fileNum.FileNum()) {
t.Fatalf("expected table %d, but found %d", j, fileNum)
}
f, err := opts.FS.Open(opts.FS.PathJoin(dir, files[j]))
require.NoError(t, err)
data, err := io.ReadAll(f)
require.NoError(t, err)
require.NoError(t, f.Close())
if !bytes.Equal(contents[j], data) {
t.Fatalf("expected %s, but found %s", contents[j], data)
}
}
}
})
}
}
func TestIngestLinkFallback(t *testing.T) {
// Verify that ingestLink succeeds if linking fails by falling back to
// copying.
mem := vfs.NewMem()
src, err := mem.Create("source")
require.NoError(t, err)
opts := &Options{FS: errorfs.Wrap(mem, errorfs.ErrInjected.If(errorfs.OnIndex(1)))}
opts.EnsureDefaults().WithFSDefaults()
objSettings := objstorageprovider.DefaultSettings(opts.FS, "")
// Prevent the provider from listing the dir (where we may get an injected error).
objSettings.FSDirInitialListing = []string{}
objProvider, err := objstorageprovider.Open(objSettings)
require.NoError(t, err)
defer objProvider.Close()
meta := []*fileMetadata{{FileNum: 1}}
meta[0].InitPhysicalBacking()
lr := ingestLoadResult{localMeta: meta, localPaths: []string{"source"}}
err = ingestLink(0, opts, objProvider, lr, nil /* shared */)
require.NoError(t, err)
dest, err := mem.Open("000001.sst")
require.NoError(t, err)
// We should be able to write bytes to src, and not have them show up in
// dest.
_, _ = src.Write([]byte("test"))
data, err := io.ReadAll(dest)
require.NoError(t, err)
if len(data) != 0 {
t.Fatalf("expected copy, but files appear to be hard linked: [%s] unexpectedly found", data)
}
}
func TestOverlappingIngestedSSTs(t *testing.T) {
dir := ""
var (
mem vfs.FS
d *DB
opts *Options
closed = false
blockFlush = false
)
defer func() {
if !closed {
require.NoError(t, d.Close())
}
}()
reset := func(strictMem bool) {
if d != nil && !closed {
require.NoError(t, d.Close())
}
blockFlush = false
if strictMem {
mem = vfs.NewStrictMem()
} else {
mem = vfs.NewMem()
}
require.NoError(t, mem.MkdirAll("ext", 0755))
opts = (&Options{
FS: mem,
MemTableStopWritesThreshold: 4,
L0CompactionThreshold: 100,
L0StopWritesThreshold: 100,
DebugCheck: DebugCheckLevels,
FormatMajorVersion: internalFormatNewest,
}).WithFSDefaults()
// Disable automatic compactions because otherwise we'll race with
// delete-only compactions triggered by ingesting range tombstones.
opts.DisableAutomaticCompactions = true
var err error
d, err = Open(dir, opts)
require.NoError(t, err)
d.TestOnlyWaitForCleaning()
}
waitForFlush := func() {
if d == nil {
return
}
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
}
reset(false)
datadriven.RunTest(t, "testdata/flushable_ingest", func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "reset":
reset(td.HasArg("strictMem"))
return ""
case "ignoreSyncs":
var ignoreSyncs bool
if len(td.CmdArgs) == 1 && td.CmdArgs[0].String() == "true" {
ignoreSyncs = true
}
mem.(*vfs.MemFS).SetIgnoreSyncs(ignoreSyncs)
return ""
case "resetToSynced":
mem.(*vfs.MemFS).ResetToSyncedState()
files, err := mem.List(dir)
sort.Strings(files)
require.NoError(t, err)
return strings.Join(files, "\n")
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 "build":
if err := runBuildCmd(td, d, mem); err != nil {
return err.Error()
}
return ""
case "ingest":
if err := runIngestCmd(td, d, mem); err != nil {
return err.Error()
}
if !blockFlush {
waitForFlush()
}
return ""
case "iter":
iter, _ := d.NewIter(nil)
return runIterCmd(td, iter, true)
case "lsm":
return runLSMCmd(td, d)
case "close":
if closed {
return "already closed"
}
require.NoError(t, d.Close())
closed = true
return ""
case "ls":
files, err := mem.List(dir)
sort.Strings(files)
require.NoError(t, err)
return strings.Join(files, "\n")
case "open":
opts.ReadOnly = td.HasArg("readOnly")
var err error
d, err = Open(dir, opts)
closed = false
require.NoError(t, err)
waitForFlush()
d.TestOnlyWaitForCleaning()
return ""
case "blockFlush":
blockFlush = true
d.mu.Lock()
d.mu.compact.flushing = true
d.mu.Unlock()
return ""
case "allowFlush":
blockFlush = false
d.mu.Lock()
d.mu.compact.flushing = false
d.mu.Unlock()
return ""
case "flush":
d.maybeScheduleFlush()
waitForFlush()
d.TestOnlyWaitForCleaning()
return ""
case "get":
return runGetCmd(t, td, d)
default:
return fmt.Sprintf("unknown command: %s", td.Cmd)
}
})
}
func TestExcise(t *testing.T) {
var mem vfs.FS
var d *DB
var flushed bool
defer func() {
require.NoError(t, d.Close())
}()
var opts *Options
reset := func() {
if d != nil {
require.NoError(t, d.Close())
}
mem = vfs.NewMem()
require.NoError(t, mem.MkdirAll("ext", 0755))
opts = &Options{
FS: mem,
L0CompactionThreshold: 100,
L0StopWritesThreshold: 100,
DebugCheck: DebugCheckLevels,
EventListener: &EventListener{FlushEnd: func(info FlushInfo) {
flushed = true
}},
FormatMajorVersion: FormatVirtualSSTables,
Comparer: testkeys.Comparer,
}
// Disable automatic compactions because otherwise we'll race with
// delete-only compactions triggered by ingesting range tombstones.
opts.DisableAutomaticCompactions = true
// Set this to true to add some testing for the virtual sstable validation
// code paths.
opts.Experimental.ValidateOnIngest = true
var err error
d, err = Open("", opts)
require.NoError(t, err)
}
reset()
datadriven.RunTest(t, "testdata/excise", func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "reset":
reset()
return ""
case "reopen":
require.NoError(t, d.Close())
var err error
d, err = Open("", opts)
require.NoError(t, err)
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 "build":
if err := runBuildCmd(td, d, mem); err != nil {
return err.Error()
}
return ""
case "flush":
if err := d.Flush(); err != nil {
return err.Error()
}
return ""
case "ingest":
flushed = false
if err := runIngestCmd(td, d, mem); err != nil {
return err.Error()
}
// Wait for a possible flush.
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
if flushed {
return "memtable flushed"
}
return ""
case "ingest-and-excise":
flushed = false
if err := runIngestAndExciseCmd(td, d, mem); err != nil {
return err.Error()
}
// Wait for a possible flush.
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
if flushed {
return "memtable flushed"
}
return ""
case "get":
return runGetCmd(t, td, d)
case "iter":
iter, _ := d.NewIter(&IterOptions{
KeyTypes: IterKeyTypePointsAndRanges,
})
return runIterCmd(td, iter, true)
case "lsm":
return runLSMCmd(td, d)
case "metrics":
// The asynchronous loading of table stats can change metrics, so
// wait for all the tables' stats to be loaded.
d.mu.Lock()
d.waitTableStats()
d.mu.Unlock()
return d.Metrics().StringForTests()
case "wait-pending-table-stats":
return runTableStatsCmd(td, d)
case "excise":
ve := &versionEdit{
DeletedFiles: map[deletedFileEntry]*fileMetadata{},
}
var exciseSpan KeyRange
if len(td.CmdArgs) != 2 {
panic("insufficient args for compact command")
}
exciseSpan.Start = []byte(td.CmdArgs[0].Key)
exciseSpan.End = []byte(td.CmdArgs[1].Key)
d.mu.Lock()
d.mu.versions.logLock()
d.mu.Unlock()
current := d.mu.versions.currentVersion()
for level := range current.Levels {
iter := current.Levels[level].Iter()
for m := iter.SeekGE(d.cmp, exciseSpan.Start); m != nil && d.cmp(m.Smallest.UserKey, exciseSpan.End) < 0; m = iter.Next() {
_, err := d.excise(exciseSpan, m, ve, level)
if err != nil {
d.mu.Lock()
d.mu.versions.logUnlock()
d.mu.Unlock()
return fmt.Sprintf("error when excising %s: %s", m.FileNum, err.Error())
}
}
}
d.mu.Lock()
d.mu.versions.logUnlock()
d.mu.Unlock()
return fmt.Sprintf("would excise %d files, use ingest-and-excise to excise.\n%s", len(ve.DeletedFiles), ve.DebugString(base.DefaultFormatter))
case "confirm-backing":
// Confirms that the files have the same FileBacking.
fileNums := make(map[base.FileNum]struct{})
for i := range td.CmdArgs {
fNum, err := strconv.Atoi(td.CmdArgs[i].Key)
if err != nil {
panic("invalid file number")
}
fileNums[base.FileNum(fNum)] = struct{}{}
}
d.mu.Lock()
currVersion := d.mu.versions.currentVersion()
var ptr *manifest.FileBacking
for _, level := range currVersion.Levels {
lIter := level.Iter()
for f := lIter.First(); f != nil; f = lIter.Next() {
if _, ok := fileNums[f.FileNum]; ok {
if ptr == nil {
ptr = f.FileBacking
continue
}
if f.FileBacking != ptr {
d.mu.Unlock()
return "file backings are not the same"
}
}
}
}
d.mu.Unlock()
return "file backings are the same"
case "compact":
if len(td.CmdArgs) != 2 {
panic("insufficient args for compact command")
}
l := td.CmdArgs[0].Key
r := td.CmdArgs[1].Key
err := d.Compact([]byte(l), []byte(r), false)
if err != nil {
return err.Error()
}
return ""
default:
return fmt.Sprintf("unknown command: %s", td.Cmd)
}
})
}
func testIngestSharedImpl(
t *testing.T, createOnShared remote.CreateOnSharedStrategy, fileName string,
) {
var d, d1, d2 *DB
var efos map[string]*EventuallyFileOnlySnapshot
defer func() {
for _, e := range efos {
require.NoError(t, e.Close())
}
if d1 != nil {
require.NoError(t, d1.Close())
}
if d2 != nil {
require.NoError(t, d2.Close())
}
}()
creatorIDCounter := uint64(1)
replicateCounter := 1
var opts1, opts2 *Options
reset := func() {
for _, e := range efos {
require.NoError(t, e.Close())
}
if d1 != nil {
require.NoError(t, d1.Close())
}
if d2 != nil {
require.NoError(t, d2.Close())
}
efos = make(map[string]*EventuallyFileOnlySnapshot)
sstorage := remote.NewInMem()
mem1 := vfs.NewMem()
mem2 := vfs.NewMem()
require.NoError(t, mem1.MkdirAll("ext", 0755))
require.NoError(t, mem2.MkdirAll("ext", 0755))
opts1 = &Options{
Comparer: testkeys.Comparer,
FS: mem1,
LBaseMaxBytes: 1,
L0CompactionThreshold: 100,
L0StopWritesThreshold: 100,
DebugCheck: DebugCheckLevels,
FormatMajorVersion: FormatVirtualSSTables,
}
// lel.
lel := MakeLoggingEventListener(DefaultLogger)
opts1.EventListener = &lel
opts1.Experimental.RemoteStorage = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
"": sstorage,
})
opts1.Experimental.CreateOnShared = createOnShared
opts1.Experimental.CreateOnSharedLocator = ""
// Disable automatic compactions because otherwise we'll race with
// delete-only compactions triggered by ingesting range tombstones.
opts1.DisableAutomaticCompactions = true
opts2 = &Options{}
*opts2 = *opts1
opts2.Experimental.RemoteStorage = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
"": sstorage,
})
opts2.Experimental.CreateOnShared = createOnShared
opts2.Experimental.CreateOnSharedLocator = ""
opts2.FS = mem2
var err error
d1, err = Open("", opts1)
require.NoError(t, err)
require.NoError(t, d1.SetCreatorID(creatorIDCounter))
creatorIDCounter++
d2, err = Open("", opts2)
require.NoError(t, err)
require.NoError(t, d2.SetCreatorID(creatorIDCounter))
creatorIDCounter++
d = d1
}
reset()
datadriven.RunTest(t, fmt.Sprintf("testdata/%s", fileName), func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "restart":
for _, e := range efos {
require.NoError(t, e.Close())
}
if d1 != nil {
require.NoError(t, d1.Close())
}
if d2 != nil {
require.NoError(t, d2.Close())
}
var err error
d1, err = Open("", opts1)
if err != nil {
return err.Error()
}
d2, err = Open("", opts2)
if err != nil {
return err.Error()
}
d = d1
return "ok, note that the active db has been set to 1 (use 'switch' to change)"
case "reset":
reset()
return ""
case "switch":
if len(td.CmdArgs) != 1 {
return "usage: switch <1 or 2>"
}
switch td.CmdArgs[0].Key {
case "1":
d = d1
case "2":
d = d2
default:
return "usage: switch <1 or 2>"
}
return "ok"
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 "build":
if err := runBuildCmd(td, d, d.opts.FS); err != nil {
return err.Error()
}
return ""
case "flush":
if err := d.Flush(); err != nil {
return err.Error()
}
return ""
case "ingest":
if err := runIngestCmd(td, d, d.opts.FS); err != nil {
return err.Error()
}
// Wait for a possible flush.
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
return ""
case "ingest-and-excise":
if err := runIngestAndExciseCmd(td, d, d.opts.FS); err != nil {
return err.Error()
}
// Wait for a possible flush.
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
return ""
case "replicate":
if len(td.CmdArgs) != 4 {
return "usage: replicate <from-db-num> <to-db-num> <start-key> <end-key>"
}
var from, to *DB
switch td.CmdArgs[0].Key {
case "1":
from = d1
case "2":
from = d2
default:
return "usage: replicate <from-db-num> <to-db-num> <start-key> <end-key>"
}
switch td.CmdArgs[1].Key {
case "1":
to = d1
case "2":
to = d2
default:
return "usage: replicate <from-db-num> <to-db-num> <start-key> <end-key>"
}
startKey := []byte(td.CmdArgs[2].Key)
endKey := []byte(td.CmdArgs[3].Key)
writeOpts := d.opts.MakeWriterOptions(0 /* level */, to.opts.FormatMajorVersion.MaxTableFormat())
sstPath := fmt.Sprintf("ext/replicate%d.sst", replicateCounter)
f, err := to.opts.FS.Create(sstPath)
require.NoError(t, err)
replicateCounter++
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), writeOpts)
var sharedSSTs []SharedSSTMeta
err = from.ScanInternal(context.TODO(), sstable.CategoryAndQoS{}, startKey, endKey,
func(key *InternalKey, value LazyValue, _ IteratorLevel) error {
val, _, err := value.Value(nil)
require.NoError(t, err)
require.NoError(t, w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val))
return nil
},
func(start, end []byte, seqNum uint64) error {
require.NoError(t, w.DeleteRange(start, end))
return nil
},
func(start, end []byte, keys []keyspan.Key) error {
s := keyspan.Span{
Start: start,
End: end,
Keys: keys,
KeysOrder: 0,
}
require.NoError(t, rangekey.Encode(&s, func(k base.InternalKey, v []byte) error {
return w.AddRangeKey(base.MakeInternalKey(k.UserKey, 0, k.Kind()), v)
}))
return nil
},
func(sst *SharedSSTMeta) error {
sharedSSTs = append(sharedSSTs, *sst)
return nil
},
)
require.NoError(t, err)
require.NoError(t, w.Close())
_, err = to.IngestAndExcise([]string{sstPath}, sharedSSTs, KeyRange{Start: startKey, End: endKey})
require.NoError(t, err)
return fmt.Sprintf("replicated %d shared SSTs", len(sharedSSTs))
case "get":
return runGetCmd(t, td, d)
case "iter":
o := &IterOptions{KeyTypes: IterKeyTypePointsAndRanges}
var reader Reader
reader = d
for _, arg := range td.CmdArgs {
switch arg.Key {
case "mask-suffix":
o.RangeKeyMasking.Suffix = []byte(arg.Vals[0])
case "mask-filter":
o.RangeKeyMasking.Filter = func() BlockPropertyFilterMask {
return sstable.NewTestKeysMaskingFilter()
}
case "snapshot":
reader = efos[arg.Vals[0]]
}
}
iter, err := reader.NewIter(o)
if err != nil {
return err.Error()
}
return runIterCmd(td, iter, true)
case "lsm":
return runLSMCmd(td, d)
case "metrics":
// The asynchronous loading of table stats can change metrics, so
// wait for all the tables' stats to be loaded.
d.mu.Lock()
d.waitTableStats()
d.mu.Unlock()
return d.Metrics().StringForTests()
case "wait-pending-table-stats":
return runTableStatsCmd(td, d)
case "excise":
ve := &versionEdit{
DeletedFiles: map[deletedFileEntry]*fileMetadata{},
}
var exciseSpan KeyRange
if len(td.CmdArgs) != 2 {
panic("insufficient args for excise command")
}
exciseSpan.Start = []byte(td.CmdArgs[0].Key)
exciseSpan.End = []byte(td.CmdArgs[1].Key)
d.mu.Lock()
d.mu.versions.logLock()
d.mu.Unlock()
current := d.mu.versions.currentVersion()
for level := range current.Levels {
iter := current.Levels[level].Iter()
for m := iter.SeekGE(d.cmp, exciseSpan.Start); m != nil && d.cmp(m.Smallest.UserKey, exciseSpan.End) < 0; m = iter.Next() {
_, err := d.excise(exciseSpan, m, ve, level)
if err != nil {
d.mu.Lock()
d.mu.versions.logUnlock()
d.mu.Unlock()
return fmt.Sprintf("error when excising %s: %s", m.FileNum, err.Error())
}
}
}
d.mu.Lock()
d.mu.versions.logUnlock()
d.mu.Unlock()
return fmt.Sprintf("would excise %d files, use ingest-and-excise to excise.\n%s", len(ve.DeletedFiles), ve.String())
case "file-only-snapshot":
if len(td.CmdArgs) != 1 {
panic("insufficient args for file-only-snapshot command")
}
name := td.CmdArgs[0].Key
var keyRanges []KeyRange
for _, line := range strings.Split(td.Input, "\n") {
fields := strings.Fields(line)
if len(fields) != 2 {
return "expected two fields for file-only snapshot KeyRanges"
}
kr := KeyRange{Start: []byte(fields[0]), End: []byte(fields[1])}
keyRanges = append(keyRanges, kr)
}
s := d.NewEventuallyFileOnlySnapshot(keyRanges)
efos[name] = s
return "ok"
case "wait-for-file-only-snapshot":
if len(td.CmdArgs) != 1 {
panic("insufficient args for file-only-snapshot command")
}
name := td.CmdArgs[0].Key
err := efos[name].WaitForFileOnlySnapshot(context.TODO(), 1*time.Millisecond)
if err != nil {
return err.Error()
}
return "ok"
case "compact":
err := runCompactCmd(td, d)
if err != nil {
return err.Error()
}
return "ok"
default:
return fmt.Sprintf("unknown command: %s", td.Cmd)
}
})
}
func TestIngestShared(t *testing.T) {
for _, strategy := range []remote.CreateOnSharedStrategy{remote.CreateOnSharedAll, remote.CreateOnSharedLower} {
strategyStr := "all"
if strategy == remote.CreateOnSharedLower {
strategyStr = "lower"
}
t.Run(fmt.Sprintf("createOnShared=%s", strategyStr), func(t *testing.T) {
fileName := "ingest_shared"
if strategy == remote.CreateOnSharedLower {
fileName = "ingest_shared_lower"
}
testIngestSharedImpl(t, strategy, fileName)
})
}
}
func TestSimpleIngestShared(t *testing.T) {
mem := vfs.NewMem()
var d *DB
var provider2 objstorage.Provider
opts2 := Options{FS: vfs.NewMem(), FormatMajorVersion: FormatVirtualSSTables}
opts2.EnsureDefaults()
// Create an objProvider where we will fake-create some sstables that can
// then be shared back to the db instance.
providerSettings := objstorageprovider.Settings{
Logger: opts2.Logger,
FS: opts2.FS,
FSDirName: "",
FSDirInitialListing: nil,
FSCleaner: opts2.Cleaner,
NoSyncOnClose: opts2.NoSyncOnClose,
BytesPerSync: opts2.BytesPerSync,
}
providerSettings.Remote.StorageFactory = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
"": remote.NewInMem(),
})
providerSettings.Remote.CreateOnShared = remote.CreateOnSharedAll
providerSettings.Remote.CreateOnSharedLocator = ""
provider2, err := objstorageprovider.Open(providerSettings)
require.NoError(t, err)
creatorIDCounter := uint64(1)
provider2.SetCreatorID(objstorage.CreatorID(creatorIDCounter))
creatorIDCounter++
defer func() {
require.NoError(t, d.Close())
}()
reset := func() {
if d != nil {
require.NoError(t, d.Close())
}
mem = vfs.NewMem()
require.NoError(t, mem.MkdirAll("ext", 0755))
opts := &Options{
FormatMajorVersion: FormatVirtualSSTables,
FS: mem,
L0CompactionThreshold: 100,
L0StopWritesThreshold: 100,
}
opts.Experimental.RemoteStorage = providerSettings.Remote.StorageFactory
opts.Experimental.CreateOnShared = providerSettings.Remote.CreateOnShared
opts.Experimental.CreateOnSharedLocator = providerSettings.Remote.CreateOnSharedLocator
var err error
d, err = Open("", opts)
require.NoError(t, err)
require.NoError(t, d.SetCreatorID(creatorIDCounter))
creatorIDCounter++
}
reset()
metaMap := map[base.DiskFileNum]objstorage.ObjectMetadata{}
require.NoError(t, d.Set([]byte("d"), []byte("unexpected"), nil))
require.NoError(t, d.Set([]byte("e"), []byte("unexpected"), nil))
require.NoError(t, d.Set([]byte("a"), []byte("unexpected"), nil))
require.NoError(t, d.Set([]byte("f"), []byte("unexpected"), nil))
d.Flush()
{
// Create a shared file.
fn := base.FileNum(2)
f, meta, err := provider2.Create(context.TODO(), fileTypeTable, fn.DiskFileNum(), objstorage.CreateOptions{PreferSharedStorage: true})
require.NoError(t, err)
w := sstable.NewWriter(f, d.opts.MakeWriterOptions(0, d.opts.FormatMajorVersion.MaxTableFormat()))
w.Set([]byte("d"), []byte("shared"))
w.Set([]byte("e"), []byte("shared"))
w.Close()
metaMap[fn.DiskFileNum()] = meta
}
m := metaMap[base.FileNum(2).DiskFileNum()]
handle, err := provider2.RemoteObjectBacking(&m)
require.NoError(t, err)
size, err := provider2.Size(m)
require.NoError(t, err)
sharedSSTMeta := SharedSSTMeta{
Backing: handle,
Smallest: base.MakeInternalKey([]byte("d"), 0, InternalKeyKindSet),
Largest: base.MakeInternalKey([]byte("e"), 0, InternalKeyKindSet),
SmallestPointKey: base.MakeInternalKey([]byte("d"), 0, InternalKeyKindSet),
LargestPointKey: base.MakeInternalKey([]byte("e"), 0, InternalKeyKindSet),
Level: 6,
Size: uint64(size + 5),
}
_, err = d.IngestAndExcise([]string{}, []SharedSSTMeta{sharedSSTMeta}, KeyRange{Start: []byte("d"), End: []byte("ee")})
require.NoError(t, err)
// TODO(bilal): Once reading of shared sstables is in, verify that the values
// of d and e have been updated.
}
type blockedCompaction struct {
startBlock, unblock chan struct{}
}
func TestConcurrentExcise(t *testing.T) {
var d, d1, d2 *DB
var efos map[string]*EventuallyFileOnlySnapshot
backgroundErrs := make(chan error, 5)
var compactions map[string]*blockedCompaction
defer func() {
for _, e := range efos {
require.NoError(t, e.Close())
}
if d1 != nil {
require.NoError(t, d1.Close())
}
if d2 != nil {
require.NoError(t, d2.Close())
}
}()
creatorIDCounter := uint64(1)
replicateCounter := 1
var wg sync.WaitGroup
defer wg.Wait()
var blockNextCompaction bool
var blockedJobID int
var blockedCompactionName string
var blockedCompactionsMu sync.Mutex // protects the above three variables.
reset := func() {
wg.Wait()
for _, e := range efos {
require.NoError(t, e.Close())
}
if d1 != nil {
require.NoError(t, d1.Close())
}
if d2 != nil {
require.NoError(t, d2.Close())
}
efos = make(map[string]*EventuallyFileOnlySnapshot)
compactions = make(map[string]*blockedCompaction)
backgroundErrs = make(chan error, 5)
var el EventListener
el.EnsureDefaults(testLogger{t: t})
el.FlushBegin = func(info FlushInfo) {
// Don't block flushes
}
el.BackgroundError = func(err error) {
backgroundErrs <- err
}
el.CompactionBegin = func(info CompactionInfo) {
if info.Reason == "move" {
return
}
blockedCompactionsMu.Lock()
defer blockedCompactionsMu.Unlock()
if blockNextCompaction {
blockNextCompaction = false
blockedJobID = info.JobID
}
}
el.TableCreated = func(info TableCreateInfo) {
blockedCompactionsMu.Lock()
if info.JobID != blockedJobID {
blockedCompactionsMu.Unlock()
return
}
blockedJobID = 0
c := compactions[blockedCompactionName]
blockedCompactionName = ""
blockedCompactionsMu.Unlock()
c.startBlock <- struct{}{}
<-c.unblock
}
sstorage := remote.NewInMem()
mem1 := vfs.NewMem()
mem2 := vfs.NewMem()
require.NoError(t, mem1.MkdirAll("ext", 0755))
require.NoError(t, mem2.MkdirAll("ext", 0755))
opts1 := &Options{
Comparer: testkeys.Comparer,
LBaseMaxBytes: 1,
FS: mem1,
L0CompactionThreshold: 100,
L0StopWritesThreshold: 100,
DebugCheck: DebugCheckLevels,
FormatMajorVersion: FormatVirtualSSTables,
}
// lel.
lel := MakeLoggingEventListener(DefaultLogger)
tel := TeeEventListener(lel, el)
opts1.EventListener = &tel
opts1.Experimental.RemoteStorage = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
"": sstorage,
})
opts1.Experimental.CreateOnShared = remote.CreateOnSharedAll
opts1.Experimental.CreateOnSharedLocator = ""
// Disable automatic compactions because otherwise we'll race with
// delete-only compactions triggered by ingesting range tombstones.
opts1.DisableAutomaticCompactions = true
opts2 := &Options{}
*opts2 = *opts1
opts2.Experimental.RemoteStorage = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
"": sstorage,
})
opts2.Experimental.CreateOnShared = remote.CreateOnSharedAll
opts2.Experimental.CreateOnSharedLocator = ""
opts2.FS = mem2
var err error
d1, err = Open("", opts1)
require.NoError(t, err)
require.NoError(t, d1.SetCreatorID(creatorIDCounter))
creatorIDCounter++
d2, err = Open("", opts2)
require.NoError(t, err)
require.NoError(t, d2.SetCreatorID(creatorIDCounter))
creatorIDCounter++
d = d1
}
reset()
datadriven.RunTest(t, "testdata/concurrent_excise", func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "reset":
reset()
return ""
case "switch":
if len(td.CmdArgs) != 1 {
return "usage: switch <1 or 2>"
}
switch td.CmdArgs[0].Key {
case "1":
d = d1
case "2":
d = d2
default:
return "usage: switch <1 or 2>"
}
return "ok"
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 "build":
if err := runBuildCmd(td, d, d.opts.FS); err != nil {
return err.Error()
}
return ""
case "flush":
if err := d.Flush(); err != nil {
return err.Error()
}
return ""
case "ingest":
if err := runIngestCmd(td, d, d.opts.FS); err != nil {
return err.Error()
}
// Wait for a possible flush.
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
return ""
case "ingest-and-excise":
if err := runIngestAndExciseCmd(td, d, d.opts.FS); err != nil {
return err.Error()
}
// Wait for a possible flush.
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
return ""
case "replicate":
if len(td.CmdArgs) != 4 {
return "usage: replicate <from-db-num> <to-db-num> <start-key> <end-key>"
}
var from, to *DB
switch td.CmdArgs[0].Key {
case "1":
from = d1
case "2":
from = d2
default:
return "usage: replicate <from-db-num> <to-db-num> <start-key> <end-key>"
}
switch td.CmdArgs[1].Key {
case "1":
to = d1
case "2":
to = d2
default:
return "usage: replicate <from-db-num> <to-db-num> <start-key> <end-key>"
}
startKey := []byte(td.CmdArgs[2].Key)
endKey := []byte(td.CmdArgs[3].Key)
writeOpts := d.opts.MakeWriterOptions(0 /* level */, to.opts.FormatMajorVersion.MaxTableFormat())
sstPath := fmt.Sprintf("ext/replicate%d.sst", replicateCounter)
f, err := to.opts.FS.Create(sstPath)
require.NoError(t, err)
replicateCounter++
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), writeOpts)
var sharedSSTs []SharedSSTMeta
err = from.ScanInternal(context.TODO(), sstable.CategoryAndQoS{}, startKey, endKey,
func(key *InternalKey, value LazyValue, _ IteratorLevel) error {
val, _, err := value.Value(nil)
require.NoError(t, err)
require.NoError(t, w.Add(base.MakeInternalKey(key.UserKey, 0, key.Kind()), val))
return nil
},
func(start, end []byte, seqNum uint64) error {
require.NoError(t, w.DeleteRange(start, end))
return nil
},
func(start, end []byte, keys []keyspan.Key) error {
s := keyspan.Span{
Start: start,
End: end,
Keys: keys,
KeysOrder: 0,
}
require.NoError(t, rangekey.Encode(&s, func(k base.InternalKey, v []byte) error {
return w.AddRangeKey(base.MakeInternalKey(k.UserKey, 0, k.Kind()), v)
}))
return nil
},
func(sst *SharedSSTMeta) error {
sharedSSTs = append(sharedSSTs, *sst)
return nil
},
)
require.NoError(t, err)
require.NoError(t, w.Close())
_, err = to.IngestAndExcise([]string{sstPath}, sharedSSTs, KeyRange{Start: startKey, End: endKey})
require.NoError(t, err)
return fmt.Sprintf("replicated %d shared SSTs", len(sharedSSTs))
case "get":
return runGetCmd(t, td, d)
case "iter":
o := &IterOptions{KeyTypes: IterKeyTypePointsAndRanges}
var reader Reader
reader = d
for _, arg := range td.CmdArgs {
switch arg.Key {
case "mask-suffix":
o.RangeKeyMasking.Suffix = []byte(arg.Vals[0])
case "mask-filter":
o.RangeKeyMasking.Filter = func() BlockPropertyFilterMask {
return sstable.NewTestKeysMaskingFilter()
}
case "snapshot":
reader = efos[arg.Vals[0]]
}
}
iter, err := reader.NewIter(o)
if err != nil {
return err.Error()
}
return runIterCmd(td, iter, true)
case "lsm":
return runLSMCmd(td, d)
case "metrics":
// The asynchronous loading of table stats can change metrics, so
// wait for all the tables' stats to be loaded.
d.mu.Lock()
d.waitTableStats()
d.mu.Unlock()
return d.Metrics().StringForTests()
case "wait-pending-table-stats":
return runTableStatsCmd(td, d)
case "excise":
ve := &versionEdit{
DeletedFiles: map[deletedFileEntry]*fileMetadata{},
}
var exciseSpan KeyRange
if len(td.CmdArgs) != 2 {
panic("insufficient args for excise command")
}
exciseSpan.Start = []byte(td.CmdArgs[0].Key)
exciseSpan.End = []byte(td.CmdArgs[1].Key)
d.mu.Lock()
d.mu.versions.logLock()
d.mu.Unlock()
current := d.mu.versions.currentVersion()
for level := range current.Levels {
iter := current.Levels[level].Iter()
for m := iter.SeekGE(d.cmp, exciseSpan.Start); m != nil && d.cmp(m.Smallest.UserKey, exciseSpan.End) < 0; m = iter.Next() {
_, err := d.excise(exciseSpan, m, ve, level)
if err != nil {
d.mu.Lock()
d.mu.versions.logUnlock()
d.mu.Unlock()
return fmt.Sprintf("error when excising %s: %s", m.FileNum, err.Error())
}
}
}
d.mu.Lock()
d.mu.versions.logUnlock()
d.mu.Unlock()
return fmt.Sprintf("would excise %d files, use ingest-and-excise to excise.\n%s", len(ve.DeletedFiles), ve.String())
case "file-only-snapshot":
if len(td.CmdArgs) != 1 {
panic("insufficient args for file-only-snapshot command")
}
name := td.CmdArgs[0].Key
var keyRanges []KeyRange
for _, line := range strings.Split(td.Input, "\n") {
fields := strings.Fields(line)
if len(fields) != 2 {
return "expected two fields for file-only snapshot KeyRanges"
}
kr := KeyRange{Start: []byte(fields[0]), End: []byte(fields[1])}
keyRanges = append(keyRanges, kr)
}
s := d.NewEventuallyFileOnlySnapshot(keyRanges)
efos[name] = s
return "ok"
case "wait-for-file-only-snapshot":
if len(td.CmdArgs) != 1 {
panic("insufficient args for file-only-snapshot command")
}
name := td.CmdArgs[0].Key
err := efos[name].WaitForFileOnlySnapshot(context.TODO(), 1*time.Millisecond)
if err != nil {
return err.Error()
}
return "ok"
case "unblock":
name := td.CmdArgs[0].Key
blockedCompactionsMu.Lock()
c := compactions[name]
delete(compactions, name)
blockedCompactionsMu.Unlock()
c.unblock <- struct{}{}
return "ok"
case "compact":
async := false
var otherArgs []datadriven.CmdArg
var bc *blockedCompaction
for i := range td.CmdArgs {
switch td.CmdArgs[i].Key {
case "block":
name := td.CmdArgs[i].Vals[0]
bc = &blockedCompaction{startBlock: make(chan struct{}), unblock: make(chan struct{})}
blockedCompactionsMu.Lock()
compactions[name] = bc
blockNextCompaction = true
blockedCompactionName = name
blockedCompactionsMu.Unlock()
async = true
default:
otherArgs = append(otherArgs, td.CmdArgs[i])
}
}
var tdClone datadriven.TestData
tdClone = *td
tdClone.CmdArgs = otherArgs
if !async {
err := runCompactCmd(td, d)
if err != nil {
return err.Error()
}
} else {
wg.Add(1)
go func() {
defer wg.Done()
_ = runCompactCmd(&tdClone, d)
}()
<-bc.startBlock
return "spun off in separate goroutine"
}
return "ok"
case "wait-for-background-error":
err := <-backgroundErrs
return err.Error()
default:
return fmt.Sprintf("unknown command: %s", td.Cmd)
}
})
}
func TestIngestExternal(t *testing.T) {
var mem vfs.FS
var d *DB
var flushed bool
defer func() {
require.NoError(t, d.Close())
}()
var remoteStorage remote.Storage
reset := func() {
if d != nil {
require.NoError(t, d.Close())
}
mem = vfs.NewMem()
require.NoError(t, mem.MkdirAll("ext", 0755))
remoteStorage = remote.NewInMem()
opts := &Options{
FS: mem,
L0CompactionThreshold: 100,
L0StopWritesThreshold: 100,
DebugCheck: DebugCheckLevels,
EventListener: &EventListener{FlushEnd: func(info FlushInfo) {
flushed = true
}},
FormatMajorVersion: FormatVirtualSSTables,
}
opts.Experimental.RemoteStorage = remote.MakeSimpleFactory(map[remote.Locator]remote.Storage{
"external-locator": remoteStorage,
})
opts.Experimental.CreateOnShared = remote.CreateOnSharedNone
// Disable automatic compactions because otherwise we'll race with
// delete-only compactions triggered by ingesting range tombstones.
opts.DisableAutomaticCompactions = true
var err error
d, err = Open("", opts)
require.NoError(t, err)
require.NoError(t, d.SetCreatorID(1))
}
reset()
datadriven.RunTest(t, "testdata/ingest_external", func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "reset":
reset()
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 "build-remote":
if err := runBuildRemoteCmd(td, d, remoteStorage); err != nil {
return err.Error()
}
return ""
case "flush":
if err := d.Flush(); err != nil {
return err.Error()
}
return ""
case "ingest-external":
flushed = false
if err := runIngestExternalCmd(td, d, "external-locator"); err != nil {
return err.Error()
}
// Wait for a possible flush.
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
if flushed {
return "memtable flushed"
}
return ""
case "get":
return runGetCmd(t, td, d)
case "iter":
iter, _ := d.NewIter(&IterOptions{
KeyTypes: IterKeyTypePointsAndRanges,
})
return runIterCmd(td, iter, true)
case "lsm":
return runLSMCmd(td, d)
case "metrics":
// The asynchronous loading of table stats can change metrics, so
// wait for all the tables' stats to be loaded.
d.mu.Lock()
d.waitTableStats()
d.mu.Unlock()
return d.Metrics().StringForTests()
case "wait-pending-table-stats":
return runTableStatsCmd(td, d)
case "compact":
if len(td.CmdArgs) != 2 {
panic("insufficient args for compact command")
}
l := td.CmdArgs[0].Key
r := td.CmdArgs[1].Key
err := d.Compact([]byte(l), []byte(r), false)
if err != nil {
return err.Error()
}
return ""
default:
return fmt.Sprintf("unknown command: %s", td.Cmd)
}
})
}
func TestIngestMemtableOverlaps(t *testing.T) {
comparers := []Comparer{
{Name: "default", Compare: DefaultComparer.Compare, FormatKey: DefaultComparer.FormatKey},
{
Name: "reverse",
Compare: func(a, b []byte) int { return DefaultComparer.Compare(b, a) },
FormatKey: DefaultComparer.FormatKey,
},
}
m := make(map[string]*Comparer)
for i := range comparers {
c := &comparers[i]
m[c.Name] = c
}
for _, comparer := range comparers {
t.Run(comparer.Name, func(t *testing.T) {
var mem *memTable
parseMeta := func(s string) *fileMetadata {
parts := strings.Split(s, "-")
meta := &fileMetadata{}
if len(parts) != 2 {
t.Fatalf("malformed table spec: %s", s)
}
var smallest, largest base.InternalKey
if strings.Contains(parts[0], ".") {
if !strings.Contains(parts[1], ".") {
t.Fatalf("malformed table spec: %s", s)
}
smallest = base.ParseInternalKey(parts[0])
largest = base.ParseInternalKey(parts[1])
} else {
smallest = InternalKey{UserKey: []byte(parts[0])}
largest = InternalKey{UserKey: []byte(parts[1])}
}
// If we're using a reverse comparer, flip the file bounds.
if mem.cmp(smallest.UserKey, largest.UserKey) > 0 {
smallest, largest = largest, smallest
}
meta.ExtendPointKeyBounds(comparer.Compare, smallest, largest)
meta.InitPhysicalBacking()
return meta
}
datadriven.RunTest(t, "testdata/ingest_memtable_overlaps", func(t *testing.T, d *datadriven.TestData) string {
switch d.Cmd {
case "define":
b := newBatch(nil)
if err := runBatchDefineCmd(d, b); err != nil {
return err.Error()
}
opts := &Options{
Comparer: &comparer,
}
opts.EnsureDefaults().WithFSDefaults()
if len(d.CmdArgs) > 1 {
return fmt.Sprintf("%s expects at most 1 argument", d.Cmd)
}
if len(d.CmdArgs) == 1 {
opts.Comparer = m[d.CmdArgs[0].String()]
if opts.Comparer == nil {
return fmt.Sprintf("%s unknown comparer: %s", d.Cmd, d.CmdArgs[0].String())
}
}
mem = newMemTable(memTableOptions{Options: opts})
if err := mem.apply(b, 0); err != nil {
return err.Error()
}
return ""
case "overlaps":
var buf bytes.Buffer
for _, data := range strings.Split(d.Input, "\n") {
var keyRanges []internalKeyRange
for _, part := range strings.Fields(data) {
meta := parseMeta(part)
keyRanges = append(keyRanges, internalKeyRange{smallest: meta.Smallest, largest: meta.Largest})
}
fmt.Fprintf(&buf, "%t\n", ingestMemtableOverlaps(mem.cmp, mem, keyRanges))
}
return buf.String()
default:
return fmt.Sprintf("unknown command: %s", d.Cmd)
}
})
})
}
}
func TestKeyRangeBasic(t *testing.T) {
cmp := base.DefaultComparer.Compare
k1 := KeyRange{Start: []byte("b"), End: []byte("c")}
// Tests for Contains()
require.True(t, k1.Contains(cmp, base.MakeInternalKey([]byte("b"), 1, InternalKeyKindSet)))
require.False(t, k1.Contains(cmp, base.MakeInternalKey([]byte("c"), 1, InternalKeyKindSet)))
require.True(t, k1.Contains(cmp, base.MakeInternalKey([]byte("bb"), 1, InternalKeyKindSet)))
require.True(t, k1.Contains(cmp, base.MakeExclusiveSentinelKey(InternalKeyKindRangeDelete, []byte("c"))))
m1 := &fileMetadata{
Smallest: base.MakeInternalKey([]byte("b"), 1, InternalKeyKindSet),
Largest: base.MakeInternalKey([]byte("c"), 1, InternalKeyKindSet),
}
require.True(t, k1.Overlaps(cmp, m1))
m2 := &fileMetadata{
Smallest: base.MakeInternalKey([]byte("c"), 1, InternalKeyKindSet),
Largest: base.MakeInternalKey([]byte("d"), 1, InternalKeyKindSet),
}
require.False(t, k1.Overlaps(cmp, m2))
m3 := &fileMetadata{
Smallest: base.MakeInternalKey([]byte("a"), 1, InternalKeyKindSet),
Largest: base.MakeExclusiveSentinelKey(InternalKeyKindRangeDelete, []byte("b")),
}
require.False(t, k1.Overlaps(cmp, m3))
m4 := &fileMetadata{
Smallest: base.MakeInternalKey([]byte("a"), 1, InternalKeyKindSet),
Largest: base.MakeInternalKey([]byte("b"), 1, InternalKeyKindSet),
}
require.True(t, k1.Overlaps(cmp, m4))
}
func BenchmarkIngestOverlappingMemtable(b *testing.B) {
assertNoError := func(err error) {
b.Helper()
if err != nil {
b.Fatal(err)
}
}
for count := 1; count < 6; count++ {
b.Run(fmt.Sprintf("memtables=%d", count), func(b *testing.B) {
for i := 0; i < b.N; i++ {
b.StopTimer()
mem := vfs.NewMem()
d, err := Open("", &Options{
FS: mem,
})
assertNoError(err)
// Create memtables.
for {
assertNoError(d.Set([]byte("a"), nil, nil))
d.mu.Lock()
done := len(d.mu.mem.queue) == count
d.mu.Unlock()
if done {
break
}
}
// Create the overlapping sstable that will force a flush when ingested.
f, err := mem.Create("ext")
assertNoError(err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
assertNoError(w.Set([]byte("a"), nil))
assertNoError(w.Close())
b.StartTimer()
assertNoError(d.Ingest([]string{"ext"}))
}
})
}
}
func TestIngestTargetLevel(t *testing.T) {
var d *DB
defer func() {
if d != nil {
// Ignore errors because this test defines fake in-progress transactions
// that prohibit clean shutdown.
_ = d.Close()
}
}()
parseMeta := func(s string) *fileMetadata {
var rkey bool
if len(s) >= 4 && s[0:4] == "rkey" {
rkey = true
s = s[5:]
}
parts := strings.Split(s, "-")
if len(parts) != 2 {
t.Fatalf("malformed table spec: %s", s)
}
var m *fileMetadata
if rkey {
m = (&fileMetadata{}).ExtendRangeKeyBounds(
d.cmp,
InternalKey{UserKey: []byte(parts[0])},
InternalKey{UserKey: []byte(parts[1])},
)
} else {
m = (&fileMetadata{}).ExtendPointKeyBounds(
d.cmp,
InternalKey{UserKey: []byte(parts[0])},
InternalKey{UserKey: []byte(parts[1])},
)
}
m.InitPhysicalBacking()
return m
}
datadriven.RunTest(t, "testdata/ingest_target_level", func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "define":
if d != nil {
// Ignore errors because this test defines fake in-progress
// transactions that prohibit clean shutdown.
_ = d.Close()
}
var err error
opts := Options{
FormatMajorVersion: internalFormatNewest,
}
opts.WithFSDefaults()
if d, err = runDBDefineCmd(td, &opts); err != nil {
return err.Error()
}
readState := d.loadReadState()
c := &checkConfig{
logger: d.opts.Logger,
comparer: d.opts.Comparer,
readState: readState,
newIters: d.newIters,
// TODO: runDBDefineCmd doesn't properly update the visible
// sequence number. So we have to explicitly configure level checker with a very large
// sequence number, otherwise the DB appears empty.
seqNum: InternalKeySeqNumMax,
}
if err := checkLevelsInternal(c); err != nil {
return err.Error()
}
readState.unref()
d.mu.Lock()
s := d.mu.versions.currentVersion().String()
d.mu.Unlock()
return s
case "target":
var buf bytes.Buffer
suggestSplit := false
for _, cmd := range td.CmdArgs {
switch cmd.Key {
case "suggest-split":
suggestSplit = true
}
}
for _, target := range strings.Split(td.Input, "\n") {
meta := parseMeta(target)
level, overlapFile, err := ingestTargetLevel(
d.newIters, d.tableNewRangeKeyIter, IterOptions{logger: d.opts.Logger},
d.opts.Comparer, d.mu.versions.currentVersion(), 1, d.mu.compact.inProgress, meta,
suggestSplit)
if err != nil {
return err.Error()
}
if overlapFile != nil {
fmt.Fprintf(&buf, "%d (split file: %s)\n", level, overlapFile.FileNum)
} else {
fmt.Fprintf(&buf, "%d\n", level)
}
}
return buf.String()
default:
return fmt.Sprintf("unknown command: %s", td.Cmd)
}
})
}
func TestIngest(t *testing.T) {
var mem vfs.FS
var d *DB
var flushed bool
defer func() {
require.NoError(t, d.Close())
}()
reset := func(split bool) {
if d != nil {
require.NoError(t, d.Close())
}
mem = vfs.NewMem()
require.NoError(t, mem.MkdirAll("ext", 0755))
opts := &Options{
FS: mem,
L0CompactionThreshold: 100,
L0StopWritesThreshold: 100,
DebugCheck: DebugCheckLevels,
EventListener: &EventListener{FlushEnd: func(info FlushInfo) {
flushed = true
}},
FormatMajorVersion: internalFormatNewest,
}
opts.Experimental.IngestSplit = func() bool {
return split
}
// Disable automatic compactions because otherwise we'll race with
// delete-only compactions triggered by ingesting range tombstones.
opts.DisableAutomaticCompactions = true
var err error
d, err = Open("", opts)
require.NoError(t, err)
}
reset(false /* split */)
datadriven.RunTest(t, "testdata/ingest", func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "reset":
split := false
for _, cmd := range td.CmdArgs {
switch cmd.Key {
case "enable-split":
split = true
default:
return fmt.Sprintf("unexpected key: %s", cmd.Key)
}
}
reset(split)
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 "build":
if err := runBuildCmd(td, d, mem); err != nil {
return err.Error()
}
return ""
case "ingest":
flushed = false
if err := runIngestCmd(td, d, mem); err != nil {
return err.Error()
}
// Wait for a possible flush.
d.mu.Lock()
for d.mu.compact.flushing {
d.mu.compact.cond.Wait()
}
d.mu.Unlock()
if flushed {
return "memtable flushed"
}
return ""
case "get":
return runGetCmd(t, td, d)
case "iter":
iter, _ := d.NewIter(&IterOptions{
KeyTypes: IterKeyTypePointsAndRanges,
})
return runIterCmd(td, iter, true)
case "lsm":
return runLSMCmd(td, d)
case "metrics":
// The asynchronous loading of table stats can change metrics, so
// wait for all the tables' stats to be loaded.
d.mu.Lock()
d.waitTableStats()
d.mu.Unlock()
return d.Metrics().StringForTests()
case "wait-pending-table-stats":
return runTableStatsCmd(td, d)
case "compact":
if len(td.CmdArgs) != 2 {
panic("insufficient args for compact command")
}
l := td.CmdArgs[0].Key
r := td.CmdArgs[1].Key
err := d.Compact([]byte(l), []byte(r), false)
if err != nil {
return err.Error()
}
return ""
default:
return fmt.Sprintf("unknown command: %s", td.Cmd)
}
})
}
func TestIngestError(t *testing.T) {
for i := int32(0); ; i++ {
mem := vfs.NewMem()
f0, err := mem.Create("ext0")
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f0), sstable.WriterOptions{})
require.NoError(t, w.Set([]byte("d"), nil))
require.NoError(t, w.Close())
f1, err := mem.Create("ext1")
require.NoError(t, err)
w = sstable.NewWriter(objstorageprovider.NewFileWritable(f1), sstable.WriterOptions{})
require.NoError(t, w.Set([]byte("d"), nil))
require.NoError(t, w.Close())
ii := errorfs.OnIndex(-1)
d, err := Open("", &Options{
FS: errorfs.Wrap(mem, errorfs.ErrInjected.If(ii)),
Logger: panicLogger{},
L0CompactionThreshold: 8,
})
require.NoError(t, err)
// Force the creation of an L0 sstable that overlaps with the tables
// we'll attempt to ingest. This ensures that we exercise filesystem
// codepaths when determining the ingest target level.
require.NoError(t, d.Set([]byte("a"), nil, nil))
require.NoError(t, d.Set([]byte("d"), nil, nil))
require.NoError(t, d.Flush())
t.Run(fmt.Sprintf("index-%d", i), func(t *testing.T) {
defer func() {
if r := recover(); r != nil {
if e, ok := r.(error); ok && errors.Is(e, errorfs.ErrInjected) {
return
}
// d.opts.Logger.Fatalf won't propagate ErrInjected
// itself, but should contain the error message.
if strings.HasSuffix(fmt.Sprint(r), errorfs.ErrInjected.Error()) {
return
}
t.Fatal(r)
}
}()
ii.Store(i)
err1 := d.Ingest([]string{"ext0"})
err2 := d.Ingest([]string{"ext1"})
err := firstError(err1, err2)
if err != nil && !errors.Is(err, errorfs.ErrInjected) {
t.Fatal(err)
}
})
// d.Close may error if we failed to flush the manifest.
_ = d.Close()
// If the injector's index is non-negative, the i-th filesystem
// operation was never executed.
if ii.Load() >= 0 {
break
}
}
}
func TestIngestIdempotence(t *testing.T) {
// Use an on-disk filesystem, because Ingest with a MemFS will copy, not
// link the ingested file.
dir, err := os.MkdirTemp("", "ingest-idempotence")
require.NoError(t, err)
defer os.RemoveAll(dir)
fs := vfs.Default
path := fs.PathJoin(dir, "ext")
f, err := fs.Create(fs.PathJoin(dir, "ext"))
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
require.NoError(t, w.Set([]byte("d"), nil))
require.NoError(t, w.Close())
d, err := Open(dir, &Options{
FS: fs,
})
require.NoError(t, err)
const count = 4
for i := 0; i < count; i++ {
ingestPath := fs.PathJoin(dir, fmt.Sprintf("ext%d", i))
require.NoError(t, fs.Link(path, ingestPath))
require.NoError(t, d.Ingest([]string{ingestPath}))
}
require.NoError(t, d.Close())
}
func TestIngestCompact(t *testing.T) {
mem := vfs.NewMem()
lel := MakeLoggingEventListener(&base.InMemLogger{})
d, err := Open("", &Options{
EventListener: &lel,
FS: mem,
L0CompactionThreshold: 1,
L0StopWritesThreshold: 1,
})
require.NoError(t, err)
src := func(i int) string {
return fmt.Sprintf("ext%d", i)
}
f, err := mem.Create(src(0))
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
key := []byte("a")
require.NoError(t, w.Add(base.MakeInternalKey(key, 0, InternalKeyKindSet), nil))
require.NoError(t, w.Close())
// Make N copies of the sstable.
const count = 20
for i := 1; i < count; i++ {
require.NoError(t, vfs.Copy(d.opts.FS, src(0), src(i)))
}
// Ingest the same sstable multiple times. Compaction should take place as
// ingestion happens, preventing an indefinite write stall from occurring.
for i := 0; i < count; i++ {
if i == 10 {
// Half-way through the ingestions, set a key in the memtable to force
// overlap with the memtable which will require the memtable to be
// flushed.
require.NoError(t, d.Set(key, nil, nil))
}
require.NoError(t, d.Ingest([]string{src(i)}))
}
require.NoError(t, d.Close())
}
func TestConcurrentIngest(t *testing.T) {
mem := vfs.NewMem()
d, err := Open("", &Options{
FS: mem,
})
require.NoError(t, err)
// Create an sstable with 2 keys. This is necessary to trigger the overlap
// bug because an sstable with a single key will not have overlap in internal
// key space and the sequence number assignment had already guaranteed
// correct ordering.
src := func(i int) string {
return fmt.Sprintf("ext%d", i)
}
f, err := mem.Create(src(0))
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
require.NoError(t, w.Set([]byte("a"), nil))
require.NoError(t, w.Set([]byte("b"), nil))
require.NoError(t, w.Close())
// Make N copies of the sstable.
errCh := make(chan error, 5)
for i := 1; i < cap(errCh); i++ {
require.NoError(t, vfs.Copy(d.opts.FS, src(0), src(i)))
}
// Perform N ingestions concurrently.
for i := 0; i < cap(errCh); i++ {
go func(i int) {
err := d.Ingest([]string{src(i)})
if err == nil {
if _, err = d.opts.FS.Stat(src(i)); oserror.IsNotExist(err) {
err = nil
}
}
errCh <- err
}(i)
}
for i := 0; i < cap(errCh); i++ {
require.NoError(t, <-errCh)
}
require.NoError(t, d.Close())
}
func TestConcurrentIngestCompact(t *testing.T) {
for i := 0; i < 2; i++ {
t.Run("", func(t *testing.T) {
mem := vfs.NewMem()
compactionReady := make(chan struct{})
compactionBegin := make(chan struct{})
d, err := Open("", &Options{
FS: mem,
EventListener: &EventListener{
TableCreated: func(info TableCreateInfo) {
if info.Reason == "compacting" {
close(compactionReady)
<-compactionBegin
}
},
},
})
require.NoError(t, err)
ingest := func(keys ...string) {
t.Helper()
f, err := mem.Create("ext")
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
for _, k := range keys {
require.NoError(t, w.Set([]byte(k), nil))
}
require.NoError(t, w.Close())
require.NoError(t, d.Ingest([]string{"ext"}))
}
compact := func(start, end string) {
t.Helper()
require.NoError(t, d.Compact([]byte(start), []byte(end), false))
}
lsm := func() string {
d.mu.Lock()
s := d.mu.versions.currentVersion().String()
d.mu.Unlock()
return s
}
expectLSM := func(expected string) {
t.Helper()
expected = strings.TrimSpace(expected)
actual := strings.TrimSpace(lsm())
if expected != actual {
t.Fatalf("expected\n%s\nbut found\n%s", expected, actual)
}
}
ingest("a")
ingest("a")
ingest("c")
ingest("c")
expectLSM(`
0.0:
000005:[a#11,SET-a#11,SET]
000007:[c#13,SET-c#13,SET]
6:
000004:[a#10,SET-a#10,SET]
000006:[c#12,SET-c#12,SET]
`)
// At this point ingestion of an sstable containing only key "b" will be
// targeted at L6. Yet a concurrent compaction of sstables 5 and 7 will
// create a new sstable in L6 spanning ["a"-"c"]. So the ingestion must
// actually target L5.
switch i {
case 0:
// Compact, then ingest.
go func() {
<-compactionReady
ingest("b")
close(compactionBegin)
}()
compact("a", "z")
expectLSM(`
0.0:
000009:[b#14,SET-b#14,SET]
6:
000008:[a#0,SET-c#0,SET]
`)
case 1:
// Ingest, then compact
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
close(compactionBegin)
compact("a", "z")
}()
ingest("b")
wg.Wait()
// Because we're performing the ingestion and compaction concurrently,
// we can't guarantee any particular LSM structure at this point. The
// test will fail with an assertion error due to overlapping sstables
// if there is insufficient synchronization between ingestion and
// compaction.
}
require.NoError(t, d.Close())
})
}
}
func TestIngestFlushQueuedMemTable(t *testing.T) {
// Verify that ingestion forces a flush of a queued memtable.
// Test with a format major version prior to FormatFlushableIngest and one
// after. Both should result in the same statistic calculations.
for _, fmv := range []FormatMajorVersion{FormatFlushableIngest - 1, internalFormatNewest} {
func(fmv FormatMajorVersion) {
mem := vfs.NewMem()
d, err := Open("", &Options{
FS: mem,
FormatMajorVersion: fmv,
})
require.NoError(t, err)
// Add the key "a" to the memtable, then fill up the memtable with the key
// "b". The ingested sstable will only overlap with the queued memtable.
require.NoError(t, d.Set([]byte("a"), nil, nil))
for {
require.NoError(t, d.Set([]byte("b"), nil, nil))
d.mu.Lock()
done := len(d.mu.mem.queue) == 2
d.mu.Unlock()
if done {
break
}
}
ingest := func(keys ...string) {
t.Helper()
f, err := mem.Create("ext")
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
TableFormat: fmv.MinTableFormat(),
})
for _, k := range keys {
require.NoError(t, w.Set([]byte(k), nil))
}
require.NoError(t, w.Close())
stats, err := d.IngestWithStats([]string{"ext"})
require.NoError(t, err)
require.Equal(t, stats.ApproxIngestedIntoL0Bytes, stats.Bytes)
require.Equal(t, stats.MemtableOverlappingFiles, 1)
require.Less(t, uint64(0), stats.Bytes)
}
ingest("a")
require.NoError(t, d.Close())
}(fmv)
}
}
func TestIngestStats(t *testing.T) {
mem := vfs.NewMem()
d, err := Open("", &Options{
FS: mem,
})
require.NoError(t, err)
ingest := func(expectedLevel int, keys ...string) {
t.Helper()
f, err := mem.Create("ext")
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
for _, k := range keys {
require.NoError(t, w.Set([]byte(k), nil))
}
require.NoError(t, w.Close())
stats, err := d.IngestWithStats([]string{"ext"})
require.NoError(t, err)
if expectedLevel == 0 {
require.Equal(t, stats.ApproxIngestedIntoL0Bytes, stats.Bytes)
} else {
require.EqualValues(t, 0, stats.ApproxIngestedIntoL0Bytes)
}
require.Less(t, uint64(0), stats.Bytes)
}
ingest(6, "a")
ingest(0, "a")
ingest(6, "b", "g")
ingest(0, "c")
require.NoError(t, d.Close())
}
func TestIngestFlushQueuedLargeBatch(t *testing.T) {
// Verify that ingestion forces a flush of a queued large batch.
mem := vfs.NewMem()
d, err := Open("", &Options{
FS: mem,
})
require.NoError(t, err)
// The default large batch threshold is slightly less than 1/2 of the
// memtable size which makes triggering a problem with flushing queued large
// batches irritating. Manually adjust the threshold to 1/8 of the memtable
// size in order to more easily create a situation where a large batch is
// queued but not automatically flushed.
d.mu.Lock()
d.largeBatchThreshold = d.opts.MemTableSize / 8
d.mu.Unlock()
// Set a record with a large value. This will be transformed into a large
// batch and placed in the flushable queue.
require.NoError(t, d.Set([]byte("a"), bytes.Repeat([]byte("v"), int(d.largeBatchThreshold)), nil))
ingest := func(keys ...string) {
t.Helper()
f, err := mem.Create("ext")
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
for _, k := range keys {
require.NoError(t, w.Set([]byte(k), nil))
}
require.NoError(t, w.Close())
require.NoError(t, d.Ingest([]string{"ext"}))
}
ingest("a")
require.NoError(t, d.Close())
}
func TestIngestMemtablePendingOverlap(t *testing.T) {
mem := vfs.NewMem()
d, err := Open("", &Options{
FS: mem,
})
require.NoError(t, err)
d.mu.Lock()
// Use a custom commit pipeline apply function to give us control over
// timing of events.
assignedBatch := make(chan struct{})
applyBatch := make(chan struct{})
originalApply := d.commit.env.apply
d.commit.env.apply = func(b *Batch, mem *memTable) error {
assignedBatch <- struct{}{}
applyBatch <- struct{}{}
return originalApply(b, mem)
}
d.mu.Unlock()
ingest := func(keys ...string) {
t.Helper()
f, err := mem.Create("ext")
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
for _, k := range keys {
require.NoError(t, w.Set([]byte(k), nil))
}
require.NoError(t, w.Close())
require.NoError(t, d.Ingest([]string{"ext"}))
}
var wg sync.WaitGroup
wg.Add(2)
// First, Set('c') begins. This call will:
//
// * enqueue the batch to the pending queue.
// * allocate a sequence number `x`.
// * write the batch to the WAL.
//
// and then block until we read from the `applyBatch` channel down below.
go func() {
err := d.Set([]byte("c"), nil, nil)
if err != nil {
t.Error(err)
}
wg.Done()
}()
// When the above Set('c') is ready to apply, it sends on the
// `assignedBatch` channel. Once that happens, we start Ingest('a', 'c').
// The Ingest('a', 'c') allocates sequence number `x + 1`.
go func() {
// Wait until the Set has grabbed a sequence number before ingesting.
<-assignedBatch
ingest("a", "c")
wg.Done()
}()
// The Set('c')#1 and Ingest('a', 'c')#2 are both pending. To maintain
// sequence number invariants, the Set needs to be applied and flushed
// before the Ingest determines its target level.
//
// Sleep a bit to ensure that the Ingest has time to call into
// AllocateSeqNum. Once it allocates its sequence number, it should see
// that there are unpublished sequence numbers below it and spin until the
// Set's sequence number is published. After sleeping, read from
// `applyBatch` to actually allow the Set to apply and publish its
// sequence number.
time.Sleep(100 * time.Millisecond)
<-applyBatch
// Wait for both calls to complete.
wg.Wait()
require.NoError(t, d.Flush())
require.NoError(t, d.CheckLevels(nil))
require.NoError(t, d.Close())
}
type testLogger struct {
t testing.TB
}
func (l testLogger) Infof(format string, args ...interface{}) {
l.t.Logf(format, args...)
}
func (l testLogger) Errorf(format string, args ...interface{}) {
l.t.Logf(format, args...)
}
func (l testLogger) Fatalf(format string, args ...interface{}) {
l.t.Fatalf(format, args...)
}
// TestIngestMemtableOverlapRace is a regression test for the race described in
// #2196. If an ingest that checks for overlap with the mutable memtable and
// finds no overlap, it must not allow overlapping keys with later sequence
// numbers to be applied to the memtable and the memtable to be flushed before
// the ingest completes.
//
// This test operates by committing the same key concurrently:
// - 1 goroutine repeatedly ingests the same sstable writing the key `foo`
// - n goroutines repeatedly apply batches writing the key `foo` and trigger
// flushes.
//
// After a while, the database is closed and the manifest is verified. Version
// edits should contain new files with monotonically increasing sequence
// numbers, since every flush and every ingest conflicts with one another.
func TestIngestMemtableOverlapRace(t *testing.T) {
mem := vfs.NewMem()
el := MakeLoggingEventListener(testLogger{t: t})
d, err := Open("", &Options{
FS: mem,
// Disable automatic compactions to keep the manifest clean; only
// flushes and ingests.
DisableAutomaticCompactions: true,
// Disable the WAL to speed up batch commits.
DisableWAL: true,
EventListener: &el,
// We're endlessly appending to L0 without clearing it, so set a maximal
// stop writes threshold.
L0StopWritesThreshold: math.MaxInt,
// Accumulating more than 1 immutable memtable doesn't help us exercise
// the bug, since the committed keys need to be flushed promptly.
MemTableStopWritesThreshold: 2,
})
require.NoError(t, err)
// Prepare a sstable `ext` deleting foo.
f, err := mem.Create("ext")
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
require.NoError(t, w.Delete([]byte("foo")))
require.NoError(t, w.Close())
var done atomic.Bool
const numSetters = 2
var wg sync.WaitGroup
wg.Add(numSetters + 1)
untilDone := func(fn func()) {
defer wg.Done()
for !done.Load() {
fn()
}
}
// Ingest in the background.
totalIngests := 0
go untilDone(func() {
filename := fmt.Sprintf("ext%d", totalIngests)
require.NoError(t, mem.Link("ext", filename))
require.NoError(t, d.Ingest([]string{filename}))
totalIngests++
})
// Apply batches and trigger flushes in the background.
wo := &WriteOptions{Sync: false}
var localCommits [numSetters]int
for i := 0; i < numSetters; i++ {
i := i
v := []byte(fmt.Sprintf("v%d", i+1))
go untilDone(func() {
// Commit a batch setting foo=vN.
b := d.NewBatch()
require.NoError(t, b.Set([]byte("foo"), v, nil))
require.NoError(t, b.Commit(wo))
localCommits[i]++
d.AsyncFlush()
})
}
time.Sleep(100 * time.Millisecond)
done.Store(true)
wg.Wait()
var totalCommits int
for i := 0; i < numSetters; i++ {
totalCommits += localCommits[i]
}
m := d.Metrics()
tot := m.Total()
t.Logf("Committed %d batches.", totalCommits)
t.Logf("Flushed %d times.", m.Flush.Count)
t.Logf("Ingested %d sstables.", tot.TablesIngested)
require.NoError(t, d.CheckLevels(nil))
require.NoError(t, d.Close())
// Replay the manifest. Every flush and ingest is a separate version edit.
// Since they all write the same key and compactions are disabled, sequence
// numbers of new files should be monotonically increasing.
//
// This check is necessary because most of these sstables are ingested into
// L0. The L0 sublevels construction will order them by LargestSeqNum, even
// if they're added to L0 out-of-order. The CheckLevels call at the end of
// the test may find that the sublevels are all appropriately ordered, but
// the manifest may reveal they were added to the LSM out-of-order.
dbDesc, err := Peek("", mem)
require.NoError(t, err)
require.True(t, dbDesc.Exists)
f, err = mem.Open(dbDesc.ManifestFilename)
require.NoError(t, err)
defer f.Close()
rr := record.NewReader(f, 0 /* logNum */)
var largest *fileMetadata
for {
r, err := rr.Next()
if err == io.EOF || err == record.ErrInvalidChunk {
break
}
require.NoError(t, err)
var ve manifest.VersionEdit
require.NoError(t, ve.Decode(r))
t.Log(ve.String())
for _, f := range ve.NewFiles {
if largest != nil {
require.Equal(t, 0, f.Level)
if largest.LargestSeqNum > f.Meta.LargestSeqNum {
t.Fatalf("previous largest file %s has sequence number > next file %s", largest, f.Meta)
}
}
largest = f.Meta
}
}
}
type ingestCrashFS struct {
vfs.FS
}
func (fs ingestCrashFS) Link(oldname, newname string) error {
if err := fs.FS.Link(oldname, newname); err != nil {
return err
}
panic(errorfs.ErrInjected)
}
type noRemoveFS struct {
vfs.FS
}
func (fs noRemoveFS) Remove(string) error {
return errorfs.ErrInjected
}
func TestIngestFileNumReuseCrash(t *testing.T) {
const count = 10
// Use an on-disk filesystem, because Ingest with a MemFS will copy, not
// link the ingested file.
dir, err := os.MkdirTemp("", "ingest-filenum-reuse")
require.NoError(t, err)
defer os.RemoveAll(dir)
fs := vfs.Default
readFile := func(s string) []byte {
f, err := fs.Open(fs.PathJoin(dir, s))
require.NoError(t, err)
b, err := io.ReadAll(f)
require.NoError(t, err)
require.NoError(t, f.Close())
return b
}
// Create sstables to ingest.
var files []string
var fileBytes [][]byte
for i := 0; i < count; i++ {
name := fmt.Sprintf("ext%d", i)
f, err := fs.Create(fs.PathJoin(dir, name))
require.NoError(t, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
require.NoError(t, w.Set([]byte(fmt.Sprintf("foo%d", i)), nil))
require.NoError(t, w.Close())
files = append(files, name)
fileBytes = append(fileBytes, readFile(name))
}
// Open a database with a filesystem that will successfully link the
// ingested files but then panic. This is an approximation of what a crash
// after linking but before updating the manifest would look like.
d, err := Open(dir, &Options{
FS: ingestCrashFS{FS: fs},
})
// A flush here ensures the file num bumps from creating OPTIONS files,
// etc get recorded in the manifest. We want the nextFileNum after the
// restart to be the same as one of our ingested sstables.
require.NoError(t, err)
require.NoError(t, d.Set([]byte("boop"), nil, nil))
require.NoError(t, d.Flush())
for _, f := range files {
func() {
defer func() { err = recover().(error) }()
err = d.Ingest([]string{fs.PathJoin(dir, f)})
}()
if err == nil || !errors.Is(err, errorfs.ErrInjected) {
t.Fatalf("expected injected error, got %v", err)
}
}
// Leave something in the WAL so that Open will flush while replaying the
// WAL.
require.NoError(t, d.Set([]byte("wal"), nil, nil))
require.NoError(t, d.Close())
// There are now two links to each external file: the original extX link
// and a numbered sstable link. The sstable files are still not a part of
// the manifest and so they may be overwritten. Open will detect the
// obsolete number sstables and try to remove them. The FS here is wrapped
// to induce errors on Remove calls. Even if we're unsuccessful in
// removing the obsolete files, the external files should not be
// overwritten.
d, err = Open(dir, &Options{FS: noRemoveFS{FS: fs}})
require.NoError(t, err)
require.NoError(t, d.Set([]byte("bar"), nil, nil))
require.NoError(t, d.Flush())
require.NoError(t, d.Close())
// None of the external files should change despite modifying the linked
// versions.
for i, f := range files {
afterBytes := readFile(f)
require.Equal(t, fileBytes[i], afterBytes)
}
}
func TestIngest_UpdateSequenceNumber(t *testing.T) {
mem := vfs.NewMem()
cmp := base.DefaultComparer.Compare
parse := func(input string) (*sstable.Writer, error) {
f, err := mem.Create("ext")
if err != nil {
return nil, err
}
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
TableFormat: sstable.TableFormatMax,
})
for _, data := range strings.Split(input, "\n") {
if strings.HasPrefix(data, "rangekey: ") {
data = strings.TrimPrefix(data, "rangekey: ")
s := keyspan.ParseSpan(data)
err := rangekey.Encode(&s, w.AddRangeKey)
if err != nil {
return nil, err
}
continue
}
j := strings.Index(data, ":")
if j < 0 {
return nil, errors.Newf("malformed input: %s\n", data)
}
key := base.ParseInternalKey(data[:j])
value := []byte(data[j+1:])
if err := w.Add(key, value); err != nil {
return nil, err
}
}
return w, nil
}
var (
seqnum uint64
err error
metas []*fileMetadata
)
datadriven.RunTest(t, "testdata/ingest_update_seqnums", func(t *testing.T, td *datadriven.TestData) string {
switch td.Cmd {
case "starting-seqnum":
seqnum, err = strconv.ParseUint(td.Input, 10, 64)
if err != nil {
return err.Error()
}
return ""
case "reset":
metas = metas[:0]
return ""
case "load":
w, err := parse(td.Input)
if err != nil {
return err.Error()
}
if err = w.Close(); err != nil {
return err.Error()
}
defer w.Close()
// Format the bounds of the table.
wm, err := w.Metadata()
if err != nil {
return err.Error()
}
// Upper bounds for range dels and range keys are expected to be sentinel
// keys.
maybeUpdateUpperBound := func(key base.InternalKey) base.InternalKey {
switch k := key.Kind(); {
case k == base.InternalKeyKindRangeDelete:
key.Trailer = base.InternalKeyRangeDeleteSentinel
case rangekey.IsRangeKey(k):
return base.MakeExclusiveSentinelKey(k, key.UserKey)
}
return key
}
// Construct the file metadata from the writer metadata.
m := &fileMetadata{
SmallestSeqNum: 0, // Simulate an ingestion.
LargestSeqNum: 0,
}
if wm.HasPointKeys {
m.ExtendPointKeyBounds(cmp, wm.SmallestPoint, wm.LargestPoint)
}
if wm.HasRangeDelKeys {
m.ExtendPointKeyBounds(
cmp,
wm.SmallestRangeDel,
maybeUpdateUpperBound(wm.LargestRangeDel),
)
}
if wm.HasRangeKeys {
m.ExtendRangeKeyBounds(
cmp,
wm.SmallestRangeKey,
maybeUpdateUpperBound(wm.LargestRangeKey),
)
}
m.InitPhysicalBacking()
if err := m.Validate(cmp, base.DefaultFormatter); err != nil {
return err.Error()
}
// Collect this file.
metas = append(metas, m)
// Return an index number for the file.
return fmt.Sprintf("file %d\n", len(metas)-1)
case "update-files":
// Update the bounds across all files.
if err = ingestUpdateSeqNum(cmp, base.DefaultFormatter, seqnum, ingestLoadResult{localMeta: metas}); err != nil {
return err.Error()
}
var buf bytes.Buffer
for i, m := range metas {
fmt.Fprintf(&buf, "file %d:\n", i)
fmt.Fprintf(&buf, " combined: %s-%s\n", m.Smallest, m.Largest)
fmt.Fprintf(&buf, " points: %s-%s\n", m.SmallestPointKey, m.LargestPointKey)
fmt.Fprintf(&buf, " ranges: %s-%s\n", m.SmallestRangeKey, m.LargestRangeKey)
}
return buf.String()
default:
return fmt.Sprintf("unknown command %s\n", td.Cmd)
}
})
}
func TestIngestCleanup(t *testing.T) {
fns := []base.FileNum{0, 1, 2}
testCases := []struct {
closeFiles []base.FileNum
cleanupFiles []base.FileNum
wantErr string
}{
// Close and remove all files.
{
closeFiles: fns,
cleanupFiles: fns,
},
// Remove a non-existent file.
{
closeFiles: fns,
cleanupFiles: []base.FileNum{3},
wantErr: "unknown to the objstorage provider",
},
// Remove a file that has not been closed.
{
closeFiles: []base.FileNum{0, 2},
cleanupFiles: fns,
wantErr: oserror.ErrInvalid.Error(),
},
// Remove all files, one of which is still open, plus a file that does not exist.
{
closeFiles: []base.FileNum{0, 2},
cleanupFiles: []base.FileNum{0, 1, 2, 3},
wantErr: oserror.ErrInvalid.Error(), // The first error encountered is due to the open file.
},
}
for _, tc := range testCases {
t.Run("", func(t *testing.T) {
mem := vfs.NewMem()
mem.UseWindowsSemantics(true)
objProvider, err := objstorageprovider.Open(objstorageprovider.DefaultSettings(mem, ""))
require.NoError(t, err)
defer objProvider.Close()
// Create the files in the VFS.
metaMap := make(map[base.FileNum]objstorage.Writable)
for _, fn := range fns {
w, _, err := objProvider.Create(context.Background(), base.FileTypeTable, fn.DiskFileNum(), objstorage.CreateOptions{})
require.NoError(t, err)
metaMap[fn] = w
}
// Close a select number of files.
for _, m := range tc.closeFiles {
w, ok := metaMap[m]
if !ok {
continue
}
require.NoError(t, w.Finish())
}
// Cleanup the set of files in the FS.
var toRemove []*fileMetadata
for _, fn := range tc.cleanupFiles {
m := &fileMetadata{FileNum: fn}
m.InitPhysicalBacking()
toRemove = append(toRemove, m)
}
err = ingestCleanup(objProvider, toRemove)
if tc.wantErr != "" {
require.Error(t, err, "got no error, expected %s", tc.wantErr)
require.Contains(t, err.Error(), tc.wantErr)
} else {
require.NoError(t, err)
}
})
}
}
// fatalCapturingLogger captures a fatal error instead of panicking.
type fatalCapturingLogger struct {
t testing.TB
err error
}
// Infof implements the Logger interface.
func (l *fatalCapturingLogger) Infof(fmt string, args ...interface{}) {
l.t.Logf(fmt, args...)
}
// Errorf implements the Logger interface.
func (l *fatalCapturingLogger) Errorf(fmt string, args ...interface{}) {
l.t.Logf(fmt, args...)
}
// Fatalf implements the Logger interface.
func (l *fatalCapturingLogger) Fatalf(_ string, args ...interface{}) {
l.err = args[0].(error)
}
func TestIngestValidation(t *testing.T) {
type keyVal struct {
key, val []byte
}
// The corruptionLocation enum defines where to corrupt an sstable if
// anywhere. corruptionLocation{Start,End} describe the start and end
// data blocks. corruptionLocationInternal describes a random data block
// that's neither the start or end blocks. The Ingest operation does not
// read the entire sstable, only the start and end blocks, so corruption
// introduced using corruptionLocationInternal will not be discovered until
// the asynchronous validation job runs.
type corruptionLocation int
const (
corruptionLocationNone corruptionLocation = iota
corruptionLocationStart
corruptionLocationEnd
corruptionLocationInternal
)
// The errReportLocation type defines an enum to allow tests to enforce
// expectations about how an error surfaced during ingestion or validation
// is reported. Asynchronous validation that uncovers corruption should call
// Fatalf on the Logger. Asychronous validation that encounters
// non-corruption errors should surface it through the
// EventListener.BackgroundError func.
type errReportLocation int
const (
errReportLocationNone errReportLocation = iota
errReportLocationIngest
errReportLocationFatal
errReportLocationBackgroundError
)
const (
nKeys = 1_000
keySize = 16
valSize = 100
blockSize = 100
ingestTableName = "ext"
)
seed := uint64(time.Now().UnixNano())
rng := rand.New(rand.NewSource(seed))
t.Logf("rng seed = %d", seed)
// errfsCounter is used by test cases that make use of an errorfs.Injector
// to inject errors into the ingest validation code path.
var errfsCounter atomic.Int32
testCases := []struct {
description string
cLoc corruptionLocation
wantErrType errReportLocation
wantErr error
errorfsInjector errorfs.Injector
}{
{
description: "no corruption",
cLoc: corruptionLocationNone,
wantErrType: errReportLocationNone,
},
{
description: "start block",
cLoc: corruptionLocationStart,
wantErr: ErrCorruption,
wantErrType: errReportLocationIngest,
},
{
description: "end block",
cLoc: corruptionLocationEnd,
wantErr: ErrCorruption,
wantErrType: errReportLocationIngest,
},
{
description: "non-end block",
cLoc: corruptionLocationInternal,
wantErr: ErrCorruption,
wantErrType: errReportLocationFatal,
},
{
description: "non-corruption error",
cLoc: corruptionLocationNone,
wantErr: errorfs.ErrInjected,
wantErrType: errReportLocationBackgroundError,
errorfsInjector: errorfs.InjectorFunc(func(op errorfs.Op) error {
// Inject an error on the first read-at operation on an sstable
// (excluding the read on the sstable before ingestion has
// linked it in).
if op.Path != "ext" && op.Kind != errorfs.OpFileReadAt || filepath.Ext(op.Path) != ".sst" {
return nil
}
if errfsCounter.Add(1) == 1 {
return errorfs.ErrInjected
}
return nil
}),
},
}
for _, tc := range testCases {
t.Run(tc.description, func(t *testing.T) {
errfsCounter.Store(0)
var wg sync.WaitGroup
wg.Add(1)
fs := vfs.NewMem()
var testFS vfs.FS = fs
if tc.errorfsInjector != nil {
testFS = errorfs.Wrap(fs, tc.errorfsInjector)
}
// backgroundErr is populated by EventListener.BackgroundError.
var backgroundErr error
logger := &fatalCapturingLogger{t: t}
opts := &Options{
FS: testFS,
Logger: logger,
EventListener: &EventListener{
TableValidated: func(i TableValidatedInfo) {
wg.Done()
},
BackgroundError: func(err error) {
backgroundErr = err
},
},
}
// Disable table stats so that injected errors can't be accidentally
// injected into the table stats collector read, and so the table
// stats collector won't prime the table+block cache such that the
// error injection won't trigger at all during ingest validation.
opts.private.disableTableStats = true
opts.Experimental.ValidateOnIngest = true
d, err := Open("", opts)
require.NoError(t, err)
defer func() { require.NoError(t, d.Close()) }()
corrupt := func(f vfs.File) {
readable, err := sstable.NewSimpleReadable(f)
require.NoError(t, err)
// Compute the layout of the sstable in order to find the
// appropriate block locations to corrupt.
r, err := sstable.NewReader(readable, sstable.ReaderOptions{})
require.NoError(t, err)
l, err := r.Layout()
require.NoError(t, err)
// Select an appropriate data block to corrupt.
var blockIdx int
switch tc.cLoc {
case corruptionLocationStart:
blockIdx = 0
case corruptionLocationEnd:
blockIdx = len(l.Data) - 1
case corruptionLocationInternal:
blockIdx = 1 + rng.Intn(len(l.Data)-2)
default:
t.Fatalf("unknown corruptionLocation: %T", tc.cLoc)
}
bh := l.Data[blockIdx]
// Corrupting a key will cause the ingestion to fail due to a
// malformed key, rather than a block checksum mismatch.
// Instead, we corrupt the last byte in the selected block,
// before the trailer, which corresponds to a value.
offset := bh.Offset + bh.Length - 1
_, err = f.WriteAt([]byte("\xff"), int64(offset))
require.NoError(t, err)
require.NoError(t, r.Close())
}
type errT struct {
errLoc errReportLocation
err error
}
runIngest := func(keyVals []keyVal) (et errT) {
f, err := fs.Create(ingestTableName)
require.NoError(t, err)
defer func() { _ = fs.Remove(ingestTableName) }()
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{
BlockSize: blockSize, // Create many smaller blocks.
Compression: NoCompression, // For simpler debugging.
})
for _, kv := range keyVals {
require.NoError(t, w.Set(kv.key, kv.val))
}
require.NoError(t, w.Close())
// Possibly corrupt the file.
if tc.cLoc != corruptionLocationNone {
f, err = fs.OpenReadWrite(ingestTableName)
require.NoError(t, err)
corrupt(f)
}
// Ingest the external table.
err = d.Ingest([]string{ingestTableName})
if err != nil {
et.errLoc = errReportLocationIngest
et.err = err
return
}
// Wait for the validation on the sstable to complete.
wg.Wait()
// Return any error encountered during validation.
if logger.err != nil {
et.errLoc = errReportLocationFatal
et.err = logger.err
} else if backgroundErr != nil {
et.errLoc = errReportLocationBackgroundError
et.err = backgroundErr
}
return
}
// Construct a set of keys to ingest.
var keyVals []keyVal
for i := 0; i < nKeys; i++ {
key := make([]byte, keySize)
_, err = rng.Read(key)
require.NoError(t, err)
val := make([]byte, valSize)
_, err = rng.Read(val)
require.NoError(t, err)
keyVals = append(keyVals, keyVal{key, val})
}
// Keys must be sorted.
slices.SortFunc(keyVals, func(a, b keyVal) int { return d.cmp(a.key, b.key) })
// Run the ingestion.
et := runIngest(keyVals)
// Assert we saw the errors we expect.
switch tc.wantErrType {
case errReportLocationNone:
require.Equal(t, errReportLocationNone, et.errLoc)
require.NoError(t, et.err)
case errReportLocationIngest:
require.Equal(t, errReportLocationIngest, et.errLoc)
require.Error(t, et.err)
require.True(t, errors.Is(et.err, tc.wantErr))
case errReportLocationFatal:
require.Equal(t, errReportLocationFatal, et.errLoc)
require.Error(t, et.err)
require.True(t, errors.Is(et.err, tc.wantErr))
case errReportLocationBackgroundError:
require.Equal(t, errReportLocationBackgroundError, et.errLoc)
require.Error(t, et.err)
require.True(t, errors.Is(et.err, tc.wantErr))
default:
t.Fatalf("unknown wantErrType %T", tc.wantErrType)
}
})
}
}
// BenchmarkManySSTables measures the cost of various operations with various
// counts of SSTables within the database.
func BenchmarkManySSTables(b *testing.B) {
counts := []int{10, 1_000, 10_000, 100_000, 1_000_000}
ops := []string{"ingest", "calculateInuseKeyRanges"}
for _, op := range ops {
b.Run(op, func(b *testing.B) {
for _, count := range counts {
b.Run(fmt.Sprintf("sstables=%d", count), func(b *testing.B) {
mem := vfs.NewMem()
d, err := Open("", &Options{
FS: mem,
})
require.NoError(b, err)
var paths []string
for i := 0; i < count; i++ {
n := fmt.Sprintf("%07d", i)
f, err := mem.Create(n)
require.NoError(b, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
require.NoError(b, w.Set([]byte(n), nil))
require.NoError(b, w.Close())
paths = append(paths, n)
}
require.NoError(b, d.Ingest(paths))
{
const broadIngest = "broad.sst"
f, err := mem.Create(broadIngest)
require.NoError(b, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
require.NoError(b, w.Set([]byte("0"), nil))
require.NoError(b, w.Set([]byte("Z"), nil))
require.NoError(b, w.Close())
require.NoError(b, d.Ingest([]string{broadIngest}))
}
switch op {
case "ingest":
runBenchmarkManySSTablesIngest(b, d, mem, count)
case "calculateInuseKeyRanges":
runBenchmarkManySSTablesInUseKeyRanges(b, d, count)
}
require.NoError(b, d.Close())
})
}
})
}
}
func runBenchmarkManySSTablesIngest(b *testing.B, d *DB, fs vfs.FS, count int) {
b.ResetTimer()
for i := 0; i < b.N; i++ {
n := fmt.Sprintf("%07d", count+i)
f, err := fs.Create(n)
require.NoError(b, err)
w := sstable.NewWriter(objstorageprovider.NewFileWritable(f), sstable.WriterOptions{})
require.NoError(b, w.Set([]byte(n), nil))
require.NoError(b, w.Close())
require.NoError(b, d.Ingest([]string{n}))
}
}
func runBenchmarkManySSTablesInUseKeyRanges(b *testing.B, d *DB, count int) {
// This benchmark is pretty contrived, but it's not easy to write a
// microbenchmark for this in a more natural way. L6 has many files, and
// L5 has 1 file spanning the entire breadth of L5.
d.mu.Lock()
defer d.mu.Unlock()
v := d.mu.versions.currentVersion()
b.ResetTimer()
smallest := []byte("0")
largest := []byte("z")
for i := 0; i < b.N; i++ {
_ = calculateInuseKeyRanges(v, d.cmp, 0, numLevels-1, smallest, largest)
}
}
Reference in New Issue View Git Blame Copy Permalink