transaction

Documentation

Index

• Variables
• type BaseTx
• type Manager
  • func NewManager(kvStore kv.TxStore) *Manager
  • func (m *Manager) StartTx(ctx context.Context) (Tx, error)
• type SessionCtx
  • func (c *SessionCtx) GetStagedDatabase() any
  • func (c *SessionCtx) StageDatabase(db any)
• type Tx
• type TxSession
  • func (s *TxSession) AtomicAdd(ctx context.Context, key keys.Key, value int64) error
  • func (s *TxSession) AtomicRead(ctx context.Context, key keys.Key) (int64, error)
  • 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) RangeSize(ctx context.Context, _ []byte, lKey keys.Key, rKey keys.Key) (size int64, err error)
  • func (s *TxSession) Read(ctx context.Context, key keys.Key, reverse bool) (kv.Iterator, error)
  • func (s *TxSession) ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key, isSnapshot bool, ...) (kv.Iterator, error)
  • func (s *TxSession) Replace(ctx context.Context, key keys.Key, data *internal.TableData, isUpdate bool) 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

Variables

var (
	// ErrSessionIsNotStarted is returned when the session is not started but is getting used.
	ErrSessionIsNotStarted = errors.Internal("session not started")

	// ErrSessionIsGone is returned when the session is gone but getting used.
	ErrSessionIsGone = errors.Internal("session is gone")
)

Types

type BaseTx

type BaseTx 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, isUpdate bool) error
	Delete(ctx context.Context, key keys.Key) error
	Read(ctx context.Context, key keys.Key, reverse bool) (kv.Iterator, error)
	ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key, isSnapshot bool, reverse bool) (kv.Iterator, error)
	Get(ctx context.Context, key []byte, isSnapshot bool) (kv.Future, error)
	SetVersionstampedValue(ctx context.Context, key []byte, value []byte) error
	SetVersionstampedKey(ctx context.Context, key []byte, value []byte) error
	AtomicAdd(ctx context.Context, key keys.Key, value int64) error
	AtomicRead(ctx context.Context, key keys.Key) (int64, error)
	RangeSize(ctx context.Context, table []byte, lKey keys.Key, rKey keys.Key) (size int64, err error)
}

BaseTx interface exposes base methods that can be used on a transactional object.

type Manager

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

func NewManager

func NewManager(kvStore kv.TxStore) *Manager

func (*Manager) StartTx

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

StartTx starts a new read-write tx session.

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() any

func (*SessionCtx) StageDatabase

func (c *SessionCtx) StageDatabase(db any)

type Tx

type Tx interface {
	BaseTx

	Commit(ctx context.Context) error
	Rollback(ctx context.Context) error
}

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) AtomicAdd

func (s *TxSession) AtomicAdd(ctx context.Context, key keys.Key, value int64) error

func (*TxSession) AtomicRead

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

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) RangeSize

func (s *TxSession) RangeSize(ctx context.Context, _ []byte, lKey keys.Key, rKey keys.Key) (size int64, err error)

func (*TxSession) Read

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

func (*TxSession) ReadRange

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

func (*TxSession) Replace

func (s *TxSession) Replace(ctx context.Context, key keys.Key, data *internal.TableData, isUpdate bool) 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