Rustin Liu

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/tablesink/table_sink.go
// 用于将数据以表为单位进行缓存。
type TableSink interface {
	// AppendRowChangedEvents 将行变更事件追加到 Table Sink 中。
	// 注意：此方法不是线程安全的，请不要并发调用。
	AppendRowChangedEvents(rows ...*model.RowChangedEvent)
	// UpdateResolvedTs 将聚合完成的数据发送到 Event Sink 中。
	// 注意：此方法是异步的且不是线程安全的。
	UpdateResolvedTs(resolvedTs model.ResolvedTs) error
	// GetCheckpointTs 返回 Table Sink 中的 CheckpointTs。
	// 注意：此方法是线程安全的。
	GetCheckpointTs() model.ResolvedTs
	// Close 关闭 Table Sink 并释放资源。
	// 注意：我们需要保证这个方法是可取消或者可中断的。
	Close(ctx context.Context)
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/tablesink/table_sink.go
type EventTableSink[E eventsink.TableEvent] struct {
	...
	// 就是具体的 Event Sink，比如 MQ Event Sink。
	backendSink     eventsink.EventSink[E]
	...
	// 用于实现不同的聚合策略。
	eventAppender   eventsink.Appender[E]
	// 注意：数据是按照 CommitTs 排序的。
	eventBuffer []E
	...
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/event.go
type TableEvent interface {
	// GetCommitTs 返回事件的 CommitTs。
	GetCommitTs() uint64
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/event.go
type CallbackFunc func()

type CallbackableEvent[E TableEvent] struct {
	Event     E
	Callback  CallbackFunc
	...
}

// RowChangeCallbackableEvent 是行变更事件，它可以被回调。
type RowChangeCallbackableEvent = CallbackableEvent[*model.RowChangedEvent]

// TxnCallbackableEvent 是单表事务事件，它可以被回调。
type TxnCallbackableEvent = CallbackableEvent[*model.SingleTableTxn]

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/event_appender.go
type Appender[E TableEvent] interface {
	// Append 添加一批行变更事件到缓存中。
	Append(buffer []E, rows ...*model.RowChangedEvent) []E
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/event_appender.go
func (r *RowChangeEventAppender) Append(
	buffer []*model.RowChangedEvent,
	rows ...*model.RowChangedEvent,
) []*model.RowChangedEvent {
	return append(buffer, rows...)
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/event_appender.go
func (t *TxnEventAppender) Append(
	buffer []*model.SingleTableTxn,
	rows ...*model.RowChangedEvent,
) []*model.SingleTableTxn {
	for _, row := range rows {
		// 这意味我们目前还没有任何事务，所以我们需要创建一个新的事务。
		if len(buffer) == 0 {
			txn := &model.SingleTableTxn{
				StartTs:   row.StartTs,
				CommitTs:  row.CommitTs,
				Table:     row.Table,
				TableInfo: row.TableInfo,
			}
			txn.Append(row)
			buffer = append(buffer, txn)
			continue
		}

		lastTxn := buffer[len(buffer)-1]

		lastCommitTs := lastTxn.GetCommitTs()
		...
		// 使用 StartTs 来判断是否是同一个事务。如果不是同一个事务，我们需要创建一个新的事务。
		if row.SplitTxn || lastTxn.StartTs != row.StartTs {
			buffer = append(buffer, &model.SingleTableTxn{
				StartTs:   row.StartTs,
				CommitTs:  row.CommitTs,
				Table:     row.Table,
				TableInfo: row.TableInfo,
			})
		}

		buffer[len(buffer)-1].Append(row)
	}

	return buffer
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/tablesink/table_sink.go
func (e *EventTableSink[E]) AppendRowChangedEvents(rows ...*model.RowChangedEvent) {
	e.eventBuffer = e.eventAppender.Append(e.eventBuffer, rows...)
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/tablesink/table_sink.go
func (e *EventTableSink[E]) UpdateResolvedTs(resolvedTs model.ResolvedTs) error {
	...
	// 从缓存中找到第一个大于 resolvedTs 的数据。
	i := sort.Search(len(e.eventBuffer), func(i int) bool {
		return e.eventBuffer[i].GetCommitTs() > resolvedTs.Ts
	})
	...
	// 将该数据之前的所有数据取出来。
	resolvedEvents := e.eventBuffer[:i]

	...

	// 为每一个取出来的数据创建一个 CallbackableEvent。
	resolvedCallbackableEvents := make([]*eventsink.CallbackableEvent[E], 0, len(resolvedEvents))
	for _, ev := range resolvedEvents {
		ce := &eventsink.CallbackableEvent[E]{
			Event:     ev,
			Callback:  e.progressTracker.addEvent(),
			SinkState: &e.state,
		}
		resolvedCallbackableEvents = append(resolvedCallbackableEvents, ce)
	}

	...

	// 将取出来的数据写入到下游。
	return e.backendSink.WriteEvents(resolvedCallbackableEvents...)
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/event_sink.go
type EventSink[E TableEvent] interface {
	// WriteEvents 将数据写入到外部系统中。
	// 注意：这是一个异步且线程安全的方法。
	WriteEvents(events ...*CallbackableEvent[E]) error
	// Close 关闭 Event Sink。
	// 注意：我们需要保证这个方法是可取消或者可中断的。
	Close() error
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/mq/mq_dml_sink.go
type dmlSink struct {
	...
	worker *worker
	...
	ctx    context.Context
	cancel context.CancelFunc
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/mq/mq_dml_sink.go
func (s *dmlSink) WriteEvents(rows ...*eventsink.RowChangeCallbackableEvent) error {
	for _, row := range rows {
		...
		// 将数据通过 channel 发送到 worker 中。
		s.worker.msgChan.In() <- mqEvent{
			...
			rowEvent: row,
		}
	}

	return nil
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/ddlsink/ddl_sink.go
type DDLEventSink interface {
	// WriteDDLEvent 将 DDL 写入到外部系统中。
	// 注意：这是一个同步且线程安全的方法。
	WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error
	// WriteCheckpointTs 将 CheckpointTs 写入到外部系统中。
	// 注意：这是一个同步且线程安全的方法。
	// 目前只有 MQ DDL Sink 会实现这个方法。
	WriteCheckpointTs(ctx context.Context, ts uint64, tables []*model.TableInfo) error
	// Close 关闭 DDL Sink 和释放相关资源。
	Close() error
}

// 断言 LogSink 实现了 EventSink 接口。
var _ eventsink.EventSink[*model.RowChangedEvent] = (*LogSink)(nil)

type LogSink struct{}

// New 创建一个新的 LogSink。
func New() *LogSink {
	return &LogSink{}
}

func (s *LogSink) WriteEvents(rows ...*eventsink.CallbackableEvent[*model.RowChangedEvent]) error {
	for _, row := range rows {
		log.Info("LogSink: WriteEvents", zap.Any("row", row.Event))
		// 不要忘记调用 Callback 方法来通知 Table Sink 该 Event 已经被处理。
		row.Callback()
	}

	return nil
}

func (s *LogSink) Close() error {
	return nil
}

// 断言 LogSink 实现了 DDLEventSink 接口。
var _ ddlsink.DDLEventSink = (*LogSink)(nil)

type LogSink struct{}

// New create a black hole DDL sink.
func New() *LogSink {
	return &LogSink{}
}

func (d *LogSink) WriteDDLEvent(ctx context.Context,
	ddl *model.DDLEvent,
) error {
	log.Info("LogSink: DDL Event", zap.Any("ddl", ddl))
	return nil
}

func (d *LogSink) WriteCheckpointTs(ctx context.Context,
	ts uint64, tables []*model.TableInfo,
) error {
	log.Info("LogSink: Checkpoint Ts Event", zap.Uint64("ts", ts), zap.Any("tables", tables))
	return nil
}

// Close do nothing.
func (d *LogSink) Close() error {
	return nil
}

// https://github.com/pingcap/tiflow/blob/master/cdc/sinkv2/eventsink/factory/factory.go
type SinkFactory struct {
	sinkType sink.Type
	rowSink  eventsink.EventSink[*model.RowChangedEvent]
	txnSink  eventsink.EventSink[*model.SingleTableTxn]
}

// New 根据协议类型创建一个 SinkFactory。
func New(ctx context.Context,
	sinkURIStr string,
	cfg *config.ReplicaConfig,
	errCh chan error,
) (*SinkFactory, error) {
	sinkURI, err := config.GetSinkURIAndAdjustConfigWithSinkURI(sinkURIStr, cfg)
	if err != nil {
		return nil, err
	}

	s := &SinkFactory{}
	schema := strings.ToLower(sinkURI.Scheme)
	switch schema {
	case sink.MySQLScheme, sink.MySQLSSLScheme, sink.TiDBScheme, sink.TiDBSSLScheme:
		txnSink, err := txn.NewMySQLSink(ctx, sinkURI, cfg, errCh, txn.DefaultConflictDetectorSlots)
		if err != nil {
			return nil, err
		}
		s.txnSink = txnSink
		s.sinkType = sink.TxnSink
	case sink.KafkaScheme, sink.KafkaSSLScheme:
		mqs, err := mq.NewKafkaDMLSink(ctx, sinkURI, cfg, errCh,
			kafka.NewSaramaAdminClient, kafka.NewSaramaClient, dmlproducer.NewKafkaDMLProducer)
		if err != nil {
			return nil, err
		}
		s.rowSink = mqs
		s.sinkType = sink.RowSink
	...
	default:
		return nil,
			cerror.ErrSinkURIInvalid.GenWithStack("the sink scheme (%s) is not supported", schema)
	}

	return s, nil
}

// CreateTableSink 创建一个 Table Sink，并且传入 Event Sink。
func (s *SinkFactory) CreateTableSink(
	changefeedID model.ChangeFeedID, span tablepb.Span, totalRowsCounter prometheus.Counter,
) tablesink.TableSink {
	switch s.sinkType {
	case sink.RowSink:
		// 我们需要在这里显式地指定 Event Sink 的类型。否则 golang 无法推断出该类型。
		return tablesink.New[*model.RowChangedEvent](changefeedID, span,
			s.rowSink, &eventsink.RowChangeEventAppender{}, totalRowsCounter)
	case sink.TxnSink:
		return tablesink.New[*model.SingleTableTxn](changefeedID, span,
			s.txnSink, &eventsink.TxnEventAppender{}, totalRowsCounter)
	default:
		panic("unknown sink type")
	}
}