From 832c035eb1e3a273b4ac07b0a2d8243cc9232caf Mon Sep 17 00:00:00 2001 From: nhsmw Date: Wed, 4 Nov 2026 16:16:54 +0800 Subject: [PATCH 1/2] This is an automated cherry-pick of #6748 Signed-off-by: ti-chi-bot --- pkg/sink/mysql/mysql_writer_dml_batch.go | 651 +++++++++++++++++++++++ pkg/sink/mysql/mysql_writer_dml_test.go | 80 +++ 2 files changed, 731 insertions(+) create mode 100644 pkg/sink/mysql/mysql_writer_dml_batch.go diff --git a/pkg/sink/mysql/mysql_writer_dml_batch.go b/pkg/sink/mysql/mysql_writer_dml_batch.go new file mode 100644 index 0000000000..f2c3a4b747 --- /dev/null +++ b/pkg/sink/mysql/mysql_writer_dml_batch.go @@ -0,0 +1,651 @@ +// Copyright 2024 PingCAP, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "github.com/pingcap/log" + "github.com/pingcap/ticdc/pkg/common" + commonEvent "github.com/pingcap/ticdc/pkg/common/event" + cerror "github.com/pingcap/ticdc/pkg/errors" + "github.com/pingcap/ticdc/pkg/sink/sqlmodel" + "github.com/pingcap/ticdc/pkg/util" + "github.com/pingcap/tidb/pkg/util/chunk" + "go.uber.org/zap" +) + +// mysql_writer DML flow exposes three batching levels: +// 1. Normal SQL: one row → one statement, no batching (`generateNormalSQL`/`generateNormalSQLs`). +// 2. Per-event batch: rows inside a single DMLEvent are grouped before execution +// (`generateSQLForSingleEvent`/`generateBatchSQLsPerEvent`). +// 3. Cross-event batch: multiple events are merged first, then emitted as a limited set of SQLs +// (`generateBatchSQL` plus the `buildRowChanges*` helpers). +// Sections below are separated with ===== comments to highlight these layers. + +// rowChangeWithKeys keeps a row change with its row/previous-row keys to assist de-duplication. +type rowChangeWithKeys struct { + rowChange *commonEvent.RowChange + rowKeys []byte + preRowKeys []byte + commitTs uint64 +} + +// ===== Normal SQL (one row -> one statement) ===== + +// generateNormalSQLs simply iterates each event and produces SQLs without batching. +func (w *Writer) generateNormalSQLs(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { + var ( + queries []string + args [][]interface{} + rowTypes []common.RowType + ) + + for _, event := range events { + if event.Len() == 0 { + continue + } + + queryList, argsList, rowTypesList := w.generateNormalSQL(event) + queries = append(queries, queryList...) + args = append(args, argsList...) + rowTypes = append(rowTypes, rowTypesList...) + } + return queries, args, rowTypes +} + +// generateNormalSQL converts a single DMLEvent into SQL statements, respecting safe mode. +func (w *Writer) generateNormalSQL(event *commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { + inSafeMode := w.cfg.SafeMode || w.isInErrorCausedSafeMode || event.CommitTs < event.ReplicatingTs + + log.Debug("inSafeMode", + zap.Bool("inSafeMode", inSafeMode), + zap.Uint64("firstRowCommitTs", event.CommitTs), + zap.Uint64("firstRowReplicatingTs", event.ReplicatingTs), + zap.Bool("cfgSafeMode", w.cfg.SafeMode), + zap.Bool("isInErrorCausedSafeMode", w.isInErrorCausedSafeMode), + zap.Int("writerID", w.id), + ) + + var ( + queries []string + argsList [][]interface{} + rowTypesList []common.RowType + ) + for { + row, ok := event.GetNextRow() + if !ok { + event.Rewind() + break + } + var ( + query string + args []interface{} + rowType common.RowType + ) + switch row.RowType { + case common.RowTypeUpdate: + if inSafeMode { + query, args = buildDelete(event.TableInfo, row) + if query != "" { + queries = append(queries, query) + argsList = append(argsList, args) + rowTypesList = append(rowTypesList, common.RowTypeDelete) + } + query, args = buildInsert(event.TableInfo, row, inSafeMode) + rowType = common.RowTypeInsert + } else { + query, args = buildUpdate(event.TableInfo, row) + rowType = common.RowTypeUpdate + } + case common.RowTypeDelete: + query, args = buildDelete(event.TableInfo, row) + rowType = common.RowTypeDelete + case common.RowTypeInsert: + query, args = buildInsert(event.TableInfo, row, inSafeMode) + rowType = common.RowTypeInsert + } + + if query != "" { + queries = append(queries, query) + argsList = append(argsList, args) + rowTypesList = append(rowTypesList, rowType) + } + } + return queries, argsList, rowTypesList +} + +// ===== Per-event batch (single DMLEvent) ===== + +// generateSQLForSingleEvent batches all row changes of one DMLEvent using the given safe-mode flag. +func (w *Writer) generateSQLForSingleEvent(event *commonEvent.DMLEvent, inDataSafeMode bool) ([]string, [][]interface{}, []common.RowType) { + tableInfo := event.TableInfo + rowLists := make([]*commonEvent.RowChange, 0, event.Len()) + for { + row, ok := event.GetNextRow() + if !ok { + event.Rewind() + break + } + rowLists = append(rowLists, &row) + } + return w.batchSingleTxnDmls(rowLists, tableInfo, inDataSafeMode) +} + +// generateBatchSQLsPerEvent falls back to per-event batching when cross-event merging is not possible. +func (w *Writer) generateBatchSQLsPerEvent(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { + var ( + queries []string + args [][]interface{} + rowTypesList []common.RowType + ) + for _, event := range events { + if event.Len() == 0 { + continue + } + inSafeMode := w.cfg.SafeMode || w.isInErrorCausedSafeMode || event.CommitTs < event.ReplicatingTs + sqls, vals, rowTypes := w.generateSQLForSingleEvent(event, inSafeMode) + queries = append(queries, sqls...) + args = append(args, vals...) + rowTypesList = append(rowTypesList, rowTypes...) + } + return queries, args, rowTypesList +} + +// ========= Cross-event batch: multiple DMLEvents ======== + +// buildRowChangesForUnSafeBatch merges row changes within the same batch following the +// unsafe-mode algorithm that splits delete/update/insert combinations. +// It returns the merged rows which can be used to generate final SQLs. +// for generate batch sql for multi events, we first need to compare the rows with the same pk, to generate the final rows. +// because for the batch sqls, we will first execute delete sqls, then update sqls, and finally insert sqls. +// Here we mainly divide it into 2 cases: +// 1. if all the events are in unsafe mode, we need to split update into delete and insert. So we first split each update row into a delete and a insert one. +// Then compare all delete and insert rows, to delete useless rows. +// if the previous row is Insert A, and the next row is Delete A -- Romove the `Insert A` one. +// +// 2. if all the events are in safe mode: +// Consider we will split the event if PK is changed, so the Update will not change the PK +// for the rows comparation, there are six situations: +// 1. the previous row is Delete A, the next row is Insert A. --- we don't need to combine the rows. +// 2. the previous row is Delete A, the next row is Update xx where A . --- we don't need to combine the rows. +// 3. the previous row is Insert A, the next row is Delete A. --- remove the row of `Insert A` +// 4. the previous row is Insert A, the next row is Update xx where A -- remove the row of `Insert A`, change the row `Update A` to `Insert A` +// 5. the previous row is Update xx where A, the next row is Delete A. --- remove the row `Update xx where A` +// 6. the previous row is Update xx where A, the next row is Update xx where A. --- we need to remove the row, and change the second Update's preRows = first Update's preRows +// +// For these all changes to row, we will continue to compare from the beginnning to the end, until there is no change. +// Then we can generate the final sql of delete/update/insert. +// +// generateBatchSQL decides which batch algorithm (safe vs unsafe) should be used and +// returns the SQLs/args pairs accordingly. +// Considering the batch algorithm in safe mode is O(n^3), which n is the number of rows. +// So we need to limit the number of rows in one batch to avoid performance issues. +func (w *Writer) generateBatchSQL(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { + if len(events) == 0 { + return []string{}, [][]interface{}{}, []common.RowType{} + } + + inSafeMode := w.cfg.SafeMode || w.isInErrorCausedSafeMode || events[0].CommitTs < events[0].ReplicatingTs + + if len(events) == 1 { + // only one event, we don't need to do batch + return w.generateSQLForSingleEvent(events[0], inSafeMode) + } + + if inSafeMode { + // Insert will translate to Replace + return w.generateBatchSQLInSafeMode(events) + } + + return w.generateBatchSQLInUnSafeMode(events) +} + +func (w *Writer) buildRowChangesForUnSafeBatch( + events []*commonEvent.DMLEvent, + tableInfo *common.TableInfo, +) ([]*commonEvent.RowChange, []uint64, error) { + // Step 1 extract all rows in these events to rowLists, and calcuate row key for each row(based on pk value) + rowLists := make([]rowChangeWithKeys, 0) + for _, event := range events { + for { + row, ok := event.GetNextRow() + if !ok { + event.Rewind() + break + } + rowCopy := row + rowChangeWithKeys := rowChangeWithKeys{ + rowChange: &rowCopy, + commitTs: event.CommitTs, + } + if !row.Row.IsEmpty() { + _, keys := genKeyAndHash(&row.Row, tableInfo) + rowChangeWithKeys.rowKeys = keys + } + if !row.PreRow.IsEmpty() { + _, keys := genKeyAndHash(&row.PreRow, tableInfo) + rowChangeWithKeys.preRowKeys = keys + } + rowLists = append(rowLists, rowChangeWithKeys) + } + } + + // Step 2 combine the rows until there is no change + // Consider we will split the event if PK is changed, so the Update will not change the PK + // for the rows comparation, there are six situations: + // 1. the previous row is Delete A, the next row is Insert A. --- we don't need to combine the rows. + // 2. the previous row is Delete A, the next row is Update xx where A . --- we don't need to combine the rows. + // 3. the previous row is Insert A, the next row is Delete A. --- remove the row of `Insert A` + // 4. the previous row is Insert A, the next row is Update xx where A -- remove the row of `Insert A`, change the row `Update A` to `Insert A` + // 5. the previous row is Update xx where A, the next row is Delete A. --- remove the row `Update xx where A` + // 6. the previous row is Update xx where A, the next row is Update xx where A. --- we need to remove the row, and change the second Update's preRows = first Update's preRows + for { + // hasUpdate to determine whether we can break the combine logic + hasUpdate := false + // flagList used to store the exists or not for this row. True means exists. + flagList := make([]bool, len(rowLists)) + for i := range flagList { + flagList[i] = true + } + for i := 0; i < len(rowLists); i++ { + if !flagList[i] { + continue + } + innerLoop: + for j := i + 1; j < len(rowLists); j++ { + if !flagList[j] { + continue + } + rowType := rowLists[i].rowChange.RowType + nextRowType := rowLists[j].rowChange.RowType + switch rowType { + case common.RowTypeInsert: + rowKey := rowLists[i].rowKeys + if nextRowType == common.RowTypeInsert { + if compareKeys(rowKey, rowLists[j].rowKeys) { + return nil, nil, cerror.ErrUnexpected.FastGenByArgs("duplicate insert rows with same key") + } + } else if nextRowType == common.RowTypeDelete { + if compareKeys(rowKey, rowLists[j].preRowKeys) { + flagList[i] = false + hasUpdate = true + break innerLoop + } + } else if nextRowType == common.RowTypeUpdate { + if !compareKeys(rowLists[j].preRowKeys, rowLists[j].rowKeys) { + return nil, nil, cerror.ErrUnexpected.FastGenByArgs("update row key mismatch") + } + if compareKeys(rowKey, rowLists[j].preRowKeys) { + flagList[i] = false + preRowChange := rowLists[j].rowChange + newRowChange := commonEvent.RowChange{ + Row: preRowChange.Row, + RowType: common.RowTypeInsert, + } + rowLists[j] = rowChangeWithKeys{ + rowChange: &newRowChange, + rowKeys: rowLists[j].rowKeys, + commitTs: rowLists[j].commitTs, + } + hasUpdate = true + break innerLoop + } + } + case common.RowTypeUpdate: + rowKey := rowLists[i].rowKeys + if !compareKeys(rowKey, rowLists[i].preRowKeys) { + return nil, nil, cerror.ErrUnexpected.FastGenByArgs("update row key mismatch") + } + if nextRowType == common.RowTypeInsert { + if compareKeys(rowKey, rowLists[j].rowKeys) { + return nil, nil, cerror.ErrUnexpected.FastGenByArgs("duplicate rows for update and insert") + } + } else if nextRowType == common.RowTypeDelete { + if compareKeys(rowKey, rowLists[j].preRowKeys) { + flagList[j] = false + preRowChange := rowLists[i].rowChange + newRowChange := commonEvent.RowChange{ + PreRow: preRowChange.PreRow, + RowType: common.RowTypeDelete, + } + rowLists[i] = rowChangeWithKeys{ + rowChange: &newRowChange, + preRowKeys: rowKey, + commitTs: rowLists[j].commitTs, // use the delete's commitTs to make LWW correctly + } + hasUpdate = true + break innerLoop + } + } else if nextRowType == common.RowTypeUpdate { + if compareKeys(rowKey, rowLists[j].preRowKeys) { + if !compareKeys(rowLists[j].preRowKeys, rowLists[j].rowKeys) { + return nil, nil, cerror.ErrUnexpected.FastGenByArgs("update row key mismatch") + } + newRowChange := commonEvent.RowChange{ + PreRow: rowLists[j].rowChange.PreRow, + Row: rowLists[j].rowChange.Row, + RowType: common.RowTypeUpdate, + } + rowLists[j] = rowChangeWithKeys{ + rowChange: &newRowChange, + preRowKeys: rowKey, + rowKeys: rowKey, + commitTs: rowLists[j].commitTs, + } + flagList[i] = false + hasUpdate = true + break innerLoop + } + } + } + } + } + + if !hasUpdate { + // means no more changes for the rows, break and generate sqls. + break + } + newRowLists := make([]rowChangeWithKeys, 0, len(rowLists)) + for i := 0; i < len(rowLists); i++ { + if flagList[i] { + newRowLists = append(newRowLists, rowLists[i]) + } + } + rowLists = newRowLists + } + + finalRowLists := make([]*commonEvent.RowChange, 0, len(rowLists)) + finalCommitTs := make([]uint64, 0, len(rowLists)) + for i := 0; i < len(rowLists); i++ { + finalRowLists = append(finalRowLists, rowLists[i].rowChange) + finalCommitTs = append(finalCommitTs, rowLists[i].commitTs) + } + return finalRowLists, finalCommitTs, nil +} + +// generateBatchSQLInUnSafeMode merges rows with the same keys and produces batch SQLs +// following the non-safe path (INSERT/UPDATE/DELETE without REPLACE semantics). +func (w *Writer) generateBatchSQLInUnSafeMode(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { + tableInfo := events[0].TableInfo + finalRowLists, _, err := w.buildRowChangesForUnSafeBatch(events, tableInfo) + if err != nil { + sql, values, rowTypes := w.generateBatchSQLsPerEvent(events) + log.Info("normal sql should be", zap.Any("sql", sql), zap.String("values", util.RedactAny(values)), zap.Any("rowTypes", rowTypes), zap.Int("writerID", w.id)) + log.Panic("invalid rows when generating batch SQL in unsafe mode", + zap.Error(err), zap.Any("events", events), zap.Int("writerID", w.id)) + } + return w.batchSingleTxnDmls(finalRowLists, tableInfo, false) +} + +// generateBatchSQLInSafeMode rewrites rows into delete/insert pairs to ensure REPLACE semantics. +func (w *Writer) generateBatchSQLInSafeMode(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { + tableInfo := events[0].TableInfo + + // step 1. divide update row to delete row and insert row, and set into map based on the key hash + rowsMap := make(map[uint64][]*commonEvent.RowChange) + hashToKeyMap := make(map[uint64][]byte) + + addRowToMap := func(row *commonEvent.RowChange, rowData *chunk.Row, event *commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType, bool) { + hashValue, keyValue := genKeyAndHash(rowData, tableInfo) + if _, ok := hashToKeyMap[hashValue]; !ok { + hashToKeyMap[hashValue] = keyValue + } else { + if !compareKeys(hashToKeyMap[hashValue], keyValue) { + log.Warn("the key hash is equal, but the keys is not the same; so we don't use batch generate sql, but use the normal generated sql instead") + event.Rewind() // reset event + // fallback to per-event batch sql + sql, args, rowTypes := w.generateBatchSQLsPerEvent(events) + return sql, args, rowTypes, false + } + } + rowsMap[hashValue] = append(rowsMap[hashValue], row) + return nil, nil, nil, true + } + + for _, event := range events { + for { + row, ok := event.GetNextRow() + if !ok { + event.Rewind() + break + } + switch row.RowType { + case common.RowTypeUpdate: + { + deleteRow := commonEvent.RowChange{RowType: common.RowTypeDelete, PreRow: row.PreRow} + sql, args, rowTypes, ok := addRowToMap(&deleteRow, &row.PreRow, event) + if !ok { + return sql, args, rowTypes + } + } + + { + insertRow := commonEvent.RowChange{RowType: common.RowTypeInsert, Row: row.Row} + sql, args, rowTypes, ok := addRowToMap(&insertRow, &row.Row, event) + if !ok { + return sql, args, rowTypes + } + } + case common.RowTypeDelete: + sql, args, rowTypes, ok := addRowToMap(&row, &row.PreRow, event) + if !ok { + return sql, args, rowTypes + } + case common.RowTypeInsert: + sql, args, rowTypes, ok := addRowToMap(&row, &row.Row, event) + if !ok { + return sql, args, rowTypes + } + } + } + } + + // step 2. compare the rows in the same key hash, to generate the final rows + rowsList := make([]*commonEvent.RowChange, 0, len(rowsMap)) + for _, rowChanges := range rowsMap { + if len(rowChanges) == 0 { + continue + } + if len(rowChanges) == 1 { + rowsList = append(rowsList, rowChanges[0]) + continue + } + // should only happen the rows like 'insert / delete / insert / delete ...' or 'delete / insert /delete ...' , + // should not happen 'insert / insert' or 'delete / delete' + // so only the last one can be the final row changes + prevType := rowChanges[0].RowType + for i := 1; i < len(rowChanges); i++ { + rowType := rowChanges[i].RowType + if rowType == prevType { + sql, values, rowTypes := w.generateBatchSQLsPerEvent(events) + log.Info("normal sql should be", zap.Any("sql", sql), zap.String("values", util.RedactAny(values)), zap.Any("rowTypes", rowTypes), zap.Int("writerID", w.id)) + log.Panic("invalid row changes", zap.String("schemaName", tableInfo.GetSchemaName()), zap.Any("PKIndex", tableInfo.GetPKIndex()), + zap.String("tableName", tableInfo.GetTableName()), zap.Any("rowChanges", rowChanges), + zap.Any("prevType", prevType), zap.Any("currentType", rowType), zap.Int("writerID", w.id)) + } + prevType = rowType + } + rowsList = append(rowsList, rowChanges[len(rowChanges)-1]) + } + // step 3. generate sqls based on rowsList + return w.batchSingleTxnDmls(rowsList, tableInfo, true) +} + +// batchSingleTxnDmls groups row changes into delete/update/insert SQLs for a single transaction. +// inSafeMode means we should use replace sql instead of insert sql to make sure there will not +// be duplicate entry error. +func (w *Writer) batchSingleTxnDmls( + rows []*commonEvent.RowChange, + tableInfo *common.TableInfo, + inSafeMode bool, +) (sqls []string, values [][]interface{}, rowTypes []common.RowType) { + insertRows, updateRows, deleteRows := w.groupRowsByType(rows, tableInfo) + + // handle delete + if len(deleteRows) > 0 { + for _, rows := range deleteRows { + sql, value := sqlmodel.GenDeleteSQL(w.cfg.whereClause, rows...) + sqls = append(sqls, sql) + values = append(values, value) + rowTypes = append(rowTypes, common.RowTypeDelete) + } + } + + // handle update + if len(updateRows) > 0 { + if w.cfg.IsTiDB { + for _, rows := range updateRows { + s, v, rowType := w.genUpdateSQL(rows...) + sqls = append(sqls, s...) + values = append(values, v...) + rowTypes = append(rowTypes, rowType...) + } + // The behavior of update statement differs between TiDB and MySQL. + // So we don't use batch update statement when downstream is MySQL. + // Ref:https://docs.pingcap.com/tidb/stable/sql-statement-update#mysql-compatibility + } else { + for _, rows := range updateRows { + for _, row := range rows { + sql, value := row.GenSQL(sqlmodel.DMLUpdate) + sqls = append(sqls, sql) + values = append(values, value) + rowTypes = append(rowTypes, common.RowTypeUpdate) + } + } + } + } + + // handle insert + if len(insertRows) > 0 { + for _, rows := range insertRows { + if inSafeMode { + sql, value := sqlmodel.GenInsertSQL(sqlmodel.DMLReplace, rows...) + sqls = append(sqls, sql) + values = append(values, value) + } else { + sql, value := sqlmodel.GenInsertSQL(sqlmodel.DMLInsert, rows...) + sqls = append(sqls, sql) + values = append(values, value) + } + rowTypes = append(rowTypes, common.RowTypeInsert) + } + } + + return +} + +// groupRowsByType categorizes row changes by operation type and splits them by configured limits. +func (w *Writer) groupRowsByType( + rows []*commonEvent.RowChange, + tableInfo *common.TableInfo, +) (insertRows, updateRows, deleteRows [][]*sqlmodel.RowChange) { + rowSize := len(rows) + if rowSize > w.cfg.MaxTxnRow { + rowSize = w.cfg.MaxTxnRow + } + + insertRow := make([]*sqlmodel.RowChange, 0, rowSize) + updateRow := make([]*sqlmodel.RowChange, 0, rowSize) + deleteRow := make([]*sqlmodel.RowChange, 0, rowSize) + + eventTableInfo := tableInfo + for _, row := range rows { + switch row.RowType { + case common.RowTypeInsert: + args := getArgsWithGeneratedColumn(&row.Row, tableInfo) + newInsertRow := sqlmodel.NewRowChange( + &tableInfo.TableName, + nil, + nil, + args, + eventTableInfo, + nil, nil) + + insertRow = append(insertRow, newInsertRow) + if len(insertRow) >= w.cfg.MaxTxnRow { + insertRows = append(insertRows, insertRow) + insertRow = make([]*sqlmodel.RowChange, 0, rowSize) + } + case common.RowTypeUpdate: + args := getArgsWithGeneratedColumn(&row.Row, tableInfo) + preArgs := getArgsWithGeneratedColumn(&row.PreRow, tableInfo) + newUpdateRow := sqlmodel.NewRowChange( + &tableInfo.TableName, + nil, + preArgs, + args, + eventTableInfo, + nil, nil) + // Include both row images in the size estimate used to limit batch updates. + var approximateSize int64 + for i := 0; i < row.Row.Len(); i++ { + approximateSize += int64(len(row.PreRow.GetRaw(i)) + len(row.Row.GetRaw(i))) + } + newUpdateRow.SetApproximateDataSize(approximateSize) + updateRow = append(updateRow, newUpdateRow) + if len(updateRow) >= w.cfg.MaxMultiUpdateRowCount { + updateRows = append(updateRows, updateRow) + updateRow = make([]*sqlmodel.RowChange, 0, rowSize) + } + case common.RowTypeDelete: + preArgs := getArgsWithGeneratedColumn(&row.PreRow, tableInfo) + newDeleteRow := sqlmodel.NewRowChange( + &tableInfo.TableName, + nil, + preArgs, + nil, + eventTableInfo, + nil, nil) + deleteRow = append(deleteRow, newDeleteRow) + if len(deleteRow) >= w.cfg.MaxTxnRow { + deleteRows = append(deleteRows, deleteRow) + deleteRow = make([]*sqlmodel.RowChange, 0, rowSize) + } + } + } + if len(insertRow) > 0 { + insertRows = append(insertRows, insertRow) + } + if len(updateRow) > 0 { + updateRows = append(updateRows, updateRow) + } + if len(deleteRow) > 0 { + deleteRows = append(deleteRows, deleteRow) + } + + return +} + +// genUpdateSQL creates batched UPDATE statements when the payload size permits. +func (w *Writer) genUpdateSQL(rows ...*sqlmodel.RowChange) ([]string, [][]interface{}, []common.RowType) { + size := 0 + for _, r := range rows { + size += int(r.GetApproximateDataSize()) + } + if size < w.cfg.MaxMultiUpdateRowSize*len(rows) { + // use multi update in one SQL + sql, value := sqlmodel.GenUpdateSQL(w.cfg.whereClause, rows...) + return []string{sql}, [][]interface{}{value}, []common.RowType{common.RowTypeUpdate} + } + // each row has one independent update SQL. + sqls := make([]string, 0, len(rows)) + values := make([][]interface{}, 0, len(rows)) + rowTypes := make([]common.RowType, 0, len(rows)) + for _, row := range rows { + sql, value := row.GenSQL(sqlmodel.DMLUpdate) + sqls = append(sqls, sql) + values = append(values, value) + rowTypes = append(rowTypes, common.RowTypeUpdate) + } + return sqls, values, rowTypes +} diff --git a/pkg/sink/mysql/mysql_writer_dml_test.go b/pkg/sink/mysql/mysql_writer_dml_test.go index 0a729bde95..b9b8b84cdd 100644 --- a/pkg/sink/mysql/mysql_writer_dml_test.go +++ b/pkg/sink/mysql/mysql_writer_dml_test.go @@ -1005,6 +1005,86 @@ func TestGroupEventsByTable(t *testing.T) { }) } +func TestPrepareDMLsMaxMultiUpdateRowSize(t *testing.T) { + helper := commonEvent.NewEventTestHelper(t) + defer helper.Close() + helper.Tk().MustExec("use test") + job := helper.DDL2Job("CREATE TABLE t (id INT PRIMARY KEY, payload TEXT)") + tableInfo := helper.GetTableInfo(job) + + testCases := []struct { + name string + prePayloadSize int + payloadSize int + maxMultiUpdateRowSize int + isTiDB bool + expectedStatements int + }{ + {"large rows default threshold", 16384, 16384, 1024, true, 2}, + {"large rows maximum threshold", 16384, 16384, 8192, true, 2}, + {"small rows", 16, 16, 1024, true, 1}, + {"batch updates disabled", 16, 16, 0, true, 2}, + // Each row image also contains an eight-byte primary key. + {"below threshold", 503, 503, 1024, true, 1}, + {"at threshold", 504, 504, 1024, true, 2}, + {"large previous row", 16384, 16, 1024, true, 2}, + {"large current row", 16, 16384, 1024, true, 2}, + {"mysql downstream", 16, 16, 1024, false, 2}, + } + for _, rowsPerEvent := range []int{1, 2} { + for _, tc := range testCases { + t.Run(fmt.Sprintf("rowsPerEvent=%d/%s", rowsPerEvent, tc.name), func(t *testing.T) { + writer, db, _ := newTestMysqlWriter(t) + defer db.Close() + writer.cfg.BatchDMLEnable = true + writer.cfg.SafeMode = false + writer.cfg.IsTiDB = tc.isTiDB + writer.cfg.MaxMultiUpdateRowCount = 40 + writer.cfg.MaxMultiUpdateRowSize = tc.maxMultiUpdateRowSize + + prePayload := strings.Repeat("a", tc.prePayloadSize) + payload := strings.Repeat("b", tc.payloadSize) + dispatcherID := common.NewDispatcherID() + var events []*commonEvent.DMLEvent + for startID := 1; startID <= 2; startID += rowsPerEvent { + commitTs := uint64(100 + startID) + event := commonEvent.NewDMLEvent(dispatcherID, tableInfo.TableName.TableID, commitTs-1, commitTs, tableInfo) + event.ReplicatingTs = 1 + event.Length = int32(rowsPerEvent) + event.Rows = chunk.NewChunkWithCapacity(tableInfo.GetFieldSlice(), 2*rowsPerEvent) + for id := startID; id < startID+rowsPerEvent; id++ { + event.RowTypes = append(event.RowTypes, common.RowTypeUpdate, common.RowTypeUpdate) + event.Rows.AppendInt64(0, int64(id)) + event.Rows.AppendString(1, prePayload) + event.Rows.AppendInt64(0, int64(id)) + event.Rows.AppendString(1, payload) + } + events = append(events, event) + } + + require.True(t, writer.shouldGenBatchSQL(tableInfo, events)) + dmls, err := writer.prepareDMLs(events) + require.NoError(t, err) + defer dmlsPool.Put(dmls) + require.Len(t, dmls.sqls, tc.expectedStatements) + require.Len(t, dmls.values, tc.expectedStatements) + require.Len(t, dmls.rowTypes, tc.expectedStatements) + for i, sql := range dmls.sqls { + require.Equal(t, common.RowTypeUpdate, dmls.rowTypes[i]) + if tc.expectedStatements == 1 { + require.Contains(t, sql, "CASE WHEN") + require.Equal(t, []any{int64(1), int64(1), int64(2), int64(2), int64(1), payload, int64(2), payload, int64(1), int64(2)}, dmls.values[i]) + } else { + require.Equal(t, "UPDATE `test`.`t` SET `id` = ?, `payload` = ? WHERE `id` = ? LIMIT 1", sql) + id := int64(i + 1) + require.Equal(t, []any{id, payload, id}, dmls.values[i]) + } + } + }) + } + } +} + func TestPrepareDMLsWithNotNullUniqueKey(t *testing.T) { writer, _, _ := newTestMysqlWriter(t) defer writer.db.Close() From 7c768df6f8db1cacb8b52dbbf687bb6eacf48269 Mon Sep 17 00:00:00 2001 From: wk989898 Date: Thu, 5 Nov 2026 08:23:58 +0000 Subject: [PATCH 2/2] mysql: adapt update row size fix to release-8.5 --- pkg/sink/mysql/mysql_writer_dml.go | 6 + pkg/sink/mysql/mysql_writer_dml_batch.go | 651 ----------------------- 2 files changed, 6 insertions(+), 651 deletions(-) delete mode 100644 pkg/sink/mysql/mysql_writer_dml_batch.go diff --git a/pkg/sink/mysql/mysql_writer_dml.go b/pkg/sink/mysql/mysql_writer_dml.go index 387bdf7f86..4ecf16bb07 100644 --- a/pkg/sink/mysql/mysql_writer_dml.go +++ b/pkg/sink/mysql/mysql_writer_dml.go @@ -989,6 +989,12 @@ func (w *Writer) groupRowsByType( args, eventTableInfo, nil, nil) + // Include both row images in the size estimate used to limit batch updates. + var approximateSize int64 + for i := 0; i < row.Row.Len(); i++ { + approximateSize += int64(len(row.PreRow.GetRaw(i)) + len(row.Row.GetRaw(i))) + } + newUpdateRow.SetApproximateDataSize(approximateSize) updateRow = append(updateRow, newUpdateRow) if len(updateRow) >= w.cfg.MaxMultiUpdateRowCount { updateRows = append(updateRows, updateRow) diff --git a/pkg/sink/mysql/mysql_writer_dml_batch.go b/pkg/sink/mysql/mysql_writer_dml_batch.go deleted file mode 100644 index f2c3a4b747..0000000000 --- a/pkg/sink/mysql/mysql_writer_dml_batch.go +++ /dev/null @@ -1,651 +0,0 @@ -// Copyright 2024 PingCAP, Inc. -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// See the License for the specific language governing permissions and -// limitations under the License. - -package mysql - -import ( - "github.com/pingcap/log" - "github.com/pingcap/ticdc/pkg/common" - commonEvent "github.com/pingcap/ticdc/pkg/common/event" - cerror "github.com/pingcap/ticdc/pkg/errors" - "github.com/pingcap/ticdc/pkg/sink/sqlmodel" - "github.com/pingcap/ticdc/pkg/util" - "github.com/pingcap/tidb/pkg/util/chunk" - "go.uber.org/zap" -) - -// mysql_writer DML flow exposes three batching levels: -// 1. Normal SQL: one row → one statement, no batching (`generateNormalSQL`/`generateNormalSQLs`). -// 2. Per-event batch: rows inside a single DMLEvent are grouped before execution -// (`generateSQLForSingleEvent`/`generateBatchSQLsPerEvent`). -// 3. Cross-event batch: multiple events are merged first, then emitted as a limited set of SQLs -// (`generateBatchSQL` plus the `buildRowChanges*` helpers). -// Sections below are separated with ===== comments to highlight these layers. - -// rowChangeWithKeys keeps a row change with its row/previous-row keys to assist de-duplication. -type rowChangeWithKeys struct { - rowChange *commonEvent.RowChange - rowKeys []byte - preRowKeys []byte - commitTs uint64 -} - -// ===== Normal SQL (one row -> one statement) ===== - -// generateNormalSQLs simply iterates each event and produces SQLs without batching. -func (w *Writer) generateNormalSQLs(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { - var ( - queries []string - args [][]interface{} - rowTypes []common.RowType - ) - - for _, event := range events { - if event.Len() == 0 { - continue - } - - queryList, argsList, rowTypesList := w.generateNormalSQL(event) - queries = append(queries, queryList...) - args = append(args, argsList...) - rowTypes = append(rowTypes, rowTypesList...) - } - return queries, args, rowTypes -} - -// generateNormalSQL converts a single DMLEvent into SQL statements, respecting safe mode. -func (w *Writer) generateNormalSQL(event *commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { - inSafeMode := w.cfg.SafeMode || w.isInErrorCausedSafeMode || event.CommitTs < event.ReplicatingTs - - log.Debug("inSafeMode", - zap.Bool("inSafeMode", inSafeMode), - zap.Uint64("firstRowCommitTs", event.CommitTs), - zap.Uint64("firstRowReplicatingTs", event.ReplicatingTs), - zap.Bool("cfgSafeMode", w.cfg.SafeMode), - zap.Bool("isInErrorCausedSafeMode", w.isInErrorCausedSafeMode), - zap.Int("writerID", w.id), - ) - - var ( - queries []string - argsList [][]interface{} - rowTypesList []common.RowType - ) - for { - row, ok := event.GetNextRow() - if !ok { - event.Rewind() - break - } - var ( - query string - args []interface{} - rowType common.RowType - ) - switch row.RowType { - case common.RowTypeUpdate: - if inSafeMode { - query, args = buildDelete(event.TableInfo, row) - if query != "" { - queries = append(queries, query) - argsList = append(argsList, args) - rowTypesList = append(rowTypesList, common.RowTypeDelete) - } - query, args = buildInsert(event.TableInfo, row, inSafeMode) - rowType = common.RowTypeInsert - } else { - query, args = buildUpdate(event.TableInfo, row) - rowType = common.RowTypeUpdate - } - case common.RowTypeDelete: - query, args = buildDelete(event.TableInfo, row) - rowType = common.RowTypeDelete - case common.RowTypeInsert: - query, args = buildInsert(event.TableInfo, row, inSafeMode) - rowType = common.RowTypeInsert - } - - if query != "" { - queries = append(queries, query) - argsList = append(argsList, args) - rowTypesList = append(rowTypesList, rowType) - } - } - return queries, argsList, rowTypesList -} - -// ===== Per-event batch (single DMLEvent) ===== - -// generateSQLForSingleEvent batches all row changes of one DMLEvent using the given safe-mode flag. -func (w *Writer) generateSQLForSingleEvent(event *commonEvent.DMLEvent, inDataSafeMode bool) ([]string, [][]interface{}, []common.RowType) { - tableInfo := event.TableInfo - rowLists := make([]*commonEvent.RowChange, 0, event.Len()) - for { - row, ok := event.GetNextRow() - if !ok { - event.Rewind() - break - } - rowLists = append(rowLists, &row) - } - return w.batchSingleTxnDmls(rowLists, tableInfo, inDataSafeMode) -} - -// generateBatchSQLsPerEvent falls back to per-event batching when cross-event merging is not possible. -func (w *Writer) generateBatchSQLsPerEvent(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { - var ( - queries []string - args [][]interface{} - rowTypesList []common.RowType - ) - for _, event := range events { - if event.Len() == 0 { - continue - } - inSafeMode := w.cfg.SafeMode || w.isInErrorCausedSafeMode || event.CommitTs < event.ReplicatingTs - sqls, vals, rowTypes := w.generateSQLForSingleEvent(event, inSafeMode) - queries = append(queries, sqls...) - args = append(args, vals...) - rowTypesList = append(rowTypesList, rowTypes...) - } - return queries, args, rowTypesList -} - -// ========= Cross-event batch: multiple DMLEvents ======== - -// buildRowChangesForUnSafeBatch merges row changes within the same batch following the -// unsafe-mode algorithm that splits delete/update/insert combinations. -// It returns the merged rows which can be used to generate final SQLs. -// for generate batch sql for multi events, we first need to compare the rows with the same pk, to generate the final rows. -// because for the batch sqls, we will first execute delete sqls, then update sqls, and finally insert sqls. -// Here we mainly divide it into 2 cases: -// 1. if all the events are in unsafe mode, we need to split update into delete and insert. So we first split each update row into a delete and a insert one. -// Then compare all delete and insert rows, to delete useless rows. -// if the previous row is Insert A, and the next row is Delete A -- Romove the `Insert A` one. -// -// 2. if all the events are in safe mode: -// Consider we will split the event if PK is changed, so the Update will not change the PK -// for the rows comparation, there are six situations: -// 1. the previous row is Delete A, the next row is Insert A. --- we don't need to combine the rows. -// 2. the previous row is Delete A, the next row is Update xx where A . --- we don't need to combine the rows. -// 3. the previous row is Insert A, the next row is Delete A. --- remove the row of `Insert A` -// 4. the previous row is Insert A, the next row is Update xx where A -- remove the row of `Insert A`, change the row `Update A` to `Insert A` -// 5. the previous row is Update xx where A, the next row is Delete A. --- remove the row `Update xx where A` -// 6. the previous row is Update xx where A, the next row is Update xx where A. --- we need to remove the row, and change the second Update's preRows = first Update's preRows -// -// For these all changes to row, we will continue to compare from the beginnning to the end, until there is no change. -// Then we can generate the final sql of delete/update/insert. -// -// generateBatchSQL decides which batch algorithm (safe vs unsafe) should be used and -// returns the SQLs/args pairs accordingly. -// Considering the batch algorithm in safe mode is O(n^3), which n is the number of rows. -// So we need to limit the number of rows in one batch to avoid performance issues. -func (w *Writer) generateBatchSQL(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { - if len(events) == 0 { - return []string{}, [][]interface{}{}, []common.RowType{} - } - - inSafeMode := w.cfg.SafeMode || w.isInErrorCausedSafeMode || events[0].CommitTs < events[0].ReplicatingTs - - if len(events) == 1 { - // only one event, we don't need to do batch - return w.generateSQLForSingleEvent(events[0], inSafeMode) - } - - if inSafeMode { - // Insert will translate to Replace - return w.generateBatchSQLInSafeMode(events) - } - - return w.generateBatchSQLInUnSafeMode(events) -} - -func (w *Writer) buildRowChangesForUnSafeBatch( - events []*commonEvent.DMLEvent, - tableInfo *common.TableInfo, -) ([]*commonEvent.RowChange, []uint64, error) { - // Step 1 extract all rows in these events to rowLists, and calcuate row key for each row(based on pk value) - rowLists := make([]rowChangeWithKeys, 0) - for _, event := range events { - for { - row, ok := event.GetNextRow() - if !ok { - event.Rewind() - break - } - rowCopy := row - rowChangeWithKeys := rowChangeWithKeys{ - rowChange: &rowCopy, - commitTs: event.CommitTs, - } - if !row.Row.IsEmpty() { - _, keys := genKeyAndHash(&row.Row, tableInfo) - rowChangeWithKeys.rowKeys = keys - } - if !row.PreRow.IsEmpty() { - _, keys := genKeyAndHash(&row.PreRow, tableInfo) - rowChangeWithKeys.preRowKeys = keys - } - rowLists = append(rowLists, rowChangeWithKeys) - } - } - - // Step 2 combine the rows until there is no change - // Consider we will split the event if PK is changed, so the Update will not change the PK - // for the rows comparation, there are six situations: - // 1. the previous row is Delete A, the next row is Insert A. --- we don't need to combine the rows. - // 2. the previous row is Delete A, the next row is Update xx where A . --- we don't need to combine the rows. - // 3. the previous row is Insert A, the next row is Delete A. --- remove the row of `Insert A` - // 4. the previous row is Insert A, the next row is Update xx where A -- remove the row of `Insert A`, change the row `Update A` to `Insert A` - // 5. the previous row is Update xx where A, the next row is Delete A. --- remove the row `Update xx where A` - // 6. the previous row is Update xx where A, the next row is Update xx where A. --- we need to remove the row, and change the second Update's preRows = first Update's preRows - for { - // hasUpdate to determine whether we can break the combine logic - hasUpdate := false - // flagList used to store the exists or not for this row. True means exists. - flagList := make([]bool, len(rowLists)) - for i := range flagList { - flagList[i] = true - } - for i := 0; i < len(rowLists); i++ { - if !flagList[i] { - continue - } - innerLoop: - for j := i + 1; j < len(rowLists); j++ { - if !flagList[j] { - continue - } - rowType := rowLists[i].rowChange.RowType - nextRowType := rowLists[j].rowChange.RowType - switch rowType { - case common.RowTypeInsert: - rowKey := rowLists[i].rowKeys - if nextRowType == common.RowTypeInsert { - if compareKeys(rowKey, rowLists[j].rowKeys) { - return nil, nil, cerror.ErrUnexpected.FastGenByArgs("duplicate insert rows with same key") - } - } else if nextRowType == common.RowTypeDelete { - if compareKeys(rowKey, rowLists[j].preRowKeys) { - flagList[i] = false - hasUpdate = true - break innerLoop - } - } else if nextRowType == common.RowTypeUpdate { - if !compareKeys(rowLists[j].preRowKeys, rowLists[j].rowKeys) { - return nil, nil, cerror.ErrUnexpected.FastGenByArgs("update row key mismatch") - } - if compareKeys(rowKey, rowLists[j].preRowKeys) { - flagList[i] = false - preRowChange := rowLists[j].rowChange - newRowChange := commonEvent.RowChange{ - Row: preRowChange.Row, - RowType: common.RowTypeInsert, - } - rowLists[j] = rowChangeWithKeys{ - rowChange: &newRowChange, - rowKeys: rowLists[j].rowKeys, - commitTs: rowLists[j].commitTs, - } - hasUpdate = true - break innerLoop - } - } - case common.RowTypeUpdate: - rowKey := rowLists[i].rowKeys - if !compareKeys(rowKey, rowLists[i].preRowKeys) { - return nil, nil, cerror.ErrUnexpected.FastGenByArgs("update row key mismatch") - } - if nextRowType == common.RowTypeInsert { - if compareKeys(rowKey, rowLists[j].rowKeys) { - return nil, nil, cerror.ErrUnexpected.FastGenByArgs("duplicate rows for update and insert") - } - } else if nextRowType == common.RowTypeDelete { - if compareKeys(rowKey, rowLists[j].preRowKeys) { - flagList[j] = false - preRowChange := rowLists[i].rowChange - newRowChange := commonEvent.RowChange{ - PreRow: preRowChange.PreRow, - RowType: common.RowTypeDelete, - } - rowLists[i] = rowChangeWithKeys{ - rowChange: &newRowChange, - preRowKeys: rowKey, - commitTs: rowLists[j].commitTs, // use the delete's commitTs to make LWW correctly - } - hasUpdate = true - break innerLoop - } - } else if nextRowType == common.RowTypeUpdate { - if compareKeys(rowKey, rowLists[j].preRowKeys) { - if !compareKeys(rowLists[j].preRowKeys, rowLists[j].rowKeys) { - return nil, nil, cerror.ErrUnexpected.FastGenByArgs("update row key mismatch") - } - newRowChange := commonEvent.RowChange{ - PreRow: rowLists[j].rowChange.PreRow, - Row: rowLists[j].rowChange.Row, - RowType: common.RowTypeUpdate, - } - rowLists[j] = rowChangeWithKeys{ - rowChange: &newRowChange, - preRowKeys: rowKey, - rowKeys: rowKey, - commitTs: rowLists[j].commitTs, - } - flagList[i] = false - hasUpdate = true - break innerLoop - } - } - } - } - } - - if !hasUpdate { - // means no more changes for the rows, break and generate sqls. - break - } - newRowLists := make([]rowChangeWithKeys, 0, len(rowLists)) - for i := 0; i < len(rowLists); i++ { - if flagList[i] { - newRowLists = append(newRowLists, rowLists[i]) - } - } - rowLists = newRowLists - } - - finalRowLists := make([]*commonEvent.RowChange, 0, len(rowLists)) - finalCommitTs := make([]uint64, 0, len(rowLists)) - for i := 0; i < len(rowLists); i++ { - finalRowLists = append(finalRowLists, rowLists[i].rowChange) - finalCommitTs = append(finalCommitTs, rowLists[i].commitTs) - } - return finalRowLists, finalCommitTs, nil -} - -// generateBatchSQLInUnSafeMode merges rows with the same keys and produces batch SQLs -// following the non-safe path (INSERT/UPDATE/DELETE without REPLACE semantics). -func (w *Writer) generateBatchSQLInUnSafeMode(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { - tableInfo := events[0].TableInfo - finalRowLists, _, err := w.buildRowChangesForUnSafeBatch(events, tableInfo) - if err != nil { - sql, values, rowTypes := w.generateBatchSQLsPerEvent(events) - log.Info("normal sql should be", zap.Any("sql", sql), zap.String("values", util.RedactAny(values)), zap.Any("rowTypes", rowTypes), zap.Int("writerID", w.id)) - log.Panic("invalid rows when generating batch SQL in unsafe mode", - zap.Error(err), zap.Any("events", events), zap.Int("writerID", w.id)) - } - return w.batchSingleTxnDmls(finalRowLists, tableInfo, false) -} - -// generateBatchSQLInSafeMode rewrites rows into delete/insert pairs to ensure REPLACE semantics. -func (w *Writer) generateBatchSQLInSafeMode(events []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) { - tableInfo := events[0].TableInfo - - // step 1. divide update row to delete row and insert row, and set into map based on the key hash - rowsMap := make(map[uint64][]*commonEvent.RowChange) - hashToKeyMap := make(map[uint64][]byte) - - addRowToMap := func(row *commonEvent.RowChange, rowData *chunk.Row, event *commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType, bool) { - hashValue, keyValue := genKeyAndHash(rowData, tableInfo) - if _, ok := hashToKeyMap[hashValue]; !ok { - hashToKeyMap[hashValue] = keyValue - } else { - if !compareKeys(hashToKeyMap[hashValue], keyValue) { - log.Warn("the key hash is equal, but the keys is not the same; so we don't use batch generate sql, but use the normal generated sql instead") - event.Rewind() // reset event - // fallback to per-event batch sql - sql, args, rowTypes := w.generateBatchSQLsPerEvent(events) - return sql, args, rowTypes, false - } - } - rowsMap[hashValue] = append(rowsMap[hashValue], row) - return nil, nil, nil, true - } - - for _, event := range events { - for { - row, ok := event.GetNextRow() - if !ok { - event.Rewind() - break - } - switch row.RowType { - case common.RowTypeUpdate: - { - deleteRow := commonEvent.RowChange{RowType: common.RowTypeDelete, PreRow: row.PreRow} - sql, args, rowTypes, ok := addRowToMap(&deleteRow, &row.PreRow, event) - if !ok { - return sql, args, rowTypes - } - } - - { - insertRow := commonEvent.RowChange{RowType: common.RowTypeInsert, Row: row.Row} - sql, args, rowTypes, ok := addRowToMap(&insertRow, &row.Row, event) - if !ok { - return sql, args, rowTypes - } - } - case common.RowTypeDelete: - sql, args, rowTypes, ok := addRowToMap(&row, &row.PreRow, event) - if !ok { - return sql, args, rowTypes - } - case common.RowTypeInsert: - sql, args, rowTypes, ok := addRowToMap(&row, &row.Row, event) - if !ok { - return sql, args, rowTypes - } - } - } - } - - // step 2. compare the rows in the same key hash, to generate the final rows - rowsList := make([]*commonEvent.RowChange, 0, len(rowsMap)) - for _, rowChanges := range rowsMap { - if len(rowChanges) == 0 { - continue - } - if len(rowChanges) == 1 { - rowsList = append(rowsList, rowChanges[0]) - continue - } - // should only happen the rows like 'insert / delete / insert / delete ...' or 'delete / insert /delete ...' , - // should not happen 'insert / insert' or 'delete / delete' - // so only the last one can be the final row changes - prevType := rowChanges[0].RowType - for i := 1; i < len(rowChanges); i++ { - rowType := rowChanges[i].RowType - if rowType == prevType { - sql, values, rowTypes := w.generateBatchSQLsPerEvent(events) - log.Info("normal sql should be", zap.Any("sql", sql), zap.String("values", util.RedactAny(values)), zap.Any("rowTypes", rowTypes), zap.Int("writerID", w.id)) - log.Panic("invalid row changes", zap.String("schemaName", tableInfo.GetSchemaName()), zap.Any("PKIndex", tableInfo.GetPKIndex()), - zap.String("tableName", tableInfo.GetTableName()), zap.Any("rowChanges", rowChanges), - zap.Any("prevType", prevType), zap.Any("currentType", rowType), zap.Int("writerID", w.id)) - } - prevType = rowType - } - rowsList = append(rowsList, rowChanges[len(rowChanges)-1]) - } - // step 3. generate sqls based on rowsList - return w.batchSingleTxnDmls(rowsList, tableInfo, true) -} - -// batchSingleTxnDmls groups row changes into delete/update/insert SQLs for a single transaction. -// inSafeMode means we should use replace sql instead of insert sql to make sure there will not -// be duplicate entry error. -func (w *Writer) batchSingleTxnDmls( - rows []*commonEvent.RowChange, - tableInfo *common.TableInfo, - inSafeMode bool, -) (sqls []string, values [][]interface{}, rowTypes []common.RowType) { - insertRows, updateRows, deleteRows := w.groupRowsByType(rows, tableInfo) - - // handle delete - if len(deleteRows) > 0 { - for _, rows := range deleteRows { - sql, value := sqlmodel.GenDeleteSQL(w.cfg.whereClause, rows...) - sqls = append(sqls, sql) - values = append(values, value) - rowTypes = append(rowTypes, common.RowTypeDelete) - } - } - - // handle update - if len(updateRows) > 0 { - if w.cfg.IsTiDB { - for _, rows := range updateRows { - s, v, rowType := w.genUpdateSQL(rows...) - sqls = append(sqls, s...) - values = append(values, v...) - rowTypes = append(rowTypes, rowType...) - } - // The behavior of update statement differs between TiDB and MySQL. - // So we don't use batch update statement when downstream is MySQL. - // Ref:https://docs.pingcap.com/tidb/stable/sql-statement-update#mysql-compatibility - } else { - for _, rows := range updateRows { - for _, row := range rows { - sql, value := row.GenSQL(sqlmodel.DMLUpdate) - sqls = append(sqls, sql) - values = append(values, value) - rowTypes = append(rowTypes, common.RowTypeUpdate) - } - } - } - } - - // handle insert - if len(insertRows) > 0 { - for _, rows := range insertRows { - if inSafeMode { - sql, value := sqlmodel.GenInsertSQL(sqlmodel.DMLReplace, rows...) - sqls = append(sqls, sql) - values = append(values, value) - } else { - sql, value := sqlmodel.GenInsertSQL(sqlmodel.DMLInsert, rows...) - sqls = append(sqls, sql) - values = append(values, value) - } - rowTypes = append(rowTypes, common.RowTypeInsert) - } - } - - return -} - -// groupRowsByType categorizes row changes by operation type and splits them by configured limits. -func (w *Writer) groupRowsByType( - rows []*commonEvent.RowChange, - tableInfo *common.TableInfo, -) (insertRows, updateRows, deleteRows [][]*sqlmodel.RowChange) { - rowSize := len(rows) - if rowSize > w.cfg.MaxTxnRow { - rowSize = w.cfg.MaxTxnRow - } - - insertRow := make([]*sqlmodel.RowChange, 0, rowSize) - updateRow := make([]*sqlmodel.RowChange, 0, rowSize) - deleteRow := make([]*sqlmodel.RowChange, 0, rowSize) - - eventTableInfo := tableInfo - for _, row := range rows { - switch row.RowType { - case common.RowTypeInsert: - args := getArgsWithGeneratedColumn(&row.Row, tableInfo) - newInsertRow := sqlmodel.NewRowChange( - &tableInfo.TableName, - nil, - nil, - args, - eventTableInfo, - nil, nil) - - insertRow = append(insertRow, newInsertRow) - if len(insertRow) >= w.cfg.MaxTxnRow { - insertRows = append(insertRows, insertRow) - insertRow = make([]*sqlmodel.RowChange, 0, rowSize) - } - case common.RowTypeUpdate: - args := getArgsWithGeneratedColumn(&row.Row, tableInfo) - preArgs := getArgsWithGeneratedColumn(&row.PreRow, tableInfo) - newUpdateRow := sqlmodel.NewRowChange( - &tableInfo.TableName, - nil, - preArgs, - args, - eventTableInfo, - nil, nil) - // Include both row images in the size estimate used to limit batch updates. - var approximateSize int64 - for i := 0; i < row.Row.Len(); i++ { - approximateSize += int64(len(row.PreRow.GetRaw(i)) + len(row.Row.GetRaw(i))) - } - newUpdateRow.SetApproximateDataSize(approximateSize) - updateRow = append(updateRow, newUpdateRow) - if len(updateRow) >= w.cfg.MaxMultiUpdateRowCount { - updateRows = append(updateRows, updateRow) - updateRow = make([]*sqlmodel.RowChange, 0, rowSize) - } - case common.RowTypeDelete: - preArgs := getArgsWithGeneratedColumn(&row.PreRow, tableInfo) - newDeleteRow := sqlmodel.NewRowChange( - &tableInfo.TableName, - nil, - preArgs, - nil, - eventTableInfo, - nil, nil) - deleteRow = append(deleteRow, newDeleteRow) - if len(deleteRow) >= w.cfg.MaxTxnRow { - deleteRows = append(deleteRows, deleteRow) - deleteRow = make([]*sqlmodel.RowChange, 0, rowSize) - } - } - } - if len(insertRow) > 0 { - insertRows = append(insertRows, insertRow) - } - if len(updateRow) > 0 { - updateRows = append(updateRows, updateRow) - } - if len(deleteRow) > 0 { - deleteRows = append(deleteRows, deleteRow) - } - - return -} - -// genUpdateSQL creates batched UPDATE statements when the payload size permits. -func (w *Writer) genUpdateSQL(rows ...*sqlmodel.RowChange) ([]string, [][]interface{}, []common.RowType) { - size := 0 - for _, r := range rows { - size += int(r.GetApproximateDataSize()) - } - if size < w.cfg.MaxMultiUpdateRowSize*len(rows) { - // use multi update in one SQL - sql, value := sqlmodel.GenUpdateSQL(w.cfg.whereClause, rows...) - return []string{sql}, [][]interface{}{value}, []common.RowType{common.RowTypeUpdate} - } - // each row has one independent update SQL. - sqls := make([]string, 0, len(rows)) - values := make([][]interface{}, 0, len(rows)) - rowTypes := make([]common.RowType, 0, len(rows)) - for _, row := range rows { - sql, value := row.GenSQL(sqlmodel.DMLUpdate) - sqls = append(sqls, sql) - values = append(values, value) - rowTypes = append(rowTypes, common.RowTypeUpdate) - } - return sqls, values, rowTypes -}