dmlsink

Documentation

Index

• type Appender
• type CallbackFunc
• type CallbackableEvent
  • func (ce *CallbackableEvent[E]) GetTableSinkState() state.TableSinkState
• type EventSink
• type RowChangeCallbackableEvent
• type RowChangeEventAppender
  • func (r *RowChangeEventAppender) Append(buffer []*model.RowChangedEvent, rows ...*model.RowChangedEvent) []*model.RowChangedEvent
• type TableEvent
• type TxnCallbackableEvent
• type TxnEventAppender
  • func (t *TxnEventAppender) Append(buffer []*model.SingleTableTxn, rows ...*model.RowChangedEvent) []*model.SingleTableTxn

Types

type Appender

type Appender[E TableEvent] interface {
	// Append appends the event to buffer.
	Append(buffer []E, rows ...*model.RowChangedEvent) []E
}

Appender is the interface for appending events to buffer.

type CallbackFunc

type CallbackFunc func()

CallbackFunc is the callback function for callbackable event.

type CallbackableEvent

type CallbackableEvent[E TableEvent] struct {
	Event     E
	Callback  CallbackFunc
	SinkState *state.TableSinkState
}

CallbackableEvent means the event can be callbacked. It also contains the table status.

func (*CallbackableEvent[E]) GetTableSinkState

func (ce *CallbackableEvent[E]) GetTableSinkState() state.TableSinkState

GetTableSinkState returns the table sink state.

type EventSink

type EventSink[E TableEvent] interface {
	// WriteEvents writes events to the sink.
	// This is an asynchronously and thread-safe method.
	WriteEvents(events ...*CallbackableEvent[E]) error
	// SchemeOption returns the sink scheme and whether the sink should output raw change event.
	SchemeOption() (scheme string, outputRawChangeEvent bool)
	// Close closes the sink. Can be called with `WriteEvents` concurrently.
	Close()
	// The EventSink meets internal errors and has been dead already.
	Dead() <-chan struct{}
}

EventSink is the interface for event sink.

type RowChangeCallbackableEvent

type RowChangeCallbackableEvent = CallbackableEvent[*model.RowChangedEvent]

RowChangeCallbackableEvent is the row change event which can be callbacked.

type RowChangeEventAppender

type RowChangeEventAppender struct{}

RowChangeEventAppender is the builder for RowChangedEvent.

func (*RowChangeEventAppender) Append

func (r *RowChangeEventAppender) Append(
	buffer []*model.RowChangedEvent,
	rows ...*model.RowChangedEvent,
) []*model.RowChangedEvent

Append appends the given rows to the given buffer.

type TableEvent

type TableEvent interface {
	// GetCommitTs returns the commit timestamp of the event.
	GetCommitTs() uint64
	// TrySplitAndSortUpdateEvent split the update to delete and insert if the unique key is updated
	TrySplitAndSortUpdateEvent(scheme string, outputRawChangeEvent bool) error
}

TableEvent is the interface for events which can be written to sink by TableSink.

type TxnCallbackableEvent

type TxnCallbackableEvent = CallbackableEvent[*model.SingleTableTxn]

TxnCallbackableEvent is the txn event which can be callbacked.

type TxnEventAppender

type TxnEventAppender struct {
	// TableSinkStartTs is the startTs of the table sink.
	TableSinkStartTs model.Ts
	// IgnoreStartTs indicates whether to ignore the startTs of the row.
	// This is used by consumer to keep compatibility with the old version.
	// Most of our protocols are ignoring the startTs of the row, so we
	// can not use the startTs to identify a transaction.
	IgnoreStartTs bool
}

TxnEventAppender is the appender for SingleTableTxn.

func (*TxnEventAppender) Append

func (t *TxnEventAppender) Append(
	buffer []*model.SingleTableTxn,
	rows ...*model.RowChangedEvent,
) []*model.SingleTableTxn

Append appends the given rows to the given txn buffer. The callers of this function should **make sure** that the commitTs and startTs of rows is **strictly increasing**. 1. Txns ordered by commitTs and startTs. 2. Rows are grouped into SingleTableTxn by startTs and big txn batch, since the startTs is the unique identifier of a transaction. After Append, the structure of the buffer is: buffer = [Txn1[row11, row12...], Txn2[row21,row22...]...], in which:

1. If Txn1.CommitTs < Txn2.CommitTs, then Txn1.startTs can be either less or larger than Txn2.startTs.
2. If Txn1.CommitTs == Txn2.CommitTs, then Txn1.startTs must be **less than** Txn2.startTs.