rdb

README

English | 简体中文

RDB is a high-performance, Bitcask-based, embeddable Key-Value storage engine implemented in Go. It employs a log-structured merge (LSM-like) storage approach with multiple index type support, offering a rich set of features and functionalities.

paper address

Features

• Multiple Index Types Support
  • B-Tree Index
  • Adaptive Radix Tree (ART) Index
  • B+ Tree Index (with persistence support)
• High-performance Read/Write Operations
• Transaction Support
• Data Persistence and Recovery
• Data File Merge Operations
• MMap Loading Support
• Batch Write Operations
• Database Backup Support
• Iterator Support
• File Locking for Data Safety

Design Details

1. Storage Engine Architecture

The storage engine uses a log-structured approach with the following components:

• Data Files: Size-based segmented files, including active and archived files
• Memory Index: Support for multiple index implementations
• Write-Ahead Log: Ensures data durability and consistency
• Merge Mechanism: Space reclamation through merge operations
• Transaction Management: Supports atomic operations

2. Data Organization

• Data Files:

  • Configurable file size (default 256MB)
  • Append-only writing
  • MMap loading support

• Index Structures:

  • B-Tree: Balanced tree index, suitable for general scenarios
  • ART: Adaptive Radix Tree, memory-efficient
  • B+ Tree: Persistent tree-based index

3. Main Configuration configs

type configs struct {
    DirPath            string      // Database directory path
    FileSize       int64       // Size of data files
    SyncWrites         bool        // Whether to sync writes
    IndexType          IndexerType // Type of index to use
    BytesPerSync       int         // Bytes to accumulate before sync
    MMapAtStartup      bool        // Whether to use MMap at startup
    DataFileMergeRatio float32     // Threshold for data file merging
}

Usage Examples

// Open database
configs := rdbrdb.DefaultOptions
configs.DirPath = "/tmp/rdb"
db, err := rdbrdb.Open(configs)

if err != nil {
    panic(err)
}
defer db.Close()

// Write data
err = db.Put([]byte("key"), []byte("value"))

// Read data
value, err := db.Get([]byte("key"))

// Delete data
err = db.Delete([]byte("key"))

// Batch write

batch := db.NewWriteBatch(rdbrdb.DefaultWriteBatchConfigs)

batch.Put([]byte("key1"), []byte("value1"))
batch.Put([]byte("key2"), []byte("value2"))
err = batch.Commit()

Advanced Features

1. Iterator

Support for iterating over data in key dictionary order:

configs := rdbrdb.DefaultIteratorConfigs
iterator := db.Iterator(configs)
for iterator.Rewind(); iterator.Valid(); iterator.Next() {
    key := iterator.Key()
    value := iterator.Value()
}

2. Database Backup

Support for online database backup:

err := db.Backup("/path/to/backup")

3. Transaction Support

Provides atomic batch write support:

batch := db.NewWriteBatch(rdbrdb.DefaultWriteBatchConfigs)
defer batch.Commit()

batch.Put([]byte("key1"), []byte("value1"))
batch.Delete([]byte("key2"))

Performance Optimizations

1. MMap Loading: Uses memory mapping for accelerated data loading
2. Batch Writing: Efficient batch write operations through WriteBatch
3. Async Persistence: Configurable data accumulation before persistence
4. Space Reclamation: Invalid data space reclamation through merge operations

Important Notes

1. Ensure proper read/write permissions for the data directory
2. Configure appropriate data file size and merge threshold
3. Choose suitable index type based on your scenario
4. Ensure all operations are complete before closing the database

Contributing

Issues and Pull Requests are welcome!

License

MIT License

Documentation

Index

• Variables
• type Configs
• type DB
  • func Open(configs Configs) (*DB, error)
  • func (db *DB) Backup(dir string) error
  • func (db *DB) Close() error
  • func (db *DB) Delete(key []byte) error
  • func (db *DB) Fold(fn func(key []byte, value []byte) bool) error
  • func (db *DB) Get(key []byte) ([]byte, error)
  • func (db *DB) ListKeys() [][]byte
  • func (db *DB) Merge() error
  • func (db *DB) NewIterator(opts IteratorConfigs) *Iterator
  • func (db *DB) NewWriteBatch(opts WriteBatchConfigs) *WriteBatch
  • func (db *DB) Put(key []byte, value []byte) error
  • func (db *DB) Stat() *Stat
  • func (db *DB) Sync() error
• type IndexerType
• type Iterator
  • func (it *Iterator) Close()
  • func (it *Iterator) Key() []byte
  • func (it *Iterator) Next()
  • func (it *Iterator) Rewind()
  • func (it *Iterator) Seek(key []byte)
  • func (it *Iterator) Valid() bool
  • func (it *Iterator) Value() ([]byte, error)
• type IteratorConfigs
• type Stat
• type WriteBatch
  • func (wb *WriteBatch) Commit() error
  • func (wb *WriteBatch) Delete(key []byte) error
  • func (wb *WriteBatch) Put(key []byte, value []byte) error
• type WriteBatchConfigs

Variables

var (
	ErrKeyIsEmpty             = errors.New("the key is empty")
	ErrKeyNotFound            = errors.New("key not found in database")
	ErrDataFileNotFound       = errors.New("data file is not found")
	ErrDataDirectoryCorrupted = errors.New("the database directory maybe corrupted")
	ErrIndexUpdateFailed      = errors.New("failed to update index")
	ErrExceedMaxBatchNum      = errors.New("exceed the max batch num")
	ErrMergeInProgress        = errors.New("merge is in progress, try again later")
	ErrMergeRatioUnreached    = errors.New("the merge ratio do not reach the option")
	ErrNoEnoughSpaceForMerge  = errors.New("no enough disk space for merge")
	ErrDatabaseIsUsing        = errors.New("database directory is using by another process")
)

var DefaultIteratorConfigs = IteratorConfigs{
	Prefix:  nil,
	Reverse: false,
}

var DefaultOptions = Configs{
	DirPath:            os.TempDir(),
	FileSize:           256 * 1024 * 1024,
	SyncWrites:         false,
	IndexType:          BTree,
	BytesPerSync:       0,
	MMapAtStartup:      true,
	DataFileMergeRatio: 0.5,
}

var DefaultWriteBatchConfigs = WriteBatchConfigs{
	MaxBatchNum: 10000,
	SyncWrites:  true,
}

Types

type Configs

type Configs struct {
	// 数据库数据目录
	DirPath string

	// 数据文件的大小
	FileSize int64

	// 每次写数据是否持久化
	SyncWrites bool

	// 索引类型
	IndexType IndexerType

	// 积累多少字节写入后进行持久化
	BytesPerSync int

	// 启动时是否使用 MMap 加载数据
	MMapAtStartup bool

	// 数据文件合并的阈值
	DataFileMergeRatio float32
}

Configs 配置项的结构体、用户可传递过来的配置项

type DB

type DB struct {
	// contains filtered or unexported fields
}

DB represents a key-value storage engine instance.

func Open

func Open(configs Configs) (*DB, error)

Open opens or creates a DB at the specified path with the given config. If the directory does not exist, it will be created.

func (*DB) Backup

func (db *DB) Backup(dir string) error

Backup backs up the database to the specified directory.

func (*DB) Close

func (db *DB) Close() error

Close 关闭数据库

func (*DB) Delete

func (db *DB) Delete(key []byte) error

Delete removes the value for the given key. If the key does not exist, no error is returned.

func (*DB) Fold

func (db *DB) Fold(fn func(key []byte, value []byte) bool) error

Fold 获取所有的数据，并执行用户指定的操作，函数返回 false 时终止遍历

func (*DB) Get

func (db *DB) Get(key []byte) ([]byte, error)

Get retrieves the value for the given key. Returns ErrKeyNotFound if the key does not exist.

func (*DB) ListKeys

func (db *DB) ListKeys() [][]byte

ListKeys 获取数据库中所有的 key

func (*DB) Merge

func (db *DB) Merge() error

func (*DB) NewIterator

func (db *DB) NewIterator(opts IteratorConfigs) *Iterator

NewIterator 初始化迭代器

func (*DB) NewWriteBatch

func (db *DB) NewWriteBatch(opts WriteBatchConfigs) *WriteBatch

NewWriteBatch new WriteBatch

func (*DB) Put

func (db *DB) Put(key []byte, value []byte) error

Put stores a key-value pair in the database. If the key already exists, its previous value will be overwritten.

func (*DB) Stat

func (db *DB) Stat() *Stat

Stat 返回数据库的相关统计信息

func (*DB) Sync

func (db *DB) Sync() error

Sync 持久化数据文件

type IndexerType

type IndexerType = int8

const (
	// BTree 索引
	BTree IndexerType = iota + 1

	// ART Adpative Radix Tree 自适应基数树索引
	ART

	// BPlusTree B+ 树索引，将索引存储到磁盘上
	BPlusTree
)

type Iterator

type Iterator struct {
	// contains filtered or unexported fields
}

Iterator 迭代器

func (*Iterator) Close

func (it *Iterator) Close()

Close 关闭迭代器，释放相应资源

func (*Iterator) Key

func (it *Iterator) Key() []byte

Key 当前遍历位置的 Key 数据

func (*Iterator) Next

func (it *Iterator) Next()

Next 跳转到下一个 key

func (*Iterator) Rewind

func (it *Iterator) Rewind()

Rewind 重新回到迭代器的起点，即第一个数据

func (*Iterator) Seek

func (it *Iterator) Seek(key []byte)

Seek 根据传入的 key 查找到第一个大于（或小于）等于的目标 key，根据从这个 key 开始遍历

func (*Iterator) Valid

func (it *Iterator) Valid() bool

Valid 是否有效，即是否已经遍历完了所有的 key，用于退出遍历

func (*Iterator) Value

func (it *Iterator) Value() ([]byte, error)

Value 当前遍历位置的 Value 数据

type IteratorConfigs

type IteratorConfigs struct {
	// 遍历前缀为指定值的 Key，默认为空
	Prefix []byte
	// 是否反向遍历，默认 false 是正向
	Reverse bool
}

IteratorConfigs 索引迭代器配置项

type Stat

type Stat struct {
	KeyNum          uint  // key 的总数量
	DataFileNum     uint  // 数据文件的数量
	ReclaimableSize int64 // 可以进行 merge 回收的数据量，字节为单位
	DiskSize        int64 // 数据目录所占磁盘空间大小
}

Stat 存储引擎统计信息

type WriteBatch

type WriteBatch struct {
	// contains filtered or unexported fields
}

func (*WriteBatch) Commit

func (wb *WriteBatch) Commit() error

Commit 提交事务，将暂存的数据写到数据文件，并更新内存索引

func (*WriteBatch) Delete

func (wb *WriteBatch) Delete(key []byte) error

Delete 删除数据

func (*WriteBatch) Put

func (wb *WriteBatch) Put(key []byte, value []byte) error

type WriteBatchConfigs

type WriteBatchConfigs struct {
	// 一个批次当中最大的数据量
	MaxBatchNum uint

	// 提交时是否 sync 持久化
	SyncWrites bool
}

WriteBatchConfigs 批量写配置项