database

Documentation

Index

• Constants
• Variables
• func EncodeEntrySpec(key []byte, md *store.KVMetadata, value []byte) *store.EntrySpec
• func EncodeKey(key []byte) []byte
• func EncodeReference(key []byte, md *store.KVMetadata, referencedKey []byte, atTx uint64) *store.EntrySpec
• func EncodeZAdd(set []byte, score float64, key []byte, atTx uint64) *store.EntrySpec
• func PreconditionFromProto(c *schema.Precondition) (store.Precondition, error)
• func TrimPrefix(prefixed []byte) []byte
• func WrapReferenceValueAt(key []byte, atTx uint64) []byte
• func WrapWithPrefix(b []byte, prefix byte) []byte
• func WrapZAddReferenceAt(set []byte, score float64, key []byte, atTx uint64) []byte
• type DB
  • func NewDB(dbName string, multidbHandler sql.MultiDBHandler, op *Options, ...) (DB, error)
  • func OpenDB(dbName string, multidbHandler sql.MultiDBHandler, op *Options, ...) (DB, error)
• type DatabaseList
  • func NewDatabaseList() DatabaseList
• type Options
  • func DefaultOption() *Options
  • func (o *Options) AsReplica(replica bool) *Options
  • func (o *Options) GetCorruptionChecker() bool
  • func (o *Options) GetDBRootPath() string
  • func (o *Options) GetStoreOptions() *store.Options
  • func (o *Options) GetTxPoolSize() int
  • func (o *Options) WithCorruptionChecker(cc bool) *Options
  • func (o *Options) WithDBRootPath(Path string) *Options
  • func (o *Options) WithReadTxPoolSize(txPoolSize int) *Options
  • func (o *Options) WithStoreOptions(storeOpts *store.Options) *Options
  • func (o *Options) WithSyncFollowers(syncFollowers int) *Options
  • func (o *Options) WithSyncReplication(syncReplication bool) *Options

Constants

const (
	DefaultDbRootPath     = "./data"
	DefaultReadTxPoolSize = 128
)

const (
	SetKeyPrefix byte = iota
	SortedSetKeyPrefix
	SQLPrefix
)

const (
	PlainValuePrefix = iota
	ReferenceValuePrefix
)

const MaxKeyResolutionLimit = 1

const MaxKeyScanLimit = 1000

Variables

var (
	ErrIndexKeyMismatch      = status.New(codes.InvalidArgument, "mismatch between provided index and key").Err()
	ErrNoReferenceProvided   = status.New(codes.InvalidArgument, "provided argument is not a reference").Err()
	ErrReferenceKeyMissing   = status.New(codes.InvalidArgument, "reference key not provided").Err()
	ErrZAddIndexMissing      = status.New(codes.InvalidArgument, "zAdd index not provided").Err()
	ErrReferenceIndexMissing = status.New(codes.InvalidArgument, "reference index not provided").Err()

	ErrDatabaseAlreadyExists      = errors.New("database already exists")
	ErrDatabaseNotExists          = errors.New("database does not exist")
	ErrCannotDeleteAnOpenDatabase = errors.New("cannot delete an open database")
	ErrTxReadPoolExhausted        = errors.New("read tx pool exhausted")
)

var ErrFinalKeyCannotBeConvertedIntoReference = errors.New("final key cannot be converted into a reference")

var ErrIllegalArguments = store.ErrIllegalArguments

var ErrIllegalState = store.ErrIllegalState

var ErrInvalidRevision = errors.New("invalid key revision number")

var ErrIsReplica = errors.New("database is read-only because it's a replica")

var ErrKeyResolutionLimitReached = errors.New("key resolution limit reached. It may be due to cyclic references")

var ErrNoWaitOperationMustBeSelfContained = fmt.Errorf("no wait operation must be self-contained: %w", store.ErrIllegalArguments)

var ErrNotReplica = errors.New("database is NOT a replica")

var ErrReferencedKeyCannotBeAReference = errors.New("referenced key cannot be a reference")

var ErrResultSizeLimitExceeded = errors.New("result size limit exceeded")

var ErrResultSizeLimitReached = errors.New("result size limit reached")

var ErrSQLNotReady = errors.New("SQL catalog not yet replicated")

Functions

func EncodeEntrySpec

func EncodeEntrySpec(
	key []byte,
	md *store.KVMetadata,
	value []byte,
) *store.EntrySpec

func EncodeKey

func EncodeKey(key []byte) []byte

func EncodeReference

func EncodeReference(
	key []byte,
	md *store.KVMetadata,
	referencedKey []byte,
	atTx uint64,
) *store.EntrySpec

func EncodeZAdd

func EncodeZAdd(set []byte, score float64, key []byte, atTx uint64) *store.EntrySpec

func PreconditionFromProto

func PreconditionFromProto(c *schema.Precondition) (store.Precondition, error)

func TrimPrefix

func TrimPrefix(prefixed []byte) []byte

func WrapReferenceValueAt

func WrapReferenceValueAt(key []byte, atTx uint64) []byte

func WrapWithPrefix

func WrapWithPrefix(b []byte, prefix byte) []byte

WrapWithPrefix ...

func WrapZAddReferenceAt

func WrapZAddReferenceAt(set []byte, score float64, key []byte, atTx uint64) []byte

Types

type DB

type DB interface {
	GetName() string

	// Setttings
	GetOptions() *Options

	Path() string

	AsReplica(asReplica bool, syncFollowers int)
	IsReplica() bool

	IsSyncReplicationEnabled() bool
	EnableSyncReplication() error
	DisableSyncReplication() error

	MaxResultSize() int
	UseTimeFunc(timeFunc store.TimeFunc) error

	// State
	Health() (waitingCount int, lastReleaseAt time.Time)
	CurrentState() (*schema.ImmutableState, error)

	Size() (uint64, error)

	// Key-Value
	Set(req *schema.SetRequest) (*schema.TxHeader, error)
	VerifiableSet(req *schema.VerifiableSetRequest) (*schema.VerifiableTx, error)

	Get(req *schema.KeyRequest) (*schema.Entry, error)
	VerifiableGet(req *schema.VerifiableGetRequest) (*schema.VerifiableEntry, error)
	GetAll(req *schema.KeyListRequest) (*schema.Entries, error)

	Delete(req *schema.DeleteKeysRequest) (*schema.TxHeader, error)

	SetReference(req *schema.ReferenceRequest) (*schema.TxHeader, error)
	VerifiableSetReference(req *schema.VerifiableReferenceRequest) (*schema.VerifiableTx, error)

	Scan(req *schema.ScanRequest) (*schema.Entries, error)

	History(req *schema.HistoryRequest) (*schema.Entries, error)

	ExecAll(operations *schema.ExecAllRequest) (*schema.TxHeader, error)

	Count(prefix *schema.KeyPrefix) (*schema.EntryCount, error)
	CountAll() (*schema.EntryCount, error)

	ZAdd(req *schema.ZAddRequest) (*schema.TxHeader, error)
	VerifiableZAdd(req *schema.VerifiableZAddRequest) (*schema.VerifiableTx, error)
	ZScan(req *schema.ZScanRequest) (*schema.ZEntries, error)

	// SQL-related
	NewSQLTx(ctx context.Context) (*sql.SQLTx, error)

	SQLExec(req *schema.SQLExecRequest, tx *sql.SQLTx) (ntx *sql.SQLTx, ctxs []*sql.SQLTx, err error)
	SQLExecPrepared(stmts []sql.SQLStmt, params map[string]interface{}, tx *sql.SQLTx) (ntx *sql.SQLTx, ctxs []*sql.SQLTx, err error)

	InferParameters(sql string, tx *sql.SQLTx) (map[string]sql.SQLValueType, error)
	InferParametersPrepared(stmt sql.SQLStmt, tx *sql.SQLTx) (map[string]sql.SQLValueType, error)

	SQLQuery(req *schema.SQLQueryRequest, tx *sql.SQLTx) (*schema.SQLQueryResult, error)
	SQLQueryPrepared(stmt sql.DataSource, namedParams []*schema.NamedParam, tx *sql.SQLTx) (*schema.SQLQueryResult, error)
	SQLQueryRowReader(stmt sql.DataSource, params map[string]interface{}, tx *sql.SQLTx) (sql.RowReader, error)

	VerifiableSQLGet(req *schema.VerifiableSQLGetRequest) (*schema.VerifiableSQLEntry, error)

	ListTables(tx *sql.SQLTx) (*schema.SQLQueryResult, error)
	DescribeTable(table string, tx *sql.SQLTx) (*schema.SQLQueryResult, error)

	// Transactional layer
	WaitForTx(txID uint64, allowPrecommitted bool, cancellation <-chan struct{}) error
	WaitForIndexingUpto(txID uint64, cancellation <-chan struct{}) error

	TxByID(req *schema.TxRequest) (*schema.Tx, error)
	ExportTxByID(req *schema.ExportTxRequest) (txbs []byte, mayCommitUpToTxID uint64, mayCommitUpToAlh [sha256.Size]byte, err error)
	ReplicateTx(exportedTx []byte) (*schema.TxHeader, error)
	AllowCommitUpto(txID uint64, alh [sha256.Size]byte) error
	DiscardPrecommittedTxsSince(txID uint64) error

	VerifiableTxByID(req *schema.VerifiableTxRequest) (*schema.VerifiableTx, error)
	TxScan(req *schema.TxScanRequest) (*schema.TxList, error)

	// Maintenance
	FlushIndex(req *schema.FlushIndexRequest) error
	CompactIndex() error

	IsClosed() bool
	Close() error
}

func NewDB

func NewDB(dbName string, multidbHandler sql.MultiDBHandler, op *Options, log logger.Logger) (DB, error)

NewDB Creates a new Database along with it's directories and files

func OpenDB

func OpenDB(dbName string, multidbHandler sql.MultiDBHandler, op *Options, log logger.Logger) (DB, error)

OpenDB Opens an existing Database from disk

type DatabaseList

type DatabaseList interface {
	Put(database DB)
	Delete(string) (DB, error)
	GetByIndex(index int) (DB, error)
	GetByName(string) (DB, error)
	GetId(dbname string) int
	Length() int
}

DatabaseList interface

func NewDatabaseList

func NewDatabaseList() DatabaseList

NewDatabaseList constructs a new database list

type Options

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

Options database instance options

func DefaultOption

func DefaultOption() *Options

DefaultOption Initialise Db Optionts to default values

func (*Options) AsReplica

func (o *Options) AsReplica(replica bool) *Options

AsReplica sets if the database is a replica

func (*Options) GetCorruptionChecker

func (o *Options) GetCorruptionChecker() bool

GetCorruptionChecker returns if corruption checker should start for this database instance

func (*Options) GetDBRootPath

func (o *Options) GetDBRootPath() string

GetDbRootPath returns the directory in which this database resides

func (*Options) GetStoreOptions

func (o *Options) GetStoreOptions() *store.Options

GetStoreOptions returns backing store options

func (*Options) GetTxPoolSize

func (o *Options) GetTxPoolSize() int

func (*Options) WithCorruptionChecker

func (o *Options) WithCorruptionChecker(cc bool) *Options

WithCorruptionChecker sets if corruption checker should start for this database instance

func (*Options) WithDBRootPath

func (o *Options) WithDBRootPath(Path string) *Options

WithDbRootPath sets the directory in which this database will reside

func (*Options) WithReadTxPoolSize

func (o *Options) WithReadTxPoolSize(txPoolSize int) *Options

func (*Options) WithStoreOptions

func (o *Options) WithStoreOptions(storeOpts *store.Options) *Options

WithStoreOptions sets backing store options

func (*Options) WithSyncFollowers

func (o *Options) WithSyncFollowers(syncFollowers int) *Options

func (*Options) WithSyncReplication

func (o *Options) WithSyncReplication(syncReplication bool) *Options