Skip to content

Commit

Permalink
executor: fix data inconsistency after `load data ... replace into ..…
Browse files Browse the repository at this point in the history
….` (#56415) (#56508)

close #56408
  • Loading branch information
ti-chi-bot authored Oct 9, 2024
1 parent cceec97 commit 74983c9
Show file tree
Hide file tree
Showing 3 changed files with 22 additions and 6 deletions.
6 changes: 1 addition & 5 deletions pkg/executor/insert_common.go
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
2 changes: 1 addition & 1 deletion pkg/executor/test/loaddatatest/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ go_test(
],
flaky = True,
race = "on",
shard_count = 10,
shard_count = 11,
deps = [
"//br/pkg/lightning/mydump",
"//pkg/config",
Expand Down
20 changes: 20 additions & 0 deletions pkg/executor/test/loaddatatest/load_data_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -476,3 +476,23 @@ 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);")
tk.MustExec("LOAD DATA LOCAL INFILE '/tmp/nonexistence.csv' REPLACE INTO TABLE a FIELDS terminated by '|';")
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 := "delete from a;"
selectSQL := "select * from a;"
checkCases(tests, ld, t, tk, ctx, selectSQL, deleteSQL)
tk.MustExec("ADMIN CHECK TABLE a")
}

0 comments on commit 74983c9

Please sign in to comment.