fix comments

cdc/model/sink.go

@@ -773,7 +773,7 @@ func (t *SingleTableTxn) GetCommitTs() uint64 {
 
 // TrySplitAndSortUpdateEvent split update events if unique key is updated
 func (t *SingleTableTxn) TrySplitAndSortUpdateEvent(sinkScheme string) error {
-	if !t.shouldSplitTxn(sinkScheme) {
+	if !t.shouldSplitUpdateEvent(sinkScheme) {
 		return nil
 	}
 
@@ -785,12 +785,14 @@ func (t *SingleTableTxn) TrySplitAndSortUpdateEvent(sinkScheme string) error {
 	return nil
 }
 
-func (t *SingleTableTxn) shouldSplitTxn(sinkScheme string) bool {
+func (t *SingleTableTxn) shouldSplitUpdateEvent(sinkScheme string) bool {
+	// For mysql sink, we do not split single-row transactions to restore
+	// upstream events as much as possible.
 	if len(t.Rows) < 2 && sink.IsMySQLCompatibleScheme(sinkScheme) {
 		return false
 	}
-	// For MQ or storage sink, we need to split the transaction with single row, since some
-	// protocols (such as csv and avro) do not support old value.
+	// For MQ and storage sink, we need to split the transaction with single row, since some
+	// protocols (such as csv and avro) do not support old value in update event.
 	return true
 }
 

cdc/processor/sinkmanager/redo_log_worker.go

@@ -281,10 +281,7 @@ func (w *redoWorker) handleTask(ctx context.Context, task *redoTask) (finalErr e
 		if e.Row != nil {
 			// For all rows, we add table replicate ts, so mysql sink can determine safe-mode.
 			e.Row.ReplicatingTs = task.tableSink.replicateTs
-			x, size, err = convertRowChangedEvents(w.changefeedID, task.tableID, w.enableOldValue, e)
-			if err != nil {
-				return errors.Trace(err)
-			}
+			x, size = convertRowChangedEvents(w.changefeedID, task.tableID, w.enableOldValue, e)
 			usedMemSize += size
 			rows = append(rows, x...)
 			rowsSize += size

cdc/processor/sinkmanager/table_sink_worker.go

@@ -381,10 +381,7 @@ func (w *sinkWorker) handleTask(ctx context.Context, task *sinkTask) (finalErr e
 		if e.Row != nil {
 			// For all rows, we add table replicate ts, so mysql sink can determine safe-mode.
 			e.Row.ReplicatingTs = task.tableSink.replicateTs
-			x, size, err := convertRowChangedEvents(w.changefeedID, task.tableID, w.enableOldValue, e)
-			if err != nil {
-				return err
-			}
+			x, size := convertRowChangedEvents(w.changefeedID, task.tableID, w.enableOldValue, e)
 			events = append(events, x...)
 			allEventSize += size
 			usedMem += size

cdc/processor/sinkmanager/table_sink_wrapper.go

@@ -442,7 +442,7 @@ func (t *tableSinkWrapper) sinkMaybeStuck(stuckCheck time.Duration) (bool, uint6
 func convertRowChangedEvents(
 	changefeed model.ChangeFeedID, tableID model.TableID, enableOldValue bool,
 	events ...*model.PolymorphicEvent,
-) ([]*model.RowChangedEvent, uint64, error) {
+) ([]*model.RowChangedEvent, uint64) {
 	size := 0
 	rowChangedEvents := make([]*model.RowChangedEvent, 0, len(events))
 	for _, e := range events {
@@ -472,7 +472,7 @@ func convertRowChangedEvents(
 		size += e.Row.ApproximateBytes()
 		rowChangedEvents = append(rowChangedEvents, e.Row)
 	}
-	return rowChangedEvents, uint64(size), nil
+	return rowChangedEvents, uint64(size)
 }
 
 func genReplicateTs(ctx context.Context, pdClient pd.Client) (model.Ts, error) {

cdc/processor/sinkmanager/table_sink_wrapper_test.go

@@ -145,8 +145,7 @@ func TestConvertNilRowChangedEvents(t *testing.T) {
 	changefeedID := model.DefaultChangeFeedID("1")
 	tableID := model.TableID(1)
 	enableOldVlaue := false
-	result, size, err := convertRowChangedEvents(changefeedID, tableID, enableOldVlaue, events...)
-	require.NoError(t, err)
+	result, size := convertRowChangedEvents(changefeedID, tableID, enableOldVlaue, events...)
 	require.Equal(t, 0, len(result))
 	require.Equal(t, uint64(0), size)
 }
@@ -167,8 +166,7 @@ func TestConvertEmptyRowChangedEvents(t *testing.T) {
 	changefeedID := model.DefaultChangeFeedID("1")
 	tableID := model.TableID(1)
 	enableOldValue := false
-	result, size, err := convertRowChangedEvents(changefeedID, tableID, enableOldValue, events...)
-	require.NoError(t, err)
+	result, size := convertRowChangedEvents(changefeedID, tableID, enableOldValue, events...)
 	require.Equal(t, 0, len(result))
 	require.Equal(t, uint64(0), size)
 }
@@ -219,8 +217,7 @@ func TestConvertRowChangedEventsWhenEnableOldValue(t *testing.T) {
 	changefeedID := model.DefaultChangeFeedID("1")
 	tableID := model.TableID(1)
 	enableOldValue := true
-	result, size, err := convertRowChangedEvents(changefeedID, tableID, enableOldValue, events...)
-	require.NoError(t, err)
+	result, size := convertRowChangedEvents(changefeedID, tableID, enableOldValue, events...)
 	require.Equal(t, 1, len(result))
 	require.Equal(t, uint64(216), size)
 }
@@ -272,8 +269,7 @@ func TestConvertRowChangedEventsWhenDisableOldValue(t *testing.T) {
 	changefeedID := model.DefaultChangeFeedID("1")
 	tableID := model.TableID(1)
 	enableOldValue := false
-	result, size, err := convertRowChangedEvents(changefeedID, tableID, enableOldValue, events...)
-	require.NoError(t, err)
+	result, size := convertRowChangedEvents(changefeedID, tableID, enableOldValue, events...)
 	require.Equal(t, 1, len(result))
 	require.Equal(t, uint64(216), size)
 
@@ -318,8 +314,7 @@ func TestConvertRowChangedEventsWhenDisableOldValue(t *testing.T) {
 			},
 		},
 	}
-	result, size, err = convertRowChangedEvents(changefeedID, tableID, enableOldValue, events...)
-	require.NoError(t, err)
+	result, size = convertRowChangedEvents(changefeedID, tableID, enableOldValue, events...)
 	require.Equal(t, 1, len(result))
 	require.Equal(t, uint64(216), size)
 }