Repository: pingcap/tidb (go)

Issue:
disable 'import-mode' when using global sort
## Enhance

<!-- a step by step guide for reproducing the bug. -->
'import mode' is enabled previously to avoid TiKV into write stale and flow control, to make import success even through there are a lot of range overlap and some SST might ingested into L0, but with global sort, we can make sure there is no range overlap, it shouldn't ingest into L0 normally, we can disable switching to 'import mode' in this case





Pull request:
importinto: disable switching to 'import mode' when global sort
<!--

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 #60361

Problem Summary:

### What changed and how does it work?
disable switch to `import mode` when using global sort, no change for local sort
### 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
None
```


Patch:
diff --git a/docs/design/2025-03-17-sys-session-manage-enhancement.md b/docs/design/2025-03-17-sys-session-manage-enhancement.md
new file mode 100644
index 0000000000..3f1e264a25
--- /dev/null
+++ b/docs/design/2025-03-17-sys-session-manage-enhancement.md
@@ -0,0 +1,173 @@
+# Enhancement of Internal System Session Management
+
+- Author: [Wang Chao](https://github.com/lcwangchao)
+- Discussion PR: <https://github.com/pingcap/tidb/pull/59875>
+- Tracking Issue: <https://github.com/pingcap/tidb/issues/60115>
+
+## Motivation and Background
+
+The internal system sessions are widely used in TiDB across multiple modules. To minimize the overhead of creating and destroying sessions, we use a session pool for management. However, we have recently encountered several issues with the current session pool implementation—most of which stem from its improper usage.
+
+For example, in issue [#56934](https://github.com/pingcap/tidb/issues/56934), the TTL module retrieves a session from the pool but fails to return it after use, leading to a memory leak. Another issue arose after PR [#32726](https://github.com/pingcap/tidb/pull/32726), where each session retrieved from the pool is stored as a reference in a map. This reference is only removed when the session is returned to the pool. As a result, if a session is closed using the `Close` method, it may still cause a memory leak because it remains referenced in the map.
+
+Some PR such as [#59546](https://github.com/pingcap/tidb/pull/59546) attempt to address these issues by introducing a `Destroy` method for the session pool. However, this approach does not fully resolve the problems and still makes it difficult to ensure that developers call the correct method.
+
+This document proposes a more effective solution for managing internal system sessions to achieve the following goals:
+
+- Prevent concurrent usage across modules: Once a session is returned to the pool, it should no longer be used to avoid potential conflicts from concurrent access.
+- Simplify session lifecycle management: Ensure that sessions are always either returned to the pool or properly closed after use, eliminating the risk of memory leaks.
+- Enforce session state reset: When a session is returned to the pool, its internal state should be forcibly reset to prevent unexpected behaviors in subsequent usage.
+
+## Design
+
+### Pool
+
+We need to introduce a new enhanced session pool to manage internal system sessions. The new session pool will provide the following methods:
+
+```go
+type Pool struct {
+    ...
+}
+
+func (p *Pool) get() (*Session, error) {
+    ...
+}
+
+func (p *Pool) put(s *Session) {
+    ...
+}
+
+func (p *Pool) WithSession(func(*Session) error) error {
+    ...
+}
+```
+
+The most important method of `Pool` is `WithSession` which is introduced to replace the `Get` and `Put` methods of the current session pool. The `WithSession` method accepts a function as an argument, which will be executed with a session retrieved from the pool. After the function completes, the session will be automatically returned to the pool. This design ensures that the session is always returned to the pool after use, preventing memory leaks.
+
+Let's see how `WithSession` works:
+
+```go
+func (p *Pool) WithSession(fn func(*Session) error) error {
+    se, err := p.get()
+    if err != nil {
+        return err
+    }
+
+    success := false
+    defer func() {
+        if success {
+            p.put(se)
+        } else {
+            se.Close()
+        }
+    }()
+
+    if err = fn(se); err != nil {
+        return err
+    }
+    success = true
+    return nil
+}
+```
+
+We can see that the `WithSession` method always calls `p.put(se)` within a defer function to ensure the session is returned to the pool. Some exceptions are when a panic or error happens. In such cases, the session is destroyed instead of being returned to the pool, as it may be in an undefined state, making it safer to discard rather than reuse.
+
+The `Pool` should be responsible for managing the internal sessions' registration and unregistering, and the callers should not care about it. These operations are handled by `Pool.get` and `Pool.put` methods:
+
+```go
+func (p *Pool) get() (*Session, error) {
+    // get a session from the pool
+    se, err := getFromPool()
+    if err != nil {
+        return nil, err
+    }
+    // register the internal session to the map
+    infosync.StoreInternalSession(se.GetSessionContext())
+    return se, nil
+}
+
+func (p *Pool) put(s *Session) {
+    // rollback the session first to reset the internal state.
+    s.Rollback()
+    // unregister the internal session from the map
+    infosync.DeleteInternalSession(s.GetSessionContext())
+    // put the session back to the pool
+    p.putToPool(s)
+}
+```
+
+The above code has been simplified for demonstration purposes. The session pool stores the internal session in a map when `Pool.get` is called and removes it from the map in `Pool.put`. Additionally, before being returned to the pool, the session is automatically rolled back to ensure its internal state is reset.
+
+With the newly implemented session pool that manages the session lifecycle, developers no longer need to worry about the session's state or lifecycle. 
+
+### Session
+
+The type `Session` is not a "real" one but a wrapper of the internal session, see:
+
+```go
+// Session is public to callers.
+type Session struct {
+    internal *session
+}
+```
+
+It provides a set of methods that proxy access to the internal session. When the caller invokes `WithSession`, a new `Session` instance is created and passed to the function. Once the function completes, the `Session` instance is destroyed, and only the internal session is returned to the pool.
+
+We have several reasons for proposing the above design:
+
+- The wrapped methods allow for additional checks to prevent misuse of the internal session. For example, if a caller attempts to invoke methods after the session has been returned to the pool, we can detect this and return an error.
+- The `Session` type can be extended with new methods, providing more convenient functionality for internal use.
+- The original session contains numerous methods, many of which may not be necessary for the caller. By filtering out unnecessary methods, we can expose only the required ones, simplifying the interface and improving usability.
+
+The internal session is not exposed to the caller and can only be accessed by the `Session` or `Pool` types. It has several fields, see:
+
+```go
+// session is private and can only be accessed by the `Pool` and `Session`.
+type session struct {
+    mu sync.Mutex
+    owner any
+    sctx sessionctx.Context
+    ...
+}
+```
+
+The `mu` protects the internal session from concurrent access, and the field `sctx` is the real session context that holds the session's state. 
+
+Let's provide further clarification on the `owner` field. The "owner" refers to the entity that holds the internal session, and only the owner is permitted to use it. When an internal session is in the pool, the pool itself is considered the owner. When a caller retrieves the session from the pool, the wrapped `Session` becomes the new owner, which is then passed to the caller.
+
+When the caller returns the session to the pool, ownership is first transferred back to the pool, which then places the session back into the pool. Since the public `Session` will not be reused, once it loses ownership, it can never regain it. This ensures that the `Session` becomes invalid once it has been used, preventing any further use after its lifecycle is completed.
+
+```go
+func (s *session) checkOwner(owner any) error {
+    s.mu.Lock()
+    defer s.mu.Unlock()
+    if s.owner == nil || s.owner != owner {
+        return errors.New("invalid owner")
+    }
+    return nil
+}
+
+func (s *Session) ExecuteInternal(ctx context.Context, sql string, args ...any) (sqlexec.RecordSet, error) {
+    if err := s.internal.checkOwner(s); err != nil {
+        return nil, err
+    }
+    return s.internal.sctx.ExecuteInternal(ctx, sql, args...)
+}
+```
+
+## Tests
+
+We should do the below tests to make sure the new introduced session pool works as expected:
+
+- Unit tests.
+- Integration tests for specified scenarios:
+  - Test 1000+ TTL tables with small `tidb_ttl_job_internal` settings to test the frequent session creation and destruction in the TTL scene.
+  - Test async load with corrupted statistics to ensure there are no memory leaks.
+  - More tests for other scenarios.
+
+## More Features to Support
+
+There are also some features we can support in the future:
+
+- Support to detect concurrent usage of the un-thread-safe methods.
+- Support to detect more session state changes such as system variables and reset them when returning to the pool.
diff --git a/pkg/bindinfo/binding_operator.go b/pkg/bindinfo/binding_operator.go
index feaf12f607..f5cbe4b8b1 100644
--- a/pkg/bindinfo/binding_operator.go
+++ b/pkg/bindinfo/binding_operator.go
@@ -97,6 +97,13 @@ func (op *bindingOperator) CreateBinding(sctx sessionctx.Context, bindings []*Bi
 			// overflow directly.
 
 			// Insert the Bindings to the storage.
+			var sqlDigest, planDigest any // null by default
+			if binding.SQLDigest != "" {
+				sqlDigest = binding.SQLDigest
+			}
+			if binding.PlanDigest != "" {
+				planDigest = binding.PlanDigest
+			}
 			_, err = exec(
 				sctx,
 				`INSERT INTO mysql.bind_info VALUES (%?,%?, %?, %?, %?, %?, %?, %?, %?, %?, %?)`,
@@ -109,8 +116,8 @@ func (op *bindingOperator) CreateBinding(sctx sessionctx.Context, bindings []*Bi
 				binding.Charset,
 				binding.Collation,
 				binding.Source,
-				binding.SQLDigest,
-				binding.PlanDigest,
+				sqlDigest,
+				planDigest,
 			)
 			failpoint.Inject("CreateGlobalBindingNthFail", func(val failpoint.Value) {
 				n := val.(int)
diff --git a/pkg/executor/importer/import.go b/pkg/executor/importer/import.go
index cbe41c3cd1..d188866fc3 100644
--- a/pkg/executor/importer/import.go
+++ b/pkg/executor/importer/import.go
@@ -83,20 +83,30 @@ const (
 	// 0 means no limit
 	unlimitedWriteSpeed = config.ByteSize(0)
 
-	characterSetOption          = "character_set"
-	fieldsTerminatedByOption    = "fields_terminated_by"
-	fieldsEnclosedByOption      = "fields_enclosed_by"
-	fieldsEscapedByOption       = "fields_escaped_by"
-	fieldsDefinedNullByOption   = "fields_defined_null_by"
-	linesTerminatedByOption     = "lines_terminated_by"
-	skipRowsOption              = "skip_rows"
-	splitFileOption             = "split_file"
-	diskQuotaOption             = "disk_quota"
-	threadOption                = "thread"
-	maxWriteSpeedOption         = "max_write_speed"
-	checksumTableOption         = "checksum_table"
-	recordErrorsOption          = "record_errors"
-	detachedOption              = "detached"
+	characterSetOption        = "character_set"
+	fieldsTerminatedByOption  = "fields_terminated_by"
+	fieldsEnclosedByOption    = "fields_enclosed_by"
+	fieldsEscapedByOption     = "fields_escaped_by"
+	fieldsDefinedNullByOption = "fields_defined_null_by"
+	linesTerminatedByOption   = "lines_terminated_by"
+	skipRowsOption            = "skip_rows"
+	splitFileOption           = "split_file"
+	diskQuotaOption           = "disk_quota"
+	threadOption              = "thread"
+	maxWriteSpeedOption       = "max_write_speed"
+	checksumTableOption       = "checksum_table"
+	recordErrorsOption        = "record_errors"
+	detachedOption            = "detached"
+	// if 'import mode' enabled, TiKV will:
+	//  - set level0_stop_writes_trigger = max(old, 1 << 30)
+	//  - set level0_slowdown_writes_trigger = max(old, 1 << 30)
+	//  - set soft_pending_compaction_bytes_limit = 0,
+	//  - set hard_pending_compaction_bytes_limit = 0,
+	//  - will not trigger flow control when SST count in L0 is large
+	//  - will not trigger region split, it might cause some region became
+	//    very large and be a hotspot, might cause latency spike.
+	//
+	// default false for local sort, true for global sort.
 	disableTiKVImportModeOption = "disable_tikv_import_mode"
 	cloudStorageURIOption       = "cloud_storage_uri"
 	disablePrecheckOption       = "disable_precheck"
@@ -792,6 +802,9 @@ func (p *Plan) adjustOptions(targetNodeCPUCnt int) {
 			zap.Int("before", p.ThreadCnt), zap.Int("after", limit))
 		p.ThreadCnt = limit
 	}
+	if p.IsGlobalSort() {
+		p.DisableTiKVImportMode = true
+	}
 }
 
 func (p *Plan) initParameters(plan *plannercore.ImportInto) error {
diff --git a/pkg/session/BUILD.bazel b/pkg/session/BUILD.bazel
index b684099d16..11d003abe3 100644
--- a/pkg/session/BUILD.bazel
+++ b/pkg/session/BUILD.bazel
@@ -155,7 +155,6 @@ go_test(
     shard_count = 50,
     deps = [
         "//pkg/autoid_service",
-        "//pkg/bindinfo",
         "//pkg/config",
         "//pkg/ddl",
         "//pkg/domain",
@@ -165,7 +164,6 @@ go_test(
         "//pkg/keyspace",
         "//pkg/kv",
         "//pkg/meta",
-        "//pkg/parser",
         "//pkg/parser/ast",
         "//pkg/parser/auth",
         "//pkg/session/types",
diff --git a/pkg/session/bootstrap.go b/pkg/session/bootstrap.go
index a0f4971310..63223c7bf7 100644
--- a/pkg/session/bootstrap.go
+++ b/pkg/session/bootstrap.go
@@ -304,10 +304,11 @@ const (
 		charset TEXT NOT NULL,
 		collation TEXT NOT NULL,
 		source VARCHAR(10) NOT NULL DEFAULT 'unknown',
-		sql_digest varchar(64),
-		plan_digest varchar(64),
+		sql_digest varchar(64) DEFAULT NULL,
+		plan_digest varchar(64) DEFAULT NULL,
 		INDEX sql_index(original_sql(700),default_db(68)) COMMENT "accelerate the speed when add global binding query",
-		INDEX time_index(update_time) COMMENT "accelerate the speed when querying with last update time"
+		INDEX time_index(update_time) COMMENT "accelerate the speed when querying with last update time",
+		UNIQUE INDEX digest_index(plan_digest, sql_digest) COMMENT "avoid duplicated records"
 	) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;`
 
 	// CreateRoleEdgesTable stores the role and user relationship information.
@@ -1275,11 +1276,14 @@ const (
 
 	// version245 updates column types of mysql.bind_info.
 	version245 = 245
+
+	// version246 adds new unique index for mysql.bind_info.
+	version246 = 246
 )
 
 // currentBootstrapVersion is defined as a variable, so we can modify its value for testing.
 // please make sure this is the largest version
-var currentBootstrapVersion int64 = version245
+var currentBootstrapVersion int64 = version246
 
 // DDL owner key's expired time is ManagerSessionTTL seconds, we should wait the time and give more time to have a chance to finish it.
 var internalSQLTimeout = owner.ManagerSessionTTL + 15
@@ -1460,6 +1464,7 @@ var (
 		upgradeToVer243,
 		upgradeToVer244,
 		upgradeToVer245,
+		upgradeToVer246,
 	}
 )
 
@@ -3406,6 +3411,60 @@ func upgradeToVer245(s sessiontypes.Session, ver int64) {
 	doReentrantDDL(s, "ALTER TABLE mysql.bind_info MODIFY COLUMN bind_sql LONGTEXT NOT NULL")
 }
 
+func upgradeToVer246(s sessiontypes.Session, ver int64) {
+	if ver >= version246 {
+		return
+	}
+
+	// log duplicated digests that will be set to null.
+	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnBootstrap)
+	rs, err := s.ExecuteInternal(ctx,
+		`select plan_digest, sql_digest from mysql.bind_info group by plan_digest, sql_digest having count(1) > 1`)
+	if err != nil {
+		logutil.BgLogger().Fatal("failed to get duplicated plan and sql digests", zap.Error(err))
+		return
+	}
+	req := rs.NewChunk(nil)
+	duplicatedDigests := make(map[string]struct{})
+	for {
+		err = rs.Next(ctx, req)
+		if err != nil {
+			logutil.BgLogger().Fatal("failed to get duplicated plan and sql digests", zap.Error(err))
+			return
+		}
+		if req.NumRows() == 0 {
+			break
+		}
+		for i := 0; i < req.NumRows(); i++ {
+			planDigest, sqlDigest := req.GetRow(i).GetString(0), req.GetRow(i).GetString(1)
+			duplicatedDigests[sqlDigest+", "+planDigest] = struct{}{}
+		}
+		req.Reset()
+	}
+	if err := rs.Close(); err != nil {
+		logutil.BgLogger().Warn("failed to close record set", zap.Error(err))
+	}
+	if len(duplicatedDigests) > 0 {
+		digestList := make([]string, 0, len(duplicatedDigests))
+		for k := range duplicatedDigests {
+			digestList = append(digestList, "("+k+")")
+		}
+		logutil.BgLogger().Warn("set the following (plan digest, sql digest) in mysql.bind_info to null " +
+			"for adding new unique index: " + strings.Join(digestList, ", "))
+	}
+
+	// to avoid the failure of adding the unique index, remove duplicated rows on these 2 digest columns first.
+	// in most cases, there should be no duplicated rows, since now we only store one binding for each sql_digest.
+	// compared with upgrading failure, it's OK to set these 2 columns to null.
+	doReentrantDDL(s, `UPDATE mysql.bind_info SET plan_digest=null, sql_digest=null
+                       WHERE (plan_digest, sql_digest) in (
+                         select plan_digest, sql_digest from mysql.bind_info
+                         group by plan_digest, sql_digest having count(1) > 1)`)
+	doReentrantDDL(s, "ALTER TABLE mysql.bind_info MODIFY COLUMN sql_digest VARCHAR(64) DEFAULT NULL")
+	doReentrantDDL(s, "ALTER TABLE mysql.bind_info MODIFY COLUMN plan_digest VARCHAR(64) DEFAULT NULL")
+	doReentrantDDL(s, "ALTER TABLE mysql.bind_info ADD UNIQUE INDEX digest_index(plan_digest, sql_digest)", dbterror.ErrDupKeyName)
+}
+
 // initGlobalVariableIfNotExists initialize a global variable with specific val if it does not exist.
 func initGlobalVariableIfNotExists(s sessiontypes.Session, name string, val any) {
 	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnBootstrap)
diff --git a/pkg/sessionctx/variable/BUILD.bazel b/pkg/sessionctx/variable/BUILD.bazel
index 29d5e6050d..436e260726 100644
--- a/pkg/sessionctx/variable/BUILD.bazel
+++ b/pkg/sessionctx/variable/BUILD.bazel
@@ -104,6 +104,7 @@ go_test(
     shard_count = 50,
     deps = [
         "//pkg/config",
+        "//pkg/executor/join/joinversion",
         "//pkg/kv",
         "//pkg/parser",
         "//pkg/parser/auth",
diff --git a/pkg/sessionctx/variable/session.go b/pkg/sessionctx/variable/session.go
index 00def0d9c9..0ba9ab9025 100644
--- a/pkg/sessionctx/variable/session.go
+++ b/pkg/sessionctx/variable/session.go
@@ -35,6 +35,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
 	"github.com/pingcap/tidb/pkg/config"
+	"github.com/pingcap/tidb/pkg/executor/join/joinversion"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/metrics"
@@ -2256,6 +2257,7 @@ func NewSessionVars(hctx HookContext) *SessionVars {
 	vars.MemTracker.IsRootTrackerOfSess = true
 	vars.MemTracker.Killer = &vars.SQLKiller
 	vars.StatsLoadSyncWait.Store(vardef.StatsLoadSyncWait.Load())
+	vars.UseHashJoinV2 = joinversion.IsOptimizedVersion(vardef.DefTiDBHashJoinVersion)
 
 	for _, engine := range config.GetGlobalConfig().IsolationRead.Engines {
 		switch engine {


Test patch:
diff --git a/br/pkg/restore/snap_client/systable_restore_test.go b/br/pkg/restore/snap_client/systable_restore_test.go
index 4797d6c7fd..a1e8371c64 100644
--- a/br/pkg/restore/snap_client/systable_restore_test.go
+++ b/br/pkg/restore/snap_client/systable_restore_test.go
@@ -117,5 +117,5 @@ func TestCheckSysTableCompatibility(t *testing.T) {
 
 // The above variables are in the file br/pkg/restore/systable_restore.go
 func TestMonitorTheSystemTableIncremental(t *testing.T) {
-	require.Equal(t, int64(245), session.CurrentBootstrapVersion)
+	require.Equal(t, int64(246), session.CurrentBootstrapVersion)
 }
diff --git a/pkg/bindinfo/binding_operator_test.go b/pkg/bindinfo/binding_operator_test.go
index 6f9ce784ce..d53e8e8ae0 100644
--- a/pkg/bindinfo/binding_operator_test.go
+++ b/pkg/bindinfo/binding_operator_test.go
@@ -260,8 +260,6 @@ func TestSetBindingStatusWithoutBindingInCache(t *testing.T) {
 
 	// Simulate creating bindings on other machines
 	_, sqlDigest := parser.NormalizeDigestForBinding("select * from `test` . `t` where `a` > ?")
-	tk.MustExec("insert into mysql.bind_info values('select * from `test` . `t` where `a` > ?', 'SELECT /*+ USE_INDEX(`t` `idx_a`)*/ * FROM `test`.`t` WHERE `a` > 10', 'test', 'deleted', '2000-01-01 09:00:00', '2000-01-01 09:00:00', '', '','" +
-		bindinfo.SourceManual + "', '" + sqlDigest.String() + "', '')")
 	tk.MustExec("insert into mysql.bind_info values('select * from `test` . `t` where `a` > ?', 'SELECT /*+ USE_INDEX(`t` `idx_a`)*/ * FROM `test`.`t` WHERE `a` > 10', 'test', 'enabled', '2000-01-02 09:00:00', '2000-01-02 09:00:00', '', '','" +
 		bindinfo.SourceManual + "', '" + sqlDigest.String() + "', '')")
 	tk.MustExec("set binding disabled for select * from t where a > 10")
@@ -274,8 +272,6 @@ func TestSetBindingStatusWithoutBindingInCache(t *testing.T) {
 	utilCleanBindingEnv(tk)
 
 	// Simulate creating bindings on other machines
-	tk.MustExec("insert into mysql.bind_info values('select * from `test` . `t` where `a` > ?', 'SELECT * FROM `test`.`t` WHERE `a` > 10', 'test', 'deleted', '2000-01-01 09:00:00', '2000-01-01 09:00:00', '', '','" +
-		bindinfo.SourceManual + "', '" + sqlDigest.String() + "', '')")
 	tk.MustExec("insert into mysql.bind_info values('select * from `test` . `t` where `a` > ?', 'SELECT * FROM `test`.`t` WHERE `a` > 10', 'test', 'disabled', '2000-01-02 09:00:00', '2000-01-02 09:00:00', '', '','" +
 		bindinfo.SourceManual + "', '" + sqlDigest.String() + "', '')")
 	tk.MustExec("set binding enabled for select * from t where a > 10")
diff --git a/pkg/executor/importer/import_test.go b/pkg/executor/importer/import_test.go
index 8dae7e46eb..e82bf6d9aa 100644
--- a/pkg/executor/importer/import_test.go
+++ b/pkg/executor/importer/import_test.go
@@ -189,11 +189,17 @@ func TestAdjustOptions(t *testing.T) {
 	plan.adjustOptions(16)
 	require.Equal(t, 16, plan.ThreadCnt)
 	require.Equal(t, config.ByteSize(10), plan.MaxWriteSpeed) // not adjusted
+	require.False(t, plan.DisableTiKVImportMode)
 
 	plan.ThreadCnt = 100000000
 	plan.DataSourceType = DataSourceTypeQuery
 	plan.adjustOptions(16)
 	require.Equal(t, 32, plan.ThreadCnt)
+	require.False(t, plan.DisableTiKVImportMode)
+
+	plan.CloudStorageURI = "s3://bucket/path"
+	plan.adjustOptions(16)
+	require.True(t, plan.DisableTiKVImportMode)
 }
 
 func TestAdjustDiskQuota(t *testing.T) {
diff --git a/pkg/session/bootstrap_test.go b/pkg/session/bootstrap_test.go
index 8bb4ecb157..503afc70c0 100644
--- a/pkg/session/bootstrap_test.go
+++ b/pkg/session/bootstrap_test.go
@@ -18,7 +18,6 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
-	"sort"
 	"strings"
 	"testing"
 	"time"
@@ -26,14 +25,12 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/keyspacepb"
 	"github.com/pingcap/log"
-	"github.com/pingcap/tidb/pkg/bindinfo"
 	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/expression/sessionexpr"
 	"github.com/pingcap/tidb/pkg/keyspace"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/meta"
-	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/auth"
 	sessiontypes "github.com/pingcap/tidb/pkg/session/types"
 	"github.com/pingcap/tidb/pkg/sessionctx"
@@ -588,53 +585,6 @@ func TestStmtSummary(t *testing.T) {
 	require.NoError(t, r.Close())
 }
 
-func TestUpdateDuplicateBindInfo(t *testing.T) {
-	ctx := context.Background()
-	store, dom := CreateStoreAndBootstrap(t)
-	defer func() { require.NoError(t, store.Close()) }()
-	defer dom.Close()
-	se := CreateSessionAndSetID(t, store)
-	MustExec(t, se, "alter table mysql.bind_info drop column if exists plan_digest")
-	MustExec(t, se, "alter table mysql.bind_info drop column if exists sql_digest")
-
-	MustExec(t, se, `insert into mysql.bind_info values('select * from t', 'select /*+ use_index(t, idx_a)*/ * from t', 'test', 'enabled', '2021-01-04 14:50:58.257', '2021-01-04 14:50:58.257', 'utf8', 'utf8_general_ci', 'manual')`)
-	// The latest one.
-	MustExec(t, se, `insert into mysql.bind_info values('select * from test . t', 'select /*+ use_index(t, idx_b)*/ * from test.t', 'test', 'enabled', '2021-01-04 14:50:58.257', '2021-01-09 14:50:58.257', 'utf8', 'utf8_general_ci', 'manual')`)
-
-	MustExec(t, se, `insert into mysql.bind_info values('select * from t where a < ?', 'select * from t use index(idx) where a < 1', 'test', 'deleted', '2021-06-04 17:04:43.333', '2021-06-04 17:04:43.335', 'utf8', 'utf8_general_ci', 'manual')`)
-	MustExec(t, se, `insert into mysql.bind_info values('select * from t where a < ?', 'select * from t ignore index(idx) where a < 1', 'test', 'enabled', '2021-06-04 17:04:43.335', '2021-06-04 17:04:43.335', 'utf8', 'utf8_general_ci', 'manual')`)
-	MustExec(t, se, `insert into mysql.bind_info values('select * from test . t where a <= ?', 'select * from test.t use index(idx) where a <= 1', '', 'deleted', '2021-06-04 17:04:43.345', '2021-06-04 17:04:45.334', 'utf8', 'utf8_general_ci', 'manual')`)
-	MustExec(t, se, `insert into mysql.bind_info values('select * from test . t where a <= ?', 'select * from test.t ignore index(idx) where a <= 1', '', 'enabled', '2021-06-04 17:04:45.334', '2021-06-04 17:04:45.334', 'utf8', 'utf8_general_ci', 'manual')`)
-
-	upgradeToVer67(se, version66)
-
-	r := MustExecToRecodeSet(t, se, `select original_sql, bind_sql, default_db, status, create_time from mysql.bind_info where source != 'builtin' order by create_time`)
-	req := r.NewChunk(nil)
-	require.NoError(t, r.Next(ctx, req))
-	require.Equal(t, 3, req.NumRows())
-	row := req.GetRow(0)
-	require.Equal(t, "select * from `test` . `t`", row.GetString(0))
-	require.Equal(t, "SELECT /*+ use_index(`t` `idx_b`)*/ * FROM `test`.`t`", row.GetString(1))
-	require.Equal(t, "", row.GetString(2))
-	require.Equal(t, bindinfo.StatusEnabled, row.GetString(3))
-	require.Equal(t, "2021-01-04 14:50:58.257", row.GetTime(4).String())
-	row = req.GetRow(1)
-	require.Equal(t, "select * from `test` . `t` where `a` < ?", row.GetString(0))
-	require.Equal(t, "SELECT * FROM `test`.`t` IGNORE INDEX (`idx`) WHERE `a` < 1", row.GetString(1))
-	require.Equal(t, "", row.GetString(2))
-	require.Equal(t, bindinfo.StatusEnabled, row.GetString(3))
-	require.Equal(t, "2021-06-04 17:04:43.335", row.GetTime(4).String())
-	row = req.GetRow(2)
-	require.Equal(t, "select * from `test` . `t` where `a` <= ?", row.GetString(0))
-	require.Equal(t, "SELECT * FROM `test`.`t` IGNORE INDEX (`idx`) WHERE `a` <= 1", row.GetString(1))
-	require.Equal(t, "", row.GetString(2))
-	require.Equal(t, bindinfo.StatusEnabled, row.GetString(3))
-	require.Equal(t, "2021-06-04 17:04:45.334", row.GetTime(4).String())
-
-	require.NoError(t, r.Close())
-	MustExec(t, se, "delete from mysql.bind_info where original_sql = 'select * from test . t'")
-}
-
 func TestUpgradeClusteredIndexDefaultValue(t *testing.T) {
 	store, dom := CreateStoreAndBootstrap(t)
 	defer func() { require.NoError(t, store.Close()) }()
@@ -922,33 +872,6 @@ func testIndexMergeUpgradeFrom400To540(t *testing.T, enable bool) {
 	}
 }
 
-func TestUpgradeToVer85(t *testing.T) {
-	ctx := context.Background()
-	store, dom := CreateStoreAndBootstrap(t)
-	defer func() { require.NoError(t, store.Close()) }()
-	defer dom.Close()
-	se := CreateSessionAndSetID(t, store)
-	MustExec(t, se, "alter table mysql.bind_info drop column if exists plan_digest")
-	MustExec(t, se, "alter table mysql.bind_info drop column if exists sql_digest")
-
-	MustExec(t, se, `insert into mysql.bind_info values('select * from t', 'select /*+ use_index(t, idx_a)*/ * from t', 'test', 'using', '2021-01-04 14:50:58.257', '2021-01-04 14:50:58.257', 'utf8', 'utf8_general_ci', 'manual')`)
-	MustExec(t, se, `insert into mysql.bind_info values('select * from t1', 'select /*+ use_index(t1, idx_a)*/ * from t1', 'test', 'enabled', '2021-01-05 14:50:58.257', '2021-01-05 14:50:58.257', 'utf8', 'utf8_general_ci', 'manual')`)
-	MustExec(t, se, `insert into mysql.bind_info values('select * from t2', 'select /*+ use_index(t2, idx_a)*/ * from t2', 'test', 'disabled', '2021-01-06 14:50:58.257', '2021-01-06 14:50:58.257', 'utf8', 'utf8_general_ci', 'manual')`)
-	MustExec(t, se, `insert into mysql.bind_info values('select * from t3', 'select /*+ use_index(t3, idx_a)*/ * from t3', 'test', 'deleted', '2021-01-07 14:50:58.257', '2021-01-07 14:50:58.257', 'utf8', 'utf8_general_ci', 'manual')`)
-	MustExec(t, se, `insert into mysql.bind_info values('select * from t4', 'select /*+ use_index(t4, idx_a)*/ * from t4', 'test', 'invalid', '2021-01-08 14:50:58.257', '2021-01-08 14:50:58.257', 'utf8', 'utf8_general_ci', 'manual')`)
-	upgradeToVer85(se, version84)
-
-	r := MustExecToRecodeSet(t, se, `select count(*) from mysql.bind_info where status = 'enabled'`)
-	req := r.NewChunk(nil)
-	require.NoError(t, r.Next(ctx, req))
-	require.Equal(t, 1, req.NumRows())
-	row := req.GetRow(0)
-	require.Equal(t, int64(2), row.GetInt64(0))
-
-	require.NoError(t, r.Close())
-	MustExec(t, se, "delete from mysql.bind_info where default_db = 'test'")
-}
-
 func TestTiDBEnablePagingVariable(t *testing.T) {
 	store, dom := CreateStoreAndBootstrap(t)
 	se := CreateSessionAndSetID(t, store)
@@ -2005,89 +1928,6 @@ func TestTiDBUpgradeToVer170(t *testing.T) {
 	dom.Close()
 }
 
-func TestTiDBBindingInListToVer175(t *testing.T) {
-	ctx := context.Background()
-	store, dom := CreateStoreAndBootstrap(t)
-	defer func() { require.NoError(t, store.Close()) }()
-
-	// bootstrap as version174
-	ver174 := version174
-	seV174 := CreateSessionAndSetID(t, store)
-	txn, err := store.Begin()
-	require.NoError(t, err)
-	m := meta.NewMutator(txn)
-	err = m.FinishBootstrap(int64(ver174))
-	require.NoError(t, err)
-	revertVersionAndVariables(t, seV174, ver174)
-	err = txn.Commit(context.Background())
-	require.NoError(t, err)
-	store.SetOption(StoreBootstrappedKey, nil)
-
-	// create some bindings at version174
-	MustExec(t, seV174, "use test")
-	MustExec(t, seV174, "create table t (a int, b int, c int, key(c))")
-	_, digest := parser.NormalizeDigestForBinding("SELECT * FROM `test`.`t` WHERE `a` IN (1,2,3)")
-	MustExec(t, seV174, fmt.Sprintf("insert into mysql.bind_info values ('select * from `test` . `t` where `a` in ( ... )', 'SELECT /*+ use_index(`t` `c`)*/ * FROM `test`.`t` WHERE `a` IN (1,2,3)', 'test', 'enabled', '2023-09-13 14:41:38.319', '2023-09-13 14:41:35.319', 'utf8', 'utf8_general_ci', 'manual', '%s', '')", digest.String()))
-	_, digest = parser.NormalizeDigestForBinding("SELECT * FROM `test`.`t` WHERE `a` IN (1)")
-	MustExec(t, seV174, fmt.Sprintf("insert into mysql.bind_info values ('select * from `test` . `t` where `a` in ( ? )', 'SELECT /*+ use_index(`t` `c`)*/ * FROM `test`.`t` WHERE `a` IN (1)', 'test', 'enabled', '2023-09-13 14:41:38.319', '2023-09-13 14:41:36.319', 'utf8', 'utf8_general_ci', 'manual', '%s', '')", digest.String()))
-	_, digest = parser.NormalizeDigestForBinding("SELECT * FROM `test`.`t` WHERE `a` IN (1) AND `b` IN (1,2,3)")
-	MustExec(t, seV174, fmt.Sprintf("insert into mysql.bind_info values ('select * from `test` . `t` where `a` in ( ? ) and `b` in ( ... )', 'SELECT /*+ use_index(`t` `c`)*/ * FROM `test`.`t` WHERE `a` IN (1) AND `b` IN (1,2,3)', 'test', 'enabled', '2023-09-13 14:41:37.319', '2023-09-13 14:41:38.319', 'utf8', 'utf8_general_ci', 'manual', '%s', '')", digest.String()))
-
-	showBindings := func(s sessiontypes.Session) (records []string) {
-		MustExec(t, s, "admin reload bindings")
-		res := MustExecToRecodeSet(t, s, "show global bindings")
-		chk := res.NewChunk(nil)
-		for {
-			require.NoError(t, res.Next(ctx, chk))
-			if chk.NumRows() == 0 {
-				break
-			}
-			for i := 0; i < chk.NumRows(); i++ {
-				originalSQL := chk.GetRow(i).GetString(0)
-				bindSQL := chk.GetRow(i).GetString(1)
-				records = append(records, fmt.Sprintf("%s:%s", bindSQL, originalSQL))
-			}
-		}
-		require.NoError(t, res.Close())
-		sort.Strings(records)
-		return
-	}
-	bindings := showBindings(seV174)
-	// on ver174, `in (1)` and `in (1,2,3)` have different normalized results: `in (?)` and `in (...)`
-	require.Equal(t, []string{"SELECT /*+ use_index(`t` `c`)*/ * FROM `test`.`t` WHERE `a` IN (1) AND `b` IN (1,2,3):select * from `test` . `t` where `a` in ( ? ) and `b` in ( ... )",
-		"SELECT /*+ use_index(`t` `c`)*/ * FROM `test`.`t` WHERE `a` IN (1):select * from `test` . `t` where `a` in ( ? )"}, bindings)
-
-	// upgrade to ver175
-	dom.Close()
-	domCurVer, err := BootstrapSession(store)
-	require.NoError(t, err)
-	defer domCurVer.Close()
-	seCurVer := CreateSessionAndSetID(t, store)
-	ver, err := getBootstrapVersion(seCurVer)
-	require.NoError(t, err)
-	require.Equal(t, currentBootstrapVersion, ver)
-
-	// `in (?)` becomes to `in ( ... )`
-	bindings = showBindings(seCurVer)
-	require.Equal(t, []string{"SELECT /*+ use_index(`t` `c`)*/ * FROM `test`.`t` WHERE `a` IN (1) AND `b` IN (1,2,3):select * from `test` . `t` where `a` in ( ... ) and `b` in ( ... )",
-		"SELECT /*+ use_index(`t` `c`)*/ * FROM `test`.`t` WHERE `a` IN (1):select * from `test` . `t` where `a` in ( ... )"}, bindings)
-
-	planFromBinding := func(s sessiontypes.Session, q string) {
-		MustExec(t, s, q)
-		res := MustExecToRecodeSet(t, s, "select @@last_plan_from_binding")
-		chk := res.NewChunk(nil)
-		require.NoError(t, res.Next(ctx, chk))
-		require.Equal(t, int64(1), chk.GetRow(0).GetInt64(0))
-		require.NoError(t, res.Close())
-	}
-	planFromBinding(seCurVer, "select * from test.t where a in (1)")
-	planFromBinding(seCurVer, "select * from test.t where a in (1,2,3)")
-	planFromBinding(seCurVer, "select * from test.t where a in (1,2,3,4,5,6,7)")
-	planFromBinding(seCurVer, "select * from test.t where a in (1,2,3,4,5,6,7) and b in(1)")
-	planFromBinding(seCurVer, "select * from test.t where a in (1,2,3,4,5,6,7) and b in(1,2,3,4)")
-	planFromBinding(seCurVer, "select * from test.t where a in (7) and b in(1,2,3,4)")
-}
-
 func TestTiDBUpgradeToVer176(t *testing.T) {
 	store, do := CreateStoreAndBootstrap(t)
 	defer func() {
@@ -2761,3 +2601,46 @@ func TestWriteClusterIDToMySQLTiDBWhenUpgradingTo242(t *testing.T) {
 	require.NoError(t, r.Close())
 	seCurVer.Close()
 }
+
+func TestBindInfoUniqueIndex(t *testing.T) {
+	ctx := context.Background()
+	store, dom := CreateStoreAndBootstrap(t)
+	defer func() { require.NoError(t, store.Close()) }()
+
+	// bootstrap as version245
+	ver245 := version245
+	seV245 := CreateSessionAndSetID(t, store)
+	txn, err := store.Begin()
+	require.NoError(t, err)
+	m := meta.NewMutator(txn)
+	err = m.FinishBootstrap(int64(ver245))
+	require.NoError(t, err)
+	revertVersionAndVariables(t, seV245, ver245)
+	err = txn.Commit(ctx)
+	require.NoError(t, err)
+	store.SetOption(StoreBootstrappedKey, nil)
+
+	// remove the unique index on mysql.bind_info for testing
+	MustExec(t, seV245, "alter table mysql.bind_info drop index digest_index")
+
+	// insert duplicated values into mysql.bind_info
+	for _, sqlDigest := range []string{"null", "'x'", "'y'"} {
+		for _, planDigest := range []string{"null", "'x'", "'y'"} {
+			insertStmt := fmt.Sprintf(`insert into mysql.bind_info values (
+             "sql", "bind_sql", "db", "disabled", NOW(), NOW(), "", "", "", %s, %s)`,
+				sqlDigest, planDigest)
+			MustExec(t, seV245, insertStmt)
+			MustExec(t, seV245, insertStmt)
+		}
+	}
+
+	// upgrade to current version
+	dom.Close()
+	domCurVer, err := BootstrapSession(store)
+	require.NoError(t, err)
+	defer domCurVer.Close()
+	seCurVer := CreateSessionAndSetID(t, store)
+	ver, err := getBootstrapVersion(seCurVer)
+	require.NoError(t, err)
+	require.Equal(t, currentBootstrapVersion, ver)
+}
diff --git a/pkg/sessionctx/variable/sysvar_test.go b/pkg/sessionctx/variable/sysvar_test.go
index 15bbce098d..6391132ffb 100644
--- a/pkg/sessionctx/variable/sysvar_test.go
+++ b/pkg/sessionctx/variable/sysvar_test.go
@@ -29,6 +29,7 @@ import (
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/config"
+	"github.com/pingcap/tidb/pkg/executor/join/joinversion"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	"github.com/pingcap/tidb/pkg/sessionctx/vardef"
@@ -1797,6 +1798,8 @@ func TestEnableWindowFunction(t *testing.T) {
 
 func TestTiDBHashJoinVersion(t *testing.T) {
 	vars := NewSessionVars(nil)
+	// test the default value
+	require.Equal(t, joinversion.IsOptimizedVersion(vardef.DefTiDBHashJoinVersion), vars.UseHashJoinV2)
 	sv := GetSysVar(vardef.TiDBHashJoinVersion)
 	// set error value
 	_, err := sv.Validation(vars, "invalid", "invalid", vardef.ScopeSession)
diff --git a/tests/realtikvtest/importintotest3/from_server_test.go b/tests/realtikvtest/importintotest3/from_server_test.go
index d41e44afe2..544446e9d7 100644
--- a/tests/realtikvtest/importintotest3/from_server_test.go
+++ b/tests/realtikvtest/importintotest3/from_server_test.go
@@ -73,4 +73,5 @@ func (s *mockGCSSuite) TestImportFromServer() {
 	var taskMeta importinto.TaskMeta
 	require.NoError(s.T(), json.Unmarshal(task.Meta, &taskMeta))
 	require.Len(s.T(), taskMeta.ChunkMap, 2)
+	require.False(s.T(), taskMeta.Plan.DisableTiKVImportMode)
 }
diff --git a/tests/realtikvtest/importintotest4/global_sort_test.go b/tests/realtikvtest/importintotest4/global_sort_test.go
index 298a7b482a..c20a7f38e5 100644
--- a/tests/realtikvtest/importintotest4/global_sort_test.go
+++ b/tests/realtikvtest/importintotest4/global_sort_test.go
@@ -124,6 +124,7 @@ func (s *mockGCSSuite) TestGlobalSortBasic() {
 	taskMeta := importinto.TaskMeta{}
 	s.NoError(json.Unmarshal(task.Meta, &taskMeta))
 	urlEqual(s.T(), redactedSortStorageURI, taskMeta.Plan.CloudStorageURI)
+	require.True(s.T(), taskMeta.Plan.DisableTiKVImportMode)
 
 	// merge-sort data kv
 	s.tk.MustExec("truncate table t")