kafka(ticdc): make column selector can work normally to filter out co…

…lumns (#9920) (#10060)

close #9967

cdc/api/v2/api_helpers.go

@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/owner"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer/columnselector"
 	"github.com/pingcap/tiflow/cdc/sink/validator"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -518,11 +519,19 @@ func (h APIV2HelpersImpl) getVerifiedTables(
 	if err != nil {
 		return nil, nil, err
 	}
-
 	err = eventRouter.VerifyTables(tableInfos)
 	if err != nil {
 		return nil, nil, err
 	}
 
+	selectors, err := columnselector.New(replicaConfig)
+	if err != nil {
+		return nil, nil, err
+	}
+	err = selectors.VerifyTables(tableInfos, eventRouter)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	return ineligibleTables, eligibleTables, nil
 }

cdc/model/schema_storage.go

@@ -341,9 +341,9 @@ func (ti *TableInfo) IndexByName(name string) ([]string, []int, bool) {
 	return names, offset, true
 }
 
-// ColumnsByNames returns the column offsets of the corresponding columns by names
+// OffsetsByNames returns the column offsets of the corresponding columns by names
 // If any column does not exist, return false
-func (ti *TableInfo) ColumnsByNames(names []string) ([]int, bool) {
+func (ti *TableInfo) OffsetsByNames(names []string) ([]int, bool) {
 	// todo: optimize it
 	columnOffsets := make(map[string]int, len(ti.Columns))
 	for _, col := range ti.Columns {

cdc/model/schema_storage_test.go

@@ -339,17 +339,17 @@ func TestColumnsByNames(t *testing.T) {
 	}
 
 	names := []string{"col1", "col2", "col3"}
-	offsets, ok := tableInfo.ColumnsByNames(names)
+	offsets, ok := tableInfo.OffsetsByNames(names)
 	require.True(t, ok)
 	require.Equal(t, []int{0, 1, 2}, offsets)
 
 	names = []string{"col2"}
-	offsets, ok = tableInfo.ColumnsByNames(names)
+	offsets, ok = tableInfo.OffsetsByNames(names)
 	require.True(t, ok)
 	require.Equal(t, []int{1}, offsets)
 
 	names = []string{"col1", "col-not-found"}
-	offsets, ok = tableInfo.ColumnsByNames(names)
+	offsets, ok = tableInfo.OffsetsByNames(names)
 	require.False(t, ok)
 	require.Nil(t, offsets)
 }

cdc/sink/dmlsink/mq/dispatcher/event_router.go

@@ -119,13 +119,14 @@ func (s *EventRouter) GetPartitionForRowChange(
 	row *model.RowChangedEvent,
 	partitionNum int32,
 ) (int32, string, error) {
-	_, partitionDispatcher := s.matchDispatcher(
-		row.Table.Schema, row.Table.Table,
-	)
+	return s.GetPartitionDispatcher(row.Table.Schema, row.Table.Table).
+		DispatchRowChangedEvent(row, partitionNum)
+}
 
-	return partitionDispatcher.DispatchRowChangedEvent(
-		row, partitionNum,
-	)
+// GetPartitionDispatcher returns the partition dispatcher for a specific table.
+func (s *EventRouter) GetPartitionDispatcher(schema, table string) partition.Dispatcher {
+	_, partitionDispatcher := s.matchDispatcher(schema, table)
+	return partitionDispatcher
 }
 
 // VerifyTables return error if any one table route rule is invalid.
@@ -147,7 +148,7 @@ func (s *EventRouter) VerifyTables(infos []*model.TableInfo) error {
 					"index is not unique when verify the table, table: %v, index: %s", table.TableName, v.IndexName)
 			}
 		case *partition.ColumnsDispatcher:
-			_, ok := table.ColumnsByNames(v.Columns)
+			_, 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)

cdc/sink/dmlsink/mq/dispatcher/partition/columns.go

@@ -55,7 +55,7 @@ func (r *ColumnsDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent,
 		dispatchCols = row.PreColumns
 	}
 
-	offsets, ok := row.TableInfo.ColumnsByNames(r.Columns)
+	offsets, ok := row.TableInfo.OffsetsByNames(r.Columns)
 	if !ok {
 		log.Error("columns not found when dispatch event",
 			zap.Any("tableName", row.Table),
@@ -65,7 +65,11 @@ func (r *ColumnsDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent,
 	}
 
 	for idx := 0; idx < len(r.Columns); idx++ {
-		r.hasher.Write([]byte(r.Columns[idx]), []byte(model.ColumnValueString(dispatchCols[offsets[idx]].Value)))
+		col := dispatchCols[offsets[idx]]
+		if col == nil {
+			continue
+		}
+		r.hasher.Write([]byte(r.Columns[idx]), []byte(model.ColumnValueString(col.Value)))
 	}
 
 	sum32 := r.hasher.Sum32()

cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go

@@ -73,7 +73,11 @@ func (r *IndexValueDispatcher) DispatchRowChangedEvent(row *model.RowChangedEven
 				"index not found when dispatch event, table: %v, index: %s", row.Table, r.IndexName)
 		}
 		for idx := 0; idx < len(names); idx++ {
-			r.hasher.Write([]byte(names[idx]), []byte(model.ColumnValueString(dispatchCols[offsets[idx]].Value)))
+			col := dispatchCols[offsets[idx]]
+			if col == nil {
+				continue
+			}
+			r.hasher.Write([]byte(names[idx]), []byte(model.ColumnValueString(col.Value)))
 		}
 	}
 

cdc/sink/dmlsink/mq/kafka_dml_sink.go

@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dmlproducer"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer/columnselector"
 	"github.com/pingcap/tiflow/cdc/sink/util"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -97,6 +98,11 @@ func NewKafkaDMLSink(
 		return nil, errors.Trace(err)
 	}
 
+	trans, err := columnselector.New(replicaConfig)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	encoderConfig, err := util.GetEncoderConfig(changefeedID, sinkURI, protocol, replicaConfig, options.MaxMessageBytes)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -118,7 +124,7 @@ func NewKafkaDMLSink(
 	concurrency := tiflowutil.GetOrZero(replicaConfig.Sink.EncoderConcurrency)
 	encoderGroup := codec.NewEncoderGroup(encoderBuilder, concurrency, changefeedID)
 	s := newDMLSink(ctx, changefeedID, dmlProducer, adminClient, topicManager,
-		eventRouter, encoderGroup, protocol, scheme, errCh)
+		eventRouter, trans, encoderGroup, protocol, scheme, errCh)
 	log.Info("DML sink producer created",
 		zap.String("namespace", changefeedID.Namespace),
 		zap.String("changefeedID", changefeedID.ID))

cdc/sink/dmlsink/mq/mq_dml_sink.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dmlproducer"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/manager"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer"
 	"github.com/pingcap/tiflow/cdc/sink/metrics"
 	"github.com/pingcap/tiflow/cdc/sink/tablesink/state"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -48,6 +49,8 @@ type dmlSink struct {
 
 	alive struct {
 		sync.RWMutex
+
+		transformer transformer.Transformer
 		// eventRouter used to route events to the right topic and partition.
 		eventRouter *dispatcher.EventRouter
 		// topicManager used to manage topics.
@@ -77,6 +80,7 @@ func newDMLSink(
 	adminClient kafka.ClusterAdminClient,
 	topicManager manager.TopicManager,
 	eventRouter *dispatcher.EventRouter,
+	transformer transformer.Transformer,
 	encoderGroup codec.EncoderGroup,
 	protocol config.Protocol,
 	scheme string,
@@ -95,6 +99,7 @@ func newDMLSink(
 		dead:        make(chan struct{}),
 		scheme:      scheme,
 	}
+	s.alive.transformer = transformer
 	s.alive.eventRouter = eventRouter
 	s.alive.topicManager = topicManager
 	s.alive.worker = worker
@@ -170,6 +175,13 @@ func (s *dmlSink) WriteEvents(txns ...*dmlsink.CallbackableEvent[*model.SingleTa
 				s.cancel(err)
 				return errors.Trace(err)
 			}
+
+			err = s.alive.transformer.Apply(row)
+			if err != nil {
+				s.cancel(err)
+				return errors.Trace(err)
+			}
+
 			index, key, err := s.alive.eventRouter.GetPartitionForRowChange(row, partitionNum)
 			if err != nil {
 				s.cancel(err)

cdc/sink/dmlsink/mq/pulsar_dml_sink.go

@@ -24,6 +24,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dmlproducer"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/manager"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer/columnselector"
 	"github.com/pingcap/tiflow/cdc/sink/util"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -104,6 +105,11 @@ func NewPulsarDMLSink(
 		return nil, errors.Trace(err)
 	}
 
+	trans, err := columnselector.New(replicaConfig)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	encoderConfig, err := util.GetEncoderConfig(changefeedID, sinkURI, protocol, replicaConfig,
 		config.DefaultMaxMessageBytes)
 	if err != nil {
@@ -119,7 +125,7 @@ func NewPulsarDMLSink(
 	encoderGroup := codec.NewEncoderGroup(encoderBuilder, concurrency, changefeedID)
 
 	s := newDMLSink(ctx, changefeedID, p, nil, topicManager,
-		eventRouter, encoderGroup, protocol, scheme, errCh)
+		eventRouter, trans, encoderGroup, protocol, scheme, errCh)
 
 	return s, nil
 }