transaction

Documentation

Index

• Variables
• type Manager
  • func NewManager(kvStore kv.KeyValueStore) *Manager
  • func (m *Manager) StartTx(ctx context.Context) (Tx, error)
• type SessionCtx
  • func (c *SessionCtx) GetStagedDatabase() StagedDB
  • func (c *SessionCtx) StageDatabase(db StagedDB)
• type StagedDB
• type Tx
• type TxSession
  • func (s *TxSession) Commit(ctx context.Context) error
  • func (s *TxSession) Context() *SessionCtx
  • func (s *TxSession) Delete(ctx context.Context, key keys.Key) error
  • func (s *TxSession) Get(ctx context.Context, key []byte, isSnapshot bool) (kv.Future, error)
  • func (s *TxSession) GetTxCtx() *api.TransactionCtx
  • func (s *TxSession) Insert(ctx context.Context, key keys.Key, data *internal.TableData) error
  • func (s *TxSession) Read(ctx context.Context, key keys.Key) (kv.Iterator, error)
  • func (s *TxSession) ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key) (kv.Iterator, error)
  • func (s *TxSession) Replace(ctx context.Context, key keys.Key, data *internal.TableData) error
  • func (s *TxSession) Rollback(ctx context.Context) error
  • func (s *TxSession) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error
  • func (s *TxSession) SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error
  • func (s *TxSession) Update(ctx context.Context, key keys.Key, ...) (int32, error)
• type TxSessionWithMetrics
  • func (m *TxSessionWithMetrics) Commit(ctx context.Context) (err error)
  • func (m *TxSessionWithMetrics) Context() *SessionCtx
  • func (m *TxSessionWithMetrics) Delete(ctx context.Context, key keys.Key) (err error)
  • func (m *TxSessionWithMetrics) Get(ctx context.Context, key []byte, isSnapshot bool) (val kv.Future, err error)
  • func (m *TxSessionWithMetrics) GetTxCtx() *api.TransactionCtx
  • func (m *TxSessionWithMetrics) Insert(ctx context.Context, key keys.Key, data *internal.TableData) (err error)
  • func (m *TxSessionWithMetrics) Read(ctx context.Context, key keys.Key) (it kv.Iterator, err error)
  • func (m *TxSessionWithMetrics) ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key) (it kv.Iterator, err error)
  • func (m *TxSessionWithMetrics) Replace(ctx context.Context, key keys.Key, data *internal.TableData) (err error)
  • func (m *TxSessionWithMetrics) Rollback(ctx context.Context) (err error)
  • func (m *TxSessionWithMetrics) SetVersionstampedKey(ctx context.Context, key []byte, value []byte) (err error)
  • func (m *TxSessionWithMetrics) SetVersionstampedValue(ctx context.Context, key []byte, value []byte) (err error)
  • func (m *TxSessionWithMetrics) Update(ctx context.Context, key keys.Key, ...) (encoded int32, err error)

Variables

var (
	// ErrSessionIsNotStarted is returned when the session is not started but is getting used
	ErrSessionIsNotStarted = api.Errorf(api.Code_INTERNAL, "session not started")

	// ErrSessionIsGone is returned when the session is gone but getting used
	ErrSessionIsGone = api.Errorf(api.Code_INTERNAL, "session is gone")
)

Types

type Manager

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

func NewManager

func NewManager(kvStore kv.KeyValueStore) *Manager

func (*Manager) StartTx

func (m *Manager) StartTx(ctx context.Context) (Tx, error)

StartTx always starts a new session and tracks the session based on the input parameter.

type SessionCtx

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

SessionCtx is used to store any baggage for the lifetime of the transaction. We use it to stage the database inside a transaction when the transaction is performing any DDLs.

func (*SessionCtx) GetStagedDatabase

func (c *SessionCtx) GetStagedDatabase() StagedDB

func (*SessionCtx) StageDatabase

func (c *SessionCtx) StageDatabase(db StagedDB)

type StagedDB

type StagedDB interface {
	Name() string
	GetCollection(string) *schema.DefaultCollection
}

type Tx

type Tx interface {
	Context() *SessionCtx
	GetTxCtx() *api.TransactionCtx
	Insert(ctx context.Context, key keys.Key, data *internal.TableData) error
	Replace(ctx context.Context, key keys.Key, data *internal.TableData) error
	Update(ctx context.Context, key keys.Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)
	Delete(ctx context.Context, key keys.Key) error
	Read(ctx context.Context, key keys.Key) (kv.Iterator, error)
	ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key) (kv.Iterator, error)
	Get(ctx context.Context, key []byte, isSnapshot bool) (kv.Future, error)
	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
	SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error
	SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error
	// contains filtered or unexported methods
}

Tx interface exposes a method to execute and then other method to end the transaction. When Tx is returned at that point transaction is already started so no need for explicit start.

type TxSession

type TxSession struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

TxSession is used to start an explicit transaction. Caller can control whether this transaction's session needs to be tracked inside session tracker. Tracker a session is useful if the object is shared across the requests otherwise it is not useful in the same request flow.

func (*TxSession) Commit

func (s *TxSession) Commit(ctx context.Context) error

func (*TxSession) Context

func (s *TxSession) Context() *SessionCtx

func (*TxSession) Delete

func (s *TxSession) Delete(ctx context.Context, key keys.Key) error

func (*TxSession) Get

func (s *TxSession) Get(ctx context.Context, key []byte, isSnapshot bool) (kv.Future, error)

func (*TxSession) GetTxCtx

func (s *TxSession) GetTxCtx() *api.TransactionCtx

func (*TxSession) Insert

func (s *TxSession) Insert(ctx context.Context, key keys.Key, data *internal.TableData) error

func (*TxSession) Read

func (s *TxSession) Read(ctx context.Context, key keys.Key) (kv.Iterator, error)

func (*TxSession) ReadRange

func (s *TxSession) ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key) (kv.Iterator, error)

func (*TxSession) Replace

func (s *TxSession) Replace(ctx context.Context, key keys.Key, data *internal.TableData) error

func (*TxSession) Rollback

func (s *TxSession) Rollback(ctx context.Context) error

func (*TxSession) SetVersionstampedKey

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

func (*TxSession) SetVersionstampedValue

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

func (*TxSession) Update

func (s *TxSession) Update(ctx context.Context, key keys.Key, apply func(*internal.TableData) (*internal.TableData, error)) (int32, error)

type TxSessionWithMetrics

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

func (*TxSessionWithMetrics) Commit

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

func (*TxSessionWithMetrics) Context

func (m *TxSessionWithMetrics) Context() *SessionCtx

func (*TxSessionWithMetrics) Delete

func (m *TxSessionWithMetrics) Delete(ctx context.Context, key keys.Key) (err error)

func (*TxSessionWithMetrics) Get

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

func (*TxSessionWithMetrics) GetTxCtx

func (m *TxSessionWithMetrics) GetTxCtx() *api.TransactionCtx

func (*TxSessionWithMetrics) Insert

func (m *TxSessionWithMetrics) Insert(ctx context.Context, key keys.Key, data *internal.TableData) (err error)

func (*TxSessionWithMetrics) Read

func (m *TxSessionWithMetrics) Read(ctx context.Context, key keys.Key) (it kv.Iterator, err error)

func (*TxSessionWithMetrics) ReadRange

func (m *TxSessionWithMetrics) ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key) (it kv.Iterator, err error)

func (*TxSessionWithMetrics) Replace

func (m *TxSessionWithMetrics) Replace(ctx context.Context, key keys.Key, data *internal.TableData) (err error)

func (*TxSessionWithMetrics) Rollback

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

func (*TxSessionWithMetrics) SetVersionstampedKey

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

func (*TxSessionWithMetrics) SetVersionstampedValue

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

func (*TxSessionWithMetrics) Update

func (m *TxSessionWithMetrics) Update(ctx context.Context, key keys.Key, apply func(*internal.TableData) (*internal.TableData, error)) (encoded int32, err error)