executor: fix data inconsistency after load data ... replace into ...

pkg/executor/insert_common.go

@@ -1179,11 +1179,7 @@ func (e *InsertValues) handleDuplicateKey(ctx context.Context, txn kv.Transactio
 	if handle == nil {
 		return false, nil
 	}
-	_, err = e.removeRow(ctx, txn, handle, r, true)
-	if err != nil {
-		return false, err
-	}
-	return false, nil
+	return e.removeRow(ctx, txn, handle, r, true)
 }
 
 // batchCheckAndInsert checks rows with duplicate errors.

pkg/executor/test/loaddatatest/BUILD.bazel

@@ -9,7 +9,7 @@ go_test(
     ],
     flaky = True,
     race = "on",
-    shard_count = 11,
+    shard_count = 12,
     deps = [
         "//pkg/config",
         "//pkg/executor",

pkg/executor/test/loaddatatest/load_data_test.go

@@ -472,3 +472,22 @@ func TestLoadDataFromServerFile(t *testing.T) {
 	err := tk.ExecToErr("load data infile 'remote.csv' into table load_data_test")
 	require.ErrorContains(t, err, "[executor:8154]Don't support load data from tidb-server's disk.")
 }
+
+func TestFix56408(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("USE test; DROP TABLE IF EXISTS load_data_replace;")
+	tk.MustExec("create table a(id int,name varchar(20),addr varchar(100),primary key (id) nonclustered);")
+	loadSQL := "LOAD DATA LOCAL INFILE '/tmp/nonexistence.csv' REPLACE INTO TABLE a FIELDS terminated by '|';"
+	ctx := tk.Session().(sessionctx.Context)
+	tests := []testCase{
+		{[]byte("1|aa|beijing\n1|aa|beijing\n1|aa|beijing\n1|aa|beijing\n2|bb|shanghai\n2|bb|shanghai\n2|bb|shanghai\n3|cc|guangzhou\n"),
+			[]string{"1 aa beijing", "2 bb shanghai", "3 cc guangzhou"},
+			"Records: 8  Deleted: 0  Skipped: 5  Warnings: 0",
+		},
+	}
+	deleteSQL := "DO 1"
+	selectSQL := "TABLE a;"
+	checkCases(tests, loadSQL, t, tk, ctx, selectSQL, deleteSQL)
+	tk.MustExec("ADMIN CHECK TABLE a")
+}