sink: check virtual columns in column dispatcher (#1867)

ref pingcap/tiflow#12241

Author: wk989898

downstreamadapter/sink/eventrouter/partition/columns.go

@@ -20,7 +20,6 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
-	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/ticdc/pkg/hash"
 	"go.uber.org/zap"
 )
@@ -54,13 +53,10 @@ func (r *ColumnsPartitionGenerator) GeneratePartitionIndexAndKey(row *commonEven
 		rowData = row.PreRow
 	}
 
-	offsets, ok := tableInfo.OffsetsByNames(r.Columns)
-	if !ok {
-		log.Error("columns not found when dispatch event",
-			zap.Any("tableName", tableInfo.GetTableName()),
-			zap.Strings("columns", r.Columns))
-		return 0, "", errors.ErrDispatcherFailed.GenWithStack(
-			"columns not found when dispatch event, table: %v, columns: %v", tableInfo.GetTableName(), r.Columns)
+	offsets, err := tableInfo.OffsetsByNames(r.Columns)
+	if err != nil {
+		log.Error("dispatch event failed", zap.Error(err))
+		return 0, "", err
 	}
 
 	for _, idx := range offsets {

pkg/common/table_info.go

@@ -23,6 +23,7 @@ import (
 	"sync/atomic"
 
 	"github.com/pingcap/log"
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/types"
@@ -375,25 +376,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.columnSchema.Columns))
+	columnOffsets := make(map[string]int)
+	virtualGeneratedColumn := make(map[string]struct{})
 	for idx, col := range ti.columnSchema.Columns {
 		if col != nil {
-			columnOffsets[col.Name.L] = idx
+			if IsColCDCVisible(col) {
+				columnOffsets[col.Name.L] = idx
+			} 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
 }
 
 func (ti *TableInfo) HasPrimaryKey() bool {

pkg/common/table_info_helper_test.go

@@ -16,6 +16,7 @@ package common
 import (
 	"testing"
 
+	"github.com/pingcap/ticdc/pkg/errors"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
@@ -207,28 +208,40 @@ func TestColumnsByNames(t *testing.T) {
 				Name: ast.NewCIStr("col3"),
 				ID:   2,
 			},
+			{
+				Name:                ast.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 TestHandleKey(t *testing.T) {