Repository: pingcap/tiflow (go)

Issue:
CDC panics when configuring the column dispatcher with virtual columns
### What did you do?

1. Start a changefeed with config
```
[sink]
dispatchers = [
    {matcher = ['dispatcher.columns'], partition = "columns", columns = ['ID', 'CoL1', 'D']}
]
```
2. run sql
```
CREATE TABLE dispatcher.columns (id int primary key, col1 int, d int AS (2 * id));
INSERT INTO dispatcher.columns(id, col1) values (1, 2);
```

### What did you expect to see?

Everything is ok

### What did you see instead?

```
panic: runtime error: index out of range [2] with length 2

goroutine 1510 [running]:
github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher/partition.(*ColumnsDispatcher).DispatchRowChangedEvent(0xc0053c5440, 0xc005bb8090, 0x3)
	github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go:68 +0x776
github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher.(*EventRouter).GetPartitionForRowChange(0xc005680b80?, 0xc005bb8090, 0x3)
	github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher/event_router.go:123 +0x48
github.com/pingcap/tiflow/cdc/sink/dmlsink/mq.(*dmlSink).WriteEvents(0xc00538da00, {0xc005a756b0, 0x6, 0x0?})
	github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/mq_dml_sink.go:188 +0x3ac
github.com/pingcap/tiflow/cdc/sink/tablesink.(*EventTableSink[...]).UpdateResolvedTs(0x60cbfa0, {0x12, 0x4706fd?, 0xc003a5f6d8?})
	github.com/pingcap/tiflow/cdc/sink/tablesink/table_sink_impl.go:175 +0x5e9
github.com/pingcap/tiflow/cdc/processor/sinkmanager.(*tableSinkWrapper).updateResolvedTs(0xc0051dc420, {0x6601cbd74c00015?, 0xffffffffffffffff?, 0xc0055f2120?})
	github.com/pingcap/tiflow/cdc/processor/sinkmanager/table_sink_wrapper.go:220 +0x14d
github.com/pingcap/tiflow/cdc/processor/sinkmanager.advanceTableSink(0xc0011d37a0, 0x6601cbd74c00015, 0xd30, 0xc0055f2120)
	github.com/pingcap/tiflow/cdc/processor/sinkmanager/table_sink_advancer.go:329 +0x567
github.com/pingcap/tiflow/cdc/processor/sinkmanager.(*tableSinkAdvancer).advance(0xc003a5fde8, 0x0)
	github.com/pingcap/tiflow/cdc/processor/sinkmanager/table_sink_advancer.go:100 +0x691
github.com/pingcap/tiflow/cdc/processor/sinkmanager.(*tableSinkAdvancer).tryAdvanceAndAcquireMem(0xc003a5fde8, 0x1, 0x1)
	github.com/pingcap/tiflow/cdc/processor/sinkmanager/table_sink_advancer.go:182 +0x3e5
github.com/pingcap/tiflow/cdc/processor/sinkmanager.(*tableSinkAdvancer).finish(...)
	github.com/pingcap/tiflow/cdc/processor/sinkmanager/table_sink_advancer.go:257
github.com/pingcap/tiflow/cdc/processor/sinkmanager.(*sinkWorker).handleTask(0xc0055d5860, {0x606fed0, 0xc0055d5810}, 0xc0011d37a0)
	github.com/pingcap/tiflow/cdc/processor/sinkmanager/table_sink_worker.go:190 +0x7b6
github.com/pingcap/tiflow/cdc/processor/sinkmanager.(*sinkWorker).handleTasks(0xc0055d5860, {0x606fed0, 0xc0055d5810}, 0xc0055cb8f0)
	github.com/pingcap/tiflow/cdc/processor/sinkmanager/table_sink_worker.go:95 +0xc7
github.com/pingcap/tiflow/cdc/processor/sinkmanager.(*SinkManager).startSinkWorkers.func1()
	github.com/pingcap/tiflow/cdc/processor/sinkmanager/manager.go:404 +0x2a
golang.org/x/sync/errgroup.(*Group).Go.func1()
	golang.org/x/sync@v0.13.0/errgroup/errgroup.go:79 +0x50
created by golang.org/x/sync/errgroup.(*Group).Go in goroutine 1508
	golang.org/x/sync@v0.13.0/errgroup/errgroup.go:76 +0x96

```

### Versions of the cluster


TiCDC version (execute `cdc version`):

```console
master
```

Pull request description:
sink(ticdc): check virtual columns in column dispatcher
<!--
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 #12241

### What is changed and how it works?
Since column values don't include virtual generated columns, we also filter the virtual generated columns in the column dispatcher.
The column dispatcher now explicitly checks for and prevents the use of virtual generated columns, returning an ErrDispatcherFailed error if such columns are specified. This prevents potential panics when configuring the dispatcher.

### 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 configuring the column dispatcher with virtual columns.
```


Patch:
diff --git a/cdc/model/schema_storage.go b/cdc/model/schema_storage.go
index 78ebd1820..b7b859a40 100644
--- a/cdc/model/schema_storage.go
+++ b/cdc/model/schema_storage.go
@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/types"
 	"github.com/pingcap/tidb/pkg/table/tables"
 	"github.com/pingcap/tidb/pkg/util/rowcodec"
+	"github.com/pingcap/tiflow/pkg/errors"
 	"go.uber.org/zap"
 )
 
@@ -469,25 +470,36 @@ func (ti *TableInfo) IndexByName(name string) ([]string, []int, bool) {
 // 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) {
+func (ti *TableInfo) OffsetsByNames(names []string) ([]int, error) {
 	// todo: optimize it
-	columnOffsets := make(map[string]int, len(ti.Columns))
+	columnOffsets := make(map[string]int)
+	virtualGeneratedColumn := make(map[string]struct{})
 	for _, col := range ti.Columns {
 		if col != nil {
-			columnOffsets[col.Name.L] = ti.columnsOffset[col.ID]
+			if IsColCDCVisible(col) {
+				columnOffsets[col.Name.L] = ti.columnsOffset[col.ID]
+			} else {
+				virtualGeneratedColumn[col.Name.L] = struct{}{}
+			}
 		}
 	}
 
 	result := make([]int, 0, len(names))
 	for _, col := range names {
-		offset, ok := columnOffsets[strings.ToLower(col)]
+		name := strings.ToLower(col)
+		if _, ok := virtualGeneratedColumn[name]; ok {
+			return nil, errors.ErrDispatcherFailed.GenWithStack(
+				"found virtual generated columns when dispatch event, table: %v, columns: %v column: %v", ti.GetTableName(), names, name)
+		}
+		offset, ok := columnOffsets[name]
 		if !ok {
-			return nil, false
+			return nil, errors.ErrDispatcherFailed.GenWithStack(
+				"columns not found when dispatch event, table: %v, columns: %v, column: %v", ti.GetTableName(), names, name)
 		}
 		result = append(result, offset)
 	}
 
-	return result, true
+	return result, nil
 }
 
 // GetPrimaryKeyColumnNames returns the primary key column names
diff --git a/cdc/sink/dmlsink/mq/dispatcher/event_router.go b/cdc/sink/dmlsink/mq/dispatcher/event_router.go
index 2d325b0c0..b15245a28 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/event_router.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/event_router.go
@@ -150,10 +150,9 @@ func (s *EventRouter) VerifyTables(infos []*model.TableInfo) error {
 				}
 			}
 		case *partition.ColumnsDispatcher:
-			_, ok := table.OffsetsByNames(v.Columns)
-			if !ok {
-				return cerror.ErrDispatcherFailed.GenWithStack(
-					"columns not found when verify the table, table: %v, columns: %v", table.TableName, v.Columns)
+			_, err := table.OffsetsByNames(v.Columns)
+			if err != nil {
+				return err
 			}
 		default:
 		}
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go b/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
index abaab5327..ab2f4016a 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
@@ -19,7 +19,6 @@ import (
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
-	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/hash"
 	"go.uber.org/zap"
 )
@@ -55,13 +54,10 @@ func (r *ColumnsDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent,
 		dispatchCols = row.PreColumns
 	}
 
-	offsets, ok := row.TableInfo.OffsetsByNames(r.Columns)
-	if !ok {
-		log.Error("columns not found when dispatch event",
-			zap.Any("tableName", row.TableInfo.GetTableName()),
-			zap.Strings("columns", r.Columns))
-		return 0, "", errors.ErrDispatcherFailed.GenWithStack(
-			"columns not found when dispatch event, table: %v, columns: %v", row.TableInfo.GetTableName(), r.Columns)
+	offsets, err := row.TableInfo.OffsetsByNames(r.Columns)
+	if err != nil {
+		log.Error("dispatch event failed", zap.Error(err))
+		return 0, "", err
 	}
 
 	for idx := 0; idx < len(r.Columns); idx++ {


Test patch:
diff --git a/cdc/model/schema_storage_test.go b/cdc/model/schema_storage_test.go
index be6fb5bfc..97adb6c0a 100644
--- a/cdc/model/schema_storage_test.go
+++ b/cdc/model/schema_storage_test.go
@@ -21,6 +21,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/charset"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	parser_types "github.com/pingcap/tidb/pkg/parser/types"
+	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/stretchr/testify/require"
 )
 
@@ -342,28 +343,40 @@ func TestColumnsByNames(t *testing.T) {
 				Name: pmodel.NewCIStr("col3"),
 				ID:   2,
 			},
+			{
+				Name:                pmodel.NewCIStr("col4"),
+				ID:                  3,
+				GeneratedExprString: "generated",
+			},
 		},
 	})
 
 	names := []string{"col1", "col2", "col3"}
-	offsets, ok := tableInfo.OffsetsByNames(names)
-	require.True(t, ok)
+	offsets, err := tableInfo.OffsetsByNames(names)
+	require.NoError(t, err)
 	require.Equal(t, []int{1, 0, 2}, offsets)
 
 	names = []string{"col2"}
-	offsets, ok = tableInfo.OffsetsByNames(names)
-	require.True(t, ok)
+	offsets, err = tableInfo.OffsetsByNames(names)
+	require.NoError(t, err)
 	require.Equal(t, []int{0}, offsets)
 
 	names = []string{"col1", "col-not-found"}
-	offsets, ok = tableInfo.OffsetsByNames(names)
-	require.False(t, ok)
+	offsets, err = tableInfo.OffsetsByNames(names)
+	require.ErrorIs(t, err, errors.ErrDispatcherFailed)
+	require.ErrorContains(t, err, "columns not found")
 	require.Nil(t, offsets)
 
 	names = []string{"Col1", "COL2", "CoL3"}
-	offsets, ok = tableInfo.OffsetsByNames(names)
-	require.True(t, ok)
+	offsets, err = tableInfo.OffsetsByNames(names)
+	require.NoError(t, err)
 	require.Equal(t, []int{1, 0, 2}, offsets)
+
+	names = []string{"Col4"}
+	offsets, err = tableInfo.OffsetsByNames(names)
+	require.ErrorIs(t, err, errors.ErrDispatcherFailed)
+	require.ErrorContains(t, err, "found virtual generated columns")
+	require.Nil(t, offsets)
 }
 
 func TestBuildTiDBTableInfoWithIntPrimaryKey(t *testing.T) {
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go b/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
index 52adf2adc..62556b7d7 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
@@ -35,6 +35,7 @@ func TestColumnsDispatcher(t *testing.T) {
 			{ID: 1, Name: pmodel.NewCIStr("col2"), Offset: 1, FieldType: *types.NewFieldType(mysql.TypeLong)},
 			{ID: 2, Name: pmodel.NewCIStr("col1"), Offset: 0, FieldType: *types.NewFieldType(mysql.TypeLong)},
 			{ID: 3, Name: pmodel.NewCIStr("col3"), Offset: 2, FieldType: *types.NewFieldType(mysql.TypeLong)},
+			{ID: 4, Name: pmodel.NewCIStr("col4"), Offset: 3, FieldType: *types.NewFieldType(mysql.TypeLong), GeneratedExprString: "generated"},
 		},
 	}
 	tableInfo := model.WrapTableInfo(100, "test", 33, tidbTableInfo)
@@ -47,9 +48,15 @@ func TestColumnsDispatcher(t *testing.T) {
 		},
 	}
 
-	p := NewColumnsDispatcher([]string{"col-2", "col-not-found"})
+	p := NewColumnsDispatcher([]string{"col2", "col-not-found"})
 	_, _, err := p.DispatchRowChangedEvent(event, 16)
 	require.ErrorIs(t, err, errors.ErrDispatcherFailed)
+	require.ErrorContains(t, err, "columns not found")
+
+	p = NewColumnsDispatcher([]string{"col4"})
+	_, _, err = p.DispatchRowChangedEvent(event, 16)
+	require.ErrorIs(t, err, errors.ErrDispatcherFailed)
+	require.ErrorContains(t, err, "found virtual generated columns")
 
 	p = NewColumnsDispatcher([]string{"col2", "col1"})
 	index, _, err := p.DispatchRowChangedEvent(event, 16)