Merge remote-tracking branch 'pingcap/release-6.5' into cherry-pick-3…

…8738-6.5

dumpling/export/BUILD.bazel

@@ -106,6 +106,7 @@ go_test(
         "@com_github_data_dog_go_sqlmock//:go-sqlmock",
         "@com_github_go_sql_driver_mysql//:mysql",
         "@com_github_pingcap_errors//:errors",
+        "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_prometheus_client_golang//prometheus/collectors",
         "@com_github_stretchr_testify//require",
         "@org_golang_x_sync//errgroup",

dumpling/export/dump.go

@@ -1531,7 +1531,7 @@ func setSessionParam(d *Dumper) error {
 				d.L().Info("cannot check whether TiDB has TiKV, will apply tidb_snapshot by default. This won't affect dump process", log.ShortError(err))
 			}
 			if conf.ServerInfo.HasTiKV {
-				sessionParam["tidb_snapshot"] = snapshot
+				sessionParam[snapshotVar] = snapshot
 			}
 		}
 	}

dumpling/export/dump_test.go

@@ -9,7 +9,9 @@ import (
 	"time"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/br/pkg/version"
 	tcontext "github.com/pingcap/tidb/dumpling/context"
 	"github.com/pingcap/tidb/parser"
@@ -224,3 +226,64 @@ func TestUnregisterMetrics(t *testing.T) {
 	// should not panic
 	require.Error(t, err)
 }
+
+func TestSetSessionParams(t *testing.T) {
+	// case 1: fail to set tidb_snapshot, should return error with hint
+	db, mock, err := sqlmock.New()
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, db.Close())
+	}()
+
+	mock.ExpectQuery("SELECT @@tidb_config").
+		WillReturnError(errors.New("mock error"))
+	mock.ExpectQuery("SELECT COUNT\\(1\\) as c FROM MYSQL.TiDB WHERE VARIABLE_NAME='tikv_gc_safe_point'").
+		WillReturnError(errors.New("mock error"))
+	tikvErr := &mysql.MySQLError{
+		Number:  1105,
+		Message: "can not get 'tikv_gc_safe_point'",
+	}
+	mock.ExpectExec("SET SESSION tidb_snapshot").
+		WillReturnError(tikvErr)
+
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/dumpling/export/SkipResetDB", "return(true)"))
+	defer failpoint.Disable("github.com/pingcap/tidb/dumpling/export/SkipResetDB=return(true)")
+
+	tctx, cancel := tcontext.Background().WithLogger(appLogger).WithCancel()
+	defer cancel()
+
+	conf := DefaultConfig()
+	conf.ServerInfo = version.ServerInfo{
+		ServerType: version.ServerTypeTiDB,
+		HasTiKV:    false,
+	}
+	conf.Snapshot = "439153276059648000"
+	conf.Consistency = ConsistencyTypeSnapshot
+	d := &Dumper{
+		tctx:      tctx,
+		conf:      conf,
+		cancelCtx: cancel,
+		dbHandle:  db,
+	}
+	err = setSessionParam(d)
+	require.ErrorContains(t, err, "consistency=none")
+
+	// case 2: fail to set other
+	conf.ServerInfo = version.ServerInfo{
+		ServerType: version.ServerTypeMySQL,
+		HasTiKV:    false,
+	}
+	conf.Snapshot = ""
+	conf.Consistency = ConsistencyTypeFlush
+	conf.SessionParams = map[string]interface{}{
+		"mock": "UTC",
+	}
+	d.dbHandle = db
+	mock.ExpectExec("SET SESSION mock").
+		WillReturnError(errors.New("Unknown system variable mock"))
+	mock.ExpectClose()
+	mock.ExpectClose()
+
+	err = setSessionParam(d)
+	require.NoError(t, err)
+}

dumpling/export/sql.go

@@ -29,6 +29,7 @@ import (
 
 const (
 	orderByTiDBRowID = "ORDER BY `_tidb_rowid`"
+	snapshotVar      = "tidb_snapshot"
 )
 
 type listTableType int
@@ -851,7 +852,9 @@ func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, cfg *mysql.Con
 		s := fmt.Sprintf("SET SESSION %s = ?", k)
 		_, err := db.ExecContext(tctx, s, pv)
 		if err != nil {
-			if isUnknownSystemVariableErr(err) {
+			if k == snapshotVar {
+				err = errors.Annotate(err, "fail to set snapshot for tidb, please set --consistency=none/--consistency=lock or fix snapshot problem")
+			} else if isUnknownSystemVariableErr(err) {
 				tctx.L().Info("session variable is not supported by db", zap.String("variable", k), zap.Reflect("value", v))
 				continue
 			}
@@ -876,6 +879,9 @@ func resetDBWithSessionParams(tctx *tcontext.Context, db *sql.DB, cfg *mysql.Con
 		}
 		cfg.Params[k] = s
 	}
+	failpoint.Inject("SkipResetDB", func(_ failpoint.Value) {
+		failpoint.Return(db, nil)
+	})
 
 	db.Close()
 	c, err := mysql.NewConnector(cfg)

executor/adapter.go

@@ -734,12 +734,7 @@ func (a *ExecStmt) handleNoDelay(ctx context.Context, e Executor, isPessimistic
 		// done in the `defer` function. If the rs is not nil, the detachment will be done in
 		// `rs.Close` in `handleStmt`
 		if handled && sc != nil && rs == nil {
-			if sc.MemTracker != nil {
-				sc.MemTracker.Detach()
-			}
-			if sc.DiskTracker != nil {
-				sc.DiskTracker.Detach()
-			}
+			sc.DetachMemDiskTracker()
 		}
 	}()
 
@@ -1417,15 +1412,7 @@ func (a *ExecStmt) FinishExecuteStmt(txnTS uint64, err error, hasMoreResults boo
 func (a *ExecStmt) CloseRecordSet(txnStartTS uint64, lastErr error) {
 	a.FinishExecuteStmt(txnStartTS, lastErr, false)
 	a.logAudit()
-	// Detach the Memory and disk tracker for the previous stmtCtx from GlobalMemoryUsageTracker and GlobalDiskUsageTracker
-	if stmtCtx := a.Ctx.GetSessionVars().StmtCtx; stmtCtx != nil {
-		if stmtCtx.DiskTracker != nil {
-			stmtCtx.DiskTracker.Detach()
-		}
-		if stmtCtx.MemTracker != nil {
-			stmtCtx.MemTracker.Detach()
-		}
-	}
+	a.Ctx.GetSessionVars().StmtCtx.DetachMemDiskTracker()
 }
 
 // LogSlowQuery is used to print the slow query in the log files.

executor/hash_table.go

@@ -153,7 +153,7 @@ func (c *hashRowContainer) GetMatchedRows(probeKey uint64, probeRow chunk.Row, h
 }
 
 func (c *hashRowContainer) GetAllMatchedRows(probeHCtx *hashContext, probeSideRow chunk.Row,
-	probeKeyNullBits *bitmap.ConcurrentBitmap, matched []chunk.Row, needCheckBuildRowPos, needCheckProbeRowPos []int) ([]chunk.Row, error) {
+	probeKeyNullBits *bitmap.ConcurrentBitmap, matched []chunk.Row, needCheckBuildColPos, needCheckProbeColPos []int, needCheckBuildTypes, needCheckProbeTypes []*types.FieldType) ([]chunk.Row, error) {
 	// for NAAJ probe row with null, we should match them with all build rows.
 	var (
 		ok        bool
@@ -180,16 +180,20 @@ func (c *hashRowContainer) GetAllMatchedRows(probeHCtx *hashContext, probeSideRo
 		// else like
 		//	   (null, 1, 2), we should use the not-null probe bit to filter rows. Only fetch rows like
 		//     (  ? , 1, 2), that exactly with value as 1 and 2 in the second and third join key column.
-		needCheckProbeRowPos = needCheckProbeRowPos[:0]
-		needCheckBuildRowPos = needCheckBuildRowPos[:0]
+		needCheckProbeColPos = needCheckProbeColPos[:0]
+		needCheckBuildColPos = needCheckBuildColPos[:0]
+		needCheckBuildTypes = needCheckBuildTypes[:0]
+		needCheckProbeTypes = needCheckProbeTypes[:0]
 		keyColLen := len(c.hCtx.naKeyColIdx)
 		for i := 0; i < keyColLen; i++ {
 			// since all bucket is from hash table (Not Null), so the buildSideNullBits check is eliminated.
 			if probeKeyNullBits.UnsafeIsSet(i) {
 				continue
 			}
-			needCheckBuildRowPos = append(needCheckBuildRowPos, c.hCtx.naKeyColIdx[i])
-			needCheckProbeRowPos = append(needCheckProbeRowPos, probeHCtx.naKeyColIdx[i])
+			needCheckBuildColPos = append(needCheckBuildColPos, c.hCtx.naKeyColIdx[i])
+			needCheckBuildTypes = append(needCheckBuildTypes, c.hCtx.allTypes[i])
+			needCheckProbeColPos = append(needCheckProbeColPos, probeHCtx.naKeyColIdx[i])
+			needCheckProbeTypes = append(needCheckProbeTypes, probeHCtx.allTypes[i])
 		}
 	}
 	var mayMatchedRow chunk.Row
@@ -200,7 +204,7 @@ func (c *hashRowContainer) GetAllMatchedRows(probeHCtx *hashContext, probeSideRo
 		}
 		if probeKeyNullBits != nil && len(probeHCtx.naKeyColIdx) > 1 {
 			// check the idxs-th value of the join columns.
-			ok, err = codec.EqualChunkRow(c.sc, mayMatchedRow, c.hCtx.allTypes, needCheckBuildRowPos, probeSideRow, probeHCtx.allTypes, needCheckProbeRowPos)
+			ok, err = codec.EqualChunkRow(c.sc, mayMatchedRow, needCheckBuildTypes, needCheckBuildColPos, probeSideRow, needCheckProbeTypes, needCheckProbeColPos)
 			if err != nil {
 				return nil, err
 			}
@@ -287,7 +291,7 @@ func (c *hashRowContainer) GetMatchedRowsAndPtrs(probeKey uint64, probeRow chunk
 }
 
 func (c *hashRowContainer) GetNullBucketRows(probeHCtx *hashContext, probeSideRow chunk.Row,
-	probeKeyNullBits *bitmap.ConcurrentBitmap, matched []chunk.Row, needCheckBuildRowPos, needCheckProbeRowPos []int) ([]chunk.Row, error) {
+	probeKeyNullBits *bitmap.ConcurrentBitmap, matched []chunk.Row, needCheckBuildColPos, needCheckProbeColPos []int, needCheckBuildTypes, needCheckProbeTypes []*types.FieldType) ([]chunk.Row, error) {
 	var (
 		ok            bool
 		err           error
@@ -306,8 +310,10 @@ func (c *hashRowContainer) GetNullBucketRows(probeHCtx *hashContext, probeSideRo
 		// case2: left side (probe side) don't have null
 		//    left side key <1, 2>, actually we should fetch <1,null>, <null, 2>, <null, null> from the null bucket because
 		//    case like <3,null> is obviously not matched with the probe key.
-		needCheckProbeRowPos = needCheckProbeRowPos[:0]
-		needCheckBuildRowPos = needCheckBuildRowPos[:0]
+		needCheckProbeColPos = needCheckProbeColPos[:0]
+		needCheckBuildColPos = needCheckBuildColPos[:0]
+		needCheckBuildTypes = needCheckBuildTypes[:0]
+		needCheckProbeTypes = needCheckProbeTypes[:0]
 		keyColLen := len(c.hCtx.naKeyColIdx)
 		if probeKeyNullBits != nil {
 			// when the probeKeyNullBits is not nil, it means the probe key has null values, where we should distinguish
@@ -325,11 +331,13 @@ func (c *hashRowContainer) GetNullBucketRows(probeHCtx *hashContext, probeSideRo
 				if probeKeyNullBits.UnsafeIsSet(i) || nullEntry.nullBitMap.UnsafeIsSet(i) {
 					continue
 				}
-				needCheckBuildRowPos = append(needCheckBuildRowPos, c.hCtx.naKeyColIdx[i])
-				needCheckProbeRowPos = append(needCheckProbeRowPos, probeHCtx.naKeyColIdx[i])
+				needCheckBuildColPos = append(needCheckBuildColPos, c.hCtx.naKeyColIdx[i])
+				needCheckBuildTypes = append(needCheckBuildTypes, c.hCtx.allTypes[i])
+				needCheckProbeColPos = append(needCheckProbeColPos, probeHCtx.naKeyColIdx[i])
+				needCheckProbeTypes = append(needCheckProbeTypes, probeHCtx.allTypes[i])
 			}
 			// check the idxs-th value of the join columns.
-			ok, err = codec.EqualChunkRow(c.sc, mayMatchedRow, c.hCtx.allTypes, needCheckBuildRowPos, probeSideRow, probeHCtx.allTypes, needCheckProbeRowPos)
+			ok, err = codec.EqualChunkRow(c.sc, mayMatchedRow, needCheckBuildTypes, needCheckBuildColPos, probeSideRow, needCheckProbeTypes, needCheckProbeColPos)
 			if err != nil {
 				return nil, err
 			}
@@ -346,11 +354,13 @@ func (c *hashRowContainer) GetNullBucketRows(probeHCtx *hashContext, probeSideRo
 				if nullEntry.nullBitMap.UnsafeIsSet(i) {
 					continue
 				}
-				needCheckBuildRowPos = append(needCheckBuildRowPos, c.hCtx.naKeyColIdx[i])
-				needCheckProbeRowPos = append(needCheckProbeRowPos, probeHCtx.naKeyColIdx[i])
+				needCheckBuildColPos = append(needCheckBuildColPos, c.hCtx.naKeyColIdx[i])
+				needCheckBuildTypes = append(needCheckBuildTypes, c.hCtx.allTypes[i])
+				needCheckProbeColPos = append(needCheckProbeColPos, probeHCtx.naKeyColIdx[i])
+				needCheckProbeTypes = append(needCheckProbeTypes, probeHCtx.allTypes[i])
 			}
 			// check the idxs-th value of the join columns.
-			ok, err = codec.EqualChunkRow(c.sc, mayMatchedRow, c.hCtx.allTypes, needCheckBuildRowPos, probeSideRow, probeHCtx.allTypes, needCheckProbeRowPos)
+			ok, err = codec.EqualChunkRow(c.sc, mayMatchedRow, needCheckBuildTypes, needCheckBuildColPos, probeSideRow, needCheckProbeTypes, needCheckProbeColPos)
 			if err != nil {
 				return nil, err
 			}
@@ -366,6 +376,11 @@ func (c *hashRowContainer) GetNullBucketRows(probeHCtx *hashContext, probeSideRo
 
 // matchJoinKey checks if join keys of buildRow and probeRow are logically equal.
 func (c *hashRowContainer) matchJoinKey(buildRow, probeRow chunk.Row, probeHCtx *hashContext) (ok bool, err error) {
+	if len(c.hCtx.naKeyColIdx) > 0 {
+		return codec.EqualChunkRow(c.sc,
+			buildRow, c.hCtx.allTypes, c.hCtx.naKeyColIdx,
+			probeRow, probeHCtx.allTypes, probeHCtx.naKeyColIdx)
+	}
 	return codec.EqualChunkRow(c.sc,
 		buildRow, c.hCtx.allTypes, c.hCtx.keyColIdx,
 		probeRow, probeHCtx.allTypes, probeHCtx.keyColIdx)

executor/issuetest/BUILD.bazel

@@ -15,10 +15,12 @@ go_test(
         "//parser/auth",
         "//parser/charset",
         "//parser/mysql",
+        "//session",
         "//sessionctx/variable",
         "//statistics",
         "//testkit",
         "//util",
+        "//util/memory",
         "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_stretchr_testify//require",
         "@com_github_tikv_client_go_v2//tikv",

executor/issuetest/executor_issue_test.go

@@ -20,17 +20,20 @@ import (
 	"math/rand"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/config"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/parser/auth"
 	"github.com/pingcap/tidb/parser/charset"
 	"github.com/pingcap/tidb/parser/mysql"
+	"github.com/pingcap/tidb/session"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/statistics"
 	"github.com/pingcap/tidb/testkit"
 	"github.com/pingcap/tidb/util"
+	"github.com/pingcap/tidb/util/memory"
 	"github.com/stretchr/testify/require"
 )
 
@@ -1348,3 +1351,46 @@ func TestIssue40158(t *testing.T) {
 	tk.MustExec("insert into t1 values (1, null);")
 	tk.MustQuery("select * from t1 where c1 is null and _id < 1;").Check(testkit.Rows())
 }
+
+func TestIssue42662(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.Session().GetSessionVars().ConnectionID = 12345
+	tk.Session().GetSessionVars().MemTracker = memory.NewTracker(memory.LabelForSession, -1)
+	tk.Session().GetSessionVars().MemTracker.SessionID = 12345
+	tk.Session().GetSessionVars().MemTracker.IsRootTrackerOfSess = true
+
+	sm := &testkit.MockSessionManager{
+		PS: []*util.ProcessInfo{tk.Session().ShowProcess()},
+	}
+	sm.Conn = make(map[uint64]session.Session)
+	sm.Conn[tk.Session().GetSessionVars().ConnectionID] = tk.Session()
+	dom.ServerMemoryLimitHandle().SetSessionManager(sm)
+	go dom.ServerMemoryLimitHandle().Run()
+
+	tk.MustExec("use test")
+	tk.MustQuery("select connection_id()").Check(testkit.Rows("12345"))
+	tk.MustExec("drop table if exists t1, t2")
+	tk.MustExec("create table t1 (a int, b int, c int)")
+	tk.MustExec("create table t2 (a int, b int, c int)")
+	tk.MustExec("insert into t1 values (1, 1, 1), (1, 2, 2), (2, 1, 3), (2, 2, 4)")
+	tk.MustExec("insert into t2 values (1, 1, 1), (1, 2, 2), (2, 1, 3), (2, 2, 4)")
+	// set tidb_server_memory_limit to 1.6GB, tidb_server_memory_limit_sess_min_size to 128MB
+	tk.MustExec("set global tidb_server_memory_limit='1600MB'")
+	tk.MustExec("set global tidb_server_memory_limit_sess_min_size=128*1024*1024")
+	tk.MustExec("set global tidb_mem_oom_action = 'cancel'")
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/executor/issue42662_1", `return(true)`))
+	// tk.Session() should be marked as MemoryTop1Tracker but not killed.
+	tk.MustQuery("select /*+ hash_join(t1)*/ * from t1 join t2 on t1.a = t2.a and t1.b = t2.b")
+
+	// try to trigger the kill top1 logic
+	require.NoError(t, failpoint.Enable("github.com/pingcap/tidb/util/servermemorylimit/issue42662_2", `return(true)`))
+	time.Sleep(1 * time.Second)
+
+	// no error should be returned
+	tk.MustQuery("select count(*) from t1")
+	tk.MustQuery("select count(*) from t1")
+
+	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/executor/issue42662_1"))
+	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/util/servermemorylimit/issue42662_2"))
+}