Repository: pingcap/tiflow (go)

Issue:
make the case-sensitive works on the whole dispatchers configuration
### Is your feature request related to a problem?

At the moment, there is one configuration named as  [case-sensitive](https://docs.pingcap.com/tidb/stable/ticdc-changefeed-config/#case-sensitive), which controls the database name and table name specified in the dispatcher part.

Let's take the following as an example, the `columns` specified the columns used to calculate the partition. 

It may be hard for the user to specify columns with case if there are a lot of tables. For example, table's may have different columns name to indicate the same thing, such as `ID`, `Id`, `id`, etc.
```
[sink]
dispatchers = [
    {matcher = ['test2.*'], partition = "columns", columns = ["id", "a"]},
]
```

This should also takes effect on the column-selector configurations
```
[sink]
column-selectors = [
    {matcher = ['test.t1'], columns = ['a', 'b']},
    {matcher = ['test.*'], columns = ["*", "!b"]},
    {matcher = ['test1.t1'], columns = ['column*', '!column1']},
    {matcher = ['test3.t'], columns = ["column?", "!column1"]},
]
```

### Describe the feature you'd like

the `case-sensitive` field, should also take affect on the columns configurations.

### Describe alternatives you've considered

_No response_

### Teachability, Documentation, Adoption, Migration Strategy

After this feature is implemented, should also update the document to mention this.

Pull request description:
sink(ticdc): Treat column and index names as case insensitive (#12132)
This is an automated cherry-pick of #12132

<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: close #12103

### What is changed and how it works?
According to [MySQL doc](https://dev.mysql.com/doc/refman/5.7/en/identifier-case-sensitivity.html), column and index names are not case-sensitive on any platform. 

Make cdc compatible with it.

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
Treat column and index names as case-insensitive in dispatchers configuration.
```


Patch:
diff --git a/cdc/kv/client.go b/cdc/kv/client.go
index 3f4bc23de..debf90b8f 100644
--- a/cdc/kv/client.go
+++ b/cdc/kv/client.go
@@ -221,7 +221,7 @@ func NewCDCClient(
 	return c
 }
 
-func (c *CDCClient) newStream(ctx context.Context, addr string, storeID uint64) (stream *eventFeedStream, newStreamErr error) {
+func (c *CDCClient) newStream(ctx context.Context, addr string) (stream *eventFeedStream, newStreamErr error) {
 	streamFunc := func() (err error) {
 		var conn *sharedConn
 		defer func() {
@@ -233,7 +233,7 @@ func (c *CDCClient) newStream(ctx context.Context, addr string, storeID uint64)
 		if err != nil {
 			return errors.Trace(err)
 		}
-		err = version.CheckStoreVersion(ctx, c.pd, storeID)
+		err = version.CheckStoreVersion(ctx, c.pd)
 		if err != nil {
 			return errors.Trace(err)
 		}
@@ -641,7 +641,7 @@ func (s *eventFeedSession) requestRegionToStore(
 			storePendingRegions[storeAddr] = pendingRegions
 			streamCtx, streamCancel := context.WithCancel(ctx)
 			_ = streamCancel // to avoid possible context leak warning from govet
-			stream, err = s.client.newStream(streamCtx, storeAddr, storeID)
+			stream, err = s.client.newStream(streamCtx, storeAddr)
 			if err != nil {
 				// get stream failed, maybe the store is down permanently, we should try to relocate the active store
 				log.Warn("get grpc stream client failed",
diff --git a/cdc/kv/shared_client.go b/cdc/kv/shared_client.go
index 60b3038cd..59d6cea21 100644
--- a/cdc/kv/shared_client.go
+++ b/cdc/kv/shared_client.go
@@ -106,7 +106,6 @@ type rangeTask struct {
 }
 
 type requestedStore struct {
-	storeID    uint64
 	storeAddr  string
 	nextStream atomic.Uint32
 	streams    []*requestedStream
@@ -360,9 +359,8 @@ func (s *SharedClient) requestRegionToStore(ctx context.Context, g *errgroup.Gro
 			continue
 		}
 
-		storeID := sri.rpcCtx.Peer.StoreId
 		storeAddr := sri.rpcCtx.Addr
-		s.appendRequest(s.requestStore(ctx, g, storeID, storeAddr), sri)
+		s.appendRequest(s.requestStore(ctx, g, storeAddr), sri)
 	}
 }
 
@@ -375,16 +373,12 @@ func (s *SharedClient) getRPCContextForRegion(ctx context.Context, id tikv.Regio
 	return rpcCtx, nil
 }
 
-func (s *SharedClient) requestStore(
-	ctx context.Context, g *errgroup.Group,
-	storeID uint64, storeAddr string,
-) *requestedStore {
+func (s *SharedClient) requestStore(ctx context.Context, g *errgroup.Group, storeAddr string) *requestedStore {
 	var rs *requestedStore
 	if rs = s.requestedStores[storeAddr]; rs != nil {
 		return rs
 	}
-
-	rs = &requestedStore{storeID: storeID, storeAddr: storeAddr}
+	rs = &requestedStore{storeAddr: storeAddr}
 	s.requestedStores[storeAddr] = rs
 	for i := uint(0); i < s.config.KVClient.GrpcStreamConcurrent; i++ {
 		stream := newStream(ctx, s, g, rs)
@@ -416,7 +410,6 @@ func (s *SharedClient) appendRequest(r *requestedStore, sri singleRegionInfo) {
 		zap.Uint64("streamID", r.streams[offset].streamID),
 		zap.Any("subscriptionID", sri.requestedTable.subscriptionID),
 		zap.Uint64("regionID", sri.verID.GetID()),
-		zap.Uint64("storeID", r.storeID),
 		zap.String("addr", r.storeAddr))
 	r.streams[offset].requests.In() <- sri
 }
diff --git a/cdc/kv/shared_stream.go b/cdc/kv/shared_stream.go
index 422962889..7d7b82309 100644
--- a/cdc/kv/shared_stream.go
+++ b/cdc/kv/shared_stream.go
@@ -66,7 +66,6 @@ func newStream(ctx context.Context, c *SharedClient, g *errgroup.Group, r *reque
 				zap.String("namespace", c.changefeed.Namespace),
 				zap.String("changefeed", c.changefeed.ID),
 				zap.Uint64("streamID", stream.streamID),
-				zap.Uint64("storeID", r.storeID),
 				zap.String("addr", r.storeAddr))
 		}
 		for {
@@ -89,17 +88,17 @@ func newStream(ctx context.Context, c *SharedClient, g *errgroup.Group, r *reque
 				return err
 			}
 			var regionErr error
-			if err := version.CheckStoreVersion(ctx, c.pd, r.storeID); err != nil {
+			if err := version.CheckStoreVersion(ctx, c.pd); err != nil {
 				log.Info("event feed check store version fails",
 					zap.String("namespace", c.changefeed.Namespace),
 					zap.String("changefeed", c.changefeed.ID),
 					zap.Uint64("streamID", stream.streamID),
-					zap.Uint64("storeID", r.storeID),
 					zap.String("addr", r.storeAddr),
 					zap.Error(err))
 				if errors.Cause(err) == context.Canceled {
 					return nil
-				} else if cerrors.Is(err, cerrors.ErrGetAllStoresFailed) {
+				}
+				if cerrors.Is(err, cerrors.ErrGetAllStoresFailed) {
 					regionErr = &getStoreErr{}
 				} else {
 					regionErr = &sendRequestToStoreErr{}
@@ -156,7 +155,6 @@ func (s *requestedStream) run(ctx context.Context, c *SharedClient, rs *requeste
 		zap.String("namespace", c.changefeed.Namespace),
 		zap.String("changefeed", c.changefeed.ID),
 		zap.Uint64("streamID", s.streamID),
-		zap.Uint64("storeID", rs.storeID),
 		zap.String("addr", rs.storeAddr))
 
 	defer func() {
@@ -164,7 +162,6 @@ func (s *requestedStream) run(ctx context.Context, c *SharedClient, rs *requeste
 			zap.String("namespace", c.changefeed.Namespace),
 			zap.String("changefeed", c.changefeed.ID),
 			zap.Uint64("streamID", s.streamID),
-			zap.Uint64("storeID", rs.storeID),
 			zap.String("addr", rs.storeAddr),
 			zap.Bool("canceled", canceled))
 		if s.multiplexing != nil {
@@ -184,7 +181,6 @@ func (s *requestedStream) run(ctx context.Context, c *SharedClient, rs *requeste
 			zap.String("namespace", c.changefeed.Namespace),
 			zap.String("changefeed", c.changefeed.ID),
 			zap.Uint64("streamID", s.streamID),
-			zap.Uint64("storeID", rs.storeID),
 			zap.String("addr", rs.storeAddr),
 			zap.Error(err))
 		return isCanceled()
@@ -198,7 +194,6 @@ func (s *requestedStream) run(ctx context.Context, c *SharedClient, rs *requeste
 			zap.String("namespace", c.changefeed.Namespace),
 			zap.String("changefeed", c.changefeed.ID),
 			zap.Uint64("streamID", s.streamID),
-			zap.Uint64("storeID", rs.storeID),
 			zap.String("addr", rs.storeAddr))
 		cc.Release()
 
@@ -236,7 +231,6 @@ func (s *requestedStream) receive(
 				zap.String("namespace", c.changefeed.Namespace),
 				zap.String("changefeed", c.changefeed.ID),
 				zap.Uint64("streamID", s.streamID),
-				zap.Uint64("storeID", rs.storeID),
 				zap.String("addr", rs.storeAddr),
 				zap.String("code", grpcstatus.Code(err).String()),
 				zap.Error(err))
@@ -268,7 +262,6 @@ func (s *requestedStream) send(ctx context.Context, c *SharedClient, rs *request
 				zap.Uint64("streamID", s.streamID),
 				zap.Any("subscriptionID", subscriptionID),
 				zap.Uint64("regionID", req.RegionId),
-				zap.Uint64("storeID", rs.storeID),
 				zap.String("addr", rs.storeAddr),
 				zap.Error(err))
 			return errors.Trace(err)
@@ -279,7 +272,6 @@ func (s *requestedStream) send(ctx context.Context, c *SharedClient, rs *request
 			zap.Uint64("streamID", s.streamID),
 			zap.Any("subscriptionID", subscriptionID),
 			zap.Uint64("regionID", req.RegionId),
-			zap.Uint64("storeID", rs.storeID),
 			zap.String("addr", rs.storeAddr))
 		return nil
 	}
@@ -341,7 +333,6 @@ func (s *requestedStream) send(ctx context.Context, c *SharedClient, rs *request
 			zap.Uint64("streamID", s.streamID),
 			zap.Any("subscriptionID", subscriptionID),
 			zap.Uint64("regionID", sri.verID.GetID()),
-			zap.Uint64("storeID", rs.storeID),
 			zap.String("addr", rs.storeAddr))
 		// It means it's a special task for stopping the table.
 		if sri.lockedRange == nil {
diff --git a/cdc/model/mounter.go b/cdc/model/mounter.go
index df4c47a48..e8292b6ef 100644
--- a/cdc/model/mounter.go
+++ b/cdc/model/mounter.go
@@ -16,6 +16,7 @@ package model
 import (
 	"context"
 	"math"
+	"sort"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/types"
@@ -24,8 +25,53 @@ import (
 
 // RowChangedDatums is used to store the changed datums of a row.
 type RowChangedDatums struct {
-	RowDatums    []types.Datum
-	PreRowDatums []types.Datum
+	RowDatums    []types.Datum // row datums (excluding virtual columns)
+	PreRowDatums []types.Datum // pre row datums (excluding virtual columns)
+}
+
+// mergeDatumWithVirtualCols returns a slice of row datums with placeholders for virtual columns placed at the specified offset.
+func mergeDatumWithVirtualCols(datums []types.Datum, virtualColsOffset []int) []types.Datum {
+	if len(virtualColsOffset) == 0 {
+		return datums
+	}
+	if !sort.IntsAreSorted(virtualColsOffset) {
+		log.Panic("virtual column offsets must be sorted",
+			zap.Ints("virtualColsOffset", virtualColsOffset))
+	}
+
+	// checks if virtual column offsets are within valid range.
+	maxAllowedIndex := len(datums) + len(virtualColsOffset)
+	for _, idx := range virtualColsOffset {
+		if idx < 0 || idx >= maxAllowedIndex {
+			log.Panic("invalid virtual column index",
+				zap.Int("index", idx),
+				zap.Int("maxAllowedIndex", maxAllowedIndex-1))
+		}
+	}
+
+	result := make([]types.Datum, 0, maxAllowedIndex)
+	originalIdx := 0
+	virtualIdx := 0
+	for originalIdx < len(datums) || virtualIdx < len(virtualColsOffset) {
+		if virtualIdx < len(virtualColsOffset) && virtualColsOffset[virtualIdx] == len(result) {
+			result = append(result, types.Datum{})
+			virtualIdx++
+		} else if originalIdx < len(datums) {
+			result = append(result, datums[originalIdx])
+			originalIdx++
+		}
+	}
+	return result
+}
+
+// RowDatumsWithVirtualCols returns the row datums with placeholders for virtual columns placed at the specified offset.
+func (r *RowChangedDatums) RowDatumsWithVirtualCols(virtualColsOffset []int) []types.Datum {
+	return mergeDatumWithVirtualCols(r.RowDatums, virtualColsOffset)
+}
+
+// PreRowDatumsWithVirtualCols returns the pre row datums with placeholders for virtual columns placed at the specified offset.
+func (r *RowChangedDatums) PreRowDatumsWithVirtualCols(virtualColsOffset []int) []types.Datum {
+	return mergeDatumWithVirtualCols(r.PreRowDatums, virtualColsOffset)
 }
 
 // IsEmpty returns true if the RowChangeDatums is empty.
diff --git a/cdc/model/schema_storage.go b/cdc/model/schema_storage.go
index d0a74f05d..772373658 100644
--- a/cdc/model/schema_storage.go
+++ b/cdc/model/schema_storage.go
@@ -15,6 +15,7 @@ package model
 
 import (
 	"fmt"
+	"strings"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/parser/model"
@@ -76,6 +77,13 @@ type TableInfo struct {
 	// it's the same length and order with the model.TableInfo.Columns
 	rowColInfos    []rowcodec.ColInfo
 	rowColFieldTps map[int64]*types.FieldType
+
+	// offset of virtual columns in TableInfo.Columns
+	// for example:
+	// Table has 4 columns: a (physical), b (physical), c (virtual), d (virtual)
+	// TableInfo.Columns order: a, b, c, d
+	// VirtualColumnsOffset will be [2, 3] (indices of virtual columns c and d)
+	VirtualColumnsOffset []int
 }
 
 // WrapTableInfo creates a TableInfo from a timodel.TableInfo
@@ -127,7 +135,8 @@ func WrapTableInfo(schemaID int64, schemaName string, version uint64, info *mode
 					ti.handleColID = append(ti.handleColID, id)
 				}
 			}
-
+		} else {
+			ti.VirtualColumnsOffset = append(ti.VirtualColumnsOffset, i)
 		}
 		ti.rowColInfos[i] = rowcodec.ColInfo{
 			ID:            col.ID,
@@ -328,7 +337,7 @@ func (ti *TableInfo) Clone() *TableInfo {
 // GetIndex return the corresponding index by the given name.
 func (ti *TableInfo) GetIndex(name string) *model.IndexInfo {
 	for _, index := range ti.Indices {
-		if index != nil && index.Name.O == name {
+		if index != nil && index.Name.L == strings.ToLower(name) {
 			return index
 		}
 	}
@@ -336,6 +345,9 @@ func (ti *TableInfo) GetIndex(name string) *model.IndexInfo {
 }
 
 // IndexByName returns the index columns and offsets of the corresponding index by name
+// Index is not case-sensitive on any platform.
+// So we always match in lowercase.
+// See also: https://dev.mysql.com/doc/refman/5.7/en/identifier-case-sensitivity.html
 func (ti *TableInfo) IndexByName(name string) ([]string, []int, bool) {
 	index := ti.GetIndex(name)
 	if index == nil {
@@ -352,18 +364,21 @@ func (ti *TableInfo) IndexByName(name string) ([]string, []int, bool) {
 
 // OffsetsByNames returns the column offsets of the corresponding columns by names
 // If any column does not exist, return false
+// Column is not case-sensitive on any platform, nor are column aliases.
+// So we always match in lowercase.
+// See also: https://dev.mysql.com/doc/refman/5.7/en/identifier-case-sensitivity.html
 func (ti *TableInfo) OffsetsByNames(names []string) ([]int, bool) {
 	// todo: optimize it
 	columnOffsets := make(map[string]int, len(ti.Columns))
 	for _, col := range ti.Columns {
 		if col != nil {
-			columnOffsets[col.Name.O] = col.Offset
+			columnOffsets[col.Name.L] = ti.columnsOffset[col.ID]
 		}
 	}
 
 	result := make([]int, 0, len(names))
 	for _, col := range names {
-		offset, ok := columnOffsets[col]
+		offset, ok := columnOffsets[strings.ToLower(col)]
 		if !ok {
 			return nil, false
 		}
diff --git a/cdc/owner/owner.go b/cdc/owner/owner.go
index 6519d3e14..b1a5f7eb9 100644
--- a/cdc/owner/owner.go
+++ b/cdc/owner/owner.go
@@ -352,7 +352,7 @@ func (o *ownerImpl) clusterVersionConsistent(captures map[model.CaptureID]*model
 
 func (o *ownerImpl) handleDrainCaptures(ctx context.Context, query *scheduler.Query, done chan<- error) {
 	if err := o.upstreamManager.Visit(func(upstream *upstream.Upstream) error {
-		if err := version.CheckStoreVersion(ctx, upstream.PDClient, 0); err != nil {
+		if err := version.CheckStoreVersion(ctx, upstream.PDClient); err != nil {
 			return errors.Trace(err)
 		}
 		return nil
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go b/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
index e63c1b17e..c82257d62 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
@@ -69,7 +69,7 @@ func (r *ColumnsDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent,
 		if col == nil {
 			continue
 		}
-		r.hasher.Write([]byte(r.Columns[idx]), []byte(model.ColumnValueString(col.Value)))
+		r.hasher.Write([]byte(col.Name), []byte(model.ColumnValueString(col.Value)))
 	}
 
 	sum32 := r.hasher.Sum32()
diff --git a/go.mod b/go.mod
index 75d2fdb16..d43a83c27 100644
--- a/go.mod
+++ b/go.mod
@@ -395,3 +395,5 @@ replace sourcegraph.com/sourcegraph/appdash => github.com/sourcegraph/appdash v0
 replace sourcegraph.com/sourcegraph/appdash-data => github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67
 
 replace gorm.io/driver/mysql v1.4.5 => gorm.io/driver/mysql v1.3.3
+
+replace github.com/IBM/sarama v1.41.2 => github.com/pingcap/sarama v1.41.2-pingcap-20250617
diff --git a/go.sum b/go.sum
index 371eec87a..ce828e7f1 100644
--- a/go.sum
+++ b/go.sum
@@ -62,8 +62,6 @@ github.com/DataDog/zstd v1.5.5 h1:oWf5W7GtOLgp6bciQYDmhHHjdhYkALu6S/5Ni9ZgSvQ=
 github.com/DataDog/zstd v1.5.5/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
 github.com/HdrHistogram/hdrhistogram-go v1.1.2 h1:5IcZpTvzydCQeHzK4Ef/D5rrSqwxob0t8PQPMybUNFM=
 github.com/HdrHistogram/hdrhistogram-go v1.1.2/go.mod h1:yDgFjdqOqDEKOvasDdhWNXYg9BVp4O+o5f6V/ehm6Oo=
-github.com/IBM/sarama v1.41.2 h1:ZDBZfGPHAD4uuAtSv4U22fRZBgst0eEwGFzLj0fb85c=
-github.com/IBM/sarama v1.41.2/go.mod h1:xdpu7sd6OE1uxNdjYTSKUfY8FaKkJES9/+EyjSgiGQk=
 github.com/Jeffail/gabs/v2 v2.5.1 h1:ANfZYjpMlfTTKebycu4X1AgkVWumFVDYQl7JwOr4mDk=
 github.com/Jeffail/gabs/v2 v2.5.1/go.mod h1:xCn81vdHKxFUuWWAaD5jCTQDNPBMh5pPs9IJ+NcziBI=
 github.com/Joker/hpp v1.0.0/go.mod h1:8x5n+M1Hp5hC0g8okX3sR3vFQwynaX/UgSOM9MeBKzY=
@@ -842,6 +840,8 @@ github.com/pingcap/log v0.0.0-20210625125904-98ed8e2eb1c7/go.mod h1:8AanEdAHATuR
 github.com/pingcap/log v1.1.0/go.mod h1:DWQW5jICDR7UJh4HtxXSM20Churx4CQL0fwL/SoOSA4=
 github.com/pingcap/log v1.1.1-0.20240314023424-862ccc32f18d h1:y3EueKVfVykdpTyfUnQGqft0ud+xVFuCdp1XkVL0X1E=
 github.com/pingcap/log v1.1.1-0.20240314023424-862ccc32f18d/go.mod h1:ORfBOFp1eteu2odzsyaxI+b8TzJwgjwyQcGhI+9SfEA=
+github.com/pingcap/sarama v1.41.2-pingcap-20250617 h1:BkN0GxmbOQZsNXC9TY3MpzpzXJWIRX8Dw+kri4KjJCg=
+github.com/pingcap/sarama v1.41.2-pingcap-20250617/go.mod h1:xdpu7sd6OE1uxNdjYTSKUfY8FaKkJES9/+EyjSgiGQk=
 github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5 h1:T4pXRhBflzDeAhmOQHNPRRogMYxP13V7BkYw3ZsoSfE=
 github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5/go.mod h1:rlimy0GcTvjiJqvD5mXTRr8O2eNZPBrcUgiWVYp9530=
 github.com/pingcap/tidb v1.1.0-beta.0.20241213034006-c7f1b95c537a h1:jEXyHyOHtzUZ/OYSkjxpiAF8mJzIOoH7RH8shmF9J14=
diff --git a/pkg/filter/expr_filter.go b/pkg/filter/expr_filter.go
index 41e771bf3..f00593cff 100644
--- a/pkg/filter/expr_filter.go
+++ b/pkg/filter/expr_filter.go
@@ -22,9 +22,11 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/parser"
+	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/planner/core"
 	"github.com/pingcap/tidb/pkg/sessionctx"
+	"github.com/pingcap/tidb/pkg/table"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/chunk"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
@@ -280,8 +282,9 @@ func (r *dmlExprFilterRule) shouldSkipDML(
 			return false, err
 		}
 		return r.skipDMLByExpression(
-			rawRow.RowDatums,
+			rawRow.RowDatumsWithVirtualCols(ti.VirtualColumnsOffset),
 			exprs,
+			ti,
 		)
 	case row.IsUpdate():
 		oldExprs, err := r.getUpdateOldExpr(ti)
@@ -293,15 +296,17 @@ func (r *dmlExprFilterRule) shouldSkipDML(
 			return false, err
 		}
 		ignoreOld, err := r.skipDMLByExpression(
-			rawRow.PreRowDatums,
+			rawRow.PreRowDatumsWithVirtualCols(ti.VirtualColumnsOffset),
 			oldExprs,
+			ti,
 		)
 		if err != nil {
 			return false, err
 		}
 		ignoreNew, err := r.skipDMLByExpression(
-			rawRow.RowDatums,
+			rawRow.RowDatumsWithVirtualCols(ti.VirtualColumnsOffset),
 			newExprs,
+			ti,
 		)
 		if err != nil {
 			return false, err
@@ -313,8 +318,9 @@ func (r *dmlExprFilterRule) shouldSkipDML(
 			return false, err
 		}
 		return r.skipDMLByExpression(
-			rawRow.PreRowDatums,
+			rawRow.PreRowDatumsWithVirtualCols(ti.VirtualColumnsOffset),
 			exprs,
+			ti,
 		)
 	default:
 		log.Warn("unknown row changed event type")
@@ -322,16 +328,55 @@ func (r *dmlExprFilterRule) shouldSkipDML(
 	}
 }
 
+func (r *dmlExprFilterRule) buildRowWithVirtualColumns(
+	rowData []types.Datum,
+	tableInfo *model.TableInfo,
+) (chunk.Row, error) {
+	row := chunk.MutRowFromDatums(rowData).ToRow()
+	if len(tableInfo.VirtualColumnsOffset) == 0 {
+		// If there is no virtual column, we can return the row directly.
+		return row, nil
+	}
+
+	columns, _, err := expression.ColumnInfos2ColumnsAndNames(r.sessCtx,
+		timodel.CIStr{} /* unused */, tableInfo.Name, tableInfo.Columns, tableInfo.TableInfo)
+	if err != nil {
+		return chunk.Row{}, err
+	}
+
+	vColOffsets, vColFts := collectVirtualColumnOffsetsAndTypes(columns)
+	err = table.FillVirtualColumnValue(vColFts, vColOffsets, columns, tableInfo.Columns, r.sessCtx, row.Chunk())
+	if err != nil {
+		return chunk.Row{}, err
+	}
+	return row, nil
+}
+
+func collectVirtualColumnOffsetsAndTypes(cols []*expression.Column) ([]int, []*types.FieldType) {
+	var offsets []int
+	var fts []*types.FieldType
+	for i, col := range cols {
+		if col.VirtualExpr != nil {
+			offsets = append(offsets, i)
+			fts = append(fts, col.GetType())
+		}
+	}
+	return offsets, fts
+}
+
 func (r *dmlExprFilterRule) skipDMLByExpression(
 	rowData []types.Datum,
 	expr expression.Expression,
+	tableInfo *model.TableInfo,
 ) (bool, error) {
 	if len(rowData) == 0 || expr == nil {
 		return false, nil
 	}
 
-	row := chunk.MutRowFromDatums(rowData).ToRow()
-
+	row, err := r.buildRowWithVirtualColumns(rowData, tableInfo)
+	if err != nil {
+		return false, errors.Trace(err)
+	}
 	d, err := expr.Eval(row)
 	if err != nil {
 		log.Error("failed to eval expression", zap.Error(err))
diff --git a/pkg/sink/codec/canal/canal_json_message.go b/pkg/sink/codec/canal/canal_json_message.go
index 0c2cb30db..3bc21acaa 100644
--- a/pkg/sink/codec/canal/canal_json_message.go
+++ b/pkg/sink/codec/canal/canal_json_message.go
@@ -18,10 +18,12 @@ import (
 	"strings"
 
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
+	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/types"
 	"github.com/pingcap/tiflow/cdc/model"
 	cerrors "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/sink/codec/internal"
+	"github.com/pingcap/tiflow/pkg/sink/codec/utils"
 	canal "github.com/pingcap/tiflow/proto/canal"
 )
 
@@ -153,6 +155,7 @@ func (c *canalJSONMessageWithTiDBExtension) getCommitTs() uint64 {
 func canalJSONMessage2RowChange(msg canalJSONMessageInterface) (*model.RowChangedEvent, error) {
 	result := new(model.RowChangedEvent)
 	result.CommitTs = msg.getCommitTs()
+	result.TableInfo = newTableInfo(msg)
 	result.Table = &model.TableName{
 		Schema: *msg.getSchema(),
 		Table:  *msg.getTable(),
@@ -190,7 +193,6 @@ func canalJSONMessage2RowChange(msg canalJSONMessageInterface) (*model.RowChange
 		}
 	}
 	result.WithHandlePrimaryFlag(msg.pkNameSet())
-
 	return result, nil
 }
 
@@ -265,3 +267,61 @@ func getDDLActionType(query string) timodel.ActionType {
 
 	return timodel.ActionNone
 }
+
+func newTableInfo(msg canalJSONMessageInterface) *model.TableInfo {
+	schemaName := *msg.getSchema()
+	tableName := *msg.getTable()
+	tableInfo := new(timodel.TableInfo)
+	tableInfo.Name = timodel.NewCIStr(tableName)
+
+	columns := newTiColumns(msg)
+	tableInfo.Columns = columns
+	tableInfo.Indices = newTiIndices(columns, msg.pkNameSet())
+	tableInfo.PKIsHandle = len(tableInfo.Indices) != 0
+	return model.WrapTableInfo(100, schemaName, 100, tableInfo)
+}
+
+func newTiColumns(msg canalJSONMessageInterface) []*timodel.ColumnInfo {
+	var nextColumnID int64
+	result := make([]*timodel.ColumnInfo, 0, len(msg.getMySQLType()))
+	for name, mysqlType := range msg.getMySQLType() {
+		col := new(timodel.ColumnInfo)
+		col.ID = nextColumnID
+		col.Name = timodel.NewCIStr(name)
+		if utils.IsBinaryMySQLType(mysqlType) {
+			col.AddFlag(mysql.BinaryFlag)
+		}
+		if _, isPK := msg.pkNameSet()[name]; isPK {
+			col.AddFlag(mysql.PriKeyFlag)
+		}
+		result = append(result, col)
+		nextColumnID++
+	}
+	return result
+}
+
+func newTiIndices(columns []*timodel.ColumnInfo, keys map[string]struct{}) []*timodel.IndexInfo {
+	indexColumns := make([]*timodel.IndexColumn, 0, len(keys))
+	for idx, col := range columns {
+		if mysql.HasPriKeyFlag(col.GetFlag()) {
+			indexColumns = append(indexColumns, &timodel.IndexColumn{
+				Name:   col.Name,
+				Offset: idx,
+			})
+		}
+	}
+
+	result := make([]*timodel.IndexInfo, 0, len(indexColumns))
+	if len(indexColumns) == 0 {
+		return result
+	}
+	indexInfo := &timodel.IndexInfo{
+		ID:      1,
+		Name:    timodel.NewCIStr("primary"),
+		Columns: indexColumns,
+		Primary: true,
+		Unique:  true,
+	}
+	result = append(result, indexInfo)
+	return result
+}
diff --git a/pkg/version/check.go b/pkg/version/check.go
index 043803354..1c6829732 100644
--- a/pkg/version/check.go
+++ b/pkg/version/check.go
@@ -25,7 +25,6 @@ import (
 	"github.com/coreos/go-semver/semver"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
-	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/util/engine"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -80,7 +79,7 @@ func CheckClusterVersion(
 	ctx context.Context, client pd.Client, pdAddrs []string,
 	credential *security.Credential, errorTiKVIncompat bool,
 ) error {
-	err := CheckStoreVersion(ctx, client, 0 /* check all TiKV */)
+	err := CheckStoreVersion(ctx, client)
 	if err != nil {
 		if errorTiKVIncompat {
 			return err
@@ -194,30 +193,23 @@ func checkPDVersion(ctx context.Context, pdAddr string, credential *security.Cre
 	return nil
 }
 
-// CheckStoreVersion checks whether the given TiKV is compatible with this CDC.
-// If storeID is 0, it checks all TiKV.
-func CheckStoreVersion(ctx context.Context, client pd.Client, storeID uint64) error {
+// CheckStoreVersion return whether all TiKV stores compatible to the TiCDC cluster.
+func CheckStoreVersion(ctx context.Context, client pd.Client) error {
 	failpoint.Inject("GetStoreFailed", func() {
-		failpoint.Return(cerror.WrapError(cerror.ErrGetAllStoresFailed, fmt.Errorf("unknown store %d", storeID)))
+		failpoint.Return(cerror.WrapError(cerror.ErrGetAllStoresFailed, errors.New("unknown store")))
 	})
-	var stores []*metapb.Store
-	var err error
-	if storeID == 0 {
-		stores, err = client.GetAllStores(ctx, pd.WithExcludeTombstone())
-	} else {
-		stores = make([]*metapb.Store, 1)
-		stores[0], err = client.GetStore(ctx, storeID)
-	}
+	stores, err := client.GetAllStores(ctx, pd.WithExcludeTombstone())
 	if err != nil {
 		return cerror.WrapError(cerror.ErrGetAllStoresFailed, err)
 	}
 
+	var ver *semver.Version
 	for _, s := range stores {
 		if engine.IsTiFlash(s) {
 			continue
 		}
 
-		ver, err := semver.NewVersion(SanitizeVersion(s.Version))
+		ver, err = semver.NewVersion(SanitizeVersion(s.Version))
 		if err != nil {
 			err = errors.Annotate(err, "invalid TiKV version")
 			return cerror.WrapError(cerror.ErrNewSemVersion, err)


Test patch:
diff --git a/cdc/kv/client_mock_test.go b/cdc/kv/client_mock_test.go
index e077006b7..4889d93ea 100644
--- a/cdc/kv/client_mock_test.go
+++ b/cdc/kv/client_mock_test.go
@@ -32,12 +32,14 @@ type mockPDClient struct {
 
 var _ pd.Client = &mockPDClient{}
 
-func (m *mockPDClient) GetStore(ctx context.Context, storeID uint64) (*metapb.Store, error) {
-	s, err := m.Client.GetStore(ctx, storeID)
+func (m *mockPDClient) GetAllStores(ctx context.Context, _ ...pd.GetStoreOption) ([]*metapb.Store, error) {
+	s, err := m.Client.GetAllStores(ctx)
 	if err != nil {
 		return nil, err
 	}
-	s.Version = m.versionGen()
+	for _, store := range s {
+		store.Version = m.versionGen()
+	}
 	return s, nil
 }
 
diff --git a/cdc/model/mounter_test.go b/cdc/model/mounter_test.go
index dd5e29a3f..0be5ce925 100644
--- a/cdc/model/mounter_test.go
+++ b/cdc/model/mounter_test.go
@@ -17,6 +17,7 @@ import (
 	"math/rand"
 	"testing"
 
+	"github.com/pingcap/tidb/pkg/types"
 	"github.com/stretchr/testify/require"
 )
 
@@ -128,3 +129,37 @@ func TestComparePolymorphicEvents(t *testing.T) {
 		require.True(t, ComparePolymorphicEvents(item.a, item.b))
 	}
 }
+
+func TestMergeDatumWithVirtualCols(t *testing.T) {
+	t.Parallel()
+	datums := []types.Datum{
+		types.NewStringDatum("I"),
+		types.NewStringDatum("employee"),
+		types.NewStringDatum("hr"),
+		types.NewStringDatum("433305438660591626"),
+		types.NewStringDatum("101"),
+		types.NewStringDatum("Smith"),
+		types.NewStringDatum("Bob"),
+		types.NewStringDatum("2014-06-04"),
+		types.NewDatum(nil),
+	}
+	virtualColsOffset := []int{0, 3, 11}
+	mergedDatums := mergeDatumWithVirtualCols(datums, virtualColsOffset)
+
+	// Original length (9) + virtual columns (3) = 12
+	require.Len(t, mergedDatums, 12)
+
+	// Verify positions of virtual columns and original data
+	for _, pos := range virtualColsOffset {
+		// Virtual columns should be empty Datums
+		require.Equal(t, types.Datum{}, mergedDatums[pos],
+			"Virtual column at position %d should be empty", pos)
+	}
+
+	// Verify original data is in correct positions (skipping virtual columns)
+	originalIndices := []int{1, 2, 4, 5, 6, 7, 8, 9, 10}
+	for i, origIdx := range originalIndices {
+		require.Equal(t, datums[i], mergedDatums[origIdx],
+			"Original datum at position %d doesn't match", origIdx)
+	}
+}
diff --git a/cdc/model/schema_storage_test.go b/cdc/model/schema_storage_test.go
index 68f392f9d..e7dcd5db3 100644
--- a/cdc/model/schema_storage_test.go
+++ b/cdc/model/schema_storage_test.go
@@ -286,14 +286,10 @@ func TestIndexByName(t *testing.T) {
 		TableInfo: &timodel.TableInfo{
 			Indices: []*timodel.IndexInfo{
 				{
-					Name: timodel.CIStr{
-						O: "idx1",
-					},
+					Name: timodel.NewCIStr("idx1"),
 					Columns: []*timodel.IndexColumn{
 						{
-							Name: timodel.CIStr{
-								O: "col1",
-							},
+							Name: timodel.NewCIStr("col1"),
 						},
 					},
 				},
@@ -310,46 +306,140 @@ func TestIndexByName(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, []string{"col1"}, names)
 	require.Equal(t, []int{0}, offsets)
+
+	names, offsets, ok = tableInfo.IndexByName("IDX1")
+	require.True(t, ok)
+	require.Equal(t, []string{"col1"}, names)
+	require.Equal(t, []int{0}, offsets)
+
+	names, offsets, ok = tableInfo.IndexByName("Idx1")
+	require.True(t, ok)
+	require.Equal(t, []string{"col1"}, names)
+	require.Equal(t, []int{0}, offsets)
 }
 
 func TestColumnsByNames(t *testing.T) {
-	tableInfo := &TableInfo{
-		TableInfo: &timodel.TableInfo{
-			Columns: []*timodel.ColumnInfo{
-				{
-					Name: timodel.CIStr{
-						O: "col2",
-					},
-					Offset: 1,
-				},
-				{
-					Name: timodel.CIStr{
-						O: "col1",
-					},
-					Offset: 0,
-				},
-				{
-					Name: timodel.CIStr{
-						O: "col3",
-					},
-					Offset: 2,
-				},
+	tableInfo := WrapTableInfo(100, "test", 100, &timodel.TableInfo{
+		Columns: []*timodel.ColumnInfo{
+			{
+				Name:   timodel.NewCIStr("col2"),
+				ID:     1,
+				Offset: 0,
+			},
+			{
+				Name:   timodel.NewCIStr("col1"),
+				ID:     0,
+				Offset: 1,
+			},
+			{
+				Name:   timodel.NewCIStr("col3"),
+				ID:     2,
+				Offset: 2,
 			},
 		},
-	}
+	})
 
 	names := []string{"col1", "col2", "col3"}
 	offsets, ok := tableInfo.OffsetsByNames(names)
 	require.True(t, ok)
-	require.Equal(t, []int{0, 1, 2}, offsets)
+	require.Equal(t, []int{1, 0, 2}, offsets)
 
 	names = []string{"col2"}
 	offsets, ok = tableInfo.OffsetsByNames(names)
 	require.True(t, ok)
-	require.Equal(t, []int{1}, offsets)
+	require.Equal(t, []int{0}, offsets)
 
 	names = []string{"col1", "col-not-found"}
 	offsets, ok = tableInfo.OffsetsByNames(names)
 	require.False(t, ok)
 	require.Nil(t, offsets)
+
+	names = []string{"Col1", "COL2", "CoL3"}
+	offsets, ok = tableInfo.OffsetsByNames(names)
+	require.True(t, ok)
+	require.Equal(t, []int{1, 0, 2}, offsets)
+}
+
+func TestWrapTableInfoWithVirtualColumns(t *testing.T) {
+	t.Parallel()
+	ftNull := parser_types.NewFieldType(mysql.TypeUnspecified)
+	ftNull.SetFlag(mysql.NotNullFlag)
+
+	ftNotNull := parser_types.NewFieldType(mysql.TypeUnspecified)
+	ftNotNull.SetFlag(mysql.NotNullFlag | mysql.MultipleKeyFlag)
+
+	tidbTableInfo := timodel.TableInfo{
+		Columns: []*timodel.ColumnInfo{
+			{
+				Name:      timodel.CIStr{O: "a"},
+				FieldType: *ftNotNull,
+				State:     timodel.StatePublic,
+			},
+			{
+				Name:      timodel.CIStr{O: "b"},
+				FieldType: *ftNotNull,
+				State:     timodel.StatePublic,
+			},
+			{
+				Name:                timodel.CIStr{O: "c"},
+				FieldType:           *ftNull,
+				State:               timodel.StatePublic,
+				GeneratedExprString: "as d",
+				GeneratedStored:     false,
+			},
+			{
+				Name:      timodel.CIStr{O: "d"},
+				FieldType: *ftNotNull,
+				State:     timodel.StatePublic,
+			},
+		},
+		Indices: []*timodel.IndexInfo{
+			{
+				ID: 10,
+				Name: timodel.CIStr{
+					O: "a,b",
+				},
+				Columns: []*timodel.IndexColumn{
+					{Name: timodel.CIStr{O: "a"}, Offset: 0},
+					{Name: timodel.CIStr{O: "b"}, Offset: 1},
+				},
+				Unique: true,
+			},
+			{
+				ID: 9,
+				Name: timodel.CIStr{
+					O: "c",
+				},
+				Columns: []*timodel.IndexColumn{
+					{Name: timodel.CIStr{O: "c"}, Offset: 2},
+				},
+				Unique: true,
+			},
+			{
+				ID: 8,
+				Name: timodel.CIStr{
+					O: "b",
+				},
+				Columns: []*timodel.IndexColumn{
+					{Name: timodel.CIStr{O: "b"}, Offset: 1},
+				},
+				Unique: true,
+			},
+			{
+				ID: 7,
+				Name: timodel.CIStr{
+					O: "d",
+				},
+				Columns: []*timodel.IndexColumn{
+					{Name: timodel.CIStr{O: "d"}, Offset: 3},
+				},
+				Unique: true,
+			},
+		},
+		IsCommonHandle: false,
+		PKIsHandle:     false,
+	}
+
+	tableInfo := WrapTableInfo(100, "test", 1000, &tidbTableInfo)
+	require.Equal(t, []int{2}, tableInfo.VirtualColumnsOffset)
 }
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go b/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
index d6908e4f7..edabd8686 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
@@ -30,30 +30,25 @@ func TestColumnsDispatcher(t *testing.T) {
 			Schema: "test",
 			Table:  "t1",
 		},
-		TableInfo: &model.TableInfo{
-			TableInfo: &timodel.TableInfo{
-				Columns: []*timodel.ColumnInfo{
-					{
-						Name: timodel.CIStr{
-							O: "col2",
-						},
-						Offset: 1,
-					},
-					{
-						Name: timodel.CIStr{
-							O: "col1",
-						},
-						Offset: 0,
-					},
-					{
-						Name: timodel.CIStr{
-							O: "col3",
-						},
-						Offset: 2,
-					},
+		TableInfo: model.WrapTableInfo(100, "test", 100, &timodel.TableInfo{
+			Columns: []*timodel.ColumnInfo{
+				{
+					ID:     0,
+					Name:   timodel.NewCIStr("col2"),
+					Offset: 1,
+				},
+				{
+					ID:     1,
+					Name:   timodel.NewCIStr("col1"),
+					Offset: 0,
+				},
+				{
+					ID:     2,
+					Name:   timodel.NewCIStr("col3"),
+					Offset: 2,
 				},
 			},
-		},
+		}),
 		Columns: []*model.Column{
 			{
 				Name:  "col1",
@@ -78,4 +73,37 @@ func TestColumnsDispatcher(t *testing.T) {
 	index, _, err := p.DispatchRowChangedEvent(event, 16)
 	require.NoError(t, err)
 	require.Equal(t, int32(15), index)
+
+	idx := index
+	p = NewColumnsDispatcher([]string{"COL2", "Col1"})
+	index, _, err = p.DispatchRowChangedEvent(event, 16)
+	require.NoError(t, err)
+	require.Equal(t, idx, index)
+
+	event.TableInfo = model.WrapTableInfo(100, "test", 100, &timodel.TableInfo{
+		Columns: []*timodel.ColumnInfo{
+			{ID: 1, Name: timodel.NewCIStr("COL2"), Offset: 1},
+			{ID: 2, Name: timodel.NewCIStr("Col1"), Offset: 0},
+			{ID: 3, Name: timodel.NewCIStr("col3"), Offset: 2},
+		},
+	})
+	event.Columns = []*model.Column{
+		{
+			Name:  "Col1",
+			Value: 11,
+		},
+		{
+			Name:  "COL2",
+			Value: 22,
+		},
+		{
+			Name:  "col3",
+			Value: 33,
+		},
+	}
+
+	p = NewColumnsDispatcher([]string{"col2", "col1"})
+	index, _, err = p.DispatchRowChangedEvent(event, 16)
+	require.NoError(t, err)
+	require.Equal(t, int32(5), index)
 }
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/index_value_test.go b/cdc/sink/dmlsink/mq/dispatcher/partition/index_value_test.go
index e7dd2e9bd..d072b5796 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/index_value_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/index_value_test.go
@@ -165,28 +165,49 @@ func TestIndexValueDispatcherWithIndexName(t *testing.T) {
 			Schema: "test",
 			Table:  "t1",
 		},
-		TableInfo: &model.TableInfo{
-			TableInfo: &timodel.TableInfo{
-				Indices: []*timodel.IndexInfo{
-					{
-						Name: timodel.CIStr{
-							O: "index1",
-						},
-						Columns: []*timodel.IndexColumn{
-							{
-								Name: timodel.CIStr{
-									O: "a",
-								},
-							},
-						},
+		TableInfo: model.WrapTableInfo(100, "test", 100, &timodel.TableInfo{
+			Indices: []*timodel.IndexInfo{
+				{
+					Name: timodel.NewCIStr("index1"),
+					Columns: []*timodel.IndexColumn{
+						{Name: timodel.NewCIStr("COL2"), Offset: 1},
+						{Name: timodel.NewCIStr("Col1"), Offset: 0},
 					},
+					Primary: true,
 				},
 			},
-		},
+			Columns: []*timodel.ColumnInfo{
+				{
+					ID:     0,
+					Name:   timodel.NewCIStr("COL2"),
+					Offset: 1,
+				},
+				{
+					ID:     1,
+					Name:   timodel.NewCIStr("Col1"),
+					Offset: 0,
+				},
+				{
+					ID:     2,
+					Name:   timodel.NewCIStr("col3"),
+					Offset: 2,
+				},
+			},
+		}),
 		Columns: []*model.Column{
 			{
-				Name:  "a",
+				Name:  "Col1",
 				Value: 11,
+				Flag:  model.HandleKeyFlag,
+			},
+			{
+				Name:  "COL2",
+				Value: 22,
+				Flag:  model.HandleKeyFlag,
+			},
+			{
+				Name:  "col3",
+				Value: 33,
 			},
 		},
 	}
@@ -198,5 +219,16 @@ func TestIndexValueDispatcherWithIndexName(t *testing.T) {
 	p = NewIndexValueDispatcher("index1")
 	index, _, err := p.DispatchRowChangedEvent(event, 16)
 	require.NoError(t, err)
-	require.Equal(t, int32(2), index)
+	require.Equal(t, int32(5), index)
+
+	idx := index
+	p = NewIndexValueDispatcher("INDEX1")
+	index, _, err = p.DispatchRowChangedEvent(event, 16)
+	require.NoError(t, err)
+	require.Equal(t, idx, index)
+
+	p = NewIndexValueDispatcher("")
+	index, _, err = p.DispatchRowChangedEvent(event, 16)
+	require.NoError(t, err)
+	require.Equal(t, idx, index)
 }
diff --git a/pkg/filter/expr_filter_test.go b/pkg/filter/expr_filter_test.go
index f86e28862..c35ae9665 100644
--- a/pkg/filter/expr_filter_test.go
+++ b/pkg/filter/expr_filter_test.go
@@ -14,10 +14,15 @@
 package filter
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/pingcap/errors"
+	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/planner/core"
+	"github.com/pingcap/tidb/pkg/sessionctx"
+	"github.com/pingcap/tidb/pkg/table"
+	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -25,6 +30,29 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// adjustBinaryProtocolForDatumWithoutVirtualCol converts the data in binlog to TiDB datum.
+func adjustBinaryProtocolForDatumWithoutVirtualCol(ctx sessionctx.Context, data []interface{}, cols []*timodel.ColumnInfo) ([]types.Datum, error) {
+	ret := make([]types.Datum, 0, len(data))
+	colIndex := 0
+	for _, d := range data {
+		datum := types.NewDatum(d)
+		// fix the next not virtual column
+		for !model.IsColCDCVisible(cols[colIndex]) {
+			if colIndex >= len(cols) {
+				return nil, fmt.Errorf("colIndex out of bounds")
+			}
+			colIndex++
+		}
+		castDatum, err := table.CastValue(ctx, datum, cols[colIndex], false, false)
+		if err != nil {
+			return nil, err
+		}
+		ret = append(ret, castDatum)
+		colIndex++
+	}
+	return ret, nil
+}
+
 func TestShouldSkipDMLBasic(t *testing.T) {
 	helper := newTestHelper(t)
 	defer helper.close()
@@ -177,6 +205,116 @@ func TestShouldSkipDMLBasic(t *testing.T) {
 				},
 			},
 		},
+		// test table with vitual columns
+		{
+			ddl: "create table test.student_score(id int primary key, name char(50), score int, total_score INT GENERATED ALWAYS AS (score + 10) VIRTUAL, gender char(10))",
+			cfg: &config.FilterConfig{
+				EventFilters: []*config.EventFilterRule{
+					{
+						Matcher:                  []string{"test.student_score"},
+						IgnoreInsertValueExpr:    "score >= 20 or gender = 'female'",
+						IgnoreDeleteValueExpr:    "score >= 32 and gender = 'female'",
+						IgnoreUpdateOldValueExpr: "gender = 'male'",
+						IgnoreUpdateNewValueExpr: "total_score > 38",
+					},
+				},
+			},
+			cases: []innerCase{
+				{ // insert
+					schema: "test",
+					table:  "student_score",
+					columns: []*model.Column{
+						{Name: "none"},
+					},
+					row:    []interface{}{1, "Dongmen", 20, "male"},
+					ignore: true,
+				},
+				{ // insert
+					schema: "test",
+					table:  "student_score",
+					columns: []*model.Column{
+						{Name: "none"},
+					},
+					row:    []interface{}{2, "Rustin", 18, "male"},
+					ignore: false,
+				},
+				{ // insert
+					schema: "test",
+					table:  "student_score",
+					columns: []*model.Column{
+						{Name: "none"},
+					},
+					row:    []interface{}{3, "Susan", 3, "female"},
+					ignore: true,
+				},
+				{ // delete
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.Column{
+						{Name: "none"},
+					},
+					preRow: []interface{}{4, "Helen", 18, "female"},
+					ignore: false,
+				},
+				{ // delete
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.Column{
+						{Name: "none"},
+					},
+					preRow: []interface{}{5, "Madonna", 32, "female"},
+					ignore: true,
+				},
+				{ // delete
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.Column{
+						{Name: "none"},
+					},
+					preRow: []interface{}{6, "Madison", 48, "male"},
+					ignore: false,
+				},
+				{ // update, filler by new value
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.Column{
+						{Name: "none"},
+					},
+					preRow: []interface{}{7, "Marry", 28, "female"},
+					columns: []*model.Column{
+						{Name: "none"},
+					},
+					row:    []interface{}{7, "Marry", 32, "female"},
+					ignore: true,
+				},
+				{ // update
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.Column{
+						{Name: "none"},
+					},
+					preRow: []interface{}{8, "Marilyn", 18, "female"},
+					columns: []*model.Column{
+						{Name: "none"},
+					},
+					row:    []interface{}{8, "Monroe", 22, "female"},
+					ignore: false,
+				},
+				{ // update, filter by old value
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.Column{
+						{Name: "none"},
+					},
+					preRow: []interface{}{9, "Andreja", 25, "male"},
+					columns: []*model.Column{
+						{Name: "none"},
+					},
+					row:    []interface{}{9, "Andreja", 25, "female"},
+					ignore: true,
+				},
+			},
+		},
 		{
 			ddl: "create table test.computer(id int primary key, brand char(50), price int)",
 			cfg: &config.FilterConfig{
@@ -325,9 +463,9 @@ func TestShouldSkipDMLBasic(t *testing.T) {
 		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		for _, c := range tc.cases {
-			rowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.row, tableInfo.Columns)
+			rowDatums, err := adjustBinaryProtocolForDatumWithoutVirtualCol(sessCtx, c.row, tableInfo.Columns)
 			require.Nil(t, err)
-			preRowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.preRow, tableInfo.Columns)
+			preRowDatums, err := adjustBinaryProtocolForDatumWithoutVirtualCol(sessCtx, c.preRow, tableInfo.Columns)
 			require.Nil(t, err)
 			row := &model.RowChangedEvent{
 				Table: &model.TableName{
@@ -442,9 +580,9 @@ func TestShouldSkipDMLError(t *testing.T) {
 		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		for _, c := range tc.cases {
-			rowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.row, tableInfo.Columns)
+			rowDatums, err := adjustBinaryProtocolForDatumWithoutVirtualCol(sessCtx, c.row, tableInfo.Columns)
 			require.Nil(t, err)
-			preRowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.preRow, tableInfo.Columns)
+			preRowDatums, err := adjustBinaryProtocolForDatumWithoutVirtualCol(sessCtx, c.preRow, tableInfo.Columns)
 			require.Nil(t, err)
 			row := &model.RowChangedEvent{
 				Table: &model.TableName{
@@ -638,9 +776,9 @@ func TestShouldSkipDMLTableUpdated(t *testing.T) {
 			if c.updateDDl != "" {
 				tableInfo = helper.execDDL(c.updateDDl)
 			}
-			rowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.row, tableInfo.Columns)
+			rowDatums, err := adjustBinaryProtocolForDatumWithoutVirtualCol(sessCtx, c.row, tableInfo.Columns)
 			require.Nil(t, err)
-			preRowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.preRow, tableInfo.Columns)
+			preRowDatums, err := adjustBinaryProtocolForDatumWithoutVirtualCol(sessCtx, c.preRow, tableInfo.Columns)
 			require.Nil(t, err)
 			row := &model.RowChangedEvent{
 				Table: &model.TableName{
diff --git a/tests/integration_tests/event_filter/conf/cf.toml b/tests/integration_tests/event_filter/conf/cf.toml
index 4597fe3e4..9864716d4 100644
--- a/tests/integration_tests/event_filter/conf/cf.toml
+++ b/tests/integration_tests/event_filter/conf/cf.toml
@@ -6,6 +6,11 @@ matcher = ["event_filter.t1"]
 ignore-event = ["drop table", "delete"]
 ignore-insert-value-expr = "id = 2 or city = 'tokyo'"
 
+# Filter rules for virtual column table
+[[filter.event-filters]]
+matcher = ["event_filter.t_virtual"]
+ignore-insert-value-expr = "id = 2 or category = 'furniture' or is_discounted = true"
+
 [[filter.event-filters]]
 matcher = ["event_filter.t_truncate"]
 ignore-event = ["truncate table"]
@@ -16,4 +21,4 @@ ignore-event = ["alter table"]
 
 [[filter.event-filters]]
 matcher = ["event_filter.t_name*"]
-ignore-event = ["rename table"]
\ No newline at end of file
+ignore-event = ["rename table"]
diff --git a/tests/integration_tests/event_filter/data/test.sql b/tests/integration_tests/event_filter/data/test.sql
index 5ffd396ae..ed03fa994 100644
--- a/tests/integration_tests/event_filter/data/test.sql
+++ b/tests/integration_tests/event_filter/data/test.sql
@@ -38,6 +38,32 @@ WHERE id = 4;
 /* ignore by event type*/
 DROP TABLE t1;
 
+
+-- Table with virtual columns for testing ignore-insert-value-expr
+CREATE TABLE t_virtual (
+    id INT PRIMARY KEY,
+    price DECIMAL(10,2),
+    quantity INT,
+    total_price DECIMAL(10,2) AS (price * quantity) VIRTUAL,
+    discount DECIMAL(10,2) AS (CASE WHEN quantity > 10 THEN price * 0.9 ELSE price END) VIRTUAL,
+    category VARCHAR(20),
+    is_discounted BOOLEAN AS (quantity > 10) VIRTUAL
+);
+
+-- These inserts should be filtered based on the rules:
+-- 1. id = 2
+-- 2. category = 'furniture'
+-- 3. is_discounted = true (quantity > 10)
+INSERT INTO t_virtual (id, price, quantity, category) VALUES (2, 100.00, 5, 'electronics');  -- filtered by id=2
+INSERT INTO t_virtual (id, price, quantity, category) VALUES (3, 200.00, 15, 'furniture');  -- filtered by category and is_discounted
+INSERT INTO t_virtual (id, price, quantity, category) VALUES (5, 300.00, 20, 'electronics'); -- filtered by is_discounted
+INSERT INTO t_virtual (id, price, quantity, category) VALUES (6, 400.00, 12, 'clothing');   -- filtered by is_discounted
+
+-- These inserts should not be filtered
+INSERT INTO t_virtual (id, price, quantity, category) VALUES (1, 50.00, 2, 'clothing');
+INSERT INTO t_virtual (id, price, quantity, category) VALUES (4, 150.00, 1, 'books');
+INSERT INTO t_virtual (id, price, quantity, category) VALUES (7, 250.00, 8, 'electronics');
+
 /* all event of t_normal will be replicated to downstream */
 CREATE TABLE t_normal (
                     id INT,
@@ -110,4 +136,4 @@ CREATE TABLE t_name3 (
                     id INT,
                     name varchar(128),
                     PRIMARY KEY (id)
-);
\ No newline at end of file
+);
diff --git a/tests/integration_tests/event_filter/run.sh b/tests/integration_tests/event_filter/run.sh
index b66cba6eb..e210a5d3b 100644
--- a/tests/integration_tests/event_filter/run.sh
+++ b/tests/integration_tests/event_filter/run.sh
@@ -45,6 +45,7 @@ function run() {
 	check_table_exists "event_filter.t_name1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_table_exists "event_filter.t_name2" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_table_exists "event_filter.t_name3" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_virtual" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
 	# check those rows that are not filtered are synced to downstream
 	run_sql "select count(1) from event_filter.t1;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
@@ -58,6 +59,36 @@ function run() {
 	run_sql "select count(5) from event_filter.t1 where id=4;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_contains "count(5): 1"
 
+	# Add these checks after the existing checks for t_virtual
+	# check virtual column table filtering with virtual column condition
+	run_sql "select count(1) from event_filter.t_virtual;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 3" # Only 3 rows should pass the filter
+
+	run_sql "select count(1) from event_filter.t_virtual where id=1;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 1"
+
+	run_sql "select count(1) from event_filter.t_virtual where id=2;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 0" # filtered by id=2
+
+	run_sql "select count(1) from event_filter.t_virtual where id=3;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 0" # filtered by category and is_discounted
+
+	run_sql "select count(1) from event_filter.t_virtual where id=4;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 1"
+
+	run_sql "select count(1) from event_filter.t_virtual where id=5;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 0" # filtered by is_discounted
+
+	run_sql "select count(1) from event_filter.t_virtual where id=6;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 0" # filtered by is_discounted
+
+	run_sql "select count(1) from event_filter.t_virtual where id=7;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 1"
+
+	# Verify that rows with is_discounted=true are filtered
+	run_sql "select count(1) from event_filter.t_virtual where quantity > 10;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_contains "count(1): 0" # All discounted items should be filtered
+
 	run_sql "TRUNCATE TABLE event_filter.t_truncate;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	run_sql_file $CUR/data/test_truncate.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql "ALTER TABLE event_filter.t_alter MODIFY t_bigint BIGINT;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
diff --git a/tests/integration_tests/mq_sink_dispatcher/conf/changefeed.toml b/tests/integration_tests/mq_sink_dispatcher/conf/changefeed.toml
index 8ee525511..a4f6d601e 100644
--- a/tests/integration_tests/mq_sink_dispatcher/conf/changefeed.toml
+++ b/tests/integration_tests/mq_sink_dispatcher/conf/changefeed.toml
@@ -1,5 +1,6 @@
 [sink]
 dispatchers = [
     {matcher = ['verify.t'], partition = "index-value"},
-    {matcher = ['dispatcher.index'], partition = "index-value", index = "idx_a"}
+    {matcher = ['dispatcher.index'], partition = "index-value", index = "idx_a"},
+    {matcher = ['dispatcher.columns'], partition = "columns", columns = ['id', 'col1']}
 ]
diff --git a/tests/integration_tests/mq_sink_dispatcher/conf/diff_config.toml b/tests/integration_tests/mq_sink_dispatcher/conf/diff_config.toml
index d3cb63bac..8f8292a3f 100644
--- a/tests/integration_tests/mq_sink_dispatcher/conf/diff_config.toml
+++ b/tests/integration_tests/mq_sink_dispatcher/conf/diff_config.toml
@@ -7,7 +7,7 @@ export-fix-sql = true
 check-struct-only = false
 
 [task]
-output-dir = "/tmp/tidb_cdc_test/dispatcher/output"
+output-dir = "/tmp/tidb_cdc_test/mq_sink_dispatcher/output"
 
 source-instances = ["tidb0"]
 
diff --git a/tests/integration_tests/mq_sink_dispatcher/conf/new_changefeed.toml b/tests/integration_tests/mq_sink_dispatcher/conf/new_changefeed.toml
index 4fbc02909..9fd5e983b 100644
--- a/tests/integration_tests/mq_sink_dispatcher/conf/new_changefeed.toml
+++ b/tests/integration_tests/mq_sink_dispatcher/conf/new_changefeed.toml
@@ -1,4 +1,5 @@
 [sink]
 dispatchers = [
-    {matcher = ['dispatcher.index'], partition = "index-value", index = ""}
+    {matcher = ['dispatcher.index'], partition = "index-value", index = ""},
+    {matcher = ['dispatcher.columns'], partition = "columns", columns = ['ID', 'CoL1']}
 ]
diff --git a/tests/integration_tests/mq_sink_dispatcher/run.sh b/tests/integration_tests/mq_sink_dispatcher/run.sh
index 6c234a6e5..1051f2aba 100644
--- a/tests/integration_tests/mq_sink_dispatcher/run.sh
+++ b/tests/integration_tests/mq_sink_dispatcher/run.sh
@@ -21,7 +21,7 @@ function run() {
 
 	cd $WORK_DIR
 
-	TOPIC_NAME="dispatcher-test"
+	TOPIC_NAME="dispatcher-test-$RANDOM"
 
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --loglevel "info"
 
@@ -42,8 +42,10 @@ function run() {
 	run_sql "DROP DATABASE if exists dispatcher;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql "CREATE DATABASE dispatcher;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql "CREATE TABLE dispatcher.index (a int primary key, b int);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "CREATE TABLE dispatcher.columns (id int primary key, col1 int);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
 	run_sql "INSERT INTO dispatcher.index values (1, 2);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "INSERT INTO dispatcher.columns values (1, 2);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
 	ensure $MAX_RETRIES check_changefeed_state http://${UP_PD_HOST_1}:${UP_PD_PORT_1} $changefeed_id "failed" "ErrDispatcherFailed"
 
@@ -62,6 +64,13 @@ function run() {
 	run_sql "UPDATE dispatcher.index set b = 6 where a = 2;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql "DELETE FROM dispatcher.index where a = 3;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
+	# make sure send rows into same partition
+	run_sql "INSERT INTO dispatcher.columns values (2, 3);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "INSERT INTO dispatcher.columns values (4, 5);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "UPDATE dispatcher.columns set col1 = 3 where id = 1;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "UPDATE dispatcher.columns set CoL1 = 4 where ID = 2;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "DELETE FROM dispatcher.columns where Id = 4;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
 	run_sql "CREATE TABLE test.finish_mark (a int primary key);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
diff --git a/tests/integration_tests/syncpoint/run.sh b/tests/integration_tests/syncpoint/run.sh
index 9098a381a..f109423a4 100755
--- a/tests/integration_tests/syncpoint/run.sh
+++ b/tests/integration_tests/syncpoint/run.sh
@@ -138,7 +138,7 @@ function checkDiff() {
 		deployConfig ${primaryArr[$i]} ${secondaryArr[$i]}
 		check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 	done
-	rm $CUR/conf/diff_config.toml
+	rm -f $CUR/conf/diff_config.toml
 }
 
 function run() {