sinkv2(ticdc): add max-multi-update-row config in mysql sink (#8191)

cdc/sinkv2/eventsink/txn/mysql/mysql.go

@@ -256,6 +256,7 @@ func convert2RowChanges(
 			tableInfo,
 			nil, nil)
 	}
+	res.SetApproximateDataSize(row.ApproximateDataSize)
 	return res
 }
 
@@ -330,7 +331,7 @@ func (s *mysqlBackend) groupRowsByType(
 				updateRow = append(
 					updateRow,
 					convert2RowChanges(row, tableInfo, sqlmodel.RowChangeUpdate))
-				if len(updateRow) >= s.cfg.MaxTxnRow {
+				if len(updateRow) >= s.cfg.MaxMultiUpdateRowCount {
 					updateRows = append(updateRows, updateRow)
 					updateRow = make([]*sqlmodel.RowChange, 0, preAllocateSize)
 				}
@@ -384,15 +385,37 @@ func (s *mysqlBackend) batchSingleTxnDmls(
 	// handle update
 	if len(updateRows) > 0 {
 		for _, rows := range updateRows {
-			sql, value := sqlmodel.GenUpdateSQL(rows...)
-			sqls = append(sqls, sql)
-			values = append(values, value)
+			s, v := s.genUpdateSQL(rows...)
+			sqls = append(sqls, s...)
+			values = append(values, v...)
 		}
 	}
 
 	return
 }
 
+func (s *mysqlBackend) genUpdateSQL(rows ...*sqlmodel.RowChange) ([]string, [][]interface{}) {
+	size, count := 0, 0
+	for _, r := range rows {
+		size += int(r.GetApproximateDataSize())
+		count++
+	}
+	if size < s.cfg.MaxMultiUpdateRowSize*count {
+		// use multi update in one SQL
+		sql, value := sqlmodel.GenUpdateSQLFast(rows...)
+		return []string{sql}, [][]interface{}{value}
+	}
+	// each row has one independent update SQL.
+	sqls := make([]string, 0, len(rows))
+	values := make([][]interface{}, 0, len(rows))
+	for _, row := range rows {
+		sql, value := row.GenSQL(sqlmodel.DMLUpdate)
+		sqls = append(sqls, sql)
+		values = append(values, value)
+	}
+	return sqls, values
+}
+
 func hasHandleKey(cols []*model.Column) bool {
 	for _, col := range cols {
 		if col == nil {

cdc/sinkv2/eventsink/txn/mysql/mysql_test.go

@@ -28,7 +28,10 @@ import (
 	dmysql "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/infoschema"
+	"github.com/pingcap/tidb/parser"
+	"github.com/pingcap/tidb/parser/ast"
 	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tiflow/cdc/contextutil"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -37,6 +40,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/sink"
 	pmysql "github.com/pingcap/tiflow/pkg/sink/mysql"
+	"github.com/pingcap/tiflow/pkg/sqlmodel"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest/observer"
@@ -1681,3 +1685,59 @@ func TestGroupRowsByType(t *testing.T) {
 		})
 	}
 }
+
+func TestBackendGenUpdateSQL(t *testing.T) {
+	ctx := context.Background()
+	ms := newMySQLBackendWithoutDB(ctx)
+	table := &model.TableName{Schema: "db", Table: "tb1"}
+
+	createSQL := "CREATE TABLE tb1 (id INT PRIMARY KEY, name varchar(20))"
+	stmt, err := parser.New().ParseOneStmt(createSQL, "", "")
+	require.NoError(t, err)
+	ti, err := ddl.BuildTableInfoFromAST(stmt.(*ast.CreateTableStmt))
+	require.NoError(t, err)
+
+	row1 := sqlmodel.NewRowChange(table, table, []any{1, "a"}, []any{1, "aa"}, ti, ti, nil)
+	row1.SetApproximateDataSize(6)
+	row2 := sqlmodel.NewRowChange(table, table, []any{2, "b"}, []any{2, "bb"}, ti, ti, nil)
+	row2.SetApproximateDataSize(6)
+
+	testCases := []struct {
+		rows                  []*sqlmodel.RowChange
+		maxMultiUpdateRowSize int
+		expectedSQLs          []string
+		expectedValues        [][]interface{}
+	}{
+		{
+			[]*sqlmodel.RowChange{row1, row2},
+			ms.cfg.MaxMultiUpdateRowCount,
+			[]string{
+				"UPDATE `db`.`tb1` SET " +
+					"`id`=CASE WHEN `id`=? THEN ? WHEN `id`=? THEN ? END, " +
+					"`name`=CASE WHEN `id`=? THEN ? WHEN `id`=? THEN ? END " +
+					"WHERE `id` IN (?,?)",
+			},
+			[][]interface{}{
+				{1, 1, 2, 2, 1, "aa", 2, "bb", 1, 2},
+			},
+		},
+		{
+			[]*sqlmodel.RowChange{row1, row2},
+			0,
+			[]string{
+				"UPDATE `db`.`tb1` SET `id` = ?, `name` = ? WHERE `id` = ? LIMIT 1",
+				"UPDATE `db`.`tb1` SET `id` = ?, `name` = ? WHERE `id` = ? LIMIT 1",
+			},
+			[][]interface{}{
+				{1, "aa", 1},
+				{2, "bb", 2},
+			},
+		},
+	}
+	for _, tc := range testCases {
+		ms.cfg.MaxMultiUpdateRowSize = tc.maxMultiUpdateRowSize
+		sqls, values := ms.genUpdateSQL(tc.rows...)
+		require.Equal(t, tc.expectedSQLs, sqls)
+		require.Equal(t, tc.expectedValues, values)
+	}
+}

pkg/sink/mysql/config.go

@@ -41,10 +41,21 @@ const (
 	defaultWorkerCount = 16
 	// defaultMaxTxnRow is the default max number of rows in a transaction.
 	defaultMaxTxnRow = 256
+	// defaultMaxMultiUpdateRowCount is the default max number of rows in a
+	// single multi update SQL.
+	defaultMaxMultiUpdateRowCount = 40
+	// defaultMaxMultiUpdateRowSize(1KB) defines the default value of MaxMultiUpdateRowSize
+	// When row average size is larger MaxMultiUpdateRowSize,
+	// disable multi update, otherwise enable multi update.
+	defaultMaxMultiUpdateRowSize = 1024
 	// The upper limit of max worker counts.
 	maxWorkerCount = 1024
 	// The upper limit of max txn rows.
 	maxMaxTxnRow = 2048
+	// The upper limit of max multi update rows in a single SQL.
+	maxMaxMultiUpdateRowCount = 256
+	// The upper limit of max multi update row size(8KB).
+	maxMaxMultiUpdateRowSize = 8192
 
 	defaultTiDBTxnMode         = txnModeOptimistic
 	defaultBatchReplaceEnabled = true
@@ -67,19 +78,21 @@ const (
 
 // Config is the configs for MySQL backend.
 type Config struct {
-	WorkerCount         int
-	MaxTxnRow           int
-	tidbTxnMode         string
-	BatchReplaceEnabled bool
-	BatchReplaceSize    int
-	ReadTimeout         string
-	WriteTimeout        string
-	DialTimeout         string
-	SafeMode            bool
-	Timezone            string
-	TLS                 string
-	ForceReplicate      bool
-	EnableOldValue      bool
+	WorkerCount            int
+	MaxTxnRow              int
+	MaxMultiUpdateRowCount int
+	MaxMultiUpdateRowSize  int
+	tidbTxnMode            string
+	BatchReplaceEnabled    bool
+	BatchReplaceSize       int
+	ReadTimeout            string
+	WriteTimeout           string
+	DialTimeout            string
+	SafeMode               bool
+	Timezone               string
+	TLS                    string
+	ForceReplicate         bool
+	EnableOldValue         bool
 
 	IsTiDB         bool // IsTiDB is true if the downstream is TiDB
 	SourceID       uint64
@@ -89,16 +102,18 @@ type Config struct {
 // NewConfig returns the default mysql backend config.
 func NewConfig() *Config {
 	return &Config{
-		WorkerCount:         defaultWorkerCount,
-		MaxTxnRow:           defaultMaxTxnRow,
-		tidbTxnMode:         defaultTiDBTxnMode,
-		BatchReplaceEnabled: defaultBatchReplaceEnabled,
-		BatchReplaceSize:    defaultBatchReplaceSize,
-		ReadTimeout:         defaultReadTimeout,
-		WriteTimeout:        defaultWriteTimeout,
-		DialTimeout:         defaultDialTimeout,
-		SafeMode:            defaultSafeMode,
-		BatchDMLEnable:      defaultBatchDMLEnable,
+		WorkerCount:            defaultWorkerCount,
+		MaxTxnRow:              defaultMaxTxnRow,
+		MaxMultiUpdateRowCount: defaultMaxMultiUpdateRowCount,
+		MaxMultiUpdateRowSize:  defaultMaxMultiUpdateRowSize,
+		tidbTxnMode:            defaultTiDBTxnMode,
+		BatchReplaceEnabled:    defaultBatchReplaceEnabled,
+		BatchReplaceSize:       defaultBatchReplaceSize,
+		ReadTimeout:            defaultReadTimeout,
+		WriteTimeout:           defaultWriteTimeout,
+		DialTimeout:            defaultDialTimeout,
+		SafeMode:               defaultSafeMode,
+		BatchDMLEnable:         defaultBatchDMLEnable,
 	}
 }
 
@@ -124,6 +139,12 @@ func (c *Config) Apply(
 	if err = getMaxTxnRow(query, &c.MaxTxnRow); err != nil {
 		return err
 	}
+	if err = getMaxMultiUpdateRowCount(query, &c.MaxMultiUpdateRowCount); err != nil {
+		return err
+	}
+	if err = getMaxMultiUpdateRowSize(query, &c.MaxMultiUpdateRowSize); err != nil {
+		return err
+	}
 	if err = getTiDBTxnMode(query, &c.tidbTxnMode); err != nil {
 		return err
 	}
@@ -205,6 +226,53 @@ func getMaxTxnRow(values url.Values, maxTxnRow *int) error {
 	return nil
 }
 
+func getMaxMultiUpdateRowCount(values url.Values, maxMultiUpdateRow *int) error {
+	s := values.Get("max-multi-update-row")
+	if len(s) == 0 {
+		return nil
+	}
+
+	c, err := strconv.Atoi(s)
+	if err != nil {
+		return cerror.WrapError(cerror.ErrMySQLInvalidConfig, err)
+	}
+	if c <= 0 {
+		return cerror.WrapError(cerror.ErrMySQLInvalidConfig,
+			fmt.Errorf("invalid max-multi-update-row %d, which must be greater than 0", c))
+	}
+	if c > maxMaxMultiUpdateRowCount {
+		log.Warn("max-multi-update-row too large",
+			zap.Int("original", c), zap.Int("override", maxMaxMultiUpdateRowCount))
+		c = maxMaxMultiUpdateRowCount
+	}
+	*maxMultiUpdateRow = c
+	return nil
+}
+
+func getMaxMultiUpdateRowSize(values url.Values, maxMultiUpdateRowSize *int) error {
+	s := values.Get("max-multi-update-row-size")
+	if len(s) == 0 {
+		return nil
+	}
+
+	c, err := strconv.Atoi(s)
+	if err != nil {
+		return cerror.WrapError(cerror.ErrMySQLInvalidConfig, err)
+	}
+	if c < 0 {
+		return cerror.WrapError(cerror.ErrMySQLInvalidConfig,
+			fmt.Errorf("invalid max-multi-update-row-size %d, "+
+				"which must be greater than or equal to 0", c))
+	}
+	if c > maxMaxMultiUpdateRowSize {
+		log.Warn("max-multi-update-row-size too large",
+			zap.Int("original", c), zap.Int("override", maxMaxMultiUpdateRowSize))
+		c = maxMaxMultiUpdateRowSize
+	}
+	*maxMultiUpdateRowSize = c
+	return nil
+}
+
 func getTiDBTxnMode(values url.Values, mode *string) error {
 	s := values.Get("tidb-txn-mode")
 	if len(s) == 0 {

pkg/sink/mysql/config_test.go

@@ -181,13 +181,16 @@ func TestApplySinkURIParamsToConfig(t *testing.T) {
 	expected := NewConfig()
 	expected.WorkerCount = 64
 	expected.MaxTxnRow = 20
+	expected.MaxMultiUpdateRowCount = 80
+	expected.MaxMultiUpdateRowSize = 512
 	expected.BatchReplaceEnabled = true
 	expected.BatchReplaceSize = 50
 	expected.SafeMode = false
 	expected.Timezone = `"UTC"`
 	expected.tidbTxnMode = "pessimistic"
 	expected.EnableOldValue = true
 	uriStr := "mysql://127.0.0.1:3306/?worker-count=64&max-txn-row=20" +
+		"&max-multi-update-row=80&max-multi-update-row-size=512" +
 		"&batch-replace-enable=true&batch-replace-size=50&safe-mode=false" +
 		"&tidb-txn-mode=pessimistic"
 	uri, err := url.Parse(uriStr)
@@ -240,6 +243,16 @@ func TestParseSinkURIOverride(t *testing.T) {
 		checker: func(sp *Config) {
 			require.EqualValues(t, sp.MaxTxnRow, maxMaxTxnRow)
 		},
+	}, {
+		uri: "mysql://127.0.0.1:3306/?max-multi-update-row=2147483648", // int32 max
+		checker: func(sp *Config) {
+			require.EqualValues(t, sp.MaxMultiUpdateRowCount, maxMaxMultiUpdateRowCount)
+		},
+	}, {
+		uri: "mysql://127.0.0.1:3306/?max-multi-update-row-size=2147483648", // int32 max
+		checker: func(sp *Config) {
+			require.EqualValues(t, sp.MaxMultiUpdateRowSize, maxMaxMultiUpdateRowSize)
+		},
 	}, {
 		uri: "mysql://127.0.0.1:3306/?tidb-txn-mode=badmode",
 		checker: func(sp *Config) {