| /* |
| * SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc. |
| * SPDX-License-Identifier: Apache-2.0 |
| */ |
| |
| package badger |
| |
| import ( |
| "bytes" |
| "errors" |
| "expvar" |
| "fmt" |
| "math" |
| "math/rand" |
| "os" |
| "reflect" |
| "sort" |
| "sync" |
| "testing" |
| "time" |
| |
| humanize "github.com/dustin/go-humanize" |
| "github.com/stretchr/testify/require" |
| |
| "github.com/dgraph-io/badger/v4/options" |
| "github.com/dgraph-io/badger/v4/y" |
| ) |
| |
| func TestDynamicValueThreshold(t *testing.T) { |
| t.Skip() |
| dir, err := os.MkdirTemp("", "badger-test") |
| y.Check(err) |
| defer removeDir(dir) |
| kv, _ := Open(getTestOptions(dir).WithValueThreshold(32).WithVLogPercentile(0.99)) |
| defer kv.Close() |
| log := &kv.vlog |
| for vl := 32; vl <= 1024; vl = vl + 4 { |
| for i := 0; i < 1000; i++ { |
| val := make([]byte, vl) |
| y.Check2(rand.Read(val)) |
| e1 := &Entry{ |
| Key: []byte(fmt.Sprintf("samplekey_%d_%d", vl, i)), |
| Value: val, |
| meta: bitValuePointer, |
| } |
| b := new(request) |
| b.Entries = []*Entry{e1} |
| require.NoError(t, log.write([]*request{b})) |
| } |
| t.Logf("value threshold is %d \n", log.db.valueThreshold()) |
| } |
| |
| for vl := 511; vl >= 31; vl = vl - 4 { |
| for i := 0; i < 5000; i++ { |
| val := make([]byte, vl) |
| y.Check2(rand.Read(val)) |
| e1 := &Entry{ |
| Key: []byte(fmt.Sprintf("samplekey_%d_%d", vl, i)), |
| Value: val, |
| meta: bitValuePointer, |
| } |
| b := new(request) |
| b.Entries = []*Entry{e1} |
| require.NoError(t, log.write([]*request{b})) |
| } |
| t.Logf("value threshold is %d \n", log.db.valueThreshold()) |
| } |
| require.Equal(t, log.db.valueThreshold(), int64(995)) |
| } |
| |
| func TestValueBasic(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| y.Check(err) |
| defer removeDir(dir) |
| |
| kv, _ := Open(getTestOptions(dir).WithValueThreshold(32)) |
| defer kv.Close() |
| log := &kv.vlog |
| |
| // Use value big enough that the value log writes them even if SyncWrites is false. |
| const val1 = "sampleval012345678901234567890123" |
| const val2 = "samplevalb012345678901234567890123" |
| require.True(t, int64(len(val1)) >= kv.vlog.db.valueThreshold()) |
| |
| e1 := &Entry{ |
| Key: []byte("samplekey"), |
| Value: []byte(val1), |
| meta: bitValuePointer, |
| } |
| e2 := &Entry{ |
| Key: []byte("samplekeyb"), |
| Value: []byte(val2), |
| meta: bitValuePointer, |
| } |
| |
| b := new(request) |
| b.Entries = []*Entry{e1, e2} |
| |
| require.NoError(t, log.write([]*request{b})) |
| require.Len(t, b.Ptrs, 2) |
| t.Logf("Pointer written: %+v %+v\n", b.Ptrs[0], b.Ptrs[1]) |
| |
| buf1, lf1, err1 := log.readValueBytes(b.Ptrs[0]) |
| buf2, lf2, err2 := log.readValueBytes(b.Ptrs[1]) |
| require.NoError(t, err1) |
| require.NoError(t, err2) |
| defer runCallback(log.getUnlockCallback(lf1)) |
| defer runCallback(log.getUnlockCallback(lf2)) |
| e1, err = lf1.decodeEntry(buf1, b.Ptrs[0].Offset) |
| require.NoError(t, err) |
| e2, err = lf1.decodeEntry(buf2, b.Ptrs[1].Offset) |
| require.NoError(t, err) |
| readEntries := []Entry{*e1, *e2} |
| require.EqualValues(t, []Entry{ |
| { |
| Key: []byte("samplekey"), |
| Value: []byte(val1), |
| meta: bitValuePointer, |
| offset: b.Ptrs[0].Offset, |
| }, |
| { |
| Key: []byte("samplekeyb"), |
| Value: []byte(val2), |
| meta: bitValuePointer, |
| offset: b.Ptrs[1].Offset, |
| }, |
| }, readEntries) |
| |
| } |
| |
| func TestValueGCManaged(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| N := 10000 |
| |
| opt := getTestOptions(dir) |
| opt.ValueLogMaxEntries = uint32(N / 10) |
| opt.managedTxns = true |
| opt.BaseTableSize = 1 << 15 |
| opt.ValueThreshold = 1 << 10 |
| opt.MemTableSize = 1 << 15 |
| |
| db, err := Open(opt) |
| require.NoError(t, err) |
| defer db.Close() |
| |
| var ts uint64 |
| newTs := func() uint64 { |
| ts++ |
| return ts |
| } |
| |
| sz := 64 << 10 |
| var wg sync.WaitGroup |
| for i := 0; i < N; i++ { |
| v := make([]byte, sz) |
| rand.Read(v[:rand.Intn(sz)]) |
| |
| wg.Add(1) |
| txn := db.NewTransactionAt(newTs(), true) |
| require.NoError(t, txn.SetEntry(NewEntry([]byte(fmt.Sprintf("key%d", i)), v))) |
| require.NoError(t, txn.CommitAt(newTs(), func(err error) { |
| wg.Done() |
| require.NoError(t, err) |
| })) |
| } |
| |
| for i := 0; i < N; i++ { |
| wg.Add(1) |
| txn := db.NewTransactionAt(newTs(), true) |
| require.NoError(t, txn.Delete([]byte(fmt.Sprintf("key%d", i)))) |
| require.NoError(t, txn.CommitAt(newTs(), func(err error) { |
| wg.Done() |
| require.NoError(t, err) |
| })) |
| } |
| wg.Wait() |
| entries, err := os.ReadDir(dir) |
| require.NoError(t, err) |
| for _, e := range entries { |
| fi, err := e.Info() |
| require.NoError(t, err) |
| t.Logf("File: %s. Size: %s\n", fi.Name(), humanize.IBytes(uint64(fi.Size()))) |
| } |
| |
| db.SetDiscardTs(math.MaxUint32) |
| require.NoError(t, db.Flatten(3)) |
| |
| for i := 0; i < 100; i++ { |
| // Try at max 100 times to GC even a single value log file. |
| if err := db.RunValueLogGC(0.0001); err == nil { |
| return // Done |
| } |
| } |
| require.Fail(t, "Unable to GC even a single value log file.") |
| } |
| |
| func TestValueGC(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| opt := getTestOptions(dir) |
| opt.ValueLogFileSize = 1 << 20 |
| opt.BaseTableSize = 1 << 15 |
| opt.ValueThreshold = 1 << 10 |
| |
| kv, _ := Open(opt) |
| defer kv.Close() |
| |
| sz := 32 << 10 |
| txn := kv.NewTransaction(true) |
| for i := 0; i < 100; i++ { |
| v := make([]byte, sz) |
| rand.Read(v[:rand.Intn(sz)]) |
| require.NoError(t, txn.SetEntry(NewEntry([]byte(fmt.Sprintf("key%d", i)), v))) |
| if i%20 == 0 { |
| require.NoError(t, txn.Commit()) |
| txn = kv.NewTransaction(true) |
| } |
| } |
| require.NoError(t, txn.Commit()) |
| |
| for i := 0; i < 45; i++ { |
| txnDelete(t, kv, []byte(fmt.Sprintf("key%d", i))) |
| } |
| |
| kv.vlog.filesLock.RLock() |
| lf := kv.vlog.filesMap[kv.vlog.sortedFids()[0]] |
| kv.vlog.filesLock.RUnlock() |
| |
| // lf.iterate(0, func(e Entry) bool { |
| // e.print("lf") |
| // return true |
| // }) |
| |
| require.NoError(t, kv.vlog.rewrite(lf)) |
| for i := 45; i < 100; i++ { |
| key := []byte(fmt.Sprintf("key%d", i)) |
| |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| item, err := txn.Get(key) |
| require.NoError(t, err) |
| val := getItemValue(t, item) |
| require.NotNil(t, val) |
| require.True(t, len(val) == sz, "Size found: %d", len(val)) |
| return nil |
| })) |
| } |
| } |
| |
| func TestValueGC2(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| opt := getTestOptions(dir) |
| opt.ValueLogFileSize = 1 << 20 |
| opt.BaseTableSize = 1 << 15 |
| opt.ValueThreshold = 1 << 10 |
| |
| kv, _ := Open(opt) |
| defer kv.Close() |
| |
| sz := 32 << 10 |
| txn := kv.NewTransaction(true) |
| for i := 0; i < 100; i++ { |
| v := make([]byte, sz) |
| rand.Read(v[:rand.Intn(sz)]) |
| require.NoError(t, txn.SetEntry(NewEntry([]byte(fmt.Sprintf("key%d", i)), v))) |
| if i%20 == 0 { |
| require.NoError(t, txn.Commit()) |
| txn = kv.NewTransaction(true) |
| } |
| } |
| require.NoError(t, txn.Commit()) |
| |
| for i := 0; i < 5; i++ { |
| txnDelete(t, kv, []byte(fmt.Sprintf("key%d", i))) |
| } |
| |
| for i := 5; i < 10; i++ { |
| v := []byte(fmt.Sprintf("value%d", i)) |
| txnSet(t, kv, []byte(fmt.Sprintf("key%d", i)), v, 0) |
| } |
| |
| kv.vlog.filesLock.RLock() |
| lf := kv.vlog.filesMap[kv.vlog.sortedFids()[0]] |
| kv.vlog.filesLock.RUnlock() |
| |
| // lf.iterate(0, func(e Entry) bool { |
| // e.print("lf") |
| // return true |
| // }) |
| |
| require.NoError(t, kv.vlog.rewrite(lf)) |
| for i := 0; i < 5; i++ { |
| key := []byte(fmt.Sprintf("key%d", i)) |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| _, err := txn.Get(key) |
| require.Equal(t, ErrKeyNotFound, err) |
| return nil |
| })) |
| } |
| for i := 5; i < 10; i++ { |
| key := []byte(fmt.Sprintf("key%d", i)) |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| item, err := txn.Get(key) |
| require.NoError(t, err) |
| val := getItemValue(t, item) |
| require.NotNil(t, val) |
| require.Equal(t, string(val), fmt.Sprintf("value%d", i)) |
| return nil |
| })) |
| } |
| // Moved entries. |
| for i := 10; i < 100; i++ { |
| key := []byte(fmt.Sprintf("key%d", i)) |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| item, err := txn.Get(key) |
| require.NoError(t, err) |
| val := getItemValue(t, item) |
| require.NotNil(t, val) |
| require.True(t, len(val) == sz, "Size found: %d", len(val)) |
| return nil |
| })) |
| } |
| } |
| |
| func TestValueGC3(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| opt := getTestOptions(dir) |
| opt.ValueLogFileSize = 1 << 20 |
| opt.BaseTableSize = 1 << 15 |
| opt.ValueThreshold = 1 << 10 |
| |
| kv, err := Open(opt) |
| require.NoError(t, err) |
| defer kv.Close() |
| |
| // We want to test whether an iterator can continue through a value log GC. |
| |
| valueSize := 32 << 10 |
| |
| var value3 []byte |
| txn := kv.NewTransaction(true) |
| for i := 0; i < 100; i++ { |
| v := make([]byte, valueSize) // 32K * 100 will take >=3'276'800 B. |
| if i == 3 { |
| value3 = v |
| } |
| rand.Read(v[:]) |
| // Keys key000, key001, key002, such that sorted order matches insertion order |
| require.NoError(t, txn.SetEntry(NewEntry([]byte(fmt.Sprintf("key%03d", i)), v))) |
| if i%20 == 0 { |
| require.NoError(t, txn.Commit()) |
| txn = kv.NewTransaction(true) |
| } |
| } |
| require.NoError(t, txn.Commit()) |
| |
| // Start an iterator to keys in the first value log file |
| itOpt := IteratorOptions{ |
| PrefetchValues: false, |
| PrefetchSize: 0, |
| Reverse: false, |
| } |
| |
| txn = kv.NewTransaction(true) |
| it := txn.NewIterator(itOpt) |
| defer it.Close() |
| // Walk a few keys |
| it.Rewind() |
| require.True(t, it.Valid()) |
| item := it.Item() |
| require.Equal(t, []byte("key000"), item.Key()) |
| it.Next() |
| require.True(t, it.Valid()) |
| item = it.Item() |
| require.Equal(t, []byte("key001"), item.Key()) |
| it.Next() |
| require.True(t, it.Valid()) |
| item = it.Item() |
| require.Equal(t, []byte("key002"), item.Key()) |
| |
| // Like other tests, we pull out a logFile to rewrite it directly |
| |
| kv.vlog.filesLock.RLock() |
| logFile := kv.vlog.filesMap[kv.vlog.sortedFids()[0]] |
| kv.vlog.filesLock.RUnlock() |
| |
| require.NoError(t, kv.vlog.rewrite(logFile)) |
| it.Next() |
| require.True(t, it.Valid()) |
| item = it.Item() |
| require.Equal(t, []byte("key003"), item.Key()) |
| |
| v3, err := item.ValueCopy(nil) |
| require.NoError(t, err) |
| require.Equal(t, value3, v3) |
| } |
| |
| func TestValueGC4(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| opt := getTestOptions(dir) |
| opt.ValueLogFileSize = 1 << 20 |
| opt.BaseTableSize = 1 << 15 |
| opt.ValueThreshold = 1 << 10 |
| |
| kv, err := Open(opt) |
| require.NoError(t, err) |
| |
| sz := 128 << 10 // 5 entries per value log file. |
| txn := kv.NewTransaction(true) |
| for i := 0; i < 24; i++ { |
| v := make([]byte, sz) |
| rand.Read(v[:rand.Intn(sz)]) |
| require.NoError(t, txn.SetEntry(NewEntry([]byte(fmt.Sprintf("key%d", i)), v))) |
| if i%3 == 0 { |
| require.NoError(t, txn.Commit()) |
| txn = kv.NewTransaction(true) |
| } |
| } |
| require.NoError(t, txn.Commit()) |
| |
| for i := 0; i < 8; i++ { |
| txnDelete(t, kv, []byte(fmt.Sprintf("key%d", i))) |
| } |
| |
| for i := 8; i < 16; i++ { |
| v := []byte(fmt.Sprintf("value%d", i)) |
| txnSet(t, kv, []byte(fmt.Sprintf("key%d", i)), v, 0) |
| } |
| |
| kv.vlog.filesLock.RLock() |
| lf0 := kv.vlog.filesMap[kv.vlog.sortedFids()[0]] |
| lf1 := kv.vlog.filesMap[kv.vlog.sortedFids()[1]] |
| kv.vlog.filesLock.RUnlock() |
| |
| // lf.iterate(0, func(e Entry) bool { |
| // e.print("lf") |
| // return true |
| // }) |
| |
| require.NoError(t, kv.vlog.rewrite(lf0)) |
| require.NoError(t, kv.vlog.rewrite(lf1)) |
| |
| require.NoError(t, kv.Close()) |
| |
| kv, err = Open(opt) |
| require.NoError(t, err) |
| |
| for i := 0; i < 8; i++ { |
| key := []byte(fmt.Sprintf("key%d", i)) |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| _, err := txn.Get(key) |
| require.Equal(t, ErrKeyNotFound, err) |
| return nil |
| })) |
| } |
| for i := 8; i < 16; i++ { |
| key := []byte(fmt.Sprintf("key%d", i)) |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| item, err := txn.Get(key) |
| require.NoError(t, err) |
| val := getItemValue(t, item) |
| require.NotNil(t, val) |
| require.Equal(t, string(val), fmt.Sprintf("value%d", i)) |
| return nil |
| })) |
| } |
| require.NoError(t, kv.Close()) |
| } |
| |
| func TestPersistLFDiscardStats(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| opt := getTestOptions(dir) |
| // Force more compaction by reducing the number of L0 tables. |
| opt.NumLevelZeroTables = 1 |
| opt.ValueLogFileSize = 1 << 20 |
| // Avoid compaction on close so that the discard map remains the same. |
| opt.CompactL0OnClose = false |
| opt.MemTableSize = 1 << 15 |
| opt.ValueThreshold = 1 << 10 |
| tChan := make(chan string, 100) |
| defer close(tChan) |
| opt.syncChan = tChan |
| |
| db, err := Open(opt) |
| require.NoError(t, err) |
| capturedDiscardStats := make(map[uint64]uint64) |
| db.onCloseDiscardCapture = capturedDiscardStats |
| |
| sz := 128 << 10 // 5 entries per value log file. |
| v := make([]byte, sz) |
| rand.Read(v[:rand.Intn(sz)]) |
| txn := db.NewTransaction(true) |
| for i := 0; i < 500; i++ { |
| require.NoError(t, txn.SetEntry(NewEntry([]byte(fmt.Sprintf("key%d", i)), v))) |
| if i%3 == 0 { |
| require.NoError(t, txn.Commit()) |
| txn = db.NewTransaction(true) |
| } |
| } |
| require.NoError(t, txn.Commit(), "error while committing txn") |
| |
| for i := 0; i < 500; i++ { |
| // use Entry.WithDiscard() to delete entries, because this causes data to be flushed on |
| // disk, creating SSTs. Simple Delete was having data in Memtables only. |
| err = db.Update(func(txn *Txn) error { |
| return txn.SetEntry(NewEntry([]byte(fmt.Sprintf("key%d", i)), v).WithDiscard()) |
| }) |
| require.NoError(t, err) |
| } |
| |
| // Wait for invocation of updateDiscardStats at least once -- timeout after 60 seconds. |
| waitForMessage(tChan, updateDiscardStatsMsg, 1, 60, t) |
| |
| db.vlog.discardStats.Lock() |
| require.True(t, db.vlog.discardStats.Len() > 1, "some discardStats should be generated") |
| |
| db.vlog.discardStats.Unlock() |
| require.NoError(t, db.Close()) |
| |
| // Avoid running compactors on reopening badger. |
| opt.NumCompactors = 0 |
| db, err = Open(opt) |
| require.NoError(t, err) |
| defer db.Close() |
| waitForMessage(tChan, endVLogInitMsg, 1, 60, t) |
| db.vlog.discardStats.Lock() |
| statsMap := make(map[uint64]uint64) |
| db.vlog.discardStats.Iterate(func(fid, val uint64) { |
| statsMap[fid] = val |
| }) |
| require.Truef(t, reflect.DeepEqual(capturedDiscardStats, statsMap), |
| "Discard maps are not equal. On Close: %+v, After Reopen: %+v", |
| capturedDiscardStats, statsMap) |
| db.vlog.discardStats.Unlock() |
| } |
| |
| func TestValueChecksums(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| // Set up SST with K1=V1 |
| opts := getTestOptions(dir) |
| opts.ValueLogFileSize = 100 * 1024 * 1024 // 100Mb |
| opts.VerifyValueChecksum = true |
| kv, err := Open(opts) |
| require.NoError(t, err) |
| require.NoError(t, kv.Close()) |
| |
| var ( |
| k0 = []byte("k0") |
| k1 = []byte("k1") |
| k2 = []byte("k2") |
| k3 = []byte("k3") |
| v0 = []byte("value0-012345678901234567890123012345678901234567890123") |
| v1 = []byte("value1-012345678901234567890123012345678901234567890123") |
| v2 = []byte("value2-012345678901234567890123012345678901234567890123") |
| v3 = []byte("value3-012345678901234567890123012345678901234567890123") |
| ) |
| |
| // Use a vlog with K0=V0 and a (corrupted) second transaction(k1,k2) |
| buf, offset := createMemFile(t, []*Entry{ |
| {Key: k0, Value: v0}, |
| {Key: k1, Value: v1}, |
| {Key: k2, Value: v2}, |
| }) |
| buf[offset-1]++ // Corrupt last byte |
| require.NoError(t, os.WriteFile(kv.mtFilePath(1), buf, 0777)) |
| |
| // K1 should exist, but K2 shouldn't. |
| kv, err = Open(opts) |
| require.NoError(t, err) |
| |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| // Replay should have added K0. |
| item, err := txn.Get(k0) |
| require.NoError(t, err) |
| require.Equal(t, getItemValue(t, item), v0) |
| |
| _, err = txn.Get(k1) |
| require.Equal(t, ErrKeyNotFound, err) |
| |
| _, err = txn.Get(k2) |
| require.Equal(t, ErrKeyNotFound, err) |
| return nil |
| })) |
| |
| // Write K3 at the end of the vlog. |
| txnSet(t, kv, k3, v3, 0) |
| require.NoError(t, kv.Close()) |
| |
| // The DB should contain K0 and K3 (K1 and k2 was lost when Badger started up |
| // last due to checksum failure). |
| kv, err = Open(opts) |
| require.NoError(t, err) |
| |
| { |
| txn := kv.NewTransaction(false) |
| |
| iter := txn.NewIterator(DefaultIteratorOptions) |
| iter.Seek(k0) |
| require.True(t, iter.Valid()) |
| it := iter.Item() |
| require.Equal(t, it.Key(), k0) |
| require.Equal(t, getItemValue(t, it), v0) |
| iter.Next() |
| require.True(t, iter.Valid()) |
| it = iter.Item() |
| require.Equal(t, it.Key(), k3) |
| require.Equal(t, getItemValue(t, it), v3) |
| |
| iter.Close() |
| txn.Discard() |
| } |
| |
| require.NoError(t, kv.Close()) |
| } |
| |
| // TODO: Do we need this test? |
| func TestPartialAppendToWAL(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| // Create skeleton files. |
| opts := getTestOptions(dir) |
| opts.ValueLogFileSize = 100 * 1024 * 1024 // 100Mb |
| opts.ValueThreshold = 32 |
| kv, err := Open(opts) |
| require.NoError(t, err) |
| require.NoError(t, kv.Close()) |
| |
| var ( |
| k0 = []byte("k0") |
| k1 = []byte("k1") |
| k2 = []byte("k2") |
| k3 = []byte("k3") |
| v0 = []byte("value0-01234567890123456789012012345678901234567890123") |
| v1 = []byte("value1-01234567890123456789012012345678901234567890123") |
| v2 = []byte("value2-01234567890123456789012012345678901234567890123") |
| v3 = []byte("value3-01234567890123456789012012345678901234567890123") |
| ) |
| // Values need to be long enough to actually get written to value log. |
| require.True(t, int64(len(v3)) >= kv.vlog.db.valueThreshold()) |
| |
| // Create truncated vlog to simulate a partial append. |
| // k0 - single transaction, k1 and k2 in another transaction |
| buf, offset := createMemFile(t, []*Entry{ |
| {Key: k0, Value: v0}, |
| {Key: k1, Value: v1}, |
| {Key: k2, Value: v2}, |
| }) |
| buf = buf[:offset-6] |
| require.NoError(t, os.WriteFile(kv.mtFilePath(1), buf, 0777)) |
| |
| // Badger should now start up |
| kv, err = Open(opts) |
| require.NoError(t, err) |
| |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| item, err := txn.Get(k0) |
| require.NoError(t, err) |
| require.Equal(t, v0, getItemValue(t, item)) |
| |
| _, err = txn.Get(k1) |
| require.Equal(t, ErrKeyNotFound, err) |
| _, err = txn.Get(k2) |
| require.Equal(t, ErrKeyNotFound, err) |
| return nil |
| })) |
| |
| // When K3 is set, it should be persisted after a restart. |
| txnSet(t, kv, k3, v3, 0) |
| require.NoError(t, kv.Close()) |
| kv, err = Open(opts) |
| require.NoError(t, err) |
| checkKeys(t, kv, [][]byte{k3}) |
| // Replay value log from beginning, badger head is past k2. |
| require.NoError(t, kv.vlog.Close()) |
| } |
| |
| func TestReadOnlyOpenWithPartialAppendToWAL(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| // Create skeleton files. |
| opts := getTestOptions(dir) |
| opts.ValueLogFileSize = 100 * 1024 * 1024 // 100Mb |
| kv, err := Open(opts) |
| require.NoError(t, err) |
| require.NoError(t, kv.Close()) |
| |
| var ( |
| k0 = []byte("k0") |
| k1 = []byte("k1") |
| k2 = []byte("k2") |
| v0 = []byte("value0-012345678901234567890123") |
| v1 = []byte("value1-012345678901234567890123") |
| v2 = []byte("value2-012345678901234567890123") |
| ) |
| |
| // Create truncated vlog to simulate a partial append. |
| // k0 - single transaction, k1 and k2 in another transaction |
| buf, offset := createMemFile(t, []*Entry{ |
| {Key: k0, Value: v0}, |
| {Key: k1, Value: v1}, |
| {Key: k2, Value: v2}, |
| }) |
| buf = buf[:offset-6] |
| require.NoError(t, os.WriteFile(kv.mtFilePath(1), buf, 0777)) |
| |
| opts.ReadOnly = true |
| // Badger should fail a read-only open with values to replay |
| _, err = Open(opts) |
| require.Error(t, err) |
| require.Regexp(t, "Log truncate required", err.Error()) |
| } |
| |
| func TestValueLogTrigger(t *testing.T) { |
| t.Skip("Difficult to trigger compaction, so skipping. Re-enable after fixing #226") |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| opt := getTestOptions(dir) |
| opt.ValueLogFileSize = 1 << 20 |
| kv, err := Open(opt) |
| require.NoError(t, err) |
| |
| // Write a lot of data, so it creates some work for valug log GC. |
| sz := 32 << 10 |
| txn := kv.NewTransaction(true) |
| for i := 0; i < 100; i++ { |
| v := make([]byte, sz) |
| rand.Read(v[:rand.Intn(sz)]) |
| require.NoError(t, txn.SetEntry(NewEntry([]byte(fmt.Sprintf("key%d", i)), v))) |
| if i%20 == 0 { |
| require.NoError(t, txn.Commit()) |
| txn = kv.NewTransaction(true) |
| } |
| } |
| require.NoError(t, txn.Commit()) |
| |
| for i := 0; i < 45; i++ { |
| txnDelete(t, kv, []byte(fmt.Sprintf("key%d", i))) |
| } |
| |
| require.NoError(t, kv.RunValueLogGC(0.5)) |
| |
| require.NoError(t, kv.Close()) |
| |
| err = kv.RunValueLogGC(0.5) |
| require.Equal(t, ErrRejected, err, "Error should be returned after closing DB.") |
| } |
| |
| // createMemFile creates a new memFile and returns the last valid offset. |
| func createMemFile(t *testing.T, entries []*Entry) ([]byte, uint32) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| opts := getTestOptions(dir) |
| opts.ValueLogFileSize = 100 * 1024 * 1024 // 100Mb |
| kv, err := Open(opts) |
| require.NoError(t, err) |
| defer kv.Close() |
| |
| txnSet(t, kv, entries[0].Key, entries[0].Value, entries[0].meta) |
| |
| entries = entries[1:] |
| txn := kv.NewTransaction(true) |
| for _, entry := range entries { |
| require.NoError(t, txn.SetEntry(NewEntry(entry.Key, entry.Value).WithMeta(entry.meta))) |
| } |
| require.NoError(t, txn.Commit()) |
| |
| filename := kv.mtFilePath(1) |
| buf, err := os.ReadFile(filename) |
| require.NoError(t, err) |
| return buf, kv.mt.wal.writeAt |
| } |
| |
| // This test creates two mem files and corrupts the last bit of the first file. |
| func TestPenultimateMemCorruption(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| opt := getTestOptions(dir) |
| |
| db0, err := Open(opt) |
| require.NoError(t, err) |
| defer func() { require.NoError(t, db0.Close()) }() |
| |
| h := testHelper{db: db0, t: t} |
| h.writeRange(0, 2) // 00001.mem |
| |
| // Move the current memtable to the db.imm and create a new memtable so |
| // that we can have more than one mem files. |
| require.Zero(t, len(db0.imm)) |
| db0.imm = append(db0.imm, db0.mt) |
| db0.mt, err = db0.newMemTable() |
| require.NoError(t, err) |
| |
| h.writeRange(3, 7) // 00002.mem |
| |
| // Verify we have all the data we wrote. |
| h.readRange(0, 7) |
| |
| for i := 2; i >= 1; i-- { |
| fpath := db0.mtFilePath(i) |
| fi, err := os.Stat(fpath) |
| require.NoError(t, err) |
| require.True(t, fi.Size() > 0, "Empty file at log=%d", i) |
| if i == 1 { |
| // This should corrupt the last entry in the first memtable (that is entry number 2) |
| wal := db0.imm[0].wal |
| _, err = wal.Fd.WriteAt([]byte{0}, int64(wal.writeAt-1)) |
| require.NoError(t, err) |
| // We have corrupted the file. We can remove it. If we don't remove |
| // the imm here, the db.close in defer will crash since db0.mt != |
| // db0.imm[0] |
| db0.imm = db0.imm[:0] |
| } |
| } |
| // Simulate a crash by not closing db0, but releasing the locks. |
| if db0.dirLockGuard != nil { |
| require.NoError(t, db0.dirLockGuard.release()) |
| db0.dirLockGuard = nil |
| } |
| if db0.valueDirGuard != nil { |
| require.NoError(t, db0.valueDirGuard.release()) |
| db0.valueDirGuard = nil |
| } |
| |
| db1, err := Open(opt) |
| require.NoError(t, err) |
| h.db = db1 |
| // Only 2 should be gone because it is at the end of 0001.mem (first memfile). |
| h.readRange(0, 1) |
| h.readRange(3, 7) |
| err = db1.View(func(txn *Txn) error { |
| _, err := txn.Get(h.key(2)) // Verify that 2 is gone. |
| require.Equal(t, ErrKeyNotFound, err) |
| return nil |
| }) |
| require.NoError(t, err) |
| require.NoError(t, db1.Close()) |
| } |
| |
| func checkKeys(t *testing.T, kv *DB, keys [][]byte) { |
| i := 0 |
| txn := kv.NewTransaction(false) |
| defer txn.Discard() |
| iter := txn.NewIterator(IteratorOptions{}) |
| defer iter.Close() |
| for iter.Seek(keys[0]); iter.Valid(); iter.Next() { |
| require.Equal(t, iter.Item().Key(), keys[i]) |
| i++ |
| } |
| require.Equal(t, i, len(keys)) |
| } |
| |
| type testHelper struct { |
| db *DB |
| t *testing.T |
| val []byte |
| } |
| |
| func (th *testHelper) key(i int) []byte { |
| return []byte(fmt.Sprintf("%010d", i)) |
| } |
| func (th *testHelper) value() []byte { |
| if len(th.val) > 0 { |
| return th.val |
| } |
| th.val = make([]byte, 100) |
| y.Check2(rand.Read(th.val)) |
| return th.val |
| } |
| |
| // writeRange [from, to]. |
| func (th *testHelper) writeRange(from, to int) { |
| for i := from; i <= to; i++ { |
| err := th.db.Update(func(txn *Txn) error { |
| return txn.SetEntry(NewEntry(th.key(i), th.value())) |
| }) |
| require.NoError(th.t, err) |
| } |
| } |
| |
| func TestValueGCRewriteSkipsLSMGetOnlyForExpiredEntriesInMixedVlogFile(t *testing.T) { |
| dir := t.TempDir() |
| |
| opt := getTestOptions(dir) |
| opt.ValueLogFileSize = 1 << 20 |
| opt.BaseTableSize = 1 << 15 |
| opt.ValueThreshold = 1 << 10 |
| |
| kv, err := Open(opt) |
| require.NoError(t, err) |
| defer kv.Close() |
| |
| rng := rand.New(rand.NewSource(2)) |
| const mixedEntryCount = 16 |
| const mixedLiveEntryCount = mixedEntryCount / 2 |
| const extraLiveEntryCount = 1 |
| // Leave enough room for entry metadata so all mixed entries stay in one vlog file. |
| valueSize := int(opt.ValueLogFileSize/int64(mixedEntryCount)) - 512 |
| values := make([][]byte, mixedEntryCount) |
| for i := range values { |
| values[i] = make([]byte, valueSize) |
| _, err = rng.Read(values[i]) |
| require.NoError(t, err) |
| } |
| rotatingVal := make([]byte, 32<<10) |
| _, err = rng.Read(rotatingVal) |
| require.NoError(t, err) |
| |
| // Keep several expired and live entries in the same old vlog file. |
| require.NoError(t, kv.Update(func(txn *Txn) error { |
| for i, val := range values { |
| key := []byte(fmt.Sprintf("mixed-%02d", i)) |
| entry := NewEntry(key, val) |
| if i%2 == 0 { |
| entry = entry.WithTTL(-time.Hour) |
| } |
| if err := txn.SetEntry(entry); err != nil { |
| return err |
| } |
| } |
| return nil |
| })) |
| require.NoError(t, kv.Update(func(txn *Txn) error { |
| return txn.SetEntry(NewEntry([]byte("rotating"), rotatingVal)) |
| })) |
| |
| fids := kv.vlog.sortedFids() |
| require.Len(t, fids, 2) |
| |
| kv.vlog.filesLock.RLock() |
| lf := kv.vlog.filesMap[fids[0]] |
| kv.vlog.filesLock.RUnlock() |
| |
| // Live entries require exactly one LSM lookup during rewrite (the Phase 1 |
| // liveness scan). The GC write-back race is guarded by clamping compaction's |
| // discard threshold during the rewrite (see subcompact), not by a second |
| // re-read, so no extra gets are incurred. Expired entries still require none. |
| clearAllMetrics() |
| require.NoError(t, kv.vlog.rewrite(lf)) |
| |
| totalGets := expvar.Get("badger_get_num_user") |
| require.NotNil(t, totalGets) |
| require.Equal(t, int64(mixedLiveEntryCount+extraLiveEntryCount), totalGets.(*expvar.Int).Value()) |
| |
| require.NoError(t, kv.View(func(txn *Txn) error { |
| for i, want := range values { |
| key := []byte(fmt.Sprintf("mixed-%02d", i)) |
| item, err := txn.Get(key) |
| if i%2 == 0 { |
| require.Equal(t, ErrKeyNotFound, err) |
| continue |
| } |
| require.NoError(t, err) |
| val := getItemValue(t, item) |
| require.Equal(t, want, val) |
| } |
| item, err := txn.Get([]byte("rotating")) |
| require.NoError(t, err) |
| val := getItemValue(t, item) |
| require.Equal(t, rotatingVal, val) |
| return nil |
| })) |
| } |
| |
| func (th *testHelper) readRange(from, to int) { |
| for i := from; i <= to; i++ { |
| err := th.db.View(func(txn *Txn) error { |
| item, err := txn.Get(th.key(i)) |
| if err != nil { |
| return err |
| } |
| return item.Value(func(val []byte) error { |
| require.Equal(th.t, val, th.value(), "key=%q", th.key(i)) |
| return nil |
| |
| }) |
| }) |
| require.NoError(th.t, err, "key=%q", th.key(i)) |
| } |
| } |
| |
| // Test Bug #578, which showed that if a value is moved during value log GC, an |
| // older version can end up at a higher level in the LSM tree than a newer |
| // version, causing the data to not be returned. |
| func TestBug578(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| y.Check(err) |
| defer removeDir(dir) |
| |
| db, err := Open(DefaultOptions(dir). |
| WithValueLogMaxEntries(64). |
| WithBaseTableSize(1 << 13)) |
| require.NoError(t, err) |
| |
| h := testHelper{db: db, t: t} |
| |
| // Let's run this whole thing a few times. |
| for j := 0; j < 10; j++ { |
| t.Logf("Cycle: %d\n", j) |
| h.writeRange(0, 32) |
| h.writeRange(0, 10) |
| h.writeRange(50, 72) |
| h.writeRange(40, 72) |
| h.writeRange(40, 72) |
| |
| // Run value log GC a few times. |
| for i := 0; i < 5; i++ { |
| if err := db.RunValueLogGC(0.5); err != nil && !errors.Is(ErrNoRewrite, err) { |
| require.NoError(t, err) |
| } |
| } |
| h.readRange(0, 10) |
| } |
| require.NoError(t, db.Close()) |
| } |
| |
| func BenchmarkReadWrite(b *testing.B) { |
| rwRatio := []float32{ |
| 0.1, 0.2, 0.5, 1.0, |
| } |
| valueSize := []int{ |
| 64, 128, 256, 512, 1024, 2048, 4096, 8192, 16384, |
| } |
| |
| for _, vsz := range valueSize { |
| for _, rw := range rwRatio { |
| b.Run(fmt.Sprintf("%3.1f,%04d", rw, vsz), func(b *testing.B) { |
| dir, err := os.MkdirTemp("", "vlog-benchmark") |
| y.Check(err) |
| defer removeDir(dir) |
| opts := getTestOptions(dir) |
| opts.ValueThreshold = 0 |
| db, err := Open(opts) |
| y.Check(err) |
| defer db.Close() |
| vl := &db.vlog |
| b.ResetTimer() |
| |
| for i := 0; i < b.N; i++ { |
| e := new(Entry) |
| e.Key = make([]byte, 16) |
| e.Value = make([]byte, vsz) |
| bl := new(request) |
| bl.Entries = []*Entry{e} |
| |
| var ptrs []valuePointer |
| |
| _ = vl.write([]*request{bl}) |
| ptrs = append(ptrs, bl.Ptrs...) |
| |
| f := rand.Float32() |
| if f < rw { |
| _ = vl.write([]*request{bl}) |
| |
| } else { |
| ln := len(ptrs) |
| if ln == 0 { |
| b.Fatalf("Zero length of ptrs") |
| } |
| idx := rand.Intn(ln) |
| buf, lf, err := vl.readValueBytes(ptrs[idx]) |
| if err != nil { |
| b.Fatalf("Benchmark Read: %v", err) |
| } |
| |
| e, err := lf.decodeEntry(buf, ptrs[idx].Offset) |
| require.NoError(b, err) |
| if len(e.Key) != 16 { |
| b.Fatalf("Key is invalid") |
| } |
| if len(e.Value) != vsz { |
| b.Fatalf("Value is invalid") |
| } |
| runCallback(db.vlog.getUnlockCallback(lf)) |
| } |
| } |
| }) |
| } |
| } |
| } |
| |
| // Regression test for https://github.com/dgraph-io/badger/issues/817 |
| // This test verifies if fully corrupted memtables are deleted on reopen. |
| func TestValueLogTruncate(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| // Initialize the data directory. |
| db, err := Open(DefaultOptions(dir)) |
| require.NoError(t, err) |
| // Insert 1 entry so that we have valid data in first mem file |
| require.NoError(t, db.Update(func(txn *Txn) error { |
| return txn.Set([]byte("foo"), nil) |
| })) |
| |
| fileCountBeforeCorruption := 1 |
| require.NoError(t, db.Close()) |
| |
| // Create two mem files with corrupted data. These will be truncated when DB starts next time |
| require.NoError(t, os.WriteFile(db.mtFilePath(2), []byte("foo"), 0664)) |
| require.NoError(t, os.WriteFile(db.mtFilePath(3), []byte("foo"), 0664)) |
| |
| db, err = Open(DefaultOptions(dir)) |
| require.NoError(t, err) |
| |
| // Ensure we have only one SST file. |
| require.Equal(t, 1, len(db.Tables())) |
| |
| // Ensure mem file with ID 4 is zero. |
| require.Equal(t, 4, int(db.mt.wal.fid)) |
| fileStat, err := db.mt.wal.Fd.Stat() |
| require.NoError(t, err) |
| require.Equal(t, 2*db.opt.MemTableSize, fileStat.Size()) |
| |
| fileCountAfterCorruption := len(db.Tables()) + len(db.imm) + 1 // +1 for db.mt |
| // We should have one memtable and one sst file. |
| require.Equal(t, fileCountBeforeCorruption+1, fileCountAfterCorruption) |
| // maxFid will be 2 because we increment the max fid on DB open every time. |
| require.Equal(t, 2, int(db.vlog.maxFid)) |
| require.NoError(t, db.Close()) |
| } |
| |
| func TestSafeEntry(t *testing.T) { |
| var s safeRead |
| s.lf = &logFile{} |
| e := NewEntry([]byte("foo"), []byte("bar")) |
| buf := bytes.NewBuffer(nil) |
| _, err := s.lf.encodeEntry(buf, e, 0) |
| require.NoError(t, err) |
| |
| ne, err := s.Entry(buf) |
| require.NoError(t, err) |
| require.Equal(t, e.Key, ne.Key, "key mismatch") |
| require.Equal(t, e.Value, ne.Value, "value mismatch") |
| require.Equal(t, e.meta, ne.meta, "meta mismatch") |
| require.Equal(t, e.UserMeta, ne.UserMeta, "usermeta mismatch") |
| require.Equal(t, e.ExpiresAt, ne.ExpiresAt, "expiresAt mismatch") |
| } |
| |
| func TestValueEntryChecksum(t *testing.T) { |
| k := []byte("KEY") |
| v := []byte(fmt.Sprintf("val%100d", 10)) |
| t.Run("ok", func(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| opt := getTestOptions(dir) |
| opt.VerifyValueChecksum = true |
| opt.ValueThreshold = 32 |
| db, err := Open(opt) |
| require.NoError(t, err) |
| |
| require.Greater(t, int64(len(v)), db.vlog.db.valueThreshold()) |
| txnSet(t, db, k, v, 0) |
| require.NoError(t, db.Close()) |
| |
| db, err = Open(opt) |
| require.NoError(t, err) |
| |
| txn := db.NewTransaction(false) |
| entry, err := txn.Get(k) |
| require.NoError(t, err) |
| |
| x, err := entry.ValueCopy(nil) |
| require.NoError(t, err) |
| require.Equal(t, v, x) |
| |
| require.NoError(t, db.Close()) |
| }) |
| // Regression test for https://github.com/dgraph-io/badger/issues/1049 |
| t.Run("Corruption", func(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| opt := getTestOptions(dir) |
| opt.VerifyValueChecksum = true |
| opt.ValueThreshold = 32 |
| db, err := Open(opt) |
| require.NoError(t, err) |
| |
| require.Greater(t, int64(len(v)), db.vlog.db.valueThreshold()) |
| txnSet(t, db, k, v, 0) |
| |
| path := db.vlog.fpath(1) |
| require.NoError(t, db.Close()) |
| |
| file, err := os.OpenFile(path, os.O_RDWR, 0644) |
| require.NoError(t, err) |
| offset := 50 |
| orig := make([]byte, 1) |
| _, err = file.ReadAt(orig, int64(offset)) |
| require.NoError(t, err) |
| // Corrupt a single bit. |
| _, err = file.WriteAt([]byte{7}, int64(offset)) |
| require.NoError(t, err) |
| require.NoError(t, file.Close()) |
| |
| db, err = Open(opt) |
| require.NoError(t, err) |
| |
| txn := db.NewTransaction(false) |
| entry, err := txn.Get(k) |
| require.NoError(t, err) |
| |
| // TODO(ibrahim): This test is broken since we're not returning errors |
| // in case we cannot read the values. This is incorrect behavior but |
| // we're doing this to debug an issue where the values are being read |
| // from old vlog files. |
| _, _ = entry.ValueCopy(nil) |
| // require.Error(t, err) |
| // require.Contains(t, err.Error(), "ErrEOF") |
| // require.Nil(t, x) |
| |
| require.NoError(t, db.Close()) |
| }) |
| } |
| |
| func TestValidateWrite(t *testing.T) { |
| // Mocking the file size, so that we don't allocate big memory while running test. |
| maxVlogFileSize = 400 |
| defer func() { |
| maxVlogFileSize = math.MaxUint32 |
| }() |
| |
| bigBuf := make([]byte, maxVlogFileSize+1) |
| log := &valueLog{ |
| opt: DefaultOptions("."), |
| } |
| |
| // Sending a request with big values which will overflow uint32. |
| key := []byte("HelloKey") |
| req := &request{ |
| Entries: []*Entry{ |
| { |
| Key: key, |
| Value: bigBuf, |
| }, |
| { |
| Key: key, |
| Value: bigBuf, |
| }, |
| { |
| Key: key, |
| Value: bigBuf, |
| }, |
| }, |
| } |
| |
| err := log.validateWrites([]*request{req}) |
| require.Error(t, err) |
| |
| // Testing with small values. |
| smallBuf := make([]byte, 4) |
| req1 := &request{ |
| Entries: []*Entry{ |
| { |
| Key: key, |
| Value: smallBuf, |
| }, |
| { |
| Key: key, |
| Value: smallBuf, |
| }, |
| { |
| Key: key, |
| Value: smallBuf, |
| }, |
| }, |
| } |
| |
| err = log.validateWrites([]*request{req1}) |
| require.NoError(t, err) |
| |
| // Batching small and big request. |
| err = log.validateWrites([]*request{req1, req}) |
| require.Error(t, err) |
| } |
| |
| func TestValueLogMeta(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| y.Check(err) |
| defer removeDir(dir) |
| |
| opt := getTestOptions(dir).WithValueThreshold(16) |
| db, _ := Open(opt) |
| defer db.Close() |
| txn := db.NewTransaction(true) |
| for i := 0; i < 10; i++ { |
| k := []byte(fmt.Sprintf("key=%d", i)) |
| v := []byte(fmt.Sprintf("val=%020d", i)) |
| require.NoError(t, txn.SetEntry(NewEntry(k, v))) |
| } |
| require.NoError(t, txn.Commit()) |
| fids := db.vlog.sortedFids() |
| require.Equal(t, 1, len(fids)) |
| |
| // vlog entries must not have txn meta. |
| _, err = db.vlog.filesMap[fids[0]].iterate(true, 0, func(e Entry, vp valuePointer) error { |
| require.Zero(t, e.meta&(bitTxn|bitFinTxn)) |
| return nil |
| }) |
| require.NoError(t, err) |
| |
| // Entries in LSM tree must have txn bit of meta set |
| txn = db.NewTransaction(false) |
| defer txn.Discard() |
| iopt := DefaultIteratorOptions |
| key := []byte("key") |
| iopt.Prefix = key |
| itr := txn.NewIterator(iopt) |
| defer itr.Close() |
| var count int |
| for itr.Seek(key); itr.ValidForPrefix(key); itr.Next() { |
| item := itr.Item() |
| require.Equal(t, bitTxn, item.meta&(bitTxn|bitFinTxn)) |
| count++ |
| } |
| require.Equal(t, 10, count) |
| } |
| |
| // This tests asserts the condition that vlog fids start from 1. |
| // TODO(naman): should this be changed to assert instead? |
| func TestFirstVlogFile(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-test") |
| require.NoError(t, err) |
| defer removeDir(dir) |
| |
| opt := DefaultOptions(dir) |
| db, err := Open(opt) |
| require.NoError(t, err) |
| defer db.Close() |
| |
| fids := db.vlog.sortedFids() |
| require.NotZero(t, len(fids)) |
| require.Equal(t, uint32(1), fids[0]) |
| } |
| |
| // TestVlogGCRaceDeletedKeyReappearsReal is the end-to-end regression test for |
| // #2286. It exercises the genuine bug machinery through real operations only: |
| // |
| // - The delete is a REAL transaction (txn.Delete + CommitAt) → it flows through |
| // the normal write path (writeToLSM), exactly as a user's delete would. |
| // - The purge is a REAL last-level compaction (flush to L0 + runCompactDef into |
| // Lmax) → it drops both the tombstone and foo@v100 through the real |
| // subcompact drop path, exactly as background compaction would. |
| // |
| // The only non-production element is vlogGCPauseHook, which merely PAUSES the |
| // rewrite between Phase 1 (scan) and Phase 2 (write-back) so the real |
| // delete+compaction lands deterministically in the race window. No tombstone is |
| // hand-built (no createAndOpen), and the LSM is never hand-edited. The test |
| // reproduces the bug on unfixed code (verified) and is independent of how the fix |
| // is implemented. |
| func TestVlogGCRaceDeletedKeyReappearsReal(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-vlog-gc-real") |
| require.NoError(t, err) |
| defer os.RemoveAll(dir) |
| |
| openDB := func(compactOnClose bool) *DB { |
| opt := DefaultOptions(dir). |
| WithNumCompactors(0). |
| WithValueThreshold(1 << 10). // 1 KB: 4 KB value → vlog |
| WithCompression(options.None) // predictable SST sizes |
| opt.managedTxns = true |
| opt = opt.WithCompactL0OnClose(compactOnClose) |
| db, err := Open(opt) |
| require.NoError(t, err) |
| return db |
| } |
| |
| key := []byte("foo") |
| largeVal := bytes.Repeat([]byte("x"), 4<<10) // 4 KB > ValueThreshold |
| |
| // ── Setup: write foo@v100 (large value → vlog), compact to L6. ── |
| db := openDB(true) |
| txn := db.NewTransactionAt(0, true) |
| require.NoError(t, txn.SetEntry(NewEntry(key, largeVal))) |
| require.NoError(t, txn.CommitAt(100, nil)) |
| require.NoError(t, db.Close()) // value bytes → vlog; LSM pointer → L6 |
| |
| // ── Reopen, set hook, call rewrite(). ── |
| db = openDB(false) |
| |
| db.vlogGCPauseHook = func() { |
| // (1) REAL delete through the normal write path. CommitAt(_, nil) blocks |
| // until the write is applied, so the tombstone@v101 is in the memtable |
| // (and has passed through writeToLSM) by the time this returns. |
| dtxn := db.NewTransactionAt(101, true) |
| require.NoError(t, dtxn.Delete(key)) |
| require.NoError(t, dtxn.CommitAt(101, nil)) |
| |
| // discardTs set AFTER the commit (else lastCleanupTs would exceed the |
| // commit ts and trip an assert). 200 > 101 so the tombstone is droppable. |
| db.SetDiscardTs(200) |
| |
| // (2) REAL flush: move the memtable holding the tombstone out to an L0 |
| // SST via the production flush routine. |
| db.lock.Lock() |
| oldMt := db.mt |
| nmt, err := db.newMemTable() |
| require.NoError(t, err) |
| db.mt = nmt |
| db.lock.Unlock() |
| require.NoError(t, db.handleMemTableFlush(oldMt, nil)) |
| oldMt.DecrRef() |
| |
| // (3) REAL last-level compaction L0 → L6. nextLevel is Lmax, so there is |
| // no level below it: hasOverlap is false and the subcompact drop path |
| // purges BOTH the tombstone@v101 and foo@v100 (discardTs=200 ≥ 101). |
| targets := db.lc.levelTargets() |
| cdef := compactDef{ |
| thisLevel: db.lc.levels[0], |
| nextLevel: db.lc.levels[6], |
| top: db.lc.levels[0].tables, |
| bot: db.lc.levels[6].tables, |
| t: targets, |
| } |
| require.NoError(t, db.lc.runCompactDef(-1, 0, cdef)) |
| |
| db.vlogGCPauseHook = nil |
| } |
| |
| db.vlog.filesLock.RLock() |
| maxFid := db.vlog.maxFid |
| var lf *logFile |
| for _, f := range db.vlog.filesMap { |
| if f.fid < maxFid { |
| lf = f |
| break |
| } |
| } |
| db.vlog.filesLock.RUnlock() |
| require.NotNil(t, lf, "data vlog file must exist after reopen") |
| |
| require.NoError(t, db.vlog.rewrite(lf)) |
| |
| err = db.View(func(txn *Txn) error { |
| _, err := txn.Get(key) |
| return err |
| }) |
| require.Equal(t, ErrKeyNotFound, err, |
| "deleted key must not reappear after real delete + real compaction during GC") |
| |
| require.NoError(t, db.Close()) |
| } |
| |
| // TestVlogGCRaceDeletedKeyReappearsMultiCycle runs several GC rewrites |
| // back-to-back on the same DB, deleting a different key during each one. It |
| // guards the cross-cycle correctness of the discardTs clamp: that gcActive is |
| // re-set and released every cycle, that a fresh gcDiscardTs is pinned each time, |
| // and that no key deleted in any cycle ever reappears in a later one. |
| func TestVlogGCRaceDeletedKeyReappearsMultiCycle(t *testing.T) { |
| dir, err := os.MkdirTemp("", "badger-vlog-gc-multicycle") |
| require.NoError(t, err) |
| defer os.RemoveAll(dir) |
| |
| openDB := func(compactOnClose bool) *DB { |
| opt := DefaultOptions(dir). |
| WithNumCompactors(0). |
| WithValueThreshold(1 << 10). |
| WithValueLogMaxEntries(1). // one value per vlog file → one GC-able file per key |
| WithCompression(options.None) |
| opt.managedTxns = true |
| opt = opt.WithCompactL0OnClose(compactOnClose) |
| db, err := Open(opt) |
| require.NoError(t, err) |
| return db |
| } |
| |
| const cycles = 3 |
| largeVal := bytes.Repeat([]byte("x"), 4<<10) // 4 KB > ValueThreshold → vlog |
| |
| // Stage each key in its own vlog file; pointers compacted to L6 on close. |
| db := openDB(true) |
| for i := 0; i < cycles; i++ { |
| txn := db.NewTransactionAt(0, true) |
| require.NoError(t, txn.SetEntry(NewEntry([]byte(fmt.Sprintf("foo%d", i)), largeVal))) |
| require.NoError(t, txn.CommitAt(uint64(10*(i+1)), nil)) |
| } |
| require.NoError(t, db.Close()) |
| |
| db = openDB(false) |
| defer func() { require.NoError(t, db.Close()) }() |
| require.False(t, db.gcActive.Load(), "gcActive must be false before any GC") |
| |
| deleted := map[string]bool{} |
| |
| for c := 0; c < cycles; c++ { |
| // Pick a data vlog file (fid < maxFid) whose live key is not yet deleted, |
| // and learn which key it holds by iterating it (robust to fid ordering). |
| db.vlog.filesLock.RLock() |
| maxFid := db.vlog.maxFid |
| fids := make([]uint32, 0, len(db.vlog.filesMap)) |
| for fid := range db.vlog.filesMap { |
| fids = append(fids, fid) |
| } |
| filesMap := db.vlog.filesMap |
| db.vlog.filesLock.RUnlock() |
| sort.Slice(fids, func(i, j int) bool { return fids[i] < fids[j] }) |
| |
| // A sealed file may hold more than one key, so scan the whole file for the |
| // first key that is still live (not deleted in a prior cycle). |
| var lf *logFile |
| var fileKey []byte |
| for _, fid := range fids { |
| if fid >= maxFid { |
| continue |
| } |
| cand := filesMap[fid] |
| var pick []byte |
| _, _ = cand.iterate(true, 0, func(e Entry, vp valuePointer) error { |
| if e.meta&bitDelete == 0 { |
| if uk := y.ParseKey(e.Key); len(uk) > 0 && !deleted[string(uk)] { |
| pick = append([]byte{}, uk...) |
| return errStop |
| } |
| } |
| return nil |
| }) |
| if pick != nil { |
| lf, fileKey = cand, pick |
| break |
| } |
| } |
| require.NotNil(t, lf, "cycle %d: a data vlog file with an undeleted key must exist", c) |
| |
| delTs := uint64(1000 + c*100) |
| var pinned uint64 |
| db.vlogGCPauseHook = func() { |
| require.True(t, db.gcActive.Load(), "cycle %d: gcActive must be true during rewrite", c) |
| pinned = db.gcDiscardTs.Load() |
| |
| // Real delete through the write path. |
| dtxn := db.NewTransactionAt(delTs, true) |
| require.NoError(t, dtxn.Delete(fileKey)) |
| require.NoError(t, dtxn.CommitAt(delTs, nil)) |
| db.SetDiscardTs(delTs + 50) |
| |
| // Real purge: flush the tombstone to L0, then compact L0→L6 (Lmax). |
| // The clamp keeps the tombstone (version delTs >= gcDiscardTs); without |
| // the fix it would be dropped here together with the value. |
| db.lock.Lock() |
| oldMt := db.mt |
| nmt, err := db.newMemTable() |
| require.NoError(t, err) |
| db.mt = nmt |
| db.lock.Unlock() |
| require.NoError(t, db.handleMemTableFlush(oldMt, nil)) |
| oldMt.DecrRef() |
| |
| targets := db.lc.levelTargets() |
| cdef := compactDef{ |
| thisLevel: db.lc.levels[0], |
| nextLevel: db.lc.levels[6], |
| top: db.lc.levels[0].tables, |
| bot: db.lc.levels[6].tables, |
| t: targets, |
| } |
| require.NoError(t, db.lc.runCompactDef(-1, 0, cdef)) |
| db.vlogGCPauseHook = nil |
| } |
| |
| require.NoError(t, db.vlog.rewrite(lf)) |
| |
| require.False(t, db.gcActive.Load(), "cycle %d: gcActive must reset to false after rewrite", c) |
| require.NotZero(t, pinned, "cycle %d: gcDiscardTs must be pinned during the rewrite", c) |
| require.GreaterOrEqual(t, delTs, pinned, |
| "cycle %d: delete ts must be >= the pinned discard ts for the clamp to protect it", c) |
| |
| deleted[string(fileKey)] = true |
| |
| // Every key deleted so far must stay gone. |
| for k := range deleted { |
| gotErr := db.View(func(txn *Txn) error { |
| _, e := txn.Get([]byte(k)) |
| return e |
| }) |
| require.Equal(t, ErrKeyNotFound, gotErr, "cycle %d: key %q must stay deleted", c, k) |
| } |
| } |
| |
| require.Len(t, deleted, cycles, "each cycle must have deleted a distinct key") |
| } |