executor: add a variable to control whether we can write _tidb_rowid (#…

…8126)

executor/rowid_test.go

@@ -19,8 +19,11 @@ import (
 )
 
 func (s *testSuite) TestExportRowID(c *C) {
-	tk := testkit.NewTestKit(c, s.store)
-	tk.MustExec("use test")
+	tk := testkit.NewTestKitWithInit(c, s.store)
+	tk.Se.GetSessionVars().AllowWriteRowID = true
+	defer func() {
+		tk.Se.GetSessionVars().AllowWriteRowID = false
+	}()
 	tk.MustExec("drop table if exists t")
 	tk.MustExec("create table t (a int, b int)")
 	tk.MustExec("insert t values (1, 7), (1, 8), (1, 9)")
@@ -49,4 +52,39 @@ func (s *testSuite) TestExportRowID(c *C) {
 	c.Assert(err, NotNil)
 	_, err = tk.Exec("delete from s where _tidb_rowid = 1")
 	c.Assert(err, NotNil)
+
+	// Make sure "AllowWriteRowID" is a session variable.
+	tk1 := testkit.NewTestKit(c, s.store)
+	tk1.MustExec("use test")
+	_, err = tk1.Exec("insert into t (a, _tidb_rowid) values(10,1);")
+	c.Assert(err.Error(), Equals, "insert, update and replace statements for _tidb_rowid are not supported.")
+}
+
+func (s *testSuite) TestNotAllowWriteRowID(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("create table tt(id binary(10), c int, primary key(id));")
+	tk.MustExec("insert tt values (1, 10);")
+
+	// select statement
+	tk.MustQuery("select *, _tidb_rowid from tt").
+		Check(testkit.Rows("1\x00\x00\x00\x00\x00\x00\x00\x00\x00 10 1"))
+	// insert statement
+	_, err := tk.Exec("insert into tt (id, c, _tidb_rowid) values(30000,10,1);")
+	c.Assert(err.Error(), Equals, "insert, update and replace statements for _tidb_rowid are not supported.")
+	// replace statement
+	_, err = tk.Exec("replace into tt (id, c, _tidb_rowid) values(30000,10,1);")
+	c.Assert(err.Error(), Equals, "insert, update and replace statements for _tidb_rowid are not supported.")
+	// update statement
+	_, err = tk.Exec("update tt set id = 2, _tidb_rowid = 1 where _tidb_rowid = 1")
+	c.Assert(err.Error(), Equals, "insert, update and replace statements for _tidb_rowid are not supported.")
+	tk.MustExec("update tt set id = 2 where _tidb_rowid = 1")
+
+	tk.MustExec("admin check table tt;")
+	tk.MustExec("drop table tt")
+
+	// There is currently no real support for inserting, updating, and replacing _tidb_rowid statements.
+	// After we support it, the following operations must be passed.
+	//	tk.MustExec("insert into tt (id, c, _tidb_rowid) values(30000,10,1);")
+	//	tk.MustExec("admin check table tt;")
 }

executor/write.go

@@ -1370,6 +1370,9 @@ func (e *InsertValues) getColumns(tableCols []*table.Column) ([]*table.Column, e
 	}
 	for _, col := range cols {
 		if col.Name.L == model.ExtraHandleName.L {
+			if !e.ctx.GetSessionVars().AllowWriteRowID {
+				return nil, errors.Errorf("insert, update and replace statements for _tidb_rowid are not supported.")
+			}
 			e.hasExtraHandle = true
 			break
 		}
@@ -1919,7 +1922,7 @@ type UpdateExec struct {
 }
 
 func (e *UpdateExec) exec(ctx context.Context, schema *expression.Schema) (types.DatumRow, error) {
-	assignFlag, err := getUpdateColumns(e.OrderedList, schema.Len())
+	assignFlag, err := getUpdateColumns(e.ctx, e.OrderedList, schema.Len())
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -1995,9 +1998,12 @@ func (e *UpdateExec) Next(ctx context.Context, chk *chunk.Chunk) error {
 	return nil
 }
 
-func getUpdateColumns(assignList []*expression.Assignment, schemaLen int) ([]bool, error) {
+func getUpdateColumns(ctx sessionctx.Context, assignList []*expression.Assignment, schemaLen int) ([]bool, error) {
 	assignFlag := make([]bool, schemaLen)
 	for _, v := range assignList {
+		if !ctx.GetSessionVars().AllowWriteRowID && v.Col.ColName.L == model.ExtraHandleName.L {
+			return nil, errors.Errorf("insert, update and replace statements for _tidb_rowid are not supported.")
+		}
 		idx := v.Col.Index
 		assignFlag[idx] = true
 	}

sessionctx/variable/session.go

@@ -238,6 +238,10 @@ type SessionVars struct {
 	// AllowAggPushDown can be set to false to forbid aggregation push down.
 	AllowAggPushDown bool
 
+	// AllowWriteRowID can be set to false to forbid write data to _tidb_rowid.
+	// This variable is currently not recommended to be turned on.
+	AllowWriteRowID bool
+
 	// AllowInSubqueryUnFolding can be set to true to fold in subquery
 	AllowInSubqueryUnFolding bool
 
@@ -517,6 +521,8 @@ func (s *SessionVars) SetSystemVar(name string, val string) error {
 		s.SkipUTF8Check = TiDBOptOn(val)
 	case TiDBOptAggPushDown:
 		s.AllowAggPushDown = TiDBOptOn(val)
+	case TiDBOptWriteRowID:
+		s.AllowWriteRowID = TiDBOptOn(val)
 	case TiDBOptInSubqUnFolding:
 		s.AllowInSubqueryUnFolding = TiDBOptOn(val)
 	case TiDBIndexLookupConcurrency:

sessionctx/variable/sysvar.go

@@ -610,6 +610,7 @@ var defaultSysVars = []*SysVar{
 	{ScopeSession, TiDBImportingData, "0"},
 	{ScopeSession, TiDBOptAggPushDown, boolToIntStr(DefOptAggPushDown)},
 	{ScopeSession, TiDBOptInSubqUnFolding, boolToIntStr(DefOptInSubqUnfolding)},
+	{ScopeSession, TiDBOptWriteRowID, boolToIntStr(DefOptWriteRowID)},
 	{ScopeSession, TiDBBuildStatsConcurrency, strconv.Itoa(DefBuildStatsConcurrency)},
 	{ScopeGlobal, TiDBAutoAnalyzeRatio, strconv.FormatFloat(DefAutoAnalyzeRatio, 'f', -1, 64)},
 	{ScopeSession, TiDBChecksumTableConcurrency, strconv.Itoa(DefChecksumTableConcurrency)},

sessionctx/variable/tidb_vars.go

@@ -39,6 +39,9 @@ const (
 	// tidb_opt_agg_push_down is used to enable/disable the optimizer rule of aggregation push down.
 	TiDBOptAggPushDown = "tidb_opt_agg_push_down"
 
+	// tidb_opt_write_row_id is used to enable/disable the operations of insert、replace and update to _tidb_rowid.
+	TiDBOptWriteRowID = "tidb_opt_write_row_id"
+
 	// tidb_opt_insubquery_unfold is used to enable/disable the optimizer rule of in subquery unfold.
 	TiDBOptInSubqUnFolding = "tidb_opt_insubquery_unfold"
 
@@ -172,6 +175,7 @@ const (
 	DefSkipUTF8Check                 = false
 	DefOptAggPushDown                = false
 	DefOptInSubqUnfolding            = false
+	DefOptWriteRowID                 = false
 	DefBatchInsert                   = false
 	DefBatchDelete                   = false
 	DefCurretTS                      = 0

sessionctx/variable/varsutil_test.go

@@ -74,6 +74,7 @@ func (s *testVarsutilSuite) TestNewSessionVars(c *C) {
 	c.Assert(vars.MemQuotaIndexLookupReader, Equals, int64(DefTiDBMemQuotaIndexLookupReader))
 	c.Assert(vars.MemQuotaIndexLookupJoin, Equals, int64(DefTiDBMemQuotaIndexLookupJoin))
 	c.Assert(vars.MemQuotaNestedLoopApply, Equals, int64(DefTiDBMemQuotaNestedLoopApply))
+	c.Assert(vars.AllowWriteRowID, Equals, DefOptWriteRowID)
 }
 
 func (s *testVarsutilSuite) TestVarsutil(c *C) {