[to #48] modify TiDB CDC to TiKV CDC by zeminzhou · Pull Request #74 · tikv/migration

zeminzhou · 2022-03-23T03:02:51Z

What problem does this PR solve?

Issue Number: #48

Problem Description:
The current TiDB CDC should not be used for RawKV CDC.

What is changed and how does it work?

Modify TiDB CDC to synchronize RawKV data

Code changes

Has exported function/method change
Has exported variable/fields change
Has interface methods change
Has persistent data change

Check List for Tests

This PR has been tested by at least one of the following methods:

Unit test

zz-jason · 2022-03-29T05:08:41Z

-	"github.com/tikv/migration/cdc/cdc/processor/pipeline/system"
-	ssystem "github.com/tikv/migration/cdc/cdc/sorter/leveldb/system"
+
+	// ssystem "github.com/tikv/migration/cdc/cdc/sorter/leveldb/system"


why not just remove it?

The sorter module is used to sort transactions by commit ts to ensure that the transactions entered by cdc are in the same order as commit ts. Now rawkv don't it (We also need to add sorter for txnkv later).

zz-jason · 2022-03-29T05:09:20Z

-	if len(changefeedConfig.FilterRules) != 0 {
-		replicaConfig.Filter.Rules = changefeedConfig.FilterRules
-	}
+	/*


FilterRules are used to filter table.

We can modify the filter table to filter keys (or keyspan) to filter keys that users do not need to synchronize.

zz-jason · 2022-03-29T05:10:08Z

-		if err != nil {
-			return nil, cerror.ErrChangefeedUpdateRefused.GenWithStackByArgs(err.Error())
+	/*
+		if len(changefeedConfig.FilterRules) != 0 {


zz-jason · 2022-03-29T05:10:57Z

-	// APIOpVarTableID is the key of table ID in HTTP API
-	APIOpVarTableID = "table-id"
+	// APIOpVarKeySpanID is the key of keyspan ID in HTTP API
+	APIOpVarKeySpanID = "keyspan-id"


what's the relationship between keyspan id and region id?

In the current implementation, keyspan and region have a one-to-one relationship. But keyspan id is not equal to region id, keyspan id only represents an interval [start, end). keyspan-id = hash(start+end).

zz-jason · 2022-03-29T05:11:31Z

 	})

-	tableID, tableName := util.TableIDFromCtx(ctx)
+	// tableID, tableName := util.KeySpanIDFromCtx(ctx)


There might be something wrong here, I'll check it later.

zz-jason · 2022-03-29T05:11:37Z

 							zap.Uint64("ts", errInfo.singleRegionInfo.ts),
 							zap.String("changefeed", cfID), zap.Stringer("span", errInfo.span),
-							zap.Int64("tableID", tableID), zap.String("tableName", tableName),
+							// 	zap.Int64("tableID", tableID), zap.String("tableName", tableName),


zz-jason · 2022-03-29T05:12:17Z

 	"github.com/tikv/migration/cdc/cdc/owner"
 	"github.com/tikv/migration/cdc/cdc/processor"
-	tablepipeline "github.com/tikv/migration/cdc/cdc/processor/pipeline"
+	keyspanpipeline "github.com/tikv/migration/cdc/cdc/processor/pipeline"


why not just use the original name pipeline?

I don't know why tablepipeline was used instead of pipeline. 🤣

zz-jason · 2022-03-29T05:13:08Z

-	unified.InitMetrics(registry)
-	leveldb.InitMetrics(registry)
-	redowriter.InitMetrics(registry)
+	// sorter.InitMetrics(registry)


zz-jason · 2022-03-29T05:16:03Z


-		newDDLPuller: newDDLPuller,
-		newSink:      newDDLSink,
+		// newDDLPuller: newDDLPuller,


In TiCDC, owner is responsible for two things:

Divide changfeed into tasks and assign them to other captures.

Synchronize DML operations of upstream TiDB.

Now synchronizing DML operations don't need.

pingyu · 2022-03-29T09:37:28Z

@@ -37,6 +38,15 @@ func (s Span) String() string {
 	return fmt.Sprintf("[%s, %s)", hex.EncodeToString(s.Start), hex.EncodeToString(s.End))
 }

+func (s Span) ID() uint64 {


There is a very small possibility that two different spans generating a same hash. Maybe we should add a seed in Span to avoid it.

Yes, it is indeed possible. But I don't think adding a seed is effective, because this is a probabilistic event.

a simple solution, https://github.com/zeminzhou/migration/blob/tikv-cdc/cdc/cdc/owner/scheduler_v1.go#L446-L453.

There is still conflict between "currentKeySpans" and the keyspans acquired before.

How about using region id as keyspan id?https://github.com/zeminzhou/migration/blob/tikv-cdc/cdc/cdc/owner/scheduler_v1.go#L444.

If we do it this way, keyspan will lose its meaning.

No. The range of a region will change after split or merge.

How about add a seed to KeySpan to avoid conflict ?

We maintain all key spans in owner. Every time we get a new key span, calculate the hash with seed starting from 0. If the hash conflicts with an existing key span, seed += 1, and calculate the hash again. Until we find a hash without conflict.
Note that the seed should be transfer to Capture & Processor, along with key span's start & end.

pingyu · 2022-03-29T10:14:24Z

 	// for ManualSchedule only
 	targetCaptureID model.CaptureID
 	// for ManualSchedule only
-	tableID model.TableID


(Not related to this line)
GC related procedure (Owner.updateGCSafepoint) need to be reviewed. At least the CDCServiceSaftPointID must be changed. Refer to rfc.

pingyu · 2022-03-29T11:58:55Z


 	s.cleanUpFinishedOperations()
-	pendingJob, err := s.syncTablesWithCurrentTables()
+	pendingJob, err := s.syncKeySpansWithCurrentKeySpans()


Keyspans to add & keyspans to remove, must be overlapped. So to be correct, I think we must finish remove jobs first, and then dispatch the add jobs.

The "remove then add" actions would be expensive, and make the replication be jitter. So I think it would be better to limit the number of keyspans to be balanced in every period.

If we limit the number of keyspans to be balanced, the keyspans should be "related", e.g, the splitted regions and the original region, the merged region and the orginal two regions.

pingyu · 2022-04-14T16:28:52Z

@@ -20,22 +20,22 @@ import (
 var (


These metrics is for mounter. Can be removed.