ttl: always enable all read engines for TTL sessions (#56604)

close #56402

pkg/ttl/ttlworker/job_manager_integration_test.go

@@ -75,6 +75,7 @@ func TestGetSession(t *testing.T) {
 	tk.MustExec("set @@tidb_retry_limit=1")
 	tk.MustExec("set @@tidb_enable_1pc=0")
 	tk.MustExec("set @@tidb_enable_async_commit=0")
+	tk.MustExec("set @@tidb_isolation_read_engines='tiflash,tidb'")
 	var getCnt atomic.Int32
 
 	pool := pools.NewResourcePool(func() (pools.Resource, error) {
@@ -97,13 +98,13 @@ func TestGetSession(t *testing.T) {
 	require.Equal(t, "Europe/Berlin", tz.String())
 
 	// session variables should be set
-	tk.MustQuery("select @@time_zone, @@tidb_retry_limit, @@tidb_enable_1pc, @@tidb_enable_async_commit").
-		Check(testkit.Rows("UTC 0 1 1"))
+	tk.MustQuery("select @@time_zone, @@tidb_retry_limit, @@tidb_enable_1pc, @@tidb_enable_async_commit, @@tidb_isolation_read_engines").
+		Check(testkit.Rows("UTC 0 1 1 tikv,tiflash,tidb"))
 
 	// all session variables should be restored after close
 	se.Close()
-	tk.MustQuery("select @@time_zone, @@tidb_retry_limit, @@tidb_enable_1pc, @@tidb_enable_async_commit").
-		Check(testkit.Rows("Asia/Shanghai 1 0 0"))
+	tk.MustQuery("select @@time_zone, @@tidb_retry_limit, @@tidb_enable_1pc, @@tidb_enable_async_commit, @@tidb_isolation_read_engines").
+		Check(testkit.Rows("Asia/Shanghai 1 0 0 tiflash,tidb"))
 }
 
 func TestParallelLockNewJob(t *testing.T) {

pkg/ttl/ttlworker/session.go

@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
@@ -55,6 +56,12 @@ var DetachStatsCollector = func(s sqlexec.SQLExecutor) sqlexec.SQLExecutor {
 	return s
 }
 
+var allIsolationReadEngines = map[kv.StoreType]struct{}{
+	kv.TiKV:    {},
+	kv.TiFlash: {},
+	kv.TiDB:    {},
+}
+
 func getSession(pool util.SessionPool) (session.Session, error) {
 	resource, err := pool.Get()
 	if err != nil {
@@ -77,6 +84,7 @@ func getSession(pool util.SessionPool) (session.Session, error) {
 	originalEnable1PC := sctx.GetSessionVars().Enable1PC
 	originalEnableAsyncCommit := sctx.GetSessionVars().EnableAsyncCommit
 	originalTimeZone, restoreTimeZone := "", false
+	originalIsolationReadEngines, restoreIsolationReadEngines := "", false
 
 	se := session.NewSession(sctx, exec, func(se session.Session) {
 		_, err = se.ExecuteSQL(context.Background(), fmt.Sprintf("set tidb_retry_limit=%d", originalRetryLimit))
@@ -103,6 +111,12 @@ func getSession(pool util.SessionPool) (session.Session, error) {
 			terror.Log(err)
 		}
 
+		if restoreIsolationReadEngines {
+			_, err = se.ExecuteSQL(context.Background(), "set tidb_isolation_read_engines=%?", originalIsolationReadEngines)
+			intest.AssertNoError(err)
+			terror.Log(err)
+		}
+
 		DetachStatsCollector(exec)
 
 		pool.Put(resource)
@@ -157,6 +171,32 @@ func getSession(pool util.SessionPool) (session.Session, error) {
 	}
 	restoreTimeZone = true
 
+	// allow the session in TTL to use all read engines.
+	_, hasTiDBEngine := se.GetSessionVars().IsolationReadEngines[kv.TiDB]
+	_, hasTiKVEngine := se.GetSessionVars().IsolationReadEngines[kv.TiKV]
+	_, hasTiFlashEngine := se.GetSessionVars().IsolationReadEngines[kv.TiFlash]
+	if !hasTiDBEngine || !hasTiKVEngine || !hasTiFlashEngine {
+		rows, err := se.ExecuteSQL(context.Background(), "select @@tidb_isolation_read_engines")
+		if err != nil {
+			se.Close()
+			return nil, err
+		}
+
+		if len(rows) == 0 || rows[0].Len() == 0 {
+			se.Close()
+			return nil, errors.New("failed to get tidb_isolation_read_engines variable")
+		}
+		originalIsolationReadEngines = rows[0].GetString(0)
+
+		_, err = se.ExecuteSQL(context.Background(), "set tidb_isolation_read_engines='tikv,tiflash,tidb'")
+		if err != nil {
+			se.Close()
+			return nil, err
+		}
+
+		restoreIsolationReadEngines = true
+	}
+
 	return se, nil
 }