kv

package
v1.0.0-alpha.33 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 24, 2022 License: Apache-2.0 Imports: 14 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	InsertEvent      = "insert"
	ReplaceEvent     = "replace"
	UpdateEvent      = "update"
	UpdateRangeEvent = "updateRange"
	DeleteEvent      = "delete"
	DeleteRangeEvent = "deleteRange"
)

Variables

View Source 
var (
	// ErrDuplicateKey is returned when an insert call is made for a key that already exist.
	ErrDuplicateKey = NewStoreError(ErrCodeDuplicateKey, "duplicate key value, violates key constraint")
	// ErrConflictingTransaction is returned when there are conflicting transactions.
	ErrConflictingTransaction = NewStoreError(ErrCodeConflictingTransaction, "transaction not committed due to conflict with another transaction")
	// ErrTransactionMaxDurationReached is returned when transaction running beyond 5seconds.
	ErrTransactionMaxDurationReached = NewStoreError(ErrCodeTransactionMaxDuration, "transaction is old to perform reads or be committed")
)

Functions

func IsTimedOut 

func IsTimedOut(err error) bool

func NewStoreError 

func NewStoreError(code StoreErrCode, msg string, args ...interface{}) error

func WrapEventListenerCtx 

func WrapEventListenerCtx(ctx context.Context) context.Context

Types

type DefaultListener 

type DefaultListener struct {
	Events []*Event
}

func (*DefaultListener) GetEvents 

func (l *DefaultListener) GetEvents() []*Event

func (*DefaultListener) OnClearRange 

func (l *DefaultListener) OnClearRange(op string, table []byte, lKey []byte, rKey []byte)

func (*DefaultListener) OnSet 

func (l *DefaultListener) OnSet(op string, table []byte, key []byte, data []byte)

type Event 

type Event struct {
	Op    string
	Table []byte
	Key   []byte `json:",omitempty"`
	LKey  []byte `json:",omitempty"`
	RKey  []byte `json:",omitempty"`
	Data  []byte `json:",omitempty"`
	Last  bool
}

type EventListener 

type EventListener interface {
	// OnSet buffers insert/replace/update events
	OnSet(op string, table []byte, key []byte, data []byte)
	// OnClearRange buffers delete events
	OnClearRange(op string, table []byte, lKey []byte, rKey []byte)
	// GetEvents is used to access buffered events. These events may be shared by different participants callers are
	// strongly discourage to modify the event and if needed copy it to some other buffer. Once transaction completes
	// session may discard all the buffered events.
	GetEvents() []*Event
}

EventListener is listener to buffer all the changes in a transaction. It is attached by server layer in the context, and it is only responsible for buffering of the events but doesn't participate in the outcome of the transaction i.e. EventListener has no knowledge whether the transaction was committed or rolled back. The lifecycle of this listener is managed by QuerySession in server package.

func GetEventListener 

func GetEventListener(ctx context.Context) EventListener

type EventListenerCtxKey 

type EventListenerCtxKey struct{}

type Future 

type Future fdb.FutureByteSlice

type Iterator 

type Iterator interface {
	Next(value *KeyValue) bool
	Err() error
}

type IteratorImpl 

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

func (*IteratorImpl) Err 

func (i *IteratorImpl) Err() error

func (*IteratorImpl) Next 

func (i *IteratorImpl) Next(value *KeyValue) bool

type KV 

type KV interface {
	Insert(ctx context.Context, table []byte, key Key, data *internal.TableData) error
	Replace(ctx context.Context, table []byte, key Key, data *internal.TableData) error
	Delete(ctx context.Context, table []byte, key Key) error
	DeleteRange(ctx context.Context, table []byte, lKey Key, rKey Key) error
	Read(ctx context.Context, table []byte, key Key) (Iterator, error)
	ReadRange(ctx context.Context, table []byte, lkey Key, rkey Key, isSnapshot bool) (Iterator, error)
	Update(ctx context.Context, table []byte, key Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)
	UpdateRange(ctx context.Context, table []byte, lKey Key, rKey Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)
	SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error
	SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error
	Get(ctx context.Context, key []byte, isSnapshot bool) (Future, error)
}

type Key 

type Key []KeyPart

func BuildKey 

func BuildKey(parts ...interface{}) Key

func (*Key) AddPart 

func (k *Key) AddPart(part interface{})

type KeyPart 

type KeyPart interface{}

type KeyValue 

type KeyValue struct {
	Key    Key
	FDBKey []byte
	Data   *internal.TableData
}

type KeyValueStore 

type KeyValueStore interface {
	KV
	BeginTx(ctx context.Context) (Tx, error)
	CreateTable(ctx context.Context, name []byte) error
	DropTable(ctx context.Context, name []byte) error
	GetInternalDatabase() (interface{}, error) // TODO: CDC remove workaround
	TableSize(ctx context.Context, name []byte) (int64, error)
}

func NewKeyValueStore 

func NewKeyValueStore(cfg *config.FoundationDBConfig) (KeyValueStore, error)

func NewKeyValueStoreWithMetrics 

func NewKeyValueStoreWithMetrics(cfg *config.FoundationDBConfig) (KeyValueStore, error)

type KeyValueStoreImpl 

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

func (KeyValueStoreImpl) Batch 

func (d KeyValueStoreImpl) Batch() (baseTx, error)

func (*KeyValueStoreImpl) BeginTx 

func (k *KeyValueStoreImpl) BeginTx(ctx context.Context) (Tx, error)

func (KeyValueStoreImpl) CreateTable 

func (d KeyValueStoreImpl) CreateTable(_ context.Context, name []byte) error

func (KeyValueStoreImpl) Delete 

func (d KeyValueStoreImpl) Delete(ctx context.Context, table []byte, key Key) error

func (KeyValueStoreImpl) DeleteRange 

func (d KeyValueStoreImpl) DeleteRange(ctx context.Context, table []byte, lKey Key, rKey Key) error

func (KeyValueStoreImpl) DropTable 

func (d KeyValueStoreImpl) DropTable(ctx context.Context, name []byte) error

func (KeyValueStoreImpl) Get 

func (d KeyValueStoreImpl) Get(ctx context.Context, key []byte, isSnapshot bool) (Future, error)

func (*KeyValueStoreImpl) GetInternalDatabase 

func (k *KeyValueStoreImpl) GetInternalDatabase() (interface{}, error)

func (*KeyValueStoreImpl) Insert 

func (k *KeyValueStoreImpl) Insert(ctx context.Context, table []byte, key Key, data *internal.TableData) error

func (*KeyValueStoreImpl) Read 

func (k *KeyValueStoreImpl) Read(ctx context.Context, table []byte, key Key) (Iterator, error)

func (*KeyValueStoreImpl) ReadRange 

func (k *KeyValueStoreImpl) ReadRange(ctx context.Context, table []byte, lkey Key, rkey Key, isSnapshot bool) (Iterator, error)

func (*KeyValueStoreImpl) Replace 

func (k *KeyValueStoreImpl) Replace(ctx context.Context, table []byte, key Key, data *internal.TableData) error

func (KeyValueStoreImpl) SetVersionstampedKey 

func (d KeyValueStoreImpl) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error

func (KeyValueStoreImpl) SetVersionstampedValue 

func (d KeyValueStoreImpl) SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error

func (KeyValueStoreImpl) TableSize 

func (d KeyValueStoreImpl) TableSize(ctx context.Context, name []byte) (int64, error)

TableSize calculates approximate table size in bytes It also works with the prefix of the table name, allowing to calculate sizes of multiple table with the same prefix

func (*KeyValueStoreImpl) Update 

func (k *KeyValueStoreImpl) Update(ctx context.Context, table []byte, key Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)

func (*KeyValueStoreImpl) UpdateRange 

func (k *KeyValueStoreImpl) UpdateRange(ctx context.Context, table []byte, lKey Key, rKey Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)

type KeyValueStoreImplWithMetrics 

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

func (*KeyValueStoreImplWithMetrics) BeginTx 

func (m *KeyValueStoreImplWithMetrics) BeginTx(ctx context.Context) (Tx, error)

func (*KeyValueStoreImplWithMetrics) CreateTable 

func (m *KeyValueStoreImplWithMetrics) CreateTable(ctx context.Context, name []byte) (err error)

func (*KeyValueStoreImplWithMetrics) Delete 

func (m *KeyValueStoreImplWithMetrics) Delete(ctx context.Context, table []byte, key Key) (err error)

func (*KeyValueStoreImplWithMetrics) DeleteRange 

func (m *KeyValueStoreImplWithMetrics) DeleteRange(ctx context.Context, table []byte, lKey Key, rKey Key) (err error)

func (*KeyValueStoreImplWithMetrics) DropTable 

func (m *KeyValueStoreImplWithMetrics) DropTable(ctx context.Context, name []byte) (err error)

func (*KeyValueStoreImplWithMetrics) Get 

func (m *KeyValueStoreImplWithMetrics) Get(ctx context.Context, key []byte, isSnapshot bool) (val Future, err error)

func (*KeyValueStoreImplWithMetrics) GetInternalDatabase 

func (m *KeyValueStoreImplWithMetrics) GetInternalDatabase() (k interface{}, err error)

func (*KeyValueStoreImplWithMetrics) Insert 

func (m *KeyValueStoreImplWithMetrics) Insert(ctx context.Context, table []byte, key Key, data *internal.TableData) (err error)

func (*KeyValueStoreImplWithMetrics) Read 

func (m *KeyValueStoreImplWithMetrics) Read(ctx context.Context, table []byte, key Key) (it Iterator, err error)

func (*KeyValueStoreImplWithMetrics) ReadRange 

func (m *KeyValueStoreImplWithMetrics) ReadRange(ctx context.Context, table []byte, lkey Key, rkey Key, isSnapshot bool) (it Iterator, err error)

func (*KeyValueStoreImplWithMetrics) Replace 

func (m *KeyValueStoreImplWithMetrics) Replace(ctx context.Context, table []byte, key Key, data *internal.TableData) (err error)

func (*KeyValueStoreImplWithMetrics) SetVersionstampedKey 

func (m *KeyValueStoreImplWithMetrics) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) (err error)

func (*KeyValueStoreImplWithMetrics) SetVersionstampedValue 

func (m *KeyValueStoreImplWithMetrics) SetVersionstampedValue(ctx context.Context, key []byte, value []byte) (err error)

func (*KeyValueStoreImplWithMetrics) TableSize 

func (m *KeyValueStoreImplWithMetrics) TableSize(ctx context.Context, name []byte) (size int64, err error)

func (*KeyValueStoreImplWithMetrics) Update 

func (m *KeyValueStoreImplWithMetrics) Update(ctx context.Context, table []byte, key Key, apply func(*internal.TableData) (*internal.TableData, error)) (encoded int32, err error)

func (*KeyValueStoreImplWithMetrics) UpdateRange 

func (m *KeyValueStoreImplWithMetrics) UpdateRange(ctx context.Context, table []byte, lKey Key, rKey Key, apply func(*internal.TableData) (*internal.TableData, error)) (encoded int32, err error)

type NoopEventListener 

type NoopEventListener struct{}

func (*NoopEventListener) GetEvents 

func (l *NoopEventListener) GetEvents() []*Event

func (*NoopEventListener) OnClearRange 

func (l *NoopEventListener) OnClearRange(op string, table []byte, lKey []byte, rKey []byte)

func (*NoopEventListener) OnSet 

func (l *NoopEventListener) OnSet(op string, table []byte, key []byte, data []byte)

type NoopIterator 

type NoopIterator struct{}

func (*NoopIterator) Err 

func (n *NoopIterator) Err() error

func (*NoopIterator) Next 

func (n *NoopIterator) Next(value *KeyValue) bool

type NoopKV 

type NoopKV struct{}

func (*NoopKV) Delete 

func (n *NoopKV) Delete(ctx context.Context, table []byte, key Key) error

func (*NoopKV) DeleteRange 

func (n *NoopKV) DeleteRange(ctx context.Context, table []byte, lKey Key, rKey Key) error

func (*NoopKV) Get 

func (n *NoopKV) Get(ctx context.Context, key []byte, isSnapshot bool) (Future, error)

func (*NoopKV) Insert 

func (n *NoopKV) Insert(ctx context.Context, table []byte, key Key, data *internal.TableData) error

func (*NoopKV) Read 

func (n *NoopKV) Read(ctx context.Context, table []byte, key Key) (Iterator, error)

func (*NoopKV) ReadRange 

func (n *NoopKV) ReadRange(ctx context.Context, table []byte, lkey Key, rkey Key, isSnapshot bool) (Iterator, error)

func (*NoopKV) Replace 

func (n *NoopKV) Replace(ctx context.Context, table []byte, key Key, data *internal.TableData) error

func (*NoopKV) SetVersionstampedKey 

func (n *NoopKV) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error

func (*NoopKV) SetVersionstampedValue 

func (n *NoopKV) SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error

func (*NoopKV) Update 

func (n *NoopKV) Update(ctx context.Context, table []byte, key Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)

func (*NoopKV) UpdateRange 

func (n *NoopKV) UpdateRange(ctx context.Context, table []byte, lKey Key, rKey Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)

type NoopKVStore 

type NoopKVStore struct {
	*NoopKV
}

NoopKVStore is a noop store, useful if we need to profile/debug only compute and not with the storage. This can be initialized in main.go instead of using default kvStore.

func (*NoopKVStore) BeginTx 

func (n *NoopKVStore) BeginTx(_ context.Context) (Tx, error)

func (*NoopKVStore) CreateTable 

func (n *NoopKVStore) CreateTable(_ context.Context, _ []byte) error

func (*NoopKVStore) DropTable 

func (n *NoopKVStore) DropTable(_ context.Context, _ []byte) error

func (*NoopKVStore) GetInternalDatabase 

func (n *NoopKVStore) GetInternalDatabase() (interface{}, error)

func (*NoopKVStore) TableSize 

func (n *NoopKVStore) TableSize(_ context.Context, _ []byte) (int64, error)

type NoopTx 

type NoopTx struct {
	*NoopKV
}

func (*NoopTx) Commit 

func (n *NoopTx) Commit(context.Context) error

func (*NoopTx) IsRetriable 

func (n *NoopTx) IsRetriable() bool

func (*NoopTx) Rollback 

func (n *NoopTx) Rollback(context.Context) error

type StoreErrCode 

type StoreErrCode byte
const (
	ErrCodeInvalid                StoreErrCode = 0x00
	ErrCodeDuplicateKey           StoreErrCode = 0x01
	ErrCodeConflictingTransaction StoreErrCode = 0x02
	ErrCodeTransactionMaxDuration StoreErrCode = 0x03
)

type StoreError 

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

func (StoreError) Error 

func (se StoreError) Error() string

type Tx 

type Tx interface {
	KV
	Commit(context.Context) error
	Rollback(context.Context) error
	IsRetriable() bool
}

type TxImpl 

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

func (TxImpl) Commit 

func (t TxImpl) Commit(_ context.Context) error

func (TxImpl) Delete 

func (t TxImpl) Delete(ctx context.Context, table []byte, key Key) error

func (TxImpl) DeleteRange 

func (t TxImpl) DeleteRange(ctx context.Context, table []byte, lKey Key, rKey Key) error

func (TxImpl) Get 

func (t TxImpl) Get(_ context.Context, key []byte, isSnapshot bool) (Future, error)

func (*TxImpl) Insert 

func (tx *TxImpl) Insert(ctx context.Context, table []byte, key Key, data *internal.TableData) error

func (TxImpl) IsRetriable 

func (t TxImpl) IsRetriable() bool

IsRetriable returns true if transaction can be retried after error

func (*TxImpl) Read 

func (tx *TxImpl) Read(ctx context.Context, table []byte, key Key) (Iterator, error)

func (*TxImpl) ReadRange 

func (tx *TxImpl) ReadRange(ctx context.Context, table []byte, lkey Key, rkey Key, isSnapshot bool) (Iterator, error)

func (*TxImpl) Replace 

func (tx *TxImpl) Replace(ctx context.Context, table []byte, key Key, data *internal.TableData) error

func (TxImpl) Rollback 

func (t TxImpl) Rollback(_ context.Context) error

func (TxImpl) SetVersionstampedKey 

func (t TxImpl) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error

func (TxImpl) SetVersionstampedValue 

func (t TxImpl) SetVersionstampedValue(_ context.Context, key []byte, value []byte) error

func (*TxImpl) Update 

func (tx *TxImpl) Update(ctx context.Context, table []byte, key Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)

func (*TxImpl) UpdateRange 

func (tx *TxImpl) UpdateRange(ctx context.Context, table []byte, lKey Key, rKey Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)

type TxImplWithMetrics 

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

func (*TxImplWithMetrics) Commit 

func (m *TxImplWithMetrics) Commit(ctx context.Context) (err error)

func (*TxImplWithMetrics) Delete 

func (m *TxImplWithMetrics) Delete(ctx context.Context, table []byte, key Key) (err error)

func (*TxImplWithMetrics) DeleteRange 

func (m *TxImplWithMetrics) DeleteRange(ctx context.Context, table []byte, lKey Key, rKey Key) (err error)

func (*TxImplWithMetrics) Get 

func (m *TxImplWithMetrics) Get(ctx context.Context, key []byte, isSnapshot bool) (val Future, err error)

func (*TxImplWithMetrics) Insert 

func (m *TxImplWithMetrics) Insert(ctx context.Context, table []byte, key Key, data *internal.TableData) (err error)

func (*TxImplWithMetrics) IsRetriable 

func (m *TxImplWithMetrics) IsRetriable() bool

func (*TxImplWithMetrics) Read 

func (m *TxImplWithMetrics) Read(ctx context.Context, table []byte, key Key) (it Iterator, err error)

func (*TxImplWithMetrics) ReadRange 

func (m *TxImplWithMetrics) ReadRange(ctx context.Context, table []byte, lkey Key, rkey Key, isSnapshot bool) (it Iterator, err error)

func (*TxImplWithMetrics) Replace 

func (m *TxImplWithMetrics) Replace(ctx context.Context, table []byte, key Key, data *internal.TableData) (err error)

func (*TxImplWithMetrics) Rollback 

func (m *TxImplWithMetrics) Rollback(ctx context.Context) (err error)

func (*TxImplWithMetrics) SetVersionstampedKey 

func (m *TxImplWithMetrics) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) (err error)

func (*TxImplWithMetrics) SetVersionstampedValue 

func (m *TxImplWithMetrics) SetVersionstampedValue(ctx context.Context, key []byte, value []byte) (err error)

func (*TxImplWithMetrics) Update 

func (m *TxImplWithMetrics) Update(ctx context.Context, table []byte, key Key, apply func(*internal.TableData) (*internal.TableData, error)) (encoded int32, err error)

func (*TxImplWithMetrics) UpdateRange 

func (m *TxImplWithMetrics) UpdateRange(ctx context.Context, table []byte, lKey Key, rKey Key, apply func(*internal.TableData) (*internal.TableData, error)) (encoded int32, err error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.