Repository: pingcap/tidb (go)

Issue:
Row count maybe inaccurate when adding index with DXF
## Bug Report

Please answer these questions before submitting your issue. Thanks!

### 1. Minimal reproduce step (Required)

Add indexes on tables with 10k rows.

<!-- a step by step guide for reproducing the bug. -->

### 2. What did you expect to see? (Required)

All adding index jobs row count should be 10000.

### 3. What did you see instead (Required)

```
mysql> admin show ddl jobs;
+--------+---------+------------+------------------------+--------------+-----------+----------+-----------+----------------------------+----------------------------+----------------------------+--------+
| JOB_ID | DB_NAME | TABLE_NAME | JOB_TYPE               | SCHEMA_STATE | SCHEMA_ID | TABLE_ID | ROW_COUNT | CREATE_TIME                | START_TIME                 | END_TIME                   | STATE  |
+--------+---------+------------+------------------------+--------------+-----------+----------+-----------+----------------------------+----------------------------+----------------------------+--------+
|    596 | test    | sbtest27   | add index /* ingest */ | public       |       195 |      257 |     10000 | 2024-10-29 05:38:35.684000 | 2024-10-29 05:40:26.534000 | 2024-10-29 05:40:28.433000 | synced |
|    595 | test    | sbtest4    | add index /* ingest */ | public       |       195 |      203 |     10000 | 2024-10-29 05:38:35.684000 | 2024-10-29 05:40:26.334000 | 2024-10-29 05:40:30.034000 | synced |
|    594 | test    | sbtest99   | add index /* ingest */ | public       |       195 |      440 |     10000 | 2024-10-29 05:38:35.684000 | 2024-10-29 05:40:24.584000 | 2024-10-29 05:40:26.484000 | synced |
|    593 | test    | sbtest81   | add index /* ingest */ | public       |       195 |      470 |     10000 | 2024-10-29 05:38:35.684000 | 2024-10-29 05:40:24.334000 | 2024-10-29 05:40:26.283000 | synced |
|    592 | test    | sbtest47   | add index /* ingest */ | public       |       195 |      313 |         0 | 2024-10-29 05:38:35.633000 | 2024-10-29 05:40:22.684000 | 2024-10-29 05:40:24.584000 | synced |
|    591 | test    | sbtest91   | add index /* ingest */ | public       |       195 |      422 |     10000 | 2024-10-29 05:38:35.633000 | 2024-10-29 05:40:22.434000 | 2024-10-29 05:40:24.334000 | synced |
|    590 | test    | sbtest85   | add index /* ingest */ | public       |       195 |      476 |         0 | 2024-10-29 05:38:35.584000 | 2024-10-29 05:40:20.683000 | 2024-10-29 05:40:22.634000 | synced |
|    589 | test    | sbtest97   | add index /* ingest */ | public       |       195 |      494 |     10000 | 2024-10-29 05:38:35.584000 | 2024-10-29 05:40:20.483000 | 2024-10-29 05:40:22.434000 | synced |
|    588 | test    | sbtest28   | add index /* ingest */ | public       |       195 |      259 |     10000 | 2024-10-29 05:38:35.584000 | 2024-10-29 05:40:16.933000 | 2024-10-29 05:40:20.683000 | synced |
|    587 | test    | sbtest9    | add index /* ingest */ | public       |       195 |      235 |     10000 | 2024-10-29 05:38:35.484000 | 2024-10-29 05:40:16.734000 | 2024-10-29 05:40:20.483000 | synced |
+--------+---------+------------+------------------------+--------------+-----------+----------+-----------+----------------------------+----------------------------+----------------------------+--------+
```

```
[2024/10/29 05:40:24.552 +00:00] [WARN] [index.go:2684] ["cannot get task"] [category=ddl] [task_key=ddl/backfill/592] [error="task not found"]
```

### 4. What is your TiDB version? (Required)

<!-- Paste the output of SELECT tidb_version() -->

c550aae6

Pull request description:
dxf: get task from both `mysql.tidb_global_task` and `mysql.tidb_global_task_history` (#57063)
This is an automated cherry-pick of #57063

<!--

Thank you for contributing to TiDB!

PR Title Format:
1. pkg [, pkg2, pkg3]: what's changed
2. *: what's changed

-->

### What problem does this PR solve?
<!--

Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.

-->

Issue Number: close #56942

Problem Summary:
https://github.com/pingcap/tidb/issues/56942#issue-2620263734
### What changed and how does it work?
Get the task from `mysql.tidb_global_task` and `mysql.tidb_global_task_history` in `updateDistTaskRowCount`, and sum up the row count from both `tidb_background_subtask` and `tidb_background_subtask_history` table.
### Check List

Tests <!-- At least one of them must be included. -->

- [ ] Unit test
- [ ] Integration test
- [x] Manual test (add detailed scripts or steps below)
1. Use sysbench create `100` tables with `10000` rows each.
2. After all ddl jobs finished, check the ddl job row count, the `row_count` should all be `10000`:
```
mysql> admin show ddl jobs 100 where row_count != 10000;
Empty set (0.03 sec)
```

- [ ] No need to test
  > - [ ] I checked and no code files have been changed.
  > <!-- Or your custom  "No need to test" reasons -->

Side effects

- [ ] Performance regression: Consumes more CPU
- [ ] Performance regression: Consumes more Memory
- [ ] Breaking backward compatibility

Documentation

- [ ] Affects user behaviors
- [ ] Contains syntax changes
- [ ] Contains variable changes
- [ ] Contains experimental features
- [ ] Changes MySQL compatibility

### Release note

<!-- compatibility change, improvement, bugfix, and new feature need a release note -->

Please refer to [Release Notes Language Style Guide](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/release-notes-style-guide.html) to write a quality release note.

```release-note
None
```


Patch:
diff --git a/pkg/ddl/index.go b/pkg/ddl/index.go
index 132ecf498a..838205e585 100644
--- a/pkg/ddl/index.go
+++ b/pkg/ddl/index.go
@@ -2266,7 +2266,7 @@ func (w *worker) updateJobRowCount(taskKey string, jobID int64) {
 		logutil.DDLLogger().Warn("cannot get task manager", zap.String("task_key", taskKey), zap.Error(err))
 		return
 	}
-	task, err := taskMgr.GetTaskByKey(w.ctx, taskKey)
+	task, err := taskMgr.GetTaskByKeyWithHistory(w.ctx, taskKey)
 	if err != nil {
 		logutil.DDLLogger().Warn("cannot get task", zap.String("task_key", taskKey), zap.Error(err))
 		return
diff --git a/pkg/disttask/framework/storage/task_table.go b/pkg/disttask/framework/storage/task_table.go
index aa8f82262f..aaeee4b446 100644
--- a/pkg/disttask/framework/storage/task_table.go
+++ b/pkg/disttask/framework/storage/task_table.go
@@ -480,8 +480,12 @@ func (mgr *TaskManager) GetAllSubtasksByStepAndState(ctx context.Context, taskID
 func (mgr *TaskManager) GetSubtaskRowCount(ctx context.Context, taskID int64, step proto.Step) (int64, error) {
 	rs, err := mgr.ExecuteSQLWithNewSession(ctx, `select
     	cast(sum(json_extract(summary, '$.row_count')) as signed) as row_count
-		from mysql.tidb_background_subtask where task_key = %? and step = %?`,
-		taskID, step)
+		from (
+			select summary from mysql.tidb_background_subtask where task_key = %? and step = %?
+			union all
+			select summary from mysql.tidb_background_subtask_history where task_key = %? and step = %?
+		) as combined`,
+		taskID, step, taskID, step)
 	if err != nil {
 		return 0, err
 	}
diff --git a/pkg/executor/importer/import.go b/pkg/executor/importer/import.go
index 9cbd8b0f99..c0467e97b0 100644
--- a/pkg/executor/importer/import.go
+++ b/pkg/executor/importer/import.go
@@ -79,20 +79,30 @@ const (
 	// 0 means no limit
 	unlimitedWriteSpeed = config.ByteSize(0)
 
-	characterSetOption          = "character_set"
-	fieldsTerminatedByOption    = "fields_terminated_by"
-	fieldsEnclosedByOption      = "fields_enclosed_by"
-	fieldsEscapedByOption       = "fields_escaped_by"
-	fieldsDefinedNullByOption   = "fields_defined_null_by"
-	linesTerminatedByOption     = "lines_terminated_by"
-	skipRowsOption              = "skip_rows"
-	splitFileOption             = "split_file"
-	diskQuotaOption             = "disk_quota"
-	threadOption                = "thread"
-	maxWriteSpeedOption         = "max_write_speed"
-	checksumTableOption         = "checksum_table"
-	recordErrorsOption          = "record_errors"
-	detachedOption              = "detached"
+	characterSetOption        = "character_set"
+	fieldsTerminatedByOption  = "fields_terminated_by"
+	fieldsEnclosedByOption    = "fields_enclosed_by"
+	fieldsEscapedByOption     = "fields_escaped_by"
+	fieldsDefinedNullByOption = "fields_defined_null_by"
+	linesTerminatedByOption   = "lines_terminated_by"
+	skipRowsOption            = "skip_rows"
+	splitFileOption           = "split_file"
+	diskQuotaOption           = "disk_quota"
+	threadOption              = "thread"
+	maxWriteSpeedOption       = "max_write_speed"
+	checksumTableOption       = "checksum_table"
+	recordErrorsOption        = "record_errors"
+	detachedOption            = "detached"
+	// if 'import mode' enabled, TiKV will:
+	//  - set level0_stop_writes_trigger = max(old, 1 << 30)
+	//  - set level0_slowdown_writes_trigger = max(old, 1 << 30)
+	//  - set soft_pending_compaction_bytes_limit = 0,
+	//  - set hard_pending_compaction_bytes_limit = 0,
+	//  - will not trigger flow control when SST count in L0 is large
+	//  - will not trigger region split, it might cause some region became
+	//    very large and be a hotspot, might cause latency spike.
+	//
+	// default false for local sort, true for global sort.
 	disableTiKVImportModeOption = "disable_tikv_import_mode"
 	cloudStorageURIOption       = "cloud_storage_uri"
 	disablePrecheckOption       = "disable_precheck"
@@ -780,6 +790,9 @@ func (p *Plan) adjustOptions(targetNodeCPUCnt int) {
 			zap.Int("before", p.ThreadCnt), zap.Int("after", limit))
 		p.ThreadCnt = limit
 	}
+	if p.IsGlobalSort() {
+		p.DisableTiKVImportMode = true
+	}
 }
 
 func (p *Plan) initParameters(plan *plannercore.ImportInto) error {


Test patch:
diff --git a/pkg/executor/importer/import_test.go b/pkg/executor/importer/import_test.go
index 9ebd90a276..014a001c4d 100644
--- a/pkg/executor/importer/import_test.go
+++ b/pkg/executor/importer/import_test.go
@@ -189,11 +189,17 @@ func TestAdjustOptions(t *testing.T) {
 	plan.adjustOptions(16)
 	require.Equal(t, 16, plan.ThreadCnt)
 	require.Equal(t, config.ByteSize(10), plan.MaxWriteSpeed) // not adjusted
+	require.False(t, plan.DisableTiKVImportMode)
 
 	plan.ThreadCnt = 100000000
 	plan.DataSourceType = DataSourceTypeQuery
 	plan.adjustOptions(16)
 	require.Equal(t, 32, plan.ThreadCnt)
+	require.False(t, plan.DisableTiKVImportMode)
+
+	plan.CloudStorageURI = "s3://bucket/path"
+	plan.adjustOptions(16)
+	require.True(t, plan.DisableTiKVImportMode)
 }
 
 func TestAdjustDiskQuota(t *testing.T) {
diff --git a/tests/realtikvtest/importintotest3/from_server_test.go b/tests/realtikvtest/importintotest3/from_server_test.go
index d41e44afe2..544446e9d7 100644
--- a/tests/realtikvtest/importintotest3/from_server_test.go
+++ b/tests/realtikvtest/importintotest3/from_server_test.go
@@ -73,4 +73,5 @@ func (s *mockGCSSuite) TestImportFromServer() {
 	var taskMeta importinto.TaskMeta
 	require.NoError(s.T(), json.Unmarshal(task.Meta, &taskMeta))
 	require.Len(s.T(), taskMeta.ChunkMap, 2)
+	require.False(s.T(), taskMeta.Plan.DisableTiKVImportMode)
 }
diff --git a/tests/realtikvtest/importintotest4/global_sort_test.go b/tests/realtikvtest/importintotest4/global_sort_test.go
index 7f9a8c6982..f6b03c8cab 100644
--- a/tests/realtikvtest/importintotest4/global_sort_test.go
+++ b/tests/realtikvtest/importintotest4/global_sort_test.go
@@ -96,6 +96,7 @@ func (s *mockGCSSuite) TestGlobalSortBasic() {
 	taskMeta := importinto.TaskMeta{}
 	s.NoError(json.Unmarshal(task.Meta, &taskMeta))
 	urlEqual(s.T(), redactedSortStorageURI, taskMeta.Plan.CloudStorageURI)
+	require.True(s.T(), taskMeta.Plan.DisableTiKVImportMode)
 
 	// merge-sort data kv
 	s.tk.MustExec("truncate table t")