Skip to content

Commit f80995c

Browse files
asddongmenasddongmen
authored
ddl (ticdc): support replicate ddl in BDR mode (#10299)
close #10301
1 parent 368d20f commit f80995c

File tree

17 files changed

+293
-101
lines changed

17 files changed

+293
-101
lines changed

‎cdc/model/sink.go

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -690,6 +690,9 @@ type DDLEvent struct {
690690
Charset string `msg:"-"`
691691
Collate string `msg:"-"`
692692
IsBootstrap bool `msg:"-"`
693+
// BDRRole is the role of the TiDB cluster, it is used to determine whether
694+
// the DDL is executed by the primary cluster.
695+
BDRRole string `msg:"-"`
693696
}
694697

695698
// FromJob fills the values with DDLEvent from DDL job
@@ -710,7 +713,7 @@ func (d *DDLEvent) FromJobWithArgs(
710713
d.TableInfo = tableInfo
711714
d.Charset = job.Charset
712715
d.Collate = job.Collate
713-
716+
d.BDRRole = job.BDRRole
714717
switch d.Type {
715718
// The query for "DROP TABLE" and "DROP VIEW" statements need
716719
// to be rebuilt. The reason is elaborated as follows:

‎cdc/owner/ddl_manager.go

Lines changed: 10 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ import (
2222
"github.com/pingcap/errors"
2323
"github.com/pingcap/failpoint"
2424
"github.com/pingcap/log"
25+
"github.com/pingcap/tidb/pkg/parser/ast"
2526
timodel "github.com/pingcap/tidb/pkg/parser/model"
2627
"github.com/pingcap/tiflow/cdc/entry"
2728
"github.com/pingcap/tiflow/cdc/model"
@@ -332,12 +333,15 @@ func (m *ddlManager) executeDDL(ctx context.Context) error {
332333
return nil
333334
}
334335

335-
// If changefeed is in BDRMode, skip ddl.
336-
if m.BDRMode {
337-
log.Info("changefeed is in BDRMode, skip a ddl event",
336+
// In a BDR mode cluster, TiCDC can receive DDLs from all roles of TiDB.
337+
// However, CDC only executes the DDLs from the TiDB that has BDRRolePrimary role.
338+
if m.BDRMode && m.executingDDL.BDRRole != string(ast.BDRRolePrimary) {
339+
log.Info("changefeed is in BDRMode and "+
340+
"the DDL is not executed by Primary Cluster, skip it",
338341
zap.String("namespace", m.changfeedID.Namespace),
339342
zap.String("ID", m.changfeedID.ID),
340-
zap.Any("ddlEvent", m.executingDDL))
343+
zap.Any("ddlEvent", m.executingDDL),
344+
zap.String("bdrRole", m.executingDDL.BDRRole))
341345
tableName := m.executingDDL.TableInfo.TableName
342346
// Set it to nil first to accelerate GC.
343347
m.pendingDDLs[tableName][0] = nil
@@ -373,6 +377,7 @@ func (m *ddlManager) executeDDL(ctx context.Context) error {
373377
tableName := m.executingDDL.TableInfo.TableName
374378
log.Info("execute a ddl event successfully",
375379
zap.String("ddl", m.executingDDL.Query),
380+
zap.String("namespace", m.executingDDL.BDRRole),
376381
zap.Uint64("commitTs", m.executingDDL.CommitTs),
377382
zap.Stringer("table", tableName),
378383
)
@@ -526,12 +531,8 @@ func (m *ddlManager) allPhysicalTables(ctx context.Context) ([]model.TableID, er
526531

527532
// getSnapshotTs returns the ts that we should use
528533
// to get the snapshot of the schema, the rules are:
529-
// 1. If the changefeed is just started, we use the startTs,
534+
// If the changefeed is just started, we use the startTs,
530535
// otherwise we use the checkpointTs.
531-
// 2. If the changefeed is in BDRMode, we use the ddlManager.ddlResolvedTs.
532-
// Since TiCDC ignore the DDLs in BDRMode, we don't need to care about whether
533-
// the DDLs are executed or not. We should use the ddlResolvedTs to get the up-to-date
534-
// schema.
535536
func (m *ddlManager) getSnapshotTs() (ts uint64) {
536537
ts = m.checkpointTs
537538

@@ -549,10 +550,6 @@ func (m *ddlManager) getSnapshotTs() (ts uint64) {
549550
return
550551
}
551552

552-
if m.BDRMode {
553-
ts = m.ddlResolvedTs
554-
}
555-
556553
log.Debug("snapshotTs", zap.Uint64("ts", ts))
557554
return ts
558555
}

‎cdc/owner/ddl_manager_test.go

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -145,18 +145,18 @@ func TestGetSnapshotTs(t *testing.T) {
145145
dm := createDDLManagerForTest(t)
146146
dm.startTs = 0
147147
dm.checkpointTs = 1
148-
require.Equal(t, dm.getSnapshotTs(), dm.startTs)
148+
require.Equal(t, dm.startTs, dm.getSnapshotTs())
149149

150150
dm.startTs = 1
151151
dm.checkpointTs = 10
152152
dm.BDRMode = true
153153
dm.ddlResolvedTs = 15
154-
require.Equal(t, dm.getSnapshotTs(), dm.ddlResolvedTs)
154+
require.Equal(t, dm.checkpointTs, dm.getSnapshotTs())
155155

156156
dm.startTs = 1
157157
dm.checkpointTs = 10
158158
dm.BDRMode = false
159-
require.Equal(t, dm.getSnapshotTs(), dm.checkpointTs)
159+
require.Equal(t, dm.checkpointTs, dm.getSnapshotTs())
160160
}
161161

162162
func TestExecRenameTablesDDL(t *testing.T) {

‎cdc/owner/schema_test.go

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,7 @@ func TestAllTables(t *testing.T) {
115115
require.Equal(t, model.TableName{
116116
Schema: "test",
117117
Table: "t1",
118-
TableID: 102,
118+
TableID: 104,
119119
}, tableName)
120120
// add ineligible table
121121
job = helper.DDL2Job("create table test.t2(id int)")
@@ -127,7 +127,7 @@ func TestAllTables(t *testing.T) {
127127
require.Equal(t, model.TableName{
128128
Schema: "test",
129129
Table: "t1",
130-
TableID: 102,
130+
TableID: 104,
131131
}, tableName)
132132
}
133133

‎cdc/sink/ddlsink/mysql/mysql_ddl_sink.go

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import (
2626
"github.com/pingcap/tiflow/cdc/sink/ddlsink"
2727
"github.com/pingcap/tiflow/cdc/sink/metrics"
2828
"github.com/pingcap/tiflow/pkg/config"
29+
"github.com/pingcap/tiflow/pkg/errors"
2930
cerror "github.com/pingcap/tiflow/pkg/errors"
3031
"github.com/pingcap/tiflow/pkg/errorutil"
3132
"github.com/pingcap/tiflow/pkg/quotes"
@@ -89,6 +90,11 @@ func NewDDLSink(
8990
return nil, err
9091
}
9192

93+
cfg.IsWriteSourceExisted, err = pmysql.CheckIfBDRModeIsSupported(ctx, db)
94+
if err != nil {
95+
return nil, err
96+
}
97+
9298
m := &DDLSink{
9399
id: changefeedID,
94100
db: db,
@@ -201,6 +207,18 @@ func (m *DDLSink) execDDL(pctx context.Context, ddl *model.DDLEvent) error {
201207
}
202208
}
203209

210+
// we try to set cdc write source for the ddl
211+
if err = pmysql.SetWriteSource(pctx, m.cfg, tx); err != nil {
212+
if rbErr := tx.Rollback(); rbErr != nil {
213+
if errors.Cause(rbErr) != context.Canceled {
214+
log.Error("Failed to rollback",
215+
zap.String("namespace", m.id.Namespace),
216+
zap.String("changefeed", m.id.ID), zap.Error(err))
217+
}
218+
}
219+
return err
220+
}
221+
204222
if _, err = tx.ExecContext(ctx, ddl.Query); err != nil {
205223
if rbErr := tx.Rollback(); rbErr != nil {
206224
log.Error("Failed to rollback", zap.String("sql", ddl.Query),

‎cdc/sink/ddlsink/mysql/mysql_ddl_sink_test.go

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,12 +48,19 @@ func TestWriteDDLEvent(t *testing.T) {
4848
require.Nil(t, err)
4949
mock.ExpectQuery("select tidb_version()").
5050
WillReturnRows(sqlmock.NewRows([]string{"tidb_version()"}).AddRow("5.7.25-TiDB-v4.0.0-beta-191-ga1b3e3b"))
51+
mock.ExpectQuery("select tidb_version()").
52+
WillReturnRows(sqlmock.NewRows([]string{"tidb_version()"}).AddRow("5.7.25-TiDB-v4.0.0-beta-191-ga1b3e3b"))
53+
mock.ExpectExec("SET SESSION tidb_cdc_write_source = 1").WillReturnResult(sqlmock.NewResult(1, 0))
54+
5155
mock.ExpectBegin()
5256
mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(1, 1))
57+
mock.ExpectExec("SET SESSION tidb_cdc_write_source = 0").WillReturnResult(sqlmock.NewResult(1, 0))
5358
mock.ExpectExec("ALTER TABLE test.t1 ADD COLUMN a int").WillReturnResult(sqlmock.NewResult(1, 1))
5459
mock.ExpectCommit()
60+
5561
mock.ExpectBegin()
5662
mock.ExpectExec("USE `test`;").WillReturnResult(sqlmock.NewResult(1, 1))
63+
mock.ExpectExec("SET SESSION tidb_cdc_write_source = 0").WillReturnResult(sqlmock.NewResult(1, 0))
5764
mock.ExpectExec("ALTER TABLE test.t1 ADD COLUMN a int").
5865
WillReturnError(&dmysql.MySQLError{
5966
Number: uint16(infoschema.ErrColumnExists.Code()),
@@ -163,6 +170,10 @@ func TestAsyncExecAddIndex(t *testing.T) {
163170
require.Nil(t, err)
164171
mock.ExpectQuery("select tidb_version()").
165172
WillReturnRows(sqlmock.NewRows([]string{"tidb_version()"}).AddRow("5.7.25-TiDB-v4.0.0-beta-191-ga1b3e3b"))
173+
mock.ExpectQuery("select tidb_version()").WillReturnError(&dmysql.MySQLError{
174+
Number: 1305,
175+
Message: "FUNCTION test.tidb_version does not exist",
176+
})
166177
mock.ExpectBegin()
167178
mock.ExpectExec("USE `test`;").
168179
WillReturnResult(sqlmock.NewResult(1, 1))

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

Lines changed: 1 addition & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -766,7 +766,7 @@ func (s *mysqlBackend) execDMLWithMaxRetries(pctx context.Context, dmls *prepare
766766
// Set session variables first and then execute the transaction.
767767
// we try to set write source for each txn,
768768
// so we can use it to trace the data source
769-
if err = s.setWriteSource(pctx, tx); err != nil {
769+
if err = pmysql.SetWriteSource(pctx, s.cfg, tx); err != nil {
770770
err := logDMLTxnErr(
771771
cerror.WrapError(cerror.ErrMySQLTxnError, err),
772772
start, s.changefeed,
@@ -872,24 +872,3 @@ func getSQLErrCode(err error) (errors.ErrCode, bool) {
872872
func (s *mysqlBackend) setDMLMaxRetry(maxRetry uint64) {
873873
s.dmlMaxRetry = maxRetry
874874
}
875-
876-
// setWriteSource sets write source for the transaction.
877-
func (s *mysqlBackend) setWriteSource(ctx context.Context, txn *sql.Tx) error {
878-
// we only set write source when donwstream is TiDB and write source is existed.
879-
if !s.cfg.IsWriteSourceExisted {
880-
return nil
881-
}
882-
// downstream is TiDB, set system variables.
883-
// We should always try to set this variable, and ignore the error if
884-
// downstream does not support this variable, it is by design.
885-
query := fmt.Sprintf("SET SESSION %s = %d", "tidb_cdc_write_source", s.cfg.SourceID)
886-
_, err := txn.ExecContext(ctx, query)
887-
if err != nil {
888-
if mysqlErr, ok := errors.Cause(err).(*dmysql.MySQLError); ok &&
889-
mysqlErr.Number == mysql.ErrUnknownSystemVariable {
890-
return nil
891-
}
892-
return err
893-
}
894-
return nil
895-
}

‎go.mod

Lines changed: 14 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -64,14 +64,14 @@ require (
6464
github.com/phayes/freeport v0.0.0-20180830031419-95f893ade6f2
6565
github.com/pierrec/lz4/v4 v4.1.18
6666
github.com/pingcap/check v0.0.0-20211026125417-57bd13f7b5f0
67-
github.com/pingcap/errors v0.11.5-0.20221009092201-b66cddb77c32
67+
github.com/pingcap/errors v0.11.5-0.20231212100244-799fae176cfb
6868
github.com/pingcap/failpoint v0.0.0-20220801062533-2eaa32854a6c
69-
github.com/pingcap/kvproto v0.0.0-20231204093812-96c40585233f
69+
github.com/pingcap/kvproto v0.0.0-20231222062942-c0c73f41d0b2
7070
github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22
71-
github.com/pingcap/tidb v1.1.0-beta.0.20231212043317-b478056bbf73
71+
github.com/pingcap/tidb v1.1.0-beta.0.20240105042433-54d8a1416ab0
7272
github.com/pingcap/tidb-tools v0.0.0-20231228035519-c4bdf178b3d6
73-
github.com/pingcap/tidb/pkg/parser v0.0.0-20231212043317-b478056bbf73
74-
github.com/prometheus/client_golang v1.17.0
73+
github.com/pingcap/tidb/pkg/parser v0.0.0-20231229060758-e19e06e1bc19
74+
github.com/prometheus/client_golang v1.18.0
7575
github.com/prometheus/client_model v0.5.0
7676
github.com/r3labs/diff v1.1.0
7777
github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475
@@ -88,9 +88,9 @@ require (
8888
github.com/swaggo/swag v1.16.2
8989
github.com/syndtr/goleveldb v1.0.1-0.20210305035536-64b5b1c73954
9090
github.com/thanhpk/randstr v1.0.6
91-
github.com/tikv/client-go/v2 v2.0.8-0.20231201024404-0ff16620f6c0
91+
github.com/tikv/client-go/v2 v2.0.8-0.20231227070846-61c486af13a5
9292
github.com/tikv/pd v1.1.0-beta.0.20231212061647-ab97b9a267f3
93-
github.com/tikv/pd/client v0.0.0-20231130081618-862eee18738e
93+
github.com/tikv/pd/client v0.0.0-20240103101103-a4d2f1ca365a
9494
github.com/tinylib/msgp v1.1.6
9595
github.com/uber-go/atomic v1.4.0
9696
github.com/vmihailenco/msgpack/v5 v5.3.5
@@ -105,21 +105,21 @@ require (
105105
go.uber.org/atomic v1.11.0
106106
go.uber.org/dig v1.13.0
107107
go.uber.org/goleak v1.3.0
108-
go.uber.org/mock v0.3.0
108+
go.uber.org/mock v0.4.0
109109
go.uber.org/multierr v1.11.0
110110
go.uber.org/ratelimit v0.2.0
111111
go.uber.org/zap v1.26.0
112-
golang.org/x/exp v0.0.0-20231206192017-f3f8817b8deb
112+
golang.org/x/exp v0.0.0-20231226003508-02704c960a9b
113113
golang.org/x/net v0.19.0
114114
golang.org/x/oauth2 v0.15.0
115115
golang.org/x/sync v0.5.0
116116
golang.org/x/sys v0.15.0
117117
golang.org/x/text v0.14.0
118118
golang.org/x/time v0.5.0
119119
google.golang.org/genproto/googleapis/api v0.0.0-20231212172506-995d672761c0
120-
google.golang.org/genproto/googleapis/rpc v0.0.0-20231211222908-989df2bf70f3
121-
google.golang.org/grpc v1.60.0
122-
google.golang.org/protobuf v1.31.0
120+
google.golang.org/genproto/googleapis/rpc v0.0.0-20231212172506-995d672761c0
121+
google.golang.org/grpc v1.60.1
122+
google.golang.org/protobuf v1.32.0
123123
gopkg.in/yaml.v2 v2.4.0
124124
gorm.io/driver/mysql v1.4.5
125125
gorm.io/gorm v1.24.5
@@ -149,6 +149,7 @@ require (
149149
github.com/go-ldap/ldap/v3 v3.4.4 // indirect
150150
github.com/go-logr/logr v1.3.0 // indirect
151151
github.com/go-logr/stdr v1.2.2 // indirect
152+
github.com/go-resty/resty/v2 v2.7.0 // indirect
152153
github.com/golang-jwt/jwt/v4 v4.4.2 // indirect
153154
github.com/golang-jwt/jwt/v5 v5.0.0 // indirect
154155
github.com/google/gofuzz v1.2.0 // indirect
@@ -363,7 +364,7 @@ require (
363364
go.opentelemetry.io/otel/sdk v1.21.0 // indirect
364365
go.opentelemetry.io/otel/trace v1.21.0 // indirect
365366
go.opentelemetry.io/proto/otlp v1.0.0 // indirect
366-
golang.org/x/crypto v0.16.0 // indirect
367+
golang.org/x/crypto v0.17.0 // indirect
367368
golang.org/x/mod v0.14.0 // indirect
368369
golang.org/x/term v0.15.0 // indirect
369370
golang.org/x/tools v0.16.1 // indirect

0 commit comments

Comments
 (0)