From 7f42fce46a9e8c2efa06b763153ca536adb4180a Mon Sep 17 00:00:00 2001
From: Jianyuan Jiang <sdojjy@qq.com>
Date: Sun, 13 Aug 2023 18:16:59 -0700
Subject: [PATCH] sink(ticdc):  split RowChangeEvent if unique key is updated
 (#9437)

close pingcap/tiflow#9430
---
 cdc/model/sink.go                             | 152 ++++++++++++++++
 cdc/model/sink_test.go                        | 167 ++++++++++++++++++
 cdc/sink/dmlsink/event.go                     |   2 +
 cdc/sink/tablesink/table_sink_impl.go         |   3 +
 .../http_api/util/test_case.py                |   2 +-
 .../http_api_tls/util/test_case.py            |   2 +-
 6 files changed, 326 insertions(+), 2 deletions(-)

diff --git a/cdc/model/sink.go b/cdc/model/sink.go
index bfc48bda2ee..fbbc30eb579 100644
--- a/cdc/model/sink.go
+++ b/cdc/model/sink.go
@@ -15,6 +15,7 @@ package model
 
 import (
 	"fmt"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -25,6 +26,7 @@ import (
 	"github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/util/rowcodec"
+	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
 	"github.com/pingcap/tiflow/pkg/quotes"
 	"github.com/pingcap/tiflow/pkg/util"
@@ -47,6 +49,13 @@ const (
 	MessageTypeResolved
 )
 
+const (
+	// the RowChangedEvent order in the same transaction
+	typeDelete = iota + 1
+	typeUpdate
+	typeInsert
+)
+
 // ColumnFlagType is for encapsulating the flag operations for different flags.
 type ColumnFlagType util.Flag
 
@@ -261,6 +270,11 @@ func (r *RedoLog) GetCommitTs() Ts {
 	return 0
 }
 
+// TrySplitAndSortUpdateEvent redo log do nothing
+func (r *RedoLog) TrySplitAndSortUpdateEvent() error {
+	return nil
+}
+
 // RedoRowChangedEvent represents the DML event used in RedoLog
 type RedoRowChangedEvent struct {
 	Row        *RowChangedEvent `msg:"row"`
@@ -332,11 +346,43 @@ type RowChangedEvent struct {
 	ReplicatingTs Ts `json:"-" msg:"-"`
 }
 
+// txnRows represents a set of events that belong to the same transaction.
+type txnRows []*RowChangedEvent
+
+// Len is the number of elements in the collection.
+func (e txnRows) Len() int {
+	return len(e)
+}
+
+// Less sort the events base on the order of event type delete<update<insert
+func (e txnRows) Less(i, j int) bool {
+	return getDMLOrder(e[i]) < getDMLOrder(e[j])
+}
+
+// getDMLOrder returns the order of the dml types: delete<update<insert
+func getDMLOrder(event *RowChangedEvent) int {
+	if event.IsDelete() {
+		return typeDelete
+	} else if event.IsUpdate() {
+		return typeUpdate
+	}
+	return typeInsert
+}
+
+func (e txnRows) Swap(i, j int) {
+	e[i], e[j] = e[j], e[i]
+}
+
 // GetCommitTs returns the commit timestamp of this event.
 func (r *RowChangedEvent) GetCommitTs() uint64 {
 	return r.CommitTs
 }
 
+// TrySplitAndSortUpdateEvent do nothing
+func (r *RowChangedEvent) TrySplitAndSortUpdateEvent() error {
+	return nil
+}
+
 // IsDelete returns true if the row is a delete event
 func (r *RowChangedEvent) IsDelete() bool {
 	return len(r.PreColumns) != 0 && len(r.Columns) == 0
@@ -720,6 +766,112 @@ func (t *SingleTableTxn) GetCommitTs() uint64 {
 	return t.CommitTs
 }
 
+// TrySplitAndSortUpdateEvent split update events if unique key is updated
+func (t *SingleTableTxn) TrySplitAndSortUpdateEvent() error {
+	if len(t.Rows) < 2 {
+		return nil
+	}
+	newRows, err := trySplitAndSortUpdateEvent(t.Rows)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	t.Rows = newRows
+	return nil
+}
+
+// trySplitAndSortUpdateEvent try to split update events if unique key is updated
+// returns true if some updated events is split
+func trySplitAndSortUpdateEvent(
+	events []*RowChangedEvent,
+) ([]*RowChangedEvent, error) {
+	rowChangedEvents := make([]*RowChangedEvent, 0, len(events))
+	split := false
+	for _, e := range events {
+		if e == nil {
+			log.Warn("skip emit nil event",
+				zap.Any("event", e))
+			continue
+		}
+
+		colLen := len(e.Columns)
+		preColLen := len(e.PreColumns)
+		// Some transactions could generate empty row change event, such as
+		// begin; insert into t (id) values (1); delete from t where id=1; commit;
+		// Just ignore these row changed events.
+		if colLen == 0 && preColLen == 0 {
+			log.Warn("skip emit empty row event",
+				zap.Any("event", e))
+			continue
+		}
+
+		// This indicates that it is an update event. if the pk or uk is updated,
+		// we need to split it into two events (delete and insert).
+		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 {
+			rowChangedEvents = append(rowChangedEvents, e)
+		}
+	}
+	// some updated events is split, need to sort
+	if split {
+		sort.Sort(txnRows(rowChangedEvents))
+	}
+	return rowChangedEvents, nil
+}
+
+// shouldSplitUpdateEvent determines if the split event is needed to align the old format based on
+// whether the handle key column or unique key has been modified.
+// If  is modified, we need to use splitUpdateEvent to split the update event into a delete and an insert event.
+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]
+		preCol := updateEvent.PreColumns[i]
+		if col != nil && (col.Flag.IsUniqueKey() || col.Flag.IsHandleKey()) &&
+			preCol != nil && (preCol.Flag.IsUniqueKey() || preCol.Flag.IsHandleKey()) {
+			colValueString := ColumnValueString(col.Value)
+			preColValueString := ColumnValueString(preCol.Value)
+			// If one unique key columns is updated, we need to split the event row.
+			if colValueString != preColValueString {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+// splitUpdateEvent splits an update event into a delete and an insert event.
+func splitUpdateEvent(
+	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.
+	// NOTICE: Here we don't need a full deep copy because
+	// our two events need Columns and PreColumns respectively,
+	// so it won't have an impact and no more full deep copy wastes memory.
+	deleteEvent := *updateEvent
+	deleteEvent.Columns = nil
+
+	insertEvent := *updateEvent
+	// NOTICE: clean up pre cols for insert event.
+	insertEvent.PreColumns = nil
+
+	return &deleteEvent, &insertEvent, nil
+}
+
 // Append adds a row changed event into SingleTableTxn
 func (t *SingleTableTxn) Append(row *RowChangedEvent) {
 	if row.StartTs != t.StartTs || row.CommitTs != t.CommitTs || row.Table.TableID != t.Table.TableID {
diff --git a/cdc/model/sink_test.go b/cdc/model/sink_test.go
index caa0ef8854e..e0a0aed9d14 100644
--- a/cdc/model/sink_test.go
+++ b/cdc/model/sink_test.go
@@ -14,11 +14,13 @@
 package model
 
 import (
+	"sort"
 	"testing"
 
 	timodel "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/parser/types"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
@@ -396,3 +398,168 @@ func TestExchangeTablePartition(t *testing.T) {
 	require.Equal(t, "ALTER TABLE `test1`.`t1` EXCHANGE PARTITION `p0` WITH TABLE `test2`.`t2`", event.Query)
 	require.Equal(t, event.Type, timodel.ActionExchangeTablePartition)
 }
+
+func TestSortRowChangedEvent(t *testing.T) {
+	events := []*RowChangedEvent{
+		{
+			PreColumns: []*Column{{}},
+			Columns:    []*Column{{}},
+		},
+		{
+			Columns: []*Column{{}},
+		},
+		{
+			PreColumns: []*Column{{}},
+		},
+	}
+	assert.True(t, events[0].IsUpdate())
+	assert.True(t, events[1].IsInsert())
+	assert.True(t, events[2].IsDelete())
+	sort.Sort(txnRows(events))
+	assert.True(t, events[0].IsDelete())
+	assert.True(t, events[1].IsUpdate())
+	assert.True(t, events[2].IsInsert())
+}
+
+func TestTrySplitAndSortUpdateEventNil(t *testing.T) {
+	t.Parallel()
+
+	events := []*RowChangedEvent{nil}
+	result, err := trySplitAndSortUpdateEvent(events)
+	require.NoError(t, err)
+	require.Equal(t, 0, len(result))
+}
+
+func TestTrySplitAndSortUpdateEventEmpty(t *testing.T) {
+	t.Parallel()
+
+	events := []*RowChangedEvent{
+		{
+			StartTs:  1,
+			CommitTs: 2,
+		},
+	}
+	result, err := trySplitAndSortUpdateEvent(events)
+	require.NoError(t, err)
+	require.Equal(t, 0, len(result))
+}
+
+func TestTrySplitAndSortUpdateEvent(t *testing.T) {
+	t.Parallel()
+
+	// Update handle key.
+	columns := []*Column{
+		{
+			Name:  "col1",
+			Flag:  BinaryFlag,
+			Value: "col1-value-updated",
+		},
+		{
+			Name:  "col2",
+			Flag:  HandleKeyFlag,
+			Value: "col2-value-updated",
+		},
+	}
+	preColumns := []*Column{
+		{
+			Name:  "col1",
+			Flag:  BinaryFlag,
+			Value: "col1-value",
+		},
+		{
+			Name:  "col2",
+			Flag:  HandleKeyFlag,
+			Value: "col2-value",
+		},
+	}
+
+	events := []*RowChangedEvent{
+		{
+			CommitTs:   1,
+			Columns:    columns,
+			PreColumns: preColumns,
+		},
+	}
+	result, err := trySplitAndSortUpdateEvent(events)
+	require.NoError(t, err)
+	require.Equal(t, 2, len(result))
+	require.True(t, result[0].IsDelete())
+	require.True(t, result[1].IsInsert())
+
+	// Update unique key.
+	columns = []*Column{
+		{
+			Name:  "col1",
+			Flag:  BinaryFlag,
+			Value: "col1-value-updated",
+		},
+		{
+			Name:  "col2",
+			Flag:  UniqueKeyFlag,
+			Value: "col2-value-updated",
+		},
+	}
+	preColumns = []*Column{
+		{
+			Name:  "col1",
+			Flag:  BinaryFlag,
+			Value: "col1-value",
+		},
+		{
+			Name:  "col2",
+			Flag:  UniqueKeyFlag,
+			Value: "col2-value",
+		},
+	}
+
+	events = []*RowChangedEvent{
+		{
+			CommitTs:   1,
+			Columns:    columns,
+			PreColumns: preColumns,
+		},
+	}
+	result, err = trySplitAndSortUpdateEvent(events)
+	require.NoError(t, err)
+	require.Equal(t, 2, len(result))
+	require.True(t, result[0].IsDelete())
+	require.True(t, result[0].IsDelete())
+	require.True(t, result[1].IsInsert())
+
+	// Update non-handle key.
+	columns = []*Column{
+		{
+			Name:  "col1",
+			Flag:  BinaryFlag,
+			Value: "col1-value-updated",
+		},
+		{
+			Name:  "col2",
+			Flag:  HandleKeyFlag,
+			Value: "col2-value",
+		},
+	}
+	preColumns = []*Column{
+		{
+			Name:  "col1",
+			Flag:  BinaryFlag,
+			Value: "col1-value",
+		},
+		{
+			Name:  "col2",
+			Flag:  HandleKeyFlag,
+			Value: "col2-value",
+		},
+	}
+
+	events = []*RowChangedEvent{
+		{
+			CommitTs:   1,
+			Columns:    columns,
+			PreColumns: preColumns,
+		},
+	}
+	result, err = trySplitAndSortUpdateEvent(events)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(result))
+}
diff --git a/cdc/sink/dmlsink/event.go b/cdc/sink/dmlsink/event.go
index f03fa28a60c..da9466a90de 100644
--- a/cdc/sink/dmlsink/event.go
+++ b/cdc/sink/dmlsink/event.go
@@ -22,6 +22,8 @@ import (
 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() error
 }
 
 // CallbackFunc is the callback function for callbackable event.
diff --git a/cdc/sink/tablesink/table_sink_impl.go b/cdc/sink/tablesink/table_sink_impl.go
index 521b8828e58..332cccba64c 100644
--- a/cdc/sink/tablesink/table_sink_impl.go
+++ b/cdc/sink/tablesink/table_sink_impl.go
@@ -110,6 +110,9 @@ func (e *EventTableSink[E, P]) UpdateResolvedTs(resolvedTs model.ResolvedTs) err
 
 	resolvedCallbackableEvents := make([]*dmlsink.CallbackableEvent[E], 0, len(resolvedEvents))
 	for _, ev := range resolvedEvents {
+		if err := ev.TrySplitAndSortUpdateEvent(); err != nil {
+			return SinkInternalError{err}
+		}
 		// We have to record the event ID for the callback.
 		ce := &dmlsink.CallbackableEvent[E]{
 			Event:     ev,
diff --git a/tests/integration_tests/http_api/util/test_case.py b/tests/integration_tests/http_api/util/test_case.py
index f85128b39c6..605aa2e9704 100644
--- a/tests/integration_tests/http_api/util/test_case.py
+++ b/tests/integration_tests/http_api/util/test_case.py
@@ -265,7 +265,7 @@ def get_processor():
     resp = rq.get(base_url)
     assert resp.status_code == rq.codes.ok
     data = resp.json()[0]
-    time.sleep(1)
+    time.sleep(2)
     url = base_url + "/" + data["changefeed_id"] + "/" + data["capture_id"]
     resp = rq.get(url)
     # print error message for debug 
diff --git a/tests/integration_tests/http_api_tls/util/test_case.py b/tests/integration_tests/http_api_tls/util/test_case.py
index b614c6c707f..16a7be18efb 100644
--- a/tests/integration_tests/http_api_tls/util/test_case.py
+++ b/tests/integration_tests/http_api_tls/util/test_case.py
@@ -254,7 +254,7 @@ def get_processor():
     resp = rq.get(base_url, cert=CERT, verify=VERIFY)
     assert resp.status_code == rq.codes.ok
     data = resp.json()[0]
-    time.sleep(1)
+    time.sleep(2)
     url = base_url + "/" + data["changefeed_id"] + "/" + data["capture_id"]
     resp = rq.get(url, cert=CERT, verify=VERIFY)
     # print error message for debug