Repository: pingcap/tidb (go)

Issue:
Child executors can not be closed when panic happens in the `Close` function
## Bug Report

Please answer these questions before submitting your issue. Thanks!

### 1. Minimal reproduce step (Required)

- Execute a sql with hash join v1 executor with disk spill feature turned off
- Let the memory usage of this sql exceeds the memory limit

<!-- a step by step guide for reproducing the bug. -->

### 2. What did you expect to see? (Required)

No goroutine is leaked

### 3. What did you see instead (Required)

Some goroutines may be leaked

### 4. What is your TiDB version? (Required)

v8.5.0

<!-- Paste the output of SELECT tidb_version() -->



Pull request description:
executor: replace `Call` with `CallWithRecover` in the close of hash join v1 (#61868)
This is an automated cherry-pick of #61868

<!--

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: close #60926

Problem Summary:

### What changed and how does it work?

We introduce a new function `CallWithRecover` to recover the panic in the close of hash join v1

### Check List

Tests <!-- At least one of them must be included. -->

- [x] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test
  > - [ ] 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
Fix the issue that the panic in the close of hash join v1 is not recovered
```


Patch:
diff --git a/br/pkg/storage/gcs.go b/br/pkg/storage/gcs.go
index 6e5499f586..be12bc1390 100644
--- a/br/pkg/storage/gcs.go
+++ b/br/pkg/storage/gcs.go
@@ -8,6 +8,7 @@ import (
 	goerrors "errors"
 	"fmt"
 	"io"
+	"net/url"
 	"os"
 	"path"
 	"strings"
@@ -509,6 +510,14 @@ func shouldRetry(err error) bool {
 		}
 	}
 
+	// workaround for https://github.com/googleapis/google-cloud-go/issues/7090
+	// seems it's a bug of golang net/http: https://github.com/golang/go/issues/53472
+	if e := (&url.Error{}); goerrors.As(err, &e) {
+		if goerrors.Is(e.Err, io.EOF) {
+			return true
+		}
+	}
+
 	errMsg := err.Error()
 	// workaround for strange unknown errors
 	retryableErrMsg := []string{
diff --git a/br/pkg/storage/ks3.go b/br/pkg/storage/ks3.go
index 919da5e3aa..0b704f8f18 100644
--- a/br/pkg/storage/ks3.go
+++ b/br/pkg/storage/ks3.go
@@ -508,8 +508,14 @@ func (rs *KS3Storage) open(
 	}
 
 	if startOffset != r.Start || (endOffset != 0 && endOffset != r.End+1) {
-		return nil, r, errors.Annotatef(berrors.ErrStorageUnknown, "open file '%s' failed, expected range: %s, got: %v",
-			path, *rangeOffset, result.ContentRange)
+		rangeStr := "<empty>"
+		if result.ContentRange != nil {
+			rangeStr = *result.ContentRange
+		}
+		return nil, r, errors.Annotatef(
+			berrors.ErrStorageUnknown,
+			"open file '%s' failed, expected range: %s, got: %s",
+			path, *rangeOffset, rangeStr)
 	}
 
 	return result.Body, r, nil
diff --git a/br/pkg/storage/s3.go b/br/pkg/storage/s3.go
index 3987512b2a..4e95bbf7bf 100644
--- a/br/pkg/storage/s3.go
+++ b/br/pkg/storage/s3.go
@@ -853,8 +853,13 @@ func (rs *S3Storage) open(
 	}
 
 	if startOffset != r.Start || (endOffset != 0 && endOffset != r.End+1) {
-		return nil, r, errors.Annotatef(berrors.ErrStorageUnknown, "open file '%s' failed, expected range: %s, got: %v",
-			path, *rangeOffset, result.ContentRange)
+		rangeStr := "<empty>"
+		if result.ContentRange != nil {
+			rangeStr = *result.ContentRange
+		}
+		return nil, r, errors.Annotatef(berrors.ErrStorageUnknown,
+			"open file '%s' failed, expected range: %s, got: %s",
+			path, *rangeOffset, rangeStr)
 	}
 
 	return result.Body, r, nil
diff --git a/pkg/ddl/backfilling_dist_executor.go b/pkg/ddl/backfilling_dist_executor.go
index edceea3486..51796f4d98 100644
--- a/pkg/ddl/backfilling_dist_executor.go
+++ b/pkg/ddl/backfilling_dist_executor.go
@@ -141,7 +141,7 @@ func (s *backfillDistExecutor) getBackendCtx(ctx context.Context) (ingest.Backen
 	ddlObj := s.d
 	discovery := ddlObj.store.(tikv.Storage).GetRegionCache().PDClient().GetServiceDiscovery()
 
-	return ingest.LitBackCtxMgr.Register(ctx, job.ID, unique, ddlObj.etcdCli, discovery, job.ReorgMeta.ResourceGroupName)
+	return ingest.LitBackCtxMgr.Register(ctx, job.ID, unique, ddlObj.etcdCli, discovery, job.ReorgMeta.ResourceGroupName, job.RealStartTS)
 }
 
 func decodeIndexUniqueness(job *model.Job) (bool, error) {
diff --git a/pkg/ddl/index.go b/pkg/ddl/index.go
index 5de8eb28dc..5a73aa952a 100644
--- a/pkg/ddl/index.go
+++ b/pkg/ddl/index.go
@@ -942,7 +942,8 @@ func runIngestReorgJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job,
 			metrics.UnregisterLightningCommonMetricsForDDL(job.ID, m)
 		}
 	}()
-	bc, err = ingest.LitBackCtxMgr.Register(ctx, job.ID, allIndexInfos[0].Unique, nil, discovery, job.ReorgMeta.ResourceGroupName)
+	bc, err = ingest.LitBackCtxMgr.Register(ctx, job.ID, allIndexInfos[0].Unique,
+		nil, discovery, job.ReorgMeta.ResourceGroupName, job.RealStartTS)
 	if err != nil {
 		ver, err = convertAddIdxJob2RollbackJob(d, t, job, tbl.Meta(), allIndexInfos, err)
 		return false, ver, errors.Trace(err)
@@ -2018,7 +2019,8 @@ func checkDuplicateForUniqueIndex(ctx context.Context, t table.Table, reorgInfo
 		if indexInfo.Unique {
 			ctx := tidblogutil.WithCategory(ctx, "ddl-ingest")
 			if bc == nil {
-				bc, err = ingest.LitBackCtxMgr.Register(ctx, reorgInfo.ID, indexInfo.Unique, nil, discovery, reorgInfo.ReorgMeta.ResourceGroupName)
+				bc, err = ingest.LitBackCtxMgr.Register(ctx, reorgInfo.ID, indexInfo.Unique, nil,
+					discovery, reorgInfo.ReorgMeta.ResourceGroupName, reorgInfo.RealStartTS)
 				if err != nil {
 					return err
 				}
diff --git a/pkg/ddl/ingest/BUILD.bazel b/pkg/ddl/ingest/BUILD.bazel
index 68b9b4de84..39ebf97d5f 100644
--- a/pkg/ddl/ingest/BUILD.bazel
+++ b/pkg/ddl/ingest/BUILD.bazel
@@ -36,6 +36,7 @@ go_library(
         "//pkg/lightning/log",
         "//pkg/meta",
         "//pkg/metrics",
+        "//pkg/owner",
         "//pkg/parser/mysql",
         "//pkg/parser/terror",
         "//pkg/sessionctx",
@@ -54,7 +55,6 @@ go_library(
         "@com_github_tikv_client_go_v2//util",
         "@com_github_tikv_pd_client//:client",
         "@io_etcd_go_etcd_client_v3//:client",
-        "@io_etcd_go_etcd_client_v3//concurrency",
         "@org_golang_x_exp//maps",
         "@org_uber_go_atomic//:atomic",
         "@org_uber_go_zap//:zap",
diff --git a/pkg/ddl/ingest/backend.go b/pkg/ddl/ingest/backend.go
index 48f9a57c10..ec074a211d 100644
--- a/pkg/ddl/ingest/backend.go
+++ b/pkg/ddl/ingest/backend.go
@@ -29,6 +29,7 @@ import (
 	lightning "github.com/pingcap/tidb/pkg/lightning/config"
 	"github.com/pingcap/tidb/pkg/lightning/errormanager"
 	"github.com/pingcap/tidb/pkg/lightning/log"
+	"github.com/pingcap/tidb/pkg/owner"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	"github.com/pingcap/tidb/pkg/table"
@@ -36,7 +37,6 @@ import (
 	"github.com/pingcap/tidb/pkg/util/generic"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	clientv3 "go.etcd.io/etcd/client/v3"
-	"go.etcd.io/etcd/client/v3/concurrency"
 	atomicutil "go.uber.org/atomic"
 	"go.uber.org/zap"
 )
@@ -93,6 +93,7 @@ type litBackendCtx struct {
 	updateInterval  time.Duration
 	checkpointMgr   *CheckpointManager
 	etcdClient      *clientv3.Client
+	initTS          uint64
 }
 
 func (bc *litBackendCtx) handleErrorAfterCollectRemoteDuplicateRows(err error, indexID int64, tbl table.Table, hasDupe bool) error {
@@ -128,9 +129,10 @@ func (bc *litBackendCtx) CollectRemoteDuplicateRows(indexID int64, tbl table.Tab
 	// backend must be a local backend.
 	dupeController := bc.backend.GetDupeController(bc.cfg.TikvImporter.RangeConcurrency*2, errorMgr)
 	hasDupe, err := dupeController.CollectRemoteDuplicateRows(bc.ctx, tbl, tbl.Meta().Name.L, &encode.SessionOptions{
-		SQLMode: mysql.ModeStrictAllTables,
-		SysVars: bc.sysVars,
-		IndexID: indexID,
+		SQLMode:     mysql.ModeStrictAllTables,
+		SysVars:     bc.sysVars,
+		IndexID:     indexID,
+		MinCommitTS: bc.initTS,
 	}, lightning.ErrorOnDup)
 	return bc.handleErrorAfterCollectRemoteDuplicateRows(err, indexID, tbl, hasDupe)
 }
@@ -160,24 +162,16 @@ func (bc *litBackendCtx) FinishImport(indexID int64, unique bool, tbl table.Tabl
 		//nolint:forcetypeassert
 		dupeController := bc.backend.GetDupeController(bc.cfg.TikvImporter.RangeConcurrency*2, errorMgr)
 		hasDupe, err := dupeController.CollectRemoteDuplicateRows(bc.ctx, tbl, tbl.Meta().Name.L, &encode.SessionOptions{
-			SQLMode: mysql.ModeStrictAllTables,
-			SysVars: bc.sysVars,
-			IndexID: ei.indexID,
+			SQLMode:     mysql.ModeStrictAllTables,
+			SysVars:     bc.sysVars,
+			IndexID:     ei.indexID,
+			MinCommitTS: bc.initTS,
 		}, lightning.ErrorOnDup)
 		return bc.handleErrorAfterCollectRemoteDuplicateRows(err, indexID, tbl, hasDupe)
 	}
 	return nil
 }
 
-func acquireLock(ctx context.Context, se *concurrency.Session, key string) (*concurrency.Mutex, error) {
-	mu := concurrency.NewMutex(se, key)
-	err := mu.Lock(ctx)
-	if err != nil {
-		return nil, err
-	}
-	return mu, nil
-}
-
 // Flush checks the disk quota and imports the current key-values in engine to the storage.
 func (bc *litBackendCtx) Flush(indexID int64, mode FlushMode) (flushed, imported bool, err error) {
 	ei, exist := bc.Load(indexID)
@@ -207,27 +201,15 @@ func (bc *litBackendCtx) Flush(indexID int64, mode FlushMode) (flushed, imported
 		return true, false, nil
 	}
 
-	// Use distributed lock if run in distributed mode).
 	if bc.etcdClient != nil {
-		distLockKey := fmt.Sprintf("/tidb/distributeLock/%d/%d", bc.jobID, indexID)
-		se, _ := concurrency.NewSession(bc.etcdClient)
-		mu, err := acquireLock(bc.ctx, se, distLockKey)
+		key := fmt.Sprintf("/tidb/distributeLock/%d", bc.jobID)
+		release, err := owner.AcquireDistributedLock(bc.ctx, bc.etcdClient, key, 10)
 		if err != nil {
-			return true, false, errors.Trace(err)
+			return true, false, err
+		}
+		if release != nil {
+			defer release()
 		}
-		logutil.Logger(bc.ctx).Info("acquire distributed flush lock success", zap.Int64("jobID", bc.jobID))
-		defer func() {
-			err = mu.Unlock(bc.ctx)
-			if err != nil {
-				logutil.Logger(bc.ctx).Warn("release distributed flush lock error", zap.Error(err), zap.Int64("jobID", bc.jobID))
-			} else {
-				logutil.Logger(bc.ctx).Info("release distributed flush lock success", zap.Int64("jobID", bc.jobID))
-			}
-			err = se.Close()
-			if err != nil {
-				logutil.Logger(bc.ctx).Warn("close session error", zap.Error(err))
-			}
-		}()
 	}
 	failpoint.Inject("mockDMLExecutionStateBeforeImport", func(_ failpoint.Value) {
 		if MockDMLExecutionStateBeforeImport != nil {
@@ -242,6 +224,8 @@ func (bc *litBackendCtx) Flush(indexID int64, mode FlushMode) (flushed, imported
 	return true, true, nil
 }
 
+const distributedLockLease = 10 // Seconds
+
 func (bc *litBackendCtx) unsafeImportAndReset(ei *engineInfo) error {
 	logutil.Logger(bc.ctx).Info(LitInfoUnsafeImport, zap.Int64("index ID", ei.indexID),
 		zap.String("usage info", bc.diskRoot.UsageInfo()))
diff --git a/pkg/ddl/ingest/backend_mgr.go b/pkg/ddl/ingest/backend_mgr.go
index 5276184a49..bffe475900 100644
--- a/pkg/ddl/ingest/backend_mgr.go
+++ b/pkg/ddl/ingest/backend_mgr.go
@@ -52,6 +52,7 @@ type BackendCtxMgr interface {
 		etcdClient *clientv3.Client,
 		pdSvcDiscovery pd.ServiceDiscovery,
 		resourceGroupName string,
+		initTS uint64,
 	) (BackendCtx, error)
 	Unregister(jobID int64)
 	// Load returns the registered BackendCtx with the given jobID.
@@ -123,6 +124,7 @@ func (m *litBackendCtxMgr) Register(
 	etcdClient *clientv3.Client,
 	pdSvcDiscovery pd.ServiceDiscovery,
 	resourceGroupName string,
+	initTS uint64,
 ) (BackendCtx, error) {
 	bc, exist := m.Load(jobID)
 	if exist {
@@ -153,7 +155,7 @@ func (m *litBackendCtxMgr) Register(
 		return nil, err
 	}
 
-	bcCtx := newBackendContext(ctx, jobID, bd, cfg.lightning, defaultImportantVariables, m.memRoot, m.diskRoot, etcdClient)
+	bcCtx := newBackendContext(ctx, jobID, bd, cfg.lightning, defaultImportantVariables, m.memRoot, m.diskRoot, etcdClient, initTS)
 	m.backends.m[jobID] = bcCtx
 	m.memRoot.Consume(StructSizeBackendCtx)
 	m.backends.mu.Unlock()
@@ -266,6 +268,7 @@ func newBackendContext(
 	memRoot MemRoot,
 	diskRoot DiskRoot,
 	etcdClient *clientv3.Client,
+	initTS uint64,
 ) *litBackendCtx {
 	bCtx := &litBackendCtx{
 		SyncMap:        generic.NewSyncMap[int64, *engineInfo](10),
@@ -279,6 +282,7 @@ func newBackendContext(
 		diskRoot:       diskRoot,
 		updateInterval: checkpointUpdateInterval,
 		etcdClient:     etcdClient,
+		initTS:         initTS,
 	}
 	bCtx.timeOfLastFlush.Store(time.Now())
 	return bCtx
diff --git a/pkg/ddl/ingest/mock.go b/pkg/ddl/ingest/mock.go
index 4393861003..8aed39aa5f 100644
--- a/pkg/ddl/ingest/mock.go
+++ b/pkg/ddl/ingest/mock.go
@@ -52,7 +52,8 @@ func (m *MockBackendCtxMgr) CheckMoreTasksAvailable(context.Context) (bool, erro
 }
 
 // Register implements BackendCtxMgr.Register interface.
-func (m *MockBackendCtxMgr) Register(ctx context.Context, jobID int64, unique bool, etcdClient *clientv3.Client, pdSvcDiscovery pd.ServiceDiscovery, resourceGroupName string) (BackendCtx, error) {
+func (m *MockBackendCtxMgr) Register(ctx context.Context, jobID int64, unique bool, etcdClient *clientv3.Client,
+	pdSvcDiscovery pd.ServiceDiscovery, resourceGroupName string, initTS uint64) (BackendCtx, error) {
 	logutil.DDLIngestLogger().Info("mock backend mgr register", zap.Int64("jobID", jobID))
 	if mockCtx, ok := m.runningJobs[jobID]; ok {
 		return mockCtx, nil
diff --git a/pkg/executor/hash_table.go b/pkg/executor/hash_table.go
index 067c1ecb23..427ea2bf39 100644
--- a/pkg/executor/hash_table.go
+++ b/pkg/executor/hash_table.go
@@ -23,6 +23,7 @@ import (
 	"unsafe"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
 	"github.com/pingcap/tidb/pkg/types"
@@ -546,6 +547,8 @@ func (c *hashRowContainer) Len() uint64 {
 }
 
 func (c *hashRowContainer) Close() error {
+	failpoint.Inject("issue60926", nil)
+
 	defer c.memTracker.Detach()
 	c.chkBuf = nil
 	return c.rowContainer.Close()
diff --git a/pkg/executor/join.go b/pkg/executor/join.go
index f9e745b3b8..1f1362ee67 100644
--- a/pkg/executor/join.go
+++ b/pkg/executor/join.go
@@ -42,9 +42,14 @@ import (
 	"github.com/pingcap/tidb/pkg/util/dbterror/exeerrors"
 	"github.com/pingcap/tidb/pkg/util/disk"
 	"github.com/pingcap/tidb/pkg/util/execdetails"
+	"github.com/pingcap/tidb/pkg/util/logutil"
 	"github.com/pingcap/tidb/pkg/util/memory"
+	"go.uber.org/zap"
 )
 
+// IsChildCloseCalledForTest is used for test
+var IsChildCloseCalledForTest atomic.Bool
+
 var (
 	_ exec.Executor = &HashJoinExec{}
 	_ exec.Executor = &NestedLoopApplyExec{}
@@ -175,7 +180,14 @@ func (e *HashJoinExec) Close() error {
 			channel.Clear(e.probeWorkers[i].joinChkResourceCh)
 		}
 		e.probeSideTupleFetcher.probeChkResourceCh = nil
-		terror.Call(e.rowContainer.Close)
+		util.WithRecovery(func() {
+			err := e.rowContainer.Close()
+			if err != nil {
+				logutil.BgLogger().Error("RowContainer encounters error",
+					zap.Error(err),
+					zap.Stack("stack trace"))
+			}
+		}, nil)
 		e.hashJoinCtx.sessCtx.GetSessionVars().MemTracker.UnbindActionFromHardLimit(e.rowContainer.ActionSpill())
 		e.waiterWg.Wait()
 	}
@@ -196,8 +208,9 @@ func (e *HashJoinExec) Close() error {
 	if e.stats != nil {
 		defer e.Ctx().GetSessionVars().StmtCtx.RuntimeStatsColl.RegisterStats(e.ID(), e.stats)
 	}
-	err := e.BaseExecutor.Close()
-	return err
+
+	IsChildCloseCalledForTest.Store(true)
+	return e.BaseExecutor.Close()
 }
 
 // Open implements the Executor Open interface.
diff --git a/pkg/infoschema/tables.go b/pkg/infoschema/tables.go
index 76e6facd17..eb74e0ac8f 100644
--- a/pkg/infoschema/tables.go
+++ b/pkg/infoschema/tables.go
@@ -1407,6 +1407,7 @@ var tableTableTiFlashTablesCols = []columnInfo{
 	{name: "TIDB_TABLE", tp: mysql.TypeVarchar, size: 64},
 	{name: "TABLE_ID", tp: mysql.TypeLonglong, size: 64},
 	{name: "IS_TOMBSTONE", tp: mysql.TypeLonglong, size: 64},
+	{name: "COLUMN_COUNT", tp: mysql.TypeLonglong, size: 21},
 	{name: "SEGMENT_COUNT", tp: mysql.TypeLonglong, size: 64},
 	{name: "TOTAL_ROWS", tp: mysql.TypeLonglong, size: 64},
 	{name: "TOTAL_SIZE", tp: mysql.TypeLonglong, size: 64},
@@ -1415,6 +1416,7 @@ var tableTableTiFlashTablesCols = []columnInfo{
 	{name: "DELTA_RATE_SEGMENTS", tp: mysql.TypeDouble, size: 64},
 	{name: "DELTA_PLACED_RATE", tp: mysql.TypeDouble, size: 64},
 	{name: "DELTA_CACHE_SIZE", tp: mysql.TypeLonglong, size: 64},
+	{name: "DELTA_CACHE_ALLOC_SIZE", tp: mysql.TypeLonglong, size: 21},
 	{name: "DELTA_CACHE_RATE", tp: mysql.TypeDouble, size: 64},
 	{name: "DELTA_CACHE_WASTED_RATE", tp: mysql.TypeDouble, size: 64},
 	{name: "DELTA_INDEX_SIZE", tp: mysql.TypeLonglong, size: 64},
@@ -1480,6 +1482,7 @@ var tableTableTiFlashSegmentsCols = []columnInfo{
 	{name: "DELTA_PERSISTED_COLUMN_FILES", tp: mysql.TypeLonglong, size: 64},
 	{name: "DELTA_PERSISTED_DELETE_RANGES", tp: mysql.TypeLonglong, size: 64},
 	{name: "DELTA_CACHE_SIZE", tp: mysql.TypeLonglong, size: 64},
+	{name: "DELTA_CACHE_ALLOC_SIZE", tp: mysql.TypeLonglong, size: 21},
 	{name: "DELTA_INDEX_SIZE", tp: mysql.TypeLonglong, size: 64},
 	{name: "STABLE_PAGE_ID", tp: mysql.TypeLonglong, size: 64},
 	{name: "STABLE_ROWS", tp: mysql.TypeLonglong, size: 64},
diff --git a/pkg/lightning/backend/encode/encode.go b/pkg/lightning/backend/encode/encode.go
index 9f870bd15c..73fe990032 100644
--- a/pkg/lightning/backend/encode/encode.go
+++ b/pkg/lightning/backend/encode/encode.go
@@ -57,8 +57,10 @@ type SessionOptions struct {
 	SysVars   map[string]string
 	// a seed used for tableKvEncoder's auto random bits value
 	AutoRandomSeed int64
-	// IndexID is used by the DuplicateManager. Only the key range with the specified index ID is scanned.
+	// IndexID is used by the dupeDetector. Only the key range with the specified index ID is scanned.
 	IndexID int64
+	// MinCommitTS is used by dupeDetector. Only records that larger than commit TS are considered.
+	MinCommitTS uint64
 }
 
 // Rows represents a collection of encoded rows.
diff --git a/pkg/lightning/backend/local/duplicate.go b/pkg/lightning/backend/local/duplicate.go
index f099cbc7b3..49c5a3c11c 100644
--- a/pkg/lightning/backend/local/duplicate.go
+++ b/pkg/lightning/backend/local/duplicate.go
@@ -310,6 +310,7 @@ func getDupDetectClient(
 	importClientFactory ImportClientFactory,
 	resourceGroupName string,
 	taskType string,
+	minCommitTS uint64,
 ) (import_sstpb.ImportSST_DuplicateDetectClient, error) {
 	leader := region.Leader
 	if leader == nil {
@@ -330,9 +331,10 @@ func getDupDetectClient(
 		RequestSource: kvutil.BuildRequestSource(true, tidbkv.InternalTxnLightning, taskType),
 	}
 	req := &import_sstpb.DuplicateDetectRequest{
-		Context:  reqCtx,
-		StartKey: keyRange.StartKey,
-		EndKey:   keyRange.EndKey,
+		Context:     reqCtx,
+		StartKey:    keyRange.StartKey,
+		EndKey:      keyRange.EndKey,
+		MinCommitTs: minCommitTS,
 	}
 	cli, err := importClient.DuplicateDetect(ctx, req)
 	if err != nil {
@@ -349,9 +351,10 @@ func NewRemoteDupKVStream(
 	importClientFactory ImportClientFactory,
 	resourceGroupName string,
 	taskType string,
+	minCommitTS uint64,
 ) (*RemoteDupKVStream, error) {
 	subCtx, cancel := context.WithCancel(ctx)
-	cli, err := getDupDetectClient(subCtx, region, keyRange, importClientFactory, resourceGroupName, taskType)
+	cli, err := getDupDetectClient(subCtx, region, keyRange, importClientFactory, resourceGroupName, taskType, minCommitTS)
 	if err != nil {
 		cancel()
 		return nil, errors.Trace(err)
@@ -422,6 +425,7 @@ type DupeDetector struct {
 	indexID           int64
 	resourceGroupName string
 	taskType          string
+	minCommitTS       uint64
 }
 
 // NewDupeDetector creates a new DupeDetector.
@@ -456,6 +460,7 @@ func NewDupeDetector(
 		indexID:           sessOpts.IndexID,
 		resourceGroupName: resourceGroupName,
 		taskType:          taskType,
+		minCommitTS:       sessOpts.MinCommitTS,
 	}, nil
 }
 
@@ -933,7 +938,7 @@ func (m *DupeDetector) processRemoteDupTaskOnce(
 				logutil.Key("dupDetectEndKey", kr.EndKey),
 			)
 			err := func() error {
-				stream, err := NewRemoteDupKVStream(ctx, region, kr, importClientFactory, m.resourceGroupName, m.taskType)
+				stream, err := NewRemoteDupKVStream(ctx, region, kr, importClientFactory, m.resourceGroupName, m.taskType, m.minCommitTS)
 				if err != nil {
 					return errors.Annotatef(err, "failed to create remote duplicate kv stream")
 				}
diff --git a/pkg/owner/BUILD.bazel b/pkg/owner/BUILD.bazel
index 319601e30e..8fe2e02fba 100644
--- a/pkg/owner/BUILD.bazel
+++ b/pkg/owner/BUILD.bazel
@@ -37,7 +37,7 @@ go_test(
     ],
     embed = [":owner"],
     flaky = True,
-    shard_count = 7,
+    shard_count = 8,
     deps = [
         "//pkg/ddl",
         "//pkg/infoschema",
@@ -54,6 +54,7 @@ go_test(
         "@com_github_stretchr_testify//require",
         "@io_etcd_go_etcd_client_v3//:client",
         "@io_etcd_go_etcd_client_v3//concurrency",
+        "@io_etcd_go_etcd_server_v3//embed",
         "@io_etcd_go_etcd_tests_v3//integration",
         "@org_golang_google_grpc//:grpc",
         "@org_uber_go_goleak//:goleak",
diff --git a/pkg/owner/manager.go b/pkg/owner/manager.go
index cfd131b091..f91a628308 100644
--- a/pkg/owner/manager.go
+++ b/pkg/owner/manager.go
@@ -490,3 +490,38 @@ func init() {
 		logutil.BgLogger().Warn("set manager session TTL failed", zap.Error(err))
 	}
 }
+
+// AcquireDistributedLock creates a mutex with ETCD client, and returns a mutex release function.
+func AcquireDistributedLock(
+	ctx context.Context,
+	cli *clientv3.Client,
+	key string,
+	ttlInSec int,
+) (release func(), err error) {
+	se, err := concurrency.NewSession(cli, concurrency.WithTTL(ttlInSec))
+	if err != nil {
+		return nil, err
+	}
+	mu := concurrency.NewMutex(se, key)
+	err = mu.Lock(ctx)
+	if err != nil {
+		err1 := se.Close()
+		if err1 != nil {
+			logutil.Logger(ctx).Warn("close session error", zap.Error(err1))
+		}
+		return nil, err
+	}
+	logutil.Logger(ctx).Info("acquire distributed flush lock success", zap.String("key", key))
+	return func() {
+		err = mu.Unlock(ctx)
+		if err != nil {
+			logutil.Logger(ctx).Warn("release distributed flush lock error", zap.Error(err), zap.String("key", key))
+		} else {
+			logutil.Logger(ctx).Info("release distributed flush lock success", zap.String("key", key))
+		}
+		err = se.Close()
+		if err != nil {
+			logutil.Logger(ctx).Warn("close session error", zap.Error(err))
+		}
+	}, nil
+}
diff --git a/pkg/planner/cardinality/row_count_column.go b/pkg/planner/cardinality/row_count_column.go
index dd8fba0dd2..3cb966c2d2 100644
--- a/pkg/planner/cardinality/row_count_column.go
+++ b/pkg/planner/cardinality/row_count_column.go
@@ -15,6 +15,8 @@
 package cardinality
 
 import (
+	"math"
+
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/planner/context"
 	"github.com/pingcap/tidb/pkg/planner/util/debugtrace"
@@ -172,12 +174,27 @@ func equalRowCountOnColumn(sctx context.PlanContext, c *statistics.Column, val t
 	// 3. use uniform distribution assumption for the rest (even when this value is not covered by the range of stats)
 	histNDV := float64(c.Histogram.NDV - int64(c.TopN.Num()))
 	if histNDV <= 0 {
-		// If the table hasn't been modified, it's safe to return 0. Otherwise, the TopN could be stale - return 1.
+		// If histNDV is zero - we have all NDV's in TopN - and no histograms. This function uses
+		// c.NotNullCount rather than c.Histogram.NotNullCount() since the histograms are empty.
+		//
+		// If the table hasn't been modified, it's safe to return 0.
 		if modifyCount == 0 {
 			return 0, nil
 		}
-		return 1, nil
+		// ELSE calculate an approximate estimate based upon newly inserted rows.
+		//
+		// Reset to the original NDV, or if no NDV - derive an NDV using sqrt
+		if c.Histogram.NDV > 0 {
+			histNDV = float64(c.Histogram.NDV)
+		} else {
+			histNDV = math.Sqrt(max(c.NotNullCount(), float64(realtimeRowCount)))
+		}
+		// As a conservative estimate - take the smaller of the orignal totalRows or the additions.
+		// "realtimeRowCount - original count" is a better measure of inserts than modifyCount
+		totalRowCount := min(c.NotNullCount(), float64(realtimeRowCount)-c.NotNullCount())
+		return max(1, totalRowCount/histNDV), nil
 	}
+	// return the average histogram rows (which excludes topN) and NDV that excluded topN
 	return c.Histogram.NotNullCount() / histNDV, nil
 }
 
diff --git a/pkg/planner/cardinality/row_count_index.go b/pkg/planner/cardinality/row_count_index.go
index 5804b80ea4..a3cbbbbd07 100644
--- a/pkg/planner/cardinality/row_count_index.go
+++ b/pkg/planner/cardinality/row_count_index.go
@@ -409,12 +409,27 @@ func equalRowCountOnIndex(sctx context.PlanContext, idx *statistics.Index, b []b
 	// 3. use uniform distribution assumption for the rest (even when this value is not covered by the range of stats)
 	histNDV := float64(idx.Histogram.NDV - int64(idx.TopN.Num()))
 	if histNDV <= 0 {
-		// If the table hasn't been modified, it's safe to return 0. Otherwise, the TopN could be stale - return 1.
+		// If histNDV is zero - we have all NDV's in TopN - and no histograms. This function uses
+		// idx.TotalRowCount rather than idx.Histogram.NotNullCount() since the histograms are empty.
+		//
+		// If the table hasn't been modified, it's safe to return 0.
 		if modifyCount == 0 {
 			return 0
 		}
-		return 1
+		// ELSE calculate an approximate estimate based upon newly inserted rows.
+		//
+		// Reset to the original NDV, or if no NDV - derive an NDV using sqrt
+		if idx.Histogram.NDV > 0 {
+			histNDV = float64(idx.Histogram.NDV)
+		} else {
+			histNDV = math.Sqrt(max(idx.TotalRowCount(), float64(realtimeRowCount)))
+		}
+		// As a conservative estimate - take the smaller of the orignal totalRows or the additions.
+		// "realtimeRowCount - original count" is a better measure of inserts than modifyCount
+		totalRowCount := min(idx.TotalRowCount(), float64(realtimeRowCount)-idx.TotalRowCount())
+		return max(1, totalRowCount/histNDV)
 	}
+	// return the average histogram rows (which excludes topN) and NDV that excluded topN
 	return idx.Histogram.NotNullCount() / histNDV
 }
 
diff --git a/pkg/session/bootstrap.go b/pkg/session/bootstrap.go
index af72214f85..374f342147 100644
--- a/pkg/session/bootstrap.go
+++ b/pkg/session/bootstrap.go
@@ -31,7 +31,6 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/pkg/bindinfo"
 	"github.com/pingcap/tidb/pkg/config"
-	"github.com/pingcap/tidb/pkg/disttask/framework/proto"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/domain/infosync"
 	"github.com/pingcap/tidb/pkg/expression"
@@ -1332,51 +1331,6 @@ var (
 	SupportUpgradeHTTPOpVer int64 = version174
 )
 
-func checkDistTask(s sessiontypes.Session, ver int64) {
-	if ver > version195 {
-		// since version195 we enable dist task by default, no need to check
-		return
-	}
-	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnBootstrap)
-	rs, err := s.ExecuteInternal(ctx, "SELECT HIGH_PRIORITY variable_value from mysql.global_variables where variable_name = %?;", variable.TiDBEnableDistTask)
-	if err != nil {
-		logutil.BgLogger().Fatal("check dist task failed, getting tidb_enable_dist_task failed", zap.Error(err))
-	}
-	defer terror.Call(rs.Close)
-	req := rs.NewChunk(nil)
-	err = rs.Next(ctx, req)
-	if err != nil {
-		logutil.BgLogger().Fatal("check dist task failed, getting tidb_enable_dist_task failed", zap.Error(err))
-	}
-	if req.NumRows() == 0 {
-		// Not set yet.
-		return
-	} else if req.GetRow(0).GetString(0) == variable.On {
-		logutil.BgLogger().Fatal("check dist task failed, tidb_enable_dist_task is enabled", zap.Error(err))
-	}
-
-	// Even if the variable is set to `off`, we still need to check the tidb_global_task.
-	rs2, err := s.ExecuteInternal(ctx, `SELECT id FROM %n.%n WHERE state not in (%?, %?, %?)`,
-		mysql.SystemDB,
-		"tidb_global_task",
-		proto.TaskStateSucceed,
-		proto.TaskStateFailed,
-		proto.TaskStateReverted,
-	)
-	if err != nil {
-		logutil.BgLogger().Fatal("check dist task failed, reading tidb_global_task failed", zap.Error(err))
-	}
-	defer terror.Call(rs2.Close)
-	req = rs2.NewChunk(nil)
-	err = rs2.Next(ctx, req)
-	if err != nil {
-		logutil.BgLogger().Fatal("check dist task failed, reading tidb_global_task failed", zap.Error(err))
-	}
-	if req.NumRows() > 0 {
-		logutil.BgLogger().Fatal("check dist task failed, some distributed tasks is still running", zap.Error(err))
-	}
-}
-
 // upgrade function  will do some upgrade works, when the system is bootstrapped by low version TiDB server
 // For example, add new system variables into mysql.global_variables table.
 func upgrade(s sessiontypes.Session) {
@@ -1387,7 +1341,6 @@ func upgrade(s sessiontypes.Session) {
 		return
 	}
 
-	checkDistTask(s, ver)
 	printClusterState(s, ver)
 
 	// Only upgrade from under version92 and this TiDB is not owner set.
diff --git a/pkg/util/prefetch/reader.go b/pkg/util/prefetch/reader.go
index f61d5bf016..863ce942f7 100644
--- a/pkg/util/prefetch/reader.go
+++ b/pkg/util/prefetch/reader.go
@@ -16,6 +16,7 @@ package prefetch
 
 import (
 	"bytes"
+	"errors"
 	"io"
 	"sync"
 )
@@ -54,7 +55,7 @@ func (r *Reader) run() {
 	for {
 		r.bufIdx = (r.bufIdx + 1) % 2
 		buf := r.buf[r.bufIdx]
-		n, err := r.r.Read(buf)
+		n, err := io.ReadFull(r.r, buf)
 		buf = buf[:n]
 		select {
 		case <-r.closedCh:
@@ -62,6 +63,12 @@ func (r *Reader) run() {
 		case r.bufCh <- buf:
 		}
 		if err != nil {
+			if errors.Is(err, io.ErrUnexpectedEOF) {
+				// this is caused by io.ReadFull. Because we are prefetching, the buffer size may
+				// be larger that caller's need. So we return io.EOF instead. Let caller check
+				// its needed size to convert io.EOF to io.ErrUnexpectedEOF.
+				err = io.EOF
+			}
 			r.err = err
 			close(r.bufCh)
 			return


Test patch:
diff --git a/br/pkg/storage/gcs_test.go b/br/pkg/storage/gcs_test.go
index 6a6f5183ef..25fa446045 100644
--- a/br/pkg/storage/gcs_test.go
+++ b/br/pkg/storage/gcs_test.go
@@ -576,8 +576,8 @@ func TestSpeedReadManyFiles(t *testing.T) {
 }
 
 func TestGCSShouldRetry(t *testing.T) {
-	require.True(t, shouldRetry(&url.Error{Err: goerrors.New("http2: server sent GOAWAY and closed the connectiont"), Op: "Get", URL: "https://storage.googleapis.com/storage/v1/"}))
 	require.True(t, shouldRetry(&url.Error{Err: goerrors.New("http2: client connection lost"), Op: "Get", URL: "https://storage.googleapis.com/storage/v1/"}))
+	require.True(t, shouldRetry(&url.Error{Err: io.EOF, Op: "Get", URL: "https://storage.googleapis.com/storage/v1/"}))
 }
 
 func TestCtxUsage(t *testing.T) {
diff --git a/br/pkg/storage/s3_test.go b/br/pkg/storage/s3_test.go
index 0f28b9514f..09ec2d6bb1 100644
--- a/br/pkg/storage/s3_test.go
+++ b/br/pkg/storage/s3_test.go
@@ -1472,3 +1472,30 @@ func TestS3ReadFileRetryable(t *testing.T) {
 	require.Error(t, err)
 	require.True(t, strings.Contains(err.Error(), errMsg))
 }
+
+func TestOpenRangeMismatchErrorMsg(t *testing.T) {
+	s := createS3Suite(t)
+	ctx := aws.BackgroundContext()
+	start, end := int64(10), int64(30)
+
+	s.s3.EXPECT().
+		GetObjectWithContext(ctx, gomock.Any()).
+		DoAndReturn(func(context.Context, *s3.GetObjectInput, ...request.Option) (*s3.GetObjectOutput, error) {
+			return &s3.GetObjectOutput{
+				ContentRange: aws.String("bytes 10-20/20"),
+			}, nil
+		})
+	reader, err := s.storage.Open(ctx, "test", &ReaderOption{StartOffset: &start, EndOffset: &end})
+	require.ErrorContains(t, err, "expected range: bytes=10-29, got: bytes 10-20/20")
+	require.Nil(t, reader)
+
+	s.s3.EXPECT().
+		GetObjectWithContext(ctx, gomock.Any()).
+		DoAndReturn(func(context.Context, *s3.GetObjectInput, ...request.Option) (*s3.GetObjectOutput, error) {
+			return &s3.GetObjectOutput{}, nil
+		})
+	reader, err = s.storage.Open(ctx, "test", &ReaderOption{StartOffset: &start, EndOffset: &end})
+	// other function will throw error
+	require.ErrorContains(t, err, "ContentRange is empty")
+	require.Nil(t, reader)
+}
diff --git a/pkg/executor/infoschema_reader_test.go b/pkg/executor/infoschema_reader_test.go
index 632b71dd28..b8653d1474 100644
--- a/pkg/executor/infoschema_reader_test.go
+++ b/pkg/executor/infoschema_reader_test.go
@@ -401,19 +401,21 @@ func TestTiFlashSystemTableWithTiFlashV620(t *testing.T) {
 		}, nil
 	})
 
+	// When try to parse the result from old version tiflash instance with the latest columns name, some columns may not exist.
+	// The missing columns will be filled with <nil> value.
 	store := testkit.CreateMockStore(t, withMockTiFlash(1), mocker.AsOpt())
 	tk := testkit.NewTestKit(t, store)
 	tk.MustQuery("select * from information_schema.TIFLASH_SEGMENTS;").Check(testkit.Rows(
-		"db_1 t_10 mysql tables_priv 10 0 1 [-9223372036854775808,9223372036854775807) <nil> 0 0 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 0 2032 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 127.0.0.1:3933",
-		"db_1 t_8 mysql db 8 0 1 [-9223372036854775808,9223372036854775807) <nil> 0 0 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 0 2032 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 127.0.0.1:3933",
-		"db_2 t_70 test segment 70 0 1 [01,FA) <nil> 30511 50813627 0.6730359542460096 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 3578860 409336 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 127.0.0.1:3933",
+		"db_1 t_10 mysql tables_priv 10 0 1 [-9223372036854775808,9223372036854775807) <nil> 0 0 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 0 <nil> 2032 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 127.0.0.1:3933",
+		"db_1 t_8 mysql db 8 0 1 [-9223372036854775808,9223372036854775807) <nil> 0 0 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 0 <nil> 2032 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 127.0.0.1:3933",
+		"db_2 t_70 test segment 70 0 1 [01,FA) <nil> 30511 50813627 0.6730359542460096 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 3578860 <nil> 409336 <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> <nil> 127.0.0.1:3933",
 	))
 	tk.MustQuery("show warnings").Check(testkit.Rows())
 
 	tk.MustQuery("select * from information_schema.TIFLASH_TABLES;").Check(testkit.Rows(
-		"db_1 t_10 mysql tables_priv 10 0 1 0 0 0 <nil> 0 <nil> 0 <nil> <nil> 0 0 0 0 0 0 <nil> <nil> <nil> 0 0 0 0 <nil> <nil> 0 <nil> <nil> <nil> <nil> 0 <nil> <nil> <nil> 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_1 t_8 mysql db 8 0 1 0 0 0 <nil> 0 <nil> 0 <nil> <nil> 0 0 0 0 0 0 <nil> <nil> <nil> 0 0 0 0 <nil> <nil> 0 <nil> <nil> <nil> <nil> 0 <nil> <nil> <nil> 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_2 t_70 test segment 70 0 1 102000 169873868 0 0 0 <nil> 0 <nil> <nil> 0 102000 169873868 0 0 0 <nil> <nil> <nil> 1 102000 169873868 43867622 102000 169873868 0 <nil> <nil> <nil> <nil> 13 13 7846.153846153846 13067220.615384616 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_1 t_10 mysql tables_priv 10 0 <nil> 1 0 0 0 <nil> 0 <nil> 0 <nil> <nil> <nil> 0 0 0 0 0 0 <nil> <nil> <nil> 0 0 0 0 <nil> <nil> 0 <nil> <nil> <nil> <nil> 0 <nil> <nil> <nil> 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_1 t_8 mysql db 8 0 <nil> 1 0 0 0 <nil> 0 <nil> 0 <nil> <nil> <nil> 0 0 0 0 0 0 <nil> <nil> <nil> 0 0 0 0 <nil> <nil> 0 <nil> <nil> <nil> <nil> 0 <nil> <nil> <nil> 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_2 t_70 test segment 70 0 <nil> 1 102000 169873868 0 0 0 <nil> 0 <nil> <nil> <nil> 0 102000 169873868 0 0 0 <nil> <nil> <nil> 1 102000 169873868 43867622 102000 169873868 0 <nil> <nil> <nil> <nil> 13 13 7846.153846153846 13067220.615384616 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
 	))
 	tk.MustQuery("show warnings").Check(testkit.Rows())
 }
@@ -438,18 +440,20 @@ func TestTiFlashSystemTableWithTiFlashV630(t *testing.T) {
 		}, nil
 	})
 
+	// When try to parse the result from old version tiflash instance with the latest columns name, some columns may not exist.
+	// The missing columns will be filled with <nil> value.
 	store := testkit.CreateMockStore(t, withMockTiFlash(1), mocker.AsOpt())
 	tk := testkit.NewTestKit(t, store)
 	tk.MustQuery("select * from information_schema.TIFLASH_SEGMENTS;").Check(testkit.Rows(
-		"db_1 t_10 mysql tables_priv 10 0 1 [-9223372036854775808,9223372036854775807) 0 0 0 <nil> 0 0 0 0 2 0 0 0 0 0 2032 3 0 0 1 1 0 0 0 0 127.0.0.1:3933",
-		"db_2 t_70 test segment 70 436272981189328904 1 [01,FA) 5 102000 169874232 0 0 0 0 0 2 0 0 0 0 0 2032 3 102000 169874232 1 68 102000 169874232 43951837 20 127.0.0.1:3933",
-		"db_2 t_75 test segment 75 0 1 [01,013130303030393535FF61653666642D6136FF61382D343032382DFF616436312D663736FF3062323736643461FF3600000000000000F8) 2 0 0 <nil> 0 0 1 1 110 0 0 4 4 0 2032 111 0 0 1 70 0 0 0 0 127.0.0.1:3933",
-		"db_2 t_75 test segment 75 0 113 [013130303030393535FF61653666642D6136FF61382D343032382DFF616436312D663736FF3062323736643461FF3600000000000000F8,013139393938363264FF33346535382D3735FF31382D343661612DFF626235392D636264FF3139333434623736FF3100000000000000F9) 2 10167 16932617 0.4887380741615029 0 0 0 0 114 4969 8275782 2 0 0 63992 112 5198 8656835 1 71 5198 8656835 2254100 1 127.0.0.1:3933",
-		"db_2 t_75 test segment 75 0 116 [013139393938363264FF33346535382D3735FF31382D343661612DFF626235392D636264FF3139333434623736FF3100000000000000F9,013330303131383034FF61323537662D6638FF63302D346466622DFF383235632D353361FF3236306338616662FF3400000000000000F8) 3 8 13322 0.5 3 4986 1 0 117 1 1668 4 3 4986 2032 115 4 6668 1 78 4 6668 6799 1 127.0.0.1:3933",
-		"db_2 t_75 test segment 75 0 125 [013330303131383034FF61323537662D6638FF63302D346466622DFF383235632D353361FF3236306338616662FF3400000000000000F8,013339393939613861FF30663062332D6537FF32372D346234642DFF396535632D363865FF3336323066383431FF6300000000000000F9) 2 8677 14451079 0.4024432407514118 3492 5816059 3 0 126 0 0 0 0 5816059 2032 124 5185 8635020 1 79 5185 8635020 2247938 1 127.0.0.1:3933",
-		"db_2 t_75 test segment 75 0 128 [013339393939613861FF30663062332D6537FF32372D346234642DFF396535632D363865FF3336323066383431FF6300000000000000F9,013730303031636230FF32663330652D3539FF62352D346134302DFF613539312D383930FF6132316364633466FF3200000000000000F8) 0 1 1668 1 0 0 0 0 129 1 1668 5 4 0 2032 127 0 0 1 78 4 6668 6799 1 127.0.0.1:3933",
-		"db_2 t_75 test segment 75 0 119 [013730303031636230FF32663330652D3539FF62352D346134302DFF613539312D383930FF6132316364633466FF3200000000000000F8,013739393939386561FF36393566612D3534FF64302D346437642DFF383136612D646335FF6432613130353533FF3200000000000000F9) 2 10303 17158730 0.489372027564787 0 0 0 0 120 5042 8397126 2 0 0 63992 118 5261 8761604 1 77 5261 8761604 2280506 1 127.0.0.1:3933",
-		"db_2 t_75 test segment 75 0 122 [013739393939386561FF36393566612D3534FF64302D346437642DFF383136612D646335FF6432613130353533FF3200000000000000F9,FA) 0 1 1663 1 0 0 0 0 123 1 1663 4 3 0 2032 121 0 0 1 78 4 6668 6799 1 127.0.0.1:3933",
+		"db_1 t_10 mysql tables_priv 10 0 1 [-9223372036854775808,9223372036854775807) 0 0 0 <nil> 0 0 0 0 2 0 0 0 0 0 <nil> 2032 3 0 0 1 1 0 0 0 0 127.0.0.1:3933",
+		"db_2 t_70 test segment 70 436272981189328904 1 [01,FA) 5 102000 169874232 0 0 0 0 0 2 0 0 0 0 0 <nil> 2032 3 102000 169874232 1 68 102000 169874232 43951837 20 127.0.0.1:3933",
+		"db_2 t_75 test segment 75 0 1 [01,013130303030393535FF61653666642D6136FF61382D343032382DFF616436312D663736FF3062323736643461FF3600000000000000F8) 2 0 0 <nil> 0 0 1 1 110 0 0 4 4 0 <nil> 2032 111 0 0 1 70 0 0 0 0 127.0.0.1:3933",
+		"db_2 t_75 test segment 75 0 113 [013130303030393535FF61653666642D6136FF61382D343032382DFF616436312D663736FF3062323736643461FF3600000000000000F8,013139393938363264FF33346535382D3735FF31382D343661612DFF626235392D636264FF3139333434623736FF3100000000000000F9) 2 10167 16932617 0.4887380741615029 0 0 0 0 114 4969 8275782 2 0 0 <nil> 63992 112 5198 8656835 1 71 5198 8656835 2254100 1 127.0.0.1:3933",
+		"db_2 t_75 test segment 75 0 116 [013139393938363264FF33346535382D3735FF31382D343661612DFF626235392D636264FF3139333434623736FF3100000000000000F9,013330303131383034FF61323537662D6638FF63302D346466622DFF383235632D353361FF3236306338616662FF3400000000000000F8) 3 8 13322 0.5 3 4986 1 0 117 1 1668 4 3 4986 <nil> 2032 115 4 6668 1 78 4 6668 6799 1 127.0.0.1:3933",
+		"db_2 t_75 test segment 75 0 125 [013330303131383034FF61323537662D6638FF63302D346466622DFF383235632D353361FF3236306338616662FF3400000000000000F8,013339393939613861FF30663062332D6537FF32372D346234642DFF396535632D363865FF3336323066383431FF6300000000000000F9) 2 8677 14451079 0.4024432407514118 3492 5816059 3 0 126 0 0 0 0 5816059 <nil> 2032 124 5185 8635020 1 79 5185 8635020 2247938 1 127.0.0.1:3933",
+		"db_2 t_75 test segment 75 0 128 [013339393939613861FF30663062332D6537FF32372D346234642DFF396535632D363865FF3336323066383431FF6300000000000000F9,013730303031636230FF32663330652D3539FF62352D346134302DFF613539312D383930FF6132316364633466FF3200000000000000F8) 0 1 1668 1 0 0 0 0 129 1 1668 5 4 0 <nil> 2032 127 0 0 1 78 4 6668 6799 1 127.0.0.1:3933",
+		"db_2 t_75 test segment 75 0 119 [013730303031636230FF32663330652D3539FF62352D346134302DFF613539312D383930FF6132316364633466FF3200000000000000F8,013739393939386561FF36393566612D3534FF64302D346437642DFF383136612D646335FF6432613130353533FF3200000000000000F9) 2 10303 17158730 0.489372027564787 0 0 0 0 120 5042 8397126 2 0 0 <nil> 63992 118 5261 8761604 1 77 5261 8761604 2280506 1 127.0.0.1:3933",
+		"db_2 t_75 test segment 75 0 122 [013739393939386561FF36393566612D3534FF64302D346437642DFF383136612D646335FF6432613130353533FF3200000000000000F9,FA) 0 1 1663 1 0 0 0 0 123 1 1663 4 3 0 <nil> 2032 121 0 0 1 78 4 6668 6799 1 127.0.0.1:3933",
 	))
 	tk.MustQuery("show warnings").Check(testkit.Rows())
 }
@@ -474,18 +478,20 @@ func TestTiFlashSystemTableWithTiFlashV640(t *testing.T) {
 		}, nil
 	})
 
+	// When try to parse the result from old version tiflash instance with the latest columns name, some columns may not exist.
+	// The missing columns will be filled with <nil> value.
 	store := testkit.CreateMockStore(t, withMockTiFlash(1), mocker.AsOpt())
 	tk := testkit.NewTestKit(t, store)
 	tk.MustQuery("select * from information_schema.TIFLASH_TABLES;").Check(testkit.Rows(
-		"db_70 t_135 tpcc customer 135 0 4 3528714 2464079200 0 0.002329177144988231 1 0 929227 0.16169850346757514 0 8128 882178.5 616019800 4 8219 5747810 2054.75 1436952.5 0 4 3520495 2458331390 1601563417 880123.75 614582847.5 24 8 6 342.4583333333333 239492.08333333334 482 120.5 7303.9315352697095 5100272.593360996 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_70 t_137 tpcc district 137 0 1 7993 1346259 0 0.8748905292130614 1 0.8055198055198055 252168 0.21407121407121407 0 147272 7993 1346259 1 6993 1178050 6993 1178050 0 1 1000 168209 91344 1000 168209 6 6 6 1165.5 196341.66666666666 10 10 100 16820.9 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_70 t_139 tpcc history 139 0 19 19379697 1629276978 0 0.0006053758219233252 0.5789473684210527 0.4626662120695534 253640 0.25434708489601093 0 293544 1019984.052631579 85751419.89473684 11 11732 997220 1066.5454545454545 90656.36363636363 0 19 19367965 1628279758 625147717 1019366.5789473684 85698934.63157895 15 4 1.3636363636363635 782.1333333333333 66481.33333333333 2378 125.15789473684211 8144.644659377628 684726.559293524 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_70 t_141 tpcc item 141 0 1 100000 10799081 0 0 0 <nil> 0 <nil> <nil> 0 100000 10799081 0 0 0 <nil> <nil> <nil> 1 100000 10799081 7357726 100000 10799081 0 0 <nil> <nil> <nil> 13 13 7692.307692307692 830698.5384615385 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_70 t_143 tpcc new_order 143 0 4 2717707 78813503 0 0.02266763856442214 1 0.9678592299201351 52809 0.029559768846178818 0 1434208 679426.75 19703375.75 4 61604 1786516 15401 446629 0 3 2656103 77026987 40906492 885367.6666666666 25675662.333333332 37 24 9.25 1664.972972972973 48284.21621621621 380 126.66666666666667 6989.744736842105 202702.59736842106 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_70 t_145 tpcc order_line 145 0 203 210607202 20007684190 0 0.0054566462546708164 0.5862068965517241 0.7810067620424135 620065 0.005679558722564825 0 22607144 1037473.9014778325 98560020.64039409 119 1149209 109174855 9657.218487394957 917435.756302521 0 203 209457993 19898509335 8724002804 1031812.7733990147 98022213.47290641 893 39 7.504201680672269 1286.9081746920492 122256.27659574468 31507 155.20689655172413 6647.982765734599 631558.3627447869 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_70 t_147 tpcc orders 147 0 22 21903301 1270391458 0 0.02021357420052804 0.7272727272727273 0.9239944527763222 260536 0.010145817899282655 0 10025264 995604.5909090909 57745066.27272727 16 442744 25679152 27671.5 1604947 0 22 21460557 1244712306 452173775 975479.8636363636 56577832.09090909 242 34 15.125 1829.5206611570247 106112.19834710743 2973 135.13636363636363 7218.485368314833 418672.15136226034 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_70 t_149 tpcc stock 149 0 42 11112720 4811805131 0 0.028085203262567582 0.9761904761904762 0.8463391893060944 10227093 0.07567373591410528 0 6719064 264588.5714285714 114566788.83333333 41 312103 135131097 7612.268292682927 3295880.4146341463 0 42 10800617 4676674034 3231872509 257157.54761904763 111349381.76190476 238 26 5.804878048780488 1311.357142857143 567777.718487395 1644 39.142857142857146 6569.718369829684 2844692.234793187 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
-		"db_70 t_151 tpcc warehouse 151 0 1 5842 923615 0 0.9828825744608011 1 0.9669104841518634 70220 0.07732497387669801 0 133048 5842 923615 1 5742 907807 5742 907807 0 1 100 15808 11642 100 15808 5 5 5 1148.4 181561.4 5 5 20 3161.6 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_135 tpcc customer 135 0 <nil> 4 3528714 2464079200 0 0.002329177144988231 1 0 929227 <nil> 0.16169850346757514 0 8128 882178.5 616019800 4 8219 5747810 2054.75 1436952.5 0 4 3520495 2458331390 1601563417 880123.75 614582847.5 24 8 6 342.4583333333333 239492.08333333334 482 120.5 7303.9315352697095 5100272.593360996 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_137 tpcc district 137 0 <nil> 1 7993 1346259 0 0.8748905292130614 1 0.8055198055198055 252168 <nil> 0.21407121407121407 0 147272 7993 1346259 1 6993 1178050 6993 1178050 0 1 1000 168209 91344 1000 168209 6 6 6 1165.5 196341.66666666666 10 10 100 16820.9 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_139 tpcc history 139 0 <nil> 19 19379697 1629276978 0 0.0006053758219233252 0.5789473684210527 0.4626662120695534 253640 <nil> 0.25434708489601093 0 293544 1019984.052631579 85751419.89473684 11 11732 997220 1066.5454545454545 90656.36363636363 0 19 19367965 1628279758 625147717 1019366.5789473684 85698934.63157895 15 4 1.3636363636363635 782.1333333333333 66481.33333333333 2378 125.15789473684211 8144.644659377628 684726.559293524 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_141 tpcc item 141 0 <nil> 1 100000 10799081 0 0 0 <nil> 0 <nil> <nil> <nil> 0 100000 10799081 0 0 0 <nil> <nil> <nil> 1 100000 10799081 7357726 100000 10799081 0 0 <nil> <nil> <nil> 13 13 7692.307692307692 830698.5384615385 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_143 tpcc new_order 143 0 <nil> 4 2717707 78813503 0 0.02266763856442214 1 0.9678592299201351 52809 <nil> 0.029559768846178818 0 1434208 679426.75 19703375.75 4 61604 1786516 15401 446629 0 3 2656103 77026987 40906492 885367.6666666666 25675662.333333332 37 24 9.25 1664.972972972973 48284.21621621621 380 126.66666666666667 6989.744736842105 202702.59736842106 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_145 tpcc order_line 145 0 <nil> 203 210607202 20007684190 0 0.0054566462546708164 0.5862068965517241 0.7810067620424135 620065 <nil> 0.005679558722564825 0 22607144 1037473.9014778325 98560020.64039409 119 1149209 109174855 9657.218487394957 917435.756302521 0 203 209457993 19898509335 8724002804 1031812.7733990147 98022213.47290641 893 39 7.504201680672269 1286.9081746920492 122256.27659574468 31507 155.20689655172413 6647.982765734599 631558.3627447869 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_147 tpcc orders 147 0 <nil> 22 21903301 1270391458 0 0.02021357420052804 0.7272727272727273 0.9239944527763222 260536 <nil> 0.010145817899282655 0 10025264 995604.5909090909 57745066.27272727 16 442744 25679152 27671.5 1604947 0 22 21460557 1244712306 452173775 975479.8636363636 56577832.09090909 242 34 15.125 1829.5206611570247 106112.19834710743 2973 135.13636363636363 7218.485368314833 418672.15136226034 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_149 tpcc stock 149 0 <nil> 42 11112720 4811805131 0 0.028085203262567582 0.9761904761904762 0.8463391893060944 10227093 <nil> 0.07567373591410528 0 6719064 264588.5714285714 114566788.83333333 41 312103 135131097 7612.268292682927 3295880.4146341463 0 42 10800617 4676674034 3231872509 257157.54761904763 111349381.76190476 238 26 5.804878048780488 1311.357142857143 567777.718487395 1644 39.142857142857146 6569.718369829684 2844692.234793187 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
+		"db_70 t_151 tpcc warehouse 151 0 <nil> 1 5842 923615 0 0.9828825744608011 1 0.9669104841518634 70220 <nil> 0.07732497387669801 0 133048 5842 923615 1 5742 907807 5742 907807 0 1 100 15808 11642 100 15808 5 5 5 1148.4 181561.4 5 5 20 3161.6 0 0 0  0 0 0  0 0 0  0 127.0.0.1:3933",
 	))
 	tk.MustQuery("show warnings").Check(testkit.Rows())
 }
diff --git a/pkg/executor/test/issuetest/BUILD.bazel b/pkg/executor/test/issuetest/BUILD.bazel
index d825d3bd34..c25cadbdbd 100644
--- a/pkg/executor/test/issuetest/BUILD.bazel
+++ b/pkg/executor/test/issuetest/BUILD.bazel
@@ -8,10 +8,11 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
-    shard_count = 22,
+    shard_count = 23,
     deps = [
         "//pkg/autoid_service",
         "//pkg/config",
+        "//pkg/executor",
         "//pkg/kv",
         "//pkg/meta/autoid",
         "//pkg/parser/auth",
@@ -19,6 +20,7 @@ go_test(
         "//pkg/parser/mysql",
         "//pkg/session/types",
         "//pkg/testkit",
+        "//pkg/testkit/testfailpoint",
         "//pkg/util",
         "//pkg/util/dbterror/exeerrors",
         "//pkg/util/memory",
diff --git a/pkg/executor/test/issuetest/executor_issue_test.go b/pkg/executor/test/issuetest/executor_issue_test.go
index 5855c1ee17..310b3c7db4 100644
--- a/pkg/executor/test/issuetest/executor_issue_test.go
+++ b/pkg/executor/test/issuetest/executor_issue_test.go
@@ -24,12 +24,14 @@ import (
 	"github.com/pingcap/failpoint"
 	_ "github.com/pingcap/tidb/pkg/autoid_service"
 	"github.com/pingcap/tidb/pkg/config"
+	"github.com/pingcap/tidb/pkg/executor"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/auth"
 	"github.com/pingcap/tidb/pkg/parser/charset"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	sessiontypes "github.com/pingcap/tidb/pkg/session/types"
 	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
 	"github.com/pingcap/tidb/pkg/util"
 	"github.com/pingcap/tidb/pkg/util/dbterror/exeerrors"
 	"github.com/pingcap/tidb/pkg/util/memory"
@@ -712,3 +714,21 @@ func TestIssue55881(t *testing.T) {
 			"(select max(value) from (select * from cte union all select * from cte union all select * from aaa where aaa.id > bbb.id)) from bbb;")
 	}
 }
+
+func TestIssue60926(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t1")
+	tk.MustExec("drop table if exists t2")
+	tk.MustExec("create table t1 (col0 int, col1 int);")
+	tk.MustExec("create table t2 (col0 int, col1 int);")
+	tk.MustExec("insert into t1 values (0, 10), (1, 10), (2, 10), (3, 10), (4, 10), (5, 10), (6, 10), (7, 10), (8, 10), (9, 10), (10, 10);")
+	tk.MustExec("insert into t2 values (0, 5), (0, 5), (1, 5), (2, 5), (2, 5), (3, 5), (4, 5), (5, 5), (5, 5), (6, 5), (7, 5), (8, 5), (8, 5), (9, 5), (9, 5), (10, 5);")
+
+	testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/executor/join/issue60926", "panic")
+	executor.IsChildCloseCalledForTest.Store(false)
+	tk.MustQuery("select * from t1 join (select col0, sum(col1) from t2 group by col0) as r on t1.col0 = r.col0;")
+	require.True(t, executor.IsChildCloseCalledForTest.Load())
+}
diff --git a/pkg/owner/manager_test.go b/pkg/owner/manager_test.go
index 6359882703..f406341ec4 100644
--- a/pkg/owner/manager_test.go
+++ b/pkg/owner/manager_test.go
@@ -17,6 +17,8 @@ package owner_test
 import (
 	"context"
 	"fmt"
+	"math/rand"
+	"net/url"
 	"runtime"
 	"testing"
 	"time"
@@ -30,10 +32,12 @@ import (
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	"github.com/pingcap/tidb/pkg/store/mockstore"
 	"github.com/pingcap/tidb/pkg/testkit"
+	"github.com/pingcap/tidb/pkg/util"
 	"github.com/pingcap/tidb/pkg/util/logutil"
 	"github.com/stretchr/testify/require"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/concurrency"
+	"go.etcd.io/etcd/server/v3/embed"
 	"go.etcd.io/etcd/tests/v3/integration"
 )
 
@@ -413,3 +417,108 @@ func deleteLeader(cli *clientv3.Client, prefixKey string) error {
 	_, err = cli.Delete(context.Background(), string(resp.Kvs[0].Key))
 	return errors.Trace(err)
 }
+
+func TestAcquireDistributedLock(t *testing.T) {
+	const addrFmt = "http://127.0.0.1:%d"
+	cfg := embed.NewConfig()
+	cfg.Dir = t.TempDir()
+	// rand port in [20000, 60000)
+	randPort := int(rand.Int31n(40000)) + 20000
+	clientAddr := fmt.Sprintf(addrFmt, randPort)
+	lcurl, _ := url.Parse(clientAddr)
+	cfg.ListenClientUrls, cfg.AdvertiseClientUrls = []url.URL{*lcurl}, []url.URL{*lcurl}
+	lpurl, _ := url.Parse(fmt.Sprintf(addrFmt, randPort+1))
+	cfg.ListenPeerUrls, cfg.AdvertisePeerUrls = []url.URL{*lpurl}, []url.URL{*lpurl}
+	cfg.InitialCluster = "default=" + lpurl.String()
+	cfg.Logger = "zap"
+	embedEtcd, err := embed.StartEtcd(cfg)
+	require.NoError(t, err)
+	<-embedEtcd.Server.ReadyNotify()
+	t.Cleanup(func() {
+		embedEtcd.Close()
+	})
+	makeEtcdCli := func(t *testing.T) (cli *clientv3.Client) {
+		cli, err := clientv3.New(clientv3.Config{
+			Endpoints: []string{lcurl.String()},
+		})
+		require.NoError(t, err)
+		t.Cleanup(func() {
+			cli.Close()
+		})
+		return cli
+	}
+	t.Run("acquire distributed lock with same client", func(t *testing.T) {
+		cli := makeEtcdCli(t)
+		getLock := make(chan struct{})
+		ctx := context.Background()
+
+		release1, err := owner.AcquireDistributedLock(ctx, cli, "test-lock", 10)
+		require.NoError(t, err)
+		var wg util.WaitGroupWrapper
+		wg.Run(func() {
+			// Acquire another distributed lock will be blocked.
+			release2, err := owner.AcquireDistributedLock(ctx, cli, "test-lock", 10)
+			require.NoError(t, err)
+			getLock <- struct{}{}
+			release2()
+		})
+		timer := time.NewTimer(300 * time.Millisecond)
+		select {
+		case <-getLock:
+			require.FailNow(t, "acquired same lock unexpectedly")
+		case <-timer.C:
+			release1()
+			<-getLock
+		}
+		wg.Wait()
+
+		release1, err = owner.AcquireDistributedLock(ctx, cli, "test-lock/1", 10)
+		require.NoError(t, err)
+		release2, err := owner.AcquireDistributedLock(ctx, cli, "test-lock/2", 10)
+		require.NoError(t, err)
+		release1()
+		release2()
+	})
+
+	t.Run("acquire distributed lock with different clients", func(t *testing.T) {
+		cli1 := makeEtcdCli(t)
+		cli2 := makeEtcdCli(t)
+
+		getLock := make(chan struct{})
+		ctx := context.Background()
+
+		release1, err := owner.AcquireDistributedLock(ctx, cli1, "test-lock", 10)
+		require.NoError(t, err)
+		var wg util.WaitGroupWrapper
+		wg.Run(func() {
+			// Acquire another distributed lock will be blocked.
+			release2, err := owner.AcquireDistributedLock(ctx, cli2, "test-lock", 10)
+			require.NoError(t, err)
+			getLock <- struct{}{}
+			release2()
+		})
+		timer := time.NewTimer(300 * time.Millisecond)
+		select {
+		case <-getLock:
+			require.FailNow(t, "acquired same lock unexpectedly")
+		case <-timer.C:
+			release1()
+			<-getLock
+		}
+		wg.Wait()
+	})
+
+	t.Run("acquire distributed lock until timeout", func(t *testing.T) {
+		cli1 := makeEtcdCli(t)
+		cli2 := makeEtcdCli(t)
+		ctx := context.Background()
+
+		_, err := owner.AcquireDistributedLock(ctx, cli1, "test-lock", 1)
+		require.NoError(t, err)
+		cli1.Close() // Note that release() is not invoked.
+
+		release2, err := owner.AcquireDistributedLock(ctx, cli2, "test-lock", 10)
+		require.NoError(t, err)
+		release2()
+	})
+}
diff --git a/pkg/session/bootstrap_test.go b/pkg/session/bootstrap_test.go
index 440d38380b..b5d88bb28d 100644
--- a/pkg/session/bootstrap_test.go
+++ b/pkg/session/bootstrap_test.go
@@ -2249,13 +2249,13 @@ func testTiDBUpgradeWithDistTask(t *testing.T, injectQuery string, fatal bool) {
 }
 
 func TestTiDBUpgradeWithDistTaskEnable(t *testing.T) {
-	t.Run("test enable dist task", func(t *testing.T) { testTiDBUpgradeWithDistTask(t, "set global tidb_enable_dist_task = 1", true) })
+	t.Run("test enable dist task", func(t *testing.T) { testTiDBUpgradeWithDistTask(t, "set global tidb_enable_dist_task = 1", false) })
 	t.Run("test disable dist task", func(t *testing.T) { testTiDBUpgradeWithDistTask(t, "set global tidb_enable_dist_task = 0", false) })
 }
 
 func TestTiDBUpgradeWithDistTaskRunning(t *testing.T) {
 	t.Run("test dist task running", func(t *testing.T) {
-		testTiDBUpgradeWithDistTask(t, "insert into mysql.tidb_global_task set id = 1, task_key = 'aaa', type= 'aaa', state = 'running'", true)
+		testTiDBUpgradeWithDistTask(t, "insert into mysql.tidb_global_task set id = 1, task_key = 'aaa', type= 'aaa', state = 'running'", false)
 	})
 	t.Run("test dist task succeed", func(t *testing.T) {
 		testTiDBUpgradeWithDistTask(t, "insert into mysql.tidb_global_task set id = 1, task_key = 'aaa', type= 'aaa', state = 'succeed'", false)
@@ -2267,10 +2267,10 @@ func TestTiDBUpgradeWithDistTaskRunning(t *testing.T) {
 		testTiDBUpgradeWithDistTask(t, "insert into mysql.tidb_global_task set id = 1, task_key = 'aaa', type= 'aaa', state = 'reverted'", false)
 	})
 	t.Run("test dist task paused", func(t *testing.T) {
-		testTiDBUpgradeWithDistTask(t, "insert into mysql.tidb_global_task set id = 1, task_key = 'aaa', type= 'aaa', state = 'paused'", true)
+		testTiDBUpgradeWithDistTask(t, "insert into mysql.tidb_global_task set id = 1, task_key = 'aaa', type= 'aaa', state = 'paused'", false)
 	})
 	t.Run("test dist task other", func(t *testing.T) {
-		testTiDBUpgradeWithDistTask(t, "insert into mysql.tidb_global_task set id = 1, task_key = 'aaa', type= 'aaa', state = 'other'", true)
+		testTiDBUpgradeWithDistTask(t, "insert into mysql.tidb_global_task set id = 1, task_key = 'aaa', type= 'aaa', state = 'other'", false)
 	})
 }
 
diff --git a/pkg/util/prefetch/reader_test.go b/pkg/util/prefetch/reader_test.go
index 0d2972ed80..671990ede1 100644
--- a/pkg/util/prefetch/reader_test.go
+++ b/pkg/util/prefetch/reader_test.go
@@ -17,7 +17,9 @@ package prefetch
 import (
 	"bytes"
 	"io"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -80,3 +82,60 @@ func TestCloseBeforeDrainRead(t *testing.T) {
 	err := r.Close()
 	require.NoError(t, err)
 }
+
+type fragmentReader struct {
+	data     []byte
+	i        atomic.Int64
+	fragSize int
+}
+
+func (r *fragmentReader) Read(p []byte) (n int, err error) {
+	i := int(r.i.Load())
+	if i >= len(r.data) {
+		return 0, io.EOF
+	}
+	copySize := min(len(p), r.fragSize)
+	n = copy(p, r.data[i:i+copySize])
+	r.i.Add(int64(n))
+	return
+}
+
+func TestFillPrefetchBuffer(t *testing.T) {
+	// test fragmentReader behaviour
+	fragReader := &fragmentReader{
+		data:     []byte("0123456789"),
+		fragSize: 3,
+	}
+	buf := make([]byte, 5)
+	n, err := fragReader.Read(buf)
+	require.NoError(t, err)
+	require.Equal(t, 3, n)
+	require.Equal(t, "012", string(buf[:n]))
+
+	buf = make([]byte, 1)
+	n, err = fragReader.Read(buf)
+	require.NoError(t, err)
+	require.Equal(t, 1, n)
+	require.Equal(t, "3", string(buf[:n]))
+
+	fragReader = &fragmentReader{
+		data:     []byte("0123456789"),
+		fragSize: 3,
+	}
+	// when prefetch is 10B, the two internal ping-pong buffers are 5B
+	prefetchReader := NewReader(io.NopCloser(fragReader), 10)
+	// when no read happens, one of ping-pong buffer is fully filled
+	require.Eventually(t, func() bool {
+		return fragReader.i.Load() == 5
+	}, time.Second, 10*time.Millisecond)
+	// when any small read happens, one of ping-pong buffer is serving the read,
+	// another buffer is fully filled
+	buf = make([]byte, 2)
+	n, err = prefetchReader.Read(buf)
+	require.NoError(t, err)
+	require.EqualValues(t, 2, n)
+	require.EqualValues(t, "01", string(buf[:n]))
+	require.Eventually(t, func() bool {
+		return fragReader.i.Load() == 10
+	}, time.Second, 10*time.Millisecond)
+}
diff --git a/tests/realtikvtest/addindextest/add_index_test.go b/tests/realtikvtest/addindextest/add_index_test.go
index 8a5750274f..c3e7c9346b 100644
--- a/tests/realtikvtest/addindextest/add_index_test.go
+++ b/tests/realtikvtest/addindextest/add_index_test.go
@@ -186,7 +186,7 @@ func TestLitBackendCtxMgr(t *testing.T) {
 
 	jobID := int64(102)
 	discovery := store.(tikv.Storage).GetRegionCache().PDClient().GetServiceDiscovery()
-	backendCtx, err := mgr.Register(ctx, jobID, false, nil, discovery, "TestLitBackendCtxMgr")
+	backendCtx, err := mgr.Register(ctx, jobID, false, nil, discovery, "TestLitBackendCtxMgr", 0)
 	require.NoError(t, err)
 	require.NotNil(t, backendCtx)