planner, session: add isolation read with engine type (pingcap#12997)

kv/kv.go

@@ -229,6 +229,14 @@ const (
 	TiFlash
 )
 
+// Name returns the name of store type.
+func (t StoreType) Name() string {
+	if t == TiFlash {
+		return "tiflash"
+	}
+	return "tikv"
+}
+
 // Request represents a kv request.
 type Request struct {
 	// Tp is the request type.

planner/core/common_plans.go

@@ -620,9 +620,9 @@ func (e *Explain) explainPlanInRowFormat(p PhysicalPlan, taskType, indent string
 		var storeType string
 		switch copPlan.StoreType {
 		case kv.TiKV:
-			storeType = "tikv"
+			storeType = kv.TiKV.Name()
 		case kv.TiFlash:
-			storeType = "tiflash"
+			storeType = kv.TiFlash.Name()
 		default:
 			err = errors.Errorf("the store type %v is unknown", copPlan.StoreType)
 			return

planner/core/integration_test.go

@@ -263,3 +263,20 @@ func (s *testIntegrationSuite) TestErrNoDB(c *C) {
 	tk.MustExec("use test")
 	tk.MustExec("grant select on test1111 to test@'%'")
 }
+
+func (s *testIntegrationSuite) TestNoneAccessPathsFoundByIsolationRead(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int primary key)")
+
+	_, err := tk.Exec("select * from t")
+	c.Assert(err, IsNil)
+
+	tk.MustExec("set @@session.tidb_isolation_read_engines = 'tiflash'")
+
+	_, err = tk.Exec("select * from t")
+	c.Assert(err, NotNil)
+	c.Assert(err.Error(), Equals, "[planner:1815]Internal : Can not find access path matching 'tidb_isolation_read_engines'(value: 'tiflash'). Available values are 'tikv'.")
+}

planner/core/logical_plan_builder.go

@@ -2341,6 +2341,10 @@ func (b *PlanBuilder) buildDataSource(ctx context.Context, tn *ast.TableName, as
 	if err != nil {
 		return nil, err
 	}
+	possiblePaths, err = b.filterPathByIsolationRead(possiblePaths)
+	if err != nil {
+		return nil, err
+	}
 
 	var columns []*table.Column
 	if b.inUpdateStmt {

planner/core/planbuilder.go

@@ -35,6 +35,7 @@ import (
 	"github.com/pingcap/tidb/planner/property"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/sessionctx/stmtctx"
+	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/statistics"
 	"github.com/pingcap/tidb/store/tikv"
 	"github.com/pingcap/tidb/table"
@@ -573,6 +574,32 @@ func (b *PlanBuilder) getPossibleAccessPaths(indexHints []*ast.IndexHint, tblInf
 	return available, nil
 }
 
+func (b *PlanBuilder) filterPathByIsolationRead(paths []*accessPath) ([]*accessPath, error) {
+	// TODO: filter paths with isolation read locations.
+	isolationReadEngines := b.ctx.GetSessionVars().GetIsolationReadEngines()
+	availableEngine := map[kv.StoreType]struct{}{}
+	var availableEngineStr string
+	for i := len(paths) - 1; i >= 0; i-- {
+		if _, ok := availableEngine[paths[i].storeType]; !ok {
+			availableEngine[paths[i].storeType] = struct{}{}
+			if availableEngineStr != "" {
+				availableEngineStr += ", "
+			}
+			availableEngineStr += paths[i].storeType.Name()
+		}
+		if _, ok := isolationReadEngines[paths[i].storeType]; !ok {
+			paths = append(paths[:i], paths[i+1:]...)
+		}
+	}
+	var err error
+	if len(paths) == 0 {
+		engineVals, _ := b.ctx.GetSessionVars().GetSystemVar(variable.TiDBIsolationReadEngines)
+		err = ErrInternal.GenWithStackByArgs(fmt.Sprintf("Can not find access path matching '%v'(value: '%v'). Available values are '%v'.",
+			variable.TiDBIsolationReadEngines, engineVals, availableEngineStr))
+	}
+	return paths, err
+}
+
 func removeIgnoredPaths(paths, ignoredPaths []*accessPath, tblInfo *model.TableInfo) []*accessPath {
 	if len(ignoredPaths) == 0 {
 		return paths

session/session.go

@@ -1811,6 +1811,7 @@ var builtinGlobalVariable = []string{
 	variable.TiDBStmtSummaryHistorySize,
 	variable.TiDBMaxDeltaSchemaCount,
 	variable.TiDBStoreLimit,
+	variable.TiDBIsolationReadEngines,
 }
 
 var (

sessionctx/variable/session.go

@@ -457,6 +457,9 @@ type SessionVars struct {
 	// LockWaitTimeout is the duration waiting for pessimistic lock in milliseconds
 	// negative value means nowait, 0 means default behavior, others means actual wait time
 	LockWaitTimeout int64
+
+	// isolationReadEngines is used to isolation read, tidb only read from the stores whose engine type is in the engines.
+	isolationReadEngines map[kv.StoreType]struct{}
 }
 
 // PreparedParams contains the parameters of the current prepared statement when executing it.
@@ -530,6 +533,7 @@ func NewSessionVars() *SessionVars {
 		ReplicaRead:                 kv.ReplicaReadLeader,
 		AllowRemoveAutoInc:          DefTiDBAllowRemoveAutoInc,
 		LockWaitTimeout:             DefInnodbLockWaitTimeout * 1000,
+		isolationReadEngines:        map[kv.StoreType]struct{}{kv.TiKV: {}, kv.TiFlash: {}},
 	}
 	vars.Concurrency = Concurrency{
 		IndexLookupConcurrency:     DefIndexLookupConcurrency,
@@ -579,6 +583,11 @@ func (s *SessionVars) GetSplitRegionTimeout() time.Duration {
 	return time.Duration(s.WaitSplitRegionTimeout) * time.Second
 }
 
+// GetIsolationReadEngines gets isolation read engines.
+func (s *SessionVars) GetIsolationReadEngines() map[kv.StoreType]struct{} {
+	return s.isolationReadEngines
+}
+
 // CleanBuffers cleans the temporary bufs
 func (s *SessionVars) CleanBuffers() {
 	if !s.LightningMode {
@@ -925,6 +934,16 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
 		}
 	case TiDBStoreLimit:
 		storeutil.StoreLimit.Store(tidbOptInt64(val, DefTiDBStoreLimit))
+	case TiDBIsolationReadEngines:
+		s.isolationReadEngines = make(map[kv.StoreType]struct{})
+		for _, engine := range strings.Split(val, ",") {
+			switch engine {
+			case kv.TiKV.Name():
+				s.isolationReadEngines[kv.TiKV] = struct{}{}
+			case kv.TiFlash.Name():
+				s.isolationReadEngines[kv.TiFlash] = struct{}{}
+			}
+		}
 	}
 	s.systems[name] = val
 	return nil

sessionctx/variable/sysvar.go

@@ -727,6 +727,7 @@ var defaultSysVars = []*SysVar{
 	{ScopeGlobal | ScopeSession, TiDBStmtSummaryRefreshInterval, strconv.Itoa(config.GetGlobalConfig().StmtSummary.RefreshInterval)},
 	{ScopeGlobal | ScopeSession, TiDBStmtSummaryHistorySize, strconv.Itoa(config.GetGlobalConfig().StmtSummary.HistorySize)},
 	{ScopeGlobal | ScopeSession, TiDBStoreLimit, strconv.FormatInt(atomic.LoadInt64(&config.GetGlobalConfig().TiKVClient.StoreLimit), 10)},
+	{ScopeGlobal | ScopeSession, TiDBIsolationReadEngines, "tikv,tiflash"},
 }
 
 // SynonymsSysVariables is synonyms of system variables.

sessionctx/variable/tidb_vars.go

@@ -323,6 +323,10 @@ const (
 
 	// TiDBStoreLimit indicates the limit of sending request to a store, 0 means without limit.
 	TiDBStoreLimit = "tidb_store_limit"
+
+	// TiDBIsolationReadEngines indicates the tidb only read from the stores whose engine type is involved in IsolationReadEngines.
+	// Now, only support TiKV and TiFlash.
+	TiDBIsolationReadEngines = "tidb_isolation_read_engines"
 )
 
 // Default TiDB system variable values.

sessionctx/variable/varsutil.go

@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/parser/ast"
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/tidb/config"
+	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/timeutil"
 )
@@ -644,6 +645,24 @@ func ValidateSetSystemVar(vars *SessionVars, name string, value string) (string,
 			return "", nil
 		}
 		return checkUInt64SystemVar(name, value, 0, math.MaxUint8, vars)
+	case TiDBIsolationReadEngines:
+		engines := strings.Split(value, ",")
+		var formatVal string
+		for i, engine := range engines {
+			engine = strings.TrimSpace(engine)
+			if i != 0 {
+				formatVal += ","
+			}
+			switch {
+			case strings.EqualFold(engine, kv.TiKV.Name()):
+				formatVal += kv.TiKV.Name()
+			case strings.EqualFold(engine, kv.TiFlash.Name()):
+				formatVal += kv.TiFlash.Name()
+			default:
+				return value, ErrWrongValueForVar.GenWithStackByArgs(name, value)
+			}
+		}
+		return formatVal, nil
 	}
 	return value, nil
 }

sessionctx/variable/varsutil_test.go

@@ -463,6 +463,10 @@ func (s *testVarsutilSuite) TestValidate(c *C) {
 		{TiDBStmtSummaryRefreshInterval, "", false},
 		{TiDBStmtSummaryHistorySize, "a", true},
 		{TiDBStmtSummaryHistorySize, "", false},
+		{TiDBIsolationReadEngines, "", true},
+		{TiDBIsolationReadEngines, "tikv", false},
+		{TiDBIsolationReadEngines, "TiKV,tiflash", false},
+		{TiDBIsolationReadEngines, "   tikv,   tiflash  ", false},
 	}
 
 	for _, t := range tests {

store/tikv/region_cache.go

@@ -1139,7 +1139,7 @@ func (s *Store) initResolve(bo *Backoffer, c *RegionCache) (addr string, err err
 		s.storeType = kv.TiKV
 		for _, label := range store.Labels {
 			if label.Key == "engine" {
-				if label.Value == "tiflash" {
+				if label.Value == kv.TiFlash.Name() {
 					s.storeType = kv.TiFlash
 				}
 				break
@@ -1184,7 +1184,7 @@ func (s *Store) reResolve(c *RegionCache) {
 	storeType := kv.TiKV
 	for _, label := range store.Labels {
 		if label.Key == "engine" {
-			if label.Value == "tiflash" {
+			if label.Value == kv.TiFlash.Name() {
 				storeType = kv.TiFlash
 			}
 			break