cdc: remove unused code and add some logs

cdc/model/sink.go

@@ -18,7 +18,6 @@
 	"sort"
 	"strconv"
 	"strings"
-	"sync"
 	"sync/atomic"
 	"unsafe"
 
@@ -272,8 +271,8 @@
 }
 
 // TrySplitAndSortUpdateEvent redo log do nothing
 func (r *RedoLog) TrySplitAndSortUpdateEvent(_ string) error {
 	return nil
 }
 
 // RedoRowChangedEvent represents the DML event used in RedoLog
@@ -380,8 +379,8 @@
 }
 
 // TrySplitAndSortUpdateEvent do nothing
 func (r *RowChangedEvent) TrySplitAndSortUpdateEvent(_ string) error {
 	return nil
 }
 
 // IsDelete returns true if the row is a delete event
@@ -420,23 +419,23 @@
 }
 
 // GetHandleKeyColumnValues returns all handle key's column values
 func (r *RowChangedEvent) GetHandleKeyColumnValues() []string {
 	var result []string
 
 	var cols []*Column
 	if r.IsDelete() {
 		cols = r.PreColumns
 	} else {
 		cols = r.Columns
 	}
 
 	result = make([]string, 0)
 	for _, col := range cols {
 		if col != nil && col.Flag.IsHandleKey() {
 			result = append(result, ColumnValueString(col.Value))
 		}
 	}
 	return result
 }
 
 // HandleKeyColInfos returns the column(s) and colInfo(s) corresponding to the handle key(s)
@@ -610,20 +609,20 @@
 			indexInfo.Unique = true
 		}
 
 		isPrimary := true
 		for _, offset := range colOffsets {
 			col := columns[offset]
 			// When only all columns in the index are primary key, then the index is primary key.
 			if col == nil || !col.Flag.IsPrimaryKey() {
 				isPrimary = false
 			}
 
 			tiCol := ret.Columns[offset]
 			indexCol := &model.IndexColumn{}
 			indexCol.Name = tiCol.Name
 			indexCol.Offset = offset
 			indexInfo.Columns = append(indexInfo.Columns, indexCol)
 			indexInfo.Primary = isPrimary
 		}
 
 		// TODO: revert the "all column set index related flag" to "only the
@@ -754,13 +753,13 @@
 // We set Bootstrap DDL event's startTs and commitTs to 0.
 // Because it is generated by the TiCDC, not from the upstream TiDB.
 // And they ere useless for a bootstrap DDL event.
 func NewBootstrapDDLEvent(tableInfo *TableInfo) *DDLEvent {
 	return &DDLEvent{
 		StartTs:     0,
 		CommitTs:    0,
 		TableInfo:   tableInfo,
 		IsBootstrap: true,
 	}
 }
 
 // SingleTableTxn represents a transaction which includes many row events in a single table
@@ -778,11 +777,6 @@
 	StartTs  uint64
 	CommitTs uint64
 	Rows     []*RowChangedEvent
-
-	// control fields of SingleTableTxn
-	// FinishWg is a barrier txn, after this txn is received, the worker must
-	// flush cached txns and call FinishWg.Done() to mark txns have been flushed.
-	FinishWg *sync.WaitGroup
 }
 
 // GetCommitTs returns the commit timestamp of the transaction.
@@ -797,8 +791,8 @@
 	}
 	newRows, err := trySplitAndSortUpdateEvent(t.Rows)
 	if err != nil {
 		return errors.Trace(err)
 	}
 	t.Rows = newRows
 	return nil
 }
@@ -849,8 +843,8 @@
 		if e.IsUpdate() && shouldSplitUpdateEvent(e) {
 			deleteEvent, insertEvent, err := splitUpdateEvent(e)
 			if err != nil {
 				return nil, errors.Trace(err)
 			}
 			split = true
 			rowChangedEvents = append(rowChangedEvents, deleteEvent, insertEvent)
 		} else {
@@ -870,8 +864,8 @@
 func shouldSplitUpdateEvent(updateEvent *RowChangedEvent) bool {
 	// nil event will never be split.
 	if updateEvent == nil {
 		return false
 	}
 
 	for i := range updateEvent.Columns {
 		col := updateEvent.Columns[i]
@@ -894,8 +888,8 @@
 	updateEvent *RowChangedEvent,
 ) (*RowChangedEvent, *RowChangedEvent, error) {
 	if updateEvent == nil {
 		return nil, nil, errors.New("nil event cannot be split")
 	}
 
 	// If there is an update to handle key columns,
 	// we need to split the event into two events to be compatible with the old format.
@@ -924,11 +918,6 @@
 	t.Rows = append(t.Rows, row)
 }
 
-// ToWaitFlush indicates whether to wait flushing after the txn is processed or not.
-func (t *SingleTableTxn) ToWaitFlush() bool {
-	return t.FinishWg != nil
-}
-
 // TopicPartitionKey contains the topic and partition key of the message.
 type TopicPartitionKey struct {
 	Topic        string

cdc/processor/sinkmanager/metrics.go

@@ -18,17 +18,6 @@ import (
 )
 
 var (
-	// MemoryQuota indicates memory usage of a changefeed.
-	MemoryQuota = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
-			Namespace: "ticdc",
-			Subsystem: "sinkmanager",
-			Name:      "memory_quota",
-			Help:      "memory quota of the changefeed",
-		},
-		// type includes total, used, component includes sink and redo.
-		[]string{"namespace", "changefeed", "type", "component"})
-
 	// RedoEventCache indicates redo event memory usage of a changefeed.
 	RedoEventCache = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{

cdc/sink/dmlsink/txn/mysql/mysql.go

@@ -207,7 +207,7 @@ func NewMySQLBackends(
 func (s *mysqlBackend) OnTxnEvent(event *dmlsink.TxnCallbackableEvent) (needFlush bool) {
 	s.events = append(s.events, event)
 	s.rows += len(event.Event.Rows)
-	return event.Event.ToWaitFlush() || s.rows >= s.cfg.MaxTxnRow
+	return s.rows >= s.cfg.MaxTxnRow
 }
 
 // Flush implements interface backend.

cdc/sink/dmlsink/txn/worker.go

@@ -92,7 +92,7 @@ func (w *worker) close() {
 	w.txnCh.CloseAndDrain()
 }
 
-// Run a loop.
+// Continuously get events from txnCh and call backend flush based on conditions.
 func (w *worker) runLoop() error {
 	defer func() {
 		if err := w.backend.Close(); err != nil {
@@ -148,7 +148,7 @@ func (w *worker) runLoop() error {
 }
 
 // onEvent is called when a new event is received.
-// It returns true if the event is sent to backend.
+// It returns true if it needs flush immediately.
 func (w *worker) onEvent(txn txnWithNotifier) bool {
 	w.hasPending = true
 
@@ -169,7 +169,6 @@ func (w *worker) onEvent(txn txnWithNotifier) bool {
 }
 
 // doFlush flushes the backend.
-// It returns true only if it can no longer be flushed.
 func (w *worker) doFlush(flushTimeSlice *time.Duration) error {
 	if w.hasPending {
 		start := time.Now()