From 0ebf6b7452252c3c6223a88ad35fbfed8d7633e8 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Tue, 7 Jul 2026 03:45:33 +0000
Subject: [PATCH 01/14] init

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 .../sink/cloudstorage/dml_writers.go          | 22 +++---
 .../sink/cloudstorage/encoder_group.go        |  3 +
 downstreamadapter/sink/cloudstorage/sink.go   | 10 ++-
 .../sink/cloudstorage/sink_test.go            | 70 +++++++++++++++++++
 downstreamadapter/sink/cloudstorage/task.go   |  8 +++
 .../sink/columnselector/column_selector.go    |  7 ++
 downstreamadapter/sink/helper/mq_row_event.go | 70 +++++++++++++++++++
 downstreamadapter/sink/kafka/sink.go          | 39 ++---------
 downstreamadapter/sink/pulsar/sink.go         | 39 ++---------
 .../codec/canal/canal_json_txn_encoder.go     |  7 ++
 .../canal/canal_json_txn_encoder_test.go      | 29 ++++++++
 pkg/sink/codec/common/encoder.go              |  4 ++
 pkg/sink/codec/csv/csv_encoder.go             | 29 ++++++--
 pkg/sink/codec/csv/csv_encoder_test.go        | 37 ++++++++++
 pkg/sink/codec/csv/csv_message.go             | 29 +++++---
 15 files changed, 308 insertions(+), 95 deletions(-)
 create mode 100644 downstreamadapter/sink/helper/mq_row_event.go

diff --git a/downstreamadapter/sink/cloudstorage/dml_writers.go b/downstreamadapter/sink/cloudstorage/dml_writers.go
index 37f54fd258..6f4d8e9eb2 100644
--- a/downstreamadapter/sink/cloudstorage/dml_writers.go
+++ b/downstreamadapter/sink/cloudstorage/dml_writers.go
@@ -18,6 +18,7 @@ import (
 	"time"
 
 	"github.com/pingcap/ticdc/downstreamadapter/sink/cloudstorage/spool"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	sinkmetrics "github.com/pingcap/ticdc/downstreamadapter/sink/metrics"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	commonType "github.com/pingcap/ticdc/pkg/common"
@@ -43,8 +44,9 @@ type dmlWriters struct {
 	encodeGroup *encoderGroup
 	spool       *spool.Spool
 
-	writers []*writer
-	closed  atomic.Bool
+	columnSelector *columnselector.ColumnSelectors
+	writers        []*writer
+	closed         atomic.Bool
 }
 
 func newDMLWriters(
@@ -54,6 +56,7 @@ func newDMLWriters(
 	encoderConfig *common.Config,
 	extension string,
 	statistics *metrics.Statistics,
+	columnSelector *columnselector.ColumnSelectors,
 ) (*dmlWriters, error) {
 	messageCh := chann.NewUnlimitedChannelDefault[*task]()
 	encoderGroup := newEncoderGroup(
@@ -76,12 +79,13 @@ func newDMLWriters(
 	}
 
 	return &dmlWriters{
-		changefeedID: changefeedID,
-		statistics:   statistics,
-		msgCh:        messageCh,
-		encodeGroup:  encoderGroup,
-		spool:        spool,
-		writers:      writers,
+		changefeedID:   changefeedID,
+		statistics:     statistics,
+		msgCh:          messageCh,
+		encodeGroup:    encoderGroup,
+		spool:          spool,
+		columnSelector: columnSelector,
+		writers:        writers,
 	}, nil
 }
 
@@ -168,7 +172,7 @@ func (d *dmlWriters) addDMLEvent(event *commonEvent.DMLEvent) {
 		TableInfoVersion: event.TableInfoVersion,
 		DispatcherID:     event.GetDispatcherID(),
 	}
-	d.msgCh.Push(newDMLTask(table, event))
+	d.msgCh.Push(newDMLTask(table, event, d.columnSelector.GetForTableInfo(event.TableInfo)))
 }
 
 func (d *dmlWriters) flushDMLBeforeBlock(ctx context.Context, event commonEvent.BlockEvent) error {
diff --git a/downstreamadapter/sink/cloudstorage/encoder_group.go b/downstreamadapter/sink/cloudstorage/encoder_group.go
index 9b36deb502..c111cfe0f7 100644
--- a/downstreamadapter/sink/cloudstorage/encoder_group.go
+++ b/downstreamadapter/sink/cloudstorage/encoder_group.go
@@ -108,6 +108,9 @@ func (eg *encoderGroup) runEncoder(ctx context.Context, index int) error {
 				continue
 			}
 
+			if setter, ok := encoder.(common.ColumnSelectorAwareTxnEventEncoder); ok {
+				setter.SetColumnSelector(task.columnSelector)
+			}
 			err = encoder.AppendTxnEvent(task.event)
 			if err != nil {
 				return err
diff --git a/downstreamadapter/sink/cloudstorage/sink.go b/downstreamadapter/sink/cloudstorage/sink.go
index dfb9b232da..bc273abc90 100644
--- a/downstreamadapter/sink/cloudstorage/sink.go
+++ b/downstreamadapter/sink/cloudstorage/sink.go
@@ -21,6 +21,7 @@ import (
 	"time"
 
 	"github.com/pingcap/log"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	"github.com/pingcap/ticdc/pkg/common"
@@ -86,6 +87,9 @@ func Verify(ctx context.Context, changefeedID common.ChangeFeedID, sinkURI *url.
 	if err != nil {
 		return err
 	}
+	if _, err = columnselector.New(sinkConfig); err != nil {
+		return err
+	}
 	_, err = helper.GetEncoderConfig(changefeedID, sinkURI, protocol, sinkConfig, math.MaxInt)
 	if err != nil {
 		return err
@@ -122,6 +126,10 @@ func New(
 	if err != nil {
 		return nil, err
 	}
+	columnSelectors, err := columnselector.New(sinkConfig)
+	if err != nil {
+		return nil, err
+	}
 	storage, err := util.GetExternalStorageWithDefaultTimeout(ctx, sinkURI.String())
 	if err != nil {
 		return nil, err
@@ -133,7 +141,7 @@ func New(
 			storage.Close()
 		}
 	}()
-	dmlWriters, err := newDMLWriters(changefeedID, storage, cfg, encoderConfig, ext, statistics)
+	dmlWriters, err := newDMLWriters(changefeedID, storage, cfg, encoderConfig, ext, statistics, columnSelectors)
 	if err != nil {
 		return nil, err
 	}
diff --git a/downstreamadapter/sink/cloudstorage/sink_test.go b/downstreamadapter/sink/cloudstorage/sink_test.go
index 5a44dd400a..6435bdbaf4 100644
--- a/downstreamadapter/sink/cloudstorage/sink_test.go
+++ b/downstreamadapter/sink/cloudstorage/sink_test.go
@@ -157,6 +157,76 @@ func TestBasicFunctionality(t *testing.T) {
 	require.Equal(t, count.Load(), int64(3))
 }
 
+func TestCloudStorageSinkWithColumnSelector(t *testing.T) {
+	parentDir := t.TempDir()
+	uri := fmt.Sprintf("file:///%s?protocol=csv&flush-interval=3600s&file-size=1024", parentDir)
+	sinkURI, err := url.Parse(uri)
+	require.NoError(t, err)
+
+	replicaConfig := config.GetDefaultReplicaConfig()
+	replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{Matcher: []string{"test.table1"}, Columns: []string{"c1"}},
+	}
+	err = replicaConfig.ValidateAndAdjust(sinkURI)
+	require.NoError(t, err)
+	replicaConfig.Sink.DateSeparator = util.AddressOf(config.DateSeparatorNone.String())
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	setPDClockForTest(t, pdutil.NewClock4Test())
+	cloudStorageSink, err := newSinkForTest(ctx, replicaConfig, sinkURI, nil)
+	require.NoError(t, err)
+
+	runDone := runSinkInBackground(t, ctx, cloudStorageSink)
+	defer cancelAndWaitSink(t, cancel, runDone)
+
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+
+	helper.Tk().MustExec("use test")
+	job := helper.DDL2Job("create table table1(c1 int primary key, c2 varchar(255))")
+	require.NotNil(t, job)
+	helper.ApplyJob(job)
+
+	dispatcherID := common.NewDispatcherID()
+	event := helper.DML2Event(job.SchemaName, job.TableName, `insert into table1 values (1, "filtered")`)
+	event.TableInfoVersion = job.BinlogInfo.FinishedTS
+	event.DispatcherID = dispatcherID
+
+	var flushed atomic.Uint64
+	event.AddPostFlushFunc(func() {
+		flushed.Add(1)
+	})
+
+	cloudStorageSink.AddDMLEvent(event)
+	err = cloudStorageSink.FlushDMLBeforeBlock(&commonEvent.DDLEvent{
+		DispatcherID: dispatcherID,
+		FinishedTs:   event.CommitTs + 1,
+	})
+	require.NoError(t, err)
+	require.Equal(t, uint64(1), flushed.Load())
+
+	tableDir := path.Join(parentDir, job.SchemaName, job.TableName, fmt.Sprint(event.TableInfoVersion))
+	var content []byte
+	require.Eventually(t, func() bool {
+		files, err := os.ReadDir(tableDir)
+		if err != nil {
+			return false
+		}
+		for _, file := range files {
+			if file.IsDir() || !strings.HasSuffix(file.Name(), ".csv") {
+				continue
+			}
+			content, err = os.ReadFile(path.Join(tableDir, file.Name()))
+			return err == nil
+		}
+		return false
+	}, testEventuallyTimeout, testEventuallyTick)
+	require.Contains(t, string(content), "1")
+	require.NotContains(t, string(content), "filtered")
+}
+
 func TestIgnoreCallsAfterRunError(t *testing.T) {
 	uri := fmt.Sprintf("file:///%s?protocol=csv", t.TempDir())
 	sinkURI, err := url.Parse(uri)
diff --git a/downstreamadapter/sink/cloudstorage/task.go b/downstreamadapter/sink/cloudstorage/task.go
index 3b9ff1a082..c3c654e543 100644
--- a/downstreamadapter/sink/cloudstorage/task.go
+++ b/downstreamadapter/sink/cloudstorage/task.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"sync/atomic"
 
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
@@ -42,6 +43,7 @@ type task struct {
 	callbacks      *txnCallbacks                   // Lightweight txn callbacks detached from event.
 	tableInfo      *commonType.TableInfo           // Table info used after event is released.
 	versionedTable cloudstorage.VersionedTableName // Versioned output identity for the DML event.
+	columnSelector commonEvent.Selector            // Column selector used by the txn encoder.
 	encodedMsgs    []*common.Message               // Encoded result built from event.
 
 	// Flush-only field.
@@ -51,16 +53,22 @@ type task struct {
 func newDMLTask(
 	version cloudstorage.VersionedTableName,
 	event *commonEvent.DMLEvent,
+	selectors ...commonEvent.Selector,
 ) *task {
 	// The dispatcher path registers progress callbacks before calling
 	// Sink.AddDMLEvent, so snapshot callbacks here and release the large event
 	// object after encoding.
+	selector := commonEvent.Selector(columnselector.NewDefaultColumnSelector())
+	if len(selectors) > 0 && selectors[0] != nil {
+		selector = selectors[0]
+	}
 	return &task{
 		kind:           taskKindDML,
 		event:          event,
 		callbacks:      newTxnCallbacks(event),
 		tableInfo:      event.TableInfo,
 		versionedTable: version,
+		columnSelector: selector,
 		dispatcherID:   event.GetDispatcherID(),
 	}
 }
diff --git a/downstreamadapter/sink/columnselector/column_selector.go b/downstreamadapter/sink/columnselector/column_selector.go
index 0c619b42eb..4e62080632 100644
--- a/downstreamadapter/sink/columnselector/column_selector.go
+++ b/downstreamadapter/sink/columnselector/column_selector.go
@@ -99,6 +99,13 @@ func (c *ColumnSelectors) Get(schema, table string) *ColumnSelector {
 	return &ColumnSelector{}
 }
 
+func (c *ColumnSelectors) GetForTableInfo(tableInfo *common.TableInfo) *ColumnSelector {
+	if c == nil || tableInfo == nil {
+		return NewDefaultColumnSelector()
+	}
+	return c.Get(tableInfo.GetSchemaName(), tableInfo.GetTableName())
+}
+
 // VerifyTables return the error if any given table cannot satisfy the column selector constraints.
 // 1. if the column is filter out, it must not be a part of handle key or the unique key.
 // 2. if the filtered out column is used in the column dispatcher, return error.
diff --git a/downstreamadapter/sink/helper/mq_row_event.go b/downstreamadapter/sink/helper/mq_row_event.go
new file mode 100644
index 0000000000..7a7ea39902
--- /dev/null
+++ b/downstreamadapter/sink/helper/mq_row_event.go
@@ -0,0 +1,70 @@
+// 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,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package helper
+
+import (
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter/partition"
+	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
+)
+
+func NewMQRowEvents(
+	event *commonEvent.DMLEvent,
+	topic string,
+	partitionNum int32,
+	partitionGenerator partition.Generator,
+	selector commonEvent.Selector,
+) ([]*commonEvent.MQRowEvent, error) {
+	if selector == nil {
+		selector = columnselector.NewDefaultColumnSelector()
+	}
+
+	rowsCount := event.Len()
+	events := make([]*commonEvent.MQRowEvent, 0, rowsCount)
+	rowCallback := NewTxnPostFlushRowCallback(event, uint64(rowsCount))
+
+	for {
+		row, ok := event.GetNextRow()
+		if !ok {
+			event.Rewind()
+			break
+		}
+
+		index, key, err := partitionGenerator.GeneratePartitionIndexAndKey(
+			&row, partitionNum, event.TableInfo, event.CommitTs)
+		if err != nil {
+			return nil, err
+		}
+
+		events = append(events, &commonEvent.MQRowEvent{
+			Key: commonEvent.TopicPartitionKey{
+				Topic:          topic,
+				Partition:      index,
+				PartitionKey:   key,
+				TotalPartition: partitionNum,
+			},
+			RowEvent: commonEvent.RowEvent{
+				PhysicalTableID: event.PhysicalTableID,
+				TableInfo:       event.TableInfo,
+				StartTs:         event.StartTs,
+				CommitTs:        event.CommitTs,
+				Event:           row,
+				Callback:        rowCallback,
+				ColumnSelector:  selector,
+				Checksum:        row.Checksum,
+			},
+		})
+	}
+	return events, nil
+}
diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go
index 2f5f7dd005..8515aa420f 100644
--- a/downstreamadapter/sink/kafka/sink.go
+++ b/downstreamadapter/sink/kafka/sink.go
@@ -242,41 +242,10 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error {
 			}
 
 			partitionGenerator := s.comp.eventRouter.GetPartitionGenerator(schema, table)
-			selector := s.comp.columnSelector.Get(schema, table)
-			rowsCount := event.Len()
-			events := make([]*commonEvent.MQRowEvent, 0, rowsCount)
-			rowCallback := helper.NewTxnPostFlushRowCallback(event, uint64(rowsCount))
-
-			for {
-				row, ok := event.GetNextRow()
-				if !ok {
-					event.Rewind()
-					break
-				}
-
-				index, key, err := partitionGenerator.GeneratePartitionIndexAndKey(&row, partitionNum, event.TableInfo, event.CommitTs)
-				if err != nil {
-					return errors.Trace(err)
-				}
-
-				events = append(events, &commonEvent.MQRowEvent{
-					Key: commonEvent.TopicPartitionKey{
-						Topic:          topic,
-						Partition:      index,
-						PartitionKey:   key,
-						TotalPartition: partitionNum,
-					},
-					RowEvent: commonEvent.RowEvent{
-						PhysicalTableID: event.PhysicalTableID,
-						TableInfo:       event.TableInfo,
-						StartTs:         event.StartTs,
-						CommitTs:        event.CommitTs,
-						Event:           row,
-						Callback:        rowCallback,
-						ColumnSelector:  selector,
-						Checksum:        row.Checksum,
-					},
-				})
+			selector := s.comp.columnSelector.GetForTableInfo(event.TableInfo)
+			events, err := helper.NewMQRowEvents(event, topic, partitionNum, partitionGenerator, selector)
+			if err != nil {
+				return errors.Trace(err)
 			}
 			s.rowChan.Push(events...)
 		}
diff --git a/downstreamadapter/sink/pulsar/sink.go b/downstreamadapter/sink/pulsar/sink.go
index 5aae70bb2f..79a33a5fff 100644
--- a/downstreamadapter/sink/pulsar/sink.go
+++ b/downstreamadapter/sink/pulsar/sink.go
@@ -402,41 +402,10 @@ func (s *sink) calculateKeyPartitions(ctx context.Context) error {
 			}
 
 			partitionGenerator := s.comp.eventRouter.GetPartitionGenerator(schema, table)
-			selector := s.comp.columnSelector.Get(schema, table)
-			rowsCount := event.Len()
-			events := make([]*commonEvent.MQRowEvent, 0, rowsCount)
-			rowCallback := helper.NewTxnPostFlushRowCallback(event, uint64(rowsCount))
-
-			for {
-				row, ok := event.GetNextRow()
-				if !ok {
-					event.Rewind()
-					break
-				}
-
-				index, key, err := partitionGenerator.GeneratePartitionIndexAndKey(&row, partitionNum, event.TableInfo, event.CommitTs)
-				if err != nil {
-					return errors.Trace(err)
-				}
-
-				events = append(events, &commonEvent.MQRowEvent{
-					Key: commonEvent.TopicPartitionKey{
-						Topic:          topic,
-						Partition:      index,
-						PartitionKey:   key,
-						TotalPartition: partitionNum,
-					},
-					RowEvent: commonEvent.RowEvent{
-						PhysicalTableID: event.PhysicalTableID,
-						TableInfo:       event.TableInfo,
-						StartTs:         event.StartTs,
-						CommitTs:        event.CommitTs,
-						Event:           row,
-						Callback:        rowCallback,
-						ColumnSelector:  selector,
-						Checksum:        row.Checksum,
-					},
-				})
+			selector := s.comp.columnSelector.GetForTableInfo(event.TableInfo)
+			events, err := helper.NewMQRowEvents(event, topic, partitionNum, partitionGenerator, selector)
+			if err != nil {
+				return errors.Trace(err)
 			}
 			s.rowChan.Push(events...)
 		}
diff --git a/pkg/sink/codec/canal/canal_json_txn_encoder.go b/pkg/sink/codec/canal/canal_json_txn_encoder.go
index 0af6f4f3f2..b761e32971 100644
--- a/pkg/sink/codec/canal/canal_json_txn_encoder.go
+++ b/pkg/sink/codec/canal/canal_json_txn_encoder.go
@@ -47,6 +47,13 @@ func NewJSONTxnEventEncoder(config *common.Config) common.TxnEventEncoder {
 	}
 }
 
+func (j *JSONTxnEventEncoder) SetColumnSelector(selector commonEvent.Selector) {
+	if selector == nil {
+		selector = columnselector.NewDefaultColumnSelector()
+	}
+	j.columnSelector = selector
+}
+
 // AppendTxnEvent appends a txn event to the encoder.
 func (j *JSONTxnEventEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error {
 	for {
diff --git a/pkg/sink/codec/canal/canal_json_txn_encoder_test.go b/pkg/sink/codec/canal/canal_json_txn_encoder_test.go
index c0c6943d26..1d1dc38028 100644
--- a/pkg/sink/codec/canal/canal_json_txn_encoder_test.go
+++ b/pkg/sink/codec/canal/canal_json_txn_encoder_test.go
@@ -16,6 +16,7 @@ package canal
 import (
 	"testing"
 
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
@@ -100,3 +101,31 @@ func TestCanalJSONAppendTxnEventEncoderWithCallback(t *testing.T) {
 	require.Equal(t, 0, encoder.(*JSONTxnEventEncoder).batchSize)
 	require.Equal(t, 0, encoder.(*JSONTxnEventEncoder).valueBuf.Len())
 }
+
+func TestCanalJSONTxnEventEncoderWithColumnSelector(t *testing.T) {
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+
+	helper.DDL2Event("create table test.t(col1 int primary key, col2 varchar(255))")
+	event := helper.DML2Event("test", "t", `insert into test.t values (1, "filtered")`)
+
+	selectors, err := columnselector.New(&config.SinkConfig{
+		ColumnSelectors: []*config.ColumnSelector{
+			{Matcher: []string{"test.t"}, Columns: []string{"col1"}},
+		},
+	})
+	require.NoError(t, err)
+
+	encoder := NewJSONTxnEventEncoder(common.NewConfig(config.ProtocolCanalJSON))
+	selectorSetter, ok := encoder.(common.ColumnSelectorAwareTxnEventEncoder)
+	require.True(t, ok)
+	selectorSetter.SetColumnSelector(selectors.GetForTableInfo(event.TableInfo))
+
+	require.NoError(t, encoder.AppendTxnEvent(event))
+	messages := encoder.Build()
+	require.Len(t, messages, 1)
+	value := string(messages[0].Value)
+	require.Contains(t, value, "col1")
+	require.NotContains(t, value, "col2")
+	require.NotContains(t, value, "filtered")
+}
diff --git a/pkg/sink/codec/common/encoder.go b/pkg/sink/codec/common/encoder.go
index bcb9afd365..27e01d4052 100644
--- a/pkg/sink/codec/common/encoder.go
+++ b/pkg/sink/codec/common/encoder.go
@@ -44,6 +44,10 @@ type TxnEventEncoder interface {
 	Build() []*Message
 }
 
+type ColumnSelectorAwareTxnEventEncoder interface {
+	SetColumnSelector(commonEvent.Selector)
+}
+
 // IsColumnValueEqual checks whether the preValue and updatedValue are equal.
 func IsColumnValueEqual(preValue, updatedValue interface{}) bool {
 	if preValue == nil || updatedValue == nil {
diff --git a/pkg/sink/codec/csv/csv_encoder.go b/pkg/sink/codec/csv/csv_encoder.go
index 0001fbd65c..05b77e9abb 100644
--- a/pkg/sink/codec/csv/csv_encoder.go
+++ b/pkg/sink/codec/csv/csv_encoder.go
@@ -16,6 +16,7 @@ package csv
 import (
 	"bytes"
 
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
@@ -28,20 +29,30 @@ type batchEncoder struct {
 	callback  func()
 	batchSize int
 	config    *common.Config
+
+	columnSelector commonEvent.Selector
 }
 
 // NewTxnEventEncoder creates a new csv BatchEncoder.
 func NewTxnEventEncoder(config *common.Config) common.TxnEventEncoder {
 	return &batchEncoder{
-		config:   config,
-		valueBuf: &bytes.Buffer{},
+		config:         config,
+		valueBuf:       &bytes.Buffer{},
+		columnSelector: columnselector.NewDefaultColumnSelector(),
 	}
 }
 
+func (b *batchEncoder) SetColumnSelector(selector commonEvent.Selector) {
+	if selector == nil {
+		selector = columnselector.NewDefaultColumnSelector()
+	}
+	b.columnSelector = selector
+}
+
 // AppendTxnEvent implements the TxnEventEncoder interface
 func (b *batchEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error {
 	if b.config.CSVOutputFieldHeader && b.batchSize == 0 {
-		b.setHeader(event.TableInfo)
+		b.setHeader(event.TableInfo, b.columnSelector)
 	}
 	for {
 		row, ok := event.GetNextRow()
@@ -50,9 +61,10 @@ func (b *batchEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error {
 			break
 		}
 		msg, err := rowChangedEvent2CSVMsg(b.config, &commonEvent.RowEvent{
-			TableInfo: event.TableInfo,
-			CommitTs:  event.CommitTs,
-			Event:     row,
+			TableInfo:      event.TableInfo,
+			CommitTs:       event.CommitTs,
+			Event:          row,
+			ColumnSelector: b.columnSelector,
 		})
 		if err != nil {
 			return err
@@ -84,10 +96,13 @@ func (b *batchEncoder) Build() (messages []*common.Message) {
 	return []*common.Message{ret}
 }
 
-func (b *batchEncoder) setHeader(tableInfo *commonType.TableInfo) {
+func (b *batchEncoder) setHeader(tableInfo *commonType.TableInfo, selector commonEvent.Selector) {
 	buf := &bytes.Buffer{}
 	colNames := make([]string, 0, len(tableInfo.GetColumns()))
 	for _, col := range tableInfo.GetColumns() {
+		if !shouldEncodeColumn(col, selector) {
+			continue
+		}
 		colNames = append(colNames, col.Name.O)
 	}
 	buf.Write(encodeHeader(b.config, colNames))
diff --git a/pkg/sink/codec/csv/csv_encoder_test.go b/pkg/sink/codec/csv/csv_encoder_test.go
index bb51daaf6c..6b46c28f88 100644
--- a/pkg/sink/codec/csv/csv_encoder_test.go
+++ b/pkg/sink/codec/csv/csv_encoder_test.go
@@ -17,7 +17,9 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
+	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
 	"github.com/stretchr/testify/require"
 )
@@ -128,3 +130,38 @@ func TestCSVBatchCodecWithHeader(t *testing.T) {
 	messages = encoder.Build()
 	require.Len(t, messages, 0)
 }
+
+func TestCSVTxnEventEncoderWithColumnSelector(t *testing.T) {
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+
+	helper.DDL2Event("create table test.table1(col1 int primary key, col2 varchar(255))")
+	event := helper.DML2Event("test", "table1", `insert into test.table1 values (1, "filtered")`)
+
+	selectors, err := columnselector.New(&config.SinkConfig{
+		ColumnSelectors: []*config.ColumnSelector{
+			{Matcher: []string{"test.table1"}, Columns: []string{"col1"}},
+		},
+	})
+	require.NoError(t, err)
+
+	cfg := &common.Config{
+		Delimiter:            ",",
+		Quote:                "\"",
+		Terminator:           "\n",
+		NullString:           "\\N",
+		IncludeCommitTs:      true,
+		CSVOutputFieldHeader: true,
+	}
+	encoder := NewTxnEventEncoder(cfg)
+	selectorSetter, ok := encoder.(common.ColumnSelectorAwareTxnEventEncoder)
+	require.True(t, ok)
+	selectorSetter.SetColumnSelector(selectors.GetForTableInfo(event.TableInfo))
+
+	require.NoError(t, encoder.AppendTxnEvent(event))
+	messages := encoder.Build()
+	require.Len(t, messages, 1)
+	require.Equal(t, "ticdc-meta$operation,ticdc-meta$table,ticdc-meta$schema,ticdc-meta$commit-ts,col1\n", string(messages[0].Key))
+	require.NotContains(t, string(messages[0].Key), "col2")
+	require.NotContains(t, string(messages[0].Value), "filtered")
+}
diff --git a/pkg/sink/codec/csv/csv_message.go b/pkg/sink/codec/csv/csv_message.go
index 9649bbfc3c..79b429d13f 100644
--- a/pkg/sink/codec/csv/csv_message.go
+++ b/pkg/sink/codec/csv/csv_message.go
@@ -332,6 +332,7 @@ func fromColValToCsvVal(csvConfig *common.Config, row *chunk.Row, idx int, colIn
 // rowChangedEvent2CSVMsg converts a RowChangedEvent to a csv record.
 func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMessage, error) {
 	tableInfo := e.TableInfo
+	selector := e.ColumnSelector
 	csvMsg := &csvMessage{
 		config:     csvConfig,
 		tableName:  tableInfo.GetTargetTableName(),
@@ -348,14 +349,14 @@ func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMe
 	var err error
 	if e.IsDelete() {
 		csvMsg.opType = operationDelete
-		csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), tableInfo)
+		csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), tableInfo, selector)
 		if err != nil {
 			return nil, err
 		}
 	} else if e.IsInsert() {
 		// This is a insert operation.
 		csvMsg.opType = operationInsert
-		csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), tableInfo)
+		csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), tableInfo, selector)
 		if err != nil {
 			return nil, err
 		}
@@ -368,12 +369,12 @@ func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMe
 					"the column length of preColumns %d doesn't equal to that of columns %d",
 					e.GetPreRows().Len(), e.GetRows().Len())
 			}
-			csvMsg.preColumns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), tableInfo)
+			csvMsg.preColumns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetPreRows(), tableInfo, selector)
 			if err != nil {
 				return nil, err
 			}
 		}
-		csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), tableInfo)
+		csvMsg.columns, err = rowChangeColumns2CSVColumns(csvConfig, e.GetRows(), tableInfo, selector)
 		if err != nil {
 			return nil, err
 		}
@@ -381,13 +382,16 @@ func rowChangedEvent2CSVMsg(csvConfig *common.Config, e *event.RowEvent) (*csvMe
 	return csvMsg, nil
 }
 
-func rowChangeColumns2CSVColumns(csvConfig *common.Config, row *chunk.Row, tableInfo *commonType.TableInfo) ([]any, error) {
+func rowChangeColumns2CSVColumns(
+	csvConfig *common.Config,
+	row *chunk.Row,
+	tableInfo *commonType.TableInfo,
+	selector event.Selector,
+) ([]any, error) {
 	var csvColumns []any
 
 	for i, col := range tableInfo.GetColumns() {
-		// column could be nil in a condition described in
-		// https://github.com/pingcap/ticdc/issues/6198#issuecomment-1191132951
-		if col == nil || col.IsVirtualGenerated() {
+		if !shouldEncodeColumn(col, selector) {
 			continue
 		}
 
@@ -402,6 +406,15 @@ func rowChangeColumns2CSVColumns(csvConfig *common.Config, row *chunk.Row, table
 	return csvColumns, nil
 }
 
+func shouldEncodeColumn(col *timodel.ColumnInfo, selector event.Selector) bool {
+	// column could be nil in a condition described in
+	// https://github.com/pingcap/ticdc/issues/6198#issuecomment-1191132951
+	if col == nil || col.IsVirtualGenerated() {
+		return false
+	}
+	return selector == nil || selector.Select(col)
+}
+
 // The header should contain the name corresponding to the file record field,
 // and should have the same number as the record field.
 // | ticdc-meta$operation | ticdc-meta$table | ticdc-meta$schema | ticdc-meta$commit-ts | ticdc-meta$is-update | col1 | col2 | ... |

From 01b79315cc68a675674269ddd14d2083bcb07bcd Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Tue, 7 Jul 2026 06:23:19 +0000
Subject: [PATCH 02/14] add test

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 .../conf/changefeed.toml                      |  3 +-
 .../column_selector/data/data.sql             | 45 ++++++++++++
 .../run.sh                                    | 70 ++++++++++++++-----
 .../kafka_column_selector/data/data.sql       | 49 -------------
 tests/integration_tests/run_heavy_it_in_ci.sh |  2 +-
 5 files changed, 100 insertions(+), 69 deletions(-)
 rename tests/integration_tests/{kafka_column_selector => column_selector}/conf/changefeed.toml (73%)
 create mode 100644 tests/integration_tests/column_selector/data/data.sql
 rename tests/integration_tests/{kafka_column_selector => column_selector}/run.sh (54%)
 mode change 100755 => 100644
 delete mode 100644 tests/integration_tests/kafka_column_selector/data/data.sql

diff --git a/tests/integration_tests/kafka_column_selector/conf/changefeed.toml b/tests/integration_tests/column_selector/conf/changefeed.toml
similarity index 73%
rename from tests/integration_tests/kafka_column_selector/conf/changefeed.toml
rename to tests/integration_tests/column_selector/conf/changefeed.toml
index addd3dd41a..fe5bfa1af5 100644
--- a/tests/integration_tests/kafka_column_selector/conf/changefeed.toml
+++ b/tests/integration_tests/column_selector/conf/changefeed.toml
@@ -1,7 +1,6 @@
 [sink]
 column-selectors = [
     {matcher = ['test.t1'], columns = ['a', 'b']},
-    {matcher = ['test.*'], columns = ["*", "!b"]},
-
+    {matcher = ['test.*'], columns = ['*', '!b']},
     {matcher = ['test1.t1'], columns = ['column*', '!column1']},
 ]
diff --git a/tests/integration_tests/column_selector/data/data.sql b/tests/integration_tests/column_selector/data/data.sql
new file mode 100644
index 0000000000..def4ddbdf2
--- /dev/null
+++ b/tests/integration_tests/column_selector/data/data.sql
@@ -0,0 +1,45 @@
+drop database if exists `test`;
+create database `test`;
+use `test`;
+
+create table t1 (
+    a int primary key,
+    b varchar(64),
+    c varchar(64)
+);
+
+insert into t1 values (1, 'keep_b_t1', 'filtered_c_t1');
+insert into t1 values (2, 'keep_b_t1_2', 'filtered_c_t1_2');
+
+create table t2 (
+    a int primary key,
+    b varchar(64),
+    c varchar(64)
+);
+
+insert into t2 values (1, 'filtered_b_t2', 'keep_c_t2');
+insert into t2 values (2, 'filtered_b_t2_2', 'keep_c_t2_2');
+
+create table t3 (
+    a int primary key,
+    b varchar(64),
+    c varchar(64)
+);
+
+insert into t3 values (1, 'filtered_b_t3', 'keep_c_t3');
+insert into t3 values (2, 'filtered_b_t3_2', 'keep_c_t3_2');
+
+drop database if exists `test1`;
+create database `test1`;
+use `test1`;
+
+create table t1 (
+    column0 int primary key,
+    column1 varchar(64),
+    column2 varchar(64)
+);
+
+insert into t1 values (1, 'filtered_column1_test1', 'keep_column2_test1');
+insert into t1 values (2, 'filtered_column1_test1_2', 'keep_column2_test1_2');
+
+create table finishmark(id int primary key);
diff --git a/tests/integration_tests/kafka_column_selector/run.sh b/tests/integration_tests/column_selector/run.sh
old mode 100755
new mode 100644
similarity index 54%
rename from tests/integration_tests/kafka_column_selector/run.sh
rename to tests/integration_tests/column_selector/run.sh
index a7897c5432..91235725af
--- a/tests/integration_tests/kafka_column_selector/run.sh
+++ b/tests/integration_tests/column_selector/run.sh
@@ -7,43 +7,79 @@ source $CUR/../_utils/test_prepare
 WORK_DIR=$OUT_DIR/$TEST_NAME
 CDC_BINARY=cdc.test
 SINK_TYPE=$1
+CHANGEFEED_ID="test"
 
-function run() {
-	# test kafka sink only in this case
-	if [ "$SINK_TYPE" != "kafka" ]; then
-		return
-	fi
-
+function prepare_cluster() {
 	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
 	start_tidb_cluster --workdir $WORK_DIR
 
-	# record tso before we create tables to skip the system table DDLs
+	# Record TSO before creating test tables to skip system table DDLs.
 	start_ts=$(run_cdc_cli_tso_query $UP_PD_HOST_1 $UP_PD_PORT_1)
 
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+}
 
-	changefeed_id="test"
-	TOPIC_NAME="column-selector-test-$RANDOM"
-	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=canal-json&partition-num=1&enable-tidb-extension=true"
-	cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c ${changefeed_id} --config="$CUR/conf/changefeed.toml"
-
-	cdc_kafka_consumer --upstream-uri $SINK_URI --downstream-uri="mysql://root@127.0.0.1:3306/?safe-mode=true&batch-dml-enable=false&enable-ddl-ts=false" --upstream-tidb-dsn="root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/?" --config="$CUR/conf/changefeed.toml" --log-file $WORK_DIR/cdc_kafka_consumer.log 2>&1 &
-
-	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-
+function build_checksum_checker() {
 	echo "Starting build checksum checker..."
+	local pwd=$(pwd)
 	cd $CUR/../../utils/checksum_checker
 	if [ ! -f ./checksum_checker ]; then
 		GO111MODULE=on go build
 	fi
+	cd $pwd
+}
 
+function run_checksum_checker() {
+	build_checksum_checker
 	check_table_exists "test1.finishmark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
-	./checksum_checker --upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" --downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" --databases "test,test1" --config="$CUR/conf/changefeed.toml"
+	$CUR/../../utils/checksum_checker/checksum_checker \
+		--upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" \
+		--downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" \
+		--databases "test,test1" \
+		--config="$CUR/conf/changefeed.toml"
+}
+
+function run_kafka() {
+	prepare_cluster
+
+	TOPIC_NAME="column-selector-test-$RANDOM"
+	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=canal-json&partition-num=1&enable-tidb-extension=true"
+	cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c $CHANGEFEED_ID --config="$CUR/conf/changefeed.toml"
+
+	cdc_kafka_consumer --upstream-uri $SINK_URI --downstream-uri="mysql://root@127.0.0.1:3306/?safe-mode=true&batch-dml-enable=false&enable-ddl-ts=false" --upstream-tidb-dsn="root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/?" --config="$CUR/conf/changefeed.toml" --log-file $WORK_DIR/cdc_kafka_consumer.log 2>&1 &
+
+	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_checksum_checker
 
 	cleanup_process $CDC_BINARY
 }
 
+function run_storage() {
+	prepare_cluster
+
+	SINK_URI="file://$WORK_DIR/storage_test?flush-interval=5s&protocol=canal-json&enable-tidb-extension=true"
+	cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c $CHANGEFEED_ID --config="$CUR/conf/changefeed.toml"
+
+	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_storage_consumer $WORK_DIR $SINK_URI $CUR/conf/changefeed.toml ""
+	run_checksum_checker
+}
+
+function run() {
+	case "$SINK_TYPE" in
+	kafka)
+		run_kafka
+		;;
+	storage)
+		run_storage
+		;;
+	*)
+		return
+		;;
+	esac
+}
+
 trap 'stop_test $WORK_DIR' EXIT
 run $*
 check_logs $WORK_DIR
diff --git a/tests/integration_tests/kafka_column_selector/data/data.sql b/tests/integration_tests/kafka_column_selector/data/data.sql
deleted file mode 100644
index 79b4dcee0d..0000000000
--- a/tests/integration_tests/kafka_column_selector/data/data.sql
+++ /dev/null
@@ -1,49 +0,0 @@
-drop database if exists `test`;
-create database `test`;
-use `test`;
-
-create table t1 (
-    a int primary key,
-    b int,
-    c int
-);
-
-insert into t1 values (1, 2, 3);
-insert into t1 values (2, 3, 4);
-insert into t1 values (3, 4, 5);
-
-create table t2 (
-    a int primary key,
-    b int,
-    c int
-);
-
-insert into t2 values (1, 2, 3);
-insert into t2 values (2, 3, 4);
-insert into t2 values (3, 4, 5);
-
-create table t3 (
-    a int primary key,
-    b int,
-    c int
-);
-
-insert into t3 values (1, 2, 3);
-insert into t3 values (2, 3, 4);
-insert into t3 values (3, 4, 5);
-
-drop database if exists `test1`;
-create database `test1`;
-use `test1`;
-
-create table t1 (
-    column0 int primary key,
-    column1 int,
-    column2 int
-);
-
-insert into t1 values (1, 2, 3);
-insert into t1 values (2, 3, 4);
-insert into t1 values (3, 4, 5);
-
-create table finishmark(id int primary key);
diff --git a/tests/integration_tests/run_heavy_it_in_ci.sh b/tests/integration_tests/run_heavy_it_in_ci.sh
index a44f6fb87a..de047acc24 100755
--- a/tests/integration_tests/run_heavy_it_in_ci.sh
+++ b/tests/integration_tests/run_heavy_it_in_ci.sh
@@ -87,7 +87,7 @@ kafka_groups=(
 	# G09
 	'kafka_simple_handle_key_only kafka_simple_handle_key_only_avro mq_sink_error_resume multi_source'
 	# G10
-	'kafka_column_selector kafka_column_selector_avro ddl_with_random_move_table'
+	'column_selector kafka_column_selector_avro ddl_with_random_move_table'
 	# G11
 	'fail_over region_merge multi_changefeeds split_table_check'
 	# G12

From e44827b4d039ac75b567e9878ce1ce42cd4ab954 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Tue, 7 Jul 2026 07:41:23 +0000
Subject: [PATCH 03/14] update

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 .../sink/cloudstorage/buffer_manager.go       |  6 +-
 .../sink/cloudstorage/encoder_group.go        |  8 +-
 .../sink/cloudstorage/encoder_group_test.go   | 24 +++++-
 downstreamadapter/sink/cloudstorage/task.go   | 82 ++++---------------
 .../sink/cloudstorage/task_test.go            | 54 +++++-------
 .../sink/cloudstorage/writer_test.go          | 24 ++----
 downstreamadapter/sink/helper/mq_row_event.go | 61 ++++++++------
 .../codec/canal/canal_json_txn_encoder.go     | 41 +++-------
 .../canal/canal_json_txn_encoder_test.go      | 30 ++++---
 pkg/sink/codec/common/encoder.go              |  6 +-
 pkg/sink/codec/csv/csv_encoder.go             | 38 +++------
 pkg/sink/codec/csv/csv_encoder_test.go        | 41 +++++-----
 12 files changed, 170 insertions(+), 245 deletions(-)

diff --git a/downstreamadapter/sink/cloudstorage/buffer_manager.go b/downstreamadapter/sink/cloudstorage/buffer_manager.go
index 6ad0451167..fa49c404e1 100644
--- a/downstreamadapter/sink/cloudstorage/buffer_manager.go
+++ b/downstreamadapter/sink/cloudstorage/buffer_manager.go
@@ -99,12 +99,14 @@ func (c *bufferManager) run(ctx context.Context) error {
 
 func (c *bufferManager) handleDMLTask(ctx context.Context, task *task) error {
 	if len(task.encodedMsgs) == 0 {
-		task.callbacks.postEnqueue()
+		if task.postEnqueue != nil {
+			task.postEnqueue()
+		}
 		return nil
 	}
 
 	for {
-		action, entry, err := c.spool.TryEnqueue(task.encodedMsgs, task.callbacks.postEnqueue)
+		action, entry, err := c.spool.TryEnqueue(task.encodedMsgs, task.postEnqueue)
 		if err != nil {
 			return err
 		}
diff --git a/downstreamadapter/sink/cloudstorage/encoder_group.go b/downstreamadapter/sink/cloudstorage/encoder_group.go
index c111cfe0f7..023bd52467 100644
--- a/downstreamadapter/sink/cloudstorage/encoder_group.go
+++ b/downstreamadapter/sink/cloudstorage/encoder_group.go
@@ -108,16 +108,12 @@ func (eg *encoderGroup) runEncoder(ctx context.Context, index int) error {
 				continue
 			}
 
-			if setter, ok := encoder.(common.ColumnSelectorAwareTxnEventEncoder); ok {
-				setter.SetColumnSelector(task.columnSelector)
-			}
-			err = encoder.AppendTxnEvent(task.event)
+			err = encoder.AppendTxnEvent(task.rowEvents)
 			if err != nil {
 				return err
 			}
 			task.encodedMsgs = encoder.Build()
-			task.replacePostFlushCallbacks()
-			task.event = nil
+			task.rowEvents = nil
 			future.Done()
 		}
 	}
diff --git a/downstreamadapter/sink/cloudstorage/encoder_group_test.go b/downstreamadapter/sink/cloudstorage/encoder_group_test.go
index 269702ec04..268dcc1963 100644
--- a/downstreamadapter/sink/cloudstorage/encoder_group_test.go
+++ b/downstreamadapter/sink/cloudstorage/encoder_group_test.go
@@ -16,6 +16,7 @@ package cloudstorage
 import (
 	"context"
 	"net/url"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -174,6 +175,15 @@ func TestEncodingGroupEncodeDMLTask(t *testing.T) {
 	})
 
 	dispatcherID := commonType.NewDispatcherID()
+	var flushCount atomic.Int64
+	var enqueueCount atomic.Int64
+	event := newTestDMLEvent(dispatcherID, 100)
+	event.AddPostFlushFunc(func() {
+		flushCount.Add(1)
+	})
+	event.AddPostEnqueueFunc(func() {
+		enqueueCount.Add(1)
+	})
 	taskValue := newDMLTask(
 		cloudstorage.VersionedTableName{
 			TableNameWithPhysicTableID: commonType.TableName{
@@ -184,7 +194,7 @@ func TestEncodingGroupEncodeDMLTask(t *testing.T) {
 			TableInfoVersion: 1,
 			DispatcherID:     dispatcherID,
 		},
-		newTestDMLEvent(dispatcherID, 100),
+		event,
 	)
 	require.NoError(t, group.add(ctx, taskValue))
 
@@ -201,7 +211,15 @@ func TestEncodingGroupEncodeDMLTask(t *testing.T) {
 				}
 				task := future.task
 				require.Equal(t, taskValue, task)
-				require.Nil(t, task.event)
+				require.Nil(t, task.rowEvents)
+				require.Len(t, task.encodedMsgs, 1)
+				require.NotNil(t, task.encodedMsgs[0].Callback)
+				task.encodedMsgs[0].Callback()
+				require.Equal(t, int64(1), flushCount.Load())
+				require.Equal(t, int64(0), enqueueCount.Load())
+				require.NotNil(t, task.postEnqueue)
+				task.postEnqueue()
+				require.Equal(t, int64(1), enqueueCount.Load())
 				done <- struct{}{}
 				return nil
 			}
@@ -250,6 +268,8 @@ func newTestDMLEvent(dispatcherID commonType.DispatcherID, tableID int64) *commo
 		PhysicalTableID:  tableID,
 		TableInfo:        tableInfo,
 		TableInfoVersion: 1,
+		Length:           1,
+		RowTypes:         []commonType.RowType{commonType.RowTypeInsert},
 		Rows:             chunk.MutRowFromValues(1, "hello world").ToRow().Chunk(),
 	}
 }
diff --git a/downstreamadapter/sink/cloudstorage/task.go b/downstreamadapter/sink/cloudstorage/task.go
index c3c654e543..43b2024a58 100644
--- a/downstreamadapter/sink/cloudstorage/task.go
+++ b/downstreamadapter/sink/cloudstorage/task.go
@@ -15,9 +15,8 @@ package cloudstorage
 
 import (
 	"context"
-	"sync/atomic"
 
-	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
@@ -39,11 +38,10 @@ type task struct {
 	dispatcherID commonType.DispatcherID
 
 	// DML-only fields.
-	event          *commonEvent.DMLEvent           // Original DML event to encode and flush.
-	callbacks      *txnCallbacks                   // Lightweight txn callbacks detached from event.
+	postEnqueue    func()                          // Transaction enqueue callback.
 	tableInfo      *commonType.TableInfo           // Table info used after event is released.
 	versionedTable cloudstorage.VersionedTableName // Versioned output identity for the DML event.
-	columnSelector commonEvent.Selector            // Column selector used by the txn encoder.
+	rowEvents      []*commonEvent.RowEvent         // Row events to encode and flush.
 	encodedMsgs    []*common.Message               // Encoded result built from event.
 
 	// Flush-only field.
@@ -56,19 +54,18 @@ func newDMLTask(
 	selectors ...commonEvent.Selector,
 ) *task {
 	// The dispatcher path registers progress callbacks before calling
-	// Sink.AddDMLEvent, so snapshot callbacks here and release the large event
-	// object after encoding.
-	selector := commonEvent.Selector(columnselector.NewDefaultColumnSelector())
+	// Sink.AddDMLEvent. Put the flush callback into RowEvents before encoding
+	// so txn encoders do not need extra callback parameters.
+	var selector commonEvent.Selector
 	if len(selectors) > 0 && selectors[0] != nil {
 		selector = selectors[0]
 	}
 	return &task{
 		kind:           taskKindDML,
-		event:          event,
-		callbacks:      newTxnCallbacks(event),
+		postEnqueue:    event.PostEnqueue,
 		tableInfo:      event.TableInfo,
 		versionedTable: version,
-		columnSelector: selector,
+		rowEvents:      helper.NewRowEvents(event, selector, newPostTxnFlushedCallback(event)),
 		dispatcherID:   event.GetDispatcherID(),
 	}
 }
@@ -88,62 +85,19 @@ func (t *task) isFlushTask() bool {
 	return t != nil && t.kind == taskKindFlush
 }
 
-func (t *task) replacePostFlushCallbacks() {
-	if len(t.encodedMsgs) == 0 {
-		return
-	}
-
-	// Txn encoders put event.PostFlush into message.Callback. That method value
-	// keeps the original DMLEvent reachable through the encoded messages, so
-	// replace it with the lightweight callback copy before releasing task.event.
-	for _, msg := range t.encodedMsgs {
-		msg.Callback = nil
-	}
-	// One callback on the last message is enough because all messages in a task
-	// are enqueued and flushed as one spool entry.
-	t.encodedMsgs[len(t.encodedMsgs)-1].Callback = t.callbacks.postFlush
-}
-
-// txnCallbacks is a lightweight copy of a DMLEvent's enqueue and flush
-// callbacks. It lets cloud storage release the full DMLEvent after encoding
-// while preserving the event callback semantics: each stage runs at most once.
-type txnCallbacks struct {
-	flushed  []func()
-	enqueued []func()
-
-	flushedCalled  atomic.Bool
-	enqueuedCalled atomic.Bool
-}
-
-func newTxnCallbacks(event *commonEvent.DMLEvent) *txnCallbacks {
+func newPostTxnFlushedCallback(event *commonEvent.DMLEvent) func() {
 	if event == nil {
-		return &txnCallbacks{}
-	}
-	return &txnCallbacks{
-		flushed:  append([]func(){}, event.PostTxnFlushed...),
-		enqueued: append([]func(){}, event.PostTxnEnqueued...),
-	}
-}
-
-func (c *txnCallbacks) postFlush() {
-	if c == nil || !c.flushedCalled.CompareAndSwap(false, true) {
-		return
-	}
-	for _, f := range c.flushed {
-		if f != nil {
-			f()
-		}
+		return nil
 	}
-	c.postEnqueue()
-}
-
-func (c *txnCallbacks) postEnqueue() {
-	if c == nil || !c.enqueuedCalled.CompareAndSwap(false, true) {
-		return
+	callbacks := append([]func(){}, event.PostTxnFlushed...)
+	if len(callbacks) == 0 {
+		return nil
 	}
-	for _, f := range c.enqueued {
-		if f != nil {
-			f()
+	return func() {
+		for _, f := range callbacks {
+			if f != nil {
+				f()
+			}
 		}
 	}
 }
diff --git a/downstreamadapter/sink/cloudstorage/task_test.go b/downstreamadapter/sink/cloudstorage/task_test.go
index 0750d199f2..0debca9b4a 100644
--- a/downstreamadapter/sink/cloudstorage/task_test.go
+++ b/downstreamadapter/sink/cloudstorage/task_test.go
@@ -16,11 +16,10 @@ package cloudstorage
 import (
 	"context"
 	"errors"
-	"sync/atomic"
 	"testing"
 	"time"
 
-	"github.com/pingcap/ticdc/pkg/sink/codec/common"
+	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/stretchr/testify/require"
 )
 
@@ -51,41 +50,26 @@ func TestFlushMarkerWaitReturnsContextCause(t *testing.T) {
 	require.ErrorIs(t, err, cause)
 }
 
-func TestReplacePostFlushCallbacksDetachesOriginalMessageCallbacks(t *testing.T) {
+func TestPostTxnFlushedCallbackDoesNotRunPostEnqueue(t *testing.T) {
 	t.Parallel()
 
-	var originalCallbackCount atomic.Int32
-	var flushCallbackCount atomic.Int32
-	var enqueueCallbackCount atomic.Int32
-	task := &task{
-		callbacks: &txnCallbacks{
-			flushed: []func(){
-				func() {
-					flushCallbackCount.Add(1)
-				},
-			},
-			enqueued: []func(){
-				func() {
-					enqueueCallbackCount.Add(1)
-				},
-			},
-		},
-		encodedMsgs: []*common.Message{
-			{Callback: func() { originalCallbackCount.Add(1) }},
-			{Callback: func() { originalCallbackCount.Add(1) }},
-			{},
-		},
-	}
+	event := &commonEvent.DMLEvent{}
+	flushCallbackCount := 0
+	enqueueCallbackCount := 0
+	event.AddPostFlushFunc(func() {
+		flushCallbackCount++
+	})
+	event.AddPostEnqueueFunc(func() {
+		enqueueCallbackCount++
+	})
 
-	task.replacePostFlushCallbacks()
+	callback := newPostTxnFlushedCallback(event)
+	require.NotNil(t, callback)
+	callback()
+	require.Equal(t, 1, flushCallbackCount)
+	require.Equal(t, 0, enqueueCallbackCount)
 
-	require.Nil(t, task.encodedMsgs[0].Callback)
-	require.Nil(t, task.encodedMsgs[1].Callback)
-	require.NotNil(t, task.encodedMsgs[2].Callback)
-
-	task.encodedMsgs[2].Callback()
-	task.encodedMsgs[2].Callback()
-	require.Equal(t, int32(0), originalCallbackCount.Load())
-	require.Equal(t, int32(1), flushCallbackCount.Load())
-	require.Equal(t, int32(1), enqueueCallbackCount.Load())
+	event.PostEnqueue()
+	event.PostEnqueue()
+	require.Equal(t, 1, enqueueCallbackCount)
 }
diff --git a/downstreamadapter/sink/cloudstorage/writer_test.go b/downstreamadapter/sink/cloudstorage/writer_test.go
index 250c7b00a6..5eb05a70eb 100644
--- a/downstreamadapter/sink/cloudstorage/writer_test.go
+++ b/downstreamadapter/sink/cloudstorage/writer_test.go
@@ -384,7 +384,7 @@ func TestWriterPostEnqueueAfterConsume(t *testing.T) {
 	require.ErrorIs(t, <-done, context.Canceled)
 }
 
-func TestWriterPostFlushRunsPausedPostEnqueueBeforeLowWatermark(t *testing.T) {
+func TestWriterPostFlushDoesNotRunPausedPostEnqueue(t *testing.T) {
 	t.Parallel()
 
 	changefeedID := commonType.NewChangefeedID4Test("test", t.Name())
@@ -411,21 +411,13 @@ func TestWriterPostFlushRunsPausedPostEnqueueBeforeLowWatermark(t *testing.T) {
 
 	var secondFlushed atomic.Int64
 	var secondEnqueued atomic.Int64
-	secondCallbacks := &txnCallbacks{
-		flushed: []func(){
-			func() {
-				secondFlushed.Add(1)
-			},
-		},
-		enqueued: []func(){
-			func() {
-				secondEnqueued.Add(1)
-			},
-		},
-	}
 	secondMsg := common.NewMsg(nil, []byte(strings.Repeat("b", 120)))
-	secondMsg.Callback = secondCallbacks.postFlush
-	secondEntry, err := spoolBuffer.Enqueue([]*common.Message{secondMsg}, secondCallbacks.postEnqueue)
+	secondMsg.Callback = func() {
+		secondFlushed.Add(1)
+	}
+	secondEntry, err := spoolBuffer.Enqueue([]*common.Message{secondMsg}, func() {
+		secondEnqueued.Add(1)
+	})
 	require.NoError(t, err)
 	defer spoolBuffer.Release(secondEntry)
 	require.Equal(t, int64(0), secondEnqueued.Load())
@@ -442,7 +434,7 @@ func TestWriterPostFlushRunsPausedPostEnqueueBeforeLowWatermark(t *testing.T) {
 	}
 
 	require.Equal(t, int64(1), secondFlushed.Load())
-	require.Equal(t, int64(1), secondEnqueued.Load())
+	require.Equal(t, int64(0), secondEnqueued.Load())
 
 	spoolBuffer.Release(firstEntry)
 	require.Equal(t, int64(1), secondEnqueued.Load())
diff --git a/downstreamadapter/sink/helper/mq_row_event.go b/downstreamadapter/sink/helper/mq_row_event.go
index 7a7ea39902..9904012eff 100644
--- a/downstreamadapter/sink/helper/mq_row_event.go
+++ b/downstreamadapter/sink/helper/mq_row_event.go
@@ -26,23 +26,12 @@ func NewMQRowEvents(
 	partitionGenerator partition.Generator,
 	selector commonEvent.Selector,
 ) ([]*commonEvent.MQRowEvent, error) {
-	if selector == nil {
-		selector = columnselector.NewDefaultColumnSelector()
-	}
-
-	rowsCount := event.Len()
-	events := make([]*commonEvent.MQRowEvent, 0, rowsCount)
-	rowCallback := NewTxnPostFlushRowCallback(event, uint64(rowsCount))
-
-	for {
-		row, ok := event.GetNextRow()
-		if !ok {
-			event.Rewind()
-			break
-		}
+	rowEvents := NewRowEvents(event, selector, NewTxnPostFlushRowCallback(event, uint64(event.Len())))
+	events := make([]*commonEvent.MQRowEvent, 0, len(rowEvents))
 
+	for _, rowEvent := range rowEvents {
 		index, key, err := partitionGenerator.GeneratePartitionIndexAndKey(
-			&row, partitionNum, event.TableInfo, event.CommitTs)
+			&rowEvent.Event, partitionNum, event.TableInfo, event.CommitTs)
 		if err != nil {
 			return nil, err
 		}
@@ -54,17 +43,39 @@ func NewMQRowEvents(
 				PartitionKey:   key,
 				TotalPartition: partitionNum,
 			},
-			RowEvent: commonEvent.RowEvent{
-				PhysicalTableID: event.PhysicalTableID,
-				TableInfo:       event.TableInfo,
-				StartTs:         event.StartTs,
-				CommitTs:        event.CommitTs,
-				Event:           row,
-				Callback:        rowCallback,
-				ColumnSelector:  selector,
-				Checksum:        row.Checksum,
-			},
+			RowEvent: *rowEvent,
 		})
 	}
 	return events, nil
 }
+
+func NewRowEvents(
+	event *commonEvent.DMLEvent,
+	selector commonEvent.Selector,
+	callback func(),
+) []*commonEvent.RowEvent {
+	if selector == nil {
+		selector = columnselector.NewDefaultColumnSelector()
+	}
+
+	events := make([]*commonEvent.RowEvent, 0, event.Len())
+	for {
+		row, ok := event.GetNextRow()
+		if !ok {
+			event.Rewind()
+			break
+		}
+
+		events = append(events, &commonEvent.RowEvent{
+			PhysicalTableID: event.PhysicalTableID,
+			TableInfo:       event.TableInfo,
+			StartTs:         event.StartTs,
+			CommitTs:        event.CommitTs,
+			Event:           row,
+			Callback:        callback,
+			ColumnSelector:  selector,
+			Checksum:        row.Checksum,
+		})
+	}
+	return events
+}
diff --git a/pkg/sink/codec/canal/canal_json_txn_encoder.go b/pkg/sink/codec/canal/canal_json_txn_encoder.go
index b761e32971..bbde5cb878 100644
--- a/pkg/sink/codec/canal/canal_json_txn_encoder.go
+++ b/pkg/sink/codec/canal/canal_json_txn_encoder.go
@@ -17,7 +17,6 @@ import (
 	"bytes"
 
 	"github.com/pingcap/log"
-	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
@@ -33,41 +32,21 @@ type JSONTxnEventEncoder struct {
 	valueBuf   *bytes.Buffer
 	batchSize  int
 	callback   func()
-
-	columnSelector commonEvent.Selector
 }
 
 // NewJSONTxnEventEncoder creates a new JSONTxnEventEncoder
 func NewJSONTxnEventEncoder(config *common.Config) common.TxnEventEncoder {
 	return &JSONTxnEventEncoder{
-		valueBuf:       &bytes.Buffer{},
-		terminator:     []byte(config.Terminator),
-		columnSelector: columnselector.NewDefaultColumnSelector(),
-		config:         config,
-	}
-}
-
-func (j *JSONTxnEventEncoder) SetColumnSelector(selector commonEvent.Selector) {
-	if selector == nil {
-		selector = columnselector.NewDefaultColumnSelector()
+		valueBuf:   &bytes.Buffer{},
+		terminator: []byte(config.Terminator),
+		config:     config,
 	}
-	j.columnSelector = selector
 }
 
 // AppendTxnEvent appends a txn event to the encoder.
-func (j *JSONTxnEventEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error {
-	for {
-		row, ok := event.GetNextRow()
-		if !ok {
-			event.Rewind()
-			break
-		}
-		value, err := newJSONMessageForDML(&commonEvent.RowEvent{
-			TableInfo:      event.TableInfo,
-			CommitTs:       event.CommitTs,
-			Event:          row,
-			ColumnSelector: j.columnSelector,
-		}, j.config, false, "")
+func (j *JSONTxnEventEncoder) AppendTxnEvent(rowEvents []*commonEvent.RowEvent) error {
+	for _, rowEvent := range rowEvents {
+		value, err := newJSONMessageForDML(rowEvent, j.config, false, "")
 		if err != nil {
 			return err
 		}
@@ -77,14 +56,16 @@ func (j *JSONTxnEventEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error
 			log.Warn("Single message is too large for canal-json",
 				zap.Int("maxMessageBytes", j.config.MaxMessageBytes),
 				zap.Int("length", length),
-				zap.Any("table", event.TableInfo.TableName))
-			return errors.ErrMessageTooLarge.GenWithStackByArgs(event.TableInfo.GetTargetTableName(), length, j.config.MaxMessageBytes)
+				zap.Any("table", rowEvent.TableInfo.TableName))
+			return errors.ErrMessageTooLarge.GenWithStackByArgs(rowEvent.TableInfo.GetTargetTableName(), length, j.config.MaxMessageBytes)
 		}
 		j.valueBuf.Write(value)
 		j.valueBuf.Write(j.terminator)
 		j.batchSize++
 	}
-	j.callback = event.PostFlush
+	if len(rowEvents) > 0 {
+		j.callback = rowEvents[len(rowEvents)-1].Callback
+	}
 	return nil
 }
 
diff --git a/pkg/sink/codec/canal/canal_json_txn_encoder_test.go b/pkg/sink/codec/canal/canal_json_txn_encoder_test.go
index 1d1dc38028..5c18f8d533 100644
--- a/pkg/sink/codec/canal/canal_json_txn_encoder_test.go
+++ b/pkg/sink/codec/canal/canal_json_txn_encoder_test.go
@@ -17,12 +17,21 @@ import (
 	"testing"
 
 	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
+	sinkhelper "github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
 	"github.com/stretchr/testify/require"
 )
 
+func appendTxnEventForTest(
+	encoder common.TxnEventEncoder,
+	event *commonEvent.DMLEvent,
+	selector commonEvent.Selector,
+) error {
+	return encoder.AppendTxnEvent(sinkhelper.NewRowEvents(event, selector, nil))
+}
+
 func TestBuildCanalJSONTxnEventEncoder(t *testing.T) {
 	t.Parallel()
 	cfg := common.NewConfig(config.ProtocolCanalJSON)
@@ -43,13 +52,13 @@ func TestCanalJSONTxnEventEncoderMaxMessageBytes(t *testing.T) {
 	maxMessageBytes := 300
 	cfg := common.NewConfig(config.ProtocolCanalJSON).WithMaxMessageBytes(maxMessageBytes)
 	encoder := NewJSONTxnEventEncoder(cfg)
-	err := encoder.AppendTxnEvent(testEvent)
+	err := appendTxnEventForTest(encoder, testEvent, nil)
 	require.Nil(t, err)
 
 	// the test message length is larger than max-message-bytes
 	cfg = cfg.WithMaxMessageBytes(100)
 	encoder = NewJSONTxnEventEncoder(cfg)
-	err = encoder.AppendTxnEvent(testEvent)
+	err = appendTxnEventForTest(encoder, testEvent, nil)
 	require.NotNil(t, err)
 }
 
@@ -57,7 +66,7 @@ func TestCanalJSONTxnEventEncoderUsesTargetNames(t *testing.T) {
 	t.Parallel()
 
 	encoder := NewJSONTxnEventEncoder(common.NewConfig(config.ProtocolCanalJSON))
-	require.NoError(t, encoder.AppendTxnEvent(common.NewRoutedDMLEvent4Test()))
+	require.NoError(t, appendTxnEventForTest(encoder, common.NewRoutedDMLEvent4Test(), nil))
 
 	messages := encoder.Build()
 	require.Len(t, messages, 1)
@@ -77,23 +86,22 @@ func TestCanalJSONAppendTxnEventEncoderWithCallback(t *testing.T) {
 	require.NotNil(t, encoder)
 
 	event := helper.DML2Event("test", "t", `insert into test.t values("aa")`, `insert into test.t values("bb")`)
-
 	count := 0
 
-	// Empty build makes sure that the callback build logic not broken.
+	// Empty build makes sure the build path handles an empty encoder.
 	msgs := encoder.Build()
 	require.Len(t, msgs, 0, "no message should be built and no panic")
 
 	// Append the events.
-	event.AddPostFlushFunc(func() {
+	err := encoder.AppendTxnEvent(sinkhelper.NewRowEvents(event, nil, func() {
 		count++
-	})
-	err := encoder.AppendTxnEvent(event)
+	}))
 	require.Nil(t, err)
 	require.Equal(t, 0, count, "nothing should be called")
 
 	msgs = encoder.Build()
 	require.Len(t, msgs, 1, "expected one message")
+	require.NotNil(t, msgs[0].Callback)
 	msgs[0].Callback()
 	require.Equal(t, 1, count, "expected one callback be called")
 	// Assert the build reset all the internal states.
@@ -117,11 +125,7 @@ func TestCanalJSONTxnEventEncoderWithColumnSelector(t *testing.T) {
 	require.NoError(t, err)
 
 	encoder := NewJSONTxnEventEncoder(common.NewConfig(config.ProtocolCanalJSON))
-	selectorSetter, ok := encoder.(common.ColumnSelectorAwareTxnEventEncoder)
-	require.True(t, ok)
-	selectorSetter.SetColumnSelector(selectors.GetForTableInfo(event.TableInfo))
-
-	require.NoError(t, encoder.AppendTxnEvent(event))
+	require.NoError(t, appendTxnEventForTest(encoder, event, selectors.GetForTableInfo(event.TableInfo)))
 	messages := encoder.Build()
 	require.Len(t, messages, 1)
 	value := string(messages[0].Value)
diff --git a/pkg/sink/codec/common/encoder.go b/pkg/sink/codec/common/encoder.go
index 27e01d4052..5a007434be 100644
--- a/pkg/sink/codec/common/encoder.go
+++ b/pkg/sink/codec/common/encoder.go
@@ -38,16 +38,12 @@ type EventEncoder interface {
 // TxnEventEncoder is an abstraction for events encoder
 type TxnEventEncoder interface {
 	// AppendTxnEvent append a txn event into the buffer.
-	AppendTxnEvent(*commonEvent.DMLEvent) error
+	AppendTxnEvent([]*commonEvent.RowEvent) error
 	// Build builds the batch and returns the bytes of key and value.
 	// Should be called after `AppendTxnEvent`
 	Build() []*Message
 }
 
-type ColumnSelectorAwareTxnEventEncoder interface {
-	SetColumnSelector(commonEvent.Selector)
-}
-
 // IsColumnValueEqual checks whether the preValue and updatedValue are equal.
 func IsColumnValueEqual(preValue, updatedValue interface{}) bool {
 	if preValue == nil || updatedValue == nil {
diff --git a/pkg/sink/codec/csv/csv_encoder.go b/pkg/sink/codec/csv/csv_encoder.go
index 05b77e9abb..4d14ebd3a8 100644
--- a/pkg/sink/codec/csv/csv_encoder.go
+++ b/pkg/sink/codec/csv/csv_encoder.go
@@ -16,7 +16,6 @@ package csv
 import (
 	"bytes"
 
-	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	commonType "github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
@@ -29,50 +28,33 @@ type batchEncoder struct {
 	callback  func()
 	batchSize int
 	config    *common.Config
-
-	columnSelector commonEvent.Selector
 }
 
 // NewTxnEventEncoder creates a new csv BatchEncoder.
 func NewTxnEventEncoder(config *common.Config) common.TxnEventEncoder {
 	return &batchEncoder{
-		config:         config,
-		valueBuf:       &bytes.Buffer{},
-		columnSelector: columnselector.NewDefaultColumnSelector(),
-	}
-}
-
-func (b *batchEncoder) SetColumnSelector(selector commonEvent.Selector) {
-	if selector == nil {
-		selector = columnselector.NewDefaultColumnSelector()
+		config:   config,
+		valueBuf: &bytes.Buffer{},
 	}
-	b.columnSelector = selector
 }
 
 // AppendTxnEvent implements the TxnEventEncoder interface
-func (b *batchEncoder) AppendTxnEvent(event *commonEvent.DMLEvent) error {
+func (b *batchEncoder) AppendTxnEvent(rowEvents []*commonEvent.RowEvent) error {
+	if len(rowEvents) == 0 {
+		return nil
+	}
 	if b.config.CSVOutputFieldHeader && b.batchSize == 0 {
-		b.setHeader(event.TableInfo, b.columnSelector)
+		b.setHeader(rowEvents[0].TableInfo, rowEvents[0].ColumnSelector)
 	}
-	for {
-		row, ok := event.GetNextRow()
-		if !ok {
-			event.Rewind()
-			break
-		}
-		msg, err := rowChangedEvent2CSVMsg(b.config, &commonEvent.RowEvent{
-			TableInfo:      event.TableInfo,
-			CommitTs:       event.CommitTs,
-			Event:          row,
-			ColumnSelector: b.columnSelector,
-		})
+	for _, rowEvent := range rowEvents {
+		msg, err := rowChangedEvent2CSVMsg(b.config, rowEvent)
 		if err != nil {
 			return err
 		}
 		b.valueBuf.Write(msg.encode())
 		b.batchSize++
 	}
-	b.callback = event.PostFlush
+	b.callback = rowEvents[len(rowEvents)-1].Callback
 	return nil
 }
 
diff --git a/pkg/sink/codec/csv/csv_encoder_test.go b/pkg/sink/codec/csv/csv_encoder_test.go
index 6b46c28f88..de1892a3c1 100644
--- a/pkg/sink/codec/csv/csv_encoder_test.go
+++ b/pkg/sink/codec/csv/csv_encoder_test.go
@@ -18,12 +18,21 @@ import (
 	"testing"
 
 	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
+	sinkhelper "github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
 	"github.com/stretchr/testify/require"
 )
 
+func appendTxnEventForTest(
+	encoder common.TxnEventEncoder,
+	event *commonEvent.DMLEvent,
+	selector commonEvent.Selector,
+) error {
+	return encoder.AppendTxnEvent(sinkhelper.NewRowEvents(event, selector, nil))
+}
+
 func TestCSVBatchCodec(t *testing.T) {
 	s := commonEvent.NewEventTestHelper(t)
 	defer s.Close()
@@ -40,7 +49,7 @@ func TestCSVBatchCodec(t *testing.T) {
 			NullString:      "\\N",
 			IncludeCommitTs: true,
 		})
-		err := encoder.AppendTxnEvent(cs)
+		err := appendTxnEventForTest(encoder, cs, nil)
 		require.Nil(t, err)
 		messages := encoder.Build()
 		if cs.Len() == 0 {
@@ -52,7 +61,7 @@ func TestCSVBatchCodec(t *testing.T) {
 	}
 }
 
-func TestCSVAppendRowChangedEventWithCallback(t *testing.T) {
+func TestCSVAppendTxnEventWithCallback(t *testing.T) {
 	encoder := NewTxnEventEncoder(&common.Config{
 		Delimiter:       ",",
 		Quote:           "\"",
@@ -62,30 +71,28 @@ func TestCSVAppendRowChangedEventWithCallback(t *testing.T) {
 	})
 	require.NotNil(t, encoder)
 
-	count := 0
-
 	s := commonEvent.NewEventTestHelper(t)
 	defer s.Close()
 	s.DDL2Job("create table test.table1(col1 int primary key)")
 	txn := s.DML2Event("test", "table1", "insert into test.table1 values (1)")
-	callback := func() {
-		count += 1
-	}
-	txn.AddPostFlushFunc(callback)
+	count := 0
 
-	// Empty build makes sure that the callback build logic not broken.
+	// Empty build makes sure the build path handles an empty encoder.
 	msgs := encoder.Build()
 	require.Len(t, msgs, 0, "no message should be built and no panic")
 
 	// Append the event.
-	err := encoder.AppendTxnEvent(txn)
+	err := encoder.AppendTxnEvent(sinkhelper.NewRowEvents(txn, nil, func() {
+		count++
+	}))
 	require.Nil(t, err)
 	require.Equal(t, 0, count, "nothing should be called")
 
 	msgs = encoder.Build()
 	require.Len(t, msgs, 1, "expected one message")
+	require.NotNil(t, msgs[0].Callback)
 	msgs[0].Callback()
-	require.Equal(t, 1, count, "expected all callbacks to be called")
+	require.Equal(t, 1, count, "expected one callback be called")
 }
 
 func TestCSVBatchCodecWithHeader(t *testing.T) {
@@ -105,7 +112,7 @@ func TestCSVBatchCodecWithHeader(t *testing.T) {
 		CSVOutputFieldHeader: true,
 	}
 	encoder := NewTxnEventEncoder(cfg)
-	err := encoder.AppendTxnEvent(event)
+	err := appendTxnEventForTest(encoder, event, nil)
 	require.Nil(t, err)
 	messages := encoder.Build()
 	require.Len(t, messages, 1)
@@ -115,7 +122,7 @@ func TestCSVBatchCodecWithHeader(t *testing.T) {
 
 	cfg.CSVOutputFieldHeader = false
 	encoder = NewTxnEventEncoder(cfg)
-	err = encoder.AppendTxnEvent(event)
+	err = appendTxnEventForTest(encoder, event, nil)
 	require.Nil(t, err)
 	messages1 := encoder.Build()
 	require.Len(t, messages1, 1)
@@ -125,7 +132,7 @@ func TestCSVBatchCodecWithHeader(t *testing.T) {
 	cfg.CSVOutputFieldHeader = true
 	event.RowTypes = nil
 	encoder = NewTxnEventEncoder(cfg)
-	err = encoder.AppendTxnEvent(event)
+	err = appendTxnEventForTest(encoder, event, nil)
 	require.Nil(t, err)
 	messages = encoder.Build()
 	require.Len(t, messages, 0)
@@ -154,11 +161,7 @@ func TestCSVTxnEventEncoderWithColumnSelector(t *testing.T) {
 		CSVOutputFieldHeader: true,
 	}
 	encoder := NewTxnEventEncoder(cfg)
-	selectorSetter, ok := encoder.(common.ColumnSelectorAwareTxnEventEncoder)
-	require.True(t, ok)
-	selectorSetter.SetColumnSelector(selectors.GetForTableInfo(event.TableInfo))
-
-	require.NoError(t, encoder.AppendTxnEvent(event))
+	require.NoError(t, appendTxnEventForTest(encoder, event, selectors.GetForTableInfo(event.TableInfo)))
 	messages := encoder.Build()
 	require.Len(t, messages, 1)
 	require.Equal(t, "ticdc-meta$operation,ticdc-meta$table,ticdc-meta$schema,ticdc-meta$commit-ts,col1\n", string(messages[0].Key))

From 99cbaa5c475ec5ca80a9c0113c952860f6d4aac8 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Tue, 7 Jul 2026 07:54:13 +0000
Subject: [PATCH 04/14] .

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 downstreamadapter/sink/cloudstorage/task.go | 11 ++---------
 1 file changed, 2 insertions(+), 9 deletions(-)

diff --git a/downstreamadapter/sink/cloudstorage/task.go b/downstreamadapter/sink/cloudstorage/task.go
index 43b2024a58..7c6a419dfb 100644
--- a/downstreamadapter/sink/cloudstorage/task.go
+++ b/downstreamadapter/sink/cloudstorage/task.go
@@ -51,21 +51,14 @@ type task struct {
 func newDMLTask(
 	version cloudstorage.VersionedTableName,
 	event *commonEvent.DMLEvent,
-	selectors ...commonEvent.Selector,
+	selector commonEvent.Selector,
 ) *task {
-	// The dispatcher path registers progress callbacks before calling
-	// Sink.AddDMLEvent. Put the flush callback into RowEvents before encoding
-	// so txn encoders do not need extra callback parameters.
-	var selector commonEvent.Selector
-	if len(selectors) > 0 && selectors[0] != nil {
-		selector = selectors[0]
-	}
 	return &task{
 		kind:           taskKindDML,
 		postEnqueue:    event.PostEnqueue,
 		tableInfo:      event.TableInfo,
 		versionedTable: version,
-		rowEvents:      helper.NewRowEvents(event, selector, newPostTxnFlushedCallback(event)),
+		rowEvents:      helper.NewRowEvents(event, selector, helper.NewTxnPostFlushRowCallback(event, uint64(event.Length))),
 		dispatcherID:   event.GetDispatcherID(),
 	}
 }

From de9deaecc361c55c46a61a09e69fa9f56959f965 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Tue, 7 Jul 2026 08:19:11 +0000
Subject: [PATCH 05/14] update

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 .../sink/cloudstorage/buffer_manager_test.go  |  2 +-
 .../sink/cloudstorage/encoder_group_test.go   |  3 ++-
 downstreamadapter/sink/cloudstorage/task.go   | 19 +-------------
 .../sink/cloudstorage/task_test.go            | 25 -------------------
 .../sink/cloudstorage/writer_test.go          |  9 ++++++-
 downstreamadapter/sink/helper/mq_row_event.go |  2 +-
 downstreamadapter/sink/helper/row_callback.go |  4 +--
 .../sink/helper/row_callback_test.go          |  2 +-
 downstreamadapter/sink/redo/sink.go           |  2 +-
 9 files changed, 17 insertions(+), 51 deletions(-)

diff --git a/downstreamadapter/sink/cloudstorage/buffer_manager_test.go b/downstreamadapter/sink/cloudstorage/buffer_manager_test.go
index 48518eadd1..370ae3a986 100644
--- a/downstreamadapter/sink/cloudstorage/buffer_manager_test.go
+++ b/downstreamadapter/sink/cloudstorage/buffer_manager_test.go
@@ -171,7 +171,7 @@ func newBufferedTask(table string, dispatcherID commonType.DispatcherID, payload
 		},
 		TableInfoVersion: 1,
 		DispatcherID:     dispatcherID,
-	}, event)
+	}, event, nil)
 	msg := common.NewMsg(nil, []byte(payload))
 	msg.SetRowsCount(1)
 	t.encodedMsgs = []*common.Message{msg}
diff --git a/downstreamadapter/sink/cloudstorage/encoder_group_test.go b/downstreamadapter/sink/cloudstorage/encoder_group_test.go
index 268dcc1963..87b6160362 100644
--- a/downstreamadapter/sink/cloudstorage/encoder_group_test.go
+++ b/downstreamadapter/sink/cloudstorage/encoder_group_test.go
@@ -195,6 +195,7 @@ func TestEncodingGroupEncodeDMLTask(t *testing.T) {
 			DispatcherID:     dispatcherID,
 		},
 		event,
+		nil,
 	)
 	require.NoError(t, group.add(ctx, taskValue))
 
@@ -216,7 +217,7 @@ func TestEncodingGroupEncodeDMLTask(t *testing.T) {
 				require.NotNil(t, task.encodedMsgs[0].Callback)
 				task.encodedMsgs[0].Callback()
 				require.Equal(t, int64(1), flushCount.Load())
-				require.Equal(t, int64(0), enqueueCount.Load())
+				require.Equal(t, int64(1), enqueueCount.Load())
 				require.NotNil(t, task.postEnqueue)
 				task.postEnqueue()
 				require.Equal(t, int64(1), enqueueCount.Load())
diff --git a/downstreamadapter/sink/cloudstorage/task.go b/downstreamadapter/sink/cloudstorage/task.go
index 7c6a419dfb..b29bff7724 100644
--- a/downstreamadapter/sink/cloudstorage/task.go
+++ b/downstreamadapter/sink/cloudstorage/task.go
@@ -58,7 +58,7 @@ func newDMLTask(
 		postEnqueue:    event.PostEnqueue,
 		tableInfo:      event.TableInfo,
 		versionedTable: version,
-		rowEvents:      helper.NewRowEvents(event, selector, helper.NewTxnPostFlushRowCallback(event, uint64(event.Length))),
+		rowEvents:      helper.NewRowEvents(event, selector, event.PostFlush),
 		dispatcherID:   event.GetDispatcherID(),
 	}
 }
@@ -78,23 +78,6 @@ func (t *task) isFlushTask() bool {
 	return t != nil && t.kind == taskKindFlush
 }
 
-func newPostTxnFlushedCallback(event *commonEvent.DMLEvent) func() {
-	if event == nil {
-		return nil
-	}
-	callbacks := append([]func(){}, event.PostTxnFlushed...)
-	if len(callbacks) == 0 {
-		return nil
-	}
-	return func() {
-		for _, f := range callbacks {
-			if f != nil {
-				f()
-			}
-		}
-	}
-}
-
 func (t *task) wait(ctx context.Context) error {
 	if !t.isFlushTask() {
 		return nil
diff --git a/downstreamadapter/sink/cloudstorage/task_test.go b/downstreamadapter/sink/cloudstorage/task_test.go
index 0debca9b4a..ae9d010613 100644
--- a/downstreamadapter/sink/cloudstorage/task_test.go
+++ b/downstreamadapter/sink/cloudstorage/task_test.go
@@ -19,7 +19,6 @@ import (
 	"testing"
 	"time"
 
-	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
 	"github.com/stretchr/testify/require"
 )
 
@@ -49,27 +48,3 @@ func TestFlushMarkerWaitReturnsContextCause(t *testing.T) {
 	err := marker.wait(ctx)
 	require.ErrorIs(t, err, cause)
 }
-
-func TestPostTxnFlushedCallbackDoesNotRunPostEnqueue(t *testing.T) {
-	t.Parallel()
-
-	event := &commonEvent.DMLEvent{}
-	flushCallbackCount := 0
-	enqueueCallbackCount := 0
-	event.AddPostFlushFunc(func() {
-		flushCallbackCount++
-	})
-	event.AddPostEnqueueFunc(func() {
-		enqueueCallbackCount++
-	})
-
-	callback := newPostTxnFlushedCallback(event)
-	require.NotNil(t, callback)
-	callback()
-	require.Equal(t, 1, flushCallbackCount)
-	require.Equal(t, 0, enqueueCallbackCount)
-
-	event.PostEnqueue()
-	event.PostEnqueue()
-	require.Equal(t, 1, enqueueCallbackCount)
-}
diff --git a/downstreamadapter/sink/cloudstorage/writer_test.go b/downstreamadapter/sink/cloudstorage/writer_test.go
index 5eb05a70eb..c641a31e05 100644
--- a/downstreamadapter/sink/cloudstorage/writer_test.go
+++ b/downstreamadapter/sink/cloudstorage/writer_test.go
@@ -128,7 +128,7 @@ func TestWriterRun(t *testing.T) {
 			TableInfo:       tableInfo,
 			Rows:            chunk.MutRowFromValues(100, "hello world").ToRow().Chunk(),
 		}
-		tableTask := newDMLTask(tableName, dmlEvent)
+		tableTask := newDMLTask(tableName, dmlEvent, nil)
 		tableTask.encodedMsgs = []*common.Message{
 			{
 				Value: []byte(fmt.Sprintf(`{"id":%d,"database":"test","table":"table1","pkNames":[],"isDdl":false,`+
@@ -201,6 +201,7 @@ func TestWriterFlushMarker(t *testing.T) {
 			PhysicalTableID: 100,
 			TableInfo:       tableInfo,
 		},
+		nil,
 	)
 	tableTask.encodedMsgs = []*common.Message{msg}
 	require.NoError(t, d.enqueueTask(ctx, tableTask))
@@ -264,6 +265,7 @@ func TestWriterFlushMarkerOnlyFlushesTargetDispatcher(t *testing.T) {
 			PhysicalTableID: 100,
 			TableInfo:       tableInfo,
 		},
+		nil,
 	)
 	msgA := common.NewMsg(nil, []byte(`{"id":"a"}`))
 	msgA.SetRowsCount(1)
@@ -292,6 +294,7 @@ func TestWriterFlushMarkerOnlyFlushesTargetDispatcher(t *testing.T) {
 				},
 			}),
 		},
+		nil,
 	)
 	msgB := common.NewMsg(nil, []byte(`{"id":"b"}`))
 	msgB.SetRowsCount(1)
@@ -361,6 +364,7 @@ func TestWriterPostEnqueueAfterConsume(t *testing.T) {
 			DispatcherID:     dispatcherID,
 		},
 		dmlEvent,
+		nil,
 	)
 	tableTask.encodedMsgs = []*common.Message{
 		{
@@ -505,6 +509,7 @@ func TestWriterStoresPendingMessagesInSpoolBeforeFlush(t *testing.T) {
 			PhysicalTableID: 100,
 			TableInfo:       tableInfo,
 		},
+		nil,
 	)
 	msg := common.NewMsg(nil, []byte(`{"id":1}`))
 	msg.SetRowsCount(1)
@@ -670,6 +675,7 @@ func TestWriterIndexWriteError(t *testing.T) {
 			PhysicalTableID: 100,
 			TableInfo:       tableInfo,
 		},
+		nil,
 	)
 	msg := common.NewMsg(nil, []byte(`{"id":1}`))
 	msg.SetRowsCount(1)
@@ -735,6 +741,7 @@ func TestWriterDataFileCloseError(t *testing.T) {
 			PhysicalTableID: 100,
 			TableInfo:       tableInfo,
 		},
+		nil,
 	)
 
 	var callbackCount atomic.Int64
diff --git a/downstreamadapter/sink/helper/mq_row_event.go b/downstreamadapter/sink/helper/mq_row_event.go
index 9904012eff..fb47ea8b6d 100644
--- a/downstreamadapter/sink/helper/mq_row_event.go
+++ b/downstreamadapter/sink/helper/mq_row_event.go
@@ -26,7 +26,7 @@ func NewMQRowEvents(
 	partitionGenerator partition.Generator,
 	selector commonEvent.Selector,
 ) ([]*commonEvent.MQRowEvent, error) {
-	rowEvents := NewRowEvents(event, selector, NewTxnPostFlushRowCallback(event, uint64(event.Len())))
+	rowEvents := NewRowEvents(event, selector, NewPostFlushRowCallback(event, uint64(event.Len())))
 	events := make([]*commonEvent.MQRowEvent, 0, len(rowEvents))
 
 	for _, rowEvent := range rowEvents {
diff --git a/downstreamadapter/sink/helper/row_callback.go b/downstreamadapter/sink/helper/row_callback.go
index e690ec16c0..9acbc6e671 100644
--- a/downstreamadapter/sink/helper/row_callback.go
+++ b/downstreamadapter/sink/helper/row_callback.go
@@ -18,9 +18,9 @@ import (
 	"go.uber.org/atomic"
 )
 
-// NewTxnPostFlushRowCallback returns a row-level callback that triggers txn-level
+// NewPostFlushRowCallback returns a row-level callback that triggers txn-level
 // PostFlush exactly once when the callback has been invoked totalCount times.
-func NewTxnPostFlushRowCallback(event *event.DMLEvent, totalCount uint64) func() {
+func NewPostFlushRowCallback(event *event.DMLEvent, totalCount uint64) func() {
 	var calledCount atomic.Uint64
 	return func() {
 		if calledCount.Inc() == totalCount {
diff --git a/downstreamadapter/sink/helper/row_callback_test.go b/downstreamadapter/sink/helper/row_callback_test.go
index b4996ba4a4..ddcf076283 100644
--- a/downstreamadapter/sink/helper/row_callback_test.go
+++ b/downstreamadapter/sink/helper/row_callback_test.go
@@ -29,7 +29,7 @@ func TestTxnPostFlushRowCallback(t *testing.T) {
 		flushCount++
 	})
 
-	rowCallback := NewTxnPostFlushRowCallback(event, 3)
+	rowCallback := NewPostFlushRowCallback(event, 3)
 	rowCallback()
 	rowCallback()
 	require.Equal(t, 0, flushCount)
diff --git a/downstreamadapter/sink/redo/sink.go b/downstreamadapter/sink/redo/sink.go
index 101a6e7311..d19fbf9c1e 100644
--- a/downstreamadapter/sink/redo/sink.go
+++ b/downstreamadapter/sink/redo/sink.go
@@ -168,7 +168,7 @@ func (s *Sink) WriteBlockEvent(event commonEvent.BlockEvent) error {
 func (s *Sink) AddDMLEvent(event *commonEvent.DMLEvent) {
 	rowsCount := event.Len()
 	events := make([]*commonEvent.RedoRowEvent, 0, rowsCount)
-	rowCallback := helper.NewTxnPostFlushRowCallback(event, uint64(rowsCount))
+	rowCallback := helper.NewPostFlushRowCallback(event, uint64(rowsCount))
 
 	var (
 		startTs         = event.GetStartTs()

From 5102f18bc74490cd4a72d3b6343681ab3d7dd6ab Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Tue, 7 Jul 2026 09:26:43 +0000
Subject: [PATCH 06/14] add test

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 tests/integration_tests/run_heavy_it_in_ci.sh | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/integration_tests/run_heavy_it_in_ci.sh b/tests/integration_tests/run_heavy_it_in_ci.sh
index de047acc24..262125eacc 100755
--- a/tests/integration_tests/run_heavy_it_in_ci.sh
+++ b/tests/integration_tests/run_heavy_it_in_ci.sh
@@ -166,7 +166,7 @@ storage_groups=(
 	# G13
 	'fail_over region_merge multi_changefeeds'
 	# G14
-	'fail_over_ddl_mix'
+	'column_selector fail_over_ddl_mix'
 	# G15
 	'random_drop_message'
 )

From aa8c082ccb973acf41066c56d9d86960f2ff69ff Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Wed, 8 Jul 2026 12:20:07 +0000
Subject: [PATCH 07/14] add comment

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 downstreamadapter/sink/cloudstorage/task.go | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/downstreamadapter/sink/cloudstorage/task.go b/downstreamadapter/sink/cloudstorage/task.go
index b29bff7724..b5efe56bb9 100644
--- a/downstreamadapter/sink/cloudstorage/task.go
+++ b/downstreamadapter/sink/cloudstorage/task.go
@@ -58,6 +58,10 @@ func newDMLTask(
 		postEnqueue:    event.PostEnqueue,
 		tableInfo:      event.TableInfo,
 		versionedTable: version,
+		// Storage txn encoders attach only the last row callback to the built
+		// batch message, so the callback is triggered once per encoded txn
+		// message. Kafka uses row-level callbacks and counts all rows before
+		// PostFlush, which cannot be reused here for multi-row txns.
 		rowEvents:      helper.NewRowEvents(event, selector, event.PostFlush),
 		dispatcherID:   event.GetDispatcherID(),
 	}

From b547286226f73421d83365ac7850f961ddee1ac1 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Fri, 10 Jul 2026 07:18:56 +0000
Subject: [PATCH 08/14] update

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 api/v2/changefeed.go                          | 12 +++-
 api/v2/changefeed_test.go                     | 46 +++++++++++++++
 downstreamadapter/sink/cloudstorage/task.go   |  7 ++-
 .../sink/columnselector/column_selector.go    | 44 +++++++-------
 .../columnselector/column_selector_test.go    | 57 +++++++++++++++++++
 pkg/common/event/dml_event.go                 | 29 ++++++++++
 pkg/common/event/dml_event_test.go            | 24 ++++++++
 7 files changed, 195 insertions(+), 24 deletions(-)

diff --git a/api/v2/changefeed.go b/api/v2/changefeed.go
index 0c9eb45b3f..4b167d12cb 100644
--- a/api/v2/changefeed.go
+++ b/api/v2/changefeed.go
@@ -1744,7 +1744,7 @@ func getVerifiedTables(
 		return nil, nil, nil, err
 	}
 
-	if err := verifyTable4MQ(replicaConfig, scheme, topic, protocol, tableInfos); err != nil {
+	if err := verifyTablesForSink(replicaConfig, scheme, topic, protocol, tableInfos); err != nil {
 		return nil, nil, nil, err
 	}
 
@@ -1759,13 +1759,21 @@ func getVerifiedTables(
 	return ineligibleTables, eligibleTables, allTables, nil
 }
 
-func verifyTable4MQ(
+func verifyTablesForSink(
 	replicaConfig *config.ReplicaConfig,
 	scheme string,
 	topic string,
 	protocol config.Protocol,
 	tableInfos []*common.TableInfo,
 ) error {
+	if config.IsStorageScheme(scheme) {
+		selectors, err := columnselector.New(replicaConfig.Sink)
+		if err != nil {
+			return err
+		}
+		return selectors.VerifyTables(tableInfos, nil)
+	}
+
 	if !config.IsMQScheme(scheme) {
 		return nil
 	}
diff --git a/api/v2/changefeed_test.go b/api/v2/changefeed_test.go
index 38fd1155b9..a8594ffa1c 100644
--- a/api/v2/changefeed_test.go
+++ b/api/v2/changefeed_test.go
@@ -32,6 +32,10 @@ import (
 	"github.com/pingcap/ticdc/pkg/node"
 	"github.com/pingcap/ticdc/pkg/server"
 	"github.com/pingcap/ticdc/pkg/util"
+	timodel "github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tidb/pkg/types"
 	"github.com/stretchr/testify/require"
 	pd "github.com/tikv/pd/client"
 )
@@ -262,3 +266,45 @@ func TestVerifyRouteConflict(t *testing.T) {
 	require.Contains(t, err.Error(), "source `db1`.`orders`")
 	require.Contains(t, err.Error(), "source `db2`.`orders`")
 }
+
+func TestVerifyTablesForSinkValidatesStorageColumnSelectors(t *testing.T) {
+	t.Parallel()
+
+	replicaCfg := config.GetDefaultReplicaConfig()
+	replicaCfg.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{Matcher: []string{"test.t"}, Columns: []string{"name"}},
+	}
+	tableInfos := []*common.TableInfo{newTableInfoWithPrimaryKeyForTest()}
+
+	err := verifyTablesForSink(replicaCfg, config.FileScheme, "", config.ProtocolCanalJSON, tableInfos)
+	require.Error(t, err)
+	require.True(t, errors.ErrColumnSelectorFailed.Equal(err))
+
+	replicaCfg.Sink.ColumnSelectors[0].Columns = []string{"id", "name"}
+	require.NoError(t, verifyTablesForSink(replicaCfg, config.FileScheme, "", config.ProtocolCanalJSON, tableInfos))
+}
+
+func newTableInfoWithPrimaryKeyForTest() *common.TableInfo {
+	idFieldType := types.NewFieldType(mysql.TypeLong)
+	idFieldType.AddFlag(mysql.PriKeyFlag | mysql.NotNullFlag)
+
+	return common.WrapTableInfo("test", &timodel.TableInfo{
+		ID:         1,
+		Name:       ast.NewCIStr("t"),
+		PKIsHandle: true,
+		Columns: []*timodel.ColumnInfo{
+			{
+				ID:        1,
+				Name:      ast.NewCIStr("id"),
+				FieldType: *idFieldType,
+				State:     timodel.StatePublic,
+			},
+			{
+				ID:        2,
+				Name:      ast.NewCIStr("name"),
+				FieldType: *types.NewFieldType(mysql.TypeVarchar),
+				State:     timodel.StatePublic,
+			},
+		},
+	})
+}
diff --git a/downstreamadapter/sink/cloudstorage/task.go b/downstreamadapter/sink/cloudstorage/task.go
index b5efe56bb9..63ef6acd2b 100644
--- a/downstreamadapter/sink/cloudstorage/task.go
+++ b/downstreamadapter/sink/cloudstorage/task.go
@@ -53,17 +53,18 @@ func newDMLTask(
 	event *commonEvent.DMLEvent,
 	selector commonEvent.Selector,
 ) *task {
+	postEnqueue, postFlush := event.DetachPostCallbacks()
 	return &task{
 		kind:           taskKindDML,
-		postEnqueue:    event.PostEnqueue,
+		postEnqueue:    postEnqueue,
 		tableInfo:      event.TableInfo,
 		versionedTable: version,
 		// Storage txn encoders attach only the last row callback to the built
 		// batch message, so the callback is triggered once per encoded txn
 		// message. Kafka uses row-level callbacks and counts all rows before
 		// PostFlush, which cannot be reused here for multi-row txns.
-		rowEvents:      helper.NewRowEvents(event, selector, event.PostFlush),
-		dispatcherID:   event.GetDispatcherID(),
+		rowEvents:    helper.NewRowEvents(event, selector, postFlush),
+		dispatcherID: event.GetDispatcherID(),
 	}
 }
 
diff --git a/downstreamadapter/sink/columnselector/column_selector.go b/downstreamadapter/sink/columnselector/column_selector.go
index 4e62080632..dc0ddec1d3 100644
--- a/downstreamadapter/sink/columnselector/column_selector.go
+++ b/downstreamadapter/sink/columnselector/column_selector.go
@@ -109,6 +109,8 @@ func (c *ColumnSelectors) GetForTableInfo(tableInfo *common.TableInfo) *ColumnSe
 // VerifyTables return the error if any given table cannot satisfy the column selector constraints.
 // 1. if the column is filter out, it must not be a part of handle key or the unique key.
 // 2. if the filtered out column is used in the column dispatcher, return error.
+//
+// The column dispatcher check is only applied when eventRouter is not nil.
 func (c *ColumnSelectors) VerifyTables(
 	infos []*common.TableInfo, eventRouter *eventrouter.EventRouter,
 ) error {
@@ -130,17 +132,19 @@ func (c *ColumnSelectors) VerifyTables(
 					continue
 				}
 
-				partitionDispatcher := eventRouter.GetPartitionGenerator(table.TableName.Schema, table.TableName.Table)
-				switch v := partitionDispatcher.(type) {
-				case *partition.ColumnsPartitionGenerator:
-					for _, col := range v.Columns {
-						if col == columnInfo.Name.O {
-							return errors.ErrColumnSelectorFailed.GenWithStack(
-								"the filtered out column is used in the column dispatcher, "+
-									"table: %v, column: %s", table.TableName, columnInfo.Name)
+				if eventRouter != nil {
+					partitionDispatcher := eventRouter.GetPartitionGenerator(table.TableName.Schema, table.TableName.Table)
+					switch v := partitionDispatcher.(type) {
+					case *partition.ColumnsPartitionGenerator:
+						for _, col := range v.Columns {
+							if col == columnInfo.Name.O {
+								return errors.ErrColumnSelectorFailed.GenWithStack(
+									"the filtered out column is used in the column dispatcher, "+
+										"table: %v, column: %s", table.TableName, columnInfo.Name)
+							}
 						}
+					default:
 					}
-				default:
 				}
 			}
 
@@ -158,16 +162,18 @@ func (c *ColumnSelectors) VerifyTables(
 func verifyIndices(table *common.TableInfo, retainedColumns map[string]struct{}) bool {
 	primaryKeyColumns := table.GetPrimaryKeyColumnNames()
 
-	retained := true
-	for _, name := range primaryKeyColumns {
-		if _, ok := retainedColumns[name]; !ok {
-			retained = false
-			break
+	if len(primaryKeyColumns) > 0 {
+		retained := true
+		for _, name := range primaryKeyColumns {
+			if _, ok := retainedColumns[name]; !ok {
+				retained = false
+				break
+			}
+		}
+		// primary key columns are retained, return true.
+		if retained {
+			return true
 		}
-	}
-	// primary key columns are retained, return true.
-	if retained {
-		return true
 	}
 
 	// at least one unique key columns are retained, return true.
@@ -176,7 +182,7 @@ func verifyIndices(table *common.TableInfo, retainedColumns map[string]struct{})
 			continue
 		}
 
-		retained = true
+		retained := true
 		for _, col := range index.Columns {
 			if _, ok := retainedColumns[col.Name.O]; !ok {
 				retained = false
diff --git a/downstreamadapter/sink/columnselector/column_selector_test.go b/downstreamadapter/sink/columnselector/column_selector_test.go
index 255704f2ac..2a051b982d 100644
--- a/downstreamadapter/sink/columnselector/column_selector_test.go
+++ b/downstreamadapter/sink/columnselector/column_selector_test.go
@@ -16,9 +16,13 @@ package columnselector
 import (
 	"testing"
 
+	commonType "github.com/pingcap/ticdc/pkg/common"
 	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/ast"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tidb/pkg/types"
 	"github.com/stretchr/testify/require"
 )
 
@@ -182,3 +186,56 @@ func TestColumnSelectorGetSelector(t *testing.T) {
 		}
 	}
 }
+
+func TestVerifyTablesRequiresFullUniqueKey(t *testing.T) {
+	t.Parallel()
+
+	replicaConfig := config.GetDefaultReplicaConfig()
+	replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{
+			Matcher: []string{"test.t"},
+			Columns: []string{"a"},
+		},
+	}
+	selectors, err := New(replicaConfig.Sink)
+	require.NoError(t, err)
+
+	tableInfo := commonType.WrapTableInfo("test", &model.TableInfo{
+		Name: ast.NewCIStr("t"),
+		Columns: []*model.ColumnInfo{
+			newColumnInfoForSelectorTest(1, "a", mysql.NotNullFlag),
+			newColumnInfoForSelectorTest(2, "b", mysql.NotNullFlag),
+		},
+		Indices: []*model.IndexInfo{
+			{
+				Name: ast.NewCIStr("uk_ab"),
+				Columns: []*model.IndexColumn{
+					{Name: ast.NewCIStr("a"), Offset: 0},
+					{Name: ast.NewCIStr("b"), Offset: 1},
+				},
+				Unique: true,
+				State:  model.StatePublic,
+			},
+		},
+	})
+
+	err = selectors.VerifyTables([]*commonType.TableInfo{tableInfo}, nil)
+	require.Error(t, err)
+	require.True(t, errors.ErrColumnSelectorFailed.Equal(err))
+
+	replicaConfig.Sink.ColumnSelectors[0].Columns = []string{"a", "b"}
+	selectors, err = New(replicaConfig.Sink)
+	require.NoError(t, err)
+	require.NoError(t, selectors.VerifyTables([]*commonType.TableInfo{tableInfo}, nil))
+}
+
+func newColumnInfoForSelectorTest(id int64, name string, flag uint) *model.ColumnInfo {
+	ft := types.NewFieldType(mysql.TypeLong)
+	ft.AddFlag(flag)
+	return &model.ColumnInfo{
+		ID:        id,
+		Name:      ast.NewCIStr(name),
+		FieldType: *ft,
+		State:     model.StatePublic,
+	}
+}
diff --git a/pkg/common/event/dml_event.go b/pkg/common/event/dml_event.go
index 9b6e8b7b6e..0d3b85504a 100644
--- a/pkg/common/event/dml_event.go
+++ b/pkg/common/event/dml_event.go
@@ -678,6 +678,35 @@ func (t *DMLEvent) PostEnqueue() {
 	}
 }
 
+// DetachPostCallbacks returns callbacks with the same PostFlush/PostEnqueue
+// semantics as this event, then removes the callback slices from the event.
+// The returned closures intentionally do not capture the DMLEvent, so sinks can
+// keep callbacks after encoding without retaining the event rows.
+func (t *DMLEvent) DetachPostCallbacks() (postEnqueue func(), postFlush func()) {
+	postTxnEnqueued := append([]func(){}, t.PostTxnEnqueued...)
+	postTxnFlushed := append([]func(){}, t.PostTxnFlushed...)
+	t.PostTxnEnqueued = nil
+	t.PostTxnFlushed = nil
+
+	var postEnqueueCalled atomic.Bool
+	postEnqueueCalled.Store(t.postEnqueueCalled.Load())
+	postEnqueue = func() {
+		if !postEnqueueCalled.CAS(false, true) {
+			return
+		}
+		for _, f := range postTxnEnqueued {
+			f()
+		}
+	}
+	postFlush = func() {
+		for _, f := range postTxnFlushed {
+			f()
+		}
+		postEnqueue()
+	}
+	return postEnqueue, postFlush
+}
+
 func (t *DMLEvent) GetSeq() uint64 {
 	return t.Seq
 }
diff --git a/pkg/common/event/dml_event_test.go b/pkg/common/event/dml_event_test.go
index f662e118c6..7a04d9df42 100644
--- a/pkg/common/event/dml_event_test.go
+++ b/pkg/common/event/dml_event_test.go
@@ -463,6 +463,7 @@ func TestDMLEventPostCallbacks(t *testing.T) {
 	t.Run("post flush triggers post enqueue once", verifyDMLEventPostFlushTriggersPostEnqueueOnce)
 	t.Run("post flush order and fallback", verifyDMLEventPostFlushRunsFlushBeforePostEnqueueFallback)
 	t.Run("post enqueue concurrent with post flush", verifyDMLEventPostEnqueueConcurrentWithPostFlush)
+	t.Run("detach callbacks", verifyDMLEventDetachPostCallbacks)
 }
 
 func verifyDMLEventPostFlushTriggersPostEnqueueOnce(t *testing.T) {
@@ -529,3 +530,26 @@ func verifyDMLEventPostEnqueueConcurrentWithPostFlush(t *testing.T) {
 
 	require.Equal(t, int64(1), enqueueCalled.Load())
 }
+
+func verifyDMLEventDetachPostCallbacks(t *testing.T) {
+	t.Parallel()
+
+	event := &DMLEvent{}
+	order := make([]string, 0, 3)
+	event.AddPostFlushFunc(func() {
+		order = append(order, "flush")
+	})
+	event.AddPostEnqueueFunc(func() {
+		order = append(order, "enqueue")
+	})
+
+	postEnqueue, postFlush := event.DetachPostCallbacks()
+	require.Empty(t, event.PostTxnFlushed)
+	require.Empty(t, event.PostTxnEnqueued)
+
+	postFlush()
+	postEnqueue()
+	event.PostFlush()
+
+	require.Equal(t, []string{"flush", "enqueue"}, order)
+}

From 6128efa558c8e3aae8595a907a19d911e841f824 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Tue, 21 Jul 2026 09:15:28 +0000
Subject: [PATCH 09/14] update

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 .../sink/columnselector/column_selector.go           | 12 ++++++------
 pkg/common/event/dml_event.go                        |  4 ++--
 2 files changed, 8 insertions(+), 8 deletions(-)

diff --git a/downstreamadapter/sink/columnselector/column_selector.go b/downstreamadapter/sink/columnselector/column_selector.go
index dc0ddec1d3..df3342f6a9 100644
--- a/downstreamadapter/sink/columnselector/column_selector.go
+++ b/downstreamadapter/sink/columnselector/column_selector.go
@@ -14,6 +14,8 @@
 package columnselector
 
 import (
+	"slices"
+
 	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter/partition"
 	"github.com/pingcap/ticdc/pkg/common"
@@ -136,12 +138,10 @@ func (c *ColumnSelectors) VerifyTables(
 					partitionDispatcher := eventRouter.GetPartitionGenerator(table.TableName.Schema, table.TableName.Table)
 					switch v := partitionDispatcher.(type) {
 					case *partition.ColumnsPartitionGenerator:
-						for _, col := range v.Columns {
-							if col == columnInfo.Name.O {
-								return errors.ErrColumnSelectorFailed.GenWithStack(
-									"the filtered out column is used in the column dispatcher, "+
-										"table: %v, column: %s", table.TableName, columnInfo.Name)
-							}
+						if slices.Contains(v.Columns, columnInfo.Name.O) {
+							return errors.ErrColumnSelectorFailed.GenWithStack(
+								"the filtered out column is used in the column dispatcher, "+
+									"table: %v, column: %s", table.TableName, columnInfo.Name)
 						}
 					default:
 					}
diff --git a/pkg/common/event/dml_event.go b/pkg/common/event/dml_event.go
index 0d3b85504a..bb0f19dcf3 100644
--- a/pkg/common/event/dml_event.go
+++ b/pkg/common/event/dml_event.go
@@ -670,7 +670,7 @@ func (t *DMLEvent) PostFlush() {
 // This stage does not mean data is already written to downstream. The method is
 // idempotent and guarantees enqueue callbacks run at most once.
 func (t *DMLEvent) PostEnqueue() {
-	if !t.postEnqueueCalled.CAS(false, true) {
+	if !t.postEnqueueCalled.CompareAndSwap(false, true) {
 		return
 	}
 	for _, f := range t.PostTxnEnqueued {
@@ -691,7 +691,7 @@ func (t *DMLEvent) DetachPostCallbacks() (postEnqueue func(), postFlush func())
 	var postEnqueueCalled atomic.Bool
 	postEnqueueCalled.Store(t.postEnqueueCalled.Load())
 	postEnqueue = func() {
-		if !postEnqueueCalled.CAS(false, true) {
+		if !postEnqueueCalled.CompareAndSwap(false, true) {
 			return
 		}
 		for _, f := range postTxnEnqueued {

From 529e6add1b1282c488804f7feb9f09de71a42b6c Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Wed, 22 Jul 2026 03:45:22 +0000
Subject: [PATCH 10/14] update

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 downstreamadapter/sink/helper/mq_row_event.go | 28 +++++++++++++++----
 1 file changed, 23 insertions(+), 5 deletions(-)

diff --git a/downstreamadapter/sink/helper/mq_row_event.go b/downstreamadapter/sink/helper/mq_row_event.go
index fb47ea8b6d..bbd8de207c 100644
--- a/downstreamadapter/sink/helper/mq_row_event.go
+++ b/downstreamadapter/sink/helper/mq_row_event.go
@@ -26,12 +26,21 @@ func NewMQRowEvents(
 	partitionGenerator partition.Generator,
 	selector commonEvent.Selector,
 ) ([]*commonEvent.MQRowEvent, error) {
-	rowEvents := NewRowEvents(event, selector, NewPostFlushRowCallback(event, uint64(event.Len())))
-	events := make([]*commonEvent.MQRowEvent, 0, len(rowEvents))
+	callback := NewPostFlushRowCallback(event, uint64(event.Len()))
+	events := make([]*commonEvent.MQRowEvent, 0, event.Len())
+	if selector == nil {
+		selector = columnselector.NewDefaultColumnSelector()
+	}
+
+	for {
+		row, ok := event.GetNextRow()
+		if !ok {
+			event.Rewind()
+			break
+		}
 
-	for _, rowEvent := range rowEvents {
 		index, key, err := partitionGenerator.GeneratePartitionIndexAndKey(
-			&rowEvent.Event, partitionNum, event.TableInfo, event.CommitTs)
+			&row, partitionNum, event.TableInfo, event.CommitTs)
 		if err != nil {
 			return nil, err
 		}
@@ -43,7 +52,16 @@ func NewMQRowEvents(
 				PartitionKey:   key,
 				TotalPartition: partitionNum,
 			},
-			RowEvent: *rowEvent,
+			RowEvent: commonEvent.RowEvent{
+				PhysicalTableID: event.PhysicalTableID,
+				TableInfo:       event.TableInfo,
+				StartTs:         event.StartTs,
+				CommitTs:        event.CommitTs,
+				Event:           row,
+				Callback:        callback,
+				ColumnSelector:  selector,
+				Checksum:        row.Checksum,
+			},
 		})
 	}
 	return events, nil

From 007bc0f28628ec314b1d9237a3fbd3228d8cf2e8 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Wed, 22 Jul 2026 07:14:55 +0000
Subject: [PATCH 11/14] add more test

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 pkg/sink/codec/csv/csv_decoder.go             | 85 +++++++++++++++++++
 pkg/sink/codec/csv/csv_decoder_test.go        | 52 ++++++++++++
 pkg/sink/codec/csv/csv_encoder_test.go        | 49 +++++++++++
 .../conf/changefeed.toml                      | 15 ++++
 .../csv_storage_column_selector/data/data.sql | 45 ++++++++++
 .../csv_storage_column_selector/run.sh        | 59 +++++++++++++
 tests/integration_tests/run_light_it_in_ci.sh |  2 +-
 7 files changed, 306 insertions(+), 1 deletion(-)
 create mode 100644 tests/integration_tests/csv_storage_column_selector/conf/changefeed.toml
 create mode 100644 tests/integration_tests/csv_storage_column_selector/data/data.sql
 create mode 100644 tests/integration_tests/csv_storage_column_selector/run.sh

diff --git a/pkg/sink/codec/csv/csv_decoder.go b/pkg/sink/codec/csv/csv_decoder.go
index 5e5f6b80be..bc6d15ab57 100644
--- a/pkg/sink/codec/csv/csv_decoder.go
+++ b/pkg/sink/codec/csv/csv_decoder.go
@@ -31,6 +31,7 @@ import (
 	"github.com/pingcap/tidb/pkg/lightning/mydump"
 	"github.com/pingcap/tidb/pkg/lightning/worker"
 	"github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/charset"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/types"
@@ -54,6 +55,16 @@ func NewDecoder(ctx context.Context,
 	codecConfig *common.Config,
 	tableInfo *commonType.TableInfo,
 	value []byte,
+) (common.Decoder, error) {
+	return NewDecoderWithColumnSelector(ctx, codecConfig, tableInfo, value, nil)
+}
+
+// NewDecoderWithColumnSelector creates a new BatchDecoder with a column selector.
+func NewDecoderWithColumnSelector(ctx context.Context,
+	codecConfig *common.Config,
+	tableInfo *commonType.TableInfo,
+	value []byte,
+	selector commonEvent.Selector,
 ) (common.Decoder, error) {
 	var backslashEscape bool
 
@@ -78,6 +89,7 @@ func NewDecoder(ctx context.Context,
 	if err != nil {
 		return nil, err
 	}
+	tableInfo = filterTableInfoByColumnSelector(tableInfo, selector)
 	if codecConfig.CSVOutputFieldHeader {
 		err := csvParser.ReadColumns()
 		if err != nil {
@@ -102,6 +114,79 @@ func NewDecoder(ctx context.Context,
 	}, nil
 }
 
+func filterTableInfoByColumnSelector(
+	tableInfo *commonType.TableInfo,
+	selector commonEvent.Selector,
+) *commonType.TableInfo {
+	if tableInfo == nil || selector == nil {
+		return tableInfo
+	}
+
+	columns := tableInfo.GetColumns()
+	selectedColumns := make([]*model.ColumnInfo, 0, len(columns))
+	selectedColumnOffsets := make(map[int64]int, len(columns))
+	for _, col := range columns {
+		if !shouldEncodeColumn(col, selector) {
+			continue
+		}
+		selectedCol := col.Clone()
+		selectedCol.Offset = len(selectedColumns)
+		selectedColumnOffsets[col.ID] = selectedCol.Offset
+		selectedColumns = append(selectedColumns, selectedCol)
+	}
+	if len(selectedColumns) == len(columns) {
+		return tableInfo
+	}
+
+	tidbTableInfo := tableInfo.ToTiDBTableInfo().Clone()
+	tidbTableInfo.Columns = selectedColumns
+	tidbTableInfo.Indices = filterSelectedIndices(tableInfo.GetIndices(), columns, selectedColumnOffsets)
+	tidbTableInfo.PKIsHandle = tableInfo.PKIsHandle() && hasSelectedPKIsHandleColumn(selectedColumns)
+
+	return commonType.NewTableInfo4Decoder(tableInfo.GetSchemaName(), tidbTableInfo)
+}
+
+func filterSelectedIndices(
+	indices []*model.IndexInfo,
+	columns []*model.ColumnInfo,
+	selectedColumnOffsets map[int64]int,
+) []*model.IndexInfo {
+	selectedIndices := make([]*model.IndexInfo, 0, len(indices))
+	for _, index := range indices {
+		selectedIndex := index.Clone()
+		selectedIndex.Columns = make([]*model.IndexColumn, 0, len(index.Columns))
+		allIndexColumnsSelected := true
+		for _, indexColumn := range index.Columns {
+			if indexColumn.Offset >= len(columns) {
+				allIndexColumnsSelected = false
+				break
+			}
+			selectedOffset, ok := selectedColumnOffsets[columns[indexColumn.Offset].ID]
+			if !ok {
+				allIndexColumnsSelected = false
+				break
+			}
+			selectedIndexColumn := indexColumn.Clone()
+			selectedIndexColumn.Offset = selectedOffset
+			selectedIndexColumn.Name = ast.NewCIStr(columns[indexColumn.Offset].Name.O)
+			selectedIndex.Columns = append(selectedIndex.Columns, selectedIndexColumn)
+		}
+		if allIndexColumnsSelected {
+			selectedIndices = append(selectedIndices, selectedIndex)
+		}
+	}
+	return selectedIndices
+}
+
+func hasSelectedPKIsHandleColumn(columns []*model.ColumnInfo) bool {
+	for _, col := range columns {
+		if mysql.HasPriKeyFlag(col.GetFlag()) {
+			return true
+		}
+	}
+	return false
+}
+
 // AddKeyValue implements the Decoder interface.
 func (b *decoder) AddKeyValue(_, _ []byte) {}
 
diff --git a/pkg/sink/codec/csv/csv_decoder_test.go b/pkg/sink/codec/csv/csv_decoder_test.go
index d974f4ecd4..17ddfb1a33 100644
--- a/pkg/sink/codec/csv/csv_decoder_test.go
+++ b/pkg/sink/codec/csv/csv_decoder_test.go
@@ -16,7 +16,9 @@ import (
 	"context"
 	"testing"
 
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
+	"github.com/pingcap/ticdc/pkg/config"
 	"github.com/pingcap/ticdc/pkg/sink/codec/common"
 	"github.com/stretchr/testify/require"
 )
@@ -56,3 +58,53 @@ func TestCSVBatchDecoder(t *testing.T) {
 	_, hasNext := decoder.HasNext()
 	require.False(t, hasNext)
 }
+
+func TestCSVBatchDecoderWithColumnSelector(t *testing.T) {
+	csvData := `"I","t","test",433305438660591626,1,"visible-value"
+`
+	ctx := context.Background()
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+
+	createTableDDL := helper.DDL2Event(
+		"create table test.t(id int primary key, visible varchar(255), secret varchar(255))")
+
+	selectors, err := columnselector.New(&config.SinkConfig{
+		ColumnSelectors: []*config.ColumnSelector{
+			{Matcher: []string{"test.t"}, Columns: []string{"id", "visible"}},
+		},
+	})
+	require.NoError(t, err)
+
+	codecConfig := &common.Config{
+		Delimiter:       ",",
+		Quote:           "\"",
+		Terminator:      "\n",
+		NullString:      "\\N",
+		IncludeCommitTs: true,
+	}
+	decoder, err := NewDecoderWithColumnSelector(
+		ctx,
+		codecConfig,
+		createTableDDL.TableInfo,
+		[]byte(csvData),
+		selectors.GetForTableInfo(createTableDDL.TableInfo),
+	)
+	require.NoError(t, err)
+
+	tp, hasNext := decoder.HasNext()
+	require.True(t, hasNext)
+	require.Equal(t, common.MessageTypeRow, tp)
+
+	event := decoder.NextDMLMessage().ToDMLEvent()
+	require.Len(t, event.TableInfo.GetColumns(), 2)
+	require.Equal(t, "id", event.TableInfo.GetColumns()[0].Name.O)
+	require.Equal(t, "visible", event.TableInfo.GetColumns()[1].Name.O)
+	row, ok := event.GetNextRow()
+	require.True(t, ok)
+	require.Equal(t, int64(1), row.Row.GetInt64(0))
+	require.Equal(t, "visible-value", string(row.Row.GetBytes(1)))
+
+	_, hasNext = decoder.HasNext()
+	require.False(t, hasNext)
+}
diff --git a/pkg/sink/codec/csv/csv_encoder_test.go b/pkg/sink/codec/csv/csv_encoder_test.go
index de1892a3c1..9482b9c868 100644
--- a/pkg/sink/codec/csv/csv_encoder_test.go
+++ b/pkg/sink/codec/csv/csv_encoder_test.go
@@ -168,3 +168,52 @@ func TestCSVTxnEventEncoderWithColumnSelector(t *testing.T) {
 	require.NotContains(t, string(messages[0].Key), "col2")
 	require.NotContains(t, string(messages[0].Value), "filtered")
 }
+
+func TestCSVTxnEventEncoderWithColumnSelectorForUpdateAndDelete(t *testing.T) {
+	helper := commonEvent.NewEventTestHelper(t)
+	defer helper.Close()
+
+	helper.DDL2Event("create table test.table1(id int primary key, visible varchar(255), secret varchar(255))")
+	updateEvent, _ := helper.DML2UpdateEvent(
+		"test",
+		"table1",
+		`insert into test.table1 values (1, "visible-before", "secret-before")`,
+		`update test.table1 set visible = "visible-after", secret = "secret-after" where id = 1`,
+	)
+	deleteEvent := helper.DML2DeleteEvent(
+		"test",
+		"table1",
+		`insert into test.table1 values (2, "delete-visible", "delete-secret")`,
+		`delete from test.table1 where id = 2`,
+	)
+
+	selectors, err := columnselector.New(&config.SinkConfig{
+		ColumnSelectors: []*config.ColumnSelector{
+			{Matcher: []string{"test.table1"}, Columns: []string{"id", "visible"}},
+		},
+	})
+	require.NoError(t, err)
+
+	cfg := &common.Config{
+		Delimiter:       ",",
+		Quote:           "\"",
+		Terminator:      "\n",
+		NullString:      "\\N",
+		OutputOldValue:  true,
+		IncludeCommitTs: false,
+	}
+	selector := selectors.GetForTableInfo(updateEvent.TableInfo)
+	encoder := NewTxnEventEncoder(cfg)
+	require.NoError(t, appendTxnEventForTest(encoder, updateEvent, selector))
+	require.NoError(t, appendTxnEventForTest(encoder, deleteEvent, selector))
+
+	messages := encoder.Build()
+	require.Len(t, messages, 1)
+	value := string(messages[0].Value)
+	require.Contains(t, value, "visible-before")
+	require.Contains(t, value, "visible-after")
+	require.Contains(t, value, "delete-visible")
+	require.NotContains(t, value, "secret-before")
+	require.NotContains(t, value, "secret-after")
+	require.NotContains(t, value, "delete-secret")
+}
diff --git a/tests/integration_tests/csv_storage_column_selector/conf/changefeed.toml b/tests/integration_tests/csv_storage_column_selector/conf/changefeed.toml
new file mode 100644
index 0000000000..836b106543
--- /dev/null
+++ b/tests/integration_tests/csv_storage_column_selector/conf/changefeed.toml
@@ -0,0 +1,15 @@
+[sink]
+protocol = "csv"
+terminator = "\n"
+date-separator = "day"
+column-selectors = [
+    {matcher = ["test.t1"], columns = ["a", "b"]},
+    {matcher = ["test.*"], columns = ["*", "!b"]},
+    {matcher = ["test1.t1"], columns = ["column*", "!column1"]},
+]
+
+[sink.csv]
+delimiter = ","
+quote = '"'
+null = '\N'
+include-commit-ts = true
diff --git a/tests/integration_tests/csv_storage_column_selector/data/data.sql b/tests/integration_tests/csv_storage_column_selector/data/data.sql
new file mode 100644
index 0000000000..def4ddbdf2
--- /dev/null
+++ b/tests/integration_tests/csv_storage_column_selector/data/data.sql
@@ -0,0 +1,45 @@
+drop database if exists `test`;
+create database `test`;
+use `test`;
+
+create table t1 (
+    a int primary key,
+    b varchar(64),
+    c varchar(64)
+);
+
+insert into t1 values (1, 'keep_b_t1', 'filtered_c_t1');
+insert into t1 values (2, 'keep_b_t1_2', 'filtered_c_t1_2');
+
+create table t2 (
+    a int primary key,
+    b varchar(64),
+    c varchar(64)
+);
+
+insert into t2 values (1, 'filtered_b_t2', 'keep_c_t2');
+insert into t2 values (2, 'filtered_b_t2_2', 'keep_c_t2_2');
+
+create table t3 (
+    a int primary key,
+    b varchar(64),
+    c varchar(64)
+);
+
+insert into t3 values (1, 'filtered_b_t3', 'keep_c_t3');
+insert into t3 values (2, 'filtered_b_t3_2', 'keep_c_t3_2');
+
+drop database if exists `test1`;
+create database `test1`;
+use `test1`;
+
+create table t1 (
+    column0 int primary key,
+    column1 varchar(64),
+    column2 varchar(64)
+);
+
+insert into t1 values (1, 'filtered_column1_test1', 'keep_column2_test1');
+insert into t1 values (2, 'filtered_column1_test1_2', 'keep_column2_test1_2');
+
+create table finishmark(id int primary key);
diff --git a/tests/integration_tests/csv_storage_column_selector/run.sh b/tests/integration_tests/csv_storage_column_selector/run.sh
new file mode 100644
index 0000000000..7a0d028fb2
--- /dev/null
+++ b/tests/integration_tests/csv_storage_column_selector/run.sh
@@ -0,0 +1,59 @@
+#!/bin/bash
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+CHANGEFEED_ID="test"
+
+function build_checksum_checker() {
+	echo "Starting build checksum checker..."
+	local pwd=$(pwd)
+	cd $CUR/../../utils/checksum_checker
+	if [ ! -f ./checksum_checker ]; then
+		GO111MODULE=on go build
+	fi
+	cd $pwd
+}
+
+function run_checksum_checker() {
+	build_checksum_checker
+	check_table_exists "test1.finishmark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	$CUR/../../utils/checksum_checker/checksum_checker \
+		--upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" \
+		--downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" \
+		--databases "test,test1" \
+		--config="$CUR/conf/changefeed.toml"
+}
+
+function run() {
+	if [ "$SINK_TYPE" != "storage" ]; then
+		return
+	fi
+
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+	start_tidb_cluster --workdir $WORK_DIR
+
+	# Record TSO before creating test tables to skip system table DDLs.
+	start_ts=$(run_cdc_cli_tso_query $UP_PD_HOST_1 $UP_PD_PORT_1)
+
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	SINK_URI="file://$WORK_DIR/storage_test?flush-interval=5s&protocol=csv"
+	cdc_cli_changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c $CHANGEFEED_ID --config="$CUR/conf/changefeed.toml"
+
+	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_storage_consumer $WORK_DIR $SINK_URI $CUR/conf/changefeed.toml ""
+	run_checksum_checker
+
+	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 54c2078406..d77344531c 100755
--- a/tests/integration_tests/run_light_it_in_ci.sh
+++ b/tests/integration_tests/run_light_it_in_ci.sh
@@ -175,7 +175,7 @@ storage_groups=(
 	# G13
 	'cli_with_auth fail_over_ddl_N maintainer_failover_when_operator'
 	# G14
-	'csv_storage_partition_table csv_storage_multi_tables_ddl table_route fail_over_ddl_O update_changefeed_check_config'
+	'csv_storage_partition_table csv_storage_multi_tables_ddl csv_storage_column_selector table_route fail_over_ddl_O update_changefeed_check_config'
 	# G15
 	'split_region autorandom gc_safepoint'
 )

From 87ad388be55bf5151e0e80ea5abc9029996d8bd8 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Wed, 22 Jul 2026 07:15:58 +0000
Subject: [PATCH 12/14] .

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 cmd/storage-consumer/consumer.go | 17 ++++++++++++++++-
 1 file changed, 16 insertions(+), 1 deletion(-)

diff --git a/cmd/storage-consumer/consumer.go b/cmd/storage-consumer/consumer.go
index 5226d488e3..ab5f922435 100644
--- a/cmd/storage-consumer/consumer.go
+++ b/cmd/storage-consumer/consumer.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/cmd/util"
 	"github.com/pingcap/ticdc/downstreamadapter/sink"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	"github.com/pingcap/ticdc/pkg/cloudstorage"
 	commonType "github.com/pingcap/ticdc/pkg/common"
@@ -68,6 +69,7 @@ type storageMetadata struct {
 type consumer struct {
 	replicationCfg  *config.ReplicaConfig
 	codecCfg        *common.Config
+	columnSelectors *columnselector.ColumnSelectors
 	externalStorage storeapi.Storage
 	fileExtension   string
 	sink            sink.Sink
@@ -132,6 +134,10 @@ func newConsumer(ctx context.Context) (*consumer, error) {
 	if err != nil {
 		return nil, err
 	}
+	columnSelectors, err := columnselector.New(replicaConfig.Sink)
+	if err != nil {
+		return nil, err
+	}
 
 	extension := helper.GetFileExtension(protocol)
 
@@ -162,6 +168,7 @@ func newConsumer(ctx context.Context) (*consumer, error) {
 	return &consumer{
 		replicationCfg:    replicaConfig,
 		codecCfg:          codecConfig,
+		columnSelectors:   columnSelectors,
 		externalStorage:   storage,
 		fileExtension:     extension,
 		sink:              sink,
@@ -328,7 +335,15 @@ func (c *consumer) appendDMLEvents(
 	var decoder common.Decoder
 	switch c.codecCfg.Protocol {
 	case config.ProtocolCsv:
-		decoder, err = csv.NewDecoder(ctx, c.codecCfg, schemaFile.TableInfo(), content)
+		tableInfo := schemaFile.TableInfo()
+		// CSV rows contain selected values without column names, so decode with the same selector.
+		decoder, err = csv.NewDecoderWithColumnSelector(
+			ctx,
+			c.codecCfg,
+			tableInfo,
+			content,
+			c.columnSelectors.GetForTableInfo(tableInfo),
+		)
 		if err != nil {
 			return errors.Trace(err)
 		}

From 36c1f7531ec91502254ac4ac8cd813a842952fa5 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Fri, 24 Jul 2026 09:30:07 +0000
Subject: [PATCH 13/14] chore

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 tests/integration_tests/run_light_it_in_ci.sh | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/integration_tests/run_light_it_in_ci.sh b/tests/integration_tests/run_light_it_in_ci.sh
index d77344531c..515fa68d9b 100755
--- a/tests/integration_tests/run_light_it_in_ci.sh
+++ b/tests/integration_tests/run_light_it_in_ci.sh
@@ -173,9 +173,9 @@ storage_groups=(
 	# G12
 	'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_with_auth csv_storage_column_selector fail_over_ddl_N maintainer_failover_when_operator'
 	# G14
-	'csv_storage_partition_table csv_storage_multi_tables_ddl csv_storage_column_selector table_route fail_over_ddl_O update_changefeed_check_config'
+	'csv_storage_partition_table csv_storage_multi_tables_ddl table_route fail_over_ddl_O update_changefeed_check_config'
 	# G15
 	'split_region autorandom gc_safepoint'
 )

From de742baaf674065b91fe78750a0fc8aae0f16659 Mon Sep 17 00:00:00 2001
From: wk989898 <nhsmwk@gmail.com>
Date: Fri, 24 Jul 2026 16:26:15 +0000
Subject: [PATCH 14/14] update

Signed-off-by: wk989898 <nhsmwk@gmail.com>
---
 tests/integration_tests/run_heavy_it_in_ci.sh | 2 +-
 tests/integration_tests/run_light_it_in_ci.sh | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/tests/integration_tests/run_heavy_it_in_ci.sh b/tests/integration_tests/run_heavy_it_in_ci.sh
index 262125eacc..c8b098f09c 100755
--- a/tests/integration_tests/run_heavy_it_in_ci.sh
+++ b/tests/integration_tests/run_heavy_it_in_ci.sh
@@ -168,7 +168,7 @@ storage_groups=(
 	# G14
 	'column_selector fail_over_ddl_mix'
 	# G15
-	'random_drop_message'
+	'csv_storage_column_selector random_drop_message'
 )
 
 # Source shared functions and check test coverage
diff --git a/tests/integration_tests/run_light_it_in_ci.sh b/tests/integration_tests/run_light_it_in_ci.sh
index 515fa68d9b..54c2078406 100755
--- a/tests/integration_tests/run_light_it_in_ci.sh
+++ b/tests/integration_tests/run_light_it_in_ci.sh
@@ -173,7 +173,7 @@ storage_groups=(
 	# G12
 	'row_format tiflash multi_rocks fail_over_ddl_M correctness_for_shared_column_schema partial_index'
 	# G13
-	'cli_with_auth csv_storage_column_selector fail_over_ddl_N maintainer_failover_when_operator'
+	'cli_with_auth fail_over_ddl_N maintainer_failover_when_operator'
 	# G14
 	'csv_storage_partition_table csv_storage_multi_tables_ddl table_route fail_over_ddl_O update_changefeed_check_config'
 	# G15