mirror of
https://github.com/luxfi/zapdb.git
synced 2026-10-30 15:12:20 +00:00
zeekayandHanzo Dev
fe8d85d663
The module has been github.com/luxfi/zapdb for a while, but the package inside it still declared itself badger, so every consumer had to write zapdb "github.com/luxfi/zapdb" // package badger and the alias existed only to paper over the mismatch. 56 root files and the 134 qualified references inside the repo now say zapdb. Backward compatible in both directions: an existing consumer with the alias keeps compiling (the alias is now merely redundant), and one without it starts compiling instead of failing on an undefined zapdb. Directory badger/ and its cmd package are untouched — that is the CLI, and its binary is named badger. Build and vet clean; suite green except TestZRestore, which wants a /tmp/zb/src benchmark fixture and fails identically without this change. Co-authored-by: Hanzo Dev <[email protected]>
840 lines
21 KiB
Go
840 lines
21 KiB
Go
/*
|
|
* SPDX-FileCopyrightText: © 2017-2025 Istari Digital, Inc.
|
|
* SPDX-License-Identifier: Apache-2.0
|
|
*/
|
|
|
|
package zapdb
|
|
|
|
import (
|
|
"fmt"
|
|
"math"
|
|
"math/rand"
|
|
"os"
|
|
"runtime"
|
|
"sync"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
|
|
"github.com/dgraph-io/ristretto/v2/z"
|
|
"github.com/luxfi/zapdb/y"
|
|
)
|
|
|
|
func val(large bool) []byte {
|
|
var buf []byte
|
|
if large {
|
|
buf = make([]byte, 8192)
|
|
} else {
|
|
buf = make([]byte, 16)
|
|
}
|
|
rand.Read(buf)
|
|
return buf
|
|
}
|
|
|
|
func numKeys(db *DB) int {
|
|
var count int
|
|
err := db.View(func(txn *Txn) error {
|
|
itr := txn.NewIterator(DefaultIteratorOptions)
|
|
defer itr.Close()
|
|
|
|
for itr.Rewind(); itr.Valid(); itr.Next() {
|
|
count++
|
|
}
|
|
return nil
|
|
})
|
|
y.Check(err)
|
|
return count
|
|
}
|
|
|
|
func numKeysManaged(db *DB, readTs uint64) int {
|
|
txn := db.NewTransactionAt(readTs, false)
|
|
defer txn.Discard()
|
|
|
|
itr := txn.NewIterator(DefaultIteratorOptions)
|
|
defer itr.Close()
|
|
|
|
var count int
|
|
for itr.Rewind(); itr.Valid(); itr.Next() {
|
|
count++
|
|
}
|
|
return count
|
|
}
|
|
|
|
func TestDropAllManaged(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.managedTxns = true
|
|
opts.ValueLogFileSize = 5 << 20
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
|
|
N := uint64(10000)
|
|
populate := func(db *DB, start uint64) {
|
|
var wg sync.WaitGroup
|
|
for i := start; i < start+N; i++ {
|
|
wg.Add(1)
|
|
txn := db.NewTransactionAt(math.MaxUint64, true)
|
|
require.NoError(t, txn.SetEntry(NewEntry([]byte(key("key", int(i))), val(true))))
|
|
require.NoError(t, txn.CommitAt(i, func(err error) {
|
|
require.NoError(t, err)
|
|
wg.Done()
|
|
}))
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
populate(db, N)
|
|
require.Equal(t, int(N), numKeysManaged(db, math.MaxUint64))
|
|
|
|
require.NoError(t, db.DropAll())
|
|
require.NoError(t, db.DropAll()) // Just call it twice, for fun.
|
|
require.Equal(t, 0, numKeysManaged(db, math.MaxUint64))
|
|
|
|
// Check that we can still write to db, and using lower timestamps.
|
|
populate(db, 1)
|
|
require.Equal(t, int(N), numKeysManaged(db, math.MaxUint64))
|
|
require.NoError(t, db.Close())
|
|
|
|
// Ensure that value log is correctly replayed, that we are preserving badgerHead.
|
|
opts.managedTxns = true
|
|
db2, err := Open(opts)
|
|
require.NoError(t, err)
|
|
require.Equal(t, int(N), numKeysManaged(db2, math.MaxUint64))
|
|
require.NoError(t, db2.Close())
|
|
}
|
|
|
|
func TestDropAll(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.ValueLogFileSize = 5 << 20
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
|
|
N := uint64(10000)
|
|
populate := func(db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(true)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
|
|
require.NoError(t, db.DropAll())
|
|
require.Equal(t, 0, numKeys(db))
|
|
|
|
// Check that we can still write to mdb, and using lower timestamps.
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
require.NoError(t, db.Close())
|
|
|
|
// Ensure that value log is correctly replayed.
|
|
db2, err := Open(opts)
|
|
require.NoError(t, err)
|
|
require.Equal(t, int(N), numKeys(db2))
|
|
require.NoError(t, db2.Close())
|
|
}
|
|
|
|
func TestDropAllTwice(t *testing.T) {
|
|
test := func(t *testing.T, opts Options) {
|
|
db, err := Open(opts)
|
|
|
|
require.NoError(t, err)
|
|
defer func() {
|
|
require.NoError(t, db.Close())
|
|
}()
|
|
|
|
N := uint64(10000)
|
|
populate := func(db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(false)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
|
|
require.NoError(t, db.DropAll())
|
|
require.Equal(t, 0, numKeys(db))
|
|
|
|
// Call DropAll again.
|
|
require.NoError(t, db.DropAll())
|
|
require.NoError(t, db.Close())
|
|
}
|
|
t.Run("disk mode", func(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.ValueLogFileSize = 5 << 20
|
|
test(t, opts)
|
|
})
|
|
t.Run("InMemory mode", func(t *testing.T) {
|
|
opts := getTestOptions("")
|
|
opts.InMemory = true
|
|
test(t, opts)
|
|
})
|
|
}
|
|
|
|
func TestDropAllWithPendingTxn(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.ValueLogFileSize = 5 << 20
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
defer func() {
|
|
require.NoError(t, db.Close())
|
|
}()
|
|
|
|
N := uint64(10000)
|
|
populate := func(db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(true)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
|
|
txn := db.NewTransaction(true)
|
|
|
|
var wg sync.WaitGroup
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
itr := txn.NewIterator(DefaultIteratorOptions)
|
|
defer itr.Close()
|
|
|
|
var keys []string
|
|
for {
|
|
var count int
|
|
for itr.Rewind(); itr.Valid(); itr.Next() {
|
|
count++
|
|
item := itr.Item()
|
|
keys = append(keys, string(item.KeyCopy(nil)))
|
|
_, err := item.ValueCopy(nil)
|
|
if err != nil {
|
|
t.Logf("Got error during value copy: %v", err)
|
|
return
|
|
}
|
|
}
|
|
t.Logf("Got number of keys: %d\n", count)
|
|
for _, key := range keys {
|
|
item, err := txn.Get([]byte(key))
|
|
if err != nil {
|
|
t.Logf("Got error during key lookup: %v", err)
|
|
return
|
|
}
|
|
if _, err := item.ValueCopy(nil); err != nil {
|
|
t.Logf("Got error during second value copy: %v", err)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
// Do not cancel txn.
|
|
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
time.Sleep(2 * time.Second)
|
|
require.NoError(t, db.DropAll())
|
|
}()
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestDropReadOnly(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.ValueLogFileSize = 5 << 20
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
N := uint64(1000)
|
|
populate := func(db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(true)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
require.NoError(t, db.Close())
|
|
|
|
opts.ReadOnly = true
|
|
db2, err := Open(opts)
|
|
// acquireDirectoryLock returns ErrWindowsNotSupported on Windows. It can be ignored safely.
|
|
if runtime.GOOS == "windows" {
|
|
require.Equal(t, err, ErrWindowsNotSupported)
|
|
} else {
|
|
require.NoError(t, err)
|
|
require.Panics(t, func() { _ = db2.DropAll() })
|
|
require.NoError(t, db2.Close())
|
|
}
|
|
}
|
|
|
|
func TestWriteAfterClose(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.ValueLogFileSize = 5 << 20
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
N := uint64(1000)
|
|
populate := func(db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(true)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
require.NoError(t, db.Close())
|
|
err = db.Update(func(txn *Txn) error {
|
|
return txn.SetEntry(NewEntry([]byte("a"), []byte("b")))
|
|
})
|
|
require.Equal(t, ErrDBClosed, err)
|
|
}
|
|
|
|
func TestDropAllRace(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.managedTxns = true
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
|
|
N := 10000
|
|
// Start a goroutine to keep trying to write to DB while DropAll happens.
|
|
closer := z.NewCloser(1)
|
|
go func() {
|
|
defer closer.Done()
|
|
ticker := time.NewTicker(time.Millisecond)
|
|
defer ticker.Stop()
|
|
|
|
i := N + 1 // Writes would happen above N.
|
|
var errors atomic.Int32
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
i++
|
|
txn := db.NewTransactionAt(math.MaxUint64, true)
|
|
require.NoError(t, txn.SetEntry(NewEntry([]byte(key("key", i)), val(false))))
|
|
if err := txn.CommitAt(uint64(i), func(err error) {
|
|
if err != nil {
|
|
errors.Add(1)
|
|
}
|
|
}); err != nil {
|
|
errors.Add(1)
|
|
}
|
|
case <-closer.HasBeenClosed():
|
|
// The following causes a data race.
|
|
// t.Logf("i: %d. Number of (expected) write errors: %d.\n", i, errors)
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 1; i <= N; i++ {
|
|
wg.Add(1)
|
|
txn := db.NewTransactionAt(math.MaxUint64, true)
|
|
require.NoError(t, txn.SetEntry(NewEntry([]byte(key("key", i)), val(false))))
|
|
require.NoError(t, txn.CommitAt(uint64(i), func(err error) {
|
|
require.NoError(t, err)
|
|
wg.Done()
|
|
}))
|
|
}
|
|
wg.Wait()
|
|
|
|
before := numKeysManaged(db, math.MaxUint64)
|
|
require.True(t, before > N)
|
|
|
|
require.NoError(t, db.DropAll())
|
|
closer.SignalAndWait()
|
|
|
|
after := numKeysManaged(db, math.MaxUint64)
|
|
t.Logf("Before: %d. After dropall: %d\n", before, after)
|
|
require.True(t, after < before)
|
|
db.Close()
|
|
}
|
|
|
|
func TestDropPrefix(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.ValueLogFileSize = 5 << 20
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
|
|
N := uint64(10000)
|
|
populate := func(db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(true)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
|
|
require.NoError(t, db.DropPrefix([]byte("key000")))
|
|
require.Equal(t, int(N)-10, numKeys(db))
|
|
|
|
require.NoError(t, db.DropPrefix([]byte("key00")))
|
|
require.Equal(t, int(N)-100, numKeys(db))
|
|
|
|
expected := int(N)
|
|
for i := 0; i < 10; i++ {
|
|
require.NoError(t, db.DropPrefix([]byte(fmt.Sprintf("key%d", i))))
|
|
expected -= 1000
|
|
require.Equal(t, expected, numKeys(db))
|
|
}
|
|
require.NoError(t, db.DropPrefix([]byte("key1")))
|
|
require.Equal(t, 0, numKeys(db))
|
|
require.NoError(t, db.DropPrefix([]byte("key")))
|
|
require.Equal(t, 0, numKeys(db))
|
|
|
|
// Check that we can still write to mdb, and using lower timestamps.
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
require.NoError(t, db.DropPrefix([]byte("key")))
|
|
require.Equal(t, 0, numKeys(db))
|
|
require.NoError(t, db.Close())
|
|
|
|
// Ensure that value log is correctly replayed.
|
|
db2, err := Open(opts)
|
|
require.NoError(t, err)
|
|
require.Equal(t, 0, numKeys(db2))
|
|
require.NoError(t, db2.Close())
|
|
}
|
|
|
|
func TestDropPrefixWithPendingTxn(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.ValueLogFileSize = 5 << 20
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
defer func() {
|
|
require.NoError(t, db.Close())
|
|
}()
|
|
|
|
N := uint64(10000)
|
|
populate := func(db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(true)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
|
|
txn := db.NewTransaction(true)
|
|
|
|
var wg sync.WaitGroup
|
|
wg.Add(2)
|
|
go func() {
|
|
defer wg.Done()
|
|
itr := txn.NewIterator(DefaultIteratorOptions)
|
|
defer itr.Close()
|
|
|
|
var keys []string
|
|
for {
|
|
var count int
|
|
for itr.Rewind(); itr.Valid(); itr.Next() {
|
|
count++
|
|
item := itr.Item()
|
|
keys = append(keys, string(item.KeyCopy(nil)))
|
|
_, err := item.ValueCopy(nil)
|
|
if err != nil {
|
|
t.Logf("Got error during value copy: %v", err)
|
|
return
|
|
}
|
|
}
|
|
t.Logf("Got number of keys: %d\n", count)
|
|
for _, key := range keys {
|
|
item, err := txn.Get([]byte(key))
|
|
if err != nil {
|
|
t.Logf("Got error during key lookup: %v", err)
|
|
return
|
|
}
|
|
if _, err := item.ValueCopy(nil); err != nil {
|
|
t.Logf("Got error during second value copy: %v", err)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
// Do not cancel txn.
|
|
|
|
go func() {
|
|
defer wg.Done()
|
|
time.Sleep(2 * time.Second)
|
|
require.NoError(t, db.DropPrefix([]byte("key0")))
|
|
require.NoError(t, db.DropPrefix([]byte("key00")))
|
|
require.NoError(t, db.DropPrefix([]byte("key")))
|
|
}()
|
|
wg.Wait()
|
|
}
|
|
|
|
func TestDropPrefixReadOnly(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.ValueLogFileSize = 5 << 20
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
N := uint64(1000)
|
|
populate := func(db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(true)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
populate(db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
require.NoError(t, db.Close())
|
|
|
|
opts.ReadOnly = true
|
|
db2, err := Open(opts)
|
|
// acquireDirectoryLock returns ErrWindowsNotSupported on Windows. It can be ignored safely.
|
|
if runtime.GOOS == "windows" {
|
|
require.Equal(t, err, ErrWindowsNotSupported)
|
|
} else {
|
|
require.NoError(t, err)
|
|
require.Panics(t, func() { _ = db2.DropPrefix([]byte("key0")) })
|
|
require.NoError(t, db2.Close())
|
|
}
|
|
}
|
|
|
|
func TestDropPrefixRace(t *testing.T) {
|
|
dir, err := os.MkdirTemp("", "badger-test")
|
|
require.NoError(t, err)
|
|
defer removeDir(dir)
|
|
opts := getTestOptions(dir)
|
|
opts.managedTxns = true
|
|
db, err := Open(opts)
|
|
require.NoError(t, err)
|
|
|
|
N := 10000
|
|
// Start a goroutine to keep trying to write to DB while DropPrefix happens.
|
|
closer := z.NewCloser(1)
|
|
go func() {
|
|
defer closer.Done()
|
|
ticker := time.NewTicker(time.Millisecond)
|
|
defer ticker.Stop()
|
|
|
|
i := N + 1 // Writes would happen above N.
|
|
var errors atomic.Int32
|
|
for {
|
|
select {
|
|
case <-ticker.C:
|
|
i++
|
|
txn := db.NewTransactionAt(math.MaxUint64, true)
|
|
require.NoError(t, txn.SetEntry(NewEntry([]byte(key("key", i)), val(false))))
|
|
if err := txn.CommitAt(uint64(i), func(err error) {
|
|
if err != nil {
|
|
errors.Add(1)
|
|
}
|
|
}); err != nil {
|
|
errors.Add(1)
|
|
}
|
|
case <-closer.HasBeenClosed():
|
|
// The following causes a data race.
|
|
// t.Logf("i: %d. Number of (expected) write errors: %d.\n", i, errors)
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
var wg sync.WaitGroup
|
|
for i := 1; i <= N; i++ {
|
|
wg.Add(1)
|
|
txn := db.NewTransactionAt(math.MaxUint64, true)
|
|
require.NoError(t, txn.SetEntry(NewEntry([]byte(key("key", i)), val(false))))
|
|
require.NoError(t, txn.CommitAt(uint64(i), func(err error) {
|
|
require.NoError(t, err)
|
|
wg.Done()
|
|
}))
|
|
}
|
|
wg.Wait()
|
|
|
|
before := numKeysManaged(db, math.MaxUint64)
|
|
require.True(t, before > N)
|
|
|
|
require.NoError(t, db.DropPrefix([]byte("key00")))
|
|
require.NoError(t, db.DropPrefix([]byte("key1")))
|
|
require.NoError(t, db.DropPrefix([]byte("key")))
|
|
closer.SignalAndWait()
|
|
|
|
after := numKeysManaged(db, math.MaxUint64)
|
|
t.Logf("Before: %d. After dropprefix: %d\n", before, after)
|
|
require.True(t, after < before)
|
|
require.NoError(t, db.Close())
|
|
}
|
|
|
|
func TestWriteBatchManagedMode(t *testing.T) {
|
|
key := func(i int) []byte {
|
|
return []byte(fmt.Sprintf("%10d", i))
|
|
}
|
|
val := func(i int) []byte {
|
|
return []byte(fmt.Sprintf("%128d", i))
|
|
}
|
|
opt := DefaultOptions("")
|
|
opt.managedTxns = true
|
|
opt.BaseTableSize = 1 << 20 // This would create multiple transactions in write batch.
|
|
runBadgerTest(t, &opt, func(t *testing.T, db *DB) {
|
|
wb := db.NewWriteBatchAt(1)
|
|
defer wb.Cancel()
|
|
|
|
N, M := 50000, 1000
|
|
start := time.Now()
|
|
|
|
for i := 0; i < N; i++ {
|
|
require.NoError(t, wb.Set(key(i), val(i)))
|
|
}
|
|
for i := 0; i < M; i++ {
|
|
require.NoError(t, wb.Delete(key(i)))
|
|
}
|
|
require.NoError(t, wb.Flush())
|
|
t.Logf("Time taken for %d writes (w/ test options): %s\n", N+M, time.Since(start))
|
|
|
|
err := db.View(func(txn *Txn) error {
|
|
itr := txn.NewIterator(DefaultIteratorOptions)
|
|
defer itr.Close()
|
|
|
|
i := M
|
|
for itr.Rewind(); itr.Valid(); itr.Next() {
|
|
item := itr.Item()
|
|
require.Equal(t, string(key(i)), string(item.Key()))
|
|
require.Equal(t, item.Version(), uint64(1))
|
|
valcopy, err := item.ValueCopy(nil)
|
|
require.NoError(t, err)
|
|
require.Equal(t, val(i), valcopy)
|
|
i++
|
|
}
|
|
require.Equal(t, N, i)
|
|
return nil
|
|
})
|
|
require.NoError(t, err)
|
|
})
|
|
}
|
|
func TestWriteBatchManaged(t *testing.T) {
|
|
key := func(i int) []byte {
|
|
return []byte(fmt.Sprintf("%10d", i))
|
|
}
|
|
val := func(i int) []byte {
|
|
return []byte(fmt.Sprintf("%128d", i))
|
|
}
|
|
opt := DefaultOptions("")
|
|
opt.managedTxns = true
|
|
opt.BaseTableSize = 1 << 15 // This would create multiple transactions in write batch.
|
|
runBadgerTest(t, &opt, func(t *testing.T, db *DB) {
|
|
wb := db.NewManagedWriteBatch()
|
|
defer wb.Cancel()
|
|
|
|
N, M := 50000, 1000
|
|
start := time.Now()
|
|
|
|
for i := 0; i < N; i++ {
|
|
require.NoError(t, wb.SetEntryAt(&Entry{Key: key(i), Value: val(i)}, 1))
|
|
}
|
|
for i := 0; i < M; i++ {
|
|
require.NoError(t, wb.DeleteAt(key(i), 2))
|
|
}
|
|
require.NoError(t, wb.Flush())
|
|
t.Logf("Time taken for %d writes (w/ test options): %s\n", N+M, time.Since(start))
|
|
|
|
err := db.View(func(txn *Txn) error {
|
|
itr := txn.NewIterator(DefaultIteratorOptions)
|
|
defer itr.Close()
|
|
|
|
i := M
|
|
for itr.Rewind(); itr.Valid(); itr.Next() {
|
|
item := itr.Item()
|
|
require.Equal(t, string(key(i)), string(item.Key()))
|
|
require.Equal(t, item.Version(), uint64(1))
|
|
valcopy, err := item.ValueCopy(nil)
|
|
require.NoError(t, err)
|
|
require.Equal(t, val(i), valcopy)
|
|
i++
|
|
}
|
|
require.Equal(t, N, i)
|
|
return nil
|
|
})
|
|
require.NoError(t, err)
|
|
})
|
|
}
|
|
|
|
func TestWriteBatchDuplicate(t *testing.T) {
|
|
N := 10
|
|
k := []byte("key")
|
|
v := []byte("val")
|
|
readVerify := func(t *testing.T, db *DB, n int, versions []int) {
|
|
err := db.View(func(txn *Txn) error {
|
|
iopt := DefaultIteratorOptions
|
|
iopt.AllVersions = true
|
|
itr := txn.NewIterator(iopt)
|
|
defer itr.Close()
|
|
|
|
i := 0
|
|
for itr.Rewind(); itr.Valid(); itr.Next() {
|
|
item := itr.Item()
|
|
require.Equal(t, k, item.Key())
|
|
require.Equal(t, uint64(versions[i]), item.Version())
|
|
err := item.Value(func(val []byte) error {
|
|
require.Equal(t, v, val)
|
|
return nil
|
|
})
|
|
require.NoError(t, err)
|
|
i++
|
|
}
|
|
require.Equal(t, n, i)
|
|
return nil
|
|
})
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
t.Run("writebatch", func(t *testing.T) {
|
|
opt := DefaultOptions("")
|
|
opt.BaseTableSize = 1 << 15 // This would create multiple transactions in write batch.
|
|
|
|
runBadgerTest(t, &opt, func(t *testing.T, db *DB) {
|
|
wb := db.NewWriteBatch()
|
|
defer wb.Cancel()
|
|
|
|
for i := uint64(0); i < uint64(N); i++ {
|
|
// Multiple versions of the same key.
|
|
require.NoError(t, wb.SetEntry(&Entry{Key: k, Value: v}))
|
|
}
|
|
require.NoError(t, wb.Flush())
|
|
readVerify(t, db, 1, []int{1})
|
|
})
|
|
})
|
|
t.Run("writebatch at", func(t *testing.T) {
|
|
opt := DefaultOptions("")
|
|
opt.BaseTableSize = 1 << 15 // This would create multiple transactions in write batch.
|
|
opt.managedTxns = true
|
|
|
|
runBadgerTest(t, &opt, func(t *testing.T, db *DB) {
|
|
wb := db.NewWriteBatchAt(10)
|
|
defer wb.Cancel()
|
|
|
|
for i := uint64(0); i < uint64(N); i++ {
|
|
// Multiple versions of the same key.
|
|
require.NoError(t, wb.SetEntry(&Entry{Key: k, Value: v}))
|
|
}
|
|
require.NoError(t, wb.Flush())
|
|
readVerify(t, db, 1, []int{10})
|
|
})
|
|
|
|
})
|
|
t.Run("managed writebatch", func(t *testing.T) {
|
|
opt := DefaultOptions("")
|
|
opt.managedTxns = true
|
|
opt.BaseTableSize = 1 << 15 // This would create multiple transactions in write batch.
|
|
runBadgerTest(t, &opt, func(t *testing.T, db *DB) {
|
|
wb := db.NewManagedWriteBatch()
|
|
defer wb.Cancel()
|
|
|
|
for i := uint64(1); i <= uint64(N); i++ {
|
|
// Multiple versions of the same key.
|
|
require.NoError(t, wb.SetEntryAt(&Entry{Key: k, Value: v}, i))
|
|
}
|
|
require.NoError(t, wb.Flush())
|
|
readVerify(t, db, N, []int{10, 9, 8, 7, 6, 5, 4, 3, 2, 1})
|
|
})
|
|
})
|
|
}
|
|
|
|
func TestZeroDiscardStats(t *testing.T) {
|
|
N := uint64(10000)
|
|
populate := func(t *testing.T, db *DB) {
|
|
writer := db.NewWriteBatch()
|
|
for i := uint64(0); i < N; i++ {
|
|
require.NoError(t, writer.Set([]byte(key("key", int(i))), val(true)))
|
|
}
|
|
require.NoError(t, writer.Flush())
|
|
}
|
|
|
|
t.Run("after rewrite", func(t *testing.T) {
|
|
opts := getTestOptions("")
|
|
opts.ValueLogFileSize = 5 << 20
|
|
opts.ValueThreshold = 1 << 10
|
|
opts.MemTableSize = 1 << 15
|
|
runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
|
|
populate(t, db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
|
|
fids := db.vlog.sortedFids()
|
|
for _, fid := range fids {
|
|
db.vlog.discardStats.Update(fid, 1)
|
|
}
|
|
|
|
// Ensure we have some valid fids.
|
|
require.True(t, len(fids) > 2)
|
|
fid := fids[0]
|
|
require.NoError(t, db.vlog.rewrite(db.vlog.filesMap[fid]))
|
|
// All data should still be present.
|
|
require.Equal(t, int(N), numKeys(db))
|
|
|
|
db.vlog.discardStats.Iterate(func(id, val uint64) {
|
|
// Vlog with id=fid has been re-written, it's discard stats should be zero.
|
|
if uint32(id) == fid {
|
|
require.Zero(t, val)
|
|
}
|
|
})
|
|
})
|
|
})
|
|
t.Run("after dropall", func(t *testing.T) {
|
|
opts := getTestOptions("")
|
|
opts.ValueLogFileSize = 5 << 20
|
|
runBadgerTest(t, &opts, func(t *testing.T, db *DB) {
|
|
populate(t, db)
|
|
require.Equal(t, int(N), numKeys(db))
|
|
|
|
// Fill discard stats. Normally these are filled by compaction.
|
|
fids := db.vlog.sortedFids()
|
|
for _, fid := range fids {
|
|
db.vlog.discardStats.Update(fid, 1)
|
|
}
|
|
|
|
db.vlog.discardStats.Iterate(func(id, val uint64) { require.NotZero(t, val) })
|
|
require.NoError(t, db.DropAll())
|
|
require.Equal(t, 0, numKeys(db))
|
|
// We've deleted everything. DS should be zero.
|
|
db.vlog.discardStats.Iterate(func(id, val uint64) { require.Zero(t, val) })
|
|
})
|
|
})
|
|
}
|