Repository: pingcap/tiflow (go)

Issue:
TiCDC crashes when the ignore_insert_value_expr configuration is enabled for tables containing virtual columns.
### What did you do?

1. create a changefeed with config
```
[filter]
[[filter.event-filters]]
matcher = ["test.t100"]
ignore-insert-value-expr = "zzz = 2"
```

2. create table
```
CREATE TABLE t100 (
    id INT PRIMARY KEY AUTO_INCREMENT,
    name VARCHAR(50),
    price DECIMAL(10,2),
    quantity INT,
    total_price DECIMAL(10,2) AS (price * quantity) VIRTUAL,
    zzz INT
);
```

3. insert data
```
INSERT INTO t100 (name, price, quantity, zzz) VALUES ('desk', 19.99, 5, 100);
```

### What did you expect to see?

TiCDC run normally

### What did you see instead?

TiCDC crash with error:
```
panic: runtime error: index out of range [5] with length 5

goroutine 130939 [running]:
github.com/pingcap/tidb/pkg/util/chunk.Row.IsNull(...)
        github.com/pingcap/tidb@v1.1.0-beta.0.20241213034006-c7f1b95c537a/pkg/util/chunk/row.go:218
github.com/pingcap/tidb/pkg/expression.(*Column).EvalInt(0x6?, {0x588aae0, 0xc0047f8340}, {0xc0083cdf40?, 0x0?})
        github.com/pingcap/tidb@v1.1.0-beta.0.20241213034006-c7f1b95c537a/pkg/expression/column.go:441 +0x29a
github.com/pingcap/tidb/pkg/expression.CompareInt({0x588aae0, 0xc0047f8340}, {0x588a1a8, 0xc004014580}, {0x588a330, 0xc003de1200}, {0xc0083cdf40?, 0xc004014d10?}, {0xc0083cdf40, 0x0})
        github.com/pingcap/tidb@v1.1.0-beta.0.20241213034006-c7f1b95c537a/pkg/expression/builtin_compare.go:2880 +0x5c
github.com/pingcap/tidb/pkg/expression.(*builtinEQIntSig).evalInt(0x425a740?, {0xc0083cdf40?, 0xc004014d10?})
        github.com/pingcap/tidb@v1.1.0-beta.0.20241213034006-c7f1b95c537a/pkg/expression/builtin_compare.go:2381 +0x67
github.com/pingcap/tidb/pkg/expression.(*ScalarFunction).EvalInt(0xc000928d90, {0x588aae0, 0xc0047f8340}, {0xc0083cdf40?, 0x0?})
        github.com/pingcap/tidb@v1.1.0-beta.0.20241213034006-c7f1b95c537a/pkg/expression/scalar_function.go:457 +0x92
github.com/pingcap/tidb/pkg/expression.(*ScalarFunction).Eval(0xc000928d90, {0xc0083cdf40?, 0xc007e115a0?})
        github.com/pingcap/tidb@v1.1.0-beta.0.20241213034006-c7f1b95c537a/pkg/expression/scalar_function.go:413 +0x249
github.com/pingcap/tiflow/pkg/filter.(*dmlExprFilterRule).skipDMLByExpression(0xc0045ff560?, {0xc00567a300?, 0xc007d51960?, 0x9?}, {0x588a4b8, 0xc000928d90})
        github.com/pingcap/tiflow/pkg/filter/expr_filter.go:335 +0xb5
github.com/pingcap/tiflow/pkg/filter.(*dmlExprFilterRule).shouldSkipDML(0xc0045ff560, 0xc000d43ce0, {{0xc00567a300, 0x5, 0x5}, {0x0, 0x0, 0x0}}, 0xc003e40270)
        github.com/pingcap/tiflow/pkg/filter/expr_filter.go:282 +0x4aa
github.com/pingcap/tiflow/pkg/filter.(*dmlExprFilter).shouldSkipDML(0xc0047f8cc0?, 0xc000d43ce0, {{0xc00567a300, 0x5, 0x5}, {0x0, 0x0, 0x0}}, 0xc003e40270)
        github.com/pingcap/tiflow/pkg/filter/expr_filter.go:428 +0x14b
github.com/pingcap/tiflow/pkg/filter.(*filter).ShouldIgnoreDMLEvent(0xc0047f8cc0, 0xc000d43ce0, {{0xc00567a300, 0x5, 0x5}, {0x0, 0x0, 0x0}}, 0x0?)
        github.com/pingcap/tiflow/pkg/filter/filter.go:157 +0x106
github.com/pingcap/tiflow/cdc/entry.(*mounter).unmarshalAndMountRowChanged.func1(0xd309e5?, 0x6c, 0xc000928460, {0xc004bb3e99, 0xa, 0x2?}, 0xc003913b00, {0x65ec6e003a00009, 0x65ec6e003a0000a, 0x6c, ...})
        github.com/pingcap/tiflow/cdc/entry/mounter.go:193 +0x3e7
github.com/pingcap/tiflow/cdc/entry.(*mounter).unmarshalAndMountRowChanged(0xc003913b00, {0x57fd3f0, 0xc004bfe960}, 0xc000928460)
        github.com/pingcap/tiflow/cdc/entry/mounter.go:206 +0x46d
github.com/pingcap/tiflow/cdc/entry.(*mounter).DecodeEvent(0xc003913b00, {0x57fd3f0, 0xc004bfe960}, 0xc004e3bb90)
        github.com/pingcap/tiflow/cdc/entry/mounter.go:118 +0x5e
github.com/pingcap/tiflow/cdc/entry.(*mounterGroup).runWorker(0xc00131c6c0, {0x57fd3f0, 0xc004bfe960})
        github.com/pingcap/tiflow/cdc/entry/mounter_group.go:121 +0x138
github.com/pingcap/tiflow/cdc/entry.(*mounterGroup).Run.func2()
        github.com/pingcap/tiflow/cdc/entry/mounter_group.go:87 +0x25
golang.org/x/sync/errgroup.(*Group).Go.func1()
        golang.org/x/sync@v0.9.0/errgroup/errgroup.go:78 +0x56
created by golang.org/x/sync/errgroup.(*Group).Go in goroutine 130912
        golang.org/x/sync@v0.9.0/errgroup/errgroup.go:75 +0x96
panic: runtime error: index out of range [5] with length 5
```

### Versions of the cluster

Upstream TiDB cluster version (execute `SELECT tidb_version();` in a MySQL client):

```console
(paste TiDB cluster version here)
```

Upstream TiKV version (execute `tikv-server --version`):

```console
(paste TiKV version here)
```

TiCDC version (execute `cdc version`):

```console
(paste TiCDC version here)
```

Pull request:
event filter: fix panic when evaluate expressions for table with virtual columns
<!--
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 #12206 

### What is changed and how it works?
Evaluate and fill the value of virtual columns before evaluate the expression on the row.

### 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
Fix panic when evaluating event-filter expressions on tables with virtual columns.
```


Patch:
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 48a3364a4..dad065a01 100644
--- a/cdc/model/schema_storage.go
+++ b/cdc/model/schema_storage.go
@@ -90,8 +90,12 @@ type TableInfo struct {
 	// only for new row format decoder
 	handleColID []int64
 
-	// number of virtual columns
-	virtualColumnCount int
+	// 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
 	// rowColInfosWithoutVirtualCols is the same as rowColInfos, but without virtual columns
 	rowColInfosWithoutVirtualCols *[]rowcodec.ColInfo
 }
@@ -121,7 +125,6 @@ func WrapTableInfo(schemaID int64, schemaName string, version uint64, info *mode
 
 	rowColumnsCurrentOffset := 0
 
-	ti.virtualColumnCount = 0
 	for i, col := range ti.Columns {
 		ti.columnsOffset[col.ID] = i
 		pkIsHandle := false
@@ -146,7 +149,7 @@ func WrapTableInfo(schemaID int64, schemaName string, version uint64, info *mode
 				}
 			}
 		} else {
-			ti.virtualColumnCount += 1
+			ti.VirtualColumnsOffset = append(ti.VirtualColumnsOffset, i)
 		}
 		ti.rowColInfos[i] = rowcodec.ColInfo{
 			ID:            col.ID,
@@ -182,21 +185,21 @@ func WrapTableInfo(schemaID int64, schemaName string, version uint64, info *mode
 }
 
 func (ti *TableInfo) initRowColInfosWithoutVirtualCols() {
-	if ti.virtualColumnCount == 0 {
+	if len(ti.VirtualColumnsOffset) == 0 {
 		ti.rowColInfosWithoutVirtualCols = &ti.rowColInfos
 		return
 	}
-	colInfos := make([]rowcodec.ColInfo, 0, len(ti.rowColInfos)-ti.virtualColumnCount)
+	colInfos := make([]rowcodec.ColInfo, 0, len(ti.rowColInfos)-len(ti.VirtualColumnsOffset))
 	for i, col := range ti.Columns {
 		if IsColCDCVisible(col) {
 			colInfos = append(colInfos, ti.rowColInfos[i])
 		}
 	}
-	if len(colInfos) != len(ti.rowColInfos)-ti.virtualColumnCount {
+	if len(colInfos) != len(ti.rowColInfos)-len(ti.VirtualColumnsOffset) {
 		log.Panic("invalid rowColInfosWithoutVirtualCols",
 			zap.Int("len(colInfos)", len(colInfos)),
 			zap.Int("len(ti.rowColInfos)", len(ti.rowColInfos)),
-			zap.Int("ti.virtualColumnCount", ti.virtualColumnCount))
+			zap.Any("ti.VirtualColumnsOffset", ti.VirtualColumnsOffset))
 	}
 	ti.rowColInfosWithoutVirtualCols = &colInfos
 }
@@ -388,7 +391,7 @@ func (ti *TableInfo) HasUniqueColumn() bool {
 
 // HasVirtualColumns returns whether the table has virtual columns
 func (ti *TableInfo) HasVirtualColumns() bool {
-	return ti.virtualColumnCount > 0
+	return len(ti.VirtualColumnsOffset) > 0
 }
 
 // IsEligible returns whether the table is a eligible table
diff --git a/pkg/filter/expr_filter.go b/pkg/filter/expr_filter.go
index 5ff6a58b3..ceb6eb06d 100644
--- a/pkg/filter/expr_filter.go
+++ b/pkg/filter/expr_filter.go
@@ -21,7 +21,9 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/parser"
+	"github.com/pingcap/tidb/pkg/parser/ast"
 	"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"
 	"github.com/pingcap/tidb/pkg/util/dbterror/plannererrors"
@@ -274,8 +276,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)
@@ -287,15 +290,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
@@ -307,8 +312,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")
@@ -316,16 +322,60 @@ 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.GetExprCtx(),
+		ast.CIStr{} /* unused */, tableInfo.Name, tableInfo.Columns, tableInfo.TableInfo)
+	if err != nil {
+		return chunk.Row{}, err
+	}
+	var fts []*types.FieldType
+	for _, col := range columns {
+		fts = append(fts, col.GetType(r.sessCtx.GetExprCtx().GetEvalCtx()))
+	}
+	ch := chunk.NewEmptyChunk(fts)
+	ch.AppendRow(row)
+
+	vColOffsets, vColFts := collectVirtualColumnOffsetsAndTypes(r.sessCtx.GetExprCtx().GetEvalCtx(), columns)
+	err = table.FillVirtualColumnValue(vColFts, vColOffsets, columns, tableInfo.Columns, r.sessCtx.GetExprCtx(), ch)
+	if err != nil {
+		return chunk.Row{}, err
+	}
+	return ch.GetRow(0), nil
+}
+
+func collectVirtualColumnOffsetsAndTypes(ctx expression.EvalContext, 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(ctx))
+		}
+	}
+	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(r.sessCtx.GetExprCtx().GetEvalCtx(), row)
 	if err != nil {
 		log.Error("failed to eval expression", zap.Error(err))


Test patch:
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 33a31aadd..be6fb5bfc 100644
--- a/cdc/model/schema_storage_test.go
+++ b/cdc/model/schema_storage_test.go
@@ -511,7 +511,7 @@ func TestBuildTiDBTableInfoWithUniqueKey(t *testing.T) {
 	require.Equal(t, columns[3].Flag, *tableInfo.ForceGetColumnFlagType(tableInfo.Columns[3].ID))
 }
 
-func TestBuildTiDBTableInfoWithoutVirtualColumns(t *testing.T) {
+func TestBuildTiDBTableInfoWithVirtualColumns(t *testing.T) {
 	t.Parallel()
 	ftNull := parser_types.NewFieldType(mysql.TypeUnspecified)
 	ftNull.SetFlag(mysql.NotNullFlag)
@@ -519,7 +519,7 @@ func TestBuildTiDBTableInfoWithoutVirtualColumns(t *testing.T) {
 	ftNotNull := parser_types.NewFieldType(mysql.TypeUnspecified)
 	ftNotNull.SetFlag(mysql.NotNullFlag | mysql.MultipleKeyFlag)
 
-	tableInfo := timodel.TableInfo{
+	tidbTableInfo := timodel.TableInfo{
 		Columns: []*timodel.ColumnInfo{
 			{
 				Name:      pmodel.CIStr{O: "a"},
@@ -590,7 +590,8 @@ func TestBuildTiDBTableInfoWithoutVirtualColumns(t *testing.T) {
 		IsCommonHandle: false,
 		PKIsHandle:     false,
 	}
-	infoWithourVirtualCols := BuildTiDBTableInfoWithoutVirtualColumns(&tableInfo)
+	// test BuildTiDBTableInfoWithoutVirtualColumns
+	infoWithourVirtualCols := BuildTiDBTableInfoWithoutVirtualColumns(&tidbTableInfo)
 	require.Equal(t, 3, len(infoWithourVirtualCols.Columns))
 	require.Equal(t, 0, infoWithourVirtualCols.Columns[0].Offset)
 	require.Equal(t, "a", infoWithourVirtualCols.Columns[0].Name.O)
@@ -598,4 +599,7 @@ func TestBuildTiDBTableInfoWithoutVirtualColumns(t *testing.T) {
 	require.Equal(t, "b", infoWithourVirtualCols.Columns[1].Name.O)
 	require.Equal(t, 2, infoWithourVirtualCols.Columns[2].Offset)
 	require.Equal(t, "d", infoWithourVirtualCols.Columns[2].Name.O)
+
+	tableInfo := WrapTableInfo(100, "test", 1000, &tidbTableInfo)
+	require.Equal(t, []int{2}, tableInfo.VirtualColumnsOffset)
 }
diff --git a/pkg/filter/expr_filter_test.go b/pkg/filter/expr_filter_test.go
index dd7ee4c34..843a1f928 100644
--- a/pkg/filter/expr_filter_test.go
+++ b/pkg/filter/expr_filter_test.go
@@ -14,9 +14,14 @@
 package filter
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/pingcap/errors"
+	timodel "github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tidb/pkg/sessionctx"
+	"github.com/pingcap/tidb/pkg/table"
+	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/dbterror/plannererrors"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
@@ -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.ColumnData{
+						{ColumnID: 0},
+					},
+					row:    []interface{}{1, "Dongmen", 20, "male"},
+					ignore: true,
+				},
+				{ // insert
+					schema: "test",
+					table:  "student_score",
+					columns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					row:    []interface{}{2, "Rustin", 18, "male"},
+					ignore: false,
+				},
+				{ // insert
+					schema: "test",
+					table:  "student_score",
+					columns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					row:    []interface{}{3, "Susan", 3, "female"},
+					ignore: true,
+				},
+				{ // delete
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					preRow: []interface{}{4, "Helen", 18, "female"},
+					ignore: false,
+				},
+				{ // delete
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					preRow: []interface{}{5, "Madonna", 32, "female"},
+					ignore: true,
+				},
+				{ // delete
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					preRow: []interface{}{6, "Madison", 48, "male"},
+					ignore: false,
+				},
+				{ // update, filler by new value
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					preRow: []interface{}{7, "Marry", 28, "female"},
+					columns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					row:    []interface{}{7, "Marry", 32, "female"},
+					ignore: true,
+				},
+				{ // update
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					preRow: []interface{}{8, "Marilyn", 18, "female"},
+					columns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					row:    []interface{}{8, "Monroe", 22, "female"},
+					ignore: false,
+				},
+				{ // update, filter by old value
+					schema: "test",
+					table:  "student_score",
+					preColumns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					preRow: []interface{}{9, "Andreja", 25, "male"},
+					columns: []*model.ColumnData{
+						{ColumnID: 0},
+					},
+					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)
 		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{
 				TableInfo: &model.TableInfo{
@@ -444,9 +582,9 @@ func TestShouldSkipDMLError(t *testing.T) {
 		f, err := newExprFilter("", tc.cfg)
 		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{
 				TableInfo: &model.TableInfo{
@@ -642,9 +780,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{
 				TableInfo: &model.TableInfo{
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 4cfa71b1a..487bab15e 100644
--- a/tests/integration_tests/event_filter/run.sh
+++ b/tests/integration_tests/event_filter/run.sh
@@ -48,6 +48,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}
@@ -61,6 +62,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/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() {