fix the issue that INSERT IGNORE doesn't lock the parent table

Signed-off-by: Yang Keao <[email protected]>

Author: YangKeao

pkg/executor/foreign_key.go

@@ -223,10 +223,19 @@ func (fkc *FKCheckExec) doCheck(ctx context.Context) error {
 		fkc.stats = &FKCheckRuntimeStats{}
 		defer fkc.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl.RegisterStats(fkc.ID(), fkc.stats)
 	}
+	start := time.Now()
 	if len(fkc.toBeCheckedKeys) == 0 && len(fkc.toBeCheckedPrefixKeys) == 0 {
-		return nil
+		// It's possible that the `toBeLockedKeys` is not empty. Because the `INSERT IGNORE` will use
+		// `checkRows` to check each row, which doesn't need this function to check the rows again, but
+		// it still needs to lock the rows.
+		err := fkc.doLock(ctx)
+		if len(fkc.toBeLockedKeys) > 0 && fkc.stats != nil {
+			fkc.stats.Lock = time.Since(start)
+			fkc.stats.Total = fkc.stats.Lock
+		}
+
+		return err
 	}
-	start := time.Now()
 	if fkc.stats != nil {
 		defer func() {
 			fkc.stats.Keys = len(fkc.toBeCheckedKeys) + len(fkc.toBeCheckedPrefixKeys)
@@ -248,6 +257,15 @@ func (fkc *FKCheckExec) doCheck(ctx context.Context) error {
 	if fkc.stats != nil {
 		fkc.stats.Check = time.Since(start)
 	}
+
+	err = fkc.doLock(ctx)
+	if fkc.stats != nil {
+		fkc.stats.Lock = time.Since(start) - fkc.stats.Check
+	}
+	return err
+}
+
+func (fkc *FKCheckExec) doLock(ctx context.Context) error {
 	if len(fkc.toBeLockedKeys) == 0 {
 		return nil
 	}
@@ -263,9 +281,6 @@ func (fkc *FKCheckExec) doCheck(ctx context.Context) error {
 	// doLockKeys may set TxnCtx.ForUpdate to 1, then if the lock meet write conflict, TiDB can't retry for update.
 	// So reset TxnCtx.ForUpdate to 0 then can be retry if meet write conflict.
 	atomic.StoreUint32(&sessVars.TxnCtx.ForUpdate, forUpdate)
-	if fkc.stats != nil {
-		fkc.stats.Lock = time.Since(start) - fkc.stats.Check
-	}
 	return err
 }
 
@@ -542,7 +557,7 @@ type fkCheckKey struct {
 	isPrefix bool
 }
 
-func (fkc FKCheckExec) checkRows(ctx context.Context, sc *stmtctx.StatementContext, txn kv.Transaction, rows []toBeCheckedRow) error {
+func (fkc *FKCheckExec) checkRows(ctx context.Context, sc *stmtctx.StatementContext, txn kv.Transaction, rows []toBeCheckedRow) error {
 	if fkc.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl != nil {
 		fkc.stats = &FKCheckRuntimeStats{}
 		defer fkc.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl.RegisterStats(fkc.ID(), fkc.stats)

pkg/executor/test/fktest/foreign_key_test.go

@@ -2088,8 +2088,8 @@ func TestExplainAnalyzeDMLWithFKInfo(t *testing.T) {
 		{
 			sql: "explain analyze insert ignore into t6 values (1,1,10)",
 			plan: "Insert_.* root  time:.* loops:.* prepare:.* check_insert.* fk_check:.*" +
-				"├─Foreign_Key_Check.* 0 root table:t5 total:0s, foreign_keys:1 foreign_key:fk_1, check_exist N/A N/A.*" +
-				"├─Foreign_Key_Check.* 0 root table:t5, index:idx2 total:0s, foreign_keys:1 foreign_key:fk_2, check_exist N/A N/A.*" +
+				"├─Foreign_Key_Check.* 0 root table:t5 total:.*, lock:.*, foreign_keys:1 foreign_key:fk_1, check_exist N/A N/A.*" +
+				"├─Foreign_Key_Check.* 0 root table:t5, index:idx2 total:.*, lock:.*, foreign_keys:1 foreign_key:fk_2, check_exist N/A N/A.*" +
 				"└─Foreign_Key_Check.* 0 root table:t5, index:idx3 total:0s, foreign_keys:1 foreign_key:fk_3, check_exist N/A N/A",
 		},
 		{

tests/realtikvtest/txntest/txn_test.go

@@ -595,3 +595,42 @@ func TestDMLWithAddForeignKey(t *testing.T) {
 
 	require.True(t, errDML != nil || errDDL != nil)
 }
+
+func TestLockKeysInInsertIgnore(t *testing.T) {
+	// FIXME: this test will fail in the unistore. After fixing the issue in unistore, please move
+	// this test to `foreign_key_test.go`.
+
+	store := realtikvtest.CreateMockStoreAndSetup(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("create table t1 (id int primary key);")
+	tk.MustExec("create table t2 (id int primary key, foreign key fk (id) references t1(id));")
+	tk.MustExec("insert into t1 values (1)")
+
+	tk.MustExec("BEGIN")
+	tk.MustExec("INSERT IGNORE INTO t2 VALUES (1)")
+
+	var wg sync.WaitGroup
+	var tk2CommitTime time.Time
+	tk2StartTime := time.Now()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+
+		tk2 := testkit.NewTestKit(t, store)
+		tk2.MustExec("use test")
+		tk2.MustExec("BEGIN")
+		require.NotNil(t, tk2.ExecToErr("UPDATE t1 SET id = 2 WHERE id = 1"))
+		tk2.MustExec("COMMIT")
+		tk2CommitTime = time.Now()
+	}()
+
+	sleepDuration := 500 * time.Millisecond
+	time.Sleep(sleepDuration)
+	tk.MustExec("COMMIT")
+	wg.Wait()
+
+	require.Greater(t, tk2CommitTime.Sub(tk2StartTime), sleepDuration)
+	tk.MustQuery("SELECT * FROM t1").Check(testkit.Rows("1"))
+	tk.MustQuery("SELECT * FROM t2").Check(testkit.Rows("1"))
+}