*: improve log about stale-read query (#52494)

close #52492

Author: crazycs520

pkg/executor/adapter.go

@@ -1579,6 +1579,10 @@ func (a *ExecStmt) LogSlowQuery(txnTS uint64, succ bool, hasMoreResults bool) {
 	if !keyspace.IsKeyspaceNameEmpty(keyspaceName) {
 		keyspaceID = uint32(a.Ctx.GetStore().GetCodec().GetKeyspaceID())
 	}
+	if txnTS == 0 {
+		// TODO: txnTS maybe ambiguous, consider logging stale-read-ts with a new field in the slow log.
+		txnTS = sessVars.TxnCtx.StaleReadTs
+	}
 
 	slowItems := &variable.SlowQueryLogItems{
 		TxnTS:             txnTS,

pkg/executor/slow_query_sql_test.go

@@ -19,6 +19,7 @@ import (
 	"math"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/executor"
@@ -122,7 +123,7 @@ func TestSlowQueryNonPrepared(t *testing.T) {
 		`0 0 select * from t where a<3;`))
 }
 
-func TestSlowQueryPrepared(t *testing.T) {
+func TestSlowQueryMisc(t *testing.T) {
 	originCfg := config.GetGlobalConfig()
 	newCfg := *originCfg
 
@@ -158,6 +159,21 @@ func TestSlowQueryPrepared(t *testing.T) {
 	tk.MustQuery("SELECT Query FROM `information_schema`.`slow_query` " +
 		"where query like 'select%sleep%' order by time desc limit 1").
 		Check(testkit.Rows("select `sleep` ( ? ) , ?;"))
+
+	// Test 3 kinds of stale-read query.
+	tk.MustExec("create table test.t_stale_read (a int)")
+	time.Sleep(time.Second + time.Millisecond*10)
+	tk.MustExec("set tidb_redact_log=0;")
+	tk.MustExec("set @@tidb_read_staleness='-1'")
+	tk.MustQuery("select a from test.t_stale_read")
+	tk.MustExec("set @@tidb_read_staleness='0'")
+	t1 := time.Now()
+	tk.MustQuery(fmt.Sprintf("select a from test.t_stale_read as of timestamp '%s'", t1.Format("2006-1-2 15:04:05")))
+	tk.MustExec(fmt.Sprintf("start transaction read only as of timestamp '%v'", t1.Format("2006-1-2 15:04:05")))
+	tk.MustQuery("select a from test.t_stale_read")
+	tk.MustExec("commit")
+	require.Len(t, tk.MustQuery("SELECT query, txn_start_ts  FROM `information_schema`.`slow_query` "+
+		"where (query = 'select a from test.t_stale_read;' or query like 'select a from test.t_stale_read as of timestamp %') and Txn_start_ts > 0").Rows(), 3)
 }
 
 func TestLogSlowLogIndex(t *testing.T) {

pkg/planner/core/preprocess.go

@@ -1759,6 +1759,7 @@ func (p *preprocessor) updateStateFromStaleReadProcessor() error {
 				if err := txnManager.OnStmtStart(context.TODO(), txnManager.GetCurrentStmt()); err != nil {
 					return err
 				}
+				p.sctx.GetSessionVars().TxnCtx.StaleReadTs = p.LastSnapshotTS
 			}
 		}
 	}

pkg/server/conn.go

@@ -1139,17 +1139,21 @@ func (cc *clientConn) Run(ctx context.Context) {
 			if storeerr.ErrLockAcquireFailAndNoWaitSet.Equal(err) {
 				logutil.Logger(ctx).Debug("Expected error for FOR UPDATE NOWAIT", zap.Error(err))
 			} else {
-				var startTS uint64
+				var timestamp uint64
 				if ctx := cc.getCtx(); ctx != nil && ctx.GetSessionVars() != nil && ctx.GetSessionVars().TxnCtx != nil {
-					startTS = ctx.GetSessionVars().TxnCtx.StartTS
+					timestamp = ctx.GetSessionVars().TxnCtx.StartTS
+					if timestamp == 0 && ctx.GetSessionVars().TxnCtx.StaleReadTs > 0 {
+						// for state-read query.
+						timestamp = ctx.GetSessionVars().TxnCtx.StaleReadTs
+					}
 				}
 				logutil.Logger(ctx).Info("command dispatched failed",
 					zap.String("connInfo", cc.String()),
 					zap.String("command", mysql.Command2Str[data[0]]),
 					zap.String("status", cc.SessionStatusToString()),
 					zap.Stringer("sql", getLastStmtInConn{cc}),
 					zap.String("txn_mode", txnMode),
-					zap.Uint64("timestamp", startTS),
+					zap.Uint64("timestamp", timestamp),
 					zap.String("err", errStrForLog(err, cc.ctx.GetSessionVars().EnableRedactLog)),
 				)
 			}

pkg/session/txnmanager.go

@@ -333,6 +333,7 @@ func (m *txnManager) newProviderWithRequest(r *sessiontxn.EnterNewTxnRequest) (s
 	}
 
 	if r.StaleReadTS > 0 {
+		m.sctx.GetSessionVars().TxnCtx.StaleReadTs = r.StaleReadTS
 		return staleread.NewStalenessTxnContextProvider(m.sctx, r.StaleReadTS, nil), nil
 	}
 

pkg/sessionctx/variable/session.go

@@ -200,6 +200,7 @@ type TxnCtxNoNeedToRestore struct {
 	InfoSchema  any
 	History     any
 	StartTS     uint64
+	StaleReadTs uint64
 
 	// ShardStep indicates the max size of continuous rowid shard in one transaction.
 	ShardStep    int