ddl, parser: implement the normal flow of distributed processing in DDL reorg phase for adding index

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:B2FNmPDaGirXpIOgQbqxiukIkT8eOT4tKEahqYE2ers=",
         version = "v2.0.5-0.20230112062023-fe5b35c5f5dc",
     )
+    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",

Makefile

@@ -408,6 +408,10 @@ bazel_coverage_test: failpoint-enable bazel_ci_prepare
 		--build_event_json_file=bazel_1.json --@io_bazel_rules_go//go/config:cover_format=go_cover --define gotags=deadlock \
 		-- //... -//cmd/... -//tests/graceshutdown/... \
 		-//tests/globalkilltest/... -//tests/readonlytest/... -//br/pkg/task:task_test -//tests/realtikvtest/...
+	bazel $(BAZEL_GLOBAL_CONFIG) coverage $(BAZEL_CMD_CONFIG) \
+		--build_event_json_file=bazel_1.json --@io_bazel_rules_go//go/config:cover_format=go_cover --define gotags=deadlock,distributereorg \
+		-- //... -//cmd/... -//tests/graceshutdown/... \
+		-//tests/globalkilltest/... -//tests/readonlytest/... -//br/pkg/task:task_test -//tests/realtikvtest/...
 
 bazel_build: bazel_ci_prepare
 	mkdir -p bin

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",
@@ -79,6 +80,8 @@ go_library(
         "//parser/opcode",
         "//parser/terror",
         "//parser/types",
+        "//resourcemanager/pooltask",
+        "//resourcemanager/util",
         "//sessionctx",
         "//sessionctx/binloginfo",
         "//sessionctx/stmtctx",
@@ -104,6 +107,7 @@ go_library(
         "//util/filter",
         "//util/gcutil",
         "//util/generic",
+        "//util/gpool/spmc",
         "//util/hack",
         "//util/logutil",
         "//util/mathutil",

ddl/attributes_sql_test.go

@@ -73,6 +73,8 @@ func TestAlterTableAttributes(t *testing.T) {
 	// without equal
 	tk.MustExec(`alter table alter_t attributes " merge_option=allow ";`)
 	tk.MustExec(`alter table alter_t attributes " merge_option=allow , key=value ";`)
+
+	tk.MustExec("drop table alter_t")
 }
 
 func TestAlterTablePartitionAttributes(t *testing.T) {
@@ -134,6 +136,8 @@ PARTITION BY RANGE (c) (
 	require.Len(t, rows4, 1)
 	require.NotEqual(t, rows3[0][3], rows4[0][3])
 	require.NotEqual(t, rows[0][3], rows4[0][3])
+
+	tk.MustExec("drop table alter_p")
 }
 
 func TestTruncateTable(t *testing.T) {