Repository: pingcap/tidb
Pull request #52629 (2024-04-16 03:30:31)

*: tools in scripts use fixed version to avoid incompatible updates
<!--

Thank you for contributing to TiDB!

PR Title Format:
1. pkg [, pkg2, pkg3]: what's changed
2. *: what's changed

-->

### What problem does this PR solve?
<!--

Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.

-->

Issue Number: ref https://github.com/pingcap/tidb/issues/52473

Problem Summary:

### What changed and how does it work?
* installing third-party tools requires specifying the version to avoid incompatible changes as time goes by.
* fix test utils load path

### Check List

Tests <!-- At least one of them must be included. -->

- [ ] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [x] No need to test
  > - [x] I checked and no code files have been changed.
  > <!-- Or your custom  "No need to test" reasons -->

Side effects

- [ ] Performance regression: Consumes more CPU
- [ ] Performance regression: Consumes more Memory
- [ ] Breaking backward compatibility

Documentation

- [ ] Affects user behaviors
- [ ] Contains syntax changes
- [ ] Contains variable changes
- [ ] Contains experimental features
- [ ] Changes MySQL compatibility

### Release note

<!-- compatibility change, improvement, bugfix, and new feature need a release note -->

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

```release-note
None
```


ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please assign [lichunzhu](https://github.com/lichunzhu) for approval. For more information see [the Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=pingcap%2Ftidb).

<details open>
Needs approval from an approver in each of these files:

- **[dumpling/OWNERS](https://github.com/pingcap/tidb/blob/master/dumpling/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["lichunzhu"]} -->
tiprow[bot]: Hi @purelind. Thanks for your PR.

PRs from untrusted users cannot be marked as trusted with `/ok-to-test` in this repo meaning untrusted PR authors can never trigger tests themselves. Collaborators can still trigger tests on the PR using `/test all`.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=pingcap%2Ftidb).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

purelind: /cc @gmhdbjd @okJiang

Patch:
diff --git a/br/pkg/conn/conn.go b/br/pkg/conn/conn.go
index 14292ae091..16c86e18c8 100644
--- a/br/pkg/conn/conn.go
+++ b/br/pkg/conn/conn.go
@@ -381,11 +381,11 @@ func (mgr *Mgr) GetConfigFromTiKV(ctx context.Context, cli *http.Client, fn func
 			if e != nil {
 				return e
 			}
+			defer resp.Body.Close()
 			err = fn(resp)
 			if err != nil {
 				return err
 			}
-			_ = resp.Body.Close()
 			return nil
 		}, utils.NewPDReqBackoffer())
 		if err != nil {
diff --git a/cmd/tidb-server/main.go b/cmd/tidb-server/main.go
index 696e43469b..0fae4d6773 100644
--- a/cmd/tidb-server/main.go
+++ b/cmd/tidb-server/main.go
@@ -906,7 +906,6 @@ func createServer(storage kv.Storage, dom *domain.Domain) *server.Server {
 		closeDomainAndStorage(storage, dom)
 		log.Fatal("failed to create the server", zap.Error(err), zap.Stack("stack"))
 	}
-	mppcoordmanager.InstanceMPPCoordinatorManager.InitServerAddr(svr.GetStatusServerAddr())
 	svr.SetDomain(dom)
 	go dom.ExpensiveQueryHandle().SetSessionManager(svr).Run()
 	go dom.MemoryUsageAlarmHandle().SetSessionManager(svr).Run()
diff --git a/dumpling/install.sh b/dumpling/install.sh
index f24d54b499..e218d2b337 100644
--- a/dumpling/install.sh
+++ b/dumpling/install.sh
@@ -16,8 +16,8 @@ chmod a+x bin/minio
 wget https://dl.minio.io/client/mc/release/linux-amd64/mc -O bin/mc
 chmod a+x bin/mc
 
-go get github.com/ma6174/snappy
-go install github.com/ma6174/snappy
+go get github.com/ma6174/snappy@15869b0666f67839ecf86cd29ef1452ddcd79cb8
+go install github.com/ma6174/snappy@15869b0666f67839ecf86cd29ef1452ddcd79cb8
 
 wget https://github.com/facebook/zstd/releases/download/v1.5.2/zstd-1.5.2.tar.gz
 tar xvfz zstd-1.5.2.tar.gz
diff --git a/pkg/ddl/backfilling_operators.go b/pkg/ddl/backfilling_operators.go
index b9bb3d438f..4c1cfe7d60 100644
--- a/pkg/ddl/backfilling_operators.go
+++ b/pkg/ddl/backfilling_operators.go
@@ -150,7 +150,7 @@ func NewAddIndexIngestPipeline(
 
 	srcOp := NewTableScanTaskSource(ctx, store, tbl, startKey, endKey)
 	scanOp := NewTableScanOperator(ctx, sessPool, copCtx, srcChkPool, readerCnt)
-	ingestOp := NewIndexIngestOperator(ctx, copCtx, sessPool, tbl, indexes, engines, srcChkPool, writerCnt, reorgMeta)
+	ingestOp := NewIndexIngestOperator(ctx, copCtx, backendCtx, sessPool, tbl, indexes, engines, srcChkPool, writerCnt, reorgMeta)
 	sinkOp := newIndexWriteResultSink(ctx, backendCtx, tbl, indexes, totalRowCount, metricCounter)
 
 	operator.Compose[TableScanTask](srcOp, scanOp)
@@ -534,16 +534,18 @@ func NewWriteExternalStoreOperator(
 				writers = append(writers, writer)
 			}
 
-			return &indexIngestWorker{
-				ctx:          ctx,
-				tbl:          tbl,
-				indexes:      indexes,
-				copCtx:       copCtx,
-				se:           nil,
-				sessPool:     sessPool,
-				writers:      writers,
-				srcChunkPool: srcChunkPool,
-				reorgMeta:    reorgMeta,
+			return &indexIngestExternalWorker{
+				indexIngestBaseWorker: indexIngestBaseWorker{
+					ctx:          ctx,
+					tbl:          tbl,
+					indexes:      indexes,
+					copCtx:       copCtx,
+					se:           nil,
+					sessPool:     sessPool,
+					writers:      writers,
+					srcChunkPool: srcChunkPool,
+					reorgMeta:    reorgMeta,
+				},
 			}
 		})
 	return &WriteExternalStoreOperator{
@@ -568,6 +570,7 @@ type IndexIngestOperator struct {
 func NewIndexIngestOperator(
 	ctx *OperatorCtx,
 	copCtx copr.CopContext,
+	backendCtx ingest.BackendCtx,
 	sessPool opSessPool,
 	tbl table.PhysicalTable,
 	indexes []table.Index,
@@ -593,16 +596,25 @@ func NewIndexIngestOperator(
 				writers = append(writers, writer)
 			}
 
-			return &indexIngestWorker{
-				ctx:          ctx,
-				tbl:          tbl,
-				indexes:      indexes,
-				copCtx:       copCtx,
-				se:           nil,
-				sessPool:     sessPool,
-				writers:      writers,
-				srcChunkPool: srcChunkPool,
-				reorgMeta:    reorgMeta,
+			indexIDs := make([]int64, len(indexes))
+			for i := 0; i < len(indexes); i++ {
+				indexIDs[i] = indexes[i].Meta().ID
+			}
+			return &indexIngestLocalWorker{
+				indexIngestBaseWorker: indexIngestBaseWorker{
+					ctx:     ctx,
+					tbl:     tbl,
+					indexes: indexes,
+					copCtx:  copCtx,
+
+					se:           nil,
+					sessPool:     sessPool,
+					writers:      writers,
+					srcChunkPool: srcChunkPool,
+					reorgMeta:    reorgMeta,
+				},
+				indexIDs:   indexIDs,
+				backendCtx: backendCtx,
 			}
 		})
 	return &IndexIngestOperator{
@@ -610,7 +622,47 @@ func NewIndexIngestOperator(
 	}
 }
 
-type indexIngestWorker struct {
+type indexIngestExternalWorker struct {
+	indexIngestBaseWorker
+}
+
+func (w *indexIngestExternalWorker) HandleTask(rs IndexRecordChunk, send func(IndexWriteResult)) {
+	defer tidbutil.Recover(metrics.LblAddIndex, "indexIngestExternalWorkerRecover", func() {
+		w.ctx.onError(errors.New("met panic in indexIngestExternalWorker"))
+	}, false)
+	defer func() {
+		if rs.Chunk != nil {
+			w.srcChunkPool <- rs.Chunk
+		}
+	}()
+	w.indexIngestBaseWorker.HandleTask(rs, send)
+}
+
+type indexIngestLocalWorker struct {
+	indexIngestBaseWorker
+	indexIDs   []int64
+	backendCtx ingest.BackendCtx
+}
+
+func (w *indexIngestLocalWorker) HandleTask(rs IndexRecordChunk, send func(IndexWriteResult)) {
+	defer tidbutil.Recover(metrics.LblAddIndex, "indexIngestLocalWorkerRecover", func() {
+		w.ctx.onError(errors.New("met panic in indexIngestLocalWorker"))
+	}, false)
+	defer func() {
+		if rs.Chunk != nil {
+			w.srcChunkPool <- rs.Chunk
+		}
+	}()
+	w.indexIngestBaseWorker.HandleTask(rs, send)
+	// needs to flush and import to avoid too much use of disk.
+	_, _, _, err := ingest.TryFlushAllIndexes(w.backendCtx, ingest.FlushModeAuto, w.indexIDs)
+	if err != nil {
+		w.ctx.onError(err)
+		return
+	}
+}
+
+type indexIngestBaseWorker struct {
 	ctx *OperatorCtx
 
 	tbl       table.PhysicalTable
@@ -626,16 +678,7 @@ type indexIngestWorker struct {
 	srcChunkPool chan *chunk.Chunk
 }
 
-func (w *indexIngestWorker) HandleTask(rs IndexRecordChunk, send func(IndexWriteResult)) {
-	defer func() {
-		if rs.Chunk != nil {
-			w.srcChunkPool <- rs.Chunk
-		}
-	}()
-	defer tidbutil.Recover(metrics.LblAddIndex, "handleIndexIngtestTaskWithRecover", func() {
-		w.ctx.onError(errors.New("met panic in indexIngestWorker"))
-	}, false)
-
+func (w *indexIngestBaseWorker) HandleTask(rs IndexRecordChunk, send func(IndexWriteResult)) {
 	failpoint.Inject("injectPanicForIndexIngest", func() {
 		panic("mock panic")
 	})
@@ -644,7 +687,7 @@ func (w *indexIngestWorker) HandleTask(rs IndexRecordChunk, send func(IndexWrite
 		ID: rs.ID,
 	}
 	w.initSessCtx()
-	count, nextKey, err := w.WriteLocal(&rs)
+	count, nextKey, err := w.WriteChunk(&rs)
 	if err != nil {
 		w.ctx.onError(err)
 		return
@@ -662,7 +705,7 @@ func (w *indexIngestWorker) HandleTask(rs IndexRecordChunk, send func(IndexWrite
 	send(result)
 }
 
-func (w *indexIngestWorker) initSessCtx() {
+func (w *indexIngestBaseWorker) initSessCtx() {
 	if w.se == nil {
 		sessCtx, err := w.sessPool.Get()
 		if err != nil {
@@ -681,7 +724,7 @@ func (w *indexIngestWorker) initSessCtx() {
 	}
 }
 
-func (w *indexIngestWorker) Close() {
+func (w *indexIngestBaseWorker) Close() {
 	for _, writer := range w.writers {
 		err := writer.Close(w.ctx)
 		if err != nil {
@@ -694,8 +737,8 @@ func (w *indexIngestWorker) Close() {
 	}
 }
 
-// WriteLocal will write index records to lightning engine.
-func (w *indexIngestWorker) WriteLocal(rs *IndexRecordChunk) (count int, nextKey kv.Key, err error) {
+// WriteChunk will write index records to lightning engine.
+func (w *indexIngestBaseWorker) WriteChunk(rs *IndexRecordChunk) (count int, nextKey kv.Key, err error) {
 	failpoint.Inject("mockWriteLocalError", func(_ failpoint.Value) {
 		failpoint.Return(0, nil, errors.New("mock write local error"))
 	})
diff --git a/pkg/ddl/ddl_worker.go b/pkg/ddl/ddl_worker.go
index 1e025eb0fd..4d209e66ae 100644
--- a/pkg/ddl/ddl_worker.go
+++ b/pkg/ddl/ddl_worker.go
@@ -600,7 +600,7 @@ func (w *worker) registerMDLInfo(job *model.Job, ver int64) error {
 }
 
 // cleanMDLInfo cleans metadata lock info.
-func cleanMDLInfo(pool *sess.Pool, jobID int64, ec *clientv3.Client) {
+func cleanMDLInfo(pool *sess.Pool, jobID int64, ec *clientv3.Client, cleanETCD bool) {
 	if !variable.EnableMDL.Load() {
 		return
 	}
@@ -614,7 +614,7 @@ func cleanMDLInfo(pool *sess.Pool, jobID int64, ec *clientv3.Client) {
 		logutil.BgLogger().Warn("unexpected error when clean mdl info", zap.Int64("job ID", jobID), zap.Error(err))
 		return
 	}
-	if ec != nil {
+	if cleanETCD && ec != nil {
 		path := fmt.Sprintf("%s/%d/", util.DDLAllSchemaVersionsByJob, jobID)
 		_, err = ec.Delete(context.Background(), path, clientv3.WithPrefix())
 		if err != nil {
diff --git a/pkg/ddl/ingest/BUILD.bazel b/pkg/ddl/ingest/BUILD.bazel
index 9e6435ed6c..d6a6fb9dae 100644
--- a/pkg/ddl/ingest/BUILD.bazel
+++ b/pkg/ddl/ingest/BUILD.bazel
@@ -11,6 +11,7 @@ go_library(
         "engine.go",
         "engine_mgr.go",
         "env.go",
+        "flush.go",
         "mem_root.go",
         "message.go",
         "mock.go",
diff --git a/pkg/ddl/ingest/checkpoint.go b/pkg/ddl/ingest/checkpoint.go
index 9be82954b3..67afd8dfd3 100644
--- a/pkg/ddl/ingest/checkpoint.go
+++ b/pkg/ddl/ingest/checkpoint.go
@@ -174,7 +174,7 @@ func (s *CheckpointManager) UpdateCurrent(taskID int, added int) error {
 	cp.currentKeys += added
 	s.mu.Unlock()
 
-	flushed, imported, err := s.tryFlushAllIndexes(FlushModeAuto)
+	flushed, imported, _, err := TryFlushAllIndexes(s.flushCtrl, FlushModeAuto, s.indexIDs)
 	if !flushed || err != nil {
 		return err
 	}
@@ -194,20 +194,6 @@ func (s *CheckpointManager) UpdateCurrent(taskID int, added int) error {
 	return nil
 }
 
-func (s *CheckpointManager) tryFlushAllIndexes(mode FlushMode) (flushed, imported bool, err error) {
-	allFlushed := true
-	allImported := true
-	for _, idxID := range s.indexIDs {
-		flushed, imported, err := s.flushCtrl.Flush(idxID, mode)
-		if err != nil {
-			return false, false, err
-		}
-		allFlushed = allFlushed && flushed
-		allImported = allImported && imported
-	}
-	return allFlushed, allImported, nil
-}
-
 func (s *CheckpointManager) progressLocalSyncMinKey() {
 	for {
 		cp := s.checkpoints[s.minTaskIDSynced+1]
@@ -232,7 +218,7 @@ func (s *CheckpointManager) Close() {
 
 // Sync syncs the checkpoint.
 func (s *CheckpointManager) Sync() {
-	_, _, err := s.tryFlushAllIndexes(FlushModeForceLocal)
+	_, _, _, err := TryFlushAllIndexes(s.flushCtrl, FlushModeForceLocal, s.indexIDs)
 	if err != nil {
 		logutil.BgLogger().Warn("flush local engine failed", zap.String("category", "ddl-ingest"), zap.Error(err))
 	}
diff --git a/pkg/ddl/ingest/flush.go b/pkg/ddl/ingest/flush.go
new file mode 100644
index 0000000000..4d407f8733
--- /dev/null
+++ b/pkg/ddl/ingest/flush.go
@@ -0,0 +1,30 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package ingest
+
+// TryFlushAllIndexes tries to flush and import all indexes.
+func TryFlushAllIndexes(flushCtrl FlushController, mode FlushMode, indexIDs []int64) (flushed, imported bool, failedIdxID int64, err error) {
+	allFlushed := true
+	allImported := true
+	for _, idxID := range indexIDs {
+		flushed, imported, err := flushCtrl.Flush(idxID, mode)
+		if err != nil {
+			return false, false, idxID, err
+		}
+		allFlushed = allFlushed && flushed
+		allImported = allImported && imported
+	}
+	return allFlushed, allImported, -1, nil
+}
diff --git a/pkg/ddl/job_table.go b/pkg/ddl/job_table.go
index 70033dac55..dddb986dc1 100644
--- a/pkg/ddl/job_table.go
+++ b/pkg/ddl/job_table.go
@@ -441,7 +441,7 @@ func (d *ddl) delivery2Worker(wk *worker, pool *workerPool, job *model.Job) {
 						return
 					}
 					d.setAlreadyRunOnce(job.ID)
-					cleanMDLInfo(d.sessPool, job.ID, d.etcdCli)
+					cleanMDLInfo(d.sessPool, job.ID, d.etcdCli, job.State == model.JobStateSynced)
 					// Don't have a worker now.
 					return
 				}
@@ -479,7 +479,7 @@ func (d *ddl) delivery2Worker(wk *worker, pool *workerPool, job *model.Job) {
 			if err != nil {
 				return
 			}
-			cleanMDLInfo(d.sessPool, job.ID, d.etcdCli)
+			cleanMDLInfo(d.sessPool, job.ID, d.etcdCli, job.State == model.JobStateSynced)
 			d.synced(job)
 
 			if RunInGoTest {
diff --git a/pkg/executor/analyze_col.go b/pkg/executor/analyze_col.go
index 989339daa4..2ba5272041 100644
--- a/pkg/executor/analyze_col.go
+++ b/pkg/executor/analyze_col.go
@@ -64,7 +64,7 @@ type AnalyzeColumnsExec struct {
 
 func analyzeColumnsPushDownEntry(gp *gp.Pool, e *AnalyzeColumnsExec) *statistics.AnalyzeResults {
 	if e.AnalyzeInfo.StatsVersion >= statistics.Version2 {
-		return e.toV2().analyzeColumnsPushDownWithRetryV2(gp)
+		return e.toV2().analyzeColumnsPushDownV2(gp)
 	}
 	return e.toV1().analyzeColumnsPushDownV1()
 }
diff --git a/pkg/executor/analyze_col_v2.go b/pkg/executor/analyze_col_v2.go
index 32032f8da1..eeba3f89bd 100644
--- a/pkg/executor/analyze_col_v2.go
+++ b/pkg/executor/analyze_col_v2.go
@@ -17,13 +17,11 @@ package executor
 import (
 	"context"
 	stderrors "errors"
-	"math"
 	"slices"
 	"time"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
-	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/metrics"
@@ -54,47 +52,6 @@ type AnalyzeColumnsExecV2 struct {
 	*AnalyzeColumnsExec
 }
 
-func (e *AnalyzeColumnsExecV2) analyzeColumnsPushDownWithRetryV2(gp *gp.Pool) *statistics.AnalyzeResults {
-	analyzeResult := e.analyzeColumnsPushDownV2(gp)
-	if e.notRetryable(analyzeResult) {
-		return analyzeResult
-	}
-
-	finishJobWithLog(e.ctx, analyzeResult.Job, analyzeResult.Err)
-	statsHandle := domain.GetDomain(e.ctx).StatsHandle()
-	if statsHandle == nil {
-		return analyzeResult
-	}
-
-	var statsTbl *statistics.Table
-	tid := e.tableID.GetStatisticsID()
-	if tid == e.tableInfo.ID {
-		statsTbl = statsHandle.GetTableStats(e.tableInfo)
-	} else {
-		statsTbl = statsHandle.GetPartitionStats(e.tableInfo, tid)
-	}
-	if statsTbl == nil || statsTbl.RealtimeCount <= 0 {
-		return analyzeResult
-	}
-
-	newSampleRate := math.Min(1, float64(config.DefRowsForSampleRate)/float64(statsTbl.RealtimeCount))
-	if newSampleRate >= *e.analyzePB.ColReq.SampleRate {
-		return analyzeResult
-	}
-	*e.analyzePB.ColReq.SampleRate = newSampleRate
-	prepareV2AnalyzeJobInfo(e.AnalyzeColumnsExec, true)
-	AddNewAnalyzeJob(e.ctx, e.job)
-	StartAnalyzeJob(e.ctx, e.job)
-	return e.analyzeColumnsPushDownV2(gp)
-}
-
-// Do **not** retry if succeed / not oom error / not auto-analyze / samplerate not set.
-func (e *AnalyzeColumnsExecV2) notRetryable(analyzeResult *statistics.AnalyzeResults) bool {
-	return analyzeResult.Err == nil || analyzeResult.Err != errAnalyzeOOM ||
-		!e.ctx.GetSessionVars().InRestrictedSQL ||
-		e.analyzePB.ColReq == nil || *e.analyzePB.ColReq.SampleRate <= 0
-}
-
 func (e *AnalyzeColumnsExecV2) analyzeColumnsPushDownV2(gp *gp.Pool) *statistics.AnalyzeResults {
 	var ranges []*ranger.Range
 	if hc := e.handleCols; hc != nil {
diff --git a/pkg/planner/core/BUILD.bazel b/pkg/planner/core/BUILD.bazel
index 2eac2f143b..9fd690b582 100644
--- a/pkg/planner/core/BUILD.bazel
+++ b/pkg/planner/core/BUILD.bazel
@@ -113,9 +113,8 @@ go_library(
         "//pkg/planner/cardinality",
         "//pkg/planner/context",
         "//pkg/planner/core/base",
-        "//pkg/planner/core/internal",
-        "//pkg/planner/core/internal/base",
         "//pkg/planner/core/metrics",
+        "//pkg/planner/core/operator/base",
         "//pkg/planner/funcdep",
         "//pkg/planner/property",
         "//pkg/planner/util",
@@ -251,9 +250,9 @@ go_test(
         "//pkg/parser/terror",
         "//pkg/planner",
         "//pkg/planner/core/base",
-        "//pkg/planner/core/internal",
         "//pkg/planner/property",
         "//pkg/planner/util",
+        "//pkg/planner/util/coretestsdk",
         "//pkg/planner/util/coreusage",
         "//pkg/session",
         "//pkg/session/types",
diff --git a/pkg/planner/core/initialize.go b/pkg/planner/core/initialize.go
index 15d7cbc991..29885a7622 100644
--- a/pkg/planner/core/initialize.go
+++ b/pkg/planner/core/initialize.go
@@ -18,7 +18,7 @@ import (
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/kv"
 	base2 "github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/internal/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/base"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/plancodec"
diff --git a/pkg/planner/core/internal/base/BUILD.bazel b/pkg/planner/core/operator/base/BUILD.bazel
similarity index 70%
rename from pkg/planner/core/internal/base/BUILD.bazel
rename to pkg/planner/core/operator/base/BUILD.bazel
index c326999fe8..05d79f3631 100644
--- a/pkg/planner/core/internal/base/BUILD.bazel
+++ b/pkg/planner/core/operator/base/BUILD.bazel
@@ -3,8 +3,8 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library")
 go_library(
     name = "base",
     srcs = ["plan.go"],
-    importpath = "github.com/pingcap/tidb/pkg/planner/core/internal/base",
-    visibility = ["//pkg/planner/core:__subpackages__"],
+    importpath = "github.com/pingcap/tidb/pkg/planner/core/operator/base",
+    visibility = ["//visibility:public"],
     deps = [
         "//pkg/expression",
         "//pkg/planner/context",
diff --git a/pkg/planner/core/plan.go b/pkg/planner/core/plan.go
index 0d0a5d7e19..f6d0f68c0e 100644
--- a/pkg/planner/core/plan.go
+++ b/pkg/planner/core/plan.go
@@ -22,7 +22,7 @@ import (
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/planner/cardinality"
 	base2 "github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/internal/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/base"
 	fd "github.com/pingcap/tidb/pkg/planner/funcdep"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util"
diff --git a/pkg/planner/core/point_get_plan.go b/pkg/planner/core/point_get_plan.go
index 342c58c1b3..da24d0066a 100644
--- a/pkg/planner/core/point_get_plan.go
+++ b/pkg/planner/core/point_get_plan.go
@@ -34,7 +34,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	ptypes "github.com/pingcap/tidb/pkg/parser/types"
 	base2 "github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/internal/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/base"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util/coreusage"
 	"github.com/pingcap/tidb/pkg/privilege"
diff --git a/pkg/planner/core/rule_inject_extra_projection.go b/pkg/planner/core/rule_inject_extra_projection.go
index 144e6e4860..11f893e544 100644
--- a/pkg/planner/core/rule_inject_extra_projection.go
+++ b/pkg/planner/core/rule_inject_extra_projection.go
@@ -21,8 +21,8 @@ import (
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
 	"github.com/pingcap/tidb/pkg/planner/util"
+	"github.com/pingcap/tidb/pkg/planner/util/coreusage"
 )
 
 // InjectExtraProjection is used to extract the expressions of specific
@@ -116,7 +116,7 @@ func injectProjBelowUnion(un *PhysicalUnionAll) *PhysicalUnionAll {
 func InjectProjBelowAgg(aggPlan base.PhysicalPlan, aggFuncs []*aggregation.AggFuncDesc, groupByItems []expression.Expression) base.PhysicalPlan {
 	hasScalarFunc := false
 
-	internal.WrapCastForAggFuncs(aggPlan.SCtx().GetExprCtx(), aggFuncs)
+	coreusage.WrapCastForAggFuncs(aggPlan.SCtx().GetExprCtx(), aggFuncs)
 	for i := 0; !hasScalarFunc && i < len(aggFuncs); i++ {
 		for _, arg := range aggFuncs[i].Args {
 			_, isScalarFunc := arg.(*expression.ScalarFunction)
diff --git a/pkg/planner/core/scalar_subq_expression.go b/pkg/planner/core/scalar_subq_expression.go
index 54f5c60093..238c9168fb 100644
--- a/pkg/planner/core/scalar_subq_expression.go
+++ b/pkg/planner/core/scalar_subq_expression.go
@@ -23,7 +23,7 @@ import (
 	"github.com/pingcap/tidb/pkg/expression"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	base2 "github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/internal/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/base"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/chunk"
 	"github.com/pingcap/tidb/pkg/util/codec"
diff --git a/pkg/planner/core/task.go b/pkg/planner/core/task.go
index 6412b6079c..9908b17bb2 100644
--- a/pkg/planner/core/task.go
+++ b/pkg/planner/core/task.go
@@ -27,7 +27,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/planner/cardinality"
 	base2 "github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/internal/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/base"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util"
 	"github.com/pingcap/tidb/pkg/types"
diff --git a/pkg/planner/core/util.go b/pkg/planner/core/util.go
index 3ed107a56e..54009a4568 100644
--- a/pkg/planner/core/util.go
+++ b/pkg/planner/core/util.go
@@ -25,7 +25,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	base2 "github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/internal/base"
+	"github.com/pingcap/tidb/pkg/planner/core/operator/base"
 	"github.com/pingcap/tidb/pkg/planner/util/coreusage"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/table"
diff --git a/pkg/planner/core/internal/BUILD.bazel b/pkg/planner/util/coretestsdk/BUILD.bazel
similarity index 56%
rename from pkg/planner/core/internal/BUILD.bazel
rename to pkg/planner/util/coretestsdk/BUILD.bazel
index 3c16228db7..0f5bb3aa34 100644
--- a/pkg/planner/core/internal/BUILD.bazel
+++ b/pkg/planner/util/coretestsdk/BUILD.bazel
@@ -1,17 +1,12 @@
 load("@io_bazel_rules_go//go:def.bzl", "go_library")
 
 go_library(
-    name = "internal",
-    srcs = [
-        "testkit.go",
-        "util.go",
-    ],
-    importpath = "github.com/pingcap/tidb/pkg/planner/core/internal",
-    visibility = ["//pkg/planner/core:__subpackages__"],
+    name = "coretestsdk",
+    srcs = ["testkit.go"],
+    importpath = "github.com/pingcap/tidb/pkg/planner/util/coretestsdk",
+    visibility = ["//visibility:public"],
     deps = [
         "//pkg/domain",
-        "//pkg/expression",
-        "//pkg/expression/aggregation",
         "//pkg/parser/model",
         "//pkg/store/mockstore",
         "//pkg/store/mockstore/unistore",
diff --git a/pkg/planner/util/coreusage/BUILD.bazel b/pkg/planner/util/coreusage/BUILD.bazel
index 205dd98fb1..ccf6c58c16 100644
--- a/pkg/planner/util/coreusage/BUILD.bazel
+++ b/pkg/planner/util/coreusage/BUILD.bazel
@@ -3,10 +3,15 @@ load("@io_bazel_rules_go//go:def.bzl", "go_library")
 go_library(
     name = "coreusage",
     srcs = [
+        "cast_misc.go",
         "cost_misc.go",
         "opt_tracer.go",
     ],
     importpath = "github.com/pingcap/tidb/pkg/planner/util/coreusage",
     visibility = ["//visibility:public"],
-    deps = ["//pkg/util/tracing"],
+    deps = [
+        "//pkg/expression",
+        "//pkg/expression/aggregation",
+        "//pkg/util/tracing",
+    ],
 )
diff --git a/pkg/planner/core/internal/util.go b/pkg/planner/util/coreusage/cast_misc.go
similarity index 98%
rename from pkg/planner/core/internal/util.go
rename to pkg/planner/util/coreusage/cast_misc.go
index b5af5a6b63..b2161b9c65 100644
--- a/pkg/planner/core/internal/util.go
+++ b/pkg/planner/util/coreusage/cast_misc.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package coreusage
 
 import (
 	"github.com/pingcap/tidb/pkg/expression"
diff --git a/pkg/server/BUILD.bazel b/pkg/server/BUILD.bazel
index 223002db73..cc6391de1d 100644
--- a/pkg/server/BUILD.bazel
+++ b/pkg/server/BUILD.bazel
@@ -200,6 +200,7 @@ go_test(
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_pingcap_kvproto//pkg/metapb",
+        "@com_github_prometheus_client_golang//prometheus/testutil",
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//error",
         "@com_github_tikv_client_go_v2//testutils",
diff --git a/pkg/server/conn.go b/pkg/server/conn.go
index 46ed2038fb..073adf4dcd 100644
--- a/pkg/server/conn.go
+++ b/pkg/server/conn.go
@@ -1206,7 +1206,7 @@ func (cc *clientConn) addMetrics(cmd byte, startTime time.Time, err error) {
 	if counter != nil {
 		counter.Inc()
 	} else {
-		label := strconv.Itoa(int(cmd))
+		label := server_metrics.CmdToString(cmd)
 		if err != nil {
 			metrics.QueryTotalCounter.WithLabelValues(label, "Error", resourceGroupName).Inc()
 		} else {
diff --git a/pkg/server/metrics/metrics.go b/pkg/server/metrics/metrics.go
index ee0567ca95..24cd448283 100644
--- a/pkg/server/metrics/metrics.go
+++ b/pkg/server/metrics/metrics.go
@@ -15,6 +15,8 @@
 package metrics
 
 import (
+	"strconv"
+
 	"github.com/pingcap/tidb/pkg/domain/resourcegroup"
 	"github.com/pingcap/tidb/pkg/metrics"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
@@ -46,6 +48,39 @@ func init() {
 	InitMetricsVars()
 }
 
+// CmdToString convert command type to string.
+func CmdToString(cmd byte) string {
+	switch cmd {
+	case mysql.ComSleep:
+		return "Sleep"
+	case mysql.ComQuit:
+		return "Quit"
+	case mysql.ComInitDB:
+		return "InitDB"
+	case mysql.ComQuery:
+		return "Query"
+	case mysql.ComPing:
+		return "Ping"
+	case mysql.ComFieldList:
+		return "FieldList"
+	case mysql.ComStmtPrepare:
+		return "StmtPrepare"
+	case mysql.ComStmtExecute:
+		return "StmtExecute"
+	case mysql.ComStmtFetch:
+		return "StmtFetch"
+	case mysql.ComStmtClose:
+		return "StmtClose"
+	case mysql.ComStmtSendLongData:
+		return "StmtSendLongData"
+	case mysql.ComStmtReset:
+		return "StmtReset"
+	case mysql.ComSetOption:
+		return "SetOption"
+	}
+	return strconv.Itoa(int(cmd))
+}
+
 // InitMetricsVars init server metrics vars.
 func InitMetricsVars() {
 	QueryTotalCountOk = []prometheus.Counter{
diff --git a/pkg/server/server.go b/pkg/server/server.go
index 394573bb58..448dd39750 100644
--- a/pkg/server/server.go
+++ b/pkg/server/server.go
@@ -53,6 +53,7 @@ import (
 	autoid "github.com/pingcap/tidb/pkg/autoid_service"
 	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/domain"
+	"github.com/pingcap/tidb/pkg/executor/mppcoordmanager"
 	"github.com/pingcap/tidb/pkg/extension"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/metrics"
@@ -436,6 +437,7 @@ func (s *Server) Run(dom *domain.Domain) error {
 			log.Error("failed to create the server", zap.Error(err), zap.Stack("stack"))
 			return err
 		}
+		mppcoordmanager.InstanceMPPCoordinatorManager.InitServerAddr(s.GetStatusServerAddr())
 	}
 	if config.GetGlobalConfig().Performance.ForceInitStats && dom != nil {
 		<-dom.StatsHandle().InitStatsDone
diff --git a/pkg/statistics/scalar.go b/pkg/statistics/scalar.go
index b84d0933e6..105dda35cd 100644
--- a/pkg/statistics/scalar.go
+++ b/pkg/statistics/scalar.go
@@ -101,23 +101,26 @@ func (hg *Histogram) PreCalculateScalar() {
 	}
 	switch hg.GetLower(0).Kind() {
 	case types.KindMysqlDecimal, types.KindMysqlTime:
+		var lower, upper types.Datum
 		hg.Scalars = make([]scalar, l)
 		for i := 0; i < l; i++ {
-			hg.Scalars[i] = scalar{
-				lower: convertDatumToScalar(hg.GetLower(i), 0),
-				upper: convertDatumToScalar(hg.GetUpper(i), 0),
-			}
+			// It's read-only, so we don't need to allocate new datum each time.
+			hg.LowerToDatum(i, &lower)
+			hg.UpperToDatum(i, &upper)
+			hg.Scalars[i].lower = convertDatumToScalar(&lower, 0)
+			hg.Scalars[i].upper = convertDatumToScalar(&upper, 0)
 		}
 	case types.KindBytes, types.KindString:
+		var lower, upper types.Datum
 		hg.Scalars = make([]scalar, l)
 		for i := 0; i < l; i++ {
-			lower, upper := hg.GetLower(i), hg.GetUpper(i)
+			// It's read-only, so we don't need to allocate new datum each time.
+			hg.LowerToDatum(i, &lower)
+			hg.UpperToDatum(i, &upper)
 			common := commonPrefixLength(lower.GetBytes(), upper.GetBytes())
-			hg.Scalars[i] = scalar{
-				commonPfxLen: common,
-				lower:        convertDatumToScalar(lower, common),
-				upper:        convertDatumToScalar(upper, common),
-			}
+			hg.Scalars[i].commonPfxLen = common
+			hg.Scalars[i].lower = convertDatumToScalar(&lower, common)
+			hg.Scalars[i].upper = convertDatumToScalar(&upper, common)
 		}
 	}
 }
diff --git a/pkg/util/execdetails/execdetails.go b/pkg/util/execdetails/execdetails.go
index 36a8574fbe..a813fbd6aa 100644
--- a/pkg/util/execdetails/execdetails.go
+++ b/pkg/util/execdetails/execdetails.go
@@ -781,7 +781,7 @@ func (crs *CopRuntimeStats) RecordOneCopTask(address string, summary *tipb.Execu
 				minLocalStreamMs:          summary.GetTiflashScanContext().GetMinLocalStreamMs(),
 				maxLocalStreamMs:          summary.GetTiflashScanContext().GetMaxLocalStreamMs(),
 				minRemoteStreamMs:         summary.GetTiflashScanContext().GetMinRemoteStreamMs(),
-				maxRemoteStreamMs:         summary.GetTiflashScanContext().GetMaxLocalStreamMs(),
+				maxRemoteStreamMs:         summary.GetTiflashScanContext().GetMaxRemoteStreamMs(),
 				regionsOfInstance:         make(map[string]uint64),
 			}}, threads: int32(summary.GetConcurrency()),
 		totalTasks: 1,


Test patch:
diff --git a/dumpling/tests/run.sh b/dumpling/tests/run.sh
index 8ccb1bbbbd..e52e72b2bb 100755
--- a/dumpling/tests/run.sh
+++ b/dumpling/tests/run.sh
@@ -38,11 +38,11 @@ function run_case_by_fullpath() {
     # run in verbose mode?
     echo "Verbose mode = $VERBOSE"
     if [ "$VERBOSE" = "true" ]; then
-      PATH="tests/_utils:$PATH" \
+      PATH="dumpling/tests/_utils:$PATH" \
         PS4='+(${BASH_SOURCE}:${LINENO}): ${FUNCNAME[0]:+${FUNCNAME[0]}(): }' \
         bash -x "$script"
     else
-      PATH="tests/_utils:$PATH" \
+      PATH="dumpling/tests/_utils:$PATH" \
         bash +x "$script"
     fi
     echo ">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> TEST: $script Passed Cleaning up test output dir: $DUMPLING_OUTPUT_DIR"
diff --git a/pkg/planner/core/casetest/enforcempp/BUILD.bazel b/pkg/planner/core/casetest/enforcempp/BUILD.bazel
index a2cd098e8e..ead5839cab 100644
--- a/pkg/planner/core/casetest/enforcempp/BUILD.bazel
+++ b/pkg/planner/core/casetest/enforcempp/BUILD.bazel
@@ -13,7 +13,7 @@ go_test(
     deps = [
         "//pkg/domain",
         "//pkg/parser/model",
-        "//pkg/planner/core/internal",
+        "//pkg/planner/util/coretestsdk",
         "//pkg/sessionctx/stmtctx",
         "//pkg/testkit",
         "//pkg/testkit/external",
diff --git a/pkg/planner/core/casetest/enforcempp/enforce_mpp_test.go b/pkg/planner/core/casetest/enforcempp/enforce_mpp_test.go
index 842b941c5b..075cb51c45 100644
--- a/pkg/planner/core/casetest/enforcempp/enforce_mpp_test.go
+++ b/pkg/planner/core/casetest/enforcempp/enforce_mpp_test.go
@@ -21,7 +21,7 @@ import (
 
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/parser/model"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
+	"github.com/pingcap/tidb/pkg/planner/util/coretestsdk"
 	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/external"
@@ -532,7 +532,7 @@ func TestMPPSingleDistinct3Stage(t *testing.T) {
 //
 //	since it doesn't change the schema out (index ref is still the right), so by now it's fine. SEE case: EXPLAIN select count(distinct a), count(distinct b), sum(c) from t.
 func TestMPPMultiDistinct3Stage(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 
 	// test table
@@ -590,7 +590,7 @@ func TestMPPMultiDistinct3Stage(t *testing.T) {
 
 // Test null-aware semi join push down for MPP mode
 func TestMPPNullAwareSemiJoinPushDown(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 
 	// test table
@@ -638,7 +638,7 @@ func TestMPPNullAwareSemiJoinPushDown(t *testing.T) {
 }
 
 func TestMPPSharedCTEScan(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 
 	// test table
@@ -686,7 +686,7 @@ func TestMPPSharedCTEScan(t *testing.T) {
 }
 
 func TestRollupMPP(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 
 	tk.MustExec("use test")
diff --git a/pkg/planner/core/casetest/hint/BUILD.bazel b/pkg/planner/core/casetest/hint/BUILD.bazel
index cf2c5aff0e..2dbd8a78c4 100644
--- a/pkg/planner/core/casetest/hint/BUILD.bazel
+++ b/pkg/planner/core/casetest/hint/BUILD.bazel
@@ -14,7 +14,7 @@ go_test(
         "//pkg/config",
         "//pkg/domain",
         "//pkg/parser/model",
-        "//pkg/planner/core/internal",
+        "//pkg/planner/util/coretestsdk",
         "//pkg/sessionctx/variable",
         "//pkg/testkit",
         "//pkg/testkit/testdata",
diff --git a/pkg/planner/core/casetest/hint/hint_test.go b/pkg/planner/core/casetest/hint/hint_test.go
index 6a757ec52c..64ab25855b 100644
--- a/pkg/planner/core/casetest/hint/hint_test.go
+++ b/pkg/planner/core/casetest/hint/hint_test.go
@@ -19,7 +19,7 @@ import (
 
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/parser/model"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
+	"github.com/pingcap/tidb/pkg/planner/util/coretestsdk"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/testdata"
@@ -74,7 +74,7 @@ func TestReadFromStorageHint(t *testing.T) {
 }
 
 func TestAllViewHintType(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 
 	tk.MustExec("use test")
@@ -140,7 +140,7 @@ func TestAllViewHintType(t *testing.T) {
 }
 
 func TestJoinHintCompatibility(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 
 	tk.MustExec("use test")
diff --git a/pkg/planner/core/casetest/partition/BUILD.bazel b/pkg/planner/core/casetest/partition/BUILD.bazel
index 0472ac87bd..70e2eb3e0b 100644
--- a/pkg/planner/core/casetest/partition/BUILD.bazel
+++ b/pkg/planner/core/casetest/partition/BUILD.bazel
@@ -13,7 +13,7 @@ go_test(
     shard_count = 7,
     deps = [
         "//pkg/config",
-        "//pkg/planner/core/internal",
+        "//pkg/planner/util/coretestsdk",
         "//pkg/sessionctx/variable",
         "//pkg/testkit",
         "//pkg/testkit/testdata",
diff --git a/pkg/planner/core/casetest/partition/partition_pruner_test.go b/pkg/planner/core/casetest/partition/partition_pruner_test.go
index aafdeebe11..23057c98fc 100644
--- a/pkg/planner/core/casetest/partition/partition_pruner_test.go
+++ b/pkg/planner/core/casetest/partition/partition_pruner_test.go
@@ -22,7 +22,7 @@ import (
 	"testing"
 
 	"github.com/pingcap/failpoint"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
+	"github.com/pingcap/tidb/pkg/planner/util/coretestsdk"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/testdata"
@@ -86,12 +86,12 @@ func getPartitionInfoFromPlan(plan []string) string {
 	infos := make([]testTablePartitionInfo, 0, 2)
 	info := testTablePartitionInfo{}
 	for _, row := range plan {
-		partitions := internal.GetFieldValue("partition:", row)
+		partitions := coretestsdk.GetFieldValue("partition:", row)
 		if partitions != "" {
 			info.Partitions = partitions
 			continue
 		}
-		tbl := internal.GetFieldValue("table:", row)
+		tbl := coretestsdk.GetFieldValue("table:", row)
 		if tbl != "" {
 			info.Table = tbl
 			infos = append(infos, info)
diff --git a/pkg/planner/core/casetest/physicalplantest/BUILD.bazel b/pkg/planner/core/casetest/physicalplantest/BUILD.bazel
index 145f74ebb7..e9b4c85d05 100644
--- a/pkg/planner/core/casetest/physicalplantest/BUILD.bazel
+++ b/pkg/planner/core/casetest/physicalplantest/BUILD.bazel
@@ -21,7 +21,7 @@ go_test(
         "//pkg/parser/model",
         "//pkg/planner",
         "//pkg/planner/core",
-        "//pkg/planner/core/internal",
+        "//pkg/planner/util/coretestsdk",
         "//pkg/sessionctx/stmtctx",
         "//pkg/testkit",
         "//pkg/testkit/external",
diff --git a/pkg/planner/core/casetest/physicalplantest/physical_plan_test.go b/pkg/planner/core/casetest/physicalplantest/physical_plan_test.go
index eb2743978a..b9160ba7b0 100644
--- a/pkg/planner/core/casetest/physicalplantest/physical_plan_test.go
+++ b/pkg/planner/core/casetest/physicalplantest/physical_plan_test.go
@@ -30,7 +30,7 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/planner"
 	"github.com/pingcap/tidb/pkg/planner/core"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
+	"github.com/pingcap/tidb/pkg/planner/util/coretestsdk"
 	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/external"
@@ -211,7 +211,7 @@ func TestUnmatchedTableInHint(t *testing.T) {
 }
 
 func TestIssue37520(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("set tidb_cost_model_version=2")
@@ -241,7 +241,7 @@ func TestIssue37520(t *testing.T) {
 }
 
 func TestMPPHints(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("set tidb_cost_model_version=2")
@@ -283,7 +283,7 @@ func TestMPPHints(t *testing.T) {
 }
 
 func TestMPPHintsScope(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("set tidb_cost_model_version=2")
@@ -340,7 +340,7 @@ func TestMPPBCJModel(t *testing.T) {
 			Probe: sizeof(Data) * 2 / 3
 			exchange size: Build + Probe = 4/3 * sizeof(Data)
 	*/
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(3))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(3))
 	{
 		cnt, err := store.GetMPPClient().GetMPPStoreCount()
 		require.Equal(t, cnt, 3)
@@ -382,7 +382,7 @@ func TestMPPBCJModel(t *testing.T) {
 }
 
 func TestMPPPreferBCJ(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(3))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(3))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("drop table if exists t1")
@@ -449,7 +449,7 @@ func TestMPPBCJModelOneTiFlash(t *testing.T) {
 			Probe: sizeof(Data) * 0 / 1
 			exchange size: Build + Probe = 0 * sizeof(Data)
 	*/
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(1))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(1))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("create table t (a int, b int, c int, index idx_a(a), index idx_b(b))")
@@ -500,7 +500,7 @@ func TestMPPBCJModelOneTiFlash(t *testing.T) {
 }
 
 func TestMPPRightSemiJoin(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(3))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(3))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("drop table if exists t1")
@@ -556,7 +556,7 @@ func TestMPPRightSemiJoin(t *testing.T) {
 }
 
 func TestMPPRightOuterJoin(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(3))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(3))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("drop table if exists t1")
diff --git a/pkg/planner/core/casetest/rule/BUILD.bazel b/pkg/planner/core/casetest/rule/BUILD.bazel
index eeef80bad4..df090e405d 100644
--- a/pkg/planner/core/casetest/rule/BUILD.bazel
+++ b/pkg/planner/core/casetest/rule/BUILD.bazel
@@ -19,7 +19,8 @@ go_test(
         "//pkg/parser/ast",
         "//pkg/parser/model",
         "//pkg/parser/mysql",
-        "//pkg/planner/core/internal",
+        "//pkg/planner/util/coretestsdk",
+        "//pkg/planner/util/coreusage",
         "//pkg/testkit",
         "//pkg/testkit/testdata",
         "//pkg/testkit/testmain",
diff --git a/pkg/planner/core/casetest/rule/rule_derive_topn_from_window_test.go b/pkg/planner/core/casetest/rule/rule_derive_topn_from_window_test.go
index b00a915790..462993df6d 100644
--- a/pkg/planner/core/casetest/rule/rule_derive_topn_from_window_test.go
+++ b/pkg/planner/core/casetest/rule/rule_derive_topn_from_window_test.go
@@ -18,7 +18,7 @@ import (
 	"testing"
 
 	"github.com/pingcap/tidb/pkg/domain"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
+	"github.com/pingcap/tidb/pkg/planner/util/coretestsdk"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/testdata"
 )
@@ -35,7 +35,7 @@ func TestPushDerivedTopnFlash(t *testing.T) {
 	tk.MustExec("use test")
 	tk.MustExec("drop table if exists t")
 	tk.MustExec("create table t(a int, b int, primary key(b,a))")
-	internal.SetTiFlashReplica(t, dom, "test", "t")
+	coretestsdk.SetTiFlashReplica(t, dom, "test", "t")
 	tk.MustExec("set tidb_enforce_mpp=1")
 	tk.MustExec("set @@session.tidb_allow_mpp=ON;")
 	var input Input
diff --git a/pkg/planner/core/casetest/rule/rule_inject_extra_projection_test.go b/pkg/planner/core/casetest/rule/rule_inject_extra_projection_test.go
index 660a63f2ef..dcc0002ce0 100644
--- a/pkg/planner/core/casetest/rule/rule_inject_extra_projection_test.go
+++ b/pkg/planner/core/casetest/rule/rule_inject_extra_projection_test.go
@@ -21,7 +21,7 @@ import (
 	"github.com/pingcap/tidb/pkg/expression/aggregation"
 	"github.com/pingcap/tidb/pkg/parser/ast"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
+	"github.com/pingcap/tidb/pkg/planner/util/coreusage"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util/mock"
 	"github.com/stretchr/testify/require"
@@ -56,7 +56,7 @@ func TestWrapCastForAggFuncs(t *testing.T) {
 		orgAggFuncs = append(orgAggFuncs, agg.Clone())
 	}
 
-	internal.WrapCastForAggFuncs(mock.NewContext(), aggFuncs)
+	coreusage.WrapCastForAggFuncs(mock.NewContext(), aggFuncs)
 	for i := range aggFuncs {
 		if aggFuncs[i].Mode != aggregation.FinalMode && aggFuncs[i].Mode != aggregation.Partial2Mode {
 			require.Equal(t, aggFuncs[i].Args[0].GetType().GetType(), aggFuncs[i].RetTp.GetType())
diff --git a/pkg/planner/core/casetest/windows/BUILD.bazel b/pkg/planner/core/casetest/windows/BUILD.bazel
index ae89816706..8e8fbb8c25 100644
--- a/pkg/planner/core/casetest/windows/BUILD.bazel
+++ b/pkg/planner/core/casetest/windows/BUILD.bazel
@@ -13,7 +13,7 @@ go_test(
     shard_count = 5,
     deps = [
         "//pkg/domain",
-        "//pkg/planner/core/internal",
+        "//pkg/planner/util/coretestsdk",
         "//pkg/testkit",
         "//pkg/testkit/testdata",
         "//pkg/testkit/testmain",
diff --git a/pkg/planner/core/casetest/windows/window_push_down_test.go b/pkg/planner/core/casetest/windows/window_push_down_test.go
index 633318cf06..f680437024 100644
--- a/pkg/planner/core/casetest/windows/window_push_down_test.go
+++ b/pkg/planner/core/casetest/windows/window_push_down_test.go
@@ -20,7 +20,7 @@ import (
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/domain"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
+	"github.com/pingcap/tidb/pkg/planner/util/coretestsdk"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/testdata"
 	"github.com/stretchr/testify/require"
@@ -63,7 +63,7 @@ func TestWindowFunctionDescCanPushDown(t *testing.T) {
 	tk.MustExec("use test")
 	tk.MustExec("drop table if exists employee")
 	tk.MustExec("create table employee (empid int, deptid int, salary decimal(10,2))")
-	internal.SetTiFlashReplica(t, dom, "test", "employee")
+	coretestsdk.SetTiFlashReplica(t, dom, "test", "employee")
 
 	var input Input
 	var output Output
@@ -80,7 +80,7 @@ func TestWindowPushDownPlans(t *testing.T) {
 	tk.MustExec("use test")
 	tk.MustExec("drop table if exists employee")
 	tk.MustExec("create table employee (empid int, deptid int, salary decimal(10,2))")
-	internal.SetTiFlashReplica(t, dom, "test", "employee")
+	coretestsdk.SetTiFlashReplica(t, dom, "test", "employee")
 
 	var input Input
 	var output Output
@@ -98,7 +98,7 @@ func TestWindowPlanWithOtherOperators(t *testing.T) {
 	tk.MustExec("set tidb_cost_model_version=2")
 	tk.MustExec("drop table if exists employee")
 	tk.MustExec("create table employee (empid int, deptid int, salary decimal(10,2))")
-	internal.SetTiFlashReplica(t, dom, "test", "employee")
+	coretestsdk.SetTiFlashReplica(t, dom, "test", "employee")
 
 	var input Input
 	var output Output
@@ -116,8 +116,8 @@ func TestIssue34765(t *testing.T) {
 	tk.MustExec("create table t1(c1 varchar(32), c2 datetime, c3 bigint, c4 varchar(64));")
 	tk.MustExec("create table t2(b2 varchar(64));")
 	tk.MustExec("set tidb_enforce_mpp=1;")
-	internal.SetTiFlashReplica(t, dom, "test", "t1")
-	internal.SetTiFlashReplica(t, dom, "test", "t2")
+	coretestsdk.SetTiFlashReplica(t, dom, "test", "t1")
+	coretestsdk.SetTiFlashReplica(t, dom, "test", "t2")
 
 	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/planner/core/CheckMPPWindowSchemaLength", "return"))
 	defer func() {
diff --git a/pkg/planner/core/internal/base/plan.go b/pkg/planner/core/operator/base/plan.go
similarity index 100%
rename from pkg/planner/core/internal/base/plan.go
rename to pkg/planner/core/operator/base/plan.go
diff --git a/pkg/planner/core/physical_plan_test.go b/pkg/planner/core/physical_plan_test.go
index 928241461e..ed75e9ab74 100644
--- a/pkg/planner/core/physical_plan_test.go
+++ b/pkg/planner/core/physical_plan_test.go
@@ -31,9 +31,9 @@ import (
 	"github.com/pingcap/tidb/pkg/planner"
 	"github.com/pingcap/tidb/pkg/planner/core"
 	"github.com/pingcap/tidb/pkg/planner/core/base"
-	"github.com/pingcap/tidb/pkg/planner/core/internal"
 	"github.com/pingcap/tidb/pkg/planner/property"
 	"github.com/pingcap/tidb/pkg/planner/util"
+	"github.com/pingcap/tidb/pkg/planner/util/coretestsdk"
 	"github.com/pingcap/tidb/pkg/session"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/external"
@@ -222,7 +222,7 @@ func TestIndexLookupCartesianJoin(t *testing.T) {
 }
 
 func TestMPPHintsWithBinding(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("set tidb_cost_model_version=2")
@@ -271,7 +271,7 @@ func TestMPPHintsWithBinding(t *testing.T) {
 }
 
 func TestJoinHintCompatibilityWithBinding(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("set tidb_cost_model_version=2")
@@ -442,7 +442,7 @@ func TestPhysicalPlanMemoryTrace(t *testing.T) {
 }
 
 func TestPhysicalTableScanExtractCorrelatedCols(t *testing.T) {
-	store := testkit.CreateMockStore(t, internal.WithMockTiFlash(2))
+	store := testkit.CreateMockStore(t, coretestsdk.WithMockTiFlash(2))
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
 	tk.MustExec("create table t1 (id int, client_type tinyint, client_no char(18), taxpayer_no varchar(50), status tinyint, update_time datetime)")
diff --git a/pkg/planner/core/internal/testkit.go b/pkg/planner/util/coretestsdk/testkit.go
similarity index 99%
rename from pkg/planner/core/internal/testkit.go
rename to pkg/planner/util/coretestsdk/testkit.go
index 54c0ca3350..ac06e52eb4 100644
--- a/pkg/planner/core/internal/testkit.go
+++ b/pkg/planner/util/coretestsdk/testkit.go
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package internal
+package coretestsdk
 
 import (
 	"fmt"
diff --git a/pkg/server/conn_test.go b/pkg/server/conn_test.go
index ee280b768d..20c501294b 100644
--- a/pkg/server/conn_test.go
+++ b/pkg/server/conn_test.go
@@ -30,11 +30,13 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/metapb"
 	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/extension"
+	"github.com/pingcap/tidb/pkg/metrics"
 	"github.com/pingcap/tidb/pkg/parser/auth"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/server/internal"
@@ -54,6 +56,7 @@ import (
 	"github.com/pingcap/tidb/pkg/util/dbterror/exeerrors"
 	"github.com/pingcap/tidb/pkg/util/plancodec"
 	"github.com/pingcap/tidb/pkg/util/sqlkiller"
+	promtestutils "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
 	tikverr "github.com/tikv/client-go/v2/error"
 	"github.com/tikv/client-go/v2/testutils"
@@ -2085,3 +2088,40 @@ func TestCloseConn(t *testing.T) {
 	}
 	wg.Wait()
 }
+
+func TestConnAddMetrics(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	re := require.New(t)
+	tk := testkit.NewTestKit(t, store)
+	cc := &clientConn{
+		alloc:      arena.NewAllocator(1024),
+		chunkAlloc: chunk.NewAllocator(),
+		pkt:        internal.NewPacketIOForTest(bufio.NewWriter(bytes.NewBuffer(nil))),
+	}
+	tk.MustExec("use test")
+	cc.SetCtx(&TiDBContext{Session: tk.Session(), stmts: make(map[int]*TiDBStatement)})
+
+	// default
+	cc.addMetrics(mysql.ComQuery, time.Now(), nil)
+	counter := metrics.QueryTotalCounter
+	v := promtestutils.ToFloat64(counter.WithLabelValues("Query", "OK", "default"))
+	require.Equal(t, 1.0, v)
+
+	// rg1
+	cc.getCtx().GetSessionVars().ResourceGroupName = "test_rg1"
+	cc.addMetrics(mysql.ComQuery, time.Now(), nil)
+	re.Equal(promtestutils.ToFloat64(counter.WithLabelValues("Query", "OK", "default")), 1.0)
+	re.Equal(promtestutils.ToFloat64(counter.WithLabelValues("Query", "OK", "test_rg1")), 1.0)
+	/// inc the counter again
+	cc.addMetrics(mysql.ComQuery, time.Now(), nil)
+	re.Equal(promtestutils.ToFloat64(counter.WithLabelValues("Query", "OK", "test_rg1")), 2.0)
+
+	// rg2
+	cc.getCtx().GetSessionVars().ResourceGroupName = "test_rg2"
+	// error
+	cc.addMetrics(mysql.ComQuery, time.Now(), errors.New("unknown error"))
+	re.Equal(promtestutils.ToFloat64(counter.WithLabelValues("Query", "Error", "test_rg2")), 1.0)
+	// ok
+	cc.addMetrics(mysql.ComStmtExecute, time.Now(), nil)
+	re.Equal(promtestutils.ToFloat64(counter.WithLabelValues("StmtExecute", "OK", "test_rg2")), 1.0)
+}
diff --git a/pkg/server/handler/tests/BUILD.bazel b/pkg/server/handler/tests/BUILD.bazel
index 415990a07d..687e94eef2 100644
--- a/pkg/server/handler/tests/BUILD.bazel
+++ b/pkg/server/handler/tests/BUILD.bazel
@@ -9,13 +9,14 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
-    shard_count = 38,
+    shard_count = 39,
     deps = [
         "//pkg/config",
         "//pkg/ddl",
         "//pkg/ddl/util",
         "//pkg/domain",
         "//pkg/domain/infosync",
+        "//pkg/executor/mppcoordmanager",
         "//pkg/infoschema",
         "//pkg/kv",
         "//pkg/meta",
diff --git a/pkg/server/handler/tests/http_handler_test.go b/pkg/server/handler/tests/http_handler_test.go
index b2726b7f5a..ea40ec0968 100644
--- a/pkg/server/handler/tests/http_handler_test.go
+++ b/pkg/server/handler/tests/http_handler_test.go
@@ -40,6 +40,7 @@ import (
 	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/domain/infosync"
+	"github.com/pingcap/tidb/pkg/executor/mppcoordmanager"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/meta"
@@ -1502,3 +1503,13 @@ func testUpgradeShow(t *testing.T, ts *basicHTTPHandlerTestSuite) {
 	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/pkg/domain/infosync/mockGetAllServerInfo", makeFailpointRes(mockedAllServerInfos)))
 	checkUpgradeShow(3, 100, 0)
 }
+
+func TestIssue52608(t *testing.T) {
+	ts := createBasicHTTPHandlerTestSuite()
+
+	ts.startServer(t)
+	defer ts.stopServer(t)
+	on, addr := mppcoordmanager.InstanceMPPCoordinatorManager.GetServerAddr()
+	require.Equal(t, on, true)
+	require.Equal(t, addr[:10], "127.0.0.1:")
+}
diff --git a/tests/realtikvtest/addindextest3/operator_test.go b/tests/realtikvtest/addindextest3/operator_test.go
index 067fdc672d..a1cf118e65 100644
--- a/tests/realtikvtest/addindextest3/operator_test.go
+++ b/tests/realtikvtest/addindextest3/operator_test.go
@@ -135,13 +135,14 @@ func TestBackfillOperators(t *testing.T) {
 		srcChkPool := make(chan *chunk.Chunk, regionCnt*2)
 		pTbl := tbl.(table.PhysicalTable)
 		index := tables.NewIndex(pTbl.GetPhysicalID(), tbl.Meta(), idxInfo)
+		mockBackendCtx := &ingest.MockBackendCtx{}
 		mockEngine := ingest.NewMockEngineInfo(nil)
 		mockEngine.SetHook(onWrite)
 
 		src := newTestSource(chunkResults...)
 		reorgMeta := ddl.NewDDLReorgMeta(tk.Session())
 		ingestOp := ddl.NewIndexIngestOperator(
-			opCtx, copCtx, sessPool, pTbl, []table.Index{index}, []ingest.Engine{mockEngine}, srcChkPool, 3, reorgMeta)
+			opCtx, copCtx, mockBackendCtx, sessPool, pTbl, []table.Index{index}, []ingest.Engine{mockEngine}, srcChkPool, 3, reorgMeta)
 		sink := newTestSink[ddl.IndexWriteResult]()
 
 		operator.Compose[ddl.IndexRecordChunk](src, ingestOp)