diff --git a/downstreamadapter/sink/mysql/causality/conflict_detector.go b/downstreamadapter/sink/mysql/causality/conflict_detector.go
index 29296e2f6f..398cb3e34f 100644
--- a/downstreamadapter/sink/mysql/causality/conflict_detector.go
+++ b/downstreamadapter/sink/mysql/causality/conflict_detector.go
@@ -21,7 +21,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/errors"
-	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/ticdc/pkg/sink/mysql"
 	"github.com/pingcap/ticdc/pkg/util"
 	"github.com/pingcap/ticdc/utils/chann"
 	"github.com/prometheus/client_golang/prometheus"
@@ -61,7 +61,7 @@ func New(
 		resolvedTxnCaches:            make([]txnCache, opt.Count),
 		slots:                        NewSlots(numSlots),
 		notifiedNodes:                chann.NewUnlimitedChannelDefault[func()](),
-		metricConflictDetectDuration: metrics.ConflictDetectDuration.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()),
+		metricConflictDetectDuration: mysql.ConflictDetectDuration.WithLabelValues(changefeedID.Keyspace(), changefeedID.Name()),
 
 		changefeedID: changefeedID,
 	}
@@ -75,7 +75,7 @@ func New(
 
 func (d *ConflictDetector) Run(ctx context.Context) error {
 	defer func() {
-		metrics.ConflictDetectDuration.DeleteLabelValues(d.changefeedID.Keyspace(), d.changefeedID.Name())
+		mysql.ConflictDetectDuration.DeleteLabelValues(d.changefeedID.Keyspace(), d.changefeedID.Name())
 		d.closeCache()
 	}()
 
diff --git a/downstreamadapter/sink/mysql/sink.go b/downstreamadapter/sink/mysql/sink.go
index 2856d5fce4..3f22094246 100644
--- a/downstreamadapter/sink/mysql/sink.go
+++ b/downstreamadapter/sink/mysql/sink.go
@@ -227,18 +227,18 @@ func (s *Sink) runDMLWriter(ctx context.Context, idx int) error {
 	keyspace := s.changefeedID.Keyspace()
 	changefeed := s.changefeedID.Name()
 
-	workerBatchFlushDuration := metrics.WorkerBatchFlushDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-	workerFlushDuration := metrics.WorkerFlushDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-	workerTotalDuration := metrics.WorkerTotalDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-	workerHandledRows := metrics.WorkerHandledRows.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-	workerEventRowCount := metrics.WorkerEventRowCount.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerBatchFlushDuration := mysql.WorkerBatchFlushDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerFlushDuration := mysql.WorkerFlushDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerTotalDuration := mysql.WorkerTotalDuration.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerHandledRows := mysql.WorkerHandledRows.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+	workerEventRowCount := mysql.WorkerEventRowCount.WithLabelValues(keyspace, changefeed, strconv.Itoa(idx))
 
 	defer func() {
-		metrics.WorkerFlushDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-		metrics.WorkerTotalDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-		metrics.WorkerHandledRows.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-		metrics.WorkerBatchFlushDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
-		metrics.WorkerEventRowCount.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerFlushDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerTotalDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerHandledRows.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerBatchFlushDuration.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
+		mysql.WorkerEventRowCount.DeleteLabelValues(keyspace, changefeed, strconv.Itoa(idx))
 	}()
 
 	inputCh := s.conflictDetector.GetOutChByCacheID(idx)
@@ -457,6 +457,7 @@ func (s *Sink) Close() {
 		s.activeActiveSyncStatsCollector.Close()
 	}
 	s.statistics.Close()
+	mysql.DeleteDMLEventRowsAffectedMetrics(s.changefeedID)
 
 	metrics.ChangefeedDownstreamIsTiDBGauge.DeleteLabelValues(s.changefeedID.Keyspace(), s.changefeedID.Name())
 }
diff --git a/pkg/metrics/changefeed.go b/pkg/metrics/changefeed.go
index fa1d54f175..f3058d0cf9 100644
--- a/pkg/metrics/changefeed.go
+++ b/pkg/metrics/changefeed.go
@@ -26,7 +26,7 @@ var (
 			Subsystem: "maintainer",
 			Name:      "checkpoint_ts",
 			Help:      "checkpoint ts of maintainer",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	MaintainerCheckpointTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -34,7 +34,7 @@ var (
 			Subsystem: "maintainer",
 			Name:      "checkpoint_ts_lag",
 			Help:      "checkpoint ts lag of maintainer in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	MaintainerResolvedTsGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -42,14 +42,14 @@ var (
 			Subsystem: "maintainer",
 			Name:      "resolved_ts",
 			Help:      "resolved ts of maintainer",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	MaintainerResolvedTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "maintainer",
 			Name:      "resolved_ts_lag",
 			Help:      "resolved ts lag of maintainer in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	CoordinatorCounter = prometheus.NewCounter(
 		prometheus.CounterOpts{
@@ -65,7 +65,7 @@ var (
 			Subsystem: "changefeed",
 			Name:      "maintainer_counter",
 			Help:      "The counter of changefeed maintainer",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	ChangefeedStatusGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -73,7 +73,7 @@ var (
 			Subsystem: "owner",
 			Name:      "status",
 			Help:      "The status of changefeeds",
-		}, []string{getKeyspaceLabel(), "changefeed", "keyspace_id"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "keyspace_id"})
 
 	// ChangefeedErrorInfoGauge records the current warning or failed reason and its occurrence time
 	// for each changefeed.
@@ -83,7 +83,7 @@ var (
 			Subsystem: "owner",
 			Name:      "changefeed_error_info",
 			Help:      "The current warning or failed reason and occurrence time of changefeeds",
-		}, []string{getKeyspaceLabel(), "changefeed", "state", "error_time", "code", "message"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "state", "error_time", "code", "message"})
 
 	// ChangefeedOperationTimeGauge records a bounded set of recent user initiated
 	// changefeed operation timestamps for the Grafana investigation panel.
@@ -93,7 +93,7 @@ var (
 			Subsystem: "owner",
 			Name:      "changefeed_operation_time",
 			Help:      "Recent user initiated changefeed operation timestamps in Unix milliseconds",
-		}, []string{getKeyspaceLabel(), "changefeed", "operation", "result", "username", "details", "error", "event_id"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "operation", "result", "username", "details", "error", "event_id"})
 
 	ChangefeedCheckpointTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -101,7 +101,7 @@ var (
 			Subsystem: "owner",
 			Name:      "checkpoint_ts_lag",
 			Help:      "changefeed checkpoint ts lag in changefeeds in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed", "keyspace_id"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "keyspace_id"})
 
 	// it's a metrics used in a large number of tcms, we should always keep this metrics
 	ChangefeedCheckpointTsGauge = prometheus.NewGaugeVec(
@@ -110,7 +110,7 @@ var (
 			Subsystem: "owner",
 			Name:      "checkpoint_ts",
 			Help:      "checkpoint ts of changefeeds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// ChangefeedDownstreamInfoGauge is a metric with a constant '1' value,
 	// labeled by the downstream type of each changefeed.
@@ -123,7 +123,7 @@ var (
 			Subsystem: "owner",
 			Name:      "changefeed_downstream_info",
 			Help:      "Downstream type information of changefeeds exposed as labels.",
-		}, []string{getKeyspaceLabel(), "changefeed", "downstream_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "downstream_type"})
 
 	// ChangefeedDownstreamIsTiDBGauge indicates whether the downstream of a
 	// MySQL-compatible sink is confirmed to be TiDB (1 means yes).
@@ -138,7 +138,7 @@ var (
 			Subsystem: "sink",
 			Name:      "changefeed_downstream_is_tidb",
 			Help:      "Whether the downstream of a changefeed is confirmed to be TiDB (1 means yes).",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 )
 
 func DeleteChangefeedCheckpointMetrics(keyspace, changefeed string, keyspaceID uint32) {
diff --git a/pkg/metrics/ddl.go b/pkg/metrics/ddl.go
index 5194e167fe..454fc96059 100644
--- a/pkg/metrics/ddl.go
+++ b/pkg/metrics/ddl.go
@@ -27,7 +27,7 @@ var (
 			Name:      "handle_duration",
 			Help:      "Bucketed histogram of handling time (s) of a ddl.",
 			Buckets:   prometheus.ExponentialBuckets(0.01, 2, 18),
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// ExecDDLHistogram records the execution time of a DDL.
 	ExecDDLHistogram = prometheus.NewHistogramVec(
@@ -37,7 +37,7 @@ var (
 			Name:      "exec_duration",
 			Help:      "Bucketed histogram of processing time (s) of a ddl.",
 			Buckets:   prometheus.ExponentialBuckets(0.01, 2, 18),
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// ExecDDLRunningGauge records the count of running DDL.
 	ExecDDLRunningGauge = prometheus.NewGaugeVec(
@@ -46,7 +46,7 @@ var (
 			Subsystem: "ddl",
 			Name:      "exec_running",
 			Help:      "Total count of running ddl.",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// ExecDDLBlockingGauge records the count of blocking DDL.
 	ExecDDLBlockingGauge = prometheus.NewGaugeVec(
@@ -55,7 +55,7 @@ var (
 			Subsystem: "ddl",
 			Name:      "exec_blocking",
 			Help:      "Total count of blocking ddl.",
-		}, []string{getKeyspaceLabel(), "changefeed", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "mode"})
 
 	// ExecDDLCounter records the execution count of different DDL types
 	ExecDDLCounter = prometheus.NewCounterVec(
@@ -64,7 +64,7 @@ var (
 			Subsystem: "ddl",
 			Name:      "execution",
 			Help:      "Total execution count of different DDL types.",
-		}, []string{getKeyspaceLabel(), "changefeed", "ddl_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "ddl_type"})
 )
 
 func initDDLMetrics(registry *prometheus.Registry) {
diff --git a/pkg/metrics/dispatcher.go b/pkg/metrics/dispatcher.go
index ec2c45bd4f..e9b58e5aba 100644
--- a/pkg/metrics/dispatcher.go
+++ b/pkg/metrics/dispatcher.go
@@ -22,7 +22,7 @@ var (
 			Subsystem: "dispatchermanagermanager",
 			Name:      "event_dispatcher_manager_count",
 			Help:      "The number of event dispatcher managers",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	TableTriggerEventDispatcherGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -30,7 +30,7 @@ var (
 			Subsystem: "dispatchermanager",
 			Name:      "table_trigger_dispatcher_count",
 			Help:      "The number of table dispatchers",
-		}, []string{getKeyspaceLabel(), "changefeed", "event_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "event_type"})
 
 	EventDispatcherGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -38,7 +38,7 @@ var (
 			Subsystem: "dispatchermanager",
 			Name:      "table_dispatcher_count",
 			Help:      "The number of table dispatchers",
-		}, []string{getKeyspaceLabel(), "changefeed", "event_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "event_type"})
 
 	CreateDispatcherDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -47,7 +47,7 @@ var (
 			Name:      "create_dispatcher_duration",
 			Help:      "Bucketed histogram of create dispatcher time (s) for table span.",
 			Buckets:   prometheus.ExponentialBuckets(0.000001, 2, 20), // 1us~524ms
-		}, []string{getKeyspaceLabel(), "changefeed", "event_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "event_type"})
 
 	DispatcherManagerResolvedTsGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -55,7 +55,7 @@ var (
 			Subsystem: "dispatchermanager",
 			Name:      "resolved_ts",
 			Help:      "Resolved ts of event dispatcher manager(changefeed)",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	DispatcherManagerResolvedTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -63,7 +63,7 @@ var (
 			Subsystem: "dispatchermanager",
 			Name:      "resolved_ts_lag",
 			Help:      "Resolved ts lag of event dispatcher manager(changefeed) in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	DispatcherManagerCheckpointTsGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -71,7 +71,7 @@ var (
 			Subsystem: "dispatchermanager",
 			Name:      "checkpoint_ts",
 			Help:      "Checkpoint ts of event dispatcher manager(changefeed)",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	DispatcherManagerCheckpointTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -79,7 +79,7 @@ var (
 			Subsystem: "dispatchermanager",
 			Name:      "checkpoint_ts_lag",
 			Help:      "Checkpoint ts lag of event dispatcher manager(changefeed) in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	DispatcherManagerBlockStatusesChanLenGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -87,7 +87,7 @@ var (
 			Subsystem: "dispatchermanager",
 			Name:      "block_statuses_chan_len",
 			Help:      "length of dispatcher manager block statuses channel",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	HandleDispatcherRequsetCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
@@ -95,7 +95,7 @@ var (
 			Subsystem: "sink",
 			Name:      "handle_dispatcher_request",
 			Help:      "Total count of dispatcher request.",
-		}, []string{getKeyspaceLabel(), "changefeed", "type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 
 	DispatcherReceivedEventCount = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Namespace: "ticdc",
diff --git a/pkg/metrics/dynamic_stream.go b/pkg/metrics/dynamic_stream.go
index 8e5b121617..c1e45c54ab 100644
--- a/pkg/metrics/dynamic_stream.go
+++ b/pkg/metrics/dynamic_stream.go
@@ -21,7 +21,7 @@ var (
 			Namespace: "ticdc",
 			Subsystem: "dynamic_stream",
 			Name:      "memory_usage",
-		}, []string{"module", "type", getKeyspaceLabel(), "area"})
+		}, []string{"module", "type", GetKeyspaceLabel(), "area"})
 	DynamicStreamEventChanSize = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
diff --git a/pkg/metrics/init.go b/pkg/metrics/init.go
index 7583b99155..67d73c86bc 100644
--- a/pkg/metrics/init.go
+++ b/pkg/metrics/init.go
@@ -47,7 +47,8 @@ func InitMetrics(registry *prometheus.Registry) {
 	initDDLMetrics(registry)
 }
 
-func getKeyspaceLabel() string {
+// GetKeyspaceLabel returns the keyspace label name used by TiCDC metrics.
+func GetKeyspaceLabel() string {
 	if kerneltype.IsNextGen() {
 		return "keyspace_name"
 	}
diff --git a/pkg/metrics/log_coordinator.go b/pkg/metrics/log_coordinator.go
index a6e3f26e28..c360570e33 100644
--- a/pkg/metrics/log_coordinator.go
+++ b/pkg/metrics/log_coordinator.go
@@ -22,14 +22,14 @@ var (
 			Subsystem: "owner",
 			Name:      "resolved_ts",
 			Help:      "resolved ts of changefeeds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	ChangefeedResolvedTsLagGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "owner",
 			Name:      "resolved_ts_lag",
 			Help:      "resolved ts lag of changefeeds in seconds",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 )
 
 func initLogCoordinatorMetrics(registry *prometheus.Registry) {
diff --git a/pkg/metrics/maintainer.go b/pkg/metrics/maintainer.go
index b77706e4f4..ad0ebc4235 100644
--- a/pkg/metrics/maintainer.go
+++ b/pkg/metrics/maintainer.go
@@ -23,7 +23,7 @@ var (
 			Name:      "handle_event_duration",
 			Help:      "Bucketed histogram of maintainer handle event time (s).",
 			Buckets:   prometheus.ExponentialBuckets(0.01 /* 10 ms */, 2, 18),
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	MaintainerEventChLenGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -31,7 +31,7 @@ var (
 			Subsystem: "maintainer",
 			Name:      "event_ch_len",
 			Help:      "length of maintainer event channel",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	OperatorCount = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -39,7 +39,7 @@ var (
 			Subsystem: "maintainer",
 			Name:      "created_count",
 			Help:      "number of created operators",
-		}, []string{getKeyspaceLabel(), "changefeed", "type", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type", "mode"})
 
 	TotalOperatorCount = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -47,7 +47,7 @@ var (
 			Subsystem: "maintainer",
 			Name:      "total_operator_count",
 			Help:      "number of total operators",
-		}, []string{getKeyspaceLabel(), "changefeed", "type", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type", "mode"})
 
 	OperatorDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -56,7 +56,7 @@ var (
 			Name:      "finish_operators_duration_seconds",
 			Help:      "Bucketed histogram of processing time (s) of finished operator.",
 			Buckets:   []float64{0.5, 1, 2, 4, 8, 16, 20, 40, 60, 90, 120, 180, 240, 300, 480, 600, 720, 900, 1200, 1800, 3600},
-		}, []string{getKeyspaceLabel(), "changefeed", "type", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type", "mode"})
 )
 
 func initMaintainerMetrics(registry *prometheus.Registry) {
diff --git a/pkg/metrics/redo.go b/pkg/metrics/redo.go
index 5400b3259f..100c894449 100644
--- a/pkg/metrics/redo.go
+++ b/pkg/metrics/redo.go
@@ -29,7 +29,7 @@ var (
 		Subsystem: subsystem,
 		Name:      "resolved_ts",
 		Help:      "Resolved ts persisted by redo meta",
-	}, []string{getKeyspaceLabel(), "changefeed"})
+	}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// RedoCheckpointTsGauge records the checkpoint ts persisted by redo meta.
 	RedoCheckpointTsGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
@@ -37,7 +37,7 @@ var (
 		Subsystem: subsystem,
 		Name:      "checkpoint_ts",
 		Help:      "Checkpoint ts persisted by redo meta",
-	}, []string{getKeyspaceLabel(), "changefeed"})
+	}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// RedoWriteBytesGauge records the total number of bytes written to redo log.
 	RedoWriteBytesGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
@@ -45,7 +45,7 @@ var (
 		Subsystem: subsystem,
 		Name:      "write_bytes_total",
 		Help:      "Total number of bytes redo log written",
-	}, []string{getKeyspaceLabel(), "changefeed", "type"})
+	}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 
 	// RedoFsyncDurationHistogram records the latency distributions of fsync called by redo writer.
 	RedoFsyncDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
@@ -54,7 +54,7 @@ var (
 		Name:      "fsync_duration_seconds",
 		Help:      "The latency distributions of fsync called by redo writer",
 		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 16),
-	}, []string{getKeyspaceLabel(), "changefeed", "type"})
+	}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 
 	// RedoFlushAllDurationHistogram records the latency distributions of flushAll
 	// called by redo writer.
@@ -64,7 +64,7 @@ var (
 		Name:      "flush_all_duration_seconds",
 		Help:      "The latency distributions of flushall called by redo writer",
 		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 16),
-	}, []string{getKeyspaceLabel(), "changefeed", "type"})
+	}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 
 	// RedoTotalRowsCountGauge records the total number of rows written to redo log.
 	RedoTotalRowsCountGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
@@ -72,7 +72,7 @@ var (
 		Subsystem: subsystem,
 		Name:      "total_rows_count",
 		Help:      "The total count of rows that are processed by redo writer",
-	}, []string{getKeyspaceLabel(), "changefeed", "type"})
+	}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 
 	// RedoWriteLogDurationHistogram records the latency distributions of writeLog.
 	RedoWriteLogDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
@@ -81,7 +81,7 @@ var (
 		Name:      "write_log_duration_seconds",
 		Help:      "The latency distributions of writeLog called by redo sink",
 		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 16),
-	}, []string{getKeyspaceLabel(), "changefeed", "type"})
+	}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 
 	// RedoFlushLogDurationHistogram records the latency distributions of flushLog.
 	RedoFlushLogDurationHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
@@ -90,7 +90,7 @@ var (
 		Name:      "flush_log_duration_seconds",
 		Help:      "The latency distributions of flushLog called by redo sink",
 		Buckets:   prometheus.ExponentialBuckets(0.001, 2.0, 16),
-	}, []string{getKeyspaceLabel(), "changefeed", "type"})
+	}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 
 	// RedoWorkerBusyRatio records the busy ratio of redo sink worker.
 	RedoWorkerBusyRatio = prometheus.NewCounterVec(
@@ -99,7 +99,7 @@ var (
 			Subsystem: subsystem,
 			Name:      "worker_busy_ratio",
 			Help:      "Busy ratio for redo sink worker.",
-		}, []string{getKeyspaceLabel(), "changefeed", "type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "type"})
 )
 
 func initRedoMetrics(registry *prometheus.Registry) {
diff --git a/pkg/metrics/scheduler.go b/pkg/metrics/scheduler.go
index 65746ec077..b17a1d291f 100644
--- a/pkg/metrics/scheduler.go
+++ b/pkg/metrics/scheduler.go
@@ -24,7 +24,7 @@ var (
 			Subsystem: "scheduler",
 			Name:      "task",
 			Help:      "The total number of scheduler tasks",
-		}, []string{getKeyspaceLabel(), "changefeed", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "mode"})
 
 	SpanCountGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -32,70 +32,70 @@ var (
 			Subsystem: "scheduler",
 			Name:      "span_count",
 			Help:      "The total number of spans",
-		}, []string{getKeyspaceLabel(), "changefeed", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "mode"})
 	TableCountGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "table_count",
 			Help:      "The total number of tables",
-		}, []string{getKeyspaceLabel(), "changefeed", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "mode"})
 	TableStateGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "table_replication_state",
 			Help:      "The total number of tables in different replication states",
-		}, []string{getKeyspaceLabel(), "changefeed", "state", "mode"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "state", "mode"})
 	SlowestTableIDGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_id",
 			Help:      "The table ID of the slowest table",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	SlowestTableCheckpointTsGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_checkpoint_ts",
 			Help:      "The checkpoint ts of the slowest table",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	SlowestTableResolvedTsGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_resolved_ts",
 			Help:      "The resolved ts of the slowest table",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	SlowestTableStageCheckpointTsGaugeVec = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_stage_checkpoint_ts",
 			Help:      "Checkpoint ts of each stage of the slowest table",
-		}, []string{getKeyspaceLabel(), "changefeed", "stage"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "stage"})
 	SlowestTableStageResolvedTsGaugeVec = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_stage_resolved_ts",
 			Help:      "Resolved ts of each stage of the slowest table",
-		}, []string{getKeyspaceLabel(), "changefeed", "stage"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "stage"})
 	SlowestTableStageCheckpointTsLagGaugeVec = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_stage_checkpoint_ts_lag",
 			Help:      "Checkpoint ts lag of each stage of the slowest table",
-		}, []string{getKeyspaceLabel(), "changefeed", "stage"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "stage"})
 	SlowestTableStageResolvedTsLagGaugeVec = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_stage_resolved_ts_lag",
 			Help:      "Resolved ts lag of each stage of the slowest table",
-		}, []string{getKeyspaceLabel(), "changefeed", "stage"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "stage"})
 	SlowestTableStageCheckpointTsLagHistogramVec = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Namespace: "ticdc",
@@ -103,7 +103,7 @@ var (
 			Name:      "slow_table_stage_checkpoint_ts_lag_histogram",
 			Help:      "Histogram of the slowest table checkpoint ts lag of each stage",
 			Buckets:   prometheus.LinearBuckets(0.5, 0.5, 36),
-		}, []string{getKeyspaceLabel(), "changefeed", "stage"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "stage"})
 	SlowestTableStageResolvedTsLagHistogramVec = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
 			Namespace: "ticdc",
@@ -111,14 +111,14 @@ var (
 			Name:      "slow_table_stage_resolved_ts_lag_histogram",
 			Help:      "Histogram of the slowest table resolved ts lag of each stage",
 			Buckets:   prometheus.LinearBuckets(0.5, 0.5, 36),
-		}, []string{getKeyspaceLabel(), "changefeed", "stage"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "stage"})
 	SlowestTableRegionGaugeVec = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_region_count",
 			Help:      "The number of regions captured by the slowest table",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	SlowestTablePullerResolvedTs = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -126,14 +126,14 @@ var (
 			Subsystem: "scheduler",
 			Name:      "slow_table_puller_resolved_ts",
 			Help:      "Puller Slowest ResolvedTs",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	SlowestTablePullerResolvedTsLag = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "scheduler",
 			Name:      "slow_table_puller_resolved_ts_lag",
 			Help:      "Puller Slowest ResolvedTs lag",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	// checker related
 	SplitSpanCheckDuration = prometheus.NewHistogramVec(
@@ -143,7 +143,7 @@ var (
 			Name:      "split_span_check_duration",
 			Help:      "Bucketed histogram of split span check time (s).",
 			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{getKeyspaceLabel(), "changefeed", "group_id"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "group_id"})
 )
 
 func initSchedulerMetrics(registry *prometheus.Registry) {
diff --git a/pkg/metrics/sink.go b/pkg/metrics/sink.go
index a45582e6c2..925cb718ad 100644
--- a/pkg/metrics/sink.go
+++ b/pkg/metrics/sink.go
@@ -28,7 +28,7 @@ var (
 			Name:      "batch_row_count",
 			Help:      "Row count number for a given batch.",
 			Buckets:   prometheus.ExponentialBuckets(1, 2, 18),
-		}, []string{getKeyspaceLabel(), "changefeed", "type", "keyspace_id"}) // type is for `sinkType`
+		}, []string{GetKeyspaceLabel(), "changefeed", "type", "keyspace_id"}) // type is for `sinkType`
 
 	// ExecBatchWriteBytesHistogram records bytes written for each batch.
 	ExecBatchWriteBytesHistogram = prometheus.NewHistogramVec(
@@ -38,7 +38,7 @@ var (
 			Name:      "batch_write_bytes",
 			Help:      "Bytes number for a given batch.",
 			Buckets:   prometheus.ExponentialBuckets(1024, 2, 18), // 1KB~128MB
-		}, []string{getKeyspaceLabel(), "changefeed", "type"}) // type is for `sinkType`
+		}, []string{GetKeyspaceLabel(), "changefeed", "type"}) // type is for `sinkType`
 
 	// ExecWriteBytesGauge records the total number of bytes written by sink.
 	TotalWriteBytesCounter = prometheus.NewCounterVec(
@@ -47,7 +47,7 @@ var (
 			Subsystem: "sink",
 			Name:      "write_bytes_total",
 			Help:      "Total number of bytes written by sink",
-		}, []string{getKeyspaceLabel(), "changefeed", "type"}) // type is for `sinkType`
+		}, []string{GetKeyspaceLabel(), "changefeed", "type"}) // type is for `sinkType`
 
 	EventSizeHistogram = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -56,7 +56,7 @@ var (
 			Name:      "event_size",
 			Help:      "The size of changed events (in bytes).",
 			Buckets:   prometheus.ExponentialBuckets(0.01, 2, 30), // 0~32M
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	ExecDMLEventCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
@@ -64,23 +64,8 @@ var (
 			Subsystem: "sink",
 			Name:      "dml_event_count",
 			Help:      "Total count of DML events.",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
-	ExecDMLEventRowsAffectedCounter = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "dml_event_affected_row_count",
-			Help:      "Total count of affected rows.",
-		}, []string{getKeyspaceLabel(), "changefeed", "count_type", "row_type"})
-
-	ActiveActiveConflictSkipRowsCounter = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "active_active_conflict_skip_rows_total",
-			Help:      "Total number of rows skipped due to last-write-wins conflict resolution in TiDB active-active replication.",
-		}, []string{getKeyspaceLabel(), "changefeed"})
 	// ExecutionErrorCounter is the counter of execution errors.
 	ExecutionErrorCounter = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
@@ -88,21 +73,11 @@ var (
 			Subsystem: "sink",
 			Name:      "execution_error",
 			Help:      "Total count of execution errors.",
-		}, []string{getKeyspaceLabel(), "changefeed", "event_type"})
+		}, []string{GetKeyspaceLabel(), "changefeed", "event_type"})
 )
 
 // ---------- Metrics for txn sink and backends. ---------- //
 var (
-	// ConflictDetectDuration records the duration of detecting conflict.
-	ConflictDetectDuration = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "txn_conflict_detect_duration",
-			Help:      "Bucketed histogram of conflict detect time (s) for single DML statement.",
-			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{getKeyspaceLabel(), "changefeed"})
-
 	// QueueDuration = ConflictDetectDuration + (queue time in txn workers).
 	QueueDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -111,50 +86,7 @@ var (
 			Name:      "txn_queue_duration",
 			Help:      "Bucketed histogram of queue time (s) for single DML statement.",
 			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{getKeyspaceLabel(), "changefeed"})
-
-	WorkerBatchFlushDuration = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "txn_worker_batch_flush_duration",
-			Help:      "Flush duration (s) for txn worker.",
-			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{getKeyspaceLabel(), "changefeed", "id"})
-
-	WorkerFlushDuration = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "txn_worker_flush_duration",
-			Help:      "Flush duration (s) for txn worker.",
-			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{getKeyspaceLabel(), "changefeed", "id"})
-
-	WorkerTotalDuration = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "txn_worker_total_duration",
-			Help:      "total duration (s) for txn worker.",
-			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{getKeyspaceLabel(), "changefeed", "id"})
-
-	WorkerHandledRows = prometheus.NewCounterVec(
-		prometheus.CounterOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "txn_worker_handled_rows",
-			Help:      "Busy ratio (X ms in 1s) for all workers.",
-		}, []string{getKeyspaceLabel(), "changefeed", "id"})
-	WorkerEventRowCount = prometheus.NewHistogramVec(
-		prometheus.HistogramOpts{
-			Namespace: "ticdc",
-			Subsystem: "sink",
-			Name:      "txn_worker_event_row_count",
-			Help:      "Row count number for a single DML event handled by txn sink worker.",
-			Buckets:   prometheus.ExponentialBuckets(1, 2, 12), // 1~2048
-		}, []string{getKeyspaceLabel(), "changefeed", "id"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	SinkDMLBatchCommit = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -163,7 +95,7 @@ var (
 			Name:      "txn_sink_dml_batch_commit",
 			Help:      "Duration of committing a DML batch",
 			Buckets:   prometheus.ExponentialBuckets(0.01, 2, 18), // 10ms~1310s
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	SinkDMLBatchCallback = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -172,7 +104,7 @@ var (
 			Name:      "txn_sink_dml_batch_callback",
 			Help:      "Duration of execuing a batch of callbacks",
 			Buckets:   prometheus.ExponentialBuckets(0.01, 2, 18), // 10ms~1300s
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	PrepareStatementErrors = prometheus.NewCounterVec(
 		prometheus.CounterOpts{
@@ -180,7 +112,7 @@ var (
 			Subsystem: "sink",
 			Name:      "txn_prepare_statement_errors",
 			Help:      "Prepare statement errors",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 )
 
 // ---------- Metrics for kafka sink and backends. ---------- //
@@ -193,7 +125,7 @@ var (
 			Name:      "mq_worker_send_message_duration",
 			Help:      "Send Message duration(s) for MQ worker.",
 			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	// WorkerBatchSize record the size of each batched messages.
 	WorkerBatchSize = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -202,7 +134,7 @@ var (
 			Name:      "mq_worker_batch_size",
 			Help:      "Batch size for MQ worker.",
 			Buckets:   prometheus.ExponentialBuckets(4, 2, 10), // 4 ~ 2048
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 	// WorkerBatchDuration record the time duration cost on batch messages.
 	WorkerBatchDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -211,7 +143,7 @@ var (
 			Name:      "mq_worker_batch_duration",
 			Help:      "Batch duration for MQ worker.",
 			Buckets:   prometheus.ExponentialBuckets(0.004, 2, 10), // 4ms ~ 2s
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	CheckpointTsMessageDuration = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -220,7 +152,7 @@ var (
 			Name:      "mq_checkpoint_ts_message_duration",
 			Help:      "Duration of sending checkpoint ts message.",
 			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 
 	CheckpointTsMessageCount = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
@@ -228,7 +160,7 @@ var (
 			Subsystem: "sink",
 			Name:      "mq_checkpoint_ts_message_count",
 			Help:      "Number of checkpoint ts messages sent.",
-		}, []string{getKeyspaceLabel(), "changefeed"})
+		}, []string{GetKeyspaceLabel(), "changefeed"})
 )
 
 // InitMetrics registers all metrics in this file.
@@ -239,18 +171,10 @@ func initSinkMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(TotalWriteBytesCounter)
 	registry.MustRegister(EventSizeHistogram)
 	registry.MustRegister(ExecDMLEventCounter)
-	registry.MustRegister(ExecDMLEventRowsAffectedCounter)
-	registry.MustRegister(ActiveActiveConflictSkipRowsCounter)
 	registry.MustRegister(ExecutionErrorCounter)
 
 	// txn sink metrics
-	registry.MustRegister(ConflictDetectDuration)
 	registry.MustRegister(QueueDuration)
-	registry.MustRegister(WorkerFlushDuration)
-	registry.MustRegister(WorkerBatchFlushDuration)
-	registry.MustRegister(WorkerTotalDuration)
-	registry.MustRegister(WorkerHandledRows)
-	registry.MustRegister(WorkerEventRowCount)
 	registry.MustRegister(SinkDMLBatchCommit)
 	registry.MustRegister(SinkDMLBatchCallback)
 	registry.MustRegister(PrepareStatementErrors)
diff --git a/pkg/metrics/statistics.go b/pkg/metrics/statistics.go
index f1815a63e5..af7f4ab1cf 100644
--- a/pkg/metrics/statistics.go
+++ b/pkg/metrics/statistics.go
@@ -14,8 +14,6 @@
 package metrics
 
 import (
-	"fmt"
-	"strings"
 	"sync"
 	"time"
 
@@ -30,11 +28,10 @@ func NewStatistics(
 	sinkType string,
 ) *Statistics {
 	statistics := &Statistics{
-		sinkType:        sinkType,
-		changefeedID:    changefeed,
-		keyspaceID:      FormatKeyspaceID(keyspaceID),
-		ddlTypes:        sync.Map{},
-		rowsAffectedMap: sync.Map{},
+		sinkType:     sinkType,
+		changefeedID: changefeed,
+		keyspaceID:   FormatKeyspaceID(keyspaceID),
+		ddlTypes:     sync.Map{},
 	}
 
 	keyspace := changefeed.Keyspace()
@@ -54,11 +51,10 @@ func NewStatistics(
 // Statistics maintains some status and metrics of the Sink
 // Note: All methods of Statistics should be thread-safe.
 type Statistics struct {
-	sinkType        string
-	changefeedID    common.ChangeFeedID
-	keyspaceID      string
-	ddlTypes        sync.Map
-	rowsAffectedMap sync.Map
+	sinkType     string
+	changefeedID common.ChangeFeedID
+	keyspaceID   string
+	ddlTypes     sync.Map
 
 	// metricExecDDLHis records each DDL execution time duration.
 	metricExecDDLHis prometheus.Observer
@@ -116,30 +112,6 @@ func (b *Statistics) RecordDDLExecution(executor func() (string, error)) error {
 	return nil
 }
 
-func (b *Statistics) RecordTotalRowsAffected(actualRowsAffected, expectedRowsAffected int64) {
-	b.getRowsAffected("actual", "total").Add(float64(actualRowsAffected))
-	b.getRowsAffected("expected", "total").Add(float64(expectedRowsAffected))
-}
-
-func (b *Statistics) RecordRowsAffected(rowsAffected int64, rowType common.RowType) {
-	b.getRowsAffected("actual", rowType.String()).Add(float64(rowsAffected))
-	b.getRowsAffected("expected", rowType.String()).Add(1)
-	b.RecordTotalRowsAffected(rowsAffected, 1)
-}
-
-func (b *Statistics) getRowsAffected(countType, rowType string) prometheus.Counter {
-	key := fmt.Sprintf("%s-%s", countType, rowType)
-	counter, loaded := b.rowsAffectedMap.Load(key)
-	if !loaded {
-		keyspace := b.changefeedID.Keyspace()
-		changefeedID := b.changefeedID.Name()
-		counter := ExecDMLEventRowsAffectedCounter.WithLabelValues(keyspace, changefeedID, countType, rowType)
-		b.rowsAffectedMap.Store(key, counter)
-		return counter
-	}
-	return counter.(prometheus.Counter)
-}
-
 // Close release some internal resources.
 func (b *Statistics) Close() {
 	keyspace := b.changefeedID.Keyspace()
@@ -155,13 +127,6 @@ func (b *Statistics) Close() {
 		ExecDDLCounter.DeleteLabelValues(keyspace, changefeedID, ddlType)
 		return true
 	})
-	b.rowsAffectedMap.Range(func(key, value any) bool {
-		countTypeAndRowType := key.(string)
-		splitTypes := strings.Split(countTypeAndRowType, "-")
-		countType, rowType := splitTypes[0], splitTypes[1]
-		ExecDMLEventRowsAffectedCounter.DeleteLabelValues(keyspace, changefeedID, countType, rowType)
-		return true
-	})
 	TotalWriteBytesCounter.DeleteLabelValues(keyspace, changefeedID, b.sinkType)
 	ExecDMLEventCounter.DeleteLabelValues(keyspace, changefeedID)
 }
diff --git a/pkg/sink/codec/encoder_group.go b/pkg/sink/codec/encoder_group.go
index eeeca21a20..6e38552658 100644
--- a/pkg/sink/codec/encoder_group.go
+++ b/pkg/sink/codec/encoder_group.go
@@ -27,6 +27,7 @@ import (
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
 	"github.com/pingcap/ticdc/pkg/sink/kafka/claimcheck"
 	"github.com/pingcap/ticdc/pkg/util"
+	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
@@ -133,6 +134,9 @@ func (g *encoderGroup) Run(ctx context.Context) error {
 			return g.runEncoder(ctx, idx)
 		})
 	}
+	eg.Go(func() error {
+		return g.collectMetrics(ctx)
+	})
 
 	if g.bootstrapWorker != nil {
 		eg.Go(func() error {
@@ -143,10 +147,12 @@ func (g *encoderGroup) Run(ctx context.Context) error {
 	return eg.Wait()
 }
 
-func (g *encoderGroup) runEncoder(ctx context.Context, idx int) error {
-	inputCh := g.inputCh[idx]
-	metric := encoderGroupInputChanSizeGauge.
-		WithLabelValues(g.changefeedID.Keyspace(), g.changefeedID.Name(), strconv.Itoa(idx))
+func (g *encoderGroup) collectMetrics(ctx context.Context) error {
+	inputMetrics := make([]prometheus.Gauge, len(g.inputCh))
+	for idx := range g.inputCh {
+		inputMetrics[idx] = encoderGroupInputChanSizeGauge.WithLabelValues(g.changefeedID.Keyspace(), g.changefeedID.Name(), strconv.Itoa(idx))
+	}
+	outputMetric := encoderGroupOutputChanSizeGauge.WithLabelValues(g.changefeedID.Keyspace(), g.changefeedID.Name())
 	ticker := time.NewTicker(defaultMetricInterval)
 	defer ticker.Stop()
 	for {
@@ -154,7 +160,20 @@ func (g *encoderGroup) runEncoder(ctx context.Context, idx int) error {
 		case <-ctx.Done():
 			return nil
 		case <-ticker.C:
-			metric.Set(float64(len(inputCh)))
+			for idx, inputCh := range g.inputCh {
+				inputMetrics[idx].Set(float64(len(inputCh)))
+			}
+			outputMetric.Set(float64(len(g.outputCh)))
+		}
+	}
+}
+
+func (g *encoderGroup) runEncoder(ctx context.Context, idx int) error {
+	inputCh := g.inputCh[idx]
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
 		case future := <-inputCh:
 			for _, event := range future.events {
 				err := g.rowEventEncoders[idx].AppendRowChangedEvent(ctx, future.Key.Topic, event)
@@ -209,7 +228,10 @@ func (g *encoderGroup) Output() <-chan *future {
 }
 
 func (g *encoderGroup) cleanMetrics() {
-	encoderGroupInputChanSizeGauge.DeleteLabelValues(g.changefeedID.Keyspace(), g.changefeedID.Name())
+	for idx := range g.inputCh {
+		encoderGroupInputChanSizeGauge.DeleteLabelValues(g.changefeedID.Keyspace(), g.changefeedID.Name(), strconv.Itoa(idx))
+	}
+	encoderGroupOutputChanSizeGauge.DeleteLabelValues(g.changefeedID.Keyspace(), g.changefeedID.Name())
 	common.CleanMetrics(g.changefeedID)
 }
 
diff --git a/pkg/sink/codec/metrics.go b/pkg/sink/codec/metrics.go
index 83aa04fce7..230d27e5ef 100644
--- a/pkg/sink/codec/metrics.go
+++ b/pkg/sink/codec/metrics.go
@@ -26,8 +26,8 @@ var (
 			Name:      "encoder_group_input_chan_size",
 			Help:      "The size of input channel of encoder group",
 		}, []string{"namespace", "changefeed", "index"})
-	// EncoderGroupOutputChanSizeGauge tracks the size of output channel of encoder group
-	EncoderGroupOutputChanSizeGauge = prometheus.NewGaugeVec(
+	// encoderGroupOutputChanSizeGauge tracks the size of output channel of encoder group
+	encoderGroupOutputChanSizeGauge = prometheus.NewGaugeVec(
 		prometheus.GaugeOpts{
 			Namespace: "ticdc",
 			Subsystem: "sink",
@@ -39,6 +39,6 @@ var (
 // InitMetrics registers all metrics in this file
 func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(encoderGroupInputChanSizeGauge)
-	registry.MustRegister(EncoderGroupOutputChanSizeGauge)
+	registry.MustRegister(encoderGroupOutputChanSizeGauge)
 	common.InitMetrics(registry)
 }
diff --git a/pkg/sink/mysql/metrics.go b/pkg/sink/mysql/metrics.go
new file mode 100644
index 0000000000..40a33d9d55
--- /dev/null
+++ b/pkg/sink/mysql/metrics.go
@@ -0,0 +1,114 @@
+// Copyright 2026 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,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mysql
+
+import (
+	"github.com/pingcap/ticdc/pkg/common"
+	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+var (
+	execDMLEventRowsAffectedCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "dml_event_affected_row_count",
+			Help:      "Total count of affected rows.",
+		}, []string{metrics.GetKeyspaceLabel(), "changefeed", "count_type", "row_type"},
+	)
+
+	activeActiveConflictSkipRowsCounter = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "active_active_conflict_skip_rows_total",
+			Help:      "Total number of rows skipped due to last-write-wins conflict resolution in TiDB active-active replication.",
+		}, []string{metrics.GetKeyspaceLabel(), "changefeed"})
+
+	// ConflictDetectDuration records the duration of detecting conflict.
+	ConflictDetectDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "txn_conflict_detect_duration",
+			Help:      "Bucketed histogram of conflict detect time (s) for single DML statement.",
+			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
+		}, []string{metrics.GetKeyspaceLabel(), "changefeed"})
+
+	WorkerBatchFlushDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "txn_worker_batch_flush_duration",
+			Help:      "Flush duration (s) for txn worker.",
+			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
+		}, []string{metrics.GetKeyspaceLabel(), "changefeed", "id"})
+
+	WorkerFlushDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "txn_worker_flush_duration",
+			Help:      "Flush duration (s) for txn worker.",
+			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
+		}, []string{metrics.GetKeyspaceLabel(), "changefeed", "id"})
+
+	WorkerTotalDuration = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "txn_worker_total_duration",
+			Help:      "total duration (s) for txn worker.",
+			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 20), // 1ms~524s
+		}, []string{metrics.GetKeyspaceLabel(), "changefeed", "id"})
+
+	WorkerHandledRows = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "txn_worker_handled_rows",
+			Help:      "Busy ratio (X ms in 1s) for all workers.",
+		}, []string{metrics.GetKeyspaceLabel(), "changefeed", "id"})
+
+	WorkerEventRowCount = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
+			Namespace: "ticdc",
+			Subsystem: "sink",
+			Name:      "txn_worker_event_row_count",
+			Help:      "Row count number for a single DML event handled by txn sink worker.",
+			Buckets:   prometheus.ExponentialBuckets(1, 2, 12), // 1~2048
+		}, []string{metrics.GetKeyspaceLabel(), "changefeed", "id"})
+)
+
+// InitMetrics registers MySQL sink metrics.
+func InitMetrics(registry *prometheus.Registry) {
+	registry.MustRegister(execDMLEventRowsAffectedCounter)
+	registry.MustRegister(activeActiveConflictSkipRowsCounter)
+	registry.MustRegister(ConflictDetectDuration)
+	registry.MustRegister(WorkerBatchFlushDuration)
+	registry.MustRegister(WorkerFlushDuration)
+	registry.MustRegister(WorkerTotalDuration)
+	registry.MustRegister(WorkerHandledRows)
+	registry.MustRegister(WorkerEventRowCount)
+}
+
+// DeleteDMLEventRowsAffectedMetrics deletes affected-row metric series for a MySQL sink.
+func DeleteDMLEventRowsAffectedMetrics(changefeedID common.ChangeFeedID) {
+	execDMLEventRowsAffectedCounter.DeletePartialMatch(prometheus.Labels{
+		metrics.GetKeyspaceLabel(): changefeedID.Keyspace(),
+		"changefeed":               changefeedID.Name(),
+	})
+}
diff --git a/pkg/sink/mysql/metrics_test.go b/pkg/sink/mysql/metrics_test.go
new file mode 100644
index 0000000000..ac1f8cc16a
--- /dev/null
+++ b/pkg/sink/mysql/metrics_test.go
@@ -0,0 +1,53 @@
+// Copyright 2026 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,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package mysql
+
+import (
+	"testing"
+
+	"github.com/pingcap/ticdc/pkg/common"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDMLEventRowsAffectedMetrics(t *testing.T) {
+	execDMLEventRowsAffectedCounter.Reset()
+	t.Cleanup(execDMLEventRowsAffectedCounter.Reset)
+
+	changefeedID := common.NewChangefeedID4Test("test-keyspace", "rows-affected")
+	writer := &Writer{ChangefeedID: changefeedID}
+	writer.recordRowsAffected(2, common.RowTypeInsert)
+	writer.recordTotalRowsAffected(3, 2)
+
+	keyspace := changefeedID.Keyspace()
+	changefeed := changefeedID.Name()
+	require.Equal(t, float64(2), testutil.ToFloat64(
+		execDMLEventRowsAffectedCounter.WithLabelValues(keyspace, changefeed, "actual", "insert")))
+	require.Equal(t, float64(1), testutil.ToFloat64(
+		execDMLEventRowsAffectedCounter.WithLabelValues(keyspace, changefeed, "expected", "insert")))
+	require.Equal(t, float64(5), testutil.ToFloat64(
+		execDMLEventRowsAffectedCounter.WithLabelValues(keyspace, changefeed, "actual", "total")))
+	require.Equal(t, float64(3), testutil.ToFloat64(
+		execDMLEventRowsAffectedCounter.WithLabelValues(keyspace, changefeed, "expected", "total")))
+
+	otherChangefeedID := common.NewChangefeedID4Test("test-keyspace", "other-changefeed")
+	otherWriter := &Writer{ChangefeedID: otherChangefeedID}
+	otherWriter.recordRowsAffected(4, common.RowTypeInsert)
+
+	DeleteDMLEventRowsAffectedMetrics(changefeedID)
+	require.Equal(t, 4, testutil.CollectAndCount(execDMLEventRowsAffectedCounter))
+	require.Equal(t, float64(4), testutil.ToFloat64(execDMLEventRowsAffectedCounter.WithLabelValues(
+		otherChangefeedID.Keyspace(), otherChangefeedID.Name(), "actual", "insert")))
+}
diff --git a/pkg/sink/mysql/mysql_writer.go b/pkg/sink/mysql/mysql_writer.go
index 4c38fadaab..a3ef8e75cb 100644
--- a/pkg/sink/mysql/mysql_writer.go
+++ b/pkg/sink/mysql/mysql_writer.go
@@ -26,6 +26,7 @@ import (
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 )
 
@@ -64,7 +65,8 @@ type Writer struct {
 	// implement stmtCache to improve performance, especially when the downstream is TiDB
 	stmtCache *lru.Cache
 
-	statistics *metrics.Statistics
+	statistics           *metrics.Statistics
+	rowsAffectedCounters sync.Map
 
 	// activeActiveSyncStatsCollector accumulates conflict statistics from TiDB session
 	// variable @@tidb_cdc_active_active_sync_stats. It is shared across all DML writers
@@ -305,3 +307,31 @@ func (w *Writer) Close() {
 	}
 	w.dmlSession.close(w)
 }
+
+type rowsAffectedLabels struct {
+	countType string
+	rowType   string
+}
+
+func (w *Writer) recordTotalRowsAffected(actualRowsAffected, expectedRowsAffected int64) {
+	w.getRowsAffectedCounter("actual", "total").Add(float64(actualRowsAffected))
+	w.getRowsAffectedCounter("expected", "total").Add(float64(expectedRowsAffected))
+}
+
+func (w *Writer) recordRowsAffected(rowsAffected int64, rowType common.RowType) {
+	w.getRowsAffectedCounter("actual", rowType.String()).Add(float64(rowsAffected))
+	w.getRowsAffectedCounter("expected", rowType.String()).Add(1)
+	w.recordTotalRowsAffected(rowsAffected, 1)
+}
+
+func (w *Writer) getRowsAffectedCounter(countType, rowType string) prometheus.Counter {
+	labels := rowsAffectedLabels{countType: countType, rowType: rowType}
+	counter, loaded := w.rowsAffectedCounters.Load(labels)
+	if !loaded {
+		counter := execDMLEventRowsAffectedCounter.WithLabelValues(
+			w.ChangefeedID.Keyspace(), w.ChangefeedID.Name(), countType, rowType)
+		w.rowsAffectedCounters.Store(labels, counter)
+		return counter
+	}
+	return counter.(prometheus.Counter)
+}
diff --git a/pkg/sink/mysql/mysql_writer_dml_exec.go b/pkg/sink/mysql/mysql_writer_dml_exec.go
index 4d0645bae7..c60d43dfa9 100644
--- a/pkg/sink/mysql/mysql_writer_dml_exec.go
+++ b/pkg/sink/mysql/mysql_writer_dml_exec.go
@@ -172,7 +172,7 @@ func (w *Writer) sequenceExecute(
 		if rowsAffected, err := res.RowsAffected(); err != nil {
 			log.Warn("get rows affected rows failed", zap.Error(err))
 		} else {
-			w.statistics.RecordRowsAffected(rowsAffected, dmls.rowTypes[i])
+			w.recordRowsAffected(rowsAffected, dmls.rowTypes[i])
 		}
 		cancelFunc()
 	}
@@ -214,7 +214,7 @@ func (w *Writer) multiStmtExecute(
 	if rowsAffected, err := res.RowsAffected(); err != nil {
 		log.Warn("get rows affected rows failed", zap.Error(err))
 	} else {
-		w.statistics.RecordTotalRowsAffected(rowsAffected, int64(len(dmls.sqls)))
+		w.recordTotalRowsAffected(rowsAffected, int64(len(dmls.sqls)))
 	}
 	return nil
 }
diff --git a/pkg/sink/mysql/mysql_writer_for_active_active_sync_stats.go b/pkg/sink/mysql/mysql_writer_for_active_active_sync_stats.go
index 072a9082ff..6f2f179fbd 100644
--- a/pkg/sink/mysql/mysql_writer_for_active_active_sync_stats.go
+++ b/pkg/sink/mysql/mysql_writer_for_active_active_sync_stats.go
@@ -23,7 +23,6 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/errors"
-	"github.com/pingcap/ticdc/pkg/metrics"
 	tidbmysql "github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
@@ -74,7 +73,7 @@ func NewActiveActiveSyncStatsCollector(changefeedID common.ChangeFeedID) *Active
 	return &ActiveActiveSyncStatsCollector{
 		keyspace:                 keyspace,
 		changefeed:               changefeed,
-		conflictSkipRows:         metrics.ActiveActiveConflictSkipRowsCounter.WithLabelValues(keyspace, changefeed),
+		conflictSkipRows:         activeActiveConflictSkipRowsCounter.WithLabelValues(keyspace, changefeed),
 		lastConflictSkipRowsByID: make(map[uint64]uint64),
 	}
 }
@@ -120,7 +119,7 @@ func (c *ActiveActiveSyncStatsCollector) ForgetConn(connID uint64) {
 // Close releases metric series held by this collector.
 func (c *ActiveActiveSyncStatsCollector) Close() {
 	// Reset the series on sink rebuild.
-	metrics.ActiveActiveConflictSkipRowsCounter.DeleteLabelValues(c.keyspace, c.changefeed)
+	activeActiveConflictSkipRowsCounter.DeleteLabelValues(c.keyspace, c.changefeed)
 }
 
 // queryActiveActiveSyncStats queries CONNECTION_ID() and the session variable
diff --git a/server/metrics.go b/server/metrics.go
index 989a2beda0..bc931f40b0 100644
--- a/server/metrics.go
+++ b/server/metrics.go
@@ -16,6 +16,7 @@ package server
 import (
 	"github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/metrics"
+	"github.com/pingcap/ticdc/pkg/sink/mysql"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -23,5 +24,6 @@ var registry = prometheus.NewRegistry()
 
 func init() {
 	metrics.InitMetrics(registry)
+	mysql.InitMetrics(registry)
 	event.InitEventMetrics(registry)
 }