sink: avoid batch DML panic when DELETE events are filtered

pkg/sink/mysql/config.go

@@ -27,6 +27,7 @@ import (
 	lru "github.com/hashicorp/golang-lru"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
+	bf "github.com/pingcap/ticdc/pkg/binlog-filter"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/errors"
@@ -35,6 +36,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/pingcap/tidb/br/pkg/version"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
+	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"go.uber.org/zap"
 )
 
@@ -137,6 +139,9 @@ type Config struct {
 	BatchDMLEnable  bool
 	MultiStmtEnable bool
 	CachePrepStmts  bool
+	// deleteEventFilters select legacy per-event SQL generation for tables whose
+	// deletes can be filtered, invalidating cross-event merging assumptions.
+	deleteEventFilters []tfilter.Filter
 
 	// sync point
 	SyncPointRetention time.Duration
@@ -230,6 +235,50 @@ func (c *Config) mergeConfig(cfg *config.ChangefeedConfig) {
 	}
 }
 
+func buildDeleteEventFilters(cfg *config.ChangefeedConfig) ([]tfilter.Filter, error) {
+	if cfg.Filter == nil {
+		return nil, nil
+	}
+	var filters []tfilter.Filter
+	for _, rule := range cfg.Filter.EventFilters {
+		filterDeleteEvent := false
+		for _, eventType := range rule.IgnoreEvent {
+			switch eventType {
+			case bf.DeleteEvent, bf.AllDML, bf.AllEvent:
+				filterDeleteEvent = true
+			}
+		}
+		filterDeleteExpr := util.GetOrZero(rule.IgnoreDeleteValueExpr) != ""
+		if !filterDeleteEvent && !filterDeleteExpr {
+			continue
+		}
+		tf, err := tfilter.Parse(rule.Matcher)
+		if err != nil {
+			return nil, errors.WrapError(errors.ErrFilterRuleInvalid, err, rule.Matcher)
+		}
+		if filterDeleteExpr {
+			// Expression filtering uses the case-sensitive matcher directly.
+			filters = append(filters, tf)
+		}
+		if filterDeleteEvent {
+			if !cfg.CaseSensitive {
+				tf = tfilter.CaseInsensitive(tf)
+			}
+			filters = append(filters, tf)
+		}
+	}
+	return filters, nil
+}
+
+func (c *Config) hasDeleteFilter(schema, table string) bool {
+	for _, filter := range c.deleteEventFilters {
+		if filter.MatchTable(schema, table) {
+			return true
+		}
+	}
+	return false
+}
+
 func (c *Config) Apply(
 	sinkURI *url.URL,
 	changefeedID common.ChangeFeedID,
@@ -247,6 +296,10 @@ func (c *Config) Apply(
 
 	if cfg != nil {
 		c.mergeConfig(cfg)
+		c.deleteEventFilters, err = buildDeleteEventFilters(cfg)
+		if err != nil {
+			return err
+		}
 	}
 
 	query := sinkURI.Query()

pkg/sink/mysql/config_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/DATA-DOG/go-sqlmock"
 	dmysql "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/failpoint"
+	bf "github.com/pingcap/ticdc/pkg/binlog-filter"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/util"
@@ -331,6 +332,103 @@ func TestApplySinkURIParamsToConfig(t *testing.T) {
 	require.Equal(t, expected, cfg)
 }
 
+func TestApplyDeleteFilter(t *testing.T) {
+	t.Parallel()
+	uri, err := url.Parse("mysql://127.0.0.1:3306/?batch-dml-enable=true")
+	require.NoError(t, err)
+	for _, tt := range []struct {
+		name     string
+		filter   *config.FilterConfig
+		expected bool
+	}{
+		{name: "no filter"},
+		{name: "table filter", filter: &config.FilterConfig{Rules: []string{"test.*"}}},
+		{name: "delete event", filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"test.t"}, IgnoreEvent: []bf.EventType{bf.DeleteEvent}},
+		}}, expected: true},
+		{name: "delete expression", filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"test.t"}, IgnoreDeleteValueExpr: new("id > 0")},
+		}}, expected: true},
+		{name: "empty delete expression", filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"test.t"}, IgnoreDeleteValueExpr: new("")},
+		}}},
+		{name: "all dml", filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"test.t"}, IgnoreEvent: []bf.EventType{bf.AllDML}},
+		}}, expected: true},
+		{name: "all events", filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"test.t"}, IgnoreEvent: []bf.EventType{bf.AllEvent}},
+		}}, expected: true},
+		{name: "other event filters", filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"test.t"}, IgnoreEvent: []bf.EventType{bf.InsertEvent, bf.UpdateEvent}},
+			{Matcher: []string{"test.t2"}, IgnoreUpdateNewValueExpr: new("id > 0")},
+		}}},
+		{name: "delete in a later rule", filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"test.t1"}, IgnoreEvent: []bf.EventType{bf.CreateTable}},
+			{Matcher: []string{"test.t"}, IgnoreEvent: []bf.EventType{bf.DeleteEvent}},
+		}}, expected: true},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := New()
+			require.NoError(t, cfg.Apply(uri, common.NewChangefeedID4Test("test", "test"), &config.ChangefeedConfig{
+				TimeZone: "UTC",
+				Filter:   tt.filter,
+				SinkConfig: &config.SinkConfig{
+					TiDBSourceID: 1,
+				},
+			}))
+			require.Equal(t, tt.expected, cfg.hasDeleteFilter("test", "t"))
+			require.False(t, cfg.hasDeleteFilter("test", "unfiltered"))
+			// Per-event batching remains enabled, including when the URI enables it.
+			require.True(t, cfg.BatchDMLEnable)
+		})
+	}
+}
+
+func TestDeleteFilterTableMatching(t *testing.T) {
+	t.Parallel()
+	for _, caseSensitive := range []bool{false, true} {
+		filters, err := buildDeleteEventFilters(&config.ChangefeedConfig{
+			CaseSensitive: caseSensitive,
+			Filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+				// Exclusion in one rule must not override a separate matching rule.
+				{Matcher: []string{"test.reincluded"}, IgnoreEvent: []bf.EventType{bf.DeleteEvent}},
+				{Matcher: []string{"test.*", "!test.unfiltered", "!test.reincluded"}, IgnoreEvent: []bf.EventType{bf.DeleteEvent}},
+				{Matcher: []string{"other.t"}, IgnoreEvent: []bf.EventType{bf.DeleteEvent}},
+				{Matcher: []string{"test.unfiltered"}, IgnoreEvent: []bf.EventType{bf.UpdateEvent}},
+				{Matcher: []string{"expr.*", "!expr.skip"}, IgnoreDeleteValueExpr: new("id > 0")},
+			}},
+		})
+		require.NoError(t, err)
+		cfg := New()
+		cfg.deleteEventFilters = filters
+		for _, tt := range []struct {
+			schema   string
+			table    string
+			expected bool
+		}{
+			{"test", "t", true},
+			{"test", "unfiltered", false},
+			{"test", "reincluded", true},
+			{"other", "t", true},
+			{"other", "unfiltered", false},
+			{"TEST", "T", !caseSensitive},
+			{"expr", "t", true},
+			{"expr", "skip", false},
+			{"EXPR", "T", false},
+		} {
+			require.Equal(t, tt.expected, cfg.hasDeleteFilter(tt.schema, tt.table),
+				"caseSensitive=%t, table=%s.%s", caseSensitive, tt.schema, tt.table)
+		}
+	}
+
+	_, err := buildDeleteEventFilters(&config.ChangefeedConfig{
+		Filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"invalid"}, IgnoreEvent: []bf.EventType{bf.DeleteEvent}},
+		}},
+	})
+	require.ErrorContains(t, err, "ErrFilterRuleInvalid")
+}
+
 func TestApplyAsyncDDLTimeout(t *testing.T) {
 	t.Parallel()
 

pkg/sink/mysql/mysql_writer_dml.go

@@ -94,6 +94,8 @@ func (w *Writer) prepareDMLs(events []*commonEvent.DMLEvent) (*preparedDMLs, err
 			tableInfo := eventsInGroup[0].TableInfo
 			if w.cfg.EnableActiveActive {
 				queryList, argsList, rowTypesList = w.genActiveActiveSQL(tableInfo, eventsInGroup)
+			} else if w.cfg.hasDeleteFilter(tableInfo.GetSchemaName(), tableInfo.GetTableName()) {
+				queryList, argsList, rowTypesList = w.generateSQLsWithDeleteFilter(eventsInGroup)
 			} else {
 				if !w.shouldGenBatchSQL(tableInfo, eventsInGroup) {
 					queryList, argsList, rowTypesList = w.generateNormalSQLs(eventsInGroup)
@@ -114,6 +116,45 @@ func (w *Writer) prepareDMLs(events []*commonEvent.DMLEvent) (*preparedDMLs, err
 	return dmls, nil
 }
 
+// generateSQLsWithDeleteFilter preserves event order as the legacy MySQL sink
+// does. Deletes removed by filtering must not cause rows from different events
+// to be merged. Batching within each event still follows the configured limits.
+func (w *Writer) generateSQLsWithDeleteFilter(events []*commonEvent.DMLEvent) ([]string, [][]any, []common.RowType) {
+	if len(events) == 0 {
+		return nil, nil, nil
+	}
+	// The caller groups events by table and schema version, so table eligibility
+	// is shared. A single event cannot contain mixed recovery safe-mode states.
+	canBatch := w.canBatchTable(events[0].TableInfo)
+	queries := make([]string, 0, len(events))
+	args := make([][]any, 0, len(events))
+	rowTypes := make([]common.RowType, 0, len(events))
+	for _, event := range events {
+		if event.Len() == 0 {
+			continue
+		}
+		var (
+			sqls  []string
+			vals  [][]any
+			types []common.RowType
+		)
+		inSafeMode := w.cfg.SafeMode || w.isInErrorCausedSafeMode || event.CommitTs <= event.ReplicatingTs
+		if canBatch && event.Len() > 1 {
+			if inSafeMode {
+				sqls, vals, types = w.generateSafeBatchSQLForSingleEvent(event)
+			} else {
+				sqls, vals, types = w.generateSQLForSingleEvent(event, false)
+			}
+		} else {
+			sqls, vals, types = w.generateNormalSQL(event)
+		}
+		queries = append(queries, sqls...)
+		args = append(args, vals...)
+		rowTypes = append(rowTypes, types...)
+	}
+	return queries, args, rowTypes
+}
+
 func (w *Writer) genActiveActiveSQL(tableInfo *common.TableInfo, eventsInGroup []*commonEvent.DMLEvent) ([]string, [][]interface{}, []common.RowType) {
 	if !w.shouldGenBatchSQL(tableInfo, eventsInGroup) {
 		return w.generateActiveActiveNormalSQLs(eventsInGroup)
@@ -129,6 +170,19 @@ func (w *Writer) genActiveActiveSQL(tableInfo *common.TableInfo, eventsInGroup [
 // 4. There's more than one row in the group
 // 5. All events have the same safe mode status
 func (w *Writer) shouldGenBatchSQL(tableInfo *common.TableInfo, events []*commonEvent.DMLEvent) bool {
+	if !w.canBatchTable(tableInfo) {
+		return false
+	}
+	if len(events) == 1 && events[0].Len() == 1 {
+		return false
+	}
+
+	return allRowInSameSafeMode(w.cfg.SafeMode, events)
+}
+
+// canBatchTable checks the configuration and table properties shared by all
+// events with the same table ID and schema version.
+func (w *Writer) canBatchTable(tableInfo *common.TableInfo) bool {
 	if !w.cfg.BatchDMLEnable {
 		return false
 	}
@@ -150,12 +204,7 @@ func (w *Writer) shouldGenBatchSQL(tableInfo *common.TableInfo, events []*common
 			return false
 		}
 	}
-	// if tableInfo.HasVirtualColumns()
-	if len(events) == 1 && events[0].Len() == 1 {
-		return false
-	}
-
-	return allRowInSameSafeMode(w.cfg.SafeMode, events)
+	return true
 }
 
 // allRowInSameSafeMode determines whether all DMLEvents in a batch have the same safe mode status.

pkg/sink/mysql/mysql_writer_dml_batch.go

@@ -142,6 +142,27 @@ func (w *Writer) generateSQLForSingleEvent(event *commonEvent.DMLEvent, inDataSa
 	return w.batchSingleTxnDmls(rowLists, tableInfo, inDataSafeMode)
 }
 
+// generateSafeBatchSQLForSingleEvent rewrites updates as delete/replace pairs
+// so safe mode can recover rows missing downstream without merging events.
+func (w *Writer) generateSafeBatchSQLForSingleEvent(event *commonEvent.DMLEvent) ([]string, [][]any, []common.RowType) {
+	rows := make([]*commonEvent.RowChange, 0, event.Len())
+	for {
+		row, ok := event.GetNextRow()
+		if !ok {
+			event.Rewind()
+			break
+		}
+		if row.RowType == common.RowTypeUpdate {
+			rows = append(rows,
+				&commonEvent.RowChange{RowType: common.RowTypeDelete, PreRow: row.PreRow},
+				&commonEvent.RowChange{RowType: common.RowTypeInsert, Row: row.Row})
+		} else {
+			rows = append(rows, &row)
+		}
+	}
+	return w.batchSingleTxnDmls(rows, event.TableInfo, true)
+}
+
 // 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 (
@@ -241,6 +262,14 @@ func (w *Writer) buildRowChangesForUnSafeBatch(
 		}
 	}
 
+	// Validate all updates before merging so a filtered duplicate cannot hide
+	// an unexpected, unsplit primary-key change later in the batch.
+	for _, row := range rowLists {
+		if row.rowChange.RowType == common.RowTypeUpdate && !compareKeys(row.preRowKeys, row.rowKeys) {
+			return nil, nil, cerror.ErrUnexpected.FastGenByArgs("update row key mismatch")
+		}
+	}
+
 	// 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:
@@ -283,9 +312,6 @@ func (w *Writer) buildRowChangesForUnSafeBatch(
 							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
@@ -304,9 +330,6 @@ func (w *Writer) buildRowChangesForUnSafeBatch(
 					}
 				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")
@@ -329,9 +352,6 @@ func (w *Writer) buildRowChangesForUnSafeBatch(
 						}
 					} 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,