Merge branch 'master' into remove-non-dry-sysvar-code

DEPS.bzl

@@ -4453,8 +4453,8 @@ def go_deps():
         name = "org_uber_go_goleak",
         build_file_proto_mode = "disable_global",
         importpath = "go.uber.org/goleak",
-        sum = "h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA=",
-        version = "v1.1.12",
+        sum = "h1:xqgm/S+aQvhWFTtR0XK3Jvg7z8kGV8P4X14IzwN3Eqk=",
+        version = "v1.2.0",
     )
     go_repository(
         name = "org_uber_go_multierr",

br/pkg/streamhelper/daemon/BUILD.bazel

@@ -0,0 +1,29 @@
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
+
+go_library(
+    name = "daemon",
+    srcs = [
+        "interface.go",
+        "owner_daemon.go",
+    ],
+    importpath = "github.com/pingcap/tidb/br/pkg/streamhelper/daemon",
+    visibility = ["//visibility:public"],
+    deps = [
+        "//br/pkg/logutil",
+        "//owner",
+        "@com_github_pingcap_log//:log",
+        "@org_uber_go_zap//:zap",
+    ],
+)
+
+go_test(
+    name = "daemon_test",
+    srcs = ["owner_daemon_test.go"],
+    flaky = True,
+    deps = [
+        ":daemon",
+        "//owner",
+        "@com_github_pingcap_log//:log",
+        "@com_github_stretchr_testify//require",
+    ],
+)

br/pkg/task/BUILD.bazel

@@ -36,6 +36,7 @@ go_library(
         "//br/pkg/stream",
         "//br/pkg/streamhelper",
         "//br/pkg/streamhelper/config",
+        "//br/pkg/streamhelper/daemon",
         "//br/pkg/summary",
         "//br/pkg/utils",
         "//br/pkg/version",

ddl/ingest/BUILD.bazel

@@ -32,6 +32,7 @@ go_library(
         "//parser/mysql",
         "//sessionctx/variable",
         "//table",
+        "//util",
         "//util/generic",
         "//util/logutil",
         "//util/mathutil",

domain/BUILD.bazel

@@ -19,6 +19,7 @@ go_library(
     deps = [
         "//bindinfo",
         "//br/pkg/streamhelper",
+        "//br/pkg/streamhelper/daemon",
         "//config",
         "//ddl",
         "//ddl/placement",
@@ -51,6 +52,7 @@ go_library(
         "//util/execdetails",
         "//util/expensivequery",
         "//util/logutil",
+        "//util/servermemorylimit",
         "//util/sqlexec",
         "@com_github_ngaut_pools//:pools",
         "@com_github_pingcap_errors//:errors",

domain/domain.go

@@ -64,6 +64,7 @@ import (
 	"github.com/pingcap/tidb/util/engine"
 	"github.com/pingcap/tidb/util/expensivequery"
 	"github.com/pingcap/tidb/util/logutil"
+	"github.com/pingcap/tidb/util/servermemorylimit"
 	"github.com/pingcap/tidb/util/sqlexec"
 	"github.com/tikv/client-go/v2/txnkv/transaction"
 	pd "github.com/tikv/pd/client"
@@ -87,30 +88,31 @@ func NewMockDomain() *Domain {
 // Domain represents a storage space. Different domains can use the same database name.
 // Multiple domains can be used in parallel without synchronization.
 type Domain struct {
-	store                kv.Storage
-	infoCache            *infoschema.InfoCache
-	privHandle           *privileges.Handle
-	bindHandle           *bindinfo.BindHandle
-	statsHandle          unsafe.Pointer
-	statsLease           time.Duration
-	ddl                  ddl.DDL
-	info                 *infosync.InfoSyncer
-	globalCfgSyncer      *globalconfigsync.GlobalConfigSyncer
-	m                    sync.Mutex
-	SchemaValidator      SchemaValidator
-	sysSessionPool       *sessionPool
-	exit                 chan struct{}
-	etcdClient           *clientv3.Client
-	sysVarCache          sysVarCache // replaces GlobalVariableCache
-	slowQuery            *topNSlowQueries
-	expensiveQueryHandle *expensivequery.Handle
-	wg                   util.WaitGroupWrapper
-	statsUpdating        atomicutil.Int32
-	cancel               context.CancelFunc
-	indexUsageSyncLease  time.Duration
-	dumpFileGcChecker    *dumpFileGcChecker
-	expiredTimeStamp4PC  types.Time
-	logBackupAdvancer    *daemon.OwnerDaemon
+	store                   kv.Storage
+	infoCache               *infoschema.InfoCache
+	privHandle              *privileges.Handle
+	bindHandle              *bindinfo.BindHandle
+	statsHandle             unsafe.Pointer
+	statsLease              time.Duration
+	ddl                     ddl.DDL
+	info                    *infosync.InfoSyncer
+	globalCfgSyncer         *globalconfigsync.GlobalConfigSyncer
+	m                       sync.Mutex
+	SchemaValidator         SchemaValidator
+	sysSessionPool          *sessionPool
+	exit                    chan struct{}
+	etcdClient              *clientv3.Client
+	sysVarCache             sysVarCache // replaces GlobalVariableCache
+	slowQuery               *topNSlowQueries
+	expensiveQueryHandle    *expensivequery.Handle
+	serverMemoryLimitHandle *servermemorylimit.Handle
+	wg                      util.WaitGroupWrapper
+	statsUpdating           atomicutil.Int32
+	cancel                  context.CancelFunc
+	indexUsageSyncLease     time.Duration
+	dumpFileGcChecker       *dumpFileGcChecker
+	expiredTimeStamp4PC     types.Time
+	logBackupAdvancer       *daemon.OwnerDaemon
 
 	serverID             uint64
 	serverIDSession      *concurrency.Session
@@ -885,6 +887,7 @@ func NewDomain(store kv.Storage, ddlLease time.Duration, statsLease time.Duratio
 
 	do.SchemaValidator = NewSchemaValidator(ddlLease, do)
 	do.expensiveQueryHandle = expensivequery.NewExpensiveQueryHandle(do.exit)
+	do.serverMemoryLimitHandle = servermemorylimit.NewServerMemoryLimitHandle(do.exit)
 	do.sysProcesses = SysProcesses{mu: &sync.RWMutex{}, procMap: make(map[uint64]sessionctx.Context)}
 	do.initDomainSysVars()
 	return do
@@ -1818,6 +1821,11 @@ func (do *Domain) ExpensiveQueryHandle() *expensivequery.Handle {
 	return do.expensiveQueryHandle
 }
 
+// ServerMemoryLimitHandle returns the expensive query handle.
+func (do *Domain) ServerMemoryLimitHandle() *servermemorylimit.Handle {
+	return do.serverMemoryLimitHandle
+}
+
 const (
 	privilegeKey   = "/tidb/privilege"
 	sysVarCacheKey = "/tidb/sysvars"

domain/infosync/BUILD.bazel

@@ -28,7 +28,6 @@ go_library(
         "//sessionctx/variable",
         "//store/helper",
         "//tablecodec",
-        "//types",
         "//util",
         "//util/codec",
         "//util/dbterror",

executor/BUILD.bazel

@@ -36,6 +36,7 @@ go_library(
         "errors.go",
         "executor.go",
         "explain.go",
+        "foreign_key.go",
         "grant.go",
         "hash_table.go",
         "index_advise.go",
@@ -270,6 +271,7 @@ go_test(
         "explain_test.go",
         "explain_unit_test.go",
         "explainfor_test.go",
+        "foreign_key_test.go",
         "grant_test.go",
         "hash_table_test.go",
         "hot_regions_history_table_test.go",

executor/analyzetest/BUILD.bazel

@@ -8,6 +8,7 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
+    race = "on",
     shard_count = 50,
     deps = [
         "//domain",

executor/executor.go

@@ -1943,6 +1943,7 @@ func ResetContextOfStmt(ctx sessionctx.Context, s ast.StmtNode) (err error) {
 	} else {
 		sc.InitMemTracker(memory.LabelForSQLText, vars.MemQuotaQuery)
 		sc.MemTracker.AttachToGlobalTracker(GlobalMemoryUsageTracker)
+		sc.MemTracker.IsRootTrackerOfSess, sc.MemTracker.SessionID = true, vars.ConnectionID
 	}
 
 	sc.InitDiskTracker(memory.LabelForSQLText, -1)

executor/executor_test.go

@@ -6063,3 +6063,66 @@ func TestTableLockPrivilege(t *testing.T) {
 	tk2.MustExec("LOCK TABLE test.t WRITE, test2.t2 WRITE")
 	tk.MustExec("LOCK TABLE test.t WRITE, test2.t2 WRITE")
 }
+
+func TestGlobalMemoryControl(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+
+	tk0 := testkit.NewTestKit(t, store)
+	tk0.MustExec("set global tidb_mem_oom_action = 'cancel'")
+	tk0.MustExec("set global tidb_server_memory_limit = 512 << 20")
+	tk0.MustExec("set global tidb_server_memory_limit_sess_min_size = 128")
+
+	tk1 := testkit.NewTestKit(t, store)
+	tracker1 := tk1.Session().GetSessionVars().StmtCtx.MemTracker
+
+	tk2 := testkit.NewTestKit(t, store)
+	tracker2 := tk2.Session().GetSessionVars().StmtCtx.MemTracker
+
+	tk3 := testkit.NewTestKit(t, store)
+	tracker3 := tk3.Session().GetSessionVars().StmtCtx.MemTracker
+
+	sm := &testkit.MockSessionManager{
+		PS: []*util.ProcessInfo{tk1.Session().ShowProcess(), tk2.Session().ShowProcess(), tk3.Session().ShowProcess()},
+	}
+	dom.ServerMemoryLimitHandle().SetSessionManager(sm)
+	go dom.ServerMemoryLimitHandle().Run()
+
+	tracker1.Consume(100 << 20) // 100 MB
+	tracker2.Consume(200 << 20) // 200 MB
+	tracker3.Consume(300 << 20) // 300 MB
+
+	test := make([]int, 128<<20)       // Keep 1GB HeapInUse
+	time.Sleep(500 * time.Millisecond) // The check goroutine checks the memory usage every 100ms. The Sleep() make sure that Top1Tracker can be Canceled.
+
+	// Kill Top1
+	require.False(t, tracker1.NeedKill.Load())
+	require.False(t, tracker2.NeedKill.Load())
+	require.True(t, tracker3.NeedKill.Load())
+	require.Equal(t, memory.MemUsageTop1Tracker.Load(), tracker3)
+	util.WithRecovery( // Next Consume() will panic and cancel the SQL
+		func() {
+			tracker3.Consume(1)
+		}, func(r interface{}) {
+			require.True(t, strings.Contains(r.(string), "Out Of Memory Quota!"))
+		})
+	tracker2.Consume(300 << 20) // Sum 500MB, Not Panic, Waiting t3 cancel finish.
+	time.Sleep(500 * time.Millisecond)
+	require.False(t, tracker2.NeedKill.Load())
+	// Kill Finished
+	tracker3.Consume(-(300 << 20))
+	// Simulated SQL is Canceled and the time is updated
+	sm.PSMu.Lock()
+	ps := *sm.PS[2]
+	ps.Time = time.Now()
+	sm.PS[2] = &ps
+	sm.PSMu.Unlock()
+	time.Sleep(500 * time.Millisecond)
+	// Kill the Next SQL
+	util.WithRecovery( // Next Consume() will panic and cancel the SQL
+		func() {
+			tracker2.Consume(1)
+		}, func(r interface{}) {
+			require.True(t, strings.Contains(r.(string), "Out Of Memory Quota!"))
+		})
+	require.Equal(t, test[0], 0) // Keep 1GB HeapInUse
+}

executor/memtest/BUILD.bazel

@@ -8,6 +8,7 @@ go_test(
         "mem_test.go",
     ],
     flaky = True,
+    race = "on",
     deps = [
         "//config",
         "//meta/autoid",

executor/seqtest/prepared_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/tidb/errno"
 	"github.com/pingcap/tidb/executor"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/metrics"
@@ -53,28 +54,26 @@ func TestPrepared(t *testing.T) {
 		tk.MustExec(`execute stmt_test_1 using @a;`)
 		tk.MustExec(`prepare stmt_test_2 from 'select 1'`)
 		// Prepare multiple statement is not allowed.
-		_, err = tk.Exec(`prepare stmt_test_3 from 'select id from prepare_test where id > ?;select id from prepare_test where id > ?;'`)
-		require.True(t, executor.ErrPrepareMulti.Equal(err))
+		tk.MustGetErrCode(`prepare stmt_test_3 from 'select id from prepare_test where id > ?;select id from prepare_test where id > ?;'`, errno.ErrPrepareMulti)
 
 		// The variable count does not match.
 		tk.MustExec(`prepare stmt_test_4 from 'select id from prepare_test where id > ? and id < ?';`)
 		tk.MustExec(`set @a = 1;`)
-		_, err = tk.Exec(`execute stmt_test_4 using @a;`)
-		require.True(t, plannercore.ErrWrongParamCount.Equal(err))
+		tk.MustGetErrCode(`execute stmt_test_4 using @a;`, errno.ErrWrongParamCount)
 		// Prepare and deallocate prepared statement immediately.
 		tk.MustExec(`prepare stmt_test_5 from 'select id from prepare_test where id > ?';`)
 		tk.MustExec(`deallocate prepare stmt_test_5;`)
 
 		// Statement not found.
-		_, err = tk.Exec("deallocate prepare stmt_test_5")
+		err = tk.ExecToErr("deallocate prepare stmt_test_5")
 		require.True(t, plannercore.ErrStmtNotFound.Equal(err))
 
 		// incorrect SQLs in prepare. issue #3738, SQL in prepare stmt is parsed in DoPrepare.
-		_, err = tk.Exec(`prepare p from "delete from t where a = 7 or 1=1/*' and b = 'p'";`)
-		require.EqualError(t, err, `[parser:1064]You have an error in your SQL syntax; check the manual that corresponds to your TiDB version for the right syntax to use near '/*' and b = 'p'' at line 1`)
+		tk.MustGetErrMsg(`prepare p from "delete from t where a = 7 or 1=1/*' and b = 'p'";`,
+			`[parser:1064]You have an error in your SQL syntax; check the manual that corresponds to your TiDB version for the right syntax to use near '/*' and b = 'p'' at line 1`)
 
 		// The `stmt_test5` should not be found.
-		_, err = tk.Exec(`set @a = 1; execute stmt_test_5 using @a;`)
+		err = tk.ExecToErr(`set @a = 1; execute stmt_test_5 using @a;`)
 		require.True(t, plannercore.ErrStmtNotFound.Equal(err))
 
 		// Use parameter marker with argument will run prepared statement.
@@ -158,8 +157,7 @@ func TestPrepared(t *testing.T) {
 
 		// Make schema change.
 		tk.MustExec("drop table if exists prepare2")
-		_, err = tk.Exec("create table prepare2 (a int)")
-		require.NoError(t, err)
+		tk.MustExec("create table prepare2 (a int)")
 
 		// Should success as the changed schema do not affect the prepared statement.
 		rs, err = tk.Session().ExecutePreparedStmt(ctx, stmtID, expression.Args2Expressions4Test(1))
@@ -185,8 +183,7 @@ func TestPrepared(t *testing.T) {
 		tk.MustExec("drop table if exists prepare3")
 		tk.MustExec("create table prepare3 (a decimal(1))")
 		tk.MustExec("prepare stmt from 'insert into prepare3 value(123)'")
-		_, err = tk.Exec("execute stmt")
-		require.Error(t, err)
+		tk.MustExecToErr("execute stmt")
 
 		_, _, fields, err := tk.Session().PrepareStmt("select a from prepare3")
 		require.NoError(t, err)
@@ -230,27 +227,21 @@ func TestPrepared(t *testing.T) {
 		tk.MustExec("drop table if exists prepare1;")
 		tk.MustExec("create table prepare1 (a decimal(1))")
 		tk.MustExec("insert into prepare1 values(1);")
-		_, err = tk.Exec("prepare stmt FROM @sql1")
-		require.EqualError(t, err, "[parser:1064]You have an error in your SQL syntax; check the manual that corresponds to your TiDB version for the right syntax to use line 1 column 4 near \"NULL\" ")
+		tk.MustGetErrMsg("prepare stmt FROM @sql1",
+			"[parser:1064]You have an error in your SQL syntax; check the manual that corresponds to your TiDB version for the right syntax to use line 1 column 4 near \"NULL\" ")
 		tk.MustExec("SET @sql = 'update prepare1 set a=5 where a=?';")
-		_, err = tk.Exec("prepare stmt FROM @sql")
-		require.NoError(t, err)
+		tk.MustExec("prepare stmt FROM @sql")
 		tk.MustExec("set @var=1;")
-		_, err = tk.Exec("execute stmt using @var")
-		require.NoError(t, err)
+		tk.MustExec("execute stmt using @var")
 		tk.MustQuery("select a from prepare1;").Check(testkit.Rows("5"))
 
 		// issue 19371
 		tk.MustExec("SET @sql = 'update prepare1 set a=a+1';")
-		_, err = tk.Exec("prepare stmt FROM @SQL")
-		require.NoError(t, err)
-		_, err = tk.Exec("execute stmt")
-		require.NoError(t, err)
+		tk.MustExec("prepare stmt FROM @SQL")
+		tk.MustExec("execute stmt")
 		tk.MustQuery("select a from prepare1;").Check(testkit.Rows("6"))
-		_, err = tk.Exec("prepare stmt FROM @Sql")
-		require.NoError(t, err)
-		_, err = tk.Exec("execute stmt")
-		require.NoError(t, err)
+		tk.MustExec("prepare stmt FROM @Sql")
+		tk.MustExec("execute stmt")
 		tk.MustQuery("select a from prepare1;").Check(testkit.Rows("7"))
 
 		// Coverage.

executor/showtest/show_test.go

@@ -1472,7 +1472,7 @@ func TestShowBuiltin(t *testing.T) {
 	res := tk.MustQuery("show builtins;")
 	require.NotNil(t, res)
 	rows := res.Rows()
-	const builtinFuncNum = 281
+	const builtinFuncNum = 282
 	require.Equal(t, builtinFuncNum, len(rows))
 	require.Equal(t, rows[0][0].(string), "abs")
 	require.Equal(t, rows[builtinFuncNum-1][0].(string), "yearweek")

executor/simpletest/BUILD.bazel

@@ -8,6 +8,7 @@ go_test(
         "simple_test.go",
     ],
     flaky = True,
+    race = "on",
     shard_count = 30,
     deps = [
         "//config",