Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[DNM] *(ticdc): make the row event physical table id field private to avoid misuse #11855

Open
wants to merge 5 commits into
base: master
Choose a base branch
from
Open

[DNM] *(ticdc): make the row event physical table id field private to avoid misuse #11855

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
ceed123
make the physical table id private
3AceShowHand Dec 10, 2024
d54fb81
make fmt
3AceShowHand Dec 10, 2024
44db34f
make fmt
3AceShowHand Dec 10, 2024
7dbc7de
fix the mysql gen key unit test
3AceShowHand Dec 10, 2024
f1993a0
Merge branch 'master' into hidden-table-id
3AceShowHand Dec 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 11 additions & 10 deletions cdc/entry/mounter.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -792,20 +792,21 @@ func (m *mounter) mountRowKVEntry(
}
}

return &model.RowChangedEvent{
StartTs: row.StartTs,
CommitTs: row.CRTs,
RowID: intRowID,
HandleKey: row.RecordID,
PhysicalTableID: row.PhysicalTableID,
TableInfo: tableInfo,
Columns: cols,
PreColumns: preCols,
event := &model.RowChangedEvent{
StartTs: row.StartTs,
CommitTs: row.CRTs,
RowID: intRowID,
HandleKey: row.RecordID,
TableInfo: tableInfo,
Columns: cols,
PreColumns: preCols,

Checksum: checksum,

ApproximateDataSize: dataSize,
}, rawRow, nil
}
event.SetTableID(row.PhysicalTableID)
return event, rawRow, nil
}

var emptyBytes = make([]byte, 0)
Expand Down
23 changes: 14 additions & 9 deletions cdc/model/sink.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -329,7 +329,7 @@ type RowChangedEvent struct {

RowID int64 // Deprecated. It is empty when the RowID comes from clustered index table.

PhysicalTableID int64
physicalTableID int64

// NOTICE: We probably store the logical ID inside TableInfo's TableName,
// not the physical ID.
Expand Down Expand Up @@ -397,13 +397,13 @@ func (r *RowChangedEventInRedoLog) ToRowChangedEvent() *RowChangedEvent {
tableInfo.TableName.TableID = r.Table.TableID
tableInfo.TableName.IsPartition = r.Table.IsPartition
row := &RowChangedEvent{
StartTs: r.StartTs,
CommitTs: r.CommitTs,
PhysicalTableID: r.Table.TableID,
TableInfo: tableInfo,
Columns: Columns2ColumnDatas(r.Columns, tableInfo),
PreColumns: Columns2ColumnDatas(r.PreColumns, tableInfo),
StartTs: r.StartTs,
CommitTs: r.CommitTs,
TableInfo: tableInfo,
Columns: Columns2ColumnDatas(r.Columns, tableInfo),
PreColumns: Columns2ColumnDatas(r.PreColumns, tableInfo),
}
row.SetTableID(r.Table.TableID)
return row
}

Expand Down Expand Up @@ -436,7 +436,12 @@ func (e txnRows) Swap(i, j int) {

// GetTableID returns the table ID of the event.
func (r *RowChangedEvent) GetTableID() int64 {
return r.PhysicalTableID
return r.physicalTableID
}

// SetTableID set the row event physical table id
func (r *RowChangedEvent) SetTableID(tableID int64) {
r.physicalTableID = tableID
}

// GetCommitTs returns the commit timestamp of this event.
Expand Down Expand Up @@ -701,7 +706,7 @@ func NewIncrementalColumnIDAllocator() *IncrementalColumnIDAllocator {
}

// GetColumnID return the next mock column id
func (d *IncrementalColumnIDAllocator) GetColumnID(name string) int64 {
func (d *IncrementalColumnIDAllocator) GetColumnID(_ string) int64 {
result := d.nextColID
d.nextColID += 1
return result
Expand Down
8 changes: 4 additions & 4 deletions cdc/model/sink_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -648,10 +648,9 @@ func TestToRedoLog(t *testing.T) {
}
tableInfo := BuildTableInfo("test", "t", cols, [][]int{{1}})
event := &RowChangedEvent{
StartTs: 100,
CommitTs: 1000,
PhysicalTableID: 1,
TableInfo: tableInfo,
StartTs: 100,
CommitTs: 1000,
TableInfo: tableInfo,
Columns: Columns2ColumnDatas([]*Column{
{
Name: "col1",
Expand All @@ -663,6 +662,7 @@ func TestToRedoLog(t *testing.T) {
},
}, tableInfo),
}
event.SetTableID(1)
eventInRedoLog := event.ToRedoLog()
require.Equal(t, event.StartTs, eventInRedoLog.RedoRow.Row.StartTs)
require.Equal(t, event.CommitTs, eventInRedoLog.RedoRow.Row.CommitTs)
Expand Down
17 changes: 9 additions & 8 deletions cdc/processor/sinkmanager/table_sink_worker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,14 +86,15 @@ func genRowChangedEvent(startTs, commitTs uint64, span tablepb.Span) *model.RowC
{Name: "a", Value: 1},
}
tableInfo := model.BuildTableInfo("table", "table", columns, nil)
return &model.RowChangedEvent{
StartTs: startTs,
CommitTs: commitTs,
PhysicalTableID: span.TableID,
TableInfo: tableInfo,
Columns: model.Columns2ColumnDatas(columns, tableInfo),
PreColumns: model.Columns2ColumnDatas(preColumns, tableInfo),
}
result := &model.RowChangedEvent{
StartTs: startTs,
CommitTs: commitTs,
TableInfo: tableInfo,
Columns: model.Columns2ColumnDatas(columns, tableInfo),
PreColumns: model.Columns2ColumnDatas(preColumns, tableInfo),
}
result.SetTableID(span.TableID)
return result
}

type tableSinkWorkerSuite struct {
Expand Down
66 changes: 46 additions & 20 deletions cdc/redo/manager_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -149,37 +149,58 @@ func TestLogManagerInProcessor(t *testing.T) {
tableInfo := &model.TableInfo{
TableName: model.TableName{Schema: "test", Table: "t"},
}
span1 := spanz.TableIDToComparableSpan(53)
event1 := &model.RowChangedEvent{CommitTs: 120, TableInfo: tableInfo}
event1.SetTableID(span1.TableID)
event2 := &model.RowChangedEvent{CommitTs: 125, TableInfo: tableInfo}
event2.SetTableID(span1.TableID)
event3 := &model.RowChangedEvent{CommitTs: 130, TableInfo: tableInfo}
event3.SetTableID(span1.TableID)

span2 := spanz.TableIDToComparableSpan(55)
event4 := &model.RowChangedEvent{CommitTs: 130, TableInfo: tableInfo}
event4.SetTableID(span2.TableID)
event5 := &model.RowChangedEvent{CommitTs: 135, TableInfo: tableInfo}
event5.SetTableID(span2.TableID)

span3 := spanz.TableIDToComparableSpan(57)
event6 := &model.RowChangedEvent{CommitTs: 130, TableInfo: tableInfo}
event6.SetTableID(span3.TableID)

span4 := spanz.TableIDToComparableSpan(59)
event7 := &model.RowChangedEvent{CommitTs: 128, TableInfo: tableInfo}
event7.SetTableID(span4.TableID)
event8 := &model.RowChangedEvent{CommitTs: 130, TableInfo: tableInfo}
event8.SetTableID(span4.TableID)
event9 := &model.RowChangedEvent{CommitTs: 133, TableInfo: tableInfo}
event9.SetTableID(span4.TableID)

testCases := []struct {
span tablepb.Span
rows []*model.RowChangedEvent
}{
{
span: spanz.TableIDToComparableSpan(53),
span: span1,
rows: []*model.RowChangedEvent{
{CommitTs: 120, PhysicalTableID: 53, TableInfo: tableInfo},
{CommitTs: 125, PhysicalTableID: 53, TableInfo: tableInfo},
{CommitTs: 130, PhysicalTableID: 53, TableInfo: tableInfo},
event1, event2, event3,
},
},
{
span: spanz.TableIDToComparableSpan(55),
span: span2,
rows: []*model.RowChangedEvent{
{CommitTs: 130, PhysicalTableID: 55, TableInfo: tableInfo},
{CommitTs: 135, PhysicalTableID: 55, TableInfo: tableInfo},
event4, event5,
},
},
{
span: spanz.TableIDToComparableSpan(57),
span: span3,
rows: []*model.RowChangedEvent{
{CommitTs: 130, PhysicalTableID: 57, TableInfo: tableInfo},
event6,
},
},
{
span: spanz.TableIDToComparableSpan(59),
span: span4,
rows: []*model.RowChangedEvent{
{CommitTs: 128, PhysicalTableID: 59, TableInfo: tableInfo},
{CommitTs: 130, PhysicalTableID: 59, TableInfo: tableInfo},
{CommitTs: 133, PhysicalTableID: 59, TableInfo: tableInfo},
event7, event8, event9,
},
},
}
Expand Down Expand Up @@ -298,16 +319,21 @@ func TestLogManagerError(t *testing.T) {
tableInfo := &model.TableInfo{
TableName: model.TableName{Schema: "test", Table: "t"},
}
span := spanz.TableIDToComparableSpan(53)
event1 := &model.RowChangedEvent{CommitTs: 120, TableInfo: tableInfo}
event1.SetTableID(span.TableID)
event2 := &model.RowChangedEvent{CommitTs: 125, TableInfo: tableInfo}
event2.SetTableID(span.TableID)
event3 := &model.RowChangedEvent{CommitTs: 130, TableInfo: tableInfo}
event3.SetTableID(span.TableID)
testCases := []struct {
span tablepb.Span
rows []writer.RedoEvent
}{
{
span: spanz.TableIDToComparableSpan(53),
span: span,
rows: []writer.RedoEvent{
&model.RowChangedEvent{CommitTs: 120, PhysicalTableID: 53, TableInfo: tableInfo},
&model.RowChangedEvent{CommitTs: 125, PhysicalTableID: 53, TableInfo: tableInfo},
&model.RowChangedEvent{CommitTs: 130, PhysicalTableID: 53, TableInfo: tableInfo},
event1, event2, event3,
},
},
}
Expand Down Expand Up @@ -385,10 +411,10 @@ func runBenchTest(b *testing.B, storage string, useFileBackend bool) {
// prepare new row change events
b.StopTimer()
*maxCommitTs += rand.Uint64() % 10
event := &model.RowChangedEvent{CommitTs: *maxCommitTs, TableInfo: tableInfo}
event.SetTableID(span.TableID)
rows = []*model.RowChangedEvent{
{CommitTs: *maxCommitTs, PhysicalTableID: span.TableID, TableInfo: tableInfo},
{CommitTs: *maxCommitTs, PhysicalTableID: span.TableID, TableInfo: tableInfo},
{CommitTs: *maxCommitTs, PhysicalTableID: span.TableID, TableInfo: tableInfo},
event, event, event,
}

b.StartTimer()
Expand Down
15 changes: 6 additions & 9 deletions cdc/redo/writer/file/file_log_writer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,8 @@ func TestLogWriterWriteLog(t *testing.T) {
Table: "t",
},
}
event := &model.RowChangedEvent{CommitTs: 1, TableInfo: tableInfo}
event.SetTableID(111)
tests := []struct {
name string
args arg
Expand All @@ -50,22 +52,17 @@ func TestLogWriterWriteLog(t *testing.T) {
{
name: "happy",
args: arg{
ctx: context.Background(),
rows: []writer.RedoEvent{
&model.RowChangedEvent{CommitTs: 1, PhysicalTableID: 111, TableInfo: tableInfo},
},
ctx: context.Background(),
rows: []writer.RedoEvent{event},
},
isRunning: true,
writerErr: nil,
},
{
name: "writer err",
args: arg{
ctx: context.Background(),
rows: []writer.RedoEvent{
nil,
&model.RowChangedEvent{CommitTs: 1, PhysicalTableID: 11, TableInfo: tableInfo},
},
ctx: context.Background(),
rows: []writer.RedoEvent{nil, event},
},
writerErr: errors.New("err"),
wantErr: errors.New("err"),
Expand Down
25 changes: 9 additions & 16 deletions cdc/redo/writer/memory/mem_log_writer_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,23 +31,16 @@ import (
func TestWriteDDL(t *testing.T) {
t.Parallel()

event1 := &model.RowChangedEvent{CommitTs: 11, TableInfo: &model.TableInfo{TableName: model.TableName{Schema: "test", Table: "t1"}}}
event1.SetTableID(11)

event2 := &model.RowChangedEvent{CommitTs: 15, TableInfo: &model.TableInfo{TableName: model.TableName{Schema: "test", Table: "t2"}}}
event2.SetTableID(12)

event3 := &model.RowChangedEvent{CommitTs: 8, TableInfo: &model.TableInfo{TableName: model.TableName{Schema: "test", Table: "t3"}}}
event3.SetTableID(12)
rows := []writer.RedoEvent{
nil,
&model.RowChangedEvent{
PhysicalTableID: 11,
CommitTs: 11,
TableInfo: &model.TableInfo{TableName: model.TableName{Schema: "test", Table: "t1"}},
},
&model.RowChangedEvent{
PhysicalTableID: 12,
CommitTs: 15,
TableInfo: &model.TableInfo{TableName: model.TableName{Schema: "test", Table: "t2"}},
},
&model.RowChangedEvent{
PhysicalTableID: 12,
CommitTs: 8,
TableInfo: &model.TableInfo{TableName: model.TableName{Schema: "test", Table: "t2"}},
},
nil, event1, event2, event3,
}
testWriteEvents(t, rows)
}
Expand Down
4 changes: 2 additions & 2 deletions cdc/sink/dmlsink/cloudstorage/defragmenter_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,13 +99,13 @@ func TestDeframenter(t *testing.T) {
n := 1 + rand.Intn(1000)
for j := 0; j < n; j++ {
row := &model.RowChangedEvent{
PhysicalTableID: 100,
TableInfo: tableInfo,
TableInfo: tableInfo,
Columns: []*model.ColumnData{
{ColumnID: 1, Value: j + 1},
{ColumnID: 2, Value: "hello world"},
},
}
row.SetTableID(100)
frag.event.Event.Rows = append(frag.event.Event.Rows, row)
}
err := encoder.AppendTxnEvent(frag.event.Event, nil)
Expand Down
19 changes: 9 additions & 10 deletions cdc/sink/dmlsink/cloudstorage/dml_worker_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,14 @@ func TestDMLWorkerRun(t *testing.T) {
}
tableInfo := model.WrapTableInfo(100, "test", 99, tidbTableInfo)
for i := 0; i < 5; i++ {
row := &model.RowChangedEvent{
TableInfo: tableInfo,
Columns: []*model.ColumnData{
{ColumnID: 1, Value: 100},
{ColumnID: 2, Value: "hello world"},
},
}
row.SetTableID(100)
frag := eventFragment{
seqNumber: uint64(i),
versionedTable: cloudstorage.VersionedTableName{
Expand All @@ -92,16 +100,7 @@ func TestDMLWorkerRun(t *testing.T) {
event: &dmlsink.TxnCallbackableEvent{
Event: &model.SingleTableTxn{
TableInfo: tableInfo,
Rows: []*model.RowChangedEvent{
{
PhysicalTableID: 100,
TableInfo: tableInfo,
Columns: []*model.ColumnData{
{ColumnID: 1, Value: 100},
{ColumnID: 2, Value: "hello world"},
},
},
},
Rows: []*model.RowChangedEvent{row},
},
},
encodedMsgs: []*common.Message{
Expand Down
Loading
Loading