ddl: speed up add index by split handle range from pd region. (#5964)

Author: winkyao

ddl/ddl.go

@@ -87,16 +87,18 @@ var (
 	errIncorrectPrefixKey   = terror.ClassDDL.New(codeIncorrectPrefixKey, "Incorrect prefix key; the used key part isn't a string, the used length is longer than the key part, or the storage engine doesn't support unique prefix keys")
 	errTooLongKey           = terror.ClassDDL.New(codeTooLongKey,
 		fmt.Sprintf("Specified key was too long; max key length is %d bytes", maxPrefixLength))
-	errKeyColumnDoesNotExits = terror.ClassDDL.New(codeKeyColumnDoesNotExits, "this key column doesn't exist in table")
-	errDupKeyName            = terror.ClassDDL.New(codeDupKeyName, "duplicate key name")
-	errUnknownTypeLength     = terror.ClassDDL.New(codeUnknownTypeLength, "Unknown length for type tp %d")
-	errUnknownFractionLength = terror.ClassDDL.New(codeUnknownFractionLength, "Unknown Length for type tp %d and fraction %d")
-	errInvalidJobVersion     = terror.ClassDDL.New(codeInvalidJobVersion, "DDL job with version %d greater than current %d")
-	errFileNotFound          = terror.ClassDDL.New(codeFileNotFound, "Can't find file: './%s/%s.frm'")
-	errErrorOnRename         = terror.ClassDDL.New(codeErrorOnRename, "Error on rename of './%s/%s' to './%s/%s'")
-	errBadField              = terror.ClassDDL.New(codeBadField, "Unknown column '%s' in '%s'")
-	errInvalidUseOfNull      = terror.ClassDDL.New(codeInvalidUseOfNull, "Invalid use of NULL value")
-	errTooManyFields         = terror.ClassDDL.New(codeTooManyFields, "Too many columns")
+	errKeyColumnDoesNotExits    = terror.ClassDDL.New(codeKeyColumnDoesNotExits, "this key column doesn't exist in table")
+	errDupKeyName               = terror.ClassDDL.New(codeDupKeyName, "duplicate key name")
+	errUnknownTypeLength        = terror.ClassDDL.New(codeUnknownTypeLength, "Unknown length for type tp %d")
+	errUnknownFractionLength    = terror.ClassDDL.New(codeUnknownFractionLength, "Unknown Length for type tp %d and fraction %d")
+	errInvalidJobVersion        = terror.ClassDDL.New(codeInvalidJobVersion, "DDL job with version %d greater than current %d")
+	errFileNotFound             = terror.ClassDDL.New(codeFileNotFound, "Can't find file: './%s/%s.frm'")
+	errErrorOnRename            = terror.ClassDDL.New(codeErrorOnRename, "Error on rename of './%s/%s' to './%s/%s'")
+	errBadField                 = terror.ClassDDL.New(codeBadField, "Unknown column '%s' in '%s'")
+	errInvalidUseOfNull         = terror.ClassDDL.New(codeInvalidUseOfNull, "Invalid use of NULL value")
+	errTooManyFields            = terror.ClassDDL.New(codeTooManyFields, "Too many columns")
+	errInvalidSplitRegionRanges = terror.ClassDDL.New(codeInvalidRanges, "Failed to split region ranges")
+	errReorgWorkerNotRunnable   = terror.ClassDDL.New(codeReorgWorkerNotRunnable, "reorg worker is not runnable.")
 
 	// errWrongKeyColumn is for table column cannot be indexed.
 	errWrongKeyColumn = terror.ClassDDL.New(codeWrongKeyColumn, mysql.MySQLErrName[mysql.ErrWrongKeyColumn])
@@ -273,7 +275,7 @@ func newDDL(ctx context.Context, etcdCli *clientv3.Client, store kv.Storage,
 		lease:        lease,
 		ddlJobCh:     make(chan struct{}, 1),
 		ddlJobDoneCh: make(chan struct{}, 1),
-		reorgCtx:     &reorgCtx{notifyCancelReorgJob: make(chan struct{}, 1)},
+		reorgCtx:     &reorgCtx{notifyCancelReorgJob: 0},
 		ownerManager: manager,
 		schemaSyncer: syncer,
 		workerVars:   variable.NewSessionVars(),
@@ -500,17 +502,19 @@ func (d *ddl) WorkerVars() *variable.SessionVars {
 
 // DDL error codes.
 const (
-	codeInvalidWorker         terror.ErrCode = 1
-	codeNotOwner                             = 2
-	codeInvalidDDLJob                        = 3
-	codeInvalidJobFlag                       = 5
-	codeRunMultiSchemaChanges                = 6
-	codeWaitReorgTimeout                     = 7
-	codeInvalidStoreVer                      = 8
-	codeUnknownTypeLength                    = 9
-	codeUnknownFractionLength                = 10
-	codeInvalidJobVersion                    = 11
-	codeCancelledDDLJob                      = 12
+	codeInvalidWorker          terror.ErrCode = 1
+	codeNotOwner                              = 2
+	codeInvalidDDLJob                         = 3
+	codeInvalidJobFlag                        = 5
+	codeRunMultiSchemaChanges                 = 6
+	codeWaitReorgTimeout                      = 7
+	codeInvalidStoreVer                       = 8
+	codeUnknownTypeLength                     = 9
+	codeUnknownFractionLength                 = 10
+	codeInvalidJobVersion                     = 11
+	codeCancelledDDLJob                       = 12
+	codeInvalidRanges                         = 13
+	codeReorgWorkerNotRunnable                = 14
 
 	codeInvalidDBState         = 100
 	codeInvalidTableState      = 101

ddl/ddl_db_test.go

@@ -313,6 +313,11 @@ func (s *testDBSuite) TestCancelAddIndex(c *C) {
 	var checkErr error
 	hook := &ddl.TestDDLCallback{}
 	first := true
+	oldReorgWaitTimeout := ddl.ReorgWaitTimeout
+	// let hook.OnJobUpdatedExported has chance to cancel the job.
+	// the hook.OnJobUpdatedExported is called when the job is updated, runReorgJob will wait ddl.ReorgWaitTimeout, then return the ddl.runDDLJob.
+	// After that ddl call d.hook.OnJobUpdated(job), so that we can canceled the job in this test case.
+	ddl.ReorgWaitTimeout = 50 * time.Millisecond
 	hook.OnJobUpdatedExported = func(job *model.Job) {
 		addIndexNotFirstReorg := job.Type == model.ActionAddIndex && job.SchemaState == model.StateWriteReorganization && job.SnapshotVer != 0
 		// If the action is adding index and the state is writing reorganization, it want to test the case of cancelling the job when backfilling indexes.
@@ -391,6 +396,7 @@ LOOP:
 	}
 
 	s.mustExec(c, "drop table t1")
+	ddl.ReorgWaitTimeout = oldReorgWaitTimeout
 }
 
 func (s *testDBSuite) TestAddAnonymousIndex(c *C) {

ddl/ddl_worker.go

@@ -283,7 +283,7 @@ func (d *ddl) runDDLJob(t *meta.Meta, job *model.Job) (ver int64, err error) {
 		// If the value of SnapshotVer isn't zero, it means the work is backfilling the indexes.
 		if job.Type == model.ActionAddIndex && job.SchemaState == model.StateWriteReorganization && job.SnapshotVer != 0 {
 			log.Infof("[ddl] run the cancelling DDL job %s", job)
-			asyncNotify(d.reorgCtx.notifyCancelReorgJob)
+			d.reorgCtx.notifyReorgCancel()
 		} else {
 			job.State = model.JobStateCancelled
 			job.Error = errCancelledDDLJob

ddl/ddl_worker_test.go

@@ -37,6 +37,11 @@ type testDDLSuite struct{}
 
 const testLease = 5 * time.Millisecond
 
+func (s *testDDLSuite) SetUpSuite(c *C) {
+	// set ReorgWaitTimeout to small value, make test to be faster.
+	ReorgWaitTimeout = 50 * time.Millisecond
+}
+
 func (s *testDDLSuite) TestCheckOwner(c *C) {
 	defer testleak.AfterTest(c)()
 	store := testCreateStore(c, "test_owner")