This is an automated cherry-pick of pingcap#56415

Signed-off-by: ti-chi-bot <[email protected]>

pkg/executor/insert_common.go

@@ -1172,11 +1172,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,11 @@ go_test(
     ],
     flaky = True,
     race = "on",
+<<<<<<< HEAD
     shard_count = 10,
+=======
+    shard_count = 12,
+>>>>>>> 7daf0265f16 (executor: fix data inconsistency after `load data ... replace into ...` (#56415))
     deps = [
         "//br/pkg/lightning/mydump",
         "//pkg/config",

pkg/executor/test/loaddatatest/load_data_test.go

@@ -476,3 +476,22 @@ func TestLoadDataIntoPartitionedTable(t *testing.T) {
 	selectSQL := "select * from range_t order by a;"
 	checkCases(tests, ld, t, tk, ctx, selectSQL, deleteSQL)
 }
+
+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")
+}