rosedb

Documentation

Index

• Variables
• type Batch
  • func (b *Batch) Commit() error
  • func (b *Batch) Delete(key []byte) error
  • func (b *Batch) Exist(key []byte) (bool, error)
  • func (b *Batch) Expire(key []byte, ttl time.Duration) error
  • func (b *Batch) Get(key []byte) ([]byte, error)
  • func (b *Batch) Persist(key []byte) error
  • func (b *Batch) Put(key []byte, value []byte) error
  • func (b *Batch) PutWithTTL(key []byte, value []byte, ttl time.Duration) error
  • func (b *Batch) Rollback() error
  • func (b *Batch) TTL(key []byte) (time.Duration, error)
• type BatchOptions
• type DB
  • func Open(options Options) (*DB, error)
  • func (db *DB) Ascend(handleFn func(k []byte, v []byte) (bool, error))
  • func (db *DB) AscendGreaterOrEqual(key []byte, handleFn func(k []byte, v []byte) (bool, error))
  • func (db *DB) AscendKeys(pattern []byte, filterExpired bool, handleFn func(k []byte) (bool, error))
  • func (db *DB) AscendRange(startKey, endKey []byte, handleFn func(k []byte, v []byte) (bool, error))
  • func (db *DB) Close() error
  • func (db *DB) Delete(key []byte) error
  • func (db *DB) DeleteExpiredKeys(timeout time.Duration) error
  • func (db *DB) Descend(handleFn func(k []byte, v []byte) (bool, error))
  • func (db *DB) DescendKeys(pattern []byte, filterExpired bool, handleFn func(k []byte) (bool, error))
  • func (db *DB) DescendLessOrEqual(key []byte, handleFn func(k []byte, v []byte) (bool, error))
  • func (db *DB) DescendRange(startKey, endKey []byte, handleFn func(k []byte, v []byte) (bool, error))
  • func (db *DB) Exist(key []byte) (bool, error)
  • func (db *DB) Expire(key []byte, ttl time.Duration) error
  • func (db *DB) Get(key []byte) ([]byte, error)
  • func (db *DB) Merge(reopenAfterDone bool) error
  • func (db *DB) NewBatch(options BatchOptions) *Batch
  • func (db *DB) Persist(key []byte) error
  • func (db *DB) Put(key []byte, value []byte) error
  • func (db *DB) PutWithTTL(key []byte, value []byte, ttl time.Duration) error
  • func (db *DB) Stat() *Stat
  • func (db *DB) Sync() error
  • func (db *DB) TTL(key []byte) (time.Duration, error)
  • func (db *DB) Watch() (<-chan *Event, error)
• type Event
• type IndexRecord
• type LogRecord
  • func (lr *LogRecord) IsExpired(now int64) bool
• type LogRecordType
• type Options
• type Stat
• type WatchActionType
• type Watcher
  • func NewWatcher(capacity uint64) *Watcher

Variables

var (
	ErrKeyIsEmpty      = errors.New("the key is empty")
	ErrKeyNotFound     = errors.New("key not found in database")
	ErrDatabaseIsUsing = errors.New("the database directory is used by another process")
	ErrReadOnlyBatch   = errors.New("the batch is read only")
	ErrBatchCommitted  = errors.New("the batch is committed")
	ErrBatchRolledBack = errors.New("the batch is rolled back")
	ErrDBClosed        = errors.New("the database is closed")
	ErrMergeRunning    = errors.New("the merge operation is running")
	ErrWatchDisabled   = errors.New("the watch is disabled")
)

var DefaultBatchOptions = BatchOptions{
	Sync:     true,
	ReadOnly: false,
}

var DefaultOptions = Options{
	DirPath:           "./data",
	SegmentSize:       1 * wal.GB,
	BlockCache:        0,
	Sync:              false,
	BytesPerSync:      0,
	WatchQueueSize:    0,
	AutoMergeCronExpr: "",
}

Types

type Batch

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

func (*Batch) Commit

func (b *Batch) Commit() error

Commit commits the batch, if the batch is readonly or empty, it will return directly.

It will iterate the pendingWrites and write the data to the database, then write a record to indicate the end of the batch to guarantee atomicity. Finally, it will write the index.

func (*Batch) Delete

func (b *Batch) Delete(key []byte) error

func (*Batch) Exist

func (b *Batch) Exist(key []byte) (bool, error)

func (*Batch) Expire

func (b *Batch) Expire(key []byte, ttl time.Duration) error

func (*Batch) Get

func (b *Batch) Get(key []byte) ([]byte, error)

func (*Batch) Persist

func (b *Batch) Persist(key []byte) error

Persist 将过期时间设置为不过期 相当于keep live

func (*Batch) Put

func (b *Batch) Put(key []byte, value []byte) error

func (*Batch) PutWithTTL

func (b *Batch) PutWithTTL(key []byte, value []byte, ttl time.Duration) error

func (*Batch) Rollback

func (b *Batch) Rollback() error

Rollback discards an uncommitted batch instance. the discard operation will clear the buffered data and release the lock.

func (*Batch) TTL

func (b *Batch) TTL(key []byte) (time.Duration, error)

type BatchOptions

type BatchOptions struct {
	Sync     bool
	ReadOnly bool
}

type DB

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

func Open

func Open(options Options) (*DB, error)

Open a database with the specified options. If the database directory does not exist, it will be created automatically.

Multiple processes can not use the same database directory at the same time, otherwise it will return ErrDatabaseIsUsing.

It will open the wal files in the database directory and load the index from them. Return the DB instance, or an error if any.

func (*DB) Ascend

func (db *DB) Ascend(handleFn func(k []byte, v []byte) (bool, error))

Ascend calls handleFn for each key/value pair in the db in ascending order.

func (*DB) AscendGreaterOrEqual

func (db *DB) AscendGreaterOrEqual(key []byte, handleFn func(k []byte, v []byte) (bool, error))

AscendGreaterOrEqual calls handleFn for each key/value pair in the db with keys greater than or equal to the given key.

func (*DB) AscendKeys

func (db *DB) AscendKeys(pattern []byte, filterExpired bool, handleFn func(k []byte) (bool, error))

AscendKeys calls handleFn for each key in the db in ascending order. Since our expiry time is stored in the value, if you want to filter expired keys, you need to set parameter filterExpired to true. But the performance will be affected. Because we need to read the value of each key to determine if it is expired.

func (*DB) AscendRange

func (db *DB) AscendRange(startKey, endKey []byte, handleFn func(k []byte, v []byte) (bool, error))

AscendRange calls handleFn for each key/value pair in the db within the range [startKey, endKey] in ascending order.

func (*DB) Close

func (db *DB) Close() error

func (*DB) Delete

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

func (*DB) DeleteExpiredKeys

func (db *DB) DeleteExpiredKeys(timeout time.Duration) error

DeleteExpiredKeys scan the entire index in ascending order to delete expired keys. It is a time-consuming operation, so we need to specify a timeout to prevent the DB from being unavailable for a long time.

func (*DB) Descend

func (db *DB) Descend(handleFn func(k []byte, v []byte) (bool, error))

Descend calls handleFn for each key/value pair in the db in descending order.

func (*DB) DescendKeys

func (db *DB) DescendKeys(pattern []byte, filterExpired bool, handleFn func(k []byte) (bool, error))

DescendKeys calls handleFn for each key in the db in descending order. Since our expiry time is stored in the value, if you want to filter expired keys, you need to set parameter filterExpired to true. But the performance will be affected. Because we need to read the value of each key to determine if it is expired.

func (*DB) DescendLessOrEqual

func (db *DB) DescendLessOrEqual(key []byte, handleFn func(k []byte, v []byte) (bool, error))

DescendLessOrEqual calls handleFn for each key/value pair in the db with keys less than or equal to the given key.

func (*DB) DescendRange

func (db *DB) DescendRange(startKey, endKey []byte, handleFn func(k []byte, v []byte) (bool, error))

DescendRange calls handleFn for each key/value pair in the db within the range [startKey, endKey] in descending order.

func (*DB) Exist

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

func (*DB) Expire

func (db *DB) Expire(key []byte, ttl time.Duration) error

func (*DB) Get

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

func (*DB) Merge

func (db *DB) Merge(reopenAfterDone bool) error

Merge merges all the data files in the database. It will iterate all the data files, find the valid data, and rewrite the data to the new data file.

Merge operation maybe a very time-consuming operation when the database is large. So it is recommended to perform this operation when the database is idle.

If reopenAfterDone is true, the original file will be replaced by the merge file, and db's index will be rebuilt after the merge completes.

func (*DB) NewBatch

func (db *DB) NewBatch(options BatchOptions) *Batch

NewBatch creates a new Batch instance.

func (*DB) Persist

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

func (*DB) Put

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

func (*DB) PutWithTTL

func (db *DB) PutWithTTL(key []byte, value []byte, ttl time.Duration) error

func (*DB) Stat

func (db *DB) Stat() *Stat

Stat returns the statistics of the database.

func (*DB) Sync

func (db *DB) Sync() error

Sync all data files to the underlying storage.

func (*DB) TTL

func (db *DB) TTL(key []byte) (time.Duration, error)

func (*DB) Watch

func (db *DB) Watch() (<-chan *Event, error)

type Event

type Event struct {
	Action  WatchActionType
	Key     []byte
	Value   []byte
	BatchId uint64
}

type IndexRecord

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

IndexRecord is the index record of the key. It contains the key, the record type and the position of the record in the wal. Only used in start up to rebuild the index.

type LogRecord

type LogRecord struct {
	Key     []byte
	Value   []byte
	Type    LogRecordType
	BatchId uint64
	Expire  int64
}

LogRecord is the log record of the key/value pair. It contains the key, the value, the record type and the batch id It will be encoded to byte slice and written to the wal.

func (*LogRecord) IsExpired

func (lr *LogRecord) IsExpired(now int64) bool

IsExpired checks whether the log record is expired.

type LogRecordType

type LogRecordType = byte

LogRecordType is the type of the log record.

const (
	// LogRecordNormal is the normal log record type.
	LogRecordNormal LogRecordType = iota
	// LogRecordDeleted is the deleted log record type.
	LogRecordDeleted
	// LogRecordBatchFinished is the batch finished log record type.
	LogRecordBatchFinished
)

type Options

type Options struct {
	// DirPath specifies the directory path where the WAL segment files will be stored.
	DirPath string

	// SegmentSize specifies the maximum size of each segment file in bytes.
	SegmentSize int64

	// BlockCache specifies the size of the block cache in number of bytes.
	// A block cache is used to store recently accessed data blocks, improving read performance.
	// If BlockCache is set to 0, no block cache will be used.
	BlockCache uint32

	// Sync is whether to synchronize writes through os buffer cache and down onto the actual disk.
	// Setting sync is required for durability of a single write operation, but also results in slower writes.
	//
	// If false, and the machine crashes, then some recent writes may be lost.
	// Note that if it is just the process that crashes (machine does not) then no writes will be lost.
	//
	// In other words, Sync being false has the same semantics as a write
	// system call. Sync being true means write followed by fsync.
	Sync bool

	// BytesPerSync specifies the number of bytes to write before calling fsync.
	BytesPerSync uint32

	// WatchQueueSize the cache length of the watch queue.
	// if the size greater than 0, which means enable the watch.
	WatchQueueSize uint64

	// AutoMergeEnable enable the auto merge.
	// auto merge will be triggered when cron expr is satisfied.
	// cron expression follows the standard cron expression.
	// e.g. "0 0 * * *" means merge at 00:00:00 every day.
	// it also supports seconds optionally.
	// when enable the second field, the cron expression will be like this: "0/10 * * * * *" (every 10 seconds).
	// when auto merge is enabled, the db will be closed and reopened after merge done.
	// do not set this shecule too frequently, it will affect the performance.
	// refer to https://en.wikipedia.org/wiki/Cron
	AutoMergeCronExpr string
}

type Stat

type Stat struct {
	KeysNum  int
	DiskSize int64
}

type WatchActionType

type WatchActionType = byte

const (
	WatchActionPut WatchActionType = iota
	WatchActionDelete
)

type Watcher

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

func NewWatcher

func NewWatcher(capacity uint64) *Watcher