simple-go-kv
/
batch_test.go

package simple_bitcask_kv

import (
	"github.com/stretchr/testify/assert"
	"simple_bitcask_kv/utils"
	"testing"
)

// / **********单个事务测试*********
// 要求不管WriteBatch是否提交，都能Get到已经写入的数据
func TestDB_WBGetCommit(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 100; i++ {
		db.RawPut(utils.GetTestKey(i), utils.GetTestKey(i))
	}

	wb := db.NewWriteBatch(DefaultWBOptions)
	for i := 100; i < 200; i++ {
		wb.Put(utils.GetTestKey(i), utils.GetTestKey(i))
	}
	// 提交前也能看到自己的写入  和之前已有写入
	for i := 0; i < 200; i++ {
		val, err := wb.Get(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}

	// 正式提交事务
	wb.Commit()

	// 提交后更加能看到自己的写入！
	for i := 0; i < 200; i++ {
		val, err := wb.Get(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}
	// 提交后直接RawGet也能看到写入
	for i := 0; i < 200; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}
}

// Rollback的数据就没法再看到了
func TestDB_WBGetRollback(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 100; i++ {
		db.RawPut(utils.GetTestKey(i), utils.GetTestKey(i))
	}

	wb := db.NewWriteBatch(DefaultWBOptions)
	for i := 100; i < 200; i++ {
		wb.Put(utils.GetTestKey(i), utils.GetTestKey(i))
	}
	// 提交前也能看到自己的写入  和之前已有写入
	for i := 0; i < 200; i++ {
		val, err := wb.Get(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}
	wb.Rollback()
	// 回滚后没法看到自己的写入！
	for i := 100; i < 200; i++ {
		val, err := wb.Get(utils.GetTestKey(i))
		assert.NotNil(t, err)
		assert.Equal(t, ErrKeyNotFound, err)
		assert.Nil(t, val)
	}
	// 通过RawGet也没法看到
	for i := 100; i < 200; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.NotNil(t, err)
		assert.Equal(t, ErrKeyNotFound, err)
		assert.Nil(t, val)
	}
	// 但是仍能看到之前的成功写入
	for i := 0; i < 100; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}
}

// / 测试重启后能否看到Commit/Rollback事务的数据
func TestDB_WBGetCommitRestart(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	db.RawPut(utils.GetTestKey(0), utils.GetTestKey(0))

	wb := db.NewWriteBatch(DefaultWBOptions)
	wb.Put(utils.GetTestKey(1), utils.GetTestKey(1))

	// 提交前也能看到自己的写入  和之前已有写入
	for i := 0; i < 2; i++ {
		val, err := wb.Get(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}

	// 正式提交事务
	wb.Commit()
	db.Sync()

	// restart
	db.Close()
	db, err = Open(opts)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	// 提交后直接RawGet也能看到所有写入
	val, err := db.RawGet(utils.GetTestKey(0))
	assert.Nil(t, err)
	assert.NotNil(t, val)
	assert.Equal(t, utils.GetTestKey(0), val)

	val, err = db.RawGet(utils.GetTestKey(1))
	assert.Nil(t, err)
	assert.NotNil(t, val)
	assert.Equal(t, utils.GetTestKey(1), val)
}

func TestDB_WBGetCommitRestartM(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 100; i++ {
		db.RawPut(utils.GetTestKey(i), utils.GetTestKey(i))
	}

	wb := db.NewWriteBatch(DefaultWBOptions)
	for i := 100; i < 200; i++ {
		wb.Put(utils.GetTestKey(i), utils.GetTestKey(i))
	}
	// 提交前也能看到自己的写入  和之前已有写入
	for i := 0; i < 200; i++ {
		val, err := wb.Get(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}

	// 正式提交事务
	wb.Commit()
	db.Sync()

	// restart
	db.Close()
	db, err = Open(opts)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	// 提交后直接RawGet也能看到所有写入
	for i := 0; i < 100; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}
	// TODO: 发现这里重启后各种seqNo读出来是98，但其实应该是2（commmit记录的seqNo确实也是2）
	for i := 100; i < 200; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}
}

// Rollback的数据就没法再看到了
func TestDB_WBGetRollbackRestart(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 100; i++ {
		db.RawPut(utils.GetTestKey(i), utils.GetTestKey(i))
	}

	wb := db.NewWriteBatch(DefaultWBOptions)
	for i := 100; i < 200; i++ {
		wb.Put(utils.GetTestKey(i), utils.GetTestKey(i))
	}
	// 提交前也能看到自己的写入  和之前已有写入
	for i := 0; i < 200; i++ {
		val, err := wb.Get(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}
	wb.Rollback()

	// restart
	db.Close()
	db, err = Open(opts)

	// 回滚后没法看到写入，但是仍能看到原有写入
	// 通过RawGet也没法看到
	for i := 100; i < 200; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.NotNil(t, err)
		assert.Equal(t, ErrKeyNotFound, err)
		assert.Nil(t, val)
	}
	// 但是仍能看到之前的成功写入
	for i := 0; i < 100; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}
}

// 测试两个事务之间的隔离性

// 保证不能看到其他事务未提交的修改
func TestWriteBatch_TxnIsolation1(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 10; i++ {
		db.RawPut(utils.GetTestKey(i), utils.GetTestKey(i))
	}

	wb1 := db.NewWriteBatch(DefaultWBOptions)
	wb2 := db.NewWriteBatch(DefaultWBOptions)

	// 先检验两个WB都能看到已有数据
	for i := 0; i < 10; i++ {
		val, err := wb1.Get(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
		val, err = wb2.Get(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.NotNil(t, val)
		assert.Equal(t, utils.GetTestKey(i), val)
	}

	// 各自修改不同的数据，能看到自己的修改，但是看不到对方的修改
	val1 := []byte("by wb1")
	wb1.Put(utils.GetTestKey(1), val1)

	// wb1能看到自己修改，不能看到wb2修改
	val, err := wb1.Get(utils.GetTestKey(1))
	assert.Nil(t, err)
	assert.Equal(t, val1, val)

	// wb2不能看到wb1的修改
	val, err = wb2.Get(utils.GetTestKey(1))
	assert.Nil(t, err)
	assert.Equal(t, utils.GetTestKey(1), val)

	wb1.Rollback()
	wb2.Rollback()
}

// 能看到已有事务提交的数据
func TestWriteBatch_TxnIsolation2(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 10; i++ {
		db.RawPut(utils.GetTestKey(i), utils.GetTestKey(i))
	}

	wb1 := db.NewWriteBatch(DefaultWBOptions)
	wb2 := db.NewWriteBatch(DefaultWBOptions)

	val1 := []byte("by wb1")
	wb1.Put(utils.GetTestKey(1), val1)
	wb1.Commit()

	// wb2能看到wb1已经提交的修改
	val, err := wb2.Get(utils.GetTestKey(1))
	assert.Nil(t, err)
	assert.Equal(t, val1, val)
}

func TestDB_WriteBatch(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	wb := db.NewWriteBatch(DefaultWBOptions)
	for i := 0; i < 100; i++ {
		wb.Put(utils.GetTestKey(i), utils.GetTestKey(i))
	}
	wb.Commit()

	db.Close()

	// restart
	db, err = Open(opts)
	wb = db.NewWriteBatch(DefaultWBOptions) // 奇怪的是我之前wb没有重新生成也正确，很奇怪，wb中的db应该要变了才对
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 100; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.Equal(t, val, utils.GetTestKey(i))
	}

	for i := 100; i < 200; i++ {
		wb.Put(utils.GetTestKey(i), utils.GetTestKey(i))
	}

	db.Close() // 100~200这部分数据应该写入不成个

	db, err = Open(opts)
	wb = db.NewWriteBatch(DefaultWBOptions)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 100; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.Equal(t, val, utils.GetTestKey(i))
	}

	for i := 100; i < 200; i++ { // 没有Commit的事务不会写入
		_, err := db.RawGet(utils.GetTestKey(i))
		assert.NotNil(t, err)
		assert.Equal(t, ErrKeyNotFound, err)
	}

	// 测试删除
	for i := 50; i < 100; i++ {
		wb.Delete(utils.GetTestKey(i))
	}
	wb.Commit()

	for i := 0; i < 50; i++ {
		val, err := db.RawGet(utils.GetTestKey(i))
		assert.Nil(t, err)
		assert.Equal(t, val, utils.GetTestKey(i))
	}

	for i := 50; i < 200; i++ {
		_, err := db.RawGet(utils.GetTestKey(i))
		assert.NotNil(t, err)
		assert.Equal(t, ErrKeyNotFound, err)
	}
}

func TestDB_WBExceedMaxNum(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	wb := db.NewWriteBatch(DefaultWBOptions)
	for i := 0; i < 1001; i++ {
		wb.Put(utils.GetTestKey(i), utils.GetTestKey(i))
	}

	err = wb.Commit()
	assert.NotNil(t, err)
	assert.Equal(t, ErrExceedMaxBatchNum, err)
}

// 写入很长一串，手动终止，看看能不能恢复
func TestDB_Kill(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	//defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	wbopts := DefaultWBOptions
	wbopts.MaxBatchNum = 100000000
	wb := db.NewWriteBatch(wbopts)
	for i := 0; i < 100000000; i++ {
		wb.Put(utils.GetTestKey(i), utils.GetTestKey(1024))
		t.Log(i)
	}
	wb.Commit()
	// 执行到一半直接kill 然后执行TestDB_Restore应该没法读到数据
}

// 试图恢复之前的数据，但是没有commit成功，应该会失败，一条都读不出来，但是数据文件中应该是能看到数据的（但是没有commit符号）
func TestDB_Restore(t *testing.T) {
	opts := DefaultOptions
	opts.DataFileMaxSize = 64 * 1024 * 1024
	db, err := Open(opts)
	defer destroyDB(db)
	assert.Nil(t, err)
	assert.NotNil(t, db)

	for i := 0; i < 100000000; i++ {
		_, err := db.RawGet(utils.GetTestKey(i))
		assert.NotNil(t, err)
		assert.Equal(t, ErrKeyNotFound, err)
	}
}