ddl, parser: remove usless file and tiny update

DEPS.bzl

@@ -2812,8 +2812,8 @@ def go_deps():
         name = "com_github_opentracing_basictracer_go",
         build_file_proto_mode = "disable_global",
         importpath = "github.com/opentracing/basictracer-go",
-        sum = "h1:YyUAhaEfjoWXclZVJ9sGoNct7j4TVk7lZWlQw5UXuoo=",
-        version = "v1.0.0",
+        sum = "h1:Oa1fTSBvAl8pa3U+IJYqrKm0NALwH9OsgwOqDv4xJW0=",
+        version = "v1.1.0",
     )
     go_repository(
         name = "com_github_opentracing_contrib_go_stdlib",
@@ -2917,6 +2917,14 @@ def go_deps():
         sum = "h1:JhzVVoYvbOACxoUmOs6V/G4D5nPVUW73rKvXxP4XUJc=",
         version = "v0.0.0-20180830031419-95f893ade6f2",
     )
+    go_repository(
+        name = "com_github_philhofer_fwd",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/philhofer/fwd",
+        sum = "h1:GdGcTjf5RNAxwS4QLsiMzJYj5KEvPJD3Abr261yRQXQ=",
+        version = "v1.1.1",
+    )
+
     go_repository(
         name = "com_github_pierrec_lz4",
         build_file_proto_mode = "disable_global",
@@ -3360,6 +3368,14 @@ def go_deps():
         sum = "h1:ug7PpSOB5RBPK1Kg6qskGBoP3Vnj/aNYFTznWvlkGo0=",
         version = "v0.0.0-20181202132449-6a9ea43bcacd",
     )
+    go_repository(
+        name = "com_github_silentred_gid",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/silentred/gid",
+        sum = "h1:JdsH8McqPUeY8IN4C0gxENnJG2zysvh+/xDJWhPvGVQ=",
+        version = "v1.0.0",
+    )
+
     go_repository(
         name = "com_github_sirupsen_logrus",
         build_file_proto_mode = "disable_global",
@@ -3585,6 +3601,14 @@ def go_deps():
         sum = "h1:RI6bs9TDIIJ96N0lR5uZoGO8QNot4qS/1l+Mobx0InM=",
         version = "v2.0.5-0.20230110071533-f313ddf58d73",
     )
+    go_repository(
+        name = "com_github_tikv_minitrace_go",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/tikv/minitrace-go",
+        sum = "h1:nvIrUVo5YJZMsCn6yTxrpgrokIo/wug5N/nL5mc7v50=",
+        version = "v0.0.0-20220923091513-8e6316bb4097",
+    )
+
     go_repository(
         name = "com_github_tikv_pd_client",
         build_file_proto_mode = "disable_global",
@@ -3614,6 +3638,13 @@ def go_deps():
         sum = "h1:phZCcypL/vtx6cGxObJgWZ5wexZF5SXFPLOM+ru0e/M=",
         version = "v0.1.0",
     )
+    go_repository(
+        name = "com_github_tinylib_msgp",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/tinylib/msgp",
+        sum = "h1:2gXmtWueD2HefZHQe1QOy9HVzmFrLOVvsXwXBQ0ayy0=",
+        version = "v1.1.5",
+    )
 
     go_repository(
         name = "com_github_tklauser_go_sysconf",
@@ -3650,6 +3681,13 @@ def go_deps():
         sum = "h1:NowYhSdyE/1zwK9QCLeRb6USWdoif80Ie+v+yU8u1Zw=",
         version = "v2.5.1",
     )
+    go_repository(
+        name = "com_github_ttacon_chalk",
+        build_file_proto_mode = "disable",
+        importpath = "github.com/ttacon/chalk",
+        sum = "h1:OXcKh35JaYsGMRzpvFkLv/MEyPuL49CThT1pZ8aSml4=",
+        version = "v0.0.0-20160626202418-22c06c80ed31",
+    )
 
     go_repository(
         name = "com_github_twmb_murmur3",

ddl/BUILD.bazel

@@ -24,6 +24,7 @@ go_library(
         "ddl_workerpool.go",
         "delete_range.go",
         "delete_range_util.go",
+        "dist_backfilling.go",
         "foreign_key.go",
         "generated_column.go",
         "index.go",
@@ -80,6 +81,8 @@ go_library(
         "//parser/opcode",
         "//parser/terror",
         "//parser/types",
+        "//resourcemanager/pooltask",
+        "//resourcemanager/util",
         "//sessionctx",
         "//sessionctx/binloginfo",
         "//sessionctx/stmtctx",
@@ -105,6 +108,7 @@ go_library(
         "//util/filter",
         "//util/gcutil",
         "//util/generic",
+        "//util/gpool/spmc",
         "//util/hack",
         "//util/logutil",
         "//util/mathutil",

ddl/backfilling.go

@@ -462,8 +462,6 @@ func (w *backfillWorker) runTask(task *reorgBackfillTask) (result *backfillResul
 	// Change the batch size dynamically.
 	w.GetCtx().batchCnt = int(variable.GetDDLReorgBatchSize())
 	result = w.handleBackfillTask(w.GetCtx().ddlCtx, task, w.backfiller)
-	finish := injectSpan(task.bfJob.JobID+100, fmt.Sprintf("handle-job-%d-task-%d", task.bfJob.JobID, task.bfJob.ID))
-	defer finish()
 	task.bfJob.RowCount = int64(result.addedCount)
 	if result.err != nil {
 		logutil.BgLogger().Warn("[ddl] backfill worker runTask failed",
@@ -1186,7 +1184,6 @@ func (dc *ddlCtx) splitTableToBackfillJobs(sess *session, reorgInfo *reorgInfo,
 }
 
 func (dc *ddlCtx) controlWritePhysicalTableRecord(sess *session, t table.PhysicalTable, bfWorkerType backfillerType, reorgInfo *reorgInfo) error {
-	defer injectSpan(reorgInfo.Job.ID, "control-write-records")()
 	startKey, endKey := reorgInfo.StartKey, reorgInfo.EndKey
 	if startKey == nil && endKey == nil {
 		return nil
@@ -1284,6 +1281,7 @@ func checkJobIsFinished(sess *session, ddlJobID int64) (bool, error) {
 	return false, errors.Trace(err)
 }
 
+// GetBackfillErr gets the error in backfill job.
 func GetBackfillErr(sess *session, ddlJobID, currEleID int64, currEleKey []byte) error {
 	var err error
 	var metas []*model.BackfillMeta

ddl/ddl.go

@@ -973,15 +973,15 @@ func getJobCheckInterval(job *model.Job, i int) (time.Duration, bool) {
 	}
 }
 
-func (d *ddlCtx) asyncNotifyWorker(ch chan struct{}, etcdPath string, jobID int64, jobType string) {
+func (dc *ddlCtx) asyncNotifyWorker(ch chan struct{}, etcdPath string, jobID int64, jobType string) {
 	// If the workers don't run, we needn't notify workers.
 	if !config.GetGlobalConfig().Instance.TiDBEnableDDL.Load() {
 		return
 	}
-	if d.isOwner() {
+	if dc.isOwner() {
 		asyncNotify(ch)
 	} else {
-		d.asyncNotifyByEtcd(etcdPath, jobID, jobType)
+		dc.asyncNotifyByEtcd(etcdPath, jobID, jobType)
 	}
 }
 

ddl/ddl_worker.go

@@ -58,7 +58,7 @@ var (
 	WaitTimeWhenErrorOccurred = int64(1 * time.Second)
 
 	mockDDLErrOnce = int64(0)
-	// TestNotifyBeginTxnCh is used for if the txn is begining in runInTxn.
+	// TestNotifyBeginTxnCh is used for if the txn is beginning in runInTxn.
 	TestNotifyBeginTxnCh = make(chan struct{})
 )
 

ddl/dist_backfilling.go

@@ -166,6 +166,7 @@ func (dc *ddlCtx) backfillJob2Task(t table.Table, bfJob *BackfillJob) (*reorgBac
 		priority:   bfJob.Meta.Priority}, nil
 }
 
+// GetTasks gets the backfill tasks associated with the non-runningJobID.
 func GetTasks(d *ddlCtx, sess *session, tbl table.Table, runningJobID int64, concurrency int) ([]*reorgBackfillTask, error) {
 	// TODO: At present, only add index is processed. In the future, different elements need to be distinguished.
 	var err error

ddl/index.go

@@ -610,7 +610,6 @@ func (w *worker) onCreateIndex(d *ddlCtx, t *meta.Meta, job *model.Job, isPK boo
 			return ver, err
 		}
 		logutil.BgLogger().Info("[ddl] run add index job", zap.String("job", job.String()), zap.Reflect("indexInfo", indexInfo))
-		initializeTrace(job.ID)
 	}
 	originalState := indexInfo.State
 	switch indexInfo.State {
@@ -656,7 +655,6 @@ func (w *worker) onCreateIndex(d *ddlCtx, t *meta.Meta, job *model.Job, isPK boo
 		job.SnapshotVer = 0
 		job.SchemaState = model.StateWriteReorganization
 	case model.StateWriteReorganization:
-		defer injectSpan(job.ID, "write-reorg")()
 		// reorganization -> public
 		tbl, err := getTable(d.store, schemaID, tblInfo)
 		if err != nil {
@@ -2059,15 +2057,13 @@ func runBackfillJobsWithLightning(d *ddl, sess *session, bfJob *BackfillJob, job
 		return err
 	}
 
-	finish := injectSpan(bfJob.JobID, "finish-import")
 	err = bc.FinishImport(bfJob.EleID, bfJob.Meta.IsUnique, tbl)
 	if err != nil {
 		logutil.BgLogger().Warn("[ddl] lightning import error", zap.String("first backfill job", bfJob.AbbrStr()), zap.Error(err))
 		ingest.LitBackCtxMgr.Unregister(bfJob.JobID)
 		return err
 	}
 	ingest.LitBackCtxMgr.Unregister(bfJob.ID)
-	finish()
 	bc.SetDone()
 	return nil
 }

ddl/job_table.go

@@ -411,7 +411,7 @@ func (d *ddl) loadBackfillJobAndRun() {
 	d.wg.Run(func() {
 		defer func() {
 			d.removeBackfillCtxJobCtx(bJob.JobID)
-			tidbutil.Recover(metrics.LabelBackfillWorker, fmt.Sprintf("runBackfillJobs"), nil, false)
+			tidbutil.Recover(metrics.LabelBackfillWorker, "runBackfillJobs", nil, false)
 		}()
 
 		if bJob.Meta.ReorgTp == model.ReorgTypeLitMerge {
@@ -847,6 +847,7 @@ func GetBackfillJobCount(sess *session, tblName, condition string, label string)
 	return int(rows[0].GetInt64(0)), nil
 }
 
+// GetBackfillMetas gets the backfill metas in the tblName table according to condition.
 func GetBackfillMetas(sess *session, tblName, condition string, label string) ([]*model.BackfillMeta, error) {
 	rows, err := sess.execute(context.Background(), fmt.Sprintf("select backfill_meta from mysql.%s where %s", tblName, condition), label)
 	if err != nil {

go.mod

@@ -91,7 +91,6 @@ require (
 	github.com/tdakkota/asciicheck v0.1.1
 	github.com/tiancaiamao/appdash v0.0.0-20181126055449-889f96f722a2
 	github.com/tikv/client-go/v2 v2.0.5-0.20230110071533-f313ddf58d73
-	github.com/tikv/minitrace-go v0.0.0-20220923091513-8e6316bb4097
 	github.com/tikv/pd/client v0.0.0-20221031025758-80f0d8ca4d07
 	github.com/timakin/bodyclose v0.0.0-20210704033933-f49887972144
 	github.com/twmb/murmur3 v1.1.3