Repository: pingcap/tidb
Pull request #51006 (2024-02-06 08:06:41)

external: use 2 membuf pools of difference block size
<!--

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: ref #50752

Problem Summary:

### What changed and how does it work?

use 2 membuf.Pool that shares one membuf.Limiter.
- smallBlockBufPool: block size 1M, used to store the KV pairs that will be sent to TiKV. KV size larger than 1M can be allocated but will not be tracked by membuf.Limiter.
- largeBlockBufPool: block size 8M, used by concurrent reader

Also, in readAllData spawn constant number of worker, each with a smallBlockBuf and largeBlockBuf. So the memory usage is no longer relevant to number of files to be read.

### Check List

Tests <!-- At least one of them must be included. -->

- [x] Unit test
- [ ] Integration test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test
  > - [ ] I checked and no code files have been changed.
  > <!-- Or your custom  "No need to test" reasons -->

run `TestReadAllData`, before this PR

```
    bench_test.go:821: readAllData time cost: 1m41.626412352s, size: 2939207462
```

and peak memory usage is 8.29G

after

```
    bench_test.go:829: readAllData time cost: 39.031899936s, size: 2939207462
```

and peak memory usage is 5.11G

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
```


ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [lance6716](https://github.com/lance6716). For more information see [the Kubernetes Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=pingcap%2Ftidb).

<details open>
Needs approval from an approver in each of these files:

- **[br/pkg/lightning/OWNERS](https://github.com/pingcap/tidb/blob/master/br/pkg/lightning/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["lance6716"]} -->
tiprow[bot]: Hi @lance6716. Thanks for your PR.

PRs from untrusted users cannot be marked as trusted with `/ok-to-test` in this repo meaning untrusted PR authors can never trigger tests themselves. Collaborators can still trigger tests on the PR using `/test all`.

I understand the commands that are listed [here](https://go.k8s.io/bot-commands?repo=pingcap%2Ftidb).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pingcap/tidb/pull/51006?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
> Merging [#51006](https://app.codecov.io/gh/pingcap/tidb/pull/51006?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (66614fc) into [master](https://app.codecov.io/gh/pingcap/tidb/commit/bb43e2db3d9ecb94e8b1c8a560c7763f7abcd977?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (bb43e2d) will **increase** coverage by `0.7992%`.
> Report is 1 commits behind head on master.
> The diff coverage is `67.2131%`.

<details><summary>Additional details and impacted files</summary>


```diff
@@               Coverage Diff                @@
##             master     #51006        +/-   ##
================================================
+ Coverage   70.5506%   71.3499%   +0.7992%     
================================================
  Files          1467       1467                
  Lines        434825     435343       +518     
================================================
+ Hits         306772     310617      +3845     
+ Misses       108763     105151      -3612     
- Partials      19290      19575       +285     
```

| [Flag](https://app.codecov.io/gh/pingcap/tidb/pull/51006/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/pingcap/tidb/pull/51006/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `23.1126% <67.2131%> (?)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Components](https://app.codecov.io/gh/pingcap/tidb/pull/51006/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [dumpling](https://app.codecov.io/gh/pingcap/tidb/pull/51006/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `54.2940% <ø> (ø)` | |
| [parser](https://app.codecov.io/gh/pingcap/tidb/pull/51006/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `∅ <ø> (∅)` | |
| [br](https://app.codecov.io/gh/pingcap/tidb/pull/51006/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `47.2055% <67.2131%> (+1.4697%)` | :arrow_up: |

</details>
ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: To remove the `do-not-merge/needs-linked-issue` label, please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>

<hr>

**Notice**: To remove the `do-not-merge/needs-tests-checked` label, please finished the tests then check the finished items in description.

For example:

> Tests <!-- At least one of them must be included. -->
> 
> - [x] Unit test
> - [ ] Integration test
> - [ ] Manual test (add detailed scripts or steps below)
> - [ ] No code

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#writing-tests) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: To remove the `do-not-merge/needs-linked-issue` label, please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: To remove the `do-not-merge/needs-linked-issue` label, please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: To remove the `do-not-merge/needs-linked-issue` label, please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->


Patch:
diff --git a/br/pkg/lightning/backend/external/byte_reader.go b/br/pkg/lightning/backend/external/byte_reader.go
index de93012060..53df918b01 100644
--- a/br/pkg/lightning/backend/external/byte_reader.go
+++ b/br/pkg/lightning/backend/external/byte_reader.go
@@ -31,8 +31,10 @@ import (
 var (
 	// ConcurrentReaderBufferSizePerConc is the buffer size for concurrent reader per
 	// concurrency.
-	ConcurrentReaderBufferSizePerConc = int(2 * size.MB)
-	readAllDataConcThreshold          = uint64(16)
+	ConcurrentReaderBufferSizePerConc = int(8 * size.MB)
+	// in readAllData, expected concurrency less than this value will not use
+	// concurrent reader.
+	readAllDataConcThreshold = uint64(4)
 )
 
 // byteReader provides structured reading on a byte stream of external storage.
diff --git a/br/pkg/lightning/backend/external/engine.go b/br/pkg/lightning/backend/external/engine.go
index cafe9cecff..98fdc27dd1 100644
--- a/br/pkg/lightning/backend/external/engine.go
+++ b/br/pkg/lightning/backend/external/engine.go
@@ -50,9 +50,11 @@ import (
 const maxCloudStorageConnections = 1000
 
 type memKVsAndBuffers struct {
-	mu           sync.Mutex
-	keys         [][]byte
-	values       [][]byte
+	mu     sync.Mutex
+	keys   [][]byte
+	values [][]byte
+	// memKVBuffers contains two types of buffer, first half are used for small block
+	// buffer, second half are used for large one.
 	memKVBuffers []*membuf.Buffer
 	size         int
 	droppedSize  int
@@ -92,14 +94,15 @@ func (b *memKVsAndBuffers) build(ctx context.Context) {
 
 // Engine stored sorted key/value pairs in an external storage.
 type Engine struct {
-	storage         storage.ExternalStorage
-	dataFiles       []string
-	statsFiles      []string
-	startKey        []byte
-	endKey          []byte
-	splitKeys       [][]byte
-	regionSplitSize int64
-	bufPool         *membuf.Pool
+	storage           storage.ExternalStorage
+	dataFiles         []string
+	statsFiles        []string
+	startKey          []byte
+	endKey            []byte
+	splitKeys         [][]byte
+	regionSplitSize   int64
+	smallBlockBufPool *membuf.Pool
+	largeBlockBufPool *membuf.Pool
 
 	memKVsAndBuffers memKVsAndBuffers
 
@@ -126,7 +129,10 @@ type Engine struct {
 	importedKVCount *atomic.Int64
 }
 
-const memLimit = 12 * units.GiB
+const (
+	memLimit       = 12 * units.GiB
+	smallBlockSize = units.MiB
+)
 
 // NewExternalEngine creates an (external) engine.
 func NewExternalEngine(
@@ -156,7 +162,12 @@ func NewExternalEngine(
 		endKey:          endKey,
 		splitKeys:       splitKeys,
 		regionSplitSize: regionSplitSize,
-		bufPool: membuf.NewPool(
+		smallBlockBufPool: membuf.NewPool(
+			membuf.WithBlockNum(0),
+			membuf.WithPoolMemoryLimiter(memLimiter),
+			membuf.WithBlockSize(smallBlockSize),
+		),
+		largeBlockBufPool: membuf.NewPool(
 			membuf.WithBlockNum(0),
 			membuf.WithPoolMemoryLimiter(memLimiter),
 			membuf.WithBlockSize(ConcurrentReaderBufferSizePerConc),
@@ -222,16 +233,28 @@ func getFilesReadConcurrency(
 	}
 	startOffs, endOffs := offsets[0], offsets[1]
 	for i := range statsFiles {
-		result[i] = (endOffs[i] - startOffs[i]) / uint64(ConcurrentReaderBufferSizePerConc)
+		expectedConc := (endOffs[i] - startOffs[i]) / uint64(ConcurrentReaderBufferSizePerConc)
 		// let the stat internals cover the [startKey, endKey) since seekPropsOffsets
 		// always return an offset that is less than or equal to the key.
-		result[i] += 1
-		logutil.Logger(ctx).Info("found hotspot file in getFilesReadConcurrency",
-			zap.String("filename", statsFiles[i]),
-			zap.Uint64("startOffset", startOffs[i]),
-			zap.Uint64("endOffset", endOffs[i]),
-			zap.Uint64("expected concurrency", result[i]),
-		)
+		expectedConc += 1
+		// readAllData will enable concurrent read and use large buffer if result[i] > 1
+		// when expectedConc < readAllDataConcThreshold, we don't use concurrent read to
+		// reduce overhead
+		if expectedConc >= readAllDataConcThreshold {
+			result[i] = expectedConc
+		} else {
+			result[i] = 1
+		}
+		// only log for files with expected concurrency > 1, to avoid too many logs
+		if expectedConc > 1 {
+			logutil.Logger(ctx).Info("found hotspot file in getFilesReadConcurrency",
+				zap.String("filename", statsFiles[i]),
+				zap.Uint64("startOffset", startOffs[i]),
+				zap.Uint64("endOffset", endOffs[i]),
+				zap.Uint64("expectedConc", expectedConc),
+				zap.Uint64("concurrency", result[i]),
+			)
+		}
 	}
 	return result, startOffs, nil
 }
@@ -252,7 +275,8 @@ func (e *Engine) loadBatchRegionData(ctx context.Context, startKey, endKey []byt
 		e.statsFiles,
 		startKey,
 		endKey,
-		e.bufPool,
+		e.smallBlockBufPool,
+		e.largeBlockBufPool,
 		&e.memKVsAndBuffers,
 	)
 	if err != nil {
@@ -443,9 +467,13 @@ func (e *Engine) SplitRanges(
 
 // Close implements common.Engine.
 func (e *Engine) Close() error {
-	if e.bufPool != nil {
-		e.bufPool.Destroy()
-		e.bufPool = nil
+	if e.smallBlockBufPool != nil {
+		e.smallBlockBufPool.Destroy()
+		e.smallBlockBufPool = nil
+	}
+	if e.largeBlockBufPool != nil {
+		e.largeBlockBufPool.Destroy()
+		e.largeBlockBufPool = nil
 	}
 	e.storage.Close()
 	return nil
@@ -453,11 +481,21 @@ func (e *Engine) Close() error {
 
 // Reset resets the memory buffer pool.
 func (e *Engine) Reset() error {
-	if e.bufPool != nil {
-		e.bufPool.Destroy()
-		memLimiter := membuf.NewLimiter(memLimit)
-		e.bufPool = membuf.NewPool(
+	memLimiter := membuf.NewLimiter(memLimit)
+	if e.smallBlockBufPool != nil {
+		e.smallBlockBufPool.Destroy()
+		e.smallBlockBufPool = membuf.NewPool(
+			membuf.WithBlockNum(0),
 			membuf.WithPoolMemoryLimiter(memLimiter),
+			membuf.WithBlockSize(smallBlockSize),
+		)
+	}
+	if e.largeBlockBufPool != nil {
+		e.largeBlockBufPool.Destroy()
+		e.largeBlockBufPool = membuf.NewPool(
+			membuf.WithBlockNum(0),
+			membuf.WithPoolMemoryLimiter(memLimiter),
+			membuf.WithBlockSize(ConcurrentReaderBufferSizePerConc),
 		)
 	}
 	return nil
diff --git a/br/pkg/lightning/backend/external/merge_v2.go b/br/pkg/lightning/backend/external/merge_v2.go
index 4a5dea8e12..5f5a6833bb 100644
--- a/br/pkg/lightning/backend/external/merge_v2.go
+++ b/br/pkg/lightning/backend/external/merge_v2.go
@@ -119,6 +119,7 @@ func MergeOverlappingFilesV2(
 			curStart,
 			curEnd,
 			bufPool,
+			bufPool,
 			loaded,
 		)
 		if err1 != nil {
diff --git a/br/pkg/lightning/backend/external/reader.go b/br/pkg/lightning/backend/external/reader.go
index 151d20736c..27a79cee02 100644
--- a/br/pkg/lightning/backend/external/reader.go
+++ b/br/pkg/lightning/backend/external/reader.go
@@ -35,7 +35,8 @@ func readAllData(
 	store storage.ExternalStorage,
 	dataFiles, statsFiles []string,
 	startKey, endKey []byte,
-	bufPool *membuf.Pool,
+	smallBlockBufPool *membuf.Pool,
+	largeBlockBufPool *membuf.Pool,
 	output *memKVsAndBuffers,
 ) (err error) {
 	task := log.BeginTask(logutil.Logger(ctx), "read all data")
@@ -69,36 +70,59 @@ func readAllData(
 		startKey,
 		endKey,
 	)
-	// TODO(lance6716): refine adjust concurrency
-	for i, c := range concurrences {
-		if c < readAllDataConcThreshold {
-			concurrences[i] = 1
-		}
-	}
-
 	if err != nil {
 		return err
 	}
+
 	eg, egCtx := util.NewErrorGroupWithRecoverWithCtx(ctx)
-	// limit the concurrency to avoid open too many connections at the same time
-	eg.SetLimit(1000)
-	for i := range dataFiles {
-		i := i
+	readConn := 1000
+	readConn = min(readConn, len(dataFiles))
+	taskCh := make(chan int)
+	output.memKVBuffers = make([]*membuf.Buffer, readConn*2)
+	for readIdx := 0; readIdx < readConn; readIdx++ {
+		readIdx := readIdx
 		eg.Go(func() error {
-			err2 := readOneFile(
-				egCtx,
-				store,
-				dataFiles[i],
-				startKey,
-				endKey,
-				startOffsets[i],
-				concurrences[i],
-				bufPool,
-				output,
-			)
-			return errors.Annotatef(err2, "failed to read file %s", dataFiles[i])
+			output.memKVBuffers[readIdx] = smallBlockBufPool.NewBuffer()
+			output.memKVBuffers[readIdx+readConn] = largeBlockBufPool.NewBuffer()
+			smallBlockBuf := output.memKVBuffers[readIdx]
+			largeBlockBuf := output.memKVBuffers[readIdx+readConn]
+
+			for {
+				select {
+				case <-egCtx.Done():
+					return egCtx.Err()
+				case fileIdx, ok := <-taskCh:
+					if !ok {
+						return nil
+					}
+					err2 := readOneFile(
+						egCtx,
+						store,
+						dataFiles[fileIdx],
+						startKey,
+						endKey,
+						startOffsets[fileIdx],
+						concurrences[fileIdx],
+						smallBlockBuf,
+						largeBlockBuf,
+						output,
+					)
+					if err2 != nil {
+						return errors.Annotatef(err2, "failed to read file %s", dataFiles[fileIdx])
+					}
+				}
+			}
 		})
 	}
+
+	for fileIdx := range dataFiles {
+		select {
+		case <-egCtx.Done():
+			return eg.Wait()
+		case taskCh <- fileIdx:
+		}
+	}
+	close(taskCh)
 	return eg.Wait()
 }
 
@@ -109,7 +133,8 @@ func readOneFile(
 	startKey, endKey []byte,
 	startOffset uint64,
 	concurrency uint64,
-	bufPool *membuf.Pool,
+	smallBlockBuf *membuf.Buffer,
+	largeBlockBuf *membuf.Buffer,
 	output *memKVsAndBuffers,
 ) error {
 	readAndSortDurHist := metrics.GlobalSortReadFromCloudStorageDuration.WithLabelValues("read_one_file")
@@ -127,7 +152,7 @@ func readOneFile(
 			dataFile,
 			int(concurrency),
 			ConcurrentReaderBufferSizePerConc,
-			bufPool.NewBuffer(),
+			largeBlockBuf,
 		)
 		err = rd.byteReader.switchConcurrentMode(true)
 		if err != nil {
@@ -135,8 +160,6 @@ func readOneFile(
 		}
 	}
 
-	// this buffer is associated with data slices and will return to caller
-	memBuf := bufPool.NewBuffer()
 	keys := make([][]byte, 0, 1024)
 	values := make([][]byte, 0, 1024)
 	size := 0
@@ -159,15 +182,14 @@ func readOneFile(
 		}
 		// TODO(lance6716): we are copying every KV from rd's buffer to memBuf, can we
 		// directly read into memBuf?
-		keys = append(keys, memBuf.AddBytes(k))
-		values = append(values, memBuf.AddBytes(v))
+		keys = append(keys, smallBlockBuf.AddBytes(k))
+		values = append(values, smallBlockBuf.AddBytes(v))
 		size += len(k) + len(v)
 	}
 	readAndSortDurHist.Observe(time.Since(ts).Seconds())
 	output.mu.Lock()
 	output.keysPerFile = append(output.keysPerFile, keys)
 	output.valuesPerFile = append(output.valuesPerFile, values)
-	output.memKVBuffers = append(output.memKVBuffers, memBuf)
 	output.size += size
 	output.droppedSizePerFile = append(output.droppedSizePerFile, droppedSize)
 	output.mu.Unlock()
diff --git a/pkg/metrics/grafana/tidb_summary.json b/pkg/metrics/grafana/tidb_summary.json
index aa2611e772..daf2f0f00c 100644
--- a/pkg/metrics/grafana/tidb_summary.json
+++ b/pkg/metrics/grafana/tidb_summary.json
@@ -466,7 +466,7 @@
                      "refId": "B"
                   },
                   {
-                     "expr": "sum(rate(tidb_server_handle_query_duration_seconds_sum{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\", sql_type!=\"internal\"}[30s])) / sum(rate(tidb_server_handle_query_duration_seconds_count{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", sql_type!=\"internal\"}[30s]))",
+                     "expr": "sum(rate(tidb_server_handle_query_duration_seconds_sum{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\", sql_type!=\"internal\"}[30s])) / sum(rate(tidb_server_handle_query_duration_seconds_count{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\", sql_type!=\"internal\"}[30s]))",
                      "format": "time_series",
                      "intervalFactor": 2,
                      "legendFormat": "avg",
@@ -653,7 +653,7 @@
                      "refId": "B"
                   },
                   {
-                     "expr": "sum(tidb_server_connections{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\"}) * sum(rate(tidb_server_handle_query_duration_seconds_count{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\"}[1m])) / sum(rate(tidb_server_handle_query_duration_seconds_sum{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\"}[1m]))",
+                     "expr": "sum(tidb_server_connections{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\"}) * sum(rate(tidb_server_handle_query_duration_seconds_count{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\"}[1m])) / sum(rate(tidb_server_handle_query_duration_seconds_sum{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", instance=~\"$instance\"}[1m]))",
                      "format": "time_series",
                      "hide": true,
                      "intervalFactor": 2,
diff --git a/pkg/metrics/grafana/tidb_summary.jsonnet b/pkg/metrics/grafana/tidb_summary.jsonnet
index e11d6587e0..75b555dfd7 100644
--- a/pkg/metrics/grafana/tidb_summary.jsonnet
+++ b/pkg/metrics/grafana/tidb_summary.jsonnet
@@ -178,7 +178,7 @@ local durationP = graphPanel.new(
 )
 .addTarget(
   prometheus.target(
-    'sum(rate(tidb_server_handle_query_duration_seconds_sum{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$instance", sql_type!="internal"}[30s])) / sum(rate(tidb_server_handle_query_duration_seconds_count{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", sql_type!="internal"}[30s]))',
+    'sum(rate(tidb_server_handle_query_duration_seconds_sum{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$instance", sql_type!="internal"}[30s])) / sum(rate(tidb_server_handle_query_duration_seconds_count{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$instance", sql_type!="internal"}[30s]))',
     legendFormat='avg',
   )
 );
@@ -219,7 +219,7 @@ local cpsP = graphPanel.new(
 )
 .addTarget(
   prometheus.target(
-    'sum(tidb_server_connections{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$instance"}) * sum(rate(tidb_server_handle_query_duration_seconds_count{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster"}[1m])) / sum(rate(tidb_server_handle_query_duration_seconds_sum{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster"}[1m]))',
+    'sum(tidb_server_connections{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$instance"}) * sum(rate(tidb_server_handle_query_duration_seconds_count{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$instance"}[1m])) / sum(rate(tidb_server_handle_query_duration_seconds_sum{k8s_cluster="$k8s_cluster", tidb_cluster="$tidb_cluster", instance=~"$instance"}[1m]))',
     legendFormat='ideal CPS',
     hide=true,
   )
diff --git a/pkg/planner/cardinality/BUILD.bazel b/pkg/planner/cardinality/BUILD.bazel
index edf31836d1..5cc11520d3 100644
--- a/pkg/planner/cardinality/BUILD.bazel
+++ b/pkg/planner/cardinality/BUILD.bazel
@@ -59,7 +59,7 @@ go_test(
     data = glob(["testdata/**"]),
     embed = [":cardinality"],
     flaky = True,
-    shard_count = 26,
+    shard_count = 27,
     deps = [
         "//pkg/config",
         "//pkg/domain",
diff --git a/pkg/planner/cardinality/row_count_index.go b/pkg/planner/cardinality/row_count_index.go
index 467b73f2fa..4fa07a72a1 100644
--- a/pkg/planner/cardinality/row_count_index.go
+++ b/pkg/planner/cardinality/row_count_index.go
@@ -478,14 +478,33 @@ func expBackoffEstimation(sctx sessionctx.Context, idx *statistics.Index, coll *
 	})
 	if l == 1 {
 		return singleColumnEstResults[0], true, nil
-	} else if l == 2 {
-		return singleColumnEstResults[0] * math.Sqrt(singleColumnEstResults[1]), true, nil
-	} else if l == 3 {
-		return singleColumnEstResults[0] * math.Sqrt(singleColumnEstResults[1]) * math.Sqrt(math.Sqrt(singleColumnEstResults[2])), true, nil
 	} else if l == 0 {
 		return 0, false, nil
 	}
-	return singleColumnEstResults[0] * math.Sqrt(singleColumnEstResults[1]) * math.Sqrt(math.Sqrt(singleColumnEstResults[2])) * math.Sqrt(math.Sqrt(math.Sqrt(singleColumnEstResults[3]))), true, nil
+	// Do not allow the exponential backoff to go below the available index bound. If the number of predicates
+	// is less than the number of index columns - use 90% of the bound to differentiate a subset from full index match.
+	// If there is an individual column selectivity that goes below this bound, use that selectivity only.
+	histNDV := coll.RealtimeCount
+	if idx.NDV > 0 {
+		histNDV = idx.NDV
+	}
+	idxLowBound := 1 / float64(min(histNDV, coll.RealtimeCount))
+	if l < len(idx.Info.Columns) {
+		idxLowBound /= 0.9
+	}
+	minTwoCol := min(singleColumnEstResults[0], singleColumnEstResults[1], idxLowBound)
+	multTwoCol := singleColumnEstResults[0] * math.Sqrt(singleColumnEstResults[1])
+	if l == 2 {
+		return max(minTwoCol, multTwoCol), true, nil
+	}
+	minThreeCol := min(minTwoCol, singleColumnEstResults[2])
+	multThreeCol := multTwoCol * math.Sqrt(math.Sqrt(singleColumnEstResults[2]))
+	if l == 3 {
+		return max(minThreeCol, multThreeCol), true, nil
+	}
+	minFourCol := min(minThreeCol, singleColumnEstResults[3])
+	multFourCol := multThreeCol * math.Sqrt(math.Sqrt(math.Sqrt(singleColumnEstResults[3])))
+	return max(minFourCol, multFourCol), true, nil
 }
 
 // outOfRangeOnIndex checks if the datum is out of the range.


Test patch:
diff --git a/br/pkg/lightning/backend/external/bench_test.go b/br/pkg/lightning/backend/external/bench_test.go
index 66144569ef..82910eb72b 100644
--- a/br/pkg/lightning/backend/external/bench_test.go
+++ b/br/pkg/lightning/backend/external/bench_test.go
@@ -20,19 +20,16 @@ import (
 	"flag"
 	"fmt"
 	"io"
-	"os"
-	"runtime/pprof"
-	"sync"
 	"testing"
 	"time"
 
 	"github.com/docker/go-units"
-	"github.com/felixge/fgprof"
 	"github.com/pingcap/tidb/br/pkg/membuf"
 	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/util/intest"
 	"github.com/pingcap/tidb/pkg/util/size"
+	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
 	"golang.org/x/sync/errgroup"
 )
@@ -171,34 +168,20 @@ func TestCompareWriter(t *testing.T) {
 	seed := time.Now().Nanosecond()
 	t.Logf("random seed: %d", seed)
 	var (
-		err     error
 		now     time.Time
 		elapsed time.Duration
 
-		fileCPU       *os.File
-		cpuProfCloser func() error
-
-		filenameHeap   string
-		heapProfDoneCh chan struct{}
-		heapWg         *sync.WaitGroup
+		p = newProfiler(true, true)
 	)
 	beforeTest := func() {
 		testIdx++
-		fileCPU, err = os.Create(fmt.Sprintf("cpu-profile-%d.prof", testIdx))
-		intest.AssertNoError(err)
-		cpuProfCloser = fgprof.Start(fileCPU, fgprof.FormatPprof)
-
-		filenameHeap = fmt.Sprintf("heap-profile-%d.prof", testIdx)
-		heapProfDoneCh, heapWg = recordHeapForMaxInUse(filenameHeap)
+		p.beforeTest()
 
 		now = time.Now()
 	}
 	afterClose := func() {
 		elapsed = time.Since(now)
-		err = cpuProfCloser()
-		intest.AssertNoError(err)
-		close(heapProfDoneCh)
-		heapWg.Wait()
+		p.afterTest()
 	}
 
 	suite := &writeTestSuite{
@@ -392,46 +375,20 @@ func TestCompareReaderEvenlyDistributedContent(t *testing.T) {
 
 	kvCnt, _, _ := createEvenlyDistributedFiles(store, fileSize, fileCnt, subDir)
 	memoryLimit := 64 * 1024 * 1024
-	fileIdx := 0
 
 	var (
-		err     error
-		now     time.Time
 		elapsed time.Duration
 
-		fileCPU       *os.File
-		cpuProfCloser func() error
-
-		filenameHeap   string
-		heapProfDoneCh chan struct{}
-		heapWg         *sync.WaitGroup
+		p = newProfiler(true, true)
 	)
-	beforeTest := func() {
-		fileIdx++
-		fileCPU, err = os.Create(fmt.Sprintf("cpu-profile-%d.prof", fileIdx))
-		intest.AssertNoError(err)
-		cpuProfCloser = fgprof.Start(fileCPU, fgprof.FormatPprof)
-
-		filenameHeap = fmt.Sprintf("heap-profile-%d.prof", fileIdx)
-		heapProfDoneCh, heapWg = recordHeapForMaxInUse(filenameHeap)
-
-		now = time.Now()
-	}
-	afterClose := func() {
-		elapsed = time.Since(now)
-		err = cpuProfCloser()
-		intest.AssertNoError(err)
-		close(heapProfDoneCh)
-		heapWg.Wait()
-	}
 
 	suite := &readTestSuite{
 		store:              store,
 		totalKVCnt:         kvCnt,
 		concurrency:        100,
 		memoryLimit:        memoryLimit,
-		beforeCreateReader: beforeTest,
-		afterReaderClose:   afterClose,
+		beforeCreateReader: p.beforeTest,
+		afterReaderClose:   p.afterTest,
 		subDir:             subDir,
 	}
 
@@ -495,41 +452,15 @@ func testCompareReaderWithContent(
 	if !*skipCreate {
 		kvCnt, _, _ = createFn(store, *fileSize, *fileCount, *objectPrefix)
 	}
-	fileIdx := 0
-
-	var (
-		err error
-
-		fileCPU       *os.File
-		cpuProfCloser func() error
-
-		filenameHeap   string
-		heapProfDoneCh chan struct{}
-		heapWg         *sync.WaitGroup
-	)
-	beforeTest := func() {
-		fileIdx++
-		fileCPU, err = os.Create(fmt.Sprintf("cpu-profile-%d.prof", fileIdx))
-		intest.AssertNoError(err)
-		cpuProfCloser = fgprof.Start(fileCPU, fgprof.FormatPprof)
-
-		filenameHeap = fmt.Sprintf("heap-profile-%d.prof", fileIdx)
-		heapProfDoneCh, heapWg = recordHeapForMaxInUse(filenameHeap)
-	}
-	afterClose := func() {
-		err = cpuProfCloser()
-		intest.AssertNoError(err)
-		close(heapProfDoneCh)
-		heapWg.Wait()
-	}
+	p := newProfiler(true, true)
 
 	suite := &readTestSuite{
 		store:              store,
 		totalKVCnt:         kvCnt,
 		concurrency:        *concurrency,
 		memoryLimit:        *memoryLimit,
-		beforeCreateReader: beforeTest,
-		afterReaderClose:   afterClose,
+		beforeCreateReader: p.beforeTest,
+		afterReaderClose:   p.afterTest,
 		subDir:             *objectPrefix,
 	}
 
@@ -645,7 +576,9 @@ func testCompareMergeWithContent(
 	t *testing.T,
 	concurrency int,
 	createFn func(store storage.ExternalStorage, fileSize int, fileCount int, objectPrefix string) (int, kv.Key, kv.Key),
-	fn func(t *testing.T, suite *mergeTestSuite)) {
+	fn func(t *testing.T, suite *mergeTestSuite),
+	p *profiler,
+) {
 	store := openTestingStorage(t)
 	kvCnt := 0
 	var minKey, maxKey kv.Key
@@ -653,29 +586,13 @@ func testCompareMergeWithContent(
 		kvCnt, minKey, maxKey = createFn(store, *fileSize, *fileCount, *objectPrefix)
 	}
 
-	fileIdx := 0
-	var (
-		file *os.File
-		err  error
-	)
-	beforeTest := func() {
-		file, err = os.Create(fmt.Sprintf("cpu-profile-%d.prof", fileIdx))
-		intest.AssertNoError(err)
-		err = pprof.StartCPUProfile(file)
-		intest.AssertNoError(err)
-	}
-
-	afterTest := func() {
-		pprof.StopCPUProfile()
-	}
-
 	suite := &mergeTestSuite{
 		store:            store,
 		totalKVCnt:       kvCnt,
 		concurrency:      concurrency,
 		memoryLimit:      *memoryLimit,
-		beforeMerge:      beforeTest,
-		afterMerge:       afterTest,
+		beforeMerge:      p.beforeTest,
+		afterMerge:       p.afterTest,
 		subDir:           *objectPrefix,
 		minKey:           minKey,
 		maxKey:           maxKey,
@@ -686,16 +603,17 @@ func testCompareMergeWithContent(
 }
 
 func TestMergeBench(t *testing.T) {
-	testCompareMergeWithContent(t, 1, createAscendingFiles, mergeStep)
-	testCompareMergeWithContent(t, 1, createEvenlyDistributedFiles, mergeStep)
-	testCompareMergeWithContent(t, 2, createAscendingFiles, mergeStep)
-	testCompareMergeWithContent(t, 2, createEvenlyDistributedFiles, mergeStep)
-	testCompareMergeWithContent(t, 4, createAscendingFiles, mergeStep)
-	testCompareMergeWithContent(t, 4, createEvenlyDistributedFiles, mergeStep)
-	testCompareMergeWithContent(t, 8, createAscendingFiles, mergeStep)
-	testCompareMergeWithContent(t, 8, createEvenlyDistributedFiles, mergeStep)
-	testCompareMergeWithContent(t, 8, createAscendingFiles, newMergeStep)
-	testCompareMergeWithContent(t, 8, createEvenlyDistributedFiles, newMergeStep)
+	p := newProfiler(true, true)
+	testCompareMergeWithContent(t, 1, createAscendingFiles, mergeStep, p)
+	testCompareMergeWithContent(t, 1, createEvenlyDistributedFiles, mergeStep, p)
+	testCompareMergeWithContent(t, 2, createAscendingFiles, mergeStep, p)
+	testCompareMergeWithContent(t, 2, createEvenlyDistributedFiles, mergeStep, p)
+	testCompareMergeWithContent(t, 4, createAscendingFiles, mergeStep, p)
+	testCompareMergeWithContent(t, 4, createEvenlyDistributedFiles, mergeStep, p)
+	testCompareMergeWithContent(t, 8, createAscendingFiles, mergeStep, p)
+	testCompareMergeWithContent(t, 8, createEvenlyDistributedFiles, mergeStep, p)
+	testCompareMergeWithContent(t, 8, createAscendingFiles, newMergeStep, p)
+	testCompareMergeWithContent(t, 8, createEvenlyDistributedFiles, newMergeStep, p)
 }
 
 func TestReadAllDataLargeFiles(t *testing.T) {
@@ -739,13 +657,174 @@ func TestReadAllDataLargeFiles(t *testing.T) {
 	intest.AssertNoError(err)
 	endKey, err := hex.DecodeString("00a00000000000000000")
 	intest.AssertNoError(err)
-	bufPool := membuf.NewPool(
+	smallBlockBufPool := membuf.NewPool(
+		membuf.WithBlockNum(0),
+		membuf.WithBlockSize(smallBlockSize),
+	)
+	largeBlockBufPool := membuf.NewPool(
 		membuf.WithBlockNum(0),
 		membuf.WithBlockSize(ConcurrentReaderBufferSizePerConc),
 	)
 	output := &memKVsAndBuffers{}
 	now := time.Now()
-	err = readAllData(ctx, store, dataFiles, statFiles, startKey, endKey, bufPool, output)
+
+	err = readAllData(ctx, store, dataFiles, statFiles, startKey, endKey, smallBlockBufPool, largeBlockBufPool, output)
 	t.Logf("read all data cost: %s", time.Since(now))
 	intest.AssertNoError(err)
 }
+
+func TestReadAllData(t *testing.T) {
+	// test the case that thread=16, where we will load ~3.2GB data once and this
+	// step will at most have 4000 files to read, test the case that we have
+	//
+	// 1000 files read one KV (~100B), 1000 files read ~900KB, 90 files read 10MB,
+	// 1 file read 1G. total read size = 1000*100B + 1000*900KB + 90*10MB + 1*1G = 2.8G
+
+	ctx := context.Background()
+	store := openTestingStorage(t)
+	readRangeStart := []byte("key00")
+	readRangeEnd := []byte("key88888888")
+	keyAfterRange := []byte("key9")
+	keyAfterRange2 := []byte("key9")
+	eg := errgroup.Group{}
+
+	fileIdx := 0
+	val := make([]byte, 90)
+	if *skipCreate {
+		goto finishCreateFiles
+	}
+
+	cleanOldFiles(ctx, store, "/")
+
+	for ; fileIdx < 1000; fileIdx++ {
+		fileIdx := fileIdx
+		eg.Go(func() error {
+			fileName := fmt.Sprintf("/test%d", fileIdx)
+			writer := NewWriterBuilder().BuildOneFile(store, fileName, "writerID")
+			err := writer.Init(ctx, 5*1024*1024)
+			require.NoError(t, err)
+			key := []byte(fmt.Sprintf("key0%d", fileIdx))
+			err = writer.WriteRow(ctx, key, val)
+			require.NoError(t, err)
+
+			// write some extra data that is greater than readRangeEnd
+			err = writer.WriteRow(ctx, keyAfterRange, val)
+			require.NoError(t, err)
+			err = writer.WriteRow(ctx, keyAfterRange2, make([]byte, 100*1024))
+			require.NoError(t, err)
+
+			return writer.Close(ctx)
+		})
+	}
+	require.NoError(t, eg.Wait())
+	t.Log("finish writing 1000 files of 100B")
+
+	for ; fileIdx < 2000; fileIdx++ {
+		fileIdx := fileIdx
+		eg.Go(func() error {
+			fileName := fmt.Sprintf("/test%d", fileIdx)
+			writer := NewWriterBuilder().BuildOneFile(store, fileName, "writerID")
+			err := writer.Init(ctx, 5*1024*1024)
+			require.NoError(t, err)
+
+			kvSize := 0
+			keyIdx := 0
+			for kvSize < 900*1024 {
+				key := []byte(fmt.Sprintf("key%06d_%d", keyIdx, fileIdx))
+				keyIdx++
+				kvSize += len(key) + len(val)
+				err = writer.WriteRow(ctx, key, val)
+				require.NoError(t, err)
+			}
+
+			// write some extra data that is greater than readRangeEnd
+			err = writer.WriteRow(ctx, keyAfterRange, val)
+			require.NoError(t, err)
+			err = writer.WriteRow(ctx, keyAfterRange2, make([]byte, 300*1024))
+			require.NoError(t, err)
+			return writer.Close(ctx)
+		})
+	}
+	require.NoError(t, eg.Wait())
+	t.Log("finish writing 1000 files of 900KB")
+
+	for ; fileIdx < 2090; fileIdx++ {
+		fileIdx := fileIdx
+		eg.Go(func() error {
+			fileName := fmt.Sprintf("/test%d", fileIdx)
+			writer := NewWriterBuilder().BuildOneFile(store, fileName, "writerID")
+			err := writer.Init(ctx, 5*1024*1024)
+			require.NoError(t, err)
+
+			kvSize := 0
+			keyIdx := 0
+			for kvSize < 10*1024*1024 {
+				key := []byte(fmt.Sprintf("key%09d_%d", keyIdx, fileIdx))
+				keyIdx++
+				kvSize += len(key) + len(val)
+				err = writer.WriteRow(ctx, key, val)
+				require.NoError(t, err)
+			}
+
+			// write some extra data that is greater than readRangeEnd
+			err = writer.WriteRow(ctx, keyAfterRange, val)
+			require.NoError(t, err)
+			err = writer.WriteRow(ctx, keyAfterRange2, make([]byte, 900*1024))
+			require.NoError(t, err)
+			return writer.Close(ctx)
+		})
+	}
+	require.NoError(t, eg.Wait())
+	t.Log("finish writing 90 files of 10MB")
+
+	for ; fileIdx < 2091; fileIdx++ {
+		fileName := fmt.Sprintf("/test%d", fileIdx)
+		writer := NewWriterBuilder().BuildOneFile(store, fileName, "writerID")
+		err := writer.Init(ctx, 5*1024*1024)
+		require.NoError(t, err)
+
+		kvSize := 0
+		keyIdx := 0
+		for kvSize < 1024*1024*1024 {
+			key := []byte(fmt.Sprintf("key%010d_%d", keyIdx, fileIdx))
+			keyIdx++
+			kvSize += len(key) + len(val)
+			err = writer.WriteRow(ctx, key, val)
+			require.NoError(t, err)
+		}
+
+		// write some extra data that is greater than readRangeEnd
+		err = writer.WriteRow(ctx, keyAfterRange, val)
+		require.NoError(t, err)
+		err = writer.WriteRow(ctx, keyAfterRange2, make([]byte, 900*1024))
+		require.NoError(t, err)
+		err = writer.Close(ctx)
+		require.NoError(t, err)
+	}
+	t.Log("finish writing 1 file of 1G")
+
+finishCreateFiles:
+
+	dataFiles, statFiles, err := GetAllFileNames(ctx, store, "/")
+	require.NoError(t, err)
+	require.Equal(t, 2091, len(dataFiles))
+
+	p := newProfiler(true, true)
+	smallBlockBufPool := membuf.NewPool(
+		membuf.WithBlockNum(0),
+		membuf.WithBlockSize(smallBlockSize),
+	)
+	largeBlockBufPool := membuf.NewPool(
+		membuf.WithBlockNum(0),
+		membuf.WithBlockSize(ConcurrentReaderBufferSizePerConc),
+	)
+	output := &memKVsAndBuffers{}
+	p.beforeTest()
+	now := time.Now()
+	err = readAllData(ctx, store, dataFiles, statFiles, readRangeStart, readRangeEnd, smallBlockBufPool, largeBlockBufPool, output)
+	require.NoError(t, err)
+	output.build(ctx)
+	elapsed := time.Since(now)
+	p.afterTest()
+	t.Logf("readAllData time cost: %s, size: %d", elapsed.String(), output.size)
+}
diff --git a/br/pkg/lightning/backend/external/misc_bench_test.go b/br/pkg/lightning/backend/external/misc_bench_test.go
index 59ff4cce60..ddfcdb80e5 100644
--- a/br/pkg/lightning/backend/external/misc_bench_test.go
+++ b/br/pkg/lightning/backend/external/misc_bench_test.go
@@ -27,6 +27,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/felixge/fgprof"
 	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tidb/pkg/kv"
 	"github.com/pingcap/tidb/pkg/util/intest"
@@ -39,6 +40,7 @@ func openTestingStorage(t *testing.T) storage.ExternalStorage {
 	}
 	s, err := storage.NewFromURL(context.Background(), *testingStorageURI)
 	require.NoError(t, err)
+	t.Cleanup(s.Close)
 	return s
 }
 
@@ -347,6 +349,48 @@ func createAscendingFiles(
 	return kvCnt, minKey, maxKey
 }
 
+type profiler struct {
+	onAndOffCPUProf bool
+	peakMemProf     bool
+
+	caseIdx int
+
+	onAndOffCPUProfCloser func() error
+	heapProfDoneCh        chan struct{}
+	heapProfWg            *sync.WaitGroup
+}
+
+func newProfiler(onAndOffCPUProf, peakMemProf bool) *profiler {
+	return &profiler{
+		onAndOffCPUProf: onAndOffCPUProf,
+		peakMemProf:     peakMemProf,
+	}
+}
+
+func (p *profiler) beforeTest() {
+	p.caseIdx++
+	if p.onAndOffCPUProf {
+		fileCPU, err := os.Create(fmt.Sprintf("on-and-off-cpu-%d.prof", p.caseIdx))
+		intest.AssertNoError(err)
+		p.onAndOffCPUProfCloser = fgprof.Start(fileCPU, fgprof.FormatPprof)
+	}
+	if p.peakMemProf {
+		fileHeap := fmt.Sprintf("heap-%d.prof", p.caseIdx)
+		p.heapProfDoneCh, p.heapProfWg = recordHeapForMaxInUse(fileHeap)
+	}
+}
+
+func (p *profiler) afterTest() {
+	if p.onAndOffCPUProf {
+		err := p.onAndOffCPUProfCloser()
+		intest.AssertNoError(err)
+	}
+	if p.peakMemProf {
+		close(p.heapProfDoneCh)
+		p.heapProfWg.Wait()
+	}
+}
+
 func recordHeapForMaxInUse(filename string) (chan struct{}, *sync.WaitGroup) {
 	doneCh := make(chan struct{})
 	var wg sync.WaitGroup
diff --git a/br/pkg/lightning/backend/external/reader_test.go b/br/pkg/lightning/backend/external/reader_test.go
index 12f6e5151f..153b25f3b4 100644
--- a/br/pkg/lightning/backend/external/reader_test.go
+++ b/br/pkg/lightning/backend/external/reader_test.go
@@ -139,7 +139,11 @@ func TestReadLargeFile(t *testing.T) {
 	failpoint.Enable("github.com/pingcap/tidb/br/pkg/lightning/backend/external/assertReloadAtMostOnce", "return()")
 	defer failpoint.Disable("github.com/pingcap/tidb/br/pkg/lightning/backend/external/assertReloadAtMostOnce")
 
-	bufPool := membuf.NewPool(
+	smallBlockBufPool := membuf.NewPool(
+		membuf.WithBlockNum(0),
+		membuf.WithBlockSize(smallBlockSize),
+	)
+	largeBlockBufPool := membuf.NewPool(
 		membuf.WithBlockNum(0),
 		membuf.WithBlockSize(ConcurrentReaderBufferSizePerConc),
 	)
@@ -147,7 +151,7 @@ func TestReadLargeFile(t *testing.T) {
 	startKey := []byte("key000000")
 	maxKey := []byte("key004998")
 	endKey := []byte("key004999")
-	err = readAllData(ctx, memStore, datas, stats, startKey, endKey, bufPool, output)
+	err = readAllData(ctx, memStore, datas, stats, startKey, endKey, smallBlockBufPool, largeBlockBufPool, output)
 	require.NoError(t, err)
 	output.build(ctx)
 	require.Equal(t, startKey, output.keys[0])
diff --git a/br/pkg/lightning/backend/external/testutil.go b/br/pkg/lightning/backend/external/testutil.go
index fe4c882550..38de84aaf9 100644
--- a/br/pkg/lightning/backend/external/testutil.go
+++ b/br/pkg/lightning/backend/external/testutil.go
@@ -79,6 +79,7 @@ func testReadAndCompare(
 			curStart,
 			curEnd,
 			bufPool,
+			bufPool,
 			loaded,
 		)
 		require.NoError(t, err)
diff --git a/pkg/executor/aggregate/agg_spill_test.go b/pkg/executor/aggregate/agg_spill_test.go
index b94e21000a..731d4c7324 100644
--- a/pkg/executor/aggregate/agg_spill_test.go
+++ b/pkg/executor/aggregate/agg_spill_test.go
@@ -347,8 +347,8 @@ func TestGetCorrectResult(t *testing.T) {
 	ctx := mock.NewContext()
 	initCtx(ctx, newRootExceedAction, hardLimitBytesNum, 256)
 
-	rowNum := 100000 + rand.Intn(100000)
-	ndv := 50000 + rand.Intn(50000)
+	rowNum := 100000
+	ndv := 50000
 	col1, col2 := generateData(rowNum, ndv)
 	result := generateResult(col1, col2)
 	opt := getMockDataSourceParameters(ctx)
@@ -378,9 +378,9 @@ func TestGetCorrectResult(t *testing.T) {
 		executeCorrecResultTest(t, ctx, aggExec, dataSource, result)
 	}
 
-	require.Equal(t, 0, newRootExceedAction.GetTriggeredNum())
 	finished.Store(true)
 	wg.Wait()
+	require.NoError(t, failpoint.Disable("github.com/pingcap/tidb/pkg/executor/aggregate/slowSomePartialWorkers"))
 }
 
 func TestFallBackAction(t *testing.T) {
diff --git a/pkg/planner/cardinality/selectivity_test.go b/pkg/planner/cardinality/selectivity_test.go
index 00ad3f4a5d..494bb3f108 100644
--- a/pkg/planner/cardinality/selectivity_test.go
+++ b/pkg/planner/cardinality/selectivity_test.go
@@ -1252,3 +1252,48 @@ func TestIgnoreRealtimeStats(t *testing.T) {
 	testKit.MustExec("set @@tidb_opt_objective = 'determinate'")
 	testKit.MustQuery("explain select * from t where a = 1 and b > 2").Check(testkit.Rows(analyzedPlan...))
 }
+func TestSubsetIdxCardinality(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	testKit := testkit.NewTestKit(t, store)
+	h := dom.StatsHandle()
+
+	testKit.MustExec("use test")
+	testKit.MustExec("drop table if exists t")
+	testKit.MustExec("create table t(a int, b int, c int, index iabc(a, b, c))")
+	// Insert enough rows with differing cardinalities to test subset vs full index cardinality estimate.
+	// Result of a 2 column match should produce more rows than 3 column match.
+	testKit.MustExec("insert into t values (1, 1, 1), (1, 1, 1), (2, 1, 1), (2, 1, 1), (3, 1, 1), (3, 1, 1), (4, 1, 1), (4, 1, 1), (5, 1, 1), (5, 1, 1)")
+	testKit.MustExec("insert into t select a + 5, a, a from t")
+	for i := 1; i < 3; i++ {
+		testKit.MustExec(fmt.Sprintf("insert into t select a + 10 + %v, b + 1, c from t", i))
+	}
+	for j := 0; j < 3; j++ {
+		testKit.MustExec("insert into t select a, b, c from t")
+	}
+	testKit.MustExec("insert into t select a, b + 10, c from t")
+	require.NoError(t, h.DumpStatsDeltaToKV(true))
+	testKit.MustExec(`analyze table t`)
+
+	var (
+		input  []string
+		output []struct {
+			Query  string
+			Result []string
+		}
+	)
+	integrationSuiteData := cardinality.GetCardinalitySuiteData()
+	integrationSuiteData.LoadTestCases(t, &input, &output)
+	for i := 0; i < len(input); i++ {
+		testdata.OnRecord(func() {
+			output[i].Query = input[i]
+		})
+		if !strings.HasPrefix(input[i], "explain") {
+			testKit.MustExec(input[i])
+			continue
+		}
+		testdata.OnRecord(func() {
+			output[i].Result = testdata.ConvertRowsToStrings(testKit.MustQuery(input[i]).Rows())
+		})
+		testKit.MustQuery(input[i]).Check(testkit.Rows(output[i].Result...))
+	}
+}
diff --git a/pkg/planner/cardinality/testdata/cardinality_suite_in.json b/pkg/planner/cardinality/testdata/cardinality_suite_in.json
index ef620f330f..32ad34f22f 100644
--- a/pkg/planner/cardinality/testdata/cardinality_suite_in.json
+++ b/pkg/planner/cardinality/testdata/cardinality_suite_in.json
@@ -342,5 +342,15 @@
       "set @@tidb_opt_fix_control = '44855:ON'",
       "explain format = 'brief' select /*+ inl_join(t2) */ * from (select * from t where t.a < 1) as t1 join t t2 where t2.a = 0 and t1.a = t2.b"
     ]
+  },
+  {
+    "name": "TestSubsetIdxCardinality",
+    "cases": [
+      "explain format = 'brief' select distinct a from t",
+      "explain format = 'brief' select distinct b from t",
+      "explain format = 'brief' select distinct a, b, c from t",
+      "explain format = 'brief' select a, b from t where a = 1 and b = 1 and c = 1",
+      "explain format = 'brief' select a, b from t where a = 1 and b = 1"
+    ]
   }
 ]
\ No newline at end of file
diff --git a/pkg/planner/cardinality/testdata/cardinality_suite_out.json b/pkg/planner/cardinality/testdata/cardinality_suite_out.json
index 4a8640c6ec..dfd9563c71 100644
--- a/pkg/planner/cardinality/testdata/cardinality_suite_out.json
+++ b/pkg/planner/cardinality/testdata/cardinality_suite_out.json
@@ -4023,5 +4023,52 @@
         ]
       }
     ]
+  },
+  {
+    "Name": "TestSubsetIdxCardinality",
+    "Cases": [
+      {
+        "Query": "explain format = 'brief' select distinct a from t",
+        "Result": [
+          "StreamAgg 31.00 root  group by:test.t.a, funcs:firstrow(test.t.a)->test.t.a",
+          "└─IndexReader 31.00 root  index:StreamAgg",
+          "  └─StreamAgg 31.00 cop[tikv]  group by:test.t.a, ",
+          "    └─IndexFullScan 1280.00 cop[tikv] table:t, index:iabc(a, b, c) keep order:true"
+        ]
+      },
+      {
+        "Query": "explain format = 'brief' select distinct b from t",
+        "Result": [
+          "HashAgg 14.00 root  group by:test.t.b, funcs:firstrow(test.t.b)->test.t.b",
+          "└─IndexReader 14.00 root  index:HashAgg",
+          "  └─HashAgg 14.00 cop[tikv]  group by:test.t.b, ",
+          "    └─IndexFullScan 1280.00 cop[tikv] table:t, index:iabc(a, b, c) keep order:false"
+        ]
+      },
+      {
+        "Query": "explain format = 'brief' select distinct a, b, c from t",
+        "Result": [
+          "StreamAgg 70.00 root  group by:test.t.a, test.t.b, test.t.c, funcs:firstrow(test.t.a)->test.t.a, funcs:firstrow(test.t.b)->test.t.b, funcs:firstrow(test.t.c)->test.t.c",
+          "└─IndexReader 70.00 root  index:StreamAgg",
+          "  └─StreamAgg 70.00 cop[tikv]  group by:test.t.a, test.t.b, test.t.c, ",
+          "    └─IndexFullScan 1280.00 cop[tikv] table:t, index:iabc(a, b, c) keep order:true"
+        ]
+      },
+      {
+        "Query": "explain format = 'brief' select a, b from t where a = 1 and b = 1 and c = 1",
+        "Result": [
+          "Projection 16.00 root  test.t.a, test.t.b",
+          "└─IndexReader 16.00 root  index:IndexRangeScan",
+          "  └─IndexRangeScan 16.00 cop[tikv] table:t, index:iabc(a, b, c) range:[1 1 1,1 1 1], keep order:false"
+        ]
+      },
+      {
+        "Query": "explain format = 'brief' select a, b from t where a = 1 and b = 1",
+        "Result": [
+          "IndexReader 20.32 root  index:IndexRangeScan",
+          "└─IndexRangeScan 20.32 cop[tikv] table:t, index:iabc(a, b, c) range:[1 1,1 1], keep order:false"
+        ]
+      }
+    ]
   }
 ]
diff --git a/tests/integrationtest/r/planner/core/casetest/index/index.result b/tests/integrationtest/r/planner/core/casetest/index/index.result
index 844629db29..9fa10d38fd 100644
--- a/tests/integrationtest/r/planner/core/casetest/index/index.result
+++ b/tests/integrationtest/r/planner/core/casetest/index/index.result
@@ -721,8 +721,8 @@ a	b	c	d
 3	333	3.3000000000	13
 explain format='brief' select * from t1 where t1.a = 1 and t1.b < "333";
 id	estRows	task	access object	operator info
-TableReader	0.82	root		data:TableRangeScan
-└─TableRangeScan	0.82	cop[tikv]	table:t1	range:[1 -inf,1 "333"), keep order:false
+TableReader	1.00	root		data:TableRangeScan
+└─TableRangeScan	1.00	cop[tikv]	table:t1	range:[1 -inf,1 "333"), keep order:false
 select * from t1 where t1.a = 1 and t1.b < "333";
 a	b	c	d
 1	111	1.1000000000	11
diff --git a/tests/integrationtest/r/subquery.result b/tests/integrationtest/r/subquery.result
index e1be6063cc..cbef81238b 100644
--- a/tests/integrationtest/r/subquery.result
+++ b/tests/integrationtest/r/subquery.result
@@ -23,11 +23,11 @@ Projection	5.00	root		Column#22
   ├─TableReader(Build)	5.00	root		data:TableFullScan
   │ └─TableFullScan	5.00	cop[tikv]	table:t	keep order:false
   └─StreamAgg(Probe)	5.00	root		funcs:count(1)->Column#21
-    └─IndexJoin	1.12	root		inner join, inner:TableReader, outer key:subquery.t.a, inner key:subquery.t.a, equal cond:eq(subquery.t.a, subquery.t.a)
-      ├─IndexReader(Build)	2.24	root		index:IndexRangeScan
-      │ └─IndexRangeScan	2.24	cop[tikv]	table:s, index:idx(b, c, d)	range: decided by [eq(subquery.t.b, 1) eq(subquery.t.c, 1) eq(subquery.t.d, subquery.t.a)], keep order:false
-      └─TableReader(Probe)	2.24	root		data:TableRangeScan
-        └─TableRangeScan	2.24	cop[tikv]	table:t1	range: decided by [subquery.t.a], keep order:false
+    └─IndexJoin	2.50	root		inner join, inner:TableReader, outer key:subquery.t.a, inner key:subquery.t.a, equal cond:eq(subquery.t.a, subquery.t.a)
+      ├─IndexReader(Build)	5.00	root		index:IndexRangeScan
+      │ └─IndexRangeScan	5.00	cop[tikv]	table:s, index:idx(b, c, d)	range: decided by [eq(subquery.t.b, 1) eq(subquery.t.c, 1) eq(subquery.t.d, subquery.t.a)], keep order:false
+      └─TableReader(Probe)	5.00	root		data:TableRangeScan
+        └─TableRangeScan	5.00	cop[tikv]	table:t1	range: decided by [subquery.t.a], keep order:false
 drop table if exists t;
 create table t(a int, b int, c int);
 explain format = 'brief' select a from t t1 where t1.a = (select max(t2.a) from t t2 where t1.b=t2.b and t1.c=t2.b);
diff --git a/tests/integrationtest/r/util/ranger.result b/tests/integrationtest/r/util/ranger.result
index d68da2e9b6..7148284aad 100644
--- a/tests/integrationtest/r/util/ranger.result
+++ b/tests/integrationtest/r/util/ranger.result
@@ -349,8 +349,8 @@ a	b	c
 1	1	1
 explain format='brief' select * from t where a = 1 and ((b = 1) or (b = 2 and c = 3));
 id	estRows	task	access object	operator info
-IndexReader	1.71	root		index:IndexRangeScan
-└─IndexRangeScan	1.71	cop[tikv]	table:t, index:PRIMARY(a, b, c)	range:[1 1,1 1], [1 2 3,1 2 3], keep order:false
+IndexReader	2.00	root		index:IndexRangeScan
+└─IndexRangeScan	2.00	cop[tikv]	table:t, index:PRIMARY(a, b, c)	range:[1 1,1 1], [1 2 3,1 2 3], keep order:false
 select * from t where a = 1 and ((b = 1) or (b = 2 and c = 3));
 a	b	c
 1	1	1
@@ -368,30 +368,30 @@ a	b	c
 2	2	3
 explain format='brief' select * from t use index(primary) where ((a = 1) or (a = 2 and b = 2)) and c = 3;
 id	estRows	task	access object	operator info
-IndexReader	0.68	root		index:Selection
-└─Selection	0.68	cop[tikv]		eq(util__ranger.t.c, 3), or(eq(util__ranger.t.a, 1), and(eq(util__ranger.t.a, 2), eq(util__ranger.t.b, 2)))
+IndexReader	0.75	root		index:Selection
+└─Selection	0.75	cop[tikv]		eq(util__ranger.t.c, 3), or(eq(util__ranger.t.a, 1), and(eq(util__ranger.t.a, 2), eq(util__ranger.t.b, 2)))
   └─IndexRangeScan	2.00	cop[tikv]	table:t, index:PRIMARY(a, b, c)	range:[1,1], [2,2], keep order:false
 select * from t use index(primary) where ((a = 1) or (a = 2 and b = 2)) and c = 3;
 a	b	c
 2	2	3
 explain format='brief' select * from t where (a,b) in ((1,1),(2,2)) and c > 2 and (a,b,c) in ((1,1,1),(2,2,3));
 id	estRows	task	access object	operator info
-Selection	0.58	root		gt(util__ranger.t.c, 2), or(and(eq(util__ranger.t.a, 1), eq(util__ranger.t.b, 1)), and(eq(util__ranger.t.a, 2), eq(util__ranger.t.b, 2)))
+Selection	0.75	root		gt(util__ranger.t.c, 2), or(and(eq(util__ranger.t.a, 1), eq(util__ranger.t.b, 1)), and(eq(util__ranger.t.a, 2), eq(util__ranger.t.b, 2)))
 └─Batch_Point_Get	2.00	root	table:t, index:PRIMARY(a, b, c)	keep order:false, desc:false
 select * from t where (a,b) in ((1,1),(2,2)) and c > 2 and (a,b,c) in ((1,1,1),(2,2,3));
 a	b	c
 2	2	3
 explain format='brief' select * from t where (a,b) in ((1,1),(2,2)) and c > 2;
 id	estRows	task	access object	operator info
-IndexReader	1.19	root		index:IndexRangeScan
-└─IndexRangeScan	1.19	cop[tikv]	table:t, index:PRIMARY(a, b, c)	range:(1 1 2,1 1 +inf], (2 2 2,2 2 +inf], keep order:false
+IndexReader	2.00	root		index:IndexRangeScan
+└─IndexRangeScan	2.00	cop[tikv]	table:t, index:PRIMARY(a, b, c)	range:(1 1 2,1 1 +inf], (2 2 2,2 2 +inf], keep order:false
 select * from t where (a,b) in ((1,1),(2,2)) and c > 2;
 a	b	c
 2	2	3
 explain format='brief' select * from t where ((a = 1 and b = 1) or (a = 2 and b = 2)) and c > 2;
 id	estRows	task	access object	operator info
-IndexReader	1.19	root		index:IndexRangeScan
-└─IndexRangeScan	1.19	cop[tikv]	table:t, index:PRIMARY(a, b, c)	range:(1 1 2,1 1 +inf], (2 2 2,2 2 +inf], keep order:false
+IndexReader	2.00	root		index:IndexRangeScan
+└─IndexRangeScan	2.00	cop[tikv]	table:t, index:PRIMARY(a, b, c)	range:(1 1 2,1 1 +inf], (2 2 2,2 2 +inf], keep order:false
 select * from t where ((a = 1 and b = 1) or (a = 2 and b = 2)) and c > 2;
 a	b	c
 2	2	3