Repository: pingcap/tidb (go)

Issue:
DDL internal txn reports "GC life time is shorter than transaction duration"
## Bug Report

Please answer these questions before submitting your issue. Thanks!

### 1. Minimal reproduce step (Required)

1. Prepare a large dataset.
2. Create an index for a wide column.
3. It may take more than 10 min to finish the partial import.

<!-- a step by step guide for reproducing the bug. -->

### 2. What did you expect to see? (Required)

Adding index complete without errors.

### 3. What did you see instead (Required)

The state of adding index becomes "rollback done". 
```
GC life time is shorter than transaction duration
```


### 4. What is your TiDB version? (Required)

<!-- Paste the output of SELECT tidb_version() -->
b314a9c6037b


Pull request:
session: make `TxnInfo()` return even if process info is empty (#57044)
This is an automated cherry-pick of #57044

<!--

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 #57043

Problem Summary:

https://github.com/pingcap/tidb/blob/c240fdf085ce24c6110a6764e8b9657a3788abfe/pkg/session/session.go#L4026-L4030

`TxnInfo()` could be nil if process info is not set(see the implementation of `TxnInfo()`). Such sessions are commonly used in DDL/DXF:

https://github.com/pingcap/tidb/blob/c240fdf085ce24c6110a6764e8b9657a3788abfe/pkg/ddl/backfilling_operators.go#L533-L539

And they only invoke `session.Begin()` before using:

https://github.com/pingcap/tidb/blob/c240fdf085ce24c6110a6764e8b9657a3788abfe/pkg/ddl/index_cop.go#L43-L48

As a result, the `startTS` from internal DDL sessions will not be reported to etcd, and GC safepoint will be advanced unexpectedly.

### What changed and how does it work?

Make TxnInfo() return even if process info is empty.

### 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/Makefile b/Makefile
index 7e954a1973..5751165faf 100644
--- a/Makefile
+++ b/Makefile
@@ -257,7 +257,7 @@ tools/bin/revive:
 	GOBIN=$(shell pwd)/tools/bin $(GO) install github.com/mgechev/revive@v1.2.1
 
 tools/bin/failpoint-ctl:
-	GOBIN=$(shell pwd)/tools/bin $(GO) install github.com/pingcap/failpoint/failpoint-ctl@2eaa328
+	GOBIN=$(shell pwd)/tools/bin $(GO) install github.com/pingcap/failpoint/failpoint-ctl@9b3b6e3
 
 tools/bin/errdoc-gen:
 	GOBIN=$(shell pwd)/tools/bin $(GO) install github.com/pingcap/errors/errdoc-gen@518f63d
diff --git a/pkg/ddl/index_cop.go b/pkg/ddl/index_cop.go
index a158e1165b..fec8074623 100644
--- a/pkg/ddl/index_cop.go
+++ b/pkg/ddl/index_cop.go
@@ -179,11 +179,13 @@ func wrapInBeginRollback(se *sess.Session, f func(startTS uint64) error) error {
 		return errors.Trace(err)
 	}
 	defer se.Rollback()
-	var startTS uint64
-	sessVars := se.GetSessionVars()
-	sessVars.TxnCtxMu.Lock()
-	startTS = sessVars.TxnCtx.StartTS
-	sessVars.TxnCtxMu.Unlock()
+
+	txn, err := se.Txn()
+	if err != nil {
+		return err
+	}
+	startTS := txn.StartTS()
+	failpoint.InjectCall("wrapInBeginRollbackStartTS", startTS)
 	return f(startTS)
 }
 
diff --git a/pkg/ddl/ingest/BUILD.bazel b/pkg/ddl/ingest/BUILD.bazel
index c8a9f76646..178d84bc45 100644
--- a/pkg/ddl/ingest/BUILD.bazel
+++ b/pkg/ddl/ingest/BUILD.bazel
@@ -67,7 +67,7 @@ go_test(
     embed = [":ingest"],
     flaky = True,
     race = "on",
-    shard_count = 17,
+    shard_count = 18,
     deps = [
         "//pkg/config",
         "//pkg/ddl",
@@ -79,7 +79,6 @@ go_test(
         "//pkg/errno",
         "//pkg/parser/model",
         "//pkg/testkit",
-        "//tests/realtikvtest",
         "@com_github_ngaut_pools//:pools",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_stretchr_testify//assert",
diff --git a/pkg/ddl/internal/session/session_pool.go b/pkg/ddl/internal/session/session_pool.go
index ac10d872e7..10874d595c 100644
--- a/pkg/ddl/internal/session/session_pool.go
+++ b/pkg/ddl/internal/session/session_pool.go
@@ -15,6 +15,7 @@
 package session
 
 import (
+	"context"
 	"fmt"
 	"sync"
 
@@ -84,6 +85,7 @@ func (sg *Pool) Put(ctx sessionctx.Context) {
 
 	// no need to protect sg.resPool, even the sg.resPool is closed, the ctx still need to
 	// Put into resPool, because when resPool is closing, it will wait all the ctx returns, then resPool finish closing.
+	ctx.RollbackTxn(context.Background())
 	sg.resPool.Put(ctx.(pools.Resource))
 	infosync.DeleteInternalSession(ctx)
 }
diff --git a/pkg/executor/infoschema_reader.go b/pkg/executor/infoschema_reader.go
index 72c5305707..764ffde78a 100644
--- a/pkg/executor/infoschema_reader.go
+++ b/pkg/executor/infoschema_reader.go
@@ -2499,7 +2499,7 @@ func (e *tidbTrxTableRetriever) retrieve(ctx context.Context, sctx sessionctx.Co
 		for _, info := range infoList {
 			// If you have the PROCESS privilege, you can see all running transactions.
 			// Otherwise, you can see only your own transactions.
-			if !hasProcessPriv && loginUser != nil && info.Username != loginUser.Username {
+			if !hasProcessPriv && loginUser != nil && info.ProcessInfo.Username != loginUser.Username {
 				continue
 			}
 			e.txnInfo = append(e.txnInfo, info)
diff --git a/pkg/server/server.go b/pkg/server/server.go
index 272d816a61..77d6a3ad20 100644
--- a/pkg/server/server.go
+++ b/pkg/server/server.go
@@ -839,7 +839,8 @@ func (s *Server) getUserProcessList() map[uint64]*util.ProcessInfo {
 	return rs
 }
 
-// ShowTxnList shows all txn info for displaying in `TIDB_TRX`
+// ShowTxnList shows all txn info for displaying in `TIDB_TRX`.
+// Internal sessions are not taken into consideration.
 func (s *Server) ShowTxnList() []*txninfo.TxnInfo {
 	s.rwlock.RLock()
 	defer s.rwlock.RUnlock()
@@ -847,7 +848,7 @@ func (s *Server) ShowTxnList() []*txninfo.TxnInfo {
 	for _, client := range s.clients {
 		if client.ctx.Session != nil {
 			info := client.ctx.Session.TxnInfo()
-			if info != nil {
+			if info != nil && info.ProcessInfo != nil {
 				rs = append(rs, info)
 			}
 		}
diff --git a/pkg/session/session.go b/pkg/session/session.go
index f18ce5ccfe..5b0fbbfa72 100644
--- a/pkg/session/session.go
+++ b/pkg/session/session.go
@@ -507,6 +507,7 @@ func (s *session) FieldList(tableName string) ([]*ast.ResultField, error) {
 }
 
 // TxnInfo returns a pointer to a *copy* of the internal TxnInfo, thus is *read only*
+// Process field may not initialize if this is a session used internally.
 func (s *session) TxnInfo() *txninfo.TxnInfo {
 	s.txn.mu.RLock()
 	// Copy on read to get a snapshot, this API shouldn't be frequently called.
@@ -519,17 +520,18 @@ func (s *session) TxnInfo() *txninfo.TxnInfo {
 
 	processInfo := s.ShowProcess()
 	if processInfo == nil {
-		return nil
+		return &txnInfo
+	}
+	txnInfo.ProcessInfo = &txninfo.ProcessInfo{
+		ConnectionID:    processInfo.ID,
+		Username:        processInfo.User,
+		CurrentDB:       processInfo.DB,
+		RelatedTableIDs: make(map[int64]struct{}),
 	}
-	txnInfo.ConnectionID = processInfo.ID
-	txnInfo.Username = processInfo.User
-	txnInfo.CurrentDB = processInfo.DB
-	txnInfo.RelatedTableIDs = make(map[int64]struct{})
 	s.GetSessionVars().GetRelatedTableForMDL().Range(func(key, value interface{}) bool {
-		txnInfo.RelatedTableIDs[key.(int64)] = struct{}{}
+		txnInfo.ProcessInfo.RelatedTableIDs[key.(int64)] = struct{}{}
 		return true
 	})
-
 	return &txnInfo
 }
 
@@ -3853,9 +3855,10 @@ func GetStartTSFromSession(se interface{}) (startTS, processInfoID uint64) {
 	txnInfo := tmp.TxnInfo()
 	if txnInfo != nil {
 		startTS = txnInfo.StartTS
-		processInfoID = txnInfo.ConnectionID
+		if txnInfo.ProcessInfo != nil {
+			processInfoID = txnInfo.ProcessInfo.ConnectionID
+		}
 	}
-
 	logutil.BgLogger().Debug(
 		"GetStartTSFromSession getting startTS of internal session",
 		zap.Uint64("startTS", startTS), zap.Time("start time", oracle.GetTimeFromTS(startTS)))
diff --git a/pkg/session/txninfo/txn_info.go b/pkg/session/txninfo/txn_info.go
index 50e60da829..5067487ebe 100644
--- a/pkg/session/txninfo/txn_info.go
+++ b/pkg/session/txninfo/txn_info.go
@@ -174,8 +174,12 @@ type TxnInfo struct {
 	// How many entries are in MemDB
 	EntriesCount uint64
 
-	// The following fields will be filled in `session` instead of `LazyTxn`
+	// The following field will be filled in `session` instead of `LazyTxn`
+	ProcessInfo *ProcessInfo
+}
 
+// ProcessInfo is part of fields of txnInfo, which will be filled in `session` instead of `LazyTxn`
+type ProcessInfo struct {
 	// Which session this transaction belongs to
 	ConnectionID uint64
 	// The user who open this session
@@ -219,13 +223,25 @@ var columnValueGetterMap = map[string]func(*TxnInfo) types.Datum{
 		return types.NewDatum(info.EntriesCount)
 	},
 	SessionIDStr: func(info *TxnInfo) types.Datum {
-		return types.NewDatum(info.ConnectionID)
+		var connectionID uint64
+		if info.ProcessInfo != nil {
+			connectionID = info.ProcessInfo.ConnectionID
+		}
+		return types.NewDatum(connectionID)
 	},
 	UserStr: func(info *TxnInfo) types.Datum {
-		return types.NewDatum(info.Username)
+		var userName string
+		if info.ProcessInfo != nil {
+			userName = info.ProcessInfo.Username
+		}
+		return types.NewDatum(userName)
 	},
 	DBStr: func(info *TxnInfo) types.Datum {
-		return types.NewDatum(info.CurrentDB)
+		var currentDB string
+		if info.ProcessInfo != nil {
+			currentDB = info.ProcessInfo.CurrentDB
+		}
+		return types.NewDatum(currentDB)
 	},
 	AllSQLDigestsStr: func(info *TxnInfo) types.Datum {
 		allSQLDigests := info.AllSQLDigests
@@ -241,7 +257,10 @@ var columnValueGetterMap = map[string]func(*TxnInfo) types.Datum{
 		return types.NewDatum(string(res))
 	},
 	RelatedTableIDsStr: func(info *TxnInfo) types.Datum {
-		relatedTableIDs := info.RelatedTableIDs
+		var relatedTableIDs map[int64]struct{}
+		if info.ProcessInfo != nil {
+			relatedTableIDs = info.ProcessInfo.RelatedTableIDs
+		}
 		str := strings.Builder{}
 		first := true
 		for tblID := range relatedTableIDs {


Test patch:
diff --git a/pkg/ddl/ingest/integration_test.go b/pkg/ddl/ingest/integration_test.go
index 27d5f83127..bab9d45e78 100644
--- a/pkg/ddl/ingest/integration_test.go
+++ b/pkg/ddl/ingest/integration_test.go
@@ -29,7 +29,6 @@ import (
 	"github.com/pingcap/tidb/pkg/errno"
 	"github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/testkit"
-	"github.com/pingcap/tidb/tests/realtikvtest"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -121,7 +120,7 @@ func TestIngestError(t *testing.T) {
 }
 
 func TestAddIndexIngestPanic(t *testing.T) {
-	store := realtikvtest.CreateMockStoreAndSetup(t)
+	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test;")
 	defer ingesttestutil.InjectMockBackendMgr(t, store)()
@@ -142,8 +141,39 @@ func TestAddIndexIngestPanic(t *testing.T) {
 	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/mockLocalWriterPanic"))
 }
 
+func TestAddIndexSetInternalSessions(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test;")
+	defer ingesttestutil.InjectMockBackendMgr(t, store)()
+
+	tk.MustExec("set global tidb_enable_dist_task = 1;")
+	tk.MustExec("set global tidb_ddl_reorg_worker_cnt = 1;")
+	tk.MustExec("create table t (a int);")
+	tk.MustExec("insert into t values (1);")
+	expectInternalTS := []uint64{}
+	actualInternalTS := []uint64{}
+	err := failpoint.EnableCall("github.com/pingcap/tidb/pkg/ddl/wrapInBeginRollbackStartTS", func(startTS uint64) {
+		expectInternalTS = append(expectInternalTS, startTS)
+	})
+	require.NoError(t, err)
+	defer failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/wrapInBeginRollbackStartTS")
+	err = failpoint.Enable("github.com/pingcap/tidb/pkg/ddl/scanRecordExec", "return")
+	require.NoError(t, err)
+	defer failpoint.Disable("github.com/pingcap/tidb/pkg/ddl/scanRecordExec")
+	ddl.OperatorCallBackForTest = func() {
+		mgr := tk.Session().GetSessionManager()
+		actualInternalTS = mgr.GetInternalSessionStartTSList()
+	}
+	tk.MustExec("alter table t add index idx(a);")
+	require.Len(t, expectInternalTS, 1)
+	for _, ts := range expectInternalTS {
+		require.Contains(t, actualInternalTS, ts)
+	}
+}
+
 func TestAddIndexIngestCancel(t *testing.T) {
-	store, dom := realtikvtest.CreateMockStoreAndDomainAndSetup(t)
+	store, dom := testkit.CreateMockStoreAndDomain(t)
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test;")
 	defer ingesttestutil.InjectMockBackendMgr(t, store)()
diff --git a/pkg/infoschema/test/clustertablestest/tables_test.go b/pkg/infoschema/test/clustertablestest/tables_test.go
index a19975a060..f8445b9f6b 100644
--- a/pkg/infoschema/test/clustertablestest/tables_test.go
+++ b/pkg/infoschema/test/clustertablestest/tables_test.go
@@ -1221,9 +1221,11 @@ func TestTiDBTrx(t *testing.T) {
 		CurrentSQLDigest: digest.String(),
 		State:            txninfo.TxnIdle,
 		EntriesCount:     1,
-		ConnectionID:     2,
-		Username:         "root",
-		CurrentDB:        "test",
+		ProcessInfo: &txninfo.ProcessInfo{
+			ConnectionID: 2,
+			Username:     "root",
+			CurrentDB:    "test",
+		},
 	}
 
 	blockTime2 := time.Date(2021, 05, 20, 13, 18, 30, 123456000, time.Local)
@@ -1232,9 +1234,11 @@ func TestTiDBTrx(t *testing.T) {
 		CurrentSQLDigest: "",
 		AllSQLDigests:    []string{"sql1", "sql2", digest.String()},
 		State:            txninfo.TxnLockAcquiring,
-		ConnectionID:     10,
-		Username:         "user1",
-		CurrentDB:        "db1",
+		ProcessInfo: &txninfo.ProcessInfo{
+			ConnectionID: 10,
+			Username:     "user1",
+			CurrentDB:    "db1",
+		},
 	}
 	sm.TxnInfo[1].BlockStartTime.Valid = true
 	sm.TxnInfo[1].BlockStartTime.Time = blockTime2
diff --git a/pkg/testkit/BUILD.bazel b/pkg/testkit/BUILD.bazel
index b9922d4065..849ec0c2d1 100644
--- a/pkg/testkit/BUILD.bazel
+++ b/pkg/testkit/BUILD.bazel
@@ -26,7 +26,6 @@ go_library(
         "//pkg/resourcemanager",
         "//pkg/session",
         "//pkg/session/txninfo",
-        "//pkg/sessionctx",
         "//pkg/sessionctx/variable",
         "//pkg/store/driver",
         "//pkg/store/mockstore",
diff --git a/pkg/testkit/mocksessionmanager.go b/pkg/testkit/mocksessionmanager.go
index af1aa10226..7f2b2fc106 100644
--- a/pkg/testkit/mocksessionmanager.go
+++ b/pkg/testkit/mocksessionmanager.go
@@ -24,7 +24,6 @@ import (
 	"github.com/pingcap/tidb/pkg/planner/core"
 	"github.com/pingcap/tidb/pkg/session"
 	"github.com/pingcap/tidb/pkg/session/txninfo"
-	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/util"
 )
 
@@ -52,7 +51,7 @@ func (msm *MockSessionManager) ShowTxnList() []*txninfo.TxnInfo {
 	rs := make([]*txninfo.TxnInfo, 0, len(msm.Conn))
 	for _, se := range msm.Conn {
 		info := se.TxnInfo()
-		if info != nil {
+		if info != nil && info.ProcessInfo != nil {
 			rs = append(rs, info)
 		}
 	}
@@ -155,12 +154,16 @@ func (msm *MockSessionManager) GetInternalSessionStartTSList() []uint64 {
 	defer msm.mu.Unlock()
 	ret := make([]uint64, 0, len(msm.internalSessions))
 	for internalSess := range msm.internalSessions {
-		se := internalSess.(sessionctx.Context)
-		sessVars := se.GetSessionVars()
-		sessVars.TxnCtxMu.Lock()
-		startTS := sessVars.TxnCtx.StartTS
-		sessVars.TxnCtxMu.Unlock()
-		ret = append(ret, startTS)
+		// Ref the implementation of `GetInternalSessionStartTSList` on the real session manager. The `TxnInfo` is more
+		// accurate, because if a session is pending, the `StartTS` in `sessVars.TxnCtx` will not be updated. For example,
+		// if there is not DDL for a long time, the minimal internal session start ts will not have any progress.
+		if se, ok := internalSess.(interface{ TxnInfo() *txninfo.TxnInfo }); ok {
+			txn := se.TxnInfo()
+			if txn != nil {
+				ret = append(ret, txn.StartTS)
+			}
+			continue
+		}
 	}
 	return ret
 }
diff --git a/tests/realtikvtest/txntest/txn_state_test.go b/tests/realtikvtest/txntest/txn_state_test.go
index 671e4b7e21..f2d8d57f02 100644
--- a/tests/realtikvtest/txntest/txn_state_test.go
+++ b/tests/realtikvtest/txntest/txn_state_test.go
@@ -74,9 +74,9 @@ func TestBasicTxnState(t *testing.T) {
 	require.Equal(t, []string{beginDigest.String(), selectTSDigest.String(), expectedDigest.String()}, info.AllSQLDigests)
 
 	// len and size will be covered in TestLenAndSize
-	require.Equal(t, tk.Session().GetSessionVars().ConnectionID, info.ConnectionID)
-	require.Equal(t, "", info.Username)
-	require.Equal(t, "test", info.CurrentDB)
+	require.Equal(t, tk.Session().GetSessionVars().ConnectionID, info.ProcessInfo.ConnectionID)
+	require.Equal(t, "", info.ProcessInfo.Username)
+	require.Equal(t, "test", info.ProcessInfo.CurrentDB)
 	require.Equal(t, startTS, info.StartTS)
 
 	require.NoError(t, failpoint.Enable("tikvclient/beforePrewrite", "pause"))