*: Add a variable to skip constraint check in session. (pingcap#2031)

When import data from mysqldump/mydumper. We could skip constraint check.

sessionctx/variable/session.go

@@ -123,6 +123,9 @@ type SessionVars struct {
 	// SnapshotInfoschema is used with SnapshotTS, when the schema version at snapshotTS less than current schema
 	// version, we load an old version schema for query.
 	SnapshotInfoschema interface{}
+
+	// SkipConstraintCheck is true when importing data.
+	SkipConstraintCheck bool
 }
 
 // sessionVarsKeyType is a dummy type to avoid naming collision in context.
@@ -269,6 +272,8 @@ func (s *SessionVars) SetSystemVar(key string, value types.Datum) error {
 	case AutocommitVar:
 		isAutocommit := strings.EqualFold(sVal, "ON") || sVal == "1"
 		s.SetStatusFlag(mysql.ServerStatusAutocommit, isAutocommit)
+	case TiDBSkipConstraintCheck:
+		s.setSkipConstraintCheck(sVal)
 	}
 	s.systems[key] = sVal
 	return nil
@@ -291,6 +296,14 @@ func (s *SessionVars) setSnapshotTS(sVal string) error {
 	return nil
 }
 
+func (s *SessionVars) setSkipConstraintCheck(sVal string) {
+	if sVal == "1" {
+		s.SkipConstraintCheck = true
+	} else {
+		s.SkipConstraintCheck = false
+	}
+}
+
 // GetSystemVar gets a system variable.
 func (s *SessionVars) GetSystemVar(key string) types.Datum {
 	var d types.Datum

sessionctx/variable/session_test.go

@@ -68,4 +68,13 @@ func (*testSessionSuite) TestSession(c *C) {
 	c.Assert(collation, Equals, "utf8_general_ci")
 
 	c.Assert(v.SetSystemVar("character_set_results", types.Datum{}), IsNil)
+
+	// Test case for tidb_skip_constraint_check
+	c.Assert(v.SkipConstraintCheck, IsFalse)
+	v.SetSystemVar(variable.TiDBSkipConstraintCheck, types.NewStringDatum("0"))
+	c.Assert(v.SkipConstraintCheck, IsFalse)
+	v.SetSystemVar(variable.TiDBSkipConstraintCheck, types.NewStringDatum("1"))
+	c.Assert(v.SkipConstraintCheck, IsTrue)
+	v.SetSystemVar(variable.TiDBSkipConstraintCheck, types.NewStringDatum("0"))
+	c.Assert(v.SkipConstraintCheck, IsFalse)
 }

sessionctx/variable/sysvar.go

@@ -84,6 +84,7 @@ func init() {
 	tidbSysVars[DistSQLScanConcurrencyVar] = true
 	tidbSysVars[DistSQLJoinConcurrencyVar] = true
 	tidbSysVars[TiDBSnapshot] = true
+	tidbSysVars[TiDBSkipConstraintCheck] = true
 }
 
 // we only support MySQL now
@@ -590,13 +591,15 @@ var defaultSysVars = []*SysVar{
 	{ScopeSession, TiDBSnapshot, ""},
 	{ScopeGlobal | ScopeSession, DistSQLScanConcurrencyVar, "10"},
 	{ScopeGlobal | ScopeSession, DistSQLJoinConcurrencyVar, "5"},
+	{ScopeSession, TiDBSkipConstraintCheck, "0"},
 }
 
 // TiDB system variables
 const (
 	TiDBSnapshot              = "tidb_snapshot"
 	DistSQLScanConcurrencyVar = "tidb_distsql_scan_concurrency"
 	DistSQLJoinConcurrencyVar = "tidb_distsql_join_concurrency"
+	TiDBSkipConstraintCheck   = "tidb_skip_constraint_check"
 )
 
 // SetNamesVariables is the system variable names related to set names statements.

table/tables/tables.go

@@ -391,6 +391,10 @@ func (t *Table) genIndexKeyStr(colVals []types.Datum) (string, error) {
 	return strings.Join(strVals, "-"), nil
 }
 
+func skipConstraintCheck(ctx context.Context) bool {
+	return variable.GetSessionVars(ctx).SkipConstraintCheck
+}
+
 // Add data into indices.
 func (t *Table) addIndices(ctx context.Context, recordID int64, r []types.Datum, bs *kv.BufferStore) (int64, error) {
 	txn, err := ctx.GetTxn(false)
@@ -399,7 +403,8 @@ func (t *Table) addIndices(ctx context.Context, recordID int64, r []types.Datum,
 	}
 	// Clean up lazy check error environment
 	defer txn.DelOption(kv.PresumeKeyNotExistsError)
-	if t.meta.PKIsHandle {
+	skipCheck := skipConstraintCheck(ctx)
+	if t.meta.PKIsHandle && !skipCheck {
 		// Check key exists.
 		recordKey := t.RecordKey(recordID)
 		e := kv.ErrKeyExists.FastGen("Duplicate entry '%d' for key 'PRIMARY'", recordID)
@@ -423,7 +428,7 @@ func (t *Table) addIndices(ctx context.Context, recordID int64, r []types.Datum,
 			return 0, errors.Trace(err2)
 		}
 		var dupKeyErr error
-		if v.Meta().Unique || v.Meta().Primary {
+		if !skipCheck && (v.Meta().Unique || v.Meta().Primary) {
 			entryKey, err1 := t.genIndexKeyStr(colVals)
 			if err1 != nil {
 				return 0, errors.Trace(err1)