Skip to content

Commit

Permalink
redo(ticdc): fix orderliness of redo (#11320)
Browse files Browse the repository at this point in the history 
close #11096
  • Loading branch information
@hongyunyan
hongyunyan authored Jun 17, 2024
1 parent 66d7a1c commit ffbe6f8
Show file tree
Hide file tree
Showing 3 changed files with 139 additions and 48 deletions.
15 changes: 15 additions & 0 deletions cdc/model/sink.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -290,6 +290,21 @@ type RedoRowChangedEvent struct {
Columns []*RedoColumn `msg:"columns"`
}

// IsDelete returns true if the row is a delete event
func (r *RedoRowChangedEvent) IsDelete() bool {
return len(r.PreColumns) != 0 && len(r.Columns) == 0
}

// IsInsert returns true if the row is an insert event
func (r *RedoRowChangedEvent) IsInsert() bool {
return len(r.PreColumns) == 0 && len(r.Columns) != 0
}

// IsUpdate returns true if the row is an update event
func (r *RedoRowChangedEvent) IsUpdate() bool {
return len(r.PreColumns) != 0 && len(r.Columns) != 0
}

// RedoDDLEvent represents DDL event used in redo log persistent
type RedoDDLEvent struct {
DDL *DDLEvent `msg:"ddl"`
Expand Down
7 changes: 4 additions & 3 deletions cdc/redo/reader/reader.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -382,11 +382,12 @@ func (h logHeap) Less(i, j int) bool {
return h[i].data.RedoRow.Row.StartTs < h[j].data.RedoRow.Row.StartTs
}
// in the same txn, we need to sort by delete/update/insert order
if h[i].data.RedoRow.Row.IsDelete() {
if h[i].data.RedoRow.IsDelete() {
return true
} else if h[i].data.RedoRow.Row.IsUpdate() {
return !h[j].data.RedoRow.Row.IsDelete()
} else if h[i].data.RedoRow.IsUpdate() {
return !h[j].data.RedoRow.IsDelete()
}

return false
}

Expand Down
165 changes: 120 additions & 45 deletions cdc/redo/reader/reader_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -267,11 +267,16 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 2,
},
Expand All @@ -292,20 +297,30 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 2,
},
},
Columns: []*model.Column{
{
},
Columns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 3,
},
Expand Down Expand Up @@ -334,20 +349,30 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 2,
},
},
Columns: []*model.Column{
{
},
Columns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 3,
},
Expand Down Expand Up @@ -401,20 +426,30 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 2,
},
},
Columns: []*model.Column{
{
},
Columns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 3,
},
Expand All @@ -435,11 +470,16 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 1,
},
Expand Down Expand Up @@ -468,20 +508,30 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 2,
},
},
Columns: []*model.Column{
{
},
Columns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 3,
},
Expand Down Expand Up @@ -544,11 +594,16 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 1,
},
Expand Down Expand Up @@ -684,20 +739,30 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 2,
},
},
Columns: []*model.Column{
{
},
Columns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 3,
},
Expand All @@ -718,20 +783,30 @@ func TestLogHeapLess(t *testing.T) {
TableID: 1,
IsPartition: false,
},
PreColumns: []*model.Column{
{
},
PreColumns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 1,
},
},
Columns: []*model.Column{
{
},
Columns: []*model.RedoColumn{
{
Column: &model.Column{
Name: "col-1",
Value: 1,
}, {
},
},
{
Column: &model.Column{
Name: "col-2",
Value: 3,
},
Expand Down

0 comments on commit ffbe6f8

Please sign in to comment.