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

….` (#56415) (#56505)

close #56408

executor/insert_common.go

@@ -1113,11 +1113,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/write_test.go

@@ -4291,3 +4291,37 @@ func TestHandleColumnWithOnUpdateCurrentTimestamp(t *testing.T) {
 	tk.MustExec("update t force index(primary) set b = 10 where a = '2023-06-11 10:00:00'")
 	tk.MustExec("admin check table t")
 }
+
+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);")
+	tk.MustExec("LOAD DATA LOCAL INFILE '/tmp/nonexistence.csv' REPLACE INTO TABLE a FIELDS terminated by '|';")
+	ctx := tk.Session().(sessionctx.Context)
+	ld, ok := ctx.Value(executor.LoadDataVarKey).(*executor.LoadDataInfo)
+	require.True(t, ok)
+	defer ctx.SetValue(executor.LoadDataVarKey, nil)
+	require.NotNil(t, ld)
+
+	tests := []testCase{
+		{
+			nil,
+			[]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"},
+			nil,
+			"Records: 8  Deleted: 0  Skipped: 5  Warnings: 0",
+		},
+		{
+			nil,
+			[]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"},
+			nil,
+			"Records: 8  Deleted: 0  Skipped: 8  Warnings: 0",
+		},
+	}
+	deleteSQL := "DO 1"
+	selectSQL := "TABLE a;"
+	checkCases(tests, ld, t, tk, ctx, selectSQL, deleteSQL)
+	tk.MustExec("ADMIN CHECK TABLE a")
+}