diff --git a/.agents/skills/tidb-test-guidelines/references/dxf-case-map.md b/.agents/skills/tidb-test-guidelines/references/dxf-case-map.md index 38adab7b4f504..49bf6e2bfe000 100644 --- a/.agents/skills/tidb-test-guidelines/references/dxf-case-map.md +++ b/.agents/skills/tidb-test-guidelines/references/dxf-case-map.md @@ -99,7 +99,7 @@ - `pkg/dxf/importinto/collect_conflicts_test.go` - dxf/importinto: Tests collect conflicts step executor. - `pkg/dxf/importinto/conflict_resolution_test.go` - dxf/importinto: Tests conflict resolution step executor. - `pkg/dxf/importinto/encode_and_sort_operator_test.go` - dxf/importinto: Tests encode and sort operator. -- `pkg/dxf/importinto/job_testkit_test.go` - dxf/importinto: Tests submit task next-gen. +- `pkg/dxf/importinto/job_testkit_test.go` - dxf/importinto: Tests task submission and exact GetJobLastUpdateTime lookups across active/history subtasks. - `pkg/dxf/importinto/metrics_test.go` - dxf/importinto: Tests metric manager. - `pkg/dxf/importinto/planner_test.go` - dxf/importinto: Tests logical plan. - `pkg/dxf/importinto/proto_test.go` - dxf/importinto: Tests KV conflict info aggregation. @@ -109,6 +109,11 @@ - `pkg/dxf/importinto/task_executor_testkit_test.go` - dxf/importinto: Tests post process step executor. - `pkg/dxf/importinto/wrapper_test.go` - dxf/importinto: Tests chunk convert. +## pkg/dxf/importinto/jobhistory + +### Tests +- `pkg/dxf/importinto/jobhistory/history_test.go` - dxf/importinto/jobhistory: Tests target-only history aggregation for adjacent task IDs around 2^53. + ## pkg/dxf/importinto/conflictedkv ### Tests diff --git a/pkg/dxf/framework/integrationtests/framework_err_handling_test.go b/pkg/dxf/framework/integrationtests/framework_err_handling_test.go index b114a6bf95487..f39e9760ce4e2 100644 --- a/pkg/dxf/framework/integrationtests/framework_err_handling_test.go +++ b/pkg/dxf/framework/integrationtests/framework_err_handling_test.go @@ -83,7 +83,7 @@ func TestOnTaskError(t *testing.T) { taskKey := "key3-1" taskID := prepareForAwaitingResolutionTestFn(t, taskKey) tk := testkit.NewTestKit(t, c.Store) - tk.MustExec(fmt.Sprintf("update mysql.tidb_background_subtask set state='pending' where state='failed' and task_key= %d", taskID)) + tk.MustExec(fmt.Sprintf("update mysql.tidb_background_subtask set state='pending' where state='failed' and task_key= '%d'", taskID)) tk.MustExec(fmt.Sprintf("update mysql.tidb_global_task set state='running' where id = %d", taskID)) task := testutil.WaitTaskDone(c.Ctx, t, taskKey) require.Equal(t, proto.TaskStateSucceed, task.State) diff --git a/pkg/dxf/framework/integrationtests/modify_test.go b/pkg/dxf/framework/integrationtests/modify_test.go index b8de807460a6a..57b741f7bbd41 100644 --- a/pkg/dxf/framework/integrationtests/modify_test.go +++ b/pkg/dxf/framework/integrationtests/modify_test.go @@ -24,6 +24,7 @@ import ( "github.com/pingcap/tidb/pkg/dxf/framework/handle" mockexecute "github.com/pingcap/tidb/pkg/dxf/framework/mock/execute" "github.com/pingcap/tidb/pkg/dxf/framework/proto" + "github.com/pingcap/tidb/pkg/dxf/framework/storage" "github.com/pingcap/tidb/pkg/dxf/framework/taskexecutor/execute" "github.com/pingcap/tidb/pkg/dxf/framework/testutil" "github.com/pingcap/tidb/pkg/sessionctx" @@ -513,7 +514,7 @@ func TestModifyTaskMaxNodeCountForSubtaskBalance(t *testing.T) { return runtimeInfo.activeSubtaskCount.Load() == 1 }, 10*time.Second, 100*time.Millisecond) require.NoError(t, c.TaskMgr.WithNewSession(func(se sessionctx.Context) error { - rows, err2 := sqlexec.ExecSQL(c.Ctx, se.GetSQLExecutor(), "select count(distinct exec_id) from mysql.tidb_background_subtask where task_key = %?", task.ID) + rows, err2 := sqlexec.ExecSQL(c.Ctx, se.GetSQLExecutor(), "select count(distinct exec_id) from mysql.tidb_background_subtask where task_key = %?", storage.TaskIDToKey(task.ID)) require.NoError(t, err2) require.Equal(t, 1, len(rows)) require.EqualValues(t, 1, rows[0].GetInt64(0)) @@ -533,7 +534,7 @@ func TestModifyTaskMaxNodeCountForSubtaskBalance(t *testing.T) { return runtimeInfo.activeSubtaskCount.Load() == 2 }, 10*time.Second, 100*time.Millisecond) require.NoError(t, c.TaskMgr.WithNewSession(func(se sessionctx.Context) error { - rows, err2 := sqlexec.ExecSQL(c.Ctx, se.GetSQLExecutor(), "select count(distinct exec_id) from mysql.tidb_background_subtask where task_key = %?", task.ID) + rows, err2 := sqlexec.ExecSQL(c.Ctx, se.GetSQLExecutor(), "select count(distinct exec_id) from mysql.tidb_background_subtask where task_key = %?", storage.TaskIDToKey(task.ID)) require.NoError(t, err2) require.Equal(t, 1, len(rows)) require.EqualValues(t, 2, rows[0].GetInt64(0)) diff --git a/pkg/dxf/framework/storage/converter.go b/pkg/dxf/framework/storage/converter.go index ee62bbf19fd03..11172e6df5f95 100644 --- a/pkg/dxf/framework/storage/converter.go +++ b/pkg/dxf/framework/storage/converter.go @@ -90,6 +90,14 @@ func Row2Task(r chunk.Row) *proto.Task { return task } +// TaskIDToKey returns the canonical decimal key stored in background-subtask tables. +// due to history reason, the task_key column inside subtask tables are defined +// as varchar, but it actually stores the ID of the task which is integer. to +// make sure TiDB query can correct handle this, we need to convert it to string. +func TaskIDToKey(taskID int64) string { + return strconv.FormatInt(taskID, 10) +} + // row2BasicSubTask converts a row to a subtask with basic info func row2BasicSubTask(r chunk.Row) *proto.SubtaskBase { taskIDStr := r.GetString(2) diff --git a/pkg/dxf/framework/storage/history.go b/pkg/dxf/framework/storage/history.go index 80cfafb681ea4..33eb75f7d6cea 100644 --- a/pkg/dxf/framework/storage/history.go +++ b/pkg/dxf/framework/storage/history.go @@ -41,12 +41,12 @@ const ( // TransferSubtasks2HistoryWithSession transfer the selected subtasks into tidb_background_subtask_history table by taskID. func (*TaskManager) TransferSubtasks2HistoryWithSession(ctx context.Context, se sessionctx.Context, taskID int64) error { exec := se.GetSQLExecutor() - _, err := sqlexec.ExecSQL(ctx, exec, `insert into mysql.tidb_background_subtask_history select * from mysql.tidb_background_subtask where task_key = %?`, taskID) + _, err := sqlexec.ExecSQL(ctx, exec, `insert into mysql.tidb_background_subtask_history select * from mysql.tidb_background_subtask where task_key = %?`, TaskIDToKey(taskID)) if err != nil { return err } // delete taskID subtask - _, err = sqlexec.ExecSQL(ctx, exec, "delete from mysql.tidb_background_subtask where task_key = %?", taskID) + _, err = sqlexec.ExecSQL(ctx, exec, "delete from mysql.tidb_background_subtask where task_key = %?", TaskIDToKey(taskID)) return err } diff --git a/pkg/dxf/framework/storage/subtask_state.go b/pkg/dxf/framework/storage/subtask_state.go index 3c3188d253120..094b2fa339c03 100644 --- a/pkg/dxf/framework/storage/subtask_state.go +++ b/pkg/dxf/framework/storage/subtask_state.go @@ -80,7 +80,7 @@ func (mgr *TaskManager) FailSubtask(ctx context.Context, execID string, taskID i proto.SubtaskStateFailed, serializeErr(err), execID, - taskID, + TaskIDToKey(taskID), proto.SubtaskStatePending, proto.SubtaskStateRunning) return err1 @@ -102,7 +102,7 @@ func (mgr *TaskManager) CancelSubtask(ctx context.Context, execID string, taskID state in (%?, %?);`, proto.SubtaskStateCanceled, execID, - taskID, + TaskIDToKey(taskID), proto.SubtaskStatePending, proto.SubtaskStateRunning) return err1 @@ -114,7 +114,7 @@ func (mgr *TaskManager) PauseSubtasks(ctx context.Context, execID string, taskID return err } _, err := mgr.ExecuteSQLWithNewSession(ctx, - `update mysql.tidb_background_subtask set state = "paused" where task_key = %? and state in ("running", "pending") and exec_id = %?`, taskID, execID) + `update mysql.tidb_background_subtask set state = "paused" where task_key = %? and state in ("running", "pending") and exec_id = %?`, TaskIDToKey(taskID), execID) return err } @@ -124,7 +124,7 @@ func (mgr *TaskManager) ResumeSubtasks(ctx context.Context, taskID int64) error return err } _, err := mgr.ExecuteSQLWithNewSession(ctx, - `update mysql.tidb_background_subtask set state = "pending", error = null where task_key = %? and state = "paused"`, taskID) + `update mysql.tidb_background_subtask set state = "pending", error = null where task_key = %? and state = "paused"`, TaskIDToKey(taskID)) return err } diff --git a/pkg/dxf/framework/storage/table_test.go b/pkg/dxf/framework/storage/table_test.go index 47bc94fedc0ff..cc6f248833840 100644 --- a/pkg/dxf/framework/storage/table_test.go +++ b/pkg/dxf/framework/storage/table_test.go @@ -329,7 +329,7 @@ func TestSwitchTaskStep(t *testing.T) { require.Equal(t, proto.ExtraParams{}, persistedTask.ExtraParams) require.Zero(t, persistedTask.StartTime) require.GreaterOrEqual(t, persistedTask.StateUpdateTime, switchTime) - tk.MustQuery(fmt.Sprintf("select count(1) from mysql.tidb_background_subtask where task_key = %d", prepareTaskID)). + tk.MustQuery(fmt.Sprintf("select count(1) from mysql.tidb_background_subtask where task_key = '%d'", prepareTaskID)). Check(testkit.Rows("0")) prepareTask.Meta = []byte(`{"prepare":"stale-owner"}`) @@ -376,7 +376,7 @@ func TestGetSubtaskSummaries(t *testing.T) { bytes, err := json.Marshal(summary) require.NoError(t, err) - tk.MustExec(fmt.Sprintf("update mysql.tidb_background_subtask set summary = '%s' where task_key = %d", string(bytes), task.ID)) + tk.MustExec(fmt.Sprintf("update mysql.tidb_background_subtask set summary = '%s' where task_key = '%d'", string(bytes), task.ID)) summaries, err := tm.GetAllSubtaskSummaryByStep(ctx, subtasks[0].TaskID, proto.StepOne) require.NoError(t, err) require.Len(t, summaries, len(subtasks)) @@ -388,7 +388,7 @@ func TestGetSubtaskSummaries(t *testing.T) { // If the JSON value is wrong, we still get an empty summary. // This can only happen if the summary field is manually updated. // It's acceptable, as the correct summary will be set by the executor later. - tk.MustExec(fmt.Sprintf(`update mysql.tidb_background_subtask set summary = '{"wrong_key": 123}' where task_key = %d`, task.ID)) + tk.MustExec(fmt.Sprintf(`update mysql.tidb_background_subtask set summary = '{"wrong_key": 123}' where task_key = '%d'`, task.ID)) summaries, err = tm.GetAllSubtaskSummaryByStep(ctx, subtasks[0].TaskID, proto.StepOne) require.NoError(t, err) for _, summary := range summaries { @@ -613,6 +613,26 @@ func TestGetActiveSubtasks(t *testing.T) { require.Equal(t, proto.SubtaskStateRunning, activeSubtasks[0].State) require.Equal(t, int64(3), activeSubtasks[1].ID) require.Equal(t, proto.SubtaskStatePending, activeSubtasks[1].State) + + const ( + adjacentTaskID = int64(9007199254740992) + targetTaskID = int64(9007199254740993) + largeIDExec = "large-id-exec" + ) + testutil.InsertSubtask(t, tm, adjacentTaskID, proto.StepOne, largeIDExec, nil, proto.SubtaskStatePending, proto.TaskTypeExample, 8) + testutil.InsertSubtask(t, tm, targetTaskID, proto.StepOne, largeIDExec, nil, proto.SubtaskStatePending, proto.TaskTypeExample, 8) + + activeSubtasks, err = tm.GetActiveSubtasks(ctx, targetTaskID) + require.NoError(t, err) + require.Len(t, activeSubtasks, 1) + require.Equal(t, targetTaskID, activeSubtasks[0].TaskID) + + filteredSubtasks, err := tm.GetSubtasksByExecIDAndStepAndStates( + ctx, largeIDExec, targetTaskID, proto.StepOne, proto.SubtaskStatePending, + ) + require.NoError(t, err) + require.Len(t, filteredSubtasks, 1) + require.Equal(t, targetTaskID, filteredSubtasks[0].TaskID) } func TestSubTaskTable(t *testing.T) { diff --git a/pkg/dxf/framework/storage/task_state.go b/pkg/dxf/framework/storage/task_state.go index 16d895c72a654..358dd28f78438 100644 --- a/pkg/dxf/framework/storage/task_state.go +++ b/pkg/dxf/framework/storage/task_state.go @@ -118,7 +118,7 @@ func (mgr *TaskManager) PauseTaskOnError(ctx context.Context, taskID int64, task state_update_time = unix_timestamp(), end_time = null where task_key = %? and step = %? and state = %?`, - proto.SubtaskStatePaused, taskID, step, proto.SubtaskStateFailed, + proto.SubtaskStatePaused, TaskIDToKey(taskID), step, proto.SubtaskStateFailed, ) return err }) @@ -302,7 +302,7 @@ func (mgr *TaskManager) ModifiedTask(ctx context.Context, task *proto.Task) erro update mysql.tidb_background_subtask set concurrency = %?, state_update_time = unix_timestamp() where task_key = %? and state in (%?, %?, %?)`, - task.RequiredSlots, task.ID, + task.RequiredSlots, TaskIDToKey(task.ID), proto.SubtaskStatePending, proto.SubtaskStateRunning, proto.SubtaskStatePaused) return err }) diff --git a/pkg/dxf/framework/storage/task_table.go b/pkg/dxf/framework/storage/task_table.go index 0e037d3e7461c..01b0e6e0dc207 100644 --- a/pkg/dxf/framework/storage/task_table.go +++ b/pkg/dxf/framework/storage/task_table.go @@ -593,7 +593,7 @@ func (mgr *TaskManager) GetSubtasksByExecIDAndStepAndStates(ctx context.Context, if err := injectfailpoint.DXFRandomErrorWithOnePercent(); err != nil { return nil, err } - args := []any{execID, taskID, step} + args := []any{execID, TaskIDToKey(taskID), step} for _, state := range states { args = append(args, state) } @@ -616,7 +616,7 @@ func (mgr *TaskManager) GetFirstSubtaskInStates(ctx context.Context, tidbID stri if err := injectfailpoint.DXFRandomErrorWithOnePercent(); err != nil { return nil, err } - args := []any{tidbID, taskID, step} + args := []any{tidbID, TaskIDToKey(taskID), step} for _, state := range states { args = append(args, state) } @@ -641,7 +641,7 @@ func (mgr *TaskManager) GetActiveSubtasks(ctx context.Context, taskID int64) ([] rs, err := mgr.ExecuteSQLWithNewSession(ctx, ` select `+basicSubtaskColumns+` from mysql.tidb_background_subtask where task_key = %? and state in (%?, %?)`, - taskID, proto.SubtaskStatePending, proto.SubtaskStateRunning) + TaskIDToKey(taskID), proto.SubtaskStatePending, proto.SubtaskStateRunning) if err != nil { return nil, err } @@ -659,7 +659,7 @@ func (mgr *TaskManager) GetAllSubtasksByStepAndState(ctx context.Context, taskID } rs, err := mgr.ExecuteSQLWithNewSession(ctx, `select `+SubtaskColumns+` from mysql.tidb_background_subtask where task_key = %? and state = %? and step = %?`, - taskID, state, step) + TaskIDToKey(taskID), state, step) if err != nil { return nil, err } @@ -684,7 +684,7 @@ func (mgr *TaskManager) GetAllSubtaskSummaryByStep( rs, err := mgr.ExecuteSQLWithNewSession(ctx, `select summary from mysql.tidb_background_subtask where task_key = %? and step = %?`, - taskID, step) + TaskIDToKey(taskID), step) if err != nil { return nil, err } @@ -714,7 +714,7 @@ func (mgr *TaskManager) GetSubtaskRowCount(ctx context.Context, taskID int64, st union all select summary from mysql.tidb_background_subtask_history where task_key = %? and step = %? ) as combined`, - taskID, step, taskID, step) + TaskIDToKey(taskID), step, TaskIDToKey(taskID), step) if err != nil { return 0, err } @@ -750,7 +750,7 @@ func (mgr *TaskManager) GetSubtaskCntGroupByStates(ctx context.Context, taskID i from mysql.tidb_background_subtask where task_key = %? and step = %? group by state`, - taskID, step) + TaskIDToKey(taskID), step) if err != nil { return nil, err } @@ -773,7 +773,7 @@ func (mgr *TaskManager) GetSubtaskStateCntAndErrorsByStep(ctx context.Context, t `select state, error from mysql.tidb_background_subtask where task_key = %? and step = %?`, - taskID, step) + TaskIDToKey(taskID), step) if err != nil { return nil, nil, err } @@ -806,7 +806,7 @@ func (mgr *TaskManager) GetSubtaskErrors(ctx context.Context, taskID int64) ([]e } rs, err := mgr.ExecuteSQLWithNewSession(ctx, `select error from mysql.tidb_background_subtask - where task_key = %? AND state in (%?, %?)`, taskID, proto.SubtaskStateFailed, proto.SubtaskStateCanceled) + where task_key = %? AND state in (%?, %?)`, TaskIDToKey(taskID), proto.SubtaskStateFailed, proto.SubtaskStateCanceled) if err != nil { return nil, err } @@ -991,7 +991,7 @@ func (mgr *TaskManager) SwitchTaskStepInBatch( // some subtasks may be inserted by other schedulers, we can skip them. rs, err := sqlexec.ExecSQL(ctx, se.GetSQLExecutor(), ` select count(1) from mysql.tidb_background_subtask - where task_key = %? and step = %?`, task.ID, nextStep) + where task_key = %? and step = %?`, TaskIDToKey(task.ID), nextStep) if err != nil { return err } @@ -1063,7 +1063,7 @@ func (mgr *TaskManager) GetSubtasksWithHistory(ctx context.Context, taskID int64 err = mgr.WithNewTxn(ctx, func(se sessionctx.Context) error { rs, err = sqlexec.ExecSQL(ctx, se.GetSQLExecutor(), `select `+SubtaskColumns+` from mysql.tidb_background_subtask where task_key = %? and step = %?`, - taskID, step, + TaskIDToKey(taskID), step, ) if err != nil { return err @@ -1073,7 +1073,7 @@ func (mgr *TaskManager) GetSubtasksWithHistory(ctx context.Context, taskID int64 // when the user show import jobs, we need to check the history table. rsFromHistory, err := sqlexec.ExecSQL(ctx, se.GetSQLExecutor(), `select `+SubtaskColumns+` from mysql.tidb_background_subtask_history where task_key = %? and step = %?`, - taskID, step, + TaskIDToKey(taskID), step, ) if err != nil { return err diff --git a/pkg/dxf/framework/testutil/table_util.go b/pkg/dxf/framework/testutil/table_util.go index d54960d9eb5a5..71aa1e6b5e354 100644 --- a/pkg/dxf/framework/testutil/table_util.go +++ b/pkg/dxf/framework/testutil/table_util.go @@ -99,7 +99,7 @@ func GetSubtasksFromHistory(ctx context.Context, mgr *storage.TaskManager) (int, // GetSubtasksFromHistoryByTaskID gets subtasks by taskID from history table for test. func GetSubtasksFromHistoryByTaskID(ctx context.Context, mgr *storage.TaskManager, taskID int64) (int, error) { rs, err := mgr.ExecuteSQLWithNewSession(ctx, - `select `+storage.SubtaskColumns+` from mysql.tidb_background_subtask_history where task_key = %?`, taskID) + `select `+storage.SubtaskColumns+` from mysql.tidb_background_subtask_history where task_key = %?`, storage.TaskIDToKey(taskID)) if err != nil { return 0, err } @@ -109,7 +109,7 @@ func GetSubtasksFromHistoryByTaskID(ctx context.Context, mgr *storage.TaskManage // GetSubtasksByTaskID gets subtasks by taskID for test. func GetSubtasksByTaskID(ctx context.Context, mgr *storage.TaskManager, taskID int64) ([]*proto.Subtask, error) { rs, err := mgr.ExecuteSQLWithNewSession(ctx, - `select `+storage.SubtaskColumns+` from mysql.tidb_background_subtask where task_key = %?`, taskID) + `select `+storage.SubtaskColumns+` from mysql.tidb_background_subtask where task_key = %?`, storage.TaskIDToKey(taskID)) if err != nil { return nil, err } @@ -170,7 +170,7 @@ func GetSubtaskNodes(ctx context.Context, mgr *storage.TaskManager, taskID int64 rs, err := mgr.ExecuteSQLWithNewSession(ctx, ` select distinct(exec_id) from mysql.tidb_background_subtask where task_key=%? union - select distinct(exec_id) from mysql.tidb_background_subtask_history where task_key=%?`, taskID, taskID) + select distinct(exec_id) from mysql.tidb_background_subtask_history where task_key=%?`, storage.TaskIDToKey(taskID), storage.TaskIDToKey(taskID)) if err != nil { return nil, err } @@ -220,7 +220,7 @@ func GetTasksFromHistoryInStates(ctx context.Context, mgr *storage.TaskManager, // DeleteSubtasksByTaskID deletes the subtask of the given task ID. func DeleteSubtasksByTaskID(ctx context.Context, mgr *storage.TaskManager, taskID int64) error { _, err := mgr.ExecuteSQLWithNewSession(ctx, `delete from mysql.tidb_background_subtask - where task_key = %?`, taskID) + where task_key = %?`, storage.TaskIDToKey(taskID)) if err != nil { return err } @@ -244,9 +244,9 @@ func IsTaskCancelling(ctx context.Context, mgr *storage.TaskManager, taskID int6 // PrintSubtaskInfo log the subtask info by taskKey for test. func PrintSubtaskInfo(ctx context.Context, mgr *storage.TaskManager, taskID int64) { rs, _ := mgr.ExecuteSQLWithNewSession(ctx, - `select `+storage.SubtaskColumns+` from mysql.tidb_background_subtask_history where task_key = %?`, taskID) + `select `+storage.SubtaskColumns+` from mysql.tidb_background_subtask_history where task_key = %?`, storage.TaskIDToKey(taskID)) rs2, _ := mgr.ExecuteSQLWithNewSession(ctx, - `select `+storage.SubtaskColumns+` from mysql.tidb_background_subtask where task_key = %?`, taskID) + `select `+storage.SubtaskColumns+` from mysql.tidb_background_subtask where task_key = %?`, storage.TaskIDToKey(taskID)) rs = append(rs, rs2...) for _, r := range rs { diff --git a/pkg/dxf/importinto/BUILD.bazel b/pkg/dxf/importinto/BUILD.bazel index 8789516ec917c..0a829d33f631f 100644 --- a/pkg/dxf/importinto/BUILD.bazel +++ b/pkg/dxf/importinto/BUILD.bazel @@ -112,7 +112,7 @@ go_test( ], embed = [":importinto"], flaky = True, - shard_count = 29, + shard_count = 30, deps = [ "//pkg/config", "//pkg/config/kerneltype", diff --git a/pkg/dxf/importinto/job.go b/pkg/dxf/importinto/job.go index bf8aa269b28b8..94e66c58b4e2e 100644 --- a/pkg/dxf/importinto/job.go +++ b/pkg/dxf/importinto/job.go @@ -404,7 +404,7 @@ func GetJobLastUpdateTime(ctx context.Context, jobID int64) (types.Time, error) union select state_update_time from mysql.tidb_background_subtask_history where task_key = %? ) t`, - task.ID, task.ID, + storage.TaskIDToKey(task.ID), storage.TaskIDToKey(task.ID), ) return err }) diff --git a/pkg/dxf/importinto/job_testkit_test.go b/pkg/dxf/importinto/job_testkit_test.go index 6979809ac208e..79aaa535e4a6a 100644 --- a/pkg/dxf/importinto/job_testkit_test.go +++ b/pkg/dxf/importinto/job_testkit_test.go @@ -314,6 +314,68 @@ func TestGetTaskImportedRows(t *testing.T) { require.Equal(t, "30", runInfo.Percent()) } +func TestGetJobLastUpdateTime(t *testing.T) { + _, manager, ctx := testutil.InitTableTest(t) + require.NoError(t, manager.InitMeta(ctx, ":4000", "")) + + const ( + jobID = int64(9527) + adjacentTaskID = int64(9007199254740992) + targetTaskID = int64(9007199254740993) + targetHistoryTime = int64(1000) + adjacentHistTime = int64(2000) + targetActiveTime = int64(1500) + adjacentLiveTime = int64(2500) + ) + _, err := manager.ExecuteSQLWithNewSession(ctx, + "alter table mysql.tidb_global_task auto_increment = 9007199254740992") + require.NoError(t, err) + + gotAdjacentTaskID, err := manager.CreateTask(ctx, importinto.TaskKey(jobID+1), proto.ImportInto, + "", 1, "", 0, proto.ExtraParams{}, nil) + require.NoError(t, err) + require.Equal(t, adjacentTaskID, gotAdjacentTaskID) + gotTargetTaskID, err := manager.CreateTask(ctx, importinto.TaskKey(jobID), proto.ImportInto, + "", 1, "", 0, proto.ExtraParams{}, nil) + require.NoError(t, err) + require.Equal(t, targetTaskID, gotTargetTaskID) + + updateSubtaskTime := func(id, updateTime int64) { + _, err = manager.ExecuteSQLWithNewSession(ctx, ` + update mysql.tidb_background_subtask + set state_update_time = %? + where id = %?`, updateTime, id) + require.NoError(t, err) + } + targetHistoryID := testutil.InsertSubtask(t, manager, targetTaskID, proto.ImportStepEncodeAndSort, + "tidb-1", nil, proto.SubtaskStateSucceed, proto.ImportInto, 1) + adjacentHistoryID := testutil.InsertSubtask(t, manager, adjacentTaskID, proto.ImportStepEncodeAndSort, + "tidb-1", nil, proto.SubtaskStateSucceed, proto.ImportInto, 1) + updateSubtaskTime(targetHistoryID, targetHistoryTime) + updateSubtaskTime(adjacentHistoryID, adjacentHistTime) + + targetTask, err := manager.GetTaskByID(ctx, targetTaskID) + require.NoError(t, err) + adjacentTask, err := manager.GetTaskByID(ctx, adjacentTaskID) + require.NoError(t, err) + require.NoError(t, manager.TransferTasks2History(ctx, []*proto.Task{targetTask, adjacentTask})) + + targetActiveID := testutil.InsertSubtask(t, manager, targetTaskID, proto.ImportStepWriteAndIngest, + "tidb-1", nil, proto.SubtaskStateRunning, proto.ImportInto, 1) + adjacentActiveID := testutil.InsertSubtask(t, manager, adjacentTaskID, proto.ImportStepWriteAndIngest, + "tidb-1", nil, proto.SubtaskStateRunning, proto.ImportInto, 1) + updateSubtaskTime(targetActiveID, targetActiveTime) + updateSubtaskTime(adjacentActiveID, adjacentLiveTime) + + expectedRows, err := manager.ExecuteSQLWithNewSession(ctx, "select from_unixtime(%?)", targetActiveTime) + require.NoError(t, err) + require.Len(t, expectedRows, 1) + + lastUpdateTime, err := importinto.GetJobLastUpdateTime(ctx, jobID) + require.NoError(t, err) + require.Equal(t, expectedRows[0].GetTime(0), lastUpdateTime) +} + func TestShowImportProgress(t *testing.T) { testfailpoint.Enable(t, "github.com/pingcap/tidb/pkg/domain/MockDisableDistTask", "return(true)") diff --git a/pkg/dxf/importinto/jobhistory/history.go b/pkg/dxf/importinto/jobhistory/history.go index 8dc04b7308f81..6e7c8c39e3657 100644 --- a/pkg/dxf/importinto/jobhistory/history.go +++ b/pkg/dxf/importinto/jobhistory/history.go @@ -137,7 +137,7 @@ func GetFromHistory( from mysql.tidb_background_subtask_history where task_key = %? group by step, kv_group`, - info.TaskID) + storage.TaskIDToKey(info.TaskID)) if err != nil { return nil, err } diff --git a/pkg/dxf/importinto/jobhistory/history_test.go b/pkg/dxf/importinto/jobhistory/history_test.go index 5b338f45edeb4..1f8eed7decf08 100644 --- a/pkg/dxf/importinto/jobhistory/history_test.go +++ b/pkg/dxf/importinto/jobhistory/history_test.go @@ -29,9 +29,19 @@ func TestGetFromHistory(t *testing.T) { require.NoError(t, tm.InitMeta(ctx, ":4000", "")) const ( - keyspace = "ks1" - jobID = int64(9527) + keyspace = "ks1" + jobID = int64(9527) + adjacentTaskID = int64(9007199254740992) + targetTaskID = int64(9007199254740993) ) + _, err := tm.ExecuteSQLWithNewSession(ctx, + "alter table mysql.tidb_global_task auto_increment = 9007199254740992") + require.NoError(t, err) + gotAdjacentTaskID, err := tm.CreateTask(ctx, taskkey.ForJobInKeyspace(keyspace, jobID+1), proto.ImportInto, + keyspace, 8, "", 4, proto.ExtraParams{}, nil) + require.NoError(t, err) + require.Equal(t, adjacentTaskID, gotAdjacentTaskID) + taskMeta := []byte(`{ "Plan": { "DistSQLScanConcurrency": 16, @@ -47,6 +57,7 @@ func TestGetFromHistory(t *testing.T) { }`) taskID, err := tm.CreateTask(ctx, taskkey.ForJobInKeyspace(keyspace, jobID), proto.ImportInto, keyspace, 8, "", 4, proto.ExtraParams{}, taskMeta) require.NoError(t, err) + require.Equal(t, targetTaskID, taskID) encodeID := testutil.InsertSubtask(t, tm, taskID, proto.ImportStepEncodeAndSort, "tidb-1", []byte(`{"kv-group":"data"}`), proto.SubtaskStateSucceed, proto.ImportInto, 8) @@ -69,10 +80,15 @@ func TestGetFromHistory(t *testing.T) { updateSubtask(dataID, 700, 2500, `{"bytes": 1073741824}`, `{"kv-group":"data"}`) updateSubtask(indexID, 900, 2100, `{"bytes": 536870912}`, `{"kv-group":"index-1"}`) updateSubtask(invalidDurationID, 0, 0, `{"bytes": 0}`, `{"kv-group":"data"}`) + adjacentID := testutil.InsertSubtask(t, tm, adjacentTaskID, proto.ImportStepWriteAndIngest, "tidb-2", + []byte(`{"kv-group":"data"}`), proto.SubtaskStateSucceed, proto.ImportInto, 8) + updateSubtask(adjacentID, 50, 4000, `{"bytes": 4294967296}`, `{"kv-group":"data"}`) task, err := tm.GetTaskByID(ctx, taskID) require.NoError(t, err) - require.NoError(t, tm.TransferTasks2History(ctx, []*proto.Task{task})) + adjacentTask, err := tm.GetTaskByID(ctx, adjacentTaskID) + require.NoError(t, err) + require.NoError(t, tm.TransferTasks2History(ctx, []*proto.Task{task, adjacentTask})) info, err := jobhistory.GetFromHistory(ctx, tm, keyspace, jobID) require.NoError(t, err) @@ -100,6 +116,6 @@ func TestGetFromHistory(t *testing.T) { require.Empty(t, info.Duration.ResolveConflicts) require.Empty(t, info.Duration.PostProcess) - _, err = jobhistory.GetFromHistory(ctx, tm, keyspace, jobID+1) + _, err = jobhistory.GetFromHistory(ctx, tm, keyspace, jobID+2) require.ErrorContains(t, err, "not found in history") } diff --git a/tests/realtikvtest/addindextest1/disttask_test.go b/tests/realtikvtest/addindextest1/disttask_test.go index fa749d9874e00..c8d618dae2552 100644 --- a/tests/realtikvtest/addindextest1/disttask_test.go +++ b/tests/realtikvtest/addindextest1/disttask_test.go @@ -570,12 +570,12 @@ func TestAddIndexScheduleAway(t *testing.T) { tk1.MustExec("use test") updateExecID := fmt.Sprintf(` update mysql.tidb_background_subtask set exec_id = 'other' where task_key in - (select id from mysql.tidb_global_task where task_key like '%%%d')`, jobID.Load()) + (select CAST(id AS CHAR) from mysql.tidb_global_task where task_key like '%%%d')`, jobID.Load()) tk1.MustExec(updateExecID) <-afterCancel updateExecID = fmt.Sprintf(` update mysql.tidb_background_subtask set exec_id = ':4000' where task_key in - (select id from mysql.tidb_global_task where task_key like '%%%d')`, jobID.Load()) + (select CAST(id AS CHAR) from mysql.tidb_global_task where task_key like '%%%d')`, jobID.Load()) tk1.MustExec(updateExecID) }) }) diff --git a/tests/realtikvtest/addindextest3/temp_index_test.go b/tests/realtikvtest/addindextest3/temp_index_test.go index 5aa12d276a585..39d3d1805ed9a 100644 --- a/tests/realtikvtest/addindextest3/temp_index_test.go +++ b/tests/realtikvtest/addindextest3/temp_index_test.go @@ -160,9 +160,9 @@ func TestMergeTempIndexBasic(t *testing.T) { require.Len(t, taskIDRows, 2) taskID := taskIDRows[0][0].(string) mergeTaskID := taskIDRows[1][0].(string) - readIdxCntSQL := fmt.Sprintf("select json_extract(summary, '$.row_count') from all_subtasks where task_key = %s and step = 1", taskID) + readIdxCntSQL := fmt.Sprintf("select json_extract(summary, '$.row_count') from all_subtasks where task_key = '%s' and step = 1", taskID) tk.MustQuery(readIdxCntSQL).Check(testkit.Rows(tc.readIdxRowCnt...)) - mergeCntSQL := fmt.Sprintf("select json_extract(summary, '$.row_count') from all_subtasks where task_key = %s and step = 4", mergeTaskID) + mergeCntSQL := fmt.Sprintf("select json_extract(summary, '$.row_count') from all_subtasks where task_key = '%s' and step = 4", mergeTaskID) tk.MustQuery(mergeCntSQL).Check(testkit.Rows(tc.mergeIdxCnt...)) } } diff --git a/tests/realtikvtest/importintotest4/import_summary_test.go b/tests/realtikvtest/importintotest4/import_summary_test.go index 2cd8c3463cc8e..b0b26477aed7d 100644 --- a/tests/realtikvtest/importintotest4/import_summary_test.go +++ b/tests/realtikvtest/importintotest4/import_summary_test.go @@ -80,8 +80,7 @@ func (s *mockGCSSuite) TestGlobalSortSummary() { idSQL := `select id from mysql.tidb_global_task where task_key = ? union select id from mysql.tidb_global_task_history where task_key = ?` rs = s.tk.MustQuery(idSQL, taskKey, taskKey).Rows() - id, err := strconv.Atoi(rs[0][0].(string)) - require.NoError(s.T(), err) + id := rs[0][0].(string) sql := ` SELECT step, summary