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() interface{}
  • func (c *SessionCtx) StageDatabase(db interface{})
• 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, table []byte, lKey keys.Key, rKey keys.Key) (size int64, err 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, 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) (kv.Iterator, error)
	ReadRange(ctx context.Context, lKey keys.Key, rKey keys.Key, isSnapshot 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() interface{}

func (*SessionCtx) StageDatabase

func (c *SessionCtx) StageDatabase(db interface{})

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, table []byte, lKey keys.Key, rKey keys.Key) (size int64, err 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, isSnapshot 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