kafka(ticdc): index-value dispatcher support set index names.

cdc/api/v2/api_helpers.go

@@ -29,11 +29,13 @@ import (
 	"github.com/pingcap/tiflow/cdc/kv"
 	"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/validator"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/security"
+	"github.com/pingcap/tiflow/pkg/sink"
 	"github.com/pingcap/tiflow/pkg/txnutil/gc"
 	"github.com/pingcap/tiflow/pkg/version"
 	"github.com/r3labs/diff"
@@ -112,9 +114,11 @@ type APIV2Helpers interface {
 		credential *security.Credential,
 	) (tidbkv.Storage, error)
 
-	// getVerfiedTables wraps entry.VerifyTables to increase testability
-	getVerfiedTables(replicaConfig *config.ReplicaConfig,
-		storage tidbkv.Storage, startTs uint64) (ineligibleTables,
+	// getVerifiedTables wraps entry.VerifyTables to increase testability
+	getVerifiedTables(replicaConfig *config.ReplicaConfig,
+		storage tidbkv.Storage, startTs uint64,
+		scheme string, topic string, protocol config.Protocol,
+	) (ineligibleTables,
 		eligibleTables []model.TableName, err error,
 	)
 }
@@ -491,15 +495,42 @@ func (h APIV2HelpersImpl) createTiStore(pdAddrs []string,
 	return kv.CreateTiStore(strings.Join(pdAddrs, ","), credential)
 }
 
-func (h APIV2HelpersImpl) getVerfiedTables(replicaConfig *config.ReplicaConfig,
-	storage tidbkv.Storage, startTs uint64) (ineligibleTables,
-	eligibleTables []model.TableName, err error,
-) {
+func (h APIV2HelpersImpl) getVerifiedTables(
+	replicaConfig *config.ReplicaConfig,
+	storage tidbkv.Storage, startTs uint64,
+	scheme string, topic string, protocol config.Protocol,
+) ([]model.TableName, []model.TableName, error) {
 	f, err := filter.NewFilter(replicaConfig, "")
 	if err != nil {
-		return
+		return nil, nil, err
 	}
-	_, ineligibleTables, eligibleTables, err = entry.
+	tableInfos, ineligibleTables, eligibleTables, err := entry.
 		VerifyTables(f, storage, startTs)
-	return
+	if err != nil {
+		return nil, nil, err
+	}
+
+	if !sink.IsMQScheme(scheme) {
+		return ineligibleTables, eligibleTables, nil
+	}
+
+	eventRouter, err := dispatcher.NewEventRouter(replicaConfig, protocol, topic, scheme)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	for _, table := range tableInfos {
+		dummyEvent := &model.RowChangedEvent{
+			Table: &model.TableName{
+				Schema: table.TableName.Schema,
+				Table:  table.TableName.Table,
+			},
+			TableInfo: table,
+		}
+		_, _, err := eventRouter.GetPartitionForRowChange(dummyEvent, 1)
+		if err != nil {
+			return nil, nil, err
+		}
+	}
+	return ineligibleTables, eligibleTables, nil
 }

cdc/api/v2/changefeed.go

@@ -17,6 +17,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"net/url"
 	"sort"
 	"strings"
 	"time"
@@ -28,6 +29,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/api"
 	"github.com/pingcap/tiflow/cdc/capture"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/retry"
 	"github.com/pingcap/tiflow/pkg/txnutil/gc"
@@ -341,9 +343,20 @@ func (h *OpenAPIV2) verifyTable(c *gin.Context) {
 		_ = c.Error(err)
 		return
 	}
+	uri, err := url.Parse(cfg.SinkURI)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+	scheme := uri.Scheme
+	topic := strings.TrimFunc(uri.Path, func(r rune) bool {
+		return r == '/'
+	})
 	replicaCfg := cfg.ReplicaConfig.ToInternalReplicaConfig()
+	protocol, _ := config.ParseSinkProtocolFromString(util.GetOrZero(replicaCfg.Sink.Protocol))
+
 	ineligibleTables, eligibleTables, err := h.helpers.
-		getVerfiedTables(replicaCfg, kvStore, cfg.StartTs)
+		getVerifiedTables(replicaCfg, kvStore, cfg.StartTs, scheme, topic, protocol)
 	if err != nil {
 		_ = c.Error(err)
 		return

cdc/api/v2/changefeed_test.go

@@ -178,7 +178,8 @@ func TestCreateChangefeed(t *testing.T) {
 	helpers.EXPECT().
 		getEtcdClient(gomock.Any(), gomock.Any()).
 		Return(testEtcdCluster.RandClient(), nil)
-	helpers.EXPECT().getVerfiedTables(gomock.Any(), gomock.Any(), gomock.Any()).
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil, nil).
 		AnyTimes()
 	helpers.EXPECT().
@@ -662,12 +663,13 @@ func TestVerifyTable(t *testing.T) {
 	require.Nil(t, err)
 	require.Contains(t, respErr.Code, "ErrNewStore")
 
-	// case 3: getVerfiedTables failed
+	// case 3: getVerifiedTables failed
 	helpers.EXPECT().
 		createTiStore(gomock.Any(), gomock.Any()).
 		Return(nil, nil).
 		AnyTimes()
-	helpers.EXPECT().getVerfiedTables(gomock.Any(), gomock.Any(), gomock.Any()).
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(nil, nil, cerrors.ErrFilterRuleInvalid).
 		Times(1)
 
@@ -688,7 +690,8 @@ func TestVerifyTable(t *testing.T) {
 	ineligible := []model.TableName{
 		{Schema: "test", Table: "invalidTable"},
 	}
-	helpers.EXPECT().getVerfiedTables(gomock.Any(), gomock.Any(), gomock.Any()).
+	helpers.EXPECT().getVerifiedTables(gomock.Any(), gomock.Any(), gomock.Any(),
+		gomock.Any(), gomock.Any(), gomock.Any()).
 		Return(eligible, ineligible, nil)
 
 	w = httptest.NewRecorder()

cdc/api/v2/model.go

@@ -68,6 +68,7 @@ type VerifyTableConfig struct {
 	PDConfig
 	ReplicaConfig *ReplicaConfig `json:"replica_config"`
 	StartTs       uint64         `json:"start_ts"`
+	SinkURI       string         `json:"sink_uri"`
 }
 
 func getDefaultVerifyTableConfig() *VerifyTableConfig {

cdc/model/sink.go

@@ -507,6 +507,22 @@ func (r *RowChangedEvent) ApproximateBytes() int {
 	return size
 }
 
+// IndexByName returns the index columns and offsets of the corresponding index by name
+func (r *RowChangedEvent) IndexByName(name string) ([]string, []int, bool) {
+	for _, index := range r.TableInfo.Indices {
+		if index.Name.O == name {
+			names := make([]string, 0, len(index.Columns))
+			offset := make([]int, 0, len(index.Columns))
+			for _, col := range index.Columns {
+				names = append(names, col.Name.O)
+				offset = append(offset, col.Offset)
+			}
+			return names, offset, true
+		}
+	}
+	return nil, nil, false
+}
+
 // Column represents a column value in row changed event
 type Column struct {
 	Name      string         `json:"name" msg:"name"`

cdc/model/sink_test.go

@@ -609,3 +609,36 @@ func TestTrySplitAndSortUpdateEventOne(t *testing.T) {
 	require.NoError(t, err)
 	require.Len(t, txn.Rows, 1)
 }
+
+func TestIndexByName(t *testing.T) {
+	event := &RowChangedEvent{
+		TableInfo: &TableInfo{
+			TableInfo: &timodel.TableInfo{
+				Indices: []*timodel.IndexInfo{
+					{
+						Name: timodel.CIStr{
+							O: "idx1",
+						},
+						Columns: []*timodel.IndexColumn{
+							{
+								Name: timodel.CIStr{
+									O: "col1",
+								},
+							},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	names, offsets, ok := event.IndexByName("idx2")
+	require.False(t, ok)
+	require.Nil(t, names)
+	require.Nil(t, offsets)
+
+	names, offsets, ok = event.IndexByName("idx1")
+	require.True(t, ok)
+	require.Equal(t, []string{"col1"}, names)
+	require.Equal(t, []int{0}, offsets)
+}

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

@@ -67,7 +67,7 @@ func NewEventRouter(
 			f = filter.CaseInsensitive(f)
 		}
 
-		d := getPartitionDispatcher(ruleConfig.PartitionRule, scheme)
+		d := getPartitionDispatcher(ruleConfig.PartitionRule, scheme, ruleConfig.IndexName)
 		t, err := getTopicDispatcher(ruleConfig.TopicRule, defaultTopic, protocol, scheme)
 		if err != nil {
 			return nil, err
@@ -116,7 +116,7 @@ func (s *EventRouter) GetTopicForDDL(ddl *model.DDLEvent) string {
 func (s *EventRouter) GetPartitionForRowChange(
 	row *model.RowChangedEvent,
 	partitionNum int32,
-) (int32, string) {
+) (int32, string, error) {
 	_, partitionDispatcher := s.matchDispatcher(
 		row.Table.Schema, row.Table.Table,
 	)
@@ -176,7 +176,7 @@ func (s *EventRouter) matchDispatcher(
 }
 
 // getPartitionDispatcher returns the partition dispatcher for a specific partition rule.
-func getPartitionDispatcher(rule string, scheme string) partition.Dispatcher {
+func getPartitionDispatcher(rule string, scheme string, indexName string) partition.Dispatcher {
 	switch strings.ToLower(rule) {
 	case "default":
 		return partition.NewDefaultDispatcher()
@@ -186,7 +186,7 @@ func getPartitionDispatcher(rule string, scheme string) partition.Dispatcher {
 		return partition.NewTableDispatcher()
 	case "index-value", "rowid":
 		log.Warn("rowid is deprecated, please use index-value instead.")
-		return partition.NewIndexValueDispatcher()
+		return partition.NewIndexValueDispatcher(indexName)
 	default:
 	}
 

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

@@ -192,7 +192,7 @@ func TestGetPartitionForRowChange(t *testing.T) {
 	d, err := NewEventRouter(replicaConfig, config.ProtocolCanalJSON, "test", sink.KafkaScheme)
 	require.NoError(t, err)
 
-	p, _ := d.GetPartitionForRowChange(&model.RowChangedEvent{
+	p, _, err := d.GetPartitionForRowChange(&model.RowChangedEvent{
 		Table: &model.TableName{Schema: "test_default1", Table: "table"},
 		Columns: []*model.Column{
 			{
@@ -203,9 +203,10 @@ func TestGetPartitionForRowChange(t *testing.T) {
 		},
 		IndexColumns: [][]int{{0}},
 	}, 16)
+	require.NoError(t, err)
 	require.Equal(t, int32(14), p)
 
-	p, _ = d.GetPartitionForRowChange(&model.RowChangedEvent{
+	p, _, err = d.GetPartitionForRowChange(&model.RowChangedEvent{
 		Table: &model.TableName{Schema: "test_default2", Table: "table"},
 		Columns: []*model.Column{
 			{
@@ -216,15 +217,17 @@ func TestGetPartitionForRowChange(t *testing.T) {
 		},
 		IndexColumns: [][]int{{0}},
 	}, 16)
+	require.NoError(t, err)
 	require.Equal(t, int32(0), p)
 
-	p, _ = d.GetPartitionForRowChange(&model.RowChangedEvent{
+	p, _, err = d.GetPartitionForRowChange(&model.RowChangedEvent{
 		Table:    &model.TableName{Schema: "test_table", Table: "table"},
 		CommitTs: 1,
 	}, 16)
+	require.NoError(t, err)
 	require.Equal(t, int32(15), p)
 
-	p, _ = d.GetPartitionForRowChange(&model.RowChangedEvent{
+	p, _, err = d.GetPartitionForRowChange(&model.RowChangedEvent{
 		Table: &model.TableName{Schema: "test_index_value", Table: "table"},
 		Columns: []*model.Column{
 			{
@@ -238,12 +241,14 @@ func TestGetPartitionForRowChange(t *testing.T) {
 			},
 		},
 	}, 10)
+	require.NoError(t, err)
 	require.Equal(t, int32(1), p)
 
-	p, _ = d.GetPartitionForRowChange(&model.RowChangedEvent{
+	p, _, err = d.GetPartitionForRowChange(&model.RowChangedEvent{
 		Table:    &model.TableName{Schema: "a", Table: "table"},
 		CommitTs: 1,
 	}, 2)
+	require.NoError(t, err)
 	require.Equal(t, int32(1), p)
 }
 

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

@@ -31,6 +31,6 @@ func NewDefaultDispatcher() *DefaultDispatcher {
 
 // DispatchRowChangedEvent returns the target partition to which
 // a row changed event should be dispatched.
-func (d *DefaultDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent, partitionNum int32) (int32, string) {
+func (d *DefaultDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent, partitionNum int32) (int32, string, error) {
 	return d.tbd.DispatchRowChangedEvent(row, partitionNum)
 }

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

@@ -38,6 +38,7 @@ func TestDefaultDispatcher(t *testing.T) {
 		IndexColumns: [][]int{{0}},
 	}
 
-	targetPartition, _ := NewDefaultDispatcher().DispatchRowChangedEvent(row, 3)
+	targetPartition, _, err := NewDefaultDispatcher().DispatchRowChangedEvent(row, 3)
+	require.NoError(t, err)
 	require.Equal(t, int32(0), targetPartition)
 }

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

@@ -22,5 +22,5 @@ type Dispatcher interface {
 	// DispatchRowChangedEvent returns an index of partitions or a partition key
 	//  according to RowChangedEvent.
 	// Concurrency Note: This method is thread-safe.
-	DispatchRowChangedEvent(row *model.RowChangedEvent, partitionNum int32) (int32, string)
+	DispatchRowChangedEvent(row *model.RowChangedEvent, partitionNum int32) (int32, string, error)
 }