diff --git a/pkg/sink/mysql/config.go b/pkg/sink/mysql/config.go
index 64712cb2cb..611ad52c4f 100644
--- a/pkg/sink/mysql/config.go
+++ b/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()
diff --git a/pkg/sink/mysql/config_test.go b/pkg/sink/mysql/config_test.go
index 0f8170c2ad..73c4c7eb30 100644
--- a/pkg/sink/mysql/config_test.go
+++ b/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()
 
diff --git a/pkg/sink/mysql/mysql_writer_dml.go b/pkg/sink/mysql/mysql_writer_dml.go
index 1d719780bc..126cbea166 100644
--- a/pkg/sink/mysql/mysql_writer_dml.go
+++ b/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.
diff --git a/pkg/sink/mysql/mysql_writer_dml_batch.go b/pkg/sink/mysql/mysql_writer_dml_batch.go
index b4bd085d84..855df6e17c 100644
--- a/pkg/sink/mysql/mysql_writer_dml_batch.go
+++ b/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,
diff --git a/pkg/sink/mysql/mysql_writer_dml_test.go b/pkg/sink/mysql/mysql_writer_dml_test.go
index 7b62ba6605..a82df93312 100644
--- a/pkg/sink/mysql/mysql_writer_dml_test.go
+++ b/pkg/sink/mysql/mysql_writer_dml_test.go
@@ -22,11 +22,13 @@ import (
 
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
+	"github.com/pingcap/ticdc/pkg/config"
 	tidbmodel "github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	tidbmysql "github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/types"
 	"github.com/pingcap/tidb/pkg/util/chunk"
+	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/stretchr/testify/require"
 )
 
@@ -336,6 +338,276 @@ func TestGenerateBatchSQL(t *testing.T) {
 	require.Equal(t, 2000, len(args[0]), "Should have 2000 arguments (1000 rows * 2 columns)")
 }
 
+func TestGenerateBatchSQLWithFilteredEvents(t *testing.T) {
+	writer, db, _ := newTestMysqlWriter(t)
+	defer db.Close()
+	writer.cfg.deleteEventFilters = []tfilter.Filter{tfilter.All()}
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+
+	helper.Tk().MustExec("use test")
+	require.NotNil(t, helper.DDL2Job("create table t (id int primary key, name varchar(32), unique key uk_name (name));"))
+
+	deleteEvent := helper.DML2DeleteEvent("test", "t", "insert into t values (1, 'deleted')", "delete from t where id = 1")
+	insertEvent := helper.DML2Event("test", "t", "insert into t values (1, 'first')")
+	helper.ExecuteDeleteDml("test", "t", "delete from t where id = 1")
+	updateEvent, _ := helper.DML2UpdateEvent("test", "t", "insert into t values (1, 'before')", "update t set name = 'updated' where id = 1")
+	// Omit the delete event to reproduce DELETE filtering before a same-key insert.
+	helper.ExecuteDeleteDml("test", "t", "delete from t where id = 1")
+	insertEvent2 := helper.DML2Event("test", "t", "insert into t values (1, 'last')", "insert into t values (2, 'other')")
+	insertEvent3 := helper.DML2Event("test", "t", "insert into t values (3, 'third')")
+
+	const (
+		updateSQL   = "UPDATE `test`.`t` SET `id` = ?,`name` = ? WHERE `id` = ? LIMIT 1"
+		deleteSQL   = "DELETE FROM `test`.`t` WHERE `id` = ? LIMIT 1"
+		insertSQL   = "INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?)"
+		insertsSQL  = "INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?)"
+		replaceSQL  = "REPLACE INTO `test`.`t` (`id`,`name`) VALUES (?,?)"
+		replacesSQL = "REPLACE INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?)"
+	)
+	tests := []struct {
+		name            string
+		events          []*commonEvent.DMLEvent
+		unsafeSQLs      []string
+		safeSQLs        []string
+		args            [][]any
+		rowTypes        []common.RowType
+		safeArgs        [][]any
+		safeRowTypes    []common.RowType
+		disableBatch    bool
+		invalidSequence bool
+	}{
+		{
+			name:         "single update",
+			events:       []*commonEvent.DMLEvent{updateEvent},
+			unsafeSQLs:   []string{updateSQL},
+			safeSQLs:     []string{deleteSQL, replaceSQL},
+			args:         [][]any{{int64(1), "updated", int64(1)}},
+			rowTypes:     []common.RowType{common.RowTypeUpdate},
+			safeArgs:     [][]any{{int64(1)}, {int64(1), "updated"}},
+			safeRowTypes: []common.RowType{common.RowTypeDelete, common.RowTypeInsert},
+		},
+		{
+			name:         "single update without batching",
+			events:       []*commonEvent.DMLEvent{updateEvent},
+			unsafeSQLs:   []string{updateSQL},
+			safeSQLs:     []string{deleteSQL, replaceSQL},
+			args:         [][]any{{int64(1), "updated", int64(1)}},
+			rowTypes:     []common.RowType{common.RowTypeUpdate},
+			safeArgs:     [][]any{{int64(1)}, {int64(1), "updated"}},
+			safeRowTypes: []common.RowType{common.RowTypeDelete, common.RowTypeInsert},
+			disableBatch: true,
+		},
+		{
+			name:       "distinct inserts stay in separate events",
+			events:     []*commonEvent.DMLEvent{insertEvent, insertEvent3},
+			unsafeSQLs: []string{insertSQL, insertSQL},
+			safeSQLs:   []string{replaceSQL, replaceSQL},
+			args:       [][]any{{int64(1), "first"}, {int64(3), "third"}},
+			rowTypes:   []common.RowType{common.RowTypeInsert, common.RowTypeInsert},
+		},
+		{
+			name:            "update then insert",
+			events:          []*commonEvent.DMLEvent{updateEvent, insertEvent2},
+			unsafeSQLs:      []string{updateSQL, insertsSQL},
+			safeSQLs:        []string{deleteSQL, replaceSQL, replacesSQL},
+			args:            [][]any{{int64(1), "updated", int64(1)}, {int64(1), "last", int64(2), "other"}},
+			rowTypes:        []common.RowType{common.RowTypeUpdate, common.RowTypeInsert},
+			safeArgs:        [][]any{{int64(1)}, {int64(1), "updated"}, {int64(1), "last", int64(2), "other"}},
+			safeRowTypes:    []common.RowType{common.RowTypeDelete, common.RowTypeInsert, common.RowTypeInsert},
+			invalidSequence: true,
+		},
+		{
+			name:            "insert then insert",
+			events:          []*commonEvent.DMLEvent{insertEvent, insertEvent2},
+			unsafeSQLs:      []string{insertSQL, insertsSQL},
+			safeSQLs:        []string{replaceSQL, replacesSQL},
+			args:            [][]any{{int64(1), "first"}, {int64(1), "last", int64(2), "other"}},
+			rowTypes:        []common.RowType{common.RowTypeInsert, common.RowTypeInsert},
+			invalidSequence: true,
+		},
+		{
+			name:            "delete then insert then insert",
+			events:          []*commonEvent.DMLEvent{deleteEvent, insertEvent, insertEvent2},
+			unsafeSQLs:      []string{deleteSQL, insertSQL, insertsSQL},
+			safeSQLs:        []string{deleteSQL, replaceSQL, replacesSQL},
+			args:            [][]any{{int64(1)}, {int64(1), "first"}, {int64(1), "last", int64(2), "other"}},
+			rowTypes:        []common.RowType{common.RowTypeDelete, common.RowTypeInsert, common.RowTypeInsert},
+			invalidSequence: true,
+		},
+	}
+
+	for _, mode := range []string{"unsafe", "configured safe mode", "error caused safe mode", "recovery safe mode"} {
+		t.Run(mode, func(t *testing.T) {
+			writer.cfg.SafeMode = mode == "configured safe mode"
+			writer.isInErrorCausedSafeMode = mode == "error caused safe mode"
+			for _, tt := range tests {
+				t.Run(tt.name, func(t *testing.T) {
+					writer.cfg.BatchDMLEnable = !tt.disableBatch
+					for i, event := range tt.events {
+						event.CommitTs = uint64(i + 2)
+						event.ReplicatingTs = 1
+						if mode == "recovery safe mode" {
+							event.ReplicatingTs = 10
+						}
+					}
+					// Preparing twice verifies every event is rewound for retries.
+					for range 2 {
+						require.NotPanics(t, func() {
+							dmls, err := writer.prepareDMLs(tt.events)
+							defer dmlsPool.Put(dmls)
+							require.NoError(t, err)
+							if mode == "unsafe" {
+								require.Equal(t, tt.unsafeSQLs, dmls.sqls)
+							} else {
+								require.Equal(t, tt.safeSQLs, dmls.sqls)
+							}
+							if mode != "unsafe" && tt.safeArgs != nil {
+								require.Equal(t, tt.safeArgs, dmls.values)
+								require.Equal(t, tt.safeRowTypes, dmls.rowTypes)
+							} else {
+								require.Equal(t, tt.args, dmls.values)
+								require.Equal(t, tt.rowTypes, dmls.rowTypes)
+							}
+						})
+					}
+				})
+			}
+		})
+	}
+
+	// The same incomplete sequences without delete filtering must still expose
+	// an invariant violation instead of silently switching SQL generation paths.
+	writer.cfg.deleteEventFilters = nil
+	writer.isInErrorCausedSafeMode = false
+	for _, safeMode := range []bool{false, true} {
+		writer.cfg.SafeMode = safeMode
+		for _, tt := range tests {
+			if !tt.invalidSequence {
+				continue
+			}
+			for _, event := range tt.events {
+				event.CommitTs = 2
+				event.ReplicatingTs = 1
+			}
+			expectedPanic := "invalid rows when generating batch SQL in unsafe mode"
+			if safeMode {
+				expectedPanic = "invalid row changes"
+			}
+			require.PanicsWithValue(t, expectedPanic, func() { writer.generateBatchSQL(tt.events) })
+		}
+	}
+}
+
+func TestDeleteFilterSelectsTableDMLMode(t *testing.T) {
+	writer, db, _ := newTestMysqlWriter(t)
+	defer db.Close()
+	filters, err := buildDeleteEventFilters(&config.ChangefeedConfig{
+		Filter: &config.FilterConfig{EventFilters: []*config.EventFilterRule{
+			{Matcher: []string{"test.filtered"}, IgnoreDeleteValueExpr: new("id > 0")},
+		}},
+	})
+	require.NoError(t, err)
+	writer.cfg.deleteEventFilters = filters
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+	helper.Tk().MustExec("use test")
+	var inserts, updates []*commonEvent.DMLEvent
+	for _, table := range []string{"filtered", "regular"} {
+		require.NotNil(t, helper.DDL2Job("create table "+table+" (id int primary key, name varchar(32));"))
+		inserts = append(inserts,
+			helper.DML2Event("test", table, "insert into "+table+" values (1, 'first')"),
+			helper.DML2Event("test", table, "insert into "+table+" values (2, 'second')"))
+		update, _ := helper.DML2UpdateEvent("test", table,
+			"insert into "+table+" values (3, 'before')", "update "+table+" set name = 'after' where id = 3")
+		updates = append(updates, update)
+	}
+	for _, event := range append(inserts, updates...) {
+		event.CommitTs = 2
+		event.ReplicatingTs = 1
+	}
+	checkSQLs := func(t *testing.T, events []*commonEvent.DMLEvent, expected map[string][][]any) {
+		t.Helper()
+		dmls, err := writer.prepareDMLs(events)
+		require.NoError(t, err)
+		defer dmlsPool.Put(dmls)
+		actual := make(map[string][][]any)
+		for i, sql := range dmls.sqls {
+			actual[sql] = append(actual[sql], dmls.values[i])
+		}
+		require.Len(t, actual, len(expected))
+		for sql, want := range expected {
+			got, ok := actual[sql]
+			require.True(t, ok, "missing SQL: %s", sql)
+			if strings.HasSuffix(sql, "VALUES (?,?),(?,?)") {
+				// Cross-event safe-mode batching may reorder independent rows.
+				require.Len(t, got, 1)
+				require.Len(t, got[0], 4)
+				require.ElementsMatch(t, [][]any{want[0][:2], want[0][2:]}, [][]any{got[0][:2], got[0][2:]})
+			} else {
+				// Per-event statements must preserve their original order.
+				require.Equal(t, want, got, "SQL: %s", sql)
+			}
+		}
+	}
+	for _, safeMode := range []bool{false, true} {
+		t.Run(fmt.Sprintf("safeMode=%t", safeMode), func(t *testing.T) {
+			writer.cfg.SafeMode = safeMode
+			insert := "INSERT"
+			if safeMode {
+				insert = "REPLACE"
+			}
+			// Both tables share one flush; only the unfiltered table merges events.
+			checkSQLs(t, inserts, map[string][][]any{
+				insert + " INTO `test`.`filtered` (`id`,`name`) VALUES (?,?)":      {{int64(1), "first"}, {int64(2), "second"}},
+				insert + " INTO `test`.`regular` (`id`,`name`) VALUES (?,?),(?,?)": {{int64(1), "first", int64(2), "second"}},
+			})
+			expected := map[string][][]any{
+				"UPDATE `test`.`filtered` SET `id` = ?,`name` = ? WHERE `id` = ? LIMIT 1": {{int64(3), "after", int64(3)}},
+			}
+			if safeMode {
+				delete(expected, "UPDATE `test`.`filtered` SET `id` = ?,`name` = ? WHERE `id` = ? LIMIT 1")
+				expected["DELETE FROM `test`.`filtered` WHERE `id` = ? LIMIT 1"] = [][]any{{int64(3)}}
+				expected["REPLACE INTO `test`.`filtered` (`id`,`name`) VALUES (?,?)"] = [][]any{{int64(3), "after"}}
+				expected["DELETE FROM `test`.`regular` WHERE `id` = ? LIMIT 1"] = [][]any{{int64(3)}}
+				expected["REPLACE INTO `test`.`regular` (`id`,`name`) VALUES (?,?)"] = [][]any{{int64(3), "after"}}
+			} else {
+				expected["UPDATE `test`.`regular` SET `id` = ?,`name` = ? WHERE `id` = ? LIMIT 1"] = [][]any{{int64(3), "after", int64(3)}}
+			}
+			checkSQLs(t, updates, expected)
+		})
+	}
+}
+
+func TestGenerateBatchSQLKeyMismatch(t *testing.T) {
+	writer, db, _ := newTestMysqlWriter(t)
+	defer db.Close()
+	writer.cfg.SafeMode = false
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+	helper.Tk().MustExec("use test")
+	require.NotNil(t, helper.DDL2Job("create table t (id int primary key, name varchar(32));"))
+
+	updateEvent, _ := helper.DML2UpdateEvent("test", "t", "insert into t values (1, 'before')", "update t set name = 'after' where id = 1")
+	insertEvent := helper.DML2Event("test", "t", "insert into t values (2, 'other')")
+	// A primary-key change must have been split before reaching the batch merger.
+	updateEvent.Rows.TruncateTo(1)
+	updateEvent.Rows.AppendRow(insertEvent.Rows.GetRow(0))
+	for _, event := range []*commonEvent.DMLEvent{updateEvent, insertEvent} {
+		event.CommitTs = 2
+		event.ReplicatingTs = 1
+	}
+
+	for _, events := range [][]*commonEvent.DMLEvent{
+		{updateEvent, insertEvent},
+		{insertEvent, insertEvent, updateEvent},
+	} {
+		require.PanicsWithValue(t, "invalid rows when generating batch SQL in unsafe mode", func() {
+			writer.generateBatchSQL(events)
+		})
+	}
+}
+
 func TestSQLGenerationSafeModeBoundary(t *testing.T) {
 	writer, db, _ := newTestMysqlWriter(t)
 	defer db.Close()
diff --git a/pkg/sink/mysql/mysql_writer_test.go b/pkg/sink/mysql/mysql_writer_test.go
index 525c6ef1e4..2dfc9d8268 100644
--- a/pkg/sink/mysql/mysql_writer_test.go
+++ b/pkg/sink/mysql/mysql_writer_test.go
@@ -16,6 +16,7 @@ package mysql
 import (
 	"context"
 	"database/sql"
+	"database/sql/driver"
 	"fmt"
 	"testing"
 	"time"
@@ -41,6 +42,7 @@ import (
 	"github.com/pingcap/tidb/pkg/dxf/framework/handle"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
+	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
 )
@@ -284,6 +286,168 @@ func TestMysqlWriter_FlushDML_DuplicateEntryRetry(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestMysqlWriterFlushFilteredEvents(t *testing.T) {
+	for _, multiStmtEnable := range []bool{false, true} {
+		for _, failRetry := range []bool{false, true} {
+			t.Run(fmt.Sprintf("multiStmtEnable=%t/failRetry=%t", multiStmtEnable, failRetry), func(t *testing.T) {
+				writer, db, mock := newTestMysqlWriter(t)
+				defer db.Close()
+				writer.cfg.SafeMode = false
+				writer.cfg.deleteEventFilters = []tfilter.Filter{tfilter.All()}
+				writer.cfg.MultiStmtEnable = multiStmtEnable
+				writer.cfg.CachePrepStmts = false
+				writer.cfg.DMLMaxRetry = 0
+
+				helper := commonEvent.NewEventTestHelper(t)
+				defer helper.Close()
+				helper.Tk().MustExec("use test")
+				require.NotNil(t, helper.DDL2Job("create table t (id int primary key, name varchar(32), unique key uk_name (name));"))
+				updateEvent, _ := helper.DML2UpdateEvent("test", "t", "insert into t values (1, 'before')", "update t set name = 'updated' where id = 1")
+				// DELETE filtering leaves an update followed by a same-key insert.
+				helper.ExecuteDeleteDml("test", "t", "delete from t where id = 1")
+				insertEvent := helper.DML2Event("test", "t", "insert into t values (1, 'last')", "insert into t values (2, 'other')")
+				events := []*commonEvent.DMLEvent{updateEvent, insertEvent}
+				flushedEvents := 0
+				for i, event := range events {
+					event.CommitTs = uint64(i + 2)
+					event.ReplicatingTs = 1
+					event.AddPostFlushFunc(func() { flushedEvents++ })
+				}
+
+				const (
+					updateSQL     = "UPDATE `test`.`t` SET `id` = ?,`name` = ? WHERE `id` = ? LIMIT 1"
+					deleteSQL     = "DELETE FROM `test`.`t` WHERE `id` = ? LIMIT 1"
+					replaceRowSQL = "REPLACE INTO `test`.`t` (`id`,`name`) VALUES (?,?)"
+					insertSQL     = "INSERT INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?)"
+					replaceSQL    = "REPLACE INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?)"
+				)
+				duplicateErr := &mysql.MySQLError{Number: 1062, Message: "Duplicate entry '1' for key 'PRIMARY'"}
+				retryErr := &mysql.MySQLError{Number: 1213, Message: "Deadlock found when trying to get lock"}
+
+				for attempt, insert := range []string{insertSQL, replaceSQL} {
+					update := updateSQL
+					updateArgs := []driver.Value{1, "updated", 1, 1, "last", 2, "other"}
+					if attempt == 1 {
+						update = deleteSQL + ";" + replaceRowSQL
+						updateArgs = []driver.Value{1, 1, "updated", 1, "last", 2, "other"}
+					}
+					var execErr error
+					if attempt == 0 {
+						execErr = duplicateErr
+					} else if failRetry {
+						execErr = retryErr
+					}
+					if multiStmtEnable {
+						expectation := mock.ExpectExec("BEGIN;" + update + ";" + insert + ";COMMIT;").
+							WithArgs(updateArgs...)
+						if execErr != nil {
+							expectation.WillReturnError(execErr)
+							mock.ExpectExec("ROLLBACK").WillReturnResult(sqlmock.NewResult(0, 0))
+						} else {
+							expectation.WillReturnResult(newTestMysqlResult(t, 0, 1, 1, 2, 0))
+						}
+					} else {
+						mock.ExpectBegin()
+						if attempt == 0 {
+							mock.ExpectExec(updateSQL).WithArgs(1, "updated", 1).
+								WillReturnResult(sqlmock.NewResult(0, 1))
+						} else {
+							mock.ExpectExec(deleteSQL).WithArgs(1).WillReturnResult(sqlmock.NewResult(0, 1))
+							mock.ExpectExec(replaceRowSQL).WithArgs(1, "updated").WillReturnResult(sqlmock.NewResult(1, 1))
+						}
+						expectation := mock.ExpectExec(insert).WithArgs(1, "last", 2, "other")
+						if execErr != nil {
+							expectation.WillReturnError(execErr)
+							mock.ExpectRollback()
+						} else {
+							expectation.WillReturnResult(sqlmock.NewResult(1, 2))
+							mock.ExpectCommit()
+						}
+					}
+				}
+
+				var err error
+				require.NotPanics(t, func() { err = writer.Flush(events) })
+				if failRetry {
+					require.ErrorContains(t, err, retryErr.Message)
+					require.ErrorIs(t, err, retryErr)
+					require.Zero(t, flushedEvents)
+				} else {
+					require.NoError(t, err)
+					require.Equal(t, len(events), flushedEvents)
+				}
+				require.True(t, writer.isInErrorCausedSafeMode)
+				require.NoError(t, mock.ExpectationsWereMet())
+			})
+		}
+	}
+}
+
+func TestMysqlWriterFlushFilteredUpdatesInSafeMode(t *testing.T) {
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+	helper.Tk().MustExec("use test")
+	require.NotNil(t, helper.DDL2Job("create table t (id int primary key, name varchar(32));"))
+	first, _ := helper.DML2UpdateEvent("test", "t", "insert into t values (1, 'before')", "update t set name = 'after' where id = 1")
+	second, _ := helper.DML2UpdateEvent("test", "t", "insert into t values (2, 'before')", "update t set name = 'after' where id = 2")
+
+	for _, mode := range []string{"configured", "recovery", "duplicate key retry"} {
+		for _, batchEnabled := range []bool{false, true} {
+			for _, rowCount := range []int{1, 2} {
+				t.Run(fmt.Sprintf("%s/batch=%t/rows=%d", mode, batchEnabled, rowCount), func(t *testing.T) {
+					writer, db, mock := newTestMysqlWriter(t)
+					defer db.Close()
+					writer.cfg.SafeMode = mode == "configured"
+					writer.cfg.DMLMaxRetry = 0
+					writer.cfg.BatchDMLEnable = batchEnabled
+					writer.cfg.MultiStmtEnable = false
+					writer.cfg.CachePrepStmts = false
+					writer.cfg.deleteEventFilters = []tfilter.Filter{tfilter.All()}
+					writer.isInErrorCausedSafeMode = mode == "duplicate key retry"
+					writer.lastErrorCausedSafeModeTime = time.Now()
+
+					event := commonEvent.NewDMLEvent(first.DispatcherID, first.PhysicalTableID, 1, 2, first.TableInfo)
+					event.Rows = first.Rows.CopyConstruct()
+					event.RowTypes = append(event.RowTypes, first.RowTypes...)
+					event.Length = 1
+					if rowCount == 2 {
+						for i := range second.Rows.NumRows() {
+							event.Rows.AppendRow(second.Rows.GetRow(i))
+						}
+						event.RowTypes = append(event.RowTypes, second.RowTypes...)
+						event.Length++
+					}
+					event.ReplicatingTs = 1
+					if mode == "recovery" {
+						event.ReplicatingTs = 3
+					}
+					flushed := false
+					event.AddPostFlushFunc(func() { flushed = true })
+
+					mock.ExpectBegin()
+					if batchEnabled && rowCount == 2 {
+						mock.ExpectExec("DELETE FROM `test`.`t` WHERE (`id`) IN ((?),(?))").WithArgs(1, 2).
+							WillReturnResult(sqlmock.NewResult(0, 0))
+						mock.ExpectExec("REPLACE INTO `test`.`t` (`id`,`name`) VALUES (?,?),(?,?)").WithArgs(1, "after", 2, "after").
+							WillReturnResult(sqlmock.NewResult(1, 2))
+					} else {
+						for id := 1; id <= rowCount; id++ {
+							mock.ExpectExec("DELETE FROM `test`.`t` WHERE `id` = ? LIMIT 1").WithArgs(id).
+								WillReturnResult(sqlmock.NewResult(0, 0))
+							mock.ExpectExec("REPLACE INTO `test`.`t` (`id`,`name`) VALUES (?,?)").WithArgs(id, "after").
+								WillReturnResult(sqlmock.NewResult(1, 1))
+						}
+					}
+					mock.ExpectCommit()
+					require.NoError(t, writer.Flush([]*commonEvent.DMLEvent{event}))
+					require.True(t, flushed)
+					require.NoError(t, mock.ExpectationsWereMet())
+				})
+			}
+		}
+	}
+}
+
 func TestMysqlWriter_FlushMultiDML(t *testing.T) {
 	writer, db, mock := newTestMysqlWriter(t)
 	defer db.Close()
diff --git a/tests/integration_tests/batch_dml_with_filter/run.sh b/tests/integration_tests/batch_dml_with_filter/run.sh
new file mode 100755
index 0000000000..e8ea44ca79
--- /dev/null
+++ b/tests/integration_tests/batch_dml_with_filter/run.sh
@@ -0,0 +1,159 @@
+#!/bin/bash
+
+set -euo pipefail
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source "$CUR/../_utils/test_prepare"
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=${CDC_BINARY:-cdc.test}
+SINK_TYPE=$1
+
+function check_rows() {
+	local table=$1
+	local expected=$2
+	local actual
+	actual=$(mysql -uroot -h"$DOWN_TIDB_HOST" -P"$DOWN_TIDB_PORT" -N -B -e \
+		"SELECT GROUP_CONCAT(CONCAT_WS(':', id, biz_key, biz_status, payload) ORDER BY id SEPARATOR ',')
+        FROM batch_dml_with_filter.$table") || return 1
+	if [ "$actual" != "$expected" ]; then
+		echo "Unexpected rows in $table: expected '$expected', got '$actual'"
+		return 1
+	fi
+}
+export -f check_rows
+
+function run_case() {
+	local safe_mode=$1
+	local scenario=$2
+	local filter_type=$3
+	local table="${scenario}_${safe_mode}_${filter_type}"
+	local control_table="${table}_control"
+	local changefeed_id="batch-filter-${scenario//_/-}-$safe_mode-$filter_type"
+	local config="$WORK_DIR/$table.toml"
+	local table_definition="(
+        id BIGINT PRIMARY KEY,
+        biz_key VARCHAR(64) NOT NULL,
+        biz_status TINYINT NOT NULL,
+        payload VARCHAR(64),
+        UNIQUE KEY uk_biz (biz_key, biz_status))"
+	local table_name
+	for table_name in "$table" "$control_table"; do
+		run_sql "CREATE TABLE batch_dml_with_filter.$table_name $table_definition" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+		run_sql "CREATE TABLE batch_dml_with_filter.$table_name $table_definition" "$DOWN_TIDB_HOST" "$DOWN_TIDB_PORT"
+	done
+	cat >"$config" <<EOF
+[filter]
+rules = ["batch_dml_with_filter.$table", "batch_dml_with_filter.$control_table"]
+
+[[filter.event-filters]]
+matcher = ["batch_dml_with_filter.$table"]
+EOF
+	if [ "$filter_type" = "event" ]; then
+		echo 'ignore-event = ["delete"]' >>"$config"
+	else
+		echo 'ignore-delete-value-expr = "id > 0"' >>"$config"
+	fi
+
+	cdc_cli_changefeed create --changefeed-id="$changefeed_id" --config="$config" \
+		--sink-uri="mysql://root@$DOWN_TIDB_HOST:$DOWN_TIDB_PORT/?batch-dml-enable=true&safe-mode=$safe_mode&worker-count=1"
+	ensure 30 check_changefeed_state "http://$UP_PD_HOST_1:$UP_PD_PORT_1" "$changefeed_id" normal null ""
+
+	# Synchronize a seed before writing the workload so unsafe-mode events are
+	# newer than ReplicatingTs and do not take the recovery safe-mode path.
+	run_sql "INSERT INTO batch_dml_with_filter.$table VALUES (99, 'retained', 0, 'kept')" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	do_retry 60 1 bash -c 'check_rows "$@"' _ "$table" "99:retained:0:kept"
+	if [ "$scenario" = "update_insert" ]; then
+		run_sql "INSERT INTO batch_dml_with_filter.$table VALUES (1, 'business', 1, 'before')" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+		do_retry 60 1 bash -c 'check_rows "$@"' _ "$table" "1:business:1:before,99:retained:0:kept"
+	fi
+	run_sql "INSERT INTO batch_dml_with_filter.$control_table VALUES
+        (1, 'control', 0, 'before'), (99, 'retained', 0, 'kept')" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	do_retry 60 1 bash -c 'check_rows "$@"' _ "$control_table" "1:control:0:before,99:retained:0:kept"
+
+	local log_start
+	log_start=$(wc -l <"$WORK_DIR/cdc.log")
+	# Reuse the same key in many separate transactions so normal batching sees
+	# UPDATE -> INSERT or INSERT -> INSERT after DELETE filtering.
+	local workload="$WORK_DIR/$table.sql"
+	{
+		echo "SET autocommit = 1;"
+		for ((i = 0; i < 200; i++)); do
+			echo "UPDATE batch_dml_with_filter.$control_table SET payload = 'round$i' WHERE id = 1;"
+			if [ "$scenario" = "update_insert" ]; then
+				echo "UPDATE batch_dml_with_filter.$table SET payload = 'updated' WHERE id = 1;"
+				echo "DELETE FROM batch_dml_with_filter.$table WHERE id = 1;"
+				echo "INSERT INTO batch_dml_with_filter.$table VALUES (1, 'business', 1, 'round');"
+			else
+				echo "INSERT INTO batch_dml_with_filter.$table VALUES (1, 'business', 1, 'first');"
+				echo "DELETE FROM batch_dml_with_filter.$table WHERE id = 1;"
+			fi
+		done
+		if [ "$scenario" = "update_insert" ]; then
+			echo "DELETE FROM batch_dml_with_filter.$table WHERE id = 1;"
+		fi
+		echo "INSERT INTO batch_dml_with_filter.$table VALUES (1, 'business', 1, 'last'), (2, 'other', 0, 'other');"
+	} >"$workload"
+	mysql -uroot -h"$UP_TIDB_HOST" -P"$UP_TIDB_PORT" <"$workload" >"$WORK_DIR/$table.workload.log" 2>&1
+	do_retry 60 1 bash -c 'check_rows "$@"' _ "$table" "1:business:1:last,2:other:0:other,99:retained:0:kept"
+
+	if [ "$safe_mode" = "false" ]; then
+		# Unsafe writes must reach the downstream duplicate-key retry, whose
+		# safe-mode regeneration must also handle the filtered event sequence.
+		tail -n +"$((log_start + 1))" "$WORK_DIR/cdc.log" >"$WORK_DIR/$table.retry.log"
+		grep -Fq "Meet Duplicate Entry Error, retry the dmls in safemode" "$WORK_DIR/$table.retry.log"
+	fi
+
+	# A distinct deleted row must remain downstream, and a later update must
+	# still make progress after any safe-mode retry.
+	run_sql "DELETE FROM batch_dml_with_filter.$table WHERE id = 99;
+        UPDATE batch_dml_with_filter.$table SET payload = 'continued' WHERE id = 2" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	do_retry 60 1 bash -c 'check_rows "$@"' _ "$table" "1:business:1:last,2:other:0:continued,99:retained:0:kept"
+
+	# A filtered delete followed by an insert with a different primary key can
+	# reuse a unique key. Match the legacy REPLACE behavior: only the new row
+	# survives this conflict, while the unrelated retained row stays untouched.
+	run_sql "INSERT INTO batch_dml_with_filter.$table VALUES (3, 'reused', 0, 'old')" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	do_retry 60 1 bash -c 'check_rows "$@"' _ "$table" "1:business:1:last,2:other:0:continued,3:reused:0:old,99:retained:0:kept"
+	run_sql "DELETE FROM batch_dml_with_filter.$table WHERE id = 3;
+        INSERT INTO batch_dml_with_filter.$table VALUES (4, 'reused', 0, 'new')" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	do_retry 60 1 bash -c 'check_rows "$@"' _ "$table" "1:business:1:last,2:other:0:continued,4:reused:0:new,99:retained:0:kept"
+	# The same sink also handles this table without DELETE filtering. Its
+	# deletes must be replicated while the filtered table retains deleted rows.
+	run_sql "DELETE FROM batch_dml_with_filter.$control_table WHERE id = 99;
+        UPDATE batch_dml_with_filter.$control_table SET payload = 'continued' WHERE id = 1" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	do_retry 60 1 bash -c 'check_rows "$@"' _ "$control_table" "1:control:0:continued"
+	check_changefeed_state "http://$UP_PD_HOST_1:$UP_PD_PORT_1" "$changefeed_id" normal null ""
+	kill -0 "$CDC_PID"
+	cdc_cli_changefeed remove --changefeed-id="$changefeed_id"
+}
+
+function run() {
+	if [ "$SINK_TYPE" != "mysql" ]; then
+		return
+	fi
+	rm -rf "$WORK_DIR" && mkdir -p "$WORK_DIR"
+	start_tidb_cluster --workdir "$WORK_DIR"
+	export GO_FAILPOINTS=''
+	run_cdc_server --workdir "$WORK_DIR" --binary "$CDC_BINARY"
+	CDC_PID=$(get_cdc_pid "$CDC_HOST" "$CDC_PORT")
+	for host_port in "$UP_TIDB_HOST:$UP_TIDB_PORT" "$DOWN_TIDB_HOST:$DOWN_TIDB_PORT"; do
+		run_sql "CREATE DATABASE batch_dml_with_filter" "${host_port%:*}" "${host_port##*:}"
+	done
+	for safe_mode in false true; do
+		for scenario in update_insert insert_insert; do
+			for filter_type in event expr; do
+				run_case "$safe_mode" "$scenario" "$filter_type"
+			done
+		done
+	done
+	if grep -Eq '^panic:|\[PANIC\]' "$WORK_DIR/stdout.log" "$WORK_DIR/cdc.log"; then
+		echo "CDC panicked while replicating filtered batch DML"
+		exit 1
+	fi
+	cleanup_process "$CDC_BINARY"
+}
+
+trap 'stop_test "$WORK_DIR"' EXIT
+run
+check_logs "$WORK_DIR"
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"
diff --git a/tests/integration_tests/run_light_it_in_ci.sh b/tests/integration_tests/run_light_it_in_ci.sh
index 7eb3faebaf..114aa9b757 100755
--- a/tests/integration_tests/run_light_it_in_ci.sh
+++ b/tests/integration_tests/run_light_it_in_ci.sh
@@ -54,14 +54,14 @@ mysql_groups=(
 	# G10
 	'overwrite_resume_with_syncpoint restart_changefeed changefeed_error bdr_mode fail_over_ddl_K split_table_check'
 	# G11
-	'kill_owner_with_ddl multi_tables_ddl ddl_attributes multi_cdc_cluster fail_over_ddl_L'
+	'pause_changefeed_with_long_time_ddl kill_owner_with_ddl multi_tables_ddl ddl_attributes multi_cdc_cluster fail_over_ddl_L'
 	# G12
 	# ds_memory_control
-	'row_format tiflash multi_rocks fail_over_ddl_M correctness_for_shared_column_schema partial_index'
+	'update_changefeed_check_config row_format tiflash multi_rocks fail_over_ddl_M correctness_for_shared_column_schema partial_index'
 	# G13
 	'cli_with_auth fail_over_ddl_N maintainer_failover_when_operator cli_missing_keyspace_error changefeed_query_toml_api'
 	# G14
-	'batch_add_table batch_update_to_no_batch fail_over_ddl_O update_changefeed_check_config pause_changefeed_with_long_time_ddl'
+	"batch_add_table batch_update_to_no_batch batch_dml_with_filter fail_over_ddl_O"
 	# G15
 	'split_region changefeed_resume_with_checkpoint_ts autorandom gc_safepoint foreign_key_check old_arch_compatibility bootstrap_retry_after_error'
 )
diff --git a/tests/integration_tests/safe_mode/data/insert.sql b/tests/integration_tests/safe_mode/data/insert.sql
index 2cd4400317..efd67147d8 100644
--- a/tests/integration_tests/safe_mode/data/insert.sql
+++ b/tests/integration_tests/safe_mode/data/insert.sql
@@ -1,3 +1,4 @@
 use `safe_mode`;
 insert into t values(1, "hello");
 insert into t values(2, "world");
+insert into t values(3, "third"), (4, "fourth");
diff --git a/tests/integration_tests/safe_mode/run.sh b/tests/integration_tests/safe_mode/run.sh
index 3d44c450bb..639ac5db39 100755
--- a/tests/integration_tests/safe_mode/run.sh
+++ b/tests/integration_tests/safe_mode/run.sh
@@ -3,47 +3,92 @@
 set -eu
 
 CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
-source $CUR/../_utils/test_prepare
+source "$CUR/../_utils/test_prepare"
 WORK_DIR=$OUT_DIR/$TEST_NAME
-CDC_BINARY=cdc.test
+CDC_BINARY=${CDC_BINARY:-cdc.test}
 SINK_TYPE=$1
 
-function run() {
-	if [ "$SINK_TYPE" != "mysql" ]; then
-		return
+function check_rows() {
+	local actual
+	actual=$(mysql -uroot -h"$DOWN_TIDB_HOST" -P"$DOWN_TIDB_PORT" -N -B -e \
+		"SELECT GROUP_CONCAT(CONCAT(id, ':', a) ORDER BY id) FROM safe_mode.t") || return 1
+	if [ "$actual" != "1:hello2,2:world2,3:multi,4:multi" ]; then
+		echo "Unexpected downstream rows: '$actual'"
+		return 1
 	fi
+}
+export -f check_rows
 
-	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
-
-	start_tidb_cluster --workdir $WORK_DIR
-
-	run_sql_file $CUR/data/create_table.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	run_sql_file $CUR/data/create_table.sql ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-
-	# insert data into upstream but not downstream
-	run_sql_file $CUR/data/insert.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-
-	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+function run_case() {
+	local filter_type=$1
+	local batch_enable=$2
+	local changefeed_id="safe-mode-$filter_type-$batch_enable"
+	local config="$WORK_DIR/$changefeed_id.toml"
+	echo "Testing safe mode: delete filter=$filter_type, batch-dml-enable=$batch_enable"
 
-	case $SINK_TYPE in
-	*) SINK_URI="mysql://normal:123456@127.0.0.1:3306/?safe-mode=true" ;;
-	esac
-	cdc_cli_changefeed create --sink-uri="$SINK_URI"
+	run_sql_file "$CUR/data/create_table.sql" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	run_sql_file "$CUR/data/create_table.sql" "$DOWN_TIDB_HOST" "$DOWN_TIDB_PORT"
+	# Seed only upstream, before the changefeed start timestamp. Every UPDATE
+	# below must create a missing downstream row rather than affect zero rows.
+	run_sql_file "$CUR/data/insert.sql" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	cat >"$config" <<EOF
+[filter]
+rules = ["safe_mode.t"]
+EOF
+	if [ "$filter_type" != "none" ]; then
+		cat >>"$config" <<EOF
 
-	# test update sql can be split into delete + replace at all times in safe mode
-	# otherwise the update sql will have no effect on the downstream and the downstream will have no data.
-	sleep 10
-	run_sql_file $CUR/data/update.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+[[filter.event-filters]]
+matcher = ["safe_mode.t"]
+EOF
+		if [ "$filter_type" = "event" ]; then
+			echo 'ignore-event = ["delete"]' >>"$config"
+		else
+			echo 'ignore-delete-value-expr = "id > 0"' >>"$config"
+		fi
+	fi
+	cdc_cli_changefeed create --changefeed-id="$changefeed_id" --config="$config" \
+		--sink-uri="mysql://normal:123456@$DOWN_TIDB_HOST:$DOWN_TIDB_PORT/?safe-mode=true&batch-dml-enable=$batch_enable"
+	ensure 30 check_changefeed_state "http://$UP_PD_HOST_1:$UP_PD_PORT_1" "$changefeed_id" normal null ""
+	local row_count
+	row_count=$(mysql -uroot -h"$DOWN_TIDB_HOST" -P"$DOWN_TIDB_PORT" -N -B -e "SELECT COUNT(*) FROM safe_mode.t")
+	[ "$row_count" = "0" ]
 
-	run_sql "CREATE TABLE safe_mode.finish_mark (a int primary key);"
-	sleep 30
-	check_table_exists "safe_mode.finish_mark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT} 60
-	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
+	run_sql_file "$CUR/data/update.sql" "$UP_TIDB_HOST" "$UP_TIDB_PORT"
+	if [ "$filter_type" = "none" ]; then
+		run_sql "UPDATE safe_mode.t SET a = 'multi' WHERE id = 3;"
+		run_sql "UPDATE safe_mode.t SET a = 'multi' WHERE id = 4;"
+	else
+		# Exercise multiple UPDATEs in one event on the DELETE filter path.
+		run_sql "BEGIN; UPDATE safe_mode.t SET a = 'multi' WHERE id IN (3, 4); COMMIT;"
+	fi
+	do_retry 60 1 bash -c check_rows
+	check_sync_diff "$WORK_DIR" "$CUR/conf/diff_config.toml"
+	check_changefeed_state "http://$UP_PD_HOST_1:$UP_PD_PORT_1" "$changefeed_id" normal null ""
+	cdc_cli_changefeed remove --changefeed-id="$changefeed_id"
+}
 
-	cleanup_process $CDC_BINARY
+function run() {
+	if [ "$SINK_TYPE" != "mysql" ]; then
+		return
+	fi
+	rm -rf "$WORK_DIR" && mkdir -p "$WORK_DIR"
+	start_tidb_cluster --workdir "$WORK_DIR"
+	export GO_FAILPOINTS=''
+	run_cdc_server --workdir "$WORK_DIR" --binary "$CDC_BINARY"
+	for filter_type in none event expr; do
+		for batch_enable in false true; do
+			run_case "$filter_type" "$batch_enable"
+		done
+	done
+	if grep -Eq '^panic:|\[PANIC\]' "$WORK_DIR/stdout.log" "$WORK_DIR/cdc.log"; then
+		echo "CDC panicked while recovering missing rows in safe mode"
+		exit 1
+	fi
+	cleanup_process "$CDC_BINARY"
 }
 
-trap 'stop_test $WORK_DIR' EXIT
-run $*
-check_logs $WORK_DIR
+trap 'stop_test "$WORK_DIR"' EXIT
+run
+check_logs "$WORK_DIR"
 echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"