domain: load system keyspace info schema

pkg/ddl/db_test.go

@@ -759,7 +759,7 @@ func TestSchemaValidator(t *testing.T) {
 
 	// For schema check, it tests for getting the result of "ResultUnknown".
 	is := dom.InfoSchema()
-	schemaChecker := domain.NewSchemaChecker(dom, is.SchemaMetaVersion(), nil, true)
+	schemaChecker := domain.NewSchemaChecker(dom.GetSchemaValidator(), is.SchemaMetaVersion(), nil, true)
 	// Make sure it will retry one time and doesn't take a long time.
 	domain.SchemaOutOfDateRetryTimes.Store(1)
 	domain.SchemaOutOfDateRetryInterval.Store(time.Millisecond * 1)

pkg/domain/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "//br/pkg/streamhelper/daemon",
         "//pkg/bindinfo",
         "//pkg/config",
+        "//pkg/config/kerneltype",
         "//pkg/ddl",
         "//pkg/ddl/notifier",
         "//pkg/ddl/placement",
@@ -143,9 +144,10 @@ go_test(
     ],
     embed = [":domain"],
     flaky = True,
-    shard_count = 28,
+    shard_count = 29,
     deps = [
         "//pkg/config",
+        "//pkg/config/kerneltype",
         "//pkg/ddl",
         "//pkg/domain/infosync",
         "//pkg/errno",
@@ -177,6 +179,7 @@ go_test(
         "@com_github_ngaut_pools//:pools",
         "@com_github_pingcap_errors//:errors",
         "@com_github_pingcap_failpoint//:failpoint",
+        "@com_github_pingcap_kvproto//pkg/keyspacepb",
         "@com_github_pingcap_kvproto//pkg/metapb",
         "@com_github_pingcap_kvproto//pkg/resource_manager",
         "@com_github_prometheus_client_model//go",

pkg/domain/domain.go

@@ -39,6 +39,7 @@ import (
 	"github.com/pingcap/tidb/br/pkg/streamhelper/daemon"
 	"github.com/pingcap/tidb/pkg/bindinfo"
 	"github.com/pingcap/tidb/pkg/config"
+	"github.com/pingcap/tidb/pkg/config/kerneltype"
 	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/ddl/notifier"
 	"github.com/pingcap/tidb/pkg/ddl/placement"
@@ -83,7 +84,7 @@ import (
 	"github.com/pingcap/tidb/pkg/statistics/handle/initstats"
 	statslogutil "github.com/pingcap/tidb/pkg/statistics/handle/logutil"
 	handleutil "github.com/pingcap/tidb/pkg/statistics/handle/util"
-	"github.com/pingcap/tidb/pkg/store"
+	kvstore "github.com/pingcap/tidb/pkg/store"
 	"github.com/pingcap/tidb/pkg/ttl/ttlworker"
 	"github.com/pingcap/tidb/pkg/types"
 	"github.com/pingcap/tidb/pkg/util"
@@ -219,6 +220,11 @@ type Domain struct {
 	instancePlanCache sessionctx.InstancePlanCache // the instance level plan cache
 
 	statsOwner owner.Manager
+
+	// only used for nextgen
+	sysksInfoCache *infoschema.InfoCache
+	sysksISLoader  *issyncer.Loader
+	sysksSessPool  util.DestroyableSessionPool
 }
 
 // InfoCache export for test.
@@ -566,11 +572,19 @@ const resourceIdleTimeout = 3 * time.Minute // resources in the ResourcePool wil
 
 // NewDomain creates a new domain. Should not create multiple domains for the same store.
 func NewDomain(store kv.Storage, schemaLease time.Duration, statsLease time.Duration, dumpFileGcLease time.Duration, factory pools.Factory) *Domain {
-	return NewDomainWithEtcdClient(store, schemaLease, statsLease, dumpFileGcLease, factory, nil)
+	return NewDomainWithEtcdClient(store, schemaLease, statsLease, dumpFileGcLease, factory, nil, nil)
 }
 
 // NewDomainWithEtcdClient creates a new domain with etcd client. Should not create multiple domains for the same store.
-func NewDomainWithEtcdClient(store kv.Storage, schemaLease time.Duration, statsLease time.Duration, dumpFileGcLease time.Duration, factory pools.Factory, etcdClient *clientv3.Client) *Domain {
+func NewDomainWithEtcdClient(
+	store kv.Storage,
+	schemaLease time.Duration,
+	statsLease time.Duration,
+	dumpFileGcLease time.Duration,
+	factory pools.Factory,
+	ksSessFactoryGetter func(currKSStore kv.Storage, targetKS string) pools.Factory,
+	etcdClient *clientv3.Client,
+) *Domain {
 	intest.Assert(schemaLease > 0, "schema lease should be a positive duration")
 	capacity := 200 // capacity of the sysSessionPool size
 	do := &Domain{
@@ -632,9 +646,38 @@ func NewDomainWithEtcdClient(store kv.Storage, schemaLease time.Duration, statsL
 		do.sysSessionPool,
 	)
 	do.initDomainSysVars()
+
+	if shouldLoadSysKSAdditionally(store) {
+		do.sysksInfoCache = infoschema.NewCache(kvstore.GetSystemStorage(), int(vardef.SchemaVersionCacheLimit.Load()))
+		do.sysksISLoader = issyncer.NewLoader(kvstore.GetSystemStorage(), do.sysksInfoCache)
+		// TODO register to a separate session manager when we start syncer for
+		// system keyspace.
+		do.sysksSessPool = util.NewSessionPool(
+			capacity, ksSessFactoryGetter(store, keyspace.System),
+			func(r pools.Resource) {
+				_, ok := r.(sessionctx.Context)
+				intest.Assert(ok)
+			},
+			func(r pools.Resource) {
+				sctx, ok := r.(sessionctx.Context)
+				intest.Assert(ok)
+				intest.AssertFunc(func() bool {
+					txn, _ := sctx.Txn(false)
+					return txn == nil || !txn.Valid()
+				})
+			},
+			func(r pools.Resource) {
+				intest.Assert(r != nil)
+			},
+		)
+	}
 	return do
 }
 
+func shouldLoadSysKSAdditionally(store kv.Storage) bool {
+	return kerneltype.IsNextGen() && store.GetKeyspace() != keyspace.System
+}
+
 const serverIDForStandalone = 1 // serverID for standalone deployment.
 
 // Init initializes a domain. after return, session can be used to do DMLs but not
@@ -645,12 +688,12 @@ func (do *Domain) Init(
 ) error {
 	do.sysExecutorFactory = sysExecutorFactory
 	perfschema.Init()
-	etcdStore, addrs, err := store.GetEtcdAddrs(do.store)
+	etcdStore, addrs, err := kvstore.GetEtcdAddrs(do.store)
 	if err != nil {
 		return errors.Trace(err)
 	}
 	if len(addrs) > 0 {
-		cli, err2 := store.NewEtcdCliWithAddrs(addrs, etcdStore)
+		cli, err2 := kvstore.NewEtcdCliWithAddrs(addrs, etcdStore)
 		if err2 != nil {
 			return errors.Trace(err2)
 		}
@@ -660,7 +703,7 @@ func (do *Domain) Init(
 
 		do.autoidClient = autoid.NewClientDiscover(cli)
 
-		unprefixedEtcdCli, err2 := store.NewEtcdCliWithAddrs(addrs, etcdStore)
+		unprefixedEtcdCli, err2 := kvstore.NewEtcdCliWithAddrs(addrs, etcdStore)
 		if err2 != nil {
 			return errors.Trace(err2)
 		}
@@ -818,9 +861,51 @@ func (do *Domain) Start(startMode ddl.StartMode) error {
 		return err
 	}
 
+	// right now we only allow access system keyspace info schema after fully bootstrap.
+	if shouldLoadSysKSAdditionally(do.store) && startMode == ddl.Normal {
+		if err = do.loadSysKSInfoSchema(); err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
+// TODO: we should sync the system keyspace info schema, not just load it,
+// currently, we assume there is no upgrade, so we only load the info schema of
+// system keyspace once, and it will not change during the lifetime of the domain,
+// it's not right, but it's enough to push subtasks which depends on it forward,
+// we will fix it in the future.
+func (do *Domain) loadSysKSInfoSchema() error {
+	logutil.BgLogger().Info("loading system keyspace info schema")
+	sysksStore := kvstore.GetSystemStorage()
+	ver, err := sysksStore.CurrentVersion(kv.GlobalTxnScope)
+	if err != nil {
+		return err
+	}
+
+	_, _, _, _, err = do.sysksISLoader.LoadWithTS(ver.Ver, false)
+	return err
+}
+
+// GetKSStore returns the kv.Storage for the given keyspace.
+func (*Domain) GetKSStore(targetKS string) kv.Storage {
+	if targetKS == keyspace.System {
+		return kvstore.GetSystemStorage()
+	}
+	// TODO, support loading store for other keyspaces.
+	panic("invalid call to GetKSStore, only system keyspace is supported now")
+}
+
+// GetKSInfoCache returns the system keyspace info cache.
+func (do *Domain) GetKSInfoCache(targetKS string) *infoschema.InfoCache {
+	if targetKS == keyspace.System {
+		return do.sysksInfoCache
+	}
+	// TODO, support loading info schema for other keyspaces.
+	panic("invalid call to GetKSInfoCache, only system keyspace is supported now")
+}
+
 // GetSchemaLease return the schema lease.
 func (do *Domain) GetSchemaLease() time.Duration {
 	return do.schemaLease

pkg/domain/domain_test.go

@@ -27,9 +27,12 @@ import (
 	"github.com/ngaut/pools"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/kvproto/pkg/keyspacepb"
 	"github.com/pingcap/kvproto/pkg/metapb"
+	"github.com/pingcap/tidb/pkg/config/kerneltype"
 	"github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/domain/infosync"
+	"github.com/pingcap/tidb/pkg/keyspace"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/metrics"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -498,3 +501,19 @@ func TestIsAnalyzeTableSQL(t *testing.T) {
 		require.True(t, isAnalyzeTableSQL(tt.sql))
 	}
 }
+
+func TestShouldLoadSysKSAdditionally(t *testing.T) {
+	if kerneltype.IsClassic() {
+		require.False(t, shouldLoadSysKSAdditionally(nil))
+	} else {
+		ksMeta := keyspacepb.KeyspaceMeta{}
+		ksMeta.Id = 2
+		ksMeta.Name = "ks1"
+		s, err := mockstore.NewMockStore(mockstore.WithKeyspaceMeta(&ksMeta))
+		require.NoError(t, err)
+		require.True(t, shouldLoadSysKSAdditionally(s))
+
+		ksMeta.Name = keyspace.System
+		require.False(t, shouldLoadSysKSAdditionally(s))
+	}
+}

pkg/domain/domainctx.go

@@ -19,6 +19,7 @@ import (
 )
 
 // GetDomain gets domain from context.
+// might return nil if the session is a cross keyspace one.
 func GetDomain(ctx contextutil.ValueStoreContext) *Domain {
 	v, ok := ctx.GetDomain().(*Domain)
 	if ok {

pkg/domain/domainctx_test.go

@@ -23,11 +23,11 @@ import (
 
 func TestDomainCtx(t *testing.T) {
 	ctx := mock.NewContext()
-	ctx.BindDomain(nil)
+	ctx.BindDomainAndSchValidator(nil, nil)
 	v := GetDomain(ctx)
 	require.Nil(t, v)
 
-	ctx.BindDomain(&Domain{})
+	ctx.BindDomainAndSchValidator(&Domain{}, nil)
 	v = GetDomain(ctx)
 	require.NotNil(t, v)
 }

pkg/domain/schema_checker.go

@@ -46,9 +46,9 @@ var (
 )
 
 // NewSchemaChecker creates a new schema checker.
-func NewSchemaChecker(do *Domain, schemaVer int64, relatedTableIDs []int64, needCheckSchema bool) *SchemaChecker {
+func NewSchemaChecker(validator validatorapi.Validator, schemaVer int64, relatedTableIDs []int64, needCheckSchema bool) *SchemaChecker {
 	return &SchemaChecker{
-		Validator:       do.GetSchemaValidator(),
+		Validator:       validator,
 		schemaVer:       schemaVer,
 		relatedTableIDs: relatedTableIDs,
 		needCheckSchema: needCheckSchema,

pkg/executor/adapter.go

@@ -52,6 +52,7 @@ import (
 	"github.com/pingcap/tidb/pkg/planner/core/operator/logicalop"
 	"github.com/pingcap/tidb/pkg/planner/core/resolve"
 	"github.com/pingcap/tidb/pkg/plugin"
+	"github.com/pingcap/tidb/pkg/resourcegroup/runaway"
 	"github.com/pingcap/tidb/pkg/sessionctx"
 	"github.com/pingcap/tidb/pkg/sessionctx/sessionstates"
 	"github.com/pingcap/tidb/pkg/sessionctx/stmtctx"
@@ -556,7 +557,12 @@ func (a *ExecStmt) Exec(ctx context.Context) (_ sqlexec.RecordSet, err error) {
 
 	// must set plan according to the `Execute` plan before getting planDigest
 	a.inheritContextFromExecuteStmt()
-	if rm := domain.GetDomain(sctx).RunawayManager(); vardef.EnableResourceControl.Load() && rm != nil {
+	var rm *runaway.Manager
+	dom := domain.GetDomain(sctx)
+	if dom != nil {
+		rm = dom.RunawayManager()
+	}
+	if vardef.EnableResourceControl.Load() && rm != nil {
 		sessionVars := sctx.GetSessionVars()
 		stmtCtx := sessionVars.StmtCtx
 		_, planDigest := GetPlanDigest(stmtCtx)
@@ -1749,22 +1755,26 @@ func (a *ExecStmt) LogSlowQuery(txnTS uint64, succ bool, hasMoreResults bool) {
 		if len(stmtCtx.TableIDs) > 0 {
 			tableIDs = strings.ReplaceAll(fmt.Sprintf("%v", stmtCtx.TableIDs), " ", ",")
 		}
-		domain.GetDomain(a.Ctx).LogSlowQuery(&domain.SlowQueryInfo{
-			SQL:        sql.String(),
-			Digest:     digest.String(),
-			Start:      sessVars.StartTime,
-			Duration:   costTime,
-			Detail:     stmtCtx.GetExecDetails(),
-			Succ:       succ,
-			ConnID:     sessVars.ConnectionID,
-			SessAlias:  sessVars.SessionAlias,
-			TxnTS:      txnTS,
-			User:       userString,
-			DB:         sessVars.CurrentDB,
-			TableIDs:   tableIDs,
-			IndexNames: indexNames,
-			Internal:   sessVars.InRestrictedSQL,
-		})
+		// TODO log slow query for cross keyspace query?
+		dom := domain.GetDomain(a.Ctx)
+		if dom != nil {
+			dom.LogSlowQuery(&domain.SlowQueryInfo{
+				SQL:        sql.String(),
+				Digest:     digest.String(),
+				Start:      sessVars.StartTime,
+				Duration:   costTime,
+				Detail:     stmtCtx.GetExecDetails(),
+				Succ:       succ,
+				ConnID:     sessVars.ConnectionID,
+				SessAlias:  sessVars.SessionAlias,
+				TxnTS:      txnTS,
+				User:       userString,
+				DB:         sessVars.CurrentDB,
+				TableIDs:   tableIDs,
+				IndexNames: indexNames,
+				Internal:   sessVars.InRestrictedSQL,
+			})
+		}
 	}
 }
 

pkg/executor/ddl.go

@@ -63,7 +63,7 @@ type DDLExec struct {
 func (e *DDLExec) toErr(err error) error {
 	// The err may be cause by schema changed, here we distinguish the ErrInfoSchemaChanged error from other errors.
 	dom := domain.GetDomain(e.Ctx())
-	checker := domain.NewSchemaChecker(dom, e.is.SchemaMetaVersion(), nil, true)
+	checker := domain.NewSchemaChecker(dom.GetSchemaValidator(), e.is.SchemaMetaVersion(), nil, true)
 	txn, err1 := e.Ctx().Txn(true)
 	if err1 != nil {
 		logutil.BgLogger().Error("active txn failed", zap.Error(err1))

pkg/executor/executor_pkg_test.go

@@ -332,7 +332,7 @@ func TestFilterTemporaryTableKeys(t *testing.T) {
 
 func TestErrLevelsForResetStmtContext(t *testing.T) {
 	ctx := mock.NewContext()
-	ctx.BindDomain(&domain.Domain{})
+	ctx.BindDomainAndSchValidator(&domain.Domain{}, nil)
 
 	cases := []struct {
 		name    string

pkg/executor/executor_required_rows_test.go

@@ -216,7 +216,7 @@ func defaultCtx() sessionctx.Context {
 	ctx.GetSessionVars().StmtCtx.MemTracker = memory.NewTracker(-1, ctx.GetSessionVars().MemQuotaQuery)
 	ctx.GetSessionVars().StmtCtx.DiskTracker = disk.NewTracker(-1, -1)
 	ctx.GetSessionVars().SnapshotTS = uint64(1)
-	ctx.BindDomain(domain.NewMockDomain())
+	ctx.BindDomainAndSchValidator(domain.NewMockDomain(), nil)
 	return ctx
 }
 

pkg/executor/join/joiner_test.go

@@ -38,7 +38,7 @@ func defaultCtx() sessionctx.Context {
 	ctx.GetSessionVars().StmtCtx.MemTracker = memory.NewTracker(-1, ctx.GetSessionVars().MemQuotaQuery)
 	ctx.GetSessionVars().StmtCtx.DiskTracker = disk.NewTracker(-1, -1)
 	ctx.GetSessionVars().SnapshotTS = uint64(1)
-	ctx.BindDomain(domain.NewMockDomain())
+	ctx.BindDomainAndSchValidator(domain.NewMockDomain(), nil)
 	return ctx
 }
 

pkg/executor/select.go

@@ -975,7 +975,11 @@ func ResetContextOfStmt(ctx sessionctx.Context, s ast.StmtNode) (err error) {
 	} else {
 		sc.InitMemTracker(memory.LabelForSQLText, -1)
 	}
-	logOnQueryExceedMemQuota := domain.GetDomain(ctx).ExpensiveQueryHandle().LogOnQueryExceedMemQuota
+	sessDom := domain.GetDomain(ctx)
+	var logOnQueryExceedMemQuota func(uint64)
+	if sessDom != nil {
+		logOnQueryExceedMemQuota = sessDom.ExpensiveQueryHandle().LogOnQueryExceedMemQuota
+	}
 	switch vardef.OOMAction.Load() {
 	case vardef.OOMActionCancel:
 		action := &memory.PanicOnExceed{ConnID: vars.ConnectionID, Killer: vars.MemTracker.Killer}