txn: Add txn state's view (#22908)

bindinfo/bind_test.go

@@ -35,6 +35,7 @@ import (
  "github.com/pingcap/tidb/metrics"
  plannercore "github.com/pingcap/tidb/planner/core"
  "github.com/pingcap/tidb/session"
+ "github.com/pingcap/tidb/session/txninfo"
  "github.com/pingcap/tidb/store/mockstore"
  "github.com/pingcap/tidb/store/tikv/mockstore/cluster"
  "github.com/pingcap/tidb/util"
@@ -70,6 +71,10 @@ type mockSessionManager struct {
  PS []*util.ProcessInfo
 }
 
+func (msm *mockSessionManager) ShowTxnList() []*txninfo.TxnInfo {
+ panic("unimplemented!")
+}
+
 func (msm *mockSessionManager) ShowProcessList() map[uint64]*util.ProcessInfo {
  ret := make(map[uint64]*util.ProcessInfo)
  for _, item := range msm.PS {

domain/domain_test.go

@@ -35,6 +35,7 @@ import (
  "github.com/pingcap/tidb/kv"
  "github.com/pingcap/tidb/meta"
  "github.com/pingcap/tidb/metrics"
+ "github.com/pingcap/tidb/session/txninfo"
  "github.com/pingcap/tidb/sessionctx/variable"
  "github.com/pingcap/tidb/store/mockstore"
  "github.com/pingcap/tidb/store/tikv"
@@ -241,6 +242,10 @@ type mockSessionManager struct {
  PS []*util.ProcessInfo
 }
 
+func (msm *mockSessionManager) ShowTxnList() []*txninfo.TxnInfo {
+ panic("unimplemented!")
+}
+
 func (msm *mockSessionManager) ShowProcessList() map[uint64]*util.ProcessInfo {
  ret := make(map[uint64]*util.ProcessInfo)
  for _, item := range msm.PS {

executor/adapter.go

@@ -55,6 +55,7 @@ import (
  "github.com/pingcap/tidb/util/sqlexec"
  "github.com/pingcap/tidb/util/stmtsummary"
  "github.com/pingcap/tidb/util/stringutil"
+
  "go.uber.org/zap"
  "go.uber.org/zap/zapcore"
 )
@@ -377,6 +378,7 @@ func (a *ExecStmt) Exec(ctx context.Context) (_ sqlexec.RecordSet, err error) {
  if txn.Valid() {
  txnStartTS = txn.StartTS()
  }
+
  return &recordSet{
  executor: e,
  stmt: a,
@@ -590,6 +592,7 @@ func (a *ExecStmt) handlePessimisticDML(ctx context.Context, e Executor) error {
  }
  e, err = a.handlePessimisticLockError(ctx, err)
  if err != nil {
+ // todo: Report deadlock
  if ErrDeadlock.Equal(err) {
  metrics.StatementDeadlockDetectDuration.Observe(time.Since(startLocking).Seconds())
  }

executor/builder.go

@@ -1531,7 +1531,9 @@ func (b *executorBuilder) buildMemTable(v *plannercore.PhysicalMemTable) Executo
  strings.ToLower(infoschema.TablePlacementPolicy),
  strings.ToLower(infoschema.TableClientErrorsSummaryGlobal),
  strings.ToLower(infoschema.TableClientErrorsSummaryByUser),
- strings.ToLower(infoschema.TableClientErrorsSummaryByHost):
+ strings.ToLower(infoschema.TableClientErrorsSummaryByHost),
+ strings.ToLower(infoschema.TableTiDBTrx),
+ strings.ToLower(infoschema.ClusterTableTiDBTrx):
  return &MemTableReaderExec{
  baseExecutor: newBaseExecutor(b.ctx, v.Schema(), v.ID()),
  table: v.Table,

executor/executor_pkg_test.go

@@ -30,6 +30,7 @@ import (
  "github.com/pingcap/tidb/executor/aggfuncs"
  "github.com/pingcap/tidb/expression"
  plannerutil "github.com/pingcap/tidb/planner/util"
+ txninfo "github.com/pingcap/tidb/session/txninfo"
  "github.com/pingcap/tidb/sessionctx/variable"
  "github.com/pingcap/tidb/types"
  "github.com/pingcap/tidb/util"
@@ -60,6 +61,10 @@ type mockSessionManager struct {
  serverID uint64
 }
 
+func (msm *mockSessionManager) ShowTxnList() []*txninfo.TxnInfo {
+ panic("unimplemented!")
+}
+
 // ShowProcessList implements the SessionManager.ShowProcessList interface.
 func (msm *mockSessionManager) ShowProcessList() map[uint64]*util.ProcessInfo {
  ret := make(map[uint64]*util.ProcessInfo)

executor/explainfor_test.go

@@ -26,6 +26,7 @@ import (
  "github.com/pingcap/parser/auth"
  "github.com/pingcap/tidb/planner/core"
  "github.com/pingcap/tidb/session"
+ txninfo "github.com/pingcap/tidb/session/txninfo"
  "github.com/pingcap/tidb/sessionctx/variable"
  "github.com/pingcap/tidb/util"
  "github.com/pingcap/tidb/util/israce"
@@ -38,6 +39,10 @@ type mockSessionManager1 struct {
  PS []*util.ProcessInfo
 }
 
+func (msm *mockSessionManager1) ShowTxnList() []*txninfo.TxnInfo {
+ return nil
+}
+
 // ShowProcessList implements the SessionManager.ShowProcessList interface.
 func (msm *mockSessionManager1) ShowProcessList() map[uint64]*util.ProcessInfo {
  ret := make(map[uint64]*util.ProcessInfo)

executor/infoschema_reader.go

@@ -149,6 +149,10 @@ func (e *memtableRetriever) retrieve(ctx context.Context, sctx sessionctx.Contex
  infoschema.TableClientErrorsSummaryByUser,
  infoschema.TableClientErrorsSummaryByHost:
  err = e.setDataForClientErrorsSummary(sctx, e.table.Name.O)
+ case infoschema.TableTiDBTrx:
+ e.setDataForTiDBTrx(sctx)
+ case infoschema.ClusterTableTiDBTrx:
+ err = e.setDataForClusterTiDBTrx(sctx)
  }
  if err != nil {
  return nil, err
@@ -2011,6 +2015,40 @@ func (e *memtableRetriever) setDataForClientErrorsSummary(ctx sessionctx.Context
  return nil
 }
 
+func (e *memtableRetriever) setDataForTiDBTrx(ctx sessionctx.Context) {
+ sm := ctx.GetSessionManager()
+ if sm == nil {
+ return
+ }
+
+ loginUser := ctx.GetSessionVars().User
+ var hasProcessPriv bool
+ if pm := privilege.GetPrivilegeManager(ctx); pm != nil {
+ if pm.RequestVerification(ctx.GetSessionVars().ActiveRoles, "", "", "", mysql.ProcessPriv) {
+ hasProcessPriv = true
+ }
+ }
+ infoList := sm.ShowTxnList()
+ 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 {
+ continue
+ }
+ e.rows = append(e.rows, info.ToDatum())
+ }
+}
+
+func (e *memtableRetriever) setDataForClusterTiDBTrx(ctx sessionctx.Context) error {
+ e.setDataForTiDBTrx(ctx)
+ rows, err := infoschema.AppendHostInfoToRows(ctx, e.rows)
+ if err != nil {
+ return err
+ }
+ e.rows = rows
+ return nil
+}
+
 type hugeMemTableRetriever struct {
  dummyCloser
  table *model.TableInfo

executor/infoschema_reader_test.go

@@ -36,6 +36,7 @@ import (
  "github.com/pingcap/tidb/kv"
  "github.com/pingcap/tidb/server"
  "github.com/pingcap/tidb/session"
+ txninfo "github.com/pingcap/tidb/session/txninfo"
  "github.com/pingcap/tidb/sessionctx/variable"
  "github.com/pingcap/tidb/statistics"
  "github.com/pingcap/tidb/statistics/handle"
@@ -728,6 +729,10 @@ type mockSessionManager struct {
  serverID uint64
 }
 
+func (sm *mockSessionManager) ShowTxnList() []*txninfo.TxnInfo {
+ panic("unimplemented!")
+}
+
 func (sm *mockSessionManager) ShowProcessList() map[uint64]*util.ProcessInfo {
  return sm.processInfoMap
 }

executor/prepared_test.go

@@ -26,6 +26,7 @@ import (
  "github.com/pingcap/tidb/domain"
  plannercore "github.com/pingcap/tidb/planner/core"
  "github.com/pingcap/tidb/session"
+ txninfo "github.com/pingcap/tidb/session/txninfo"
  "github.com/pingcap/tidb/sessionctx/variable"
  "github.com/pingcap/tidb/util"
  "github.com/pingcap/tidb/util/israce"
@@ -135,6 +136,10 @@ type mockSessionManager2 struct {
  killed bool
 }
 
+func (sm *mockSessionManager2) ShowTxnList() []*txninfo.TxnInfo {
+ panic("unimplemented!")
+}
+
 func (sm *mockSessionManager2) ShowProcessList() map[uint64]*util.ProcessInfo {
  pl := make(map[uint64]*util.ProcessInfo)
  if pi, ok := sm.GetProcessInfo(0); ok {

executor/seqtest/prepared_test.go

@@ -25,6 +25,7 @@ import (
  "github.com/pingcap/tidb/metrics"
  plannercore "github.com/pingcap/tidb/planner/core"
  "github.com/pingcap/tidb/session"
+ "github.com/pingcap/tidb/session/txninfo"
  "github.com/pingcap/tidb/types"
  "github.com/pingcap/tidb/util"
  "github.com/pingcap/tidb/util/kvcache"
@@ -796,6 +797,10 @@ type mockSessionManager1 struct {
  Se session.Session
 }
 
+func (msm *mockSessionManager1) ShowTxnList() []*txninfo.TxnInfo {
+ panic("unimplemented!")
+}
+
 // ShowProcessList implements the SessionManager.ShowProcessList interface.
 func (msm *mockSessionManager1) ShowProcessList() map[uint64]*util.ProcessInfo {
  ret := make(map[uint64]*util.ProcessInfo)

infoschema/cluster.go

@@ -37,6 +37,8 @@ const (
  ClusterTableStatementsSummary = "CLUSTER_STATEMENTS_SUMMARY"
  // ClusterTableStatementsSummaryHistory is the string constant of cluster statement summary history table.
  ClusterTableStatementsSummaryHistory = "CLUSTER_STATEMENTS_SUMMARY_HISTORY"
+ // ClusterTableTiDBTrx is the string constant of cluster transaction running table.
+ ClusterTableTiDBTrx = "CLUSTER_TIDB_TRX"
 )
 
 // memTableToClusterTables means add memory table to cluster table.
@@ -45,6 +47,7 @@ var memTableToClusterTables = map[string]string{
  TableProcesslist: ClusterTableProcesslist,
  TableStatementsSummary: ClusterTableStatementsSummary,
  TableStatementsSummaryHistory: ClusterTableStatementsSummaryHistory,
+ TableTiDBTrx: ClusterTableTiDBTrx,
 }
 
 func init() {

infoschema/infoschema_test.go

@@ -332,6 +332,7 @@ func (*testSuite) TestInfoTables(c *C) {
  "TABLESPACES",
  "COLLATION_CHARACTER_SET_APPLICABILITY",
  "PROCESSLIST",
+ "TIDB_TRX",
  }
  for _, t := range infoTables {
  tb, err1 := is.TableByName(util.InformationSchemaName, model.NewCIStr(t))

infoschema/tables.go

@@ -31,11 +31,13 @@ import (
  "github.com/pingcap/parser/model"
  "github.com/pingcap/parser/mysql"
  "github.com/pingcap/parser/terror"
+
  "github.com/pingcap/tidb/config"
  "github.com/pingcap/tidb/ddl/placement"
  "github.com/pingcap/tidb/domain/infosync"
  "github.com/pingcap/tidb/kv"
  "github.com/pingcap/tidb/meta/autoid"
+ "github.com/pingcap/tidb/session/txninfo"
  "github.com/pingcap/tidb/sessionctx"
  "github.com/pingcap/tidb/sessionctx/variable"
  "github.com/pingcap/tidb/store/tikv"
@@ -161,6 +163,8 @@ const (
  TableClientErrorsSummaryByUser = "CLIENT_ERRORS_SUMMARY_BY_USER"
  // TableClientErrorsSummaryByHost is the string constant of client errors table.
  TableClientErrorsSummaryByHost = "CLIENT_ERRORS_SUMMARY_BY_HOST"
+ // TableTiDBTrx is current running transaction status table.
+ TableTiDBTrx = "TIDB_TRX"
 )
 
 var tableIDMap = map[string]int64{
@@ -233,22 +237,25 @@ var tableIDMap = map[string]int64{
  TableClientErrorsSummaryGlobal: autoid.InformationSchemaDBID + 67,
  TableClientErrorsSummaryByUser: autoid.InformationSchemaDBID + 68,
  TableClientErrorsSummaryByHost: autoid.InformationSchemaDBID + 69,
+ TableTiDBTrx: autoid.InformationSchemaDBID + 70,
+ ClusterTableTiDBTrx: autoid.InformationSchemaDBID + 71,
 }
 
 type columnInfo struct {
- name string
- tp byte
- size int
- decimal int
- flag uint
- deflt interface{}
- comment string
+ name string
+ tp byte
+ size int
+ decimal int
+ flag uint
+ deflt interface{}
+ comment string
+ enumElems []string
 }
 
 func buildColumnInfo(col columnInfo) *model.ColumnInfo {
  mCharset := charset.CharsetBin
  mCollation := charset.CharsetBin
- if col.tp == mysql.TypeVarchar || col.tp == mysql.TypeBlob || col.tp == mysql.TypeLongBlob {
+ if col.tp == mysql.TypeVarchar || col.tp == mysql.TypeBlob || col.tp == mysql.TypeLongBlob || col.tp == mysql.TypeEnum {
  mCharset = charset.CharsetUTF8MB4
  mCollation = charset.CollationUTF8MB4
  }
@@ -259,6 +266,7 @@ func buildColumnInfo(col columnInfo) *model.ColumnInfo {
  Flen: col.size,
  Decimal: col.decimal,
  Flag: col.flag,
+ Elems: col.enumElems,
  }
  return &model.ColumnInfo{
  Name: model.NewCIStr(col.name),
@@ -1332,6 +1340,19 @@ var tableClientErrorsSummaryByHostCols = []columnInfo{
  {name: "LAST_SEEN", tp: mysql.TypeTimestamp, size: 26},
 }
 
+var tableTiDBTrxCols = []columnInfo{
+ {name: "ID", tp: mysql.TypeLonglong, size: 21, flag: mysql.PriKeyFlag | mysql.NotNullFlag | mysql.UnsignedFlag},
+ {name: "START_TIME", tp: mysql.TypeTimestamp, size: 26, comment: "Start time of the transaction"},
+ {name: "DIGEST", tp: mysql.TypeVarchar, size: 64, comment: "Digest of the sql the transaction are currently running"},
+ {name: "STATE", tp: mysql.TypeEnum, enumElems: txninfo.TxnRunningStateStrs, comment: "Current running state of the transaction"},
+ {name: "WAITING_START_TIME", tp: mysql.TypeTimestamp, size: 26, comment: "Current lock waiting's start time"},
+ {name: "LEN", tp: mysql.TypeLonglong, size: 64, comment: "How many entries are in MemDB"},
+ {name: "SIZE", tp: mysql.TypeLonglong, size: 64, comment: "MemDB used memory"},
+ {name: "SESSION_ID", tp: mysql.TypeLonglong, size: 21, flag: mysql.UnsignedFlag, comment: "Which session this transaction belongs to"},
+ {name: "USER", tp: mysql.TypeVarchar, size: 16, comment: "The user who open this session"},
+ {name: "DB", tp: mysql.TypeVarchar, size: 64, comment: "The schema this transaction works on"},
+}
+
 // GetShardingInfo returns a nil or description string for the sharding information of given TableInfo.
 // The returned description string may be:
 // - "NOT_SHARDED": for tables that SHARD_ROW_ID_BITS is not specified.
@@ -1701,6 +1722,7 @@ var tableNameToColumns = map[string][]columnInfo{
  TableClientErrorsSummaryGlobal: tableClientErrorsSummaryGlobalCols,
  TableClientErrorsSummaryByUser: tableClientErrorsSummaryByUserCols,
  TableClientErrorsSummaryByHost: tableClientErrorsSummaryByHostCols,
+ TableTiDBTrx: tableTiDBTrxCols,
 }
 
 func createInfoSchemaTable(_ autoid.Allocators, meta *model.TableInfo) (table.Table, error) {