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

….` (#56415) (#56876)

close #56408

executor/insert_common.go

@@ -1186,11 +1186,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.

executor/loaddatatest/BUILD.bazel

@@ -9,7 +9,7 @@ go_test(
     ],
     flaky = True,
     race = "on",
-    shard_count = 10,
+    shard_count = 11,
     deps = [
         "//br/pkg/lightning/mydump",
         "//config",

executor/loaddatatest/load_data_test.go

@@ -498,3 +498,24 @@ 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 '|';"
+	tk.MustExec(loadSQL)
+	ctx := tk.Session().(sessionctx.Context)
+	ld := ctx.Value(executor.LoadDataVarKey).(*executor.LoadDataWorker)
+	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, ld, t, tk, ctx, selectSQL, deleteSQL)
+	tk.MustExec("ADMIN CHECK TABLE a")
+}