Repository: pingcap/tidb
Pull request #59163 (2025-01-24 00:28:11)

domain: move deltaUpdateTicker/gcstats into a new goroutine (#58926)
This is an automated cherry-pick of #58926

<!--

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 #58920

Problem Summary:

### What changed and how does it work?

### 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 -->

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
move deltaUpdateTicker/gcstats into a new goroutine to avoid blocking problem

将deltaupdateticker/gcstats移至新的goroutine中，以避免阻塞问题
```


ti-chi-bot[bot]: This cherry pick PR is for a release branch and has not yet been approved by triage owners.
Adding the `do-not-merge/cherry-pick-not-approved` label.

To merge this cherry pick:
1. It must be approved by the approvers firstly.
2. **AFTER** it has been approved by approvers, please wait for the cherry-pick merging approval from triage owners.


<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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository.
</details>
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 assign [d3hunter](https://github.com/d3hunter) for approval. For more information see [the 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:

- **[pkg/domain/OWNERS](https://github.com/pingcap/tidb/blob/release-8.5/pkg/domain/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":["d3hunter"]} -->
ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/pingcap/tidb/pull/59163#pullrequestreview-2571397538" title="Approved">hawkingrei</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign [d3hunter](https://github.com/d3hunter) for approval. For more information see [the 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:

- **[pkg/domain/OWNERS](https://github.com/pingcap/tidb/blob/release-8.5/pkg/domain/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":["d3hunter"]} -->
ti-chi-bot[bot]: [LGTM Timeline notifier]
---
Timeline:

- `2025-01-24 00:28:33.760460901 +0000 UTC m=+399841.091380305`: :ballot_box_with_check: agreed by [hawkingrei](https://github.com/hawkingrei).
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pingcap/tidb/pull/59163?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
Attention: Patch coverage is `79.74684%` with `16 lines` in your changes missing coverage. Please review.
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`release-8.5@b313166`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#section-missing-base-commit) about missing BASE report.

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


```diff
@@               Coverage Diff                @@
##             release-8.5     #59163   +/-   ##
================================================
  Coverage               ?   16.7568%           
================================================
  Files                  ?       1621           
  Lines                  ?     597087           
  Branches               ?          0           
================================================
  Hits                   ?     100053           
  Misses                 ?     483228           
  Partials               ?      13806           
```

| [Flag](https://app.codecov.io/gh/pingcap/tidb/pull/59163/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/59163/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `16.7568% <79.7468%> (?)` | |

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/59163/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/59163/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `∅ <0.0000%> (?)` | |
| [parser](https://app.codecov.io/gh/pingcap/tidb/pull/59163/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `∅ <0.0000%> (?)` | |
| [br](https://app.codecov.io/gh/pingcap/tidb/pull/59163/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `21.0354% <0.0000%> (?)` | |

</details>
ti-chi-bot[bot]: @ti-chi-bot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
idc-jenkins-ci-tidb/unit-test | 72af863aababbed7cf2f24b604523c418dddb438 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tidb/job/release-8.5/job/pull_unit_test/667/display/redirect) | true | `/test unit-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tidb&pr=59163). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Ati-chi-bot).

<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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
hawkingrei: /retest
ti-chi-bot[bot]: @ti-chi-bot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
idc-jenkins-ci-tidb/unit-test | 72af863aababbed7cf2f24b604523c418dddb438 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tidb/job/release-8.5/job/pull_unit_test/668/display/redirect) | true | `/test unit-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tidb&pr=59163). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Ati-chi-bot).

<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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/pingcap/tidb/pull/59163#pullrequestreview-2571397538" title="Approved">hawkingrei</a>*, *<a href="https://github.com/pingcap/tidb/pull/59163#pullrequestreview-2571508121" title="Approved">winoros</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign [d3hunter](https://github.com/d3hunter) for approval. For more information see [the 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:

- **[pkg/domain/OWNERS](https://github.com/pingcap/tidb/blob/release-8.5/pkg/domain/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":["d3hunter"]} -->
ti-chi-bot[bot]: [LGTM Timeline notifier]
---
Timeline:

- `2025-01-24 00:28:33.760460901 +0000 UTC m=+399841.091380305`: :ballot_box_with_check: agreed by [hawkingrei](https://github.com/hawkingrei).
- `2025-01-24 02:22:36.524222634 +0000 UTC m=+406683.855142033`: :ballot_box_with_check: agreed by [winoros](https://github.com/winoros).
hawkingrei: /retest
ti-chi-bot[bot]: @ti-chi-bot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
idc-jenkins-ci-tidb/unit-test | 72af863aababbed7cf2f24b604523c418dddb438 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tidb/job/release-8.5/job/pull_unit_test/669/display/redirect) | true | `/test unit-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tidb&pr=59163). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Ati-chi-bot).

<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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/pingcap/tidb/pull/59163#pullrequestreview-2571397538" title="Approved">hawkingrei</a>*, *<a href="https://github.com/pingcap/tidb/pull/59163#pullrequestreview-2571508121" title="Approved">winoros</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign [gmhdbjd](https://github.com/gmhdbjd) for approval. For more information see [the 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:

- ~~[OWNERS](https://github.com/pingcap/tidb/blob/release-8.5/OWNERS)~~ [hawkingrei,winoros]
- **[pkg/domain/OWNERS](https://github.com/pingcap/tidb/blob/release-8.5/pkg/domain/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":["gmhdbjd"]} -->
ti-chi-bot[bot]: @ti-chi-bot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
idc-jenkins-ci-tidb/unit-test | 3a05c1c8a051bfa677355c9819e785ac7ed23e3c | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tidb/job/release-8.5/job/pull_unit_test/670/display/redirect) | true | `/test unit-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tidb&pr=59163). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Ati-chi-bot).

<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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
ti-chi-bot[bot]: @ti-chi-bot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
idc-jenkins-ci-tidb/unit-test | 377c626b150bbdc575cc06cd0d2066d72cd147c4 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tidb/job/release-8.5/job/pull_unit_test/671/display/redirect) | true | `/test unit-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tidb&pr=59163). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Ati-chi-bot).

<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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
ti-chi-bot[bot]: @ti-chi-bot: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
idc-jenkins-ci-tidb/unit-test | 377c626b150bbdc575cc06cd0d2066d72cd147c4 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tidb/job/release-8.5/job/pull_unit_test/671/display/redirect) | true | `/test unit-test`
idc-jenkins-ci-tidb/build | 377c626b150bbdc575cc06cd0d2066d72cd147c4 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tidb/job/release-8.5/job/pull_build/629/display/redirect) | true | `/test build`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tidb&pr=59163). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Ati-chi-bot).

<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-sigs/prow](https://github.com/kubernetes-sigs/prow/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/WORKSPACE b/WORKSPACE
index c6806481a5..b2e63a9aa5 100644
--- a/WORKSPACE
+++ b/WORKSPACE
@@ -85,7 +85,7 @@ go_download_sdk(
         "https://mirrors.aliyun.com/golang/{}",
         "https://dl.google.com/go/{}",
     ],
-    version = "1.23.3",
+    version = "1.23.6",
 )
 
 go_register_toolchains(
diff --git a/build/image/base b/build/image/base
index 3a9c8ca381..630e900886 100644
--- a/build/image/base
+++ b/build/image/base
@@ -30,7 +30,7 @@ RUN --mount=type=cache,target=/var/cache/dnf \
 
 # install golang toolchain
 # renovate: datasource=docker depName=golang
-ARG GOLANG_VERSION=1.23.3
+ARG GOLANG_VERSION=1.23.6
 RUN OS=linux; ARCH=$([ "$(arch)" = "x86_64" ] && echo amd64 || echo arm64); \
     curl -fsSL https://dl.google.com/go/go${GOLANG_VERSION}.linux-${ARCH}.tar.gz | tar -C /usr/local -xz
 ENV PATH /usr/local/go/bin/:$PATH
diff --git a/dumpling/README.md b/dumpling/README.md
index 076ae57a4f..389259ab9b 100644
--- a/dumpling/README.md
+++ b/dumpling/README.md
@@ -29,7 +29,7 @@ Building
 --------
 
 0. Under directory `tidb`
-1. Install Go 1.16 or above
+1. Install Go 1.23.6 or above
 2. Run `make build_dumpling` to compile. The output is in `bin/dumpling`.
 3. Run `make dumpling_unit_test` to run the unit tests.
 4. Run `make dumpling_integration_test` to run integration tests. For integration test:
diff --git a/go.mod b/go.mod
index bd66e04ea5..a0070c6bb5 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,6 @@
 module github.com/pingcap/tidb
 
-go 1.23
+go 1.23.6
 
 require (
 	cloud.google.com/go/kms v1.15.7
diff --git a/pkg/ddl/notifier/BUILD.bazel b/pkg/ddl/notifier/BUILD.bazel
index 29ac7a6724..cf7088767c 100644
--- a/pkg/ddl/notifier/BUILD.bazel
+++ b/pkg/ddl/notifier/BUILD.bazel
@@ -18,6 +18,7 @@ go_library(
         "//pkg/parser/model",
         "//pkg/sessionctx",
         "//pkg/util",
+        "//pkg/util/chunk",
         "//pkg/util/intest",
         "//pkg/util/logutil",
         "@com_github_pingcap_errors//:errors",
@@ -30,11 +31,12 @@ go_test(
     timeout = "short",
     srcs = [
         "events_test.go",
+        "store_test.go",
         "testkit_test.go",
     ],
     embed = [":notifier"],
     flaky = True,
-    shard_count = 7,
+    shard_count = 10,
     deps = [
         "//pkg/ddl",
         "//pkg/ddl/session",
@@ -43,9 +45,12 @@ go_test(
         "//pkg/sessionctx",
         "//pkg/testkit",
         "//pkg/testkit/testfailpoint",
+        "//pkg/types",
         "//pkg/util",
+        "//pkg/util/chunk",
         "@com_github_ngaut_pools//:pools",
         "@com_github_pingcap_log//:log",
         "@com_github_stretchr_testify//require",
+        "@org_uber_go_atomic//:atomic",
     ],
 )
diff --git a/pkg/ddl/notifier/publish.go b/pkg/ddl/notifier/publish.go
index 6358ccd6a4..88289c0be1 100644
--- a/pkg/ddl/notifier/publish.go
+++ b/pkg/ddl/notifier/publish.go
@@ -22,30 +22,30 @@ import (
 
 // PubSchemeChangeToStore publishes schema changes to the store to notify
 // subscribers on the Store. It stages changes in given `se` so they will be
-// visible when `se` further commits. When the schema change is not from
-// multi-schema change DDL, `multiSchemaChangeSeq` is -1. Otherwise,
-// `multiSchemaChangeSeq` is the sub-job index of the multi-schema change DDL.
+// visible when `se` further commits. When the DDL contains only one schema
+// change, `subJobID` is -1. Otherwise, `subJobID` is the sub-job index of the
+// DDL, like multi-schema change or batched create table.
 func PubSchemeChangeToStore(
 	ctx context.Context,
 	se *sess.Session,
 	ddlJobID int64,
-	multiSchemaChangeSeq int64,
+	subJobID int64,
 	event *SchemaChangeEvent,
 	store Store,
 ) error {
-	change := &schemaChange{
-		ddlJobID:             ddlJobID,
-		multiSchemaChangeSeq: multiSchemaChangeSeq,
-		event:                event,
+	change := &SchemaChange{
+		ddlJobID: ddlJobID,
+		subJobID: subJobID,
+		event:    event,
 	}
 	return store.Insert(ctx, se, change)
 }
 
-// schemaChange is the Golang representation of the persistent data. (ddlJobID,
-// multiSchemaChangeSeq) should be unique in the cluster.
-type schemaChange struct {
-	ddlJobID             int64
-	multiSchemaChangeSeq int64
-	event                *SchemaChangeEvent
-	processedByFlag      uint64
+// SchemaChange is the Golang representation of the persistent data. (ddlJobID,
+// subJobID) should be unique in the cluster.
+type SchemaChange struct {
+	ddlJobID        int64
+	subJobID        int64
+	event           *SchemaChangeEvent
+	processedByFlag uint64
 }
diff --git a/pkg/ddl/notifier/store.go b/pkg/ddl/notifier/store.go
index a46b390702..cd168f3d50 100644
--- a/pkg/ddl/notifier/store.go
+++ b/pkg/ddl/notifier/store.go
@@ -21,11 +21,15 @@ import (
 
 	"github.com/pingcap/errors"
 	sess "github.com/pingcap/tidb/pkg/ddl/session"
+	"github.com/pingcap/tidb/pkg/util/chunk"
 )
 
+// CloseFn is the function to release the resource.
+type CloseFn func()
+
 // Store is the (de)serialization and persistent layer.
 type Store interface {
-	Insert(context.Context, *sess.Session, *schemaChange) error
+	Insert(context.Context, *sess.Session, *SchemaChange) error
 	UpdateProcessed(
 		ctx context.Context,
 		se *sess.Session,
@@ -34,7 +38,20 @@ type Store interface {
 		processedBy uint64,
 	) error
 	DeleteAndCommit(ctx context.Context, se *sess.Session, ddlJobID int64, multiSchemaChangeID int) error
-	List(ctx context.Context, se *sess.Session) ([]*schemaChange, error)
+	// List will start a transaction of given session and read all schema changes
+	// through a ListResult. The ownership of session is occupied by Store until
+	// CloseFn is called.
+	List(ctx context.Context, se *sess.Session) (ListResult, CloseFn)
+}
+
+// ListResult is the result stream of a List operation.
+type ListResult interface {
+	// Read tries to decode at most `len(changes)` SchemaChange into given slices. It
+	// returns the number of schemaChanges decoded, 0 means no more schemaChanges.
+	//
+	// Note that the previous SchemaChange in the slice will be overwritten when call
+	// Read.
+	Read(changes []*SchemaChange) (int, error)
 }
 
 type tableStore struct {
@@ -42,7 +59,8 @@ type tableStore struct {
 	table string
 }
 
-func (t *tableStore) Insert(ctx context.Context, s *sess.Session, change *schemaChange) error {
+// Insert implements Store interface.
+func (t *tableStore) Insert(ctx context.Context, s *sess.Session, change *SchemaChange) error {
 	event, err := json.Marshal(change.event)
 	if err != nil {
 		return errors.Trace(err)
@@ -58,11 +76,12 @@ func (t *tableStore) Insert(ctx context.Context, s *sess.Session, change *schema
 	)
 	_, err = s.Execute(
 		ctx, sql, "ddl_notifier",
-		change.ddlJobID, change.multiSchemaChangeSeq, event,
+		change.ddlJobID, change.subJobID, event,
 	)
 	return err
 }
 
+// UpdateProcessed implements Store interface.
 func (t *tableStore) UpdateProcessed(
 	ctx context.Context,
 	se *sess.Session,
@@ -81,6 +100,7 @@ func (t *tableStore) UpdateProcessed(
 	return err
 }
 
+// DeleteAndCommit implements Store interface.
 func (t *tableStore) DeleteAndCommit(
 	ctx context.Context,
 	se *sess.Session,
@@ -106,34 +126,85 @@ func (t *tableStore) DeleteAndCommit(
 	return errors.Trace(err)
 }
 
-func (t *tableStore) List(ctx context.Context, se *sess.Session) ([]*schemaChange, error) {
-	sql := fmt.Sprintf(`
-		SELECT
-			ddl_job_id,
-			sub_job_id,
-			schema_change,
-			processed_by_flag
-		FROM %s.%s ORDER BY ddl_job_id, sub_job_id`,
-		t.db, t.table)
-	rows, err := se.Execute(ctx, sql, "ddl_notifier")
+// List implements Store interface.
+func (t *tableStore) List(ctx context.Context, se *sess.Session) (ListResult, CloseFn) {
+	return &listResult{
+		ctx: ctx,
+		se:  se,
+		sqlTemplate: fmt.Sprintf(`
+			SELECT
+				ddl_job_id,
+				sub_job_id,
+				schema_change,
+				processed_by_flag
+			FROM %s.%s
+			WHERE (ddl_job_id, sub_job_id) > (%%?, %%?)
+			ORDER BY ddl_job_id, sub_job_id
+			LIMIT %%?`,
+			t.db, t.table),
+		// DDL job ID are always positive, so we can use 0 as the initial value.
+		maxReturnedDDLJobID: 0,
+		maxReturnedSubJobID: 0,
+	}, se.Rollback
+}
+
+type listResult struct {
+	ctx                 context.Context
+	se                  *sess.Session
+	sqlTemplate         string
+	maxReturnedDDLJobID int64
+	maxReturnedSubJobID int64
+}
+
+// Read implements ListResult interface.
+func (r *listResult) Read(changes []*SchemaChange) (int, error) {
+	if r.maxReturnedDDLJobID == 0 && r.maxReturnedSubJobID == 0 {
+		err := r.se.Begin(r.ctx)
+		if err != nil {
+			return 0, errors.Trace(err)
+		}
+	}
+
+	rows, err := r.se.Execute(
+		r.ctx, r.sqlTemplate, "ddl_notifier",
+		r.maxReturnedDDLJobID, r.maxReturnedSubJobID, len(changes),
+	)
 	if err != nil {
-		return nil, err
+		return 0, errors.Trace(err)
 	}
-	ret := make([]*schemaChange, 0, len(rows))
-	for _, row := range rows {
-		event := SchemaChangeEvent{}
-		err = json.Unmarshal(row.GetBytes(2), &event)
+
+	if err = r.unmarshalSchemaChanges(rows, changes); err != nil {
+		return 0, errors.Trace(err)
+	}
+	return len(rows), nil
+}
+
+func (r *listResult) unmarshalSchemaChanges(rows []chunk.Row, changes []*SchemaChange) error {
+	for i, row := range rows {
+		if changes[i] == nil {
+			changes[i] = new(SchemaChange)
+		}
+		if changes[i].event == nil {
+			changes[i].event = new(SchemaChangeEvent)
+		}
+		if changes[i].event.inner == nil {
+			changes[i].event.inner = new(jsonSchemaChangeEvent)
+		}
+
+		err := json.Unmarshal(row.GetBytes(2), changes[i].event.inner)
 		if err != nil {
-			return nil, errors.Trace(err)
+			return errors.Trace(err)
+		}
+		changes[i].ddlJobID = row.GetInt64(0)
+		changes[i].subJobID = row.GetInt64(1)
+		changes[i].processedByFlag = row.GetUint64(3)
+
+		if i == len(rows)-1 {
+			r.maxReturnedDDLJobID = changes[i].ddlJobID
+			r.maxReturnedSubJobID = changes[i].subJobID
 		}
-		ret = append(ret, &schemaChange{
-			ddlJobID:             row.GetInt64(0),
-			multiSchemaChangeSeq: row.GetInt64(1),
-			event:                &event,
-			processedByFlag:      row.GetUint64(3),
-		})
 	}
-	return ret, nil
+	return nil
 }
 
 // OpenTableStore opens a store on a created table `db`.`table`. The table should
diff --git a/pkg/ddl/notifier/subscribe.go b/pkg/ddl/notifier/subscribe.go
index 2bb040f888..9975d29021 100644
--- a/pkg/ddl/notifier/subscribe.go
+++ b/pkg/ddl/notifier/subscribe.go
@@ -166,61 +166,89 @@ func (n *DDLNotifier) start() {
 	}
 }
 
+// ProcessEventsBatchSize is the number of events to process in a SQL query. It's
+// exposed for testing.
+var ProcessEventsBatchSize = 1024
+
 func (n *DDLNotifier) processEvents(ctx context.Context) error {
-	sysSession, err := n.sysSessionPool.Get()
+	s, err := n.sysSessionPool.Get()
 	if err != nil {
 		return errors.Trace(err)
 	}
-	defer n.sysSessionPool.Put(sysSession)
+	defer n.sysSessionPool.Put(s)
+	sess4List := sess.NewSession(s.(sessionctx.Context))
+	result, closeFn := n.store.List(ctx, sess4List)
+	defer closeFn()
 
-	session := sess.NewSession(sysSession.(sessionctx.Context))
-	changes, err := n.store.List(ctx, session)
+	s2, err := n.sysSessionPool.Get()
 	if err != nil {
 		return errors.Trace(err)
 	}
+	defer n.sysSessionPool.Put(s2)
+	sess4Process := sess.NewSession(s2.(sessionctx.Context))
 
 	// we should ensure deliver order of events to a handler, so if a handler returns
 	// error for previous events it should not receive later events.
 	skipHandlers := make(map[HandlerID]struct{})
-	for _, change := range changes {
-		for handlerID, handler := range n.handlers {
-			if _, ok := skipHandlers[handlerID]; ok {
-				continue
-			}
-			if err2 := n.processEventForHandler(ctx, session, change, handlerID, handler); err2 != nil {
-				skipHandlers[handlerID] = struct{}{}
 
-				if !goerr.Is(err2, ErrNotReadyRetryLater) {
-					logutil.Logger(ctx).Error("Error processing change",
-						zap.Int64("ddlJobID", change.ddlJobID),
-						zap.Int64("multiSchemaChangeSeq", change.multiSchemaChangeSeq),
-						zap.Stringer("handler", handlerID),
-						zap.Error(err2))
+	changes := make([]*SchemaChange, ProcessEventsBatchSize)
+
+	for {
+		count, err2 := result.Read(changes)
+		if err2 != nil {
+			return errors.Trace(err2)
+		}
+		if count == 0 {
+			break
+		}
+
+		for _, change := range changes[:count] {
+			for handlerID, handler := range n.handlers {
+				if _, ok := skipHandlers[handlerID]; ok {
+					continue
+				}
+				if err3 := n.processEventForHandler(ctx, sess4Process, change, handlerID, handler); err3 != nil {
+					skipHandlers[handlerID] = struct{}{}
+
+					if !goerr.Is(err3, ErrNotReadyRetryLater) {
+						logutil.Logger(ctx).Error("Error processing change",
+							zap.Int64("ddlJobID", change.ddlJobID),
+							zap.Int64("subJobID", change.subJobID),
+							zap.Stringer("handler", handlerID),
+							zap.Error(err3))
+					}
+					continue
 				}
-				continue
 			}
-		}
 
-		if intest.InTest {
-			if n.handlersBitMap == 0 {
-				// There are unit tests that directly check the system table while no subscriber
-				// is registered. We continue the loop to skip DELETE the events in table so
-				// tests can check them.
-				continue
+			if intest.InTest {
+				if n.handlersBitMap == 0 {
+					// There are unit tests that directly check the system table while no subscriber
+					// is registered. We continue the loop to skip DELETE the events in table so
+					// tests can check them.
+					continue
+				}
 			}
-		}
 
-		if change.processedByFlag == n.handlersBitMap {
-			if err2 := n.store.DeleteAndCommit(
-				ctx,
-				session,
-				change.ddlJobID,
-				int(change.multiSchemaChangeSeq),
-			); err2 != nil {
-				logutil.Logger(ctx).Error("Error deleting change",
-					zap.Int64("ddlJobID", change.ddlJobID),
-					zap.Int64("multiSchemaChangeSeq", change.multiSchemaChangeSeq),
-					zap.Error(err2))
+			if change.processedByFlag == n.handlersBitMap {
+				s3, err3 := n.sysSessionPool.Get()
+				if err3 != nil {
+					return errors.Trace(err3)
+				}
+				sess4Del := sess.NewSession(s3.(sessionctx.Context))
+				err3 = n.store.DeleteAndCommit(
+					ctx,
+					sess4Del,
+					change.ddlJobID,
+					int(change.subJobID),
+				)
+				n.sysSessionPool.Put(s3)
+				if err3 != nil {
+					logutil.Logger(ctx).Error("Error deleting change",
+						zap.Int64("ddlJobID", change.ddlJobID),
+						zap.Int64("subJobID", change.subJobID),
+						zap.Error(err3))
+				}
 			}
 		}
 	}
@@ -233,7 +261,7 @@ const slowHandlerLogThreshold = time.Second * 5
 func (n *DDLNotifier) processEventForHandler(
 	ctx context.Context,
 	session *sess.Session,
-	change *schemaChange,
+	change *SchemaChange,
 	handlerID HandlerID,
 	handler SchemaChangeHandler,
 ) (err error) {
@@ -260,7 +288,7 @@ func (n *DDLNotifier) processEventForHandler(
 		logutil.Logger(ctx).Warn("Slow process event",
 			zap.Stringer("handler", handlerID),
 			zap.Int64("ddlJobID", change.ddlJobID),
-			zap.Int64("multiSchemaChangeSeq", change.multiSchemaChangeSeq),
+			zap.Int64("subJobID", change.subJobID),
 			zap.Stringer("event", change.event),
 			zap.Duration("duration", time.Since(now)))
 	}
@@ -270,7 +298,7 @@ func (n *DDLNotifier) processEventForHandler(
 		ctx,
 		session,
 		change.ddlJobID,
-		change.multiSchemaChangeSeq,
+		change.subJobID,
 		newFlag,
 	); err != nil {
 		return errors.Trace(err)
diff --git a/pkg/domain/domain.go b/pkg/domain/domain.go
index 478b86fbf8..e021517ff4 100644
--- a/pkg/domain/domain.go
+++ b/pkg/domain/domain.go
@@ -2365,40 +2365,31 @@ func (do *Domain) UpdateTableStatsLoop(ctx, initStatsCtx sessionctx.Context) err
 		do.indexUsageWorker()
 	}, "indexUsageWorker")
 	if do.statsLease <= 0 {
-		// For statsLease > 0, `updateStatsWorker` handles the quit of stats owner.
+		// For statsLease > 0, `gcStatsWorker` handles the quit of stats owner.
 		do.wg.Run(func() { quitStatsOwner(do, do.statsOwner) }, "quitStatsOwner")
 		return nil
 	}
+	waitStartTask := func(do *Domain, fn func()) {
+		select {
+		case <-do.StatsHandle().InitStatsDone:
+		case <-do.exit: // It may happen that before initStatsDone, tidb receive Ctrl+C
+			return
+		}
+		fn()
+	}
 	do.SetStatsUpdating(true)
+	// The asyncLoadHistogram/dumpColStatsUsageWorker/deltaUpdateTickerWorker doesn't require the stats initialization to be completed.
+	// This is because thos workers' primary responsibilities are to update the change delta and handle DDL operations.
+	// These tasks need to be in work mod as soon as possible to avoid the problem.
 	do.wg.Run(do.asyncLoadHistogram, "asyncLoadHistogram")
-	// The stats updated worker doesn't require the stats initialization to be completed.
-	// This is because the updated worker's primary responsibilities are to update the change delta and handle DDL operations.
-	// These tasks do not interfere with or depend on the initialization process.
-	do.wg.Run(func() { do.updateStatsWorker(ctx) }, "updateStatsWorker")
+	do.wg.Run(do.deltaUpdateTickerWorker, "deltaUpdateTickerWorker")
+	do.wg.Run(do.dumpColStatsUsageWorker, "dumpColStatsUsageWorker")
+	do.wg.Run(func() { waitStartTask(do, do.gcStatsWorker) }, "gcStatsWorker")
+
 	// Wait for the stats worker to finish the initialization.
 	// Otherwise, we may start the auto analyze worker before the stats cache is initialized.
-	do.wg.Run(
-		func() {
-			select {
-			case <-do.StatsHandle().InitStatsDone:
-			case <-do.exit: // It may happen that before initStatsDone, tidb receive Ctrl+C
-				return
-			}
-			do.autoAnalyzeWorker()
-		},
-		"autoAnalyzeWorker",
-	)
-	do.wg.Run(
-		func() {
-			select {
-			case <-do.StatsHandle().InitStatsDone:
-			case <-do.exit: // It may happen that before initStatsDone, tidb receive Ctrl+C
-				return
-			}
-			do.analyzeJobsCleanupWorker()
-		},
-		"analyzeJobsCleanupWorker",
-	)
+	do.wg.Run(func() { waitStartTask(do, do.autoAnalyzeWorker) }, "autoAnalyzeWorker")
+	do.wg.Run(func() { waitStartTask(do, do.analyzeJobsCleanupWorker) }, "analyzeJobsCleanupWorker")
 	do.wg.Run(
 		func() {
 			// The initStatsCtx is used to store the internal session for initializing stats,
@@ -2579,60 +2570,66 @@ func (do *Domain) indexUsageWorker() {
 	}
 }
 
-func (do *Domain) updateStatsWorkerExitPreprocessing(statsHandle *handle.Handle) {
+func (do *Domain) gcStatsWorkerExitPreprocessing() {
 	ch := make(chan struct{}, 1)
 	timeout, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	go func() {
-		logutil.BgLogger().Info("updateStatsWorker is going to exit, start to flush stats")
-		statsHandle.FlushStats()
-		logutil.BgLogger().Info("updateStatsWorker ready to release owner")
+		logutil.BgLogger().Info("gcStatsWorker ready to release owner")
 		do.statsOwner.Close()
 		ch <- struct{}{}
 	}()
 	select {
 	case <-ch:
-		logutil.BgLogger().Info("updateStatsWorker exit preprocessing finished")
+		logutil.BgLogger().Info("gcStatsWorker exit preprocessing finished")
 		return
 	case <-timeout.Done():
-		logutil.BgLogger().Warn("updateStatsWorker exit preprocessing timeout, force exiting")
+		logutil.BgLogger().Warn("gcStatsWorker exit preprocessing timeout, force exiting")
 		return
 	}
 }
 
-func (do *Domain) updateStatsWorker(_ sessionctx.Context) {
-	defer util.Recover(metrics.LabelDomain, "updateStatsWorker", nil, false)
-	logutil.BgLogger().Info("updateStatsWorker started.")
+func (*Domain) deltaUpdateTickerWorkerExitPreprocessing(statsHandle *handle.Handle) {
+	ch := make(chan struct{}, 1)
+	timeout, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	go func() {
+		logutil.BgLogger().Info("deltaUpdateTicker is going to exit, start to flush stats")
+		statsHandle.FlushStats()
+		ch <- struct{}{}
+	}()
+	select {
+	case <-ch:
+		logutil.BgLogger().Info("deltaUpdateTicker exit preprocessing finished")
+		return
+	case <-timeout.Done():
+		logutil.BgLogger().Warn("deltaUpdateTicker exit preprocessing timeout, force exiting")
+		return
+	}
+}
+
+func (do *Domain) gcStatsWorker() {
+	defer util.Recover(metrics.LabelDomain, "gcStatsWorker", nil, false)
+	logutil.BgLogger().Info("gcStatsWorker started.")
 	lease := do.statsLease
-	// We need to have different nodes trigger tasks at different times to avoid the herd effect.
-	randDuration := time.Duration(rand.Int63n(int64(time.Minute)))
-	deltaUpdateTicker := time.NewTicker(20*lease + randDuration)
 	gcStatsTicker := time.NewTicker(100 * lease)
-	dumpColStatsUsageTicker := time.NewTicker(100 * lease)
 	updateStatsHealthyTicker := time.NewTicker(20 * lease)
 	readMemTicker := time.NewTicker(memory.ReadMemInterval)
 	statsHandle := do.StatsHandle()
 	defer func() {
-		dumpColStatsUsageTicker.Stop()
 		gcStatsTicker.Stop()
-		deltaUpdateTicker.Stop()
 		readMemTicker.Stop()
 		updateStatsHealthyTicker.Stop()
 		do.SetStatsUpdating(false)
-		logutil.BgLogger().Info("updateStatsWorker exited.")
+		logutil.BgLogger().Info("gcStatsWorker exited.")
 	}()
-	defer util.Recover(metrics.LabelDomain, "updateStatsWorker", nil, false)
+	defer util.Recover(metrics.LabelDomain, "gcStatsWorker", nil, false)
 
 	for {
 		select {
 		case <-do.exit:
-			do.updateStatsWorkerExitPreprocessing(statsHandle)
+			do.gcStatsWorkerExitPreprocessing()
 			return
-		case <-deltaUpdateTicker.C:
-			err := statsHandle.DumpStatsDeltaToKV(false)
-			if err != nil {
-				logutil.BgLogger().Warn("dump stats delta failed", zap.Error(err))
-			}
 		case <-gcStatsTicker.C:
 			if !do.statsOwner.IsOwner() {
 				continue
@@ -2642,11 +2639,6 @@ func (do *Domain) updateStatsWorker(_ sessionctx.Context) {
 				logutil.BgLogger().Warn("GC stats failed", zap.Error(err))
 			}
 			do.CheckAutoAnalyzeWindows()
-		case <-dumpColStatsUsageTicker.C:
-			err := statsHandle.DumpColStatsUsageToKV()
-			if err != nil {
-				logutil.BgLogger().Warn("dump column stats usage failed", zap.Error(err))
-			}
 		case <-readMemTicker.C:
 			memory.ForceReadMemStats()
 			do.StatsHandle().StatsCache.TriggerEvict()
@@ -2656,6 +2648,52 @@ func (do *Domain) updateStatsWorker(_ sessionctx.Context) {
 	}
 }
 
+func (do *Domain) dumpColStatsUsageWorker() {
+	logutil.BgLogger().Info("dumpColStatsUsageWorker started.")
+	lease := do.statsLease
+	dumpColStatsUsageTicker := time.NewTicker(100 * lease)
+	statsHandle := do.StatsHandle()
+	defer func() {
+		dumpColStatsUsageTicker.Stop()
+		logutil.BgLogger().Info("dumpColStatsUsageWorker exited.")
+	}()
+	defer util.Recover(metrics.LabelDomain, "dumpColStatsUsageWorker", nil, false)
+
+	for {
+		select {
+		case <-do.exit:
+			return
+		case <-dumpColStatsUsageTicker.C:
+			err := statsHandle.DumpColStatsUsageToKV()
+			if err != nil {
+				logutil.BgLogger().Warn("dump column stats usage failed", zap.Error(err))
+			}
+		}
+	}
+}
+
+func (do *Domain) deltaUpdateTickerWorker() {
+	defer util.Recover(metrics.LabelDomain, "deltaUpdateTickerWorker", nil, false)
+	logutil.BgLogger().Info("deltaUpdateTickerWorker started.")
+	lease := do.statsLease
+	// We need to have different nodes trigger tasks at different times to avoid the herd effect.
+	randDuration := time.Duration(rand.Int63n(int64(time.Minute)))
+	deltaUpdateTicker := time.NewTicker(20*lease + randDuration)
+	statsHandle := do.StatsHandle()
+	for {
+		select {
+		case <-do.exit:
+			do.deltaUpdateTickerWorkerExitPreprocessing(statsHandle)
+			return
+		case <-deltaUpdateTicker.C:
+			err := statsHandle.DumpStatsDeltaToKV(false)
+			if err != nil {
+				logutil.BgLogger().Warn("dump stats delta failed", zap.Error(err))
+			}
+		}
+	}
+}
+
 func (do *Domain) autoAnalyzeWorker() {
 	defer util.Recover(metrics.LabelDomain, "autoAnalyzeWorker", nil, false)
 	statsHandle := do.StatsHandle()
diff --git a/pkg/executor/BUILD.bazel b/pkg/executor/BUILD.bazel
index 3b65218307..ffe77ba739 100644
--- a/pkg/executor/BUILD.bazel
+++ b/pkg/executor/BUILD.bazel
@@ -475,7 +475,6 @@ go_test(
         "//pkg/util/sqlexec",
         "//pkg/util/sqlkiller",
         "//pkg/util/stmtsummary/v2:stmtsummary",
-        "//pkg/util/stringutil",
         "//pkg/util/syncutil",
         "//pkg/util/tableutil",
         "//pkg/util/topsql/state",
diff --git a/pkg/infoschema/tables.go b/pkg/infoschema/tables.go
index 12f295231d..dd0ce00fb1 100644
--- a/pkg/infoschema/tables.go
+++ b/pkg/infoschema/tables.go
@@ -1098,7 +1098,6 @@ var tableTiDBServersInfoCols = []columnInfo{
 	{name: "LEASE", tp: mysql.TypeVarchar, size: 64},
 	{name: "VERSION", tp: mysql.TypeVarchar, size: 64},
 	{name: "GIT_HASH", tp: mysql.TypeVarchar, size: 64},
-	{name: "BINLOG_STATUS", tp: mysql.TypeVarchar, size: 64},
 	{name: "LABELS", tp: mysql.TypeVarchar, size: 128},
 }
 
diff --git a/pkg/planner/cardinality/BUILD.bazel b/pkg/planner/cardinality/BUILD.bazel
index e478223ec4..bebf6b8dcc 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 = 29,
+    shard_count = 30,
     deps = [
         "//pkg/config",
         "//pkg/domain",
diff --git a/pkg/planner/core/find_best_task.go b/pkg/planner/core/find_best_task.go
index e090d3ff6d..796fba6b39 100644
--- a/pkg/planner/core/find_best_task.go
+++ b/pkg/planner/core/find_best_task.go
@@ -711,13 +711,37 @@ func compareGlobalIndex(lhs, rhs *candidatePath) int {
 
 // compareCandidates is the core of skyline pruning, which is used to decide which candidate path is better.
 // The return value is 1 if lhs is better, -1 if rhs is better, 0 if they are equivalent or not comparable.
-func compareCandidates(sctx base.PlanContext, prop *property.PhysicalProperty, lhs, rhs *candidatePath) int {
+func compareCandidates(sctx base.PlanContext, statsTbl *statistics.Table, prop *property.PhysicalProperty, lhs, rhs *candidatePath) int {
 	// Due to #50125, full scan on MVIndex has been disabled, so MVIndex path might lead to 'can't find a proper plan' error at the end.
 	// Avoid MVIndex path to exclude all other paths and leading to 'can't find a proper plan' error, see #49438 for an example.
 	if isMVIndexPath(lhs.path) || isMVIndexPath(rhs.path) {
 		return 0
 	}
 
+	// If one index has statistics and the other does not, choose the index with statistics if it
+	// has the same or higher number of equal/IN predicates.
+	lhsHasStatistics := statsTbl.Pseudo
+	if statsTbl != nil && lhs.path.Index != nil {
+		lhsHasStatistics = statsTbl.ColAndIdxExistenceMap.HasAnalyzed(lhs.path.Index.ID, true)
+	}
+	rhsHasStatistics := statsTbl.Pseudo
+	if statsTbl != nil && rhs.path.Index != nil {
+		rhsHasStatistics = statsTbl.ColAndIdxExistenceMap.HasAnalyzed(rhs.path.Index.ID, true)
+	}
+	if !lhs.path.IsTablePath() && !rhs.path.IsTablePath() && // Not a table scan
+		(lhsHasStatistics || rhsHasStatistics) && // At least one index has statistics
+		(!lhsHasStatistics || !rhsHasStatistics) && // At least one index doesn't have statistics
+		len(lhs.path.PartialIndexPaths) == 0 && len(rhs.path.PartialIndexPaths) == 0 { // not IndexMerge due to unreliability
+		lhsTotalEqual := lhs.path.EqCondCount + lhs.path.EqOrInCondCount
+		rhsTotalEqual := rhs.path.EqCondCount + rhs.path.EqOrInCondCount
+		if lhsHasStatistics && lhsTotalEqual > 0 && lhsTotalEqual >= rhsTotalEqual {
+			return 1
+		}
+		if rhsHasStatistics && rhsTotalEqual > 0 && rhsTotalEqual >= lhsTotalEqual {
+			return -1
+		}
+	}
+
 	// This rule is empirical but not always correct.
 	// If x's range row count is significantly lower than y's, for example, 1000 times, we think x is better.
 	if lhs.path.CountAfterAccess > 100 && rhs.path.CountAfterAccess > 100 && // to prevent some extreme cases, e.g. 0.01 : 10
@@ -1162,7 +1186,7 @@ func skylinePruning(ds *logicalop.DataSource, prop *property.PhysicalProperty) [
 			if candidates[i].path.StoreType == kv.TiFlash {
 				continue
 			}
-			result := compareCandidates(ds.SCtx(), prop, candidates[i], currentCandidate)
+			result := compareCandidates(ds.SCtx(), ds.StatisticTable, prop, candidates[i], currentCandidate)
 			if result == 1 {
 				pruned = true
 				// We can break here because the current candidate cannot prune others anymore.
diff --git a/pkg/planner/core/plan_cost_ver2.go b/pkg/planner/core/plan_cost_ver2.go
index f57e319d0c..2939c74001 100644
--- a/pkg/planner/core/plan_cost_ver2.go
+++ b/pkg/planner/core/plan_cost_ver2.go
@@ -262,7 +262,7 @@ func (p *PhysicalIndexLookUpReader) GetPlanCostVer2(taskType property.TaskType,
 	}
 
 	indexRows := getCardinality(p.indexPlan, option.CostFlag)
-	tableRows := getCardinality(p.indexPlan, option.CostFlag)
+	tableRows := getCardinality(p.tablePlan, option.CostFlag)
 	indexRowSize := cardinality.GetAvgRowSize(p.SCtx(), getTblStats(p.indexPlan), p.indexPlan.Schema().Columns, true, false)
 	tableRowSize := cardinality.GetAvgRowSize(p.SCtx(), getTblStats(p.tablePlan), p.tablePlan.Schema().Columns, false, false)
 	cpuFactor := getTaskCPUFactorVer2(p, taskType)
diff --git a/pkg/statistics/handle/storage/BUILD.bazel b/pkg/statistics/handle/storage/BUILD.bazel
index 288b8aa255..f6b9e78166 100644
--- a/pkg/statistics/handle/storage/BUILD.bazel
+++ b/pkg/statistics/handle/storage/BUILD.bazel
@@ -58,7 +58,7 @@ go_test(
         "stats_read_writer_test.go",
     ],
     flaky = True,
-    shard_count = 23,
+    shard_count = 24,
     deps = [
         ":storage",
         "//pkg/domain",
@@ -75,6 +75,7 @@ go_test(
         "//pkg/testkit/analyzehelper",
         "//pkg/types",
         "//pkg/util",
+        "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_stretchr_testify//require",
     ],
 )
diff --git a/pkg/statistics/handle/storage/gc.go b/pkg/statistics/handle/storage/gc.go
index 0d12327857..9b42a224ff 100644
--- a/pkg/statistics/handle/storage/gc.go
+++ b/pkg/statistics/handle/storage/gc.go
@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/parser/terror"
 	"github.com/pingcap/tidb/pkg/sessionctx"
@@ -91,6 +92,10 @@ func GCStats(
 		return nil
 	}
 
+	failpoint.Inject("injectGCStatsLastTSOffset", func(val failpoint.Value) {
+		offset = uint64(val.(int))
+	})
+
 	// Get the last gc time.
 	gcVer := now - offset
 	lastGC, err := getLastGCTimestamp(sctx)
@@ -266,29 +271,36 @@ func removeDeletedExtendedStats(sctx sessionctx.Context, version uint64) (err er
 }
 
 // gcTableStats GC this table's stats.
+// The GC of a table will be a two-phase process:
+// 1. Delete the column/index's stats from storage. Then other TiDB nodes will be aware that those stats are deleted.
+// 2. Then delete the record in stats_meta.
 func gcTableStats(sctx sessionctx.Context,
 	statsHandler types.StatsHandle,
 	is infoschema.InfoSchema, physicalID int64) error {
+	tbl, ok := statsHandler.TableInfoByID(is, physicalID)
 	rows, _, err := util.ExecRows(sctx, "select is_index, hist_id from mysql.stats_histograms where table_id = %?", physicalID)
 	if err != nil {
 		return errors.Trace(err)
 	}
-	// The table has already been deleted in stats and acknowledged to all tidb,
-	// we can safely remove the meta info now.
-	if len(rows) == 0 {
+	if !ok {
+		if len(rows) > 0 {
+			// It's the first time to run into it. Delete column/index stats to notify other TiDB nodes.
+			logutil.BgLogger().Info("remove stats in GC due to dropped table", zap.Int64("tableID", physicalID))
+			return util.WrapTxn(sctx, func(sctx sessionctx.Context) error {
+				return errors.Trace(DeleteTableStatsFromKV(sctx, []int64{physicalID}))
+			})
+		}
+		// len(rows) == 0 => The table's stats is empty.
+		// The table has already been deleted in stats and acknowledged to all tidb,
+		// We can safely remove the meta info now.
 		_, _, err = util.ExecRows(sctx, "delete from mysql.stats_meta where table_id = %?", physicalID)
 		if err != nil {
 			return errors.Trace(err)
 		}
 		cache.TableRowStatsCache.Invalidate(physicalID)
+		return nil
 	}
-	tbl, ok := statsHandler.TableInfoByID(is, physicalID)
-	if !ok {
-		logutil.BgLogger().Info("remove stats in GC due to dropped table", zap.Int64("table_id", physicalID))
-		return util.WrapTxn(sctx, func(sctx sessionctx.Context) error {
-			return errors.Trace(DeleteTableStatsFromKV(sctx, []int64{physicalID}))
-		})
-	}
+
 	tblInfo := tbl.Meta()
 	for _, row := range rows {
 		isIndex, histID := row.GetInt64(0), row.GetInt64(1)


Test patch:
diff --git a/build/image/parser_test b/build/image/parser_test
index c419500b65..dd4218a4a8 100644
--- a/build/image/parser_test
+++ b/build/image/parser_test
@@ -14,7 +14,7 @@
 
 FROM rockylinux:9
 
-ENV GOLANG_VERSION 1.23.3
+ENV GOLANG_VERSION 1.23.6
 ENV ARCH amd64
 ENV GOLANG_DOWNLOAD_URL https://dl.google.com/go/go$GOLANG_VERSION.linux-$ARCH.tar.gz
 ENV GOPATH /home/prow/go
diff --git a/pkg/ddl/notifier/store_test.go b/pkg/ddl/notifier/store_test.go
new file mode 100644
index 0000000000..69e13945cd
--- /dev/null
+++ b/pkg/ddl/notifier/store_test.go
@@ -0,0 +1,80 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package notifier
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/pingcap/tidb/pkg/meta/model"
+	pmodel "github.com/pingcap/tidb/pkg/parser/model"
+	"github.com/pingcap/tidb/pkg/types"
+	"github.com/pingcap/tidb/pkg/util/chunk"
+	"github.com/stretchr/testify/require"
+)
+
+func TestLeftoverWhenUnmarshal(t *testing.T) {
+	r := &listResult{}
+	changesReused := []*SchemaChange{
+		{event: &SchemaChangeEvent{inner: &jsonSchemaChangeEvent{
+			TableInfo: &model.TableInfo{
+				Name:    pmodel.NewCIStr("old"),
+				Columns: []*model.ColumnInfo{{Name: pmodel.NewCIStr("c1")}},
+				Indices: []*model.IndexInfo{
+					{Name: pmodel.NewCIStr("i1")},
+					{Name: pmodel.NewCIStr("i2")},
+					{Name: pmodel.NewCIStr("i3")},
+				},
+			},
+		}}},
+		{event: &SchemaChangeEvent{inner: &jsonSchemaChangeEvent{
+			AddedPartInfo: &model.PartitionInfo{Expr: "test"},
+		}}},
+		nil,
+	}
+
+	newTableInfo := &model.TableInfo{
+		Name: pmodel.NewCIStr("new"),
+		Columns: []*model.ColumnInfo{
+			{Name: pmodel.NewCIStr("c2")},
+			{Name: pmodel.NewCIStr("c3")},
+		},
+		Indices: []*model.IndexInfo{
+			{Name: pmodel.NewCIStr("i4")},
+		},
+		Constraints: []*model.ConstraintInfo{
+			{Name: pmodel.NewCIStr("c1")},
+		},
+	}
+
+	newTableInfoJSON, err := json.Marshal(jsonSchemaChangeEvent{TableInfo: newTableInfo})
+	require.NoError(t, err)
+	sameRow := chunk.MutRowFromDatums([]types.Datum{
+		types.NewIntDatum(1), types.NewIntDatum(1),
+		types.NewBytesDatum(newTableInfoJSON), types.NewUintDatum(0),
+	}).ToRow()
+	rows := []chunk.Row{sameRow, sameRow, sameRow}
+
+	err = r.unmarshalSchemaChanges(rows, changesReused)
+	require.NoError(t, err)
+
+	require.Equal(t, newTableInfo, changesReused[0].event.inner.TableInfo)
+	require.Equal(t, newTableInfo, changesReused[1].event.inner.TableInfo)
+	// The leftover will not be cleaned right after unmarshal. It will be cleaned be
+	// GC later. Because we use type field to determine read which field, so the
+	// leftover will not affect the result.
+	require.NotNil(t, changesReused[1].event.inner.AddedPartInfo)
+	require.Equal(t, newTableInfo, changesReused[2].event.inner.TableInfo)
+}
diff --git a/pkg/ddl/notifier/testkit_test.go b/pkg/ddl/notifier/testkit_test.go
index aa3ee81696..9b0e4ccc0a 100644
--- a/pkg/ddl/notifier/testkit_test.go
+++ b/pkg/ddl/notifier/testkit_test.go
@@ -37,6 +37,7 @@ import (
 	"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
 	"github.com/pingcap/tidb/pkg/util"
 	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
 )
 
 func TestPublishToTableStore(t *testing.T) {
@@ -55,9 +56,12 @@ func TestPublishToTableStore(t *testing.T) {
 	event2 := notifier.NewDropTableEvent(&model.TableInfo{ID: 1001, Name: pmodel.NewCIStr("t2")})
 	err = notifier.PubSchemeChangeToStore(ctx, se, 2, -1, event2, s)
 	require.NoError(t, err)
-	got, err := s.List(ctx, se)
+	changes := make([]*notifier.SchemaChange, 8)
+	result, closeFn := s.List(ctx, se)
+	n, err := result.Read(changes)
 	require.NoError(t, err)
-	require.Len(t, got, 2)
+	require.Equal(t, 2, n)
+	closeFn()
 }
 
 func TestBasicPubSub(t *testing.T) {
@@ -69,9 +73,9 @@ func TestBasicPubSub(t *testing.T) {
 
 	s := notifier.OpenTableStore("test", ddl.NotifierTableName)
 	sessionPool := util.NewSessionPool(
-		1,
+		2,
 		func() (pools.Resource, error) {
-			return tk.Session(), nil
+			return testkit.NewTestKit(t, store).Session(), nil
 		},
 		nil,
 		nil,
@@ -104,12 +108,7 @@ func TestBasicPubSub(t *testing.T) {
 		return nil
 	}
 	n.RegisterHandler(notifier.TestHandlerID, testHandler)
-
-	done := make(chan struct{})
-	go func() {
-		n.OnBecomeOwner()
-		close(done)
-	}()
+	n.OnBecomeOwner()
 
 	tk2 := testkit.NewTestKit(t, store)
 	se := sess.NewSession(tk2.Session())
@@ -134,7 +133,6 @@ func TestBasicPubSub(t *testing.T) {
 	require.Equal(t, event2, seenChanges[1])
 	require.Equal(t, event3, seenChanges[2])
 	n.OnRetireOwner()
-	<-done
 }
 
 func TestDeliverOrderAndCleanup(t *testing.T) {
@@ -182,12 +180,7 @@ func TestDeliverOrderAndCleanup(t *testing.T) {
 	n.RegisterHandler(3, h1)
 	n.RegisterHandler(4, h2)
 	n.RegisterHandler(9, h3)
-
-	done := make(chan struct{})
-	go func() {
-		n.OnBecomeOwner()
-		close(done)
-	}()
+	n.OnBecomeOwner()
 
 	tk2 := testkit.NewTestKit(t, store)
 	se := sess.NewSession(tk2.Session())
@@ -203,9 +196,12 @@ func TestDeliverOrderAndCleanup(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
-		changes, err2 := s.List(ctx, se)
+		changes := make([]*notifier.SchemaChange, 8)
+		result, closeFn := s.List(ctx, se)
+		count, err2 := result.Read(changes)
 		require.NoError(t, err2)
-		return len(changes) == 0
+		closeFn()
+		return count == 0
 	}, time.Second, 50*time.Millisecond)
 
 	require.Equal(t, []int64{1000, 1001, 1002}, *id1)
@@ -213,16 +209,15 @@ func TestDeliverOrderAndCleanup(t *testing.T) {
 	require.Equal(t, []int64{1000, 1001, 1002}, *id3)
 
 	n.OnRetireOwner()
-	<-done
 }
 
 func TestPubSub(t *testing.T) {
-	events := make([]*notifier.SchemaChangeEvent, 0, 32)
-	eventsLock := sync.Mutex{}
+	tps := make([]model.ActionType, 0, 32)
+	tpsLock := sync.Mutex{}
 	handler := func(_ context.Context, _ sessionctx.Context, c *notifier.SchemaChangeEvent) error {
-		eventsLock.Lock()
-		defer eventsLock.Unlock()
-		events = append(events, c)
+		tpsLock.Lock()
+		defer tpsLock.Unlock()
+		tps = append(tps, c.GetType())
 		return nil
 	}
 	testfailpoint.EnableCall(
@@ -257,15 +252,11 @@ func TestPubSub(t *testing.T) {
 	tk.MustExec("drop database test")                                                                    // ActionDropSchema
 
 	require.Eventually(t, func() bool {
-		eventsLock.Lock()
-		defer eventsLock.Unlock()
-		return len(events) == 18
+		tpsLock.Lock()
+		defer tpsLock.Unlock()
+		return len(tps) == 18
 	}, 5*time.Second, 500*time.Millisecond)
 
-	tps := make([]model.ActionType, len(events))
-	for i, event := range events {
-		tps[i] = event.GetType()
-	}
 	require.Equal(t, []model.ActionType{
 		model.ActionCreateTable,
 		model.ActionAlterTablePartitioning,
@@ -349,12 +340,7 @@ func Test2OwnerForAShortTime(t *testing.T) {
 		return nil
 	}
 	n.RegisterHandler(notifier.TestHandlerID, testHandler)
-
-	done := make(chan struct{})
-	go func() {
-		n.OnBecomeOwner()
-		close(done)
-	}()
+	n.OnBecomeOwner()
 
 	tk2 := testkit.NewTestKit(t, store)
 	se := sess.NewSession(tk2.Session())
@@ -380,5 +366,126 @@ func Test2OwnerForAShortTime(t *testing.T) {
 	tk2.MustQuery("SELECT * FROM test.result").Check(testkit.Rows())
 
 	n.OnRetireOwner()
-	<-done
+}
+
+func TestPaginatedList(t *testing.T) {
+	backup := notifier.ProcessEventsBatchSize
+	notifier.ProcessEventsBatchSize = 3
+	t.Cleanup(func() {
+		notifier.ProcessEventsBatchSize = backup
+	})
+
+	names := make([]string, 0, 32)
+	namesLock := sync.Mutex{}
+	handler := func(_ context.Context, _ sessionctx.Context, c *notifier.SchemaChangeEvent) error {
+		namesLock.Lock()
+		defer namesLock.Unlock()
+		switch c.GetType() {
+		case model.ActionCreateTable:
+			names = append(names, c.GetCreateTableInfo().Name.O)
+		case model.ActionAddColumn:
+			_, colInfo := c.GetAddColumnInfo()
+			names = append(names, colInfo[0].Name.O)
+		default:
+			t.Fatalf("unexpected event type: %s", c.GetType().String())
+		}
+		return nil
+	}
+
+	blocking := atomic.NewBool(true)
+	count := atomic.NewInt32(0)
+	blockingHandler := func(context.Context, sessionctx.Context, *notifier.SchemaChangeEvent) error {
+		if blocking.Load() {
+			return notifier.ErrNotReadyRetryLater
+		}
+		count.Inc()
+		return nil
+	}
+
+	testfailpoint.EnableCall(
+		t,
+		"github.com/pingcap/tidb/pkg/domain/afterDDLNotifierCreated",
+		func(registry *notifier.DDLNotifier) {
+			registry.RegisterHandler(notifier.TestHandlerID, handler)
+			registry.RegisterHandler(10, blockingHandler)
+		},
+	)
+
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("USE test")
+	tk.MustExec("create table t1 (a int)")
+	tk.MustExec("create table t2 (a int)")
+	tk.MustExec("create table t3 (a int)")
+	tk.MustExec("create table t4 (a int)")
+	tk.MustExec("alter table t1 add column c5 int, add column c6 int, add column c7 int, add column c8 int")
+
+	require.Eventually(t, func() bool {
+		namesLock.Lock()
+		defer namesLock.Unlock()
+		return len(names) == 8
+	}, 5*time.Second, 500*time.Millisecond)
+
+	require.Equal(t, []string{"t1", "t2", "t3", "t4", "c5", "c6", "c7", "c8"}, names)
+
+	blocking.Store(false)
+	require.Eventually(t, func() bool {
+		return count.Load() == 8
+	}, 5*time.Second, 500*time.Millisecond)
+}
+
+func TestBeginTwice(t *testing.T) {
+	conf := new(log.Config)
+	logFilename := path.Join(t.TempDir(), "/testBeginTwice.log")
+	conf.File.Filename = logFilename
+	lg, p, e := log.InitLogger(conf)
+	require.NoError(t, e)
+	rs := log.ReplaceGlobals(lg, p)
+	defer rs()
+
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("USE test")
+	tk.MustExec("DROP TABLE IF EXISTS " + ddl.NotifierTableName)
+	tk.MustExec(ddl.NotifierTableSQL)
+
+	s := notifier.OpenTableStore("test", ddl.NotifierTableName)
+	sessionPool := util.NewSessionPool(
+		5,
+		func() (pools.Resource, error) {
+			return testkit.NewTestKit(t, store).Session(), nil
+		},
+		nil,
+		nil,
+	)
+
+	n := notifier.NewDDLNotifier(sessionPool, s, 50*time.Millisecond)
+
+	testHandler := func(context.Context, sessionctx.Context, *notifier.SchemaChangeEvent) error {
+		return nil
+	}
+	n.RegisterHandler(notifier.TestHandlerID, testHandler)
+	n.OnBecomeOwner()
+
+	tk2 := testkit.NewTestKit(t, store)
+	se := sess.NewSession(tk2.Session())
+	ctx := context.Background()
+	event1 := notifier.NewCreateTableEvent(&model.TableInfo{ID: 1000, Name: pmodel.NewCIStr("t1")})
+	err := notifier.PubSchemeChangeToStore(ctx, se, 1, -1, event1, s)
+	require.NoError(t, err)
+
+	// after handler processed the event, wait to ensure the record is deleted by DDL notifier
+	require.Eventually(t, func() bool {
+		changes := make([]*notifier.SchemaChange, 8)
+		result, closeFn := s.List(ctx, se)
+		count, err2 := result.Read(changes)
+		require.NoError(t, err2)
+		closeFn()
+		return count == 0
+	}, time.Second, 50*time.Millisecond)
+
+	content, err := os.ReadFile(logFilename)
+	require.NoError(t, err)
+	require.NotContains(t, string(content), "context provider not set")
 }
diff --git a/pkg/domain/domain_test.go b/pkg/domain/domain_test.go
index 92eff2e83d..ee7f7a1cbb 100644
--- a/pkg/domain/domain_test.go
+++ b/pkg/domain/domain_test.go
@@ -181,7 +181,7 @@ func TestStatWorkRecoverFromPanic(t *testing.T) {
 	metrics.PanicCounter.Reset()
 	// Since the stats lease is 0 now, so create a new ticker will panic.
 	// Test that they can recover from panic correctly.
-	dom.updateStatsWorker(mock.NewContext())
+	dom.gcStatsWorker()
 	dom.autoAnalyzeWorker()
 	counter := metrics.PanicCounter.WithLabelValues(metrics.LabelDomain)
 	pb := &dto.Metric{}
diff --git a/pkg/executor/infoschema_reader_test.go b/pkg/executor/infoschema_reader_test.go
index 704ab1a8a0..14150542d4 100644
--- a/pkg/executor/infoschema_reader_test.go
+++ b/pkg/executor/infoschema_reader_test.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/kvproto/pkg/kvrpcpb"
+	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/domain/infosync"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/parser/auth"
@@ -35,7 +36,6 @@ import (
 	"github.com/pingcap/tidb/pkg/store/mockstore"
 	"github.com/pingcap/tidb/pkg/testkit"
 	"github.com/pingcap/tidb/pkg/testkit/testfailpoint"
-	"github.com/pingcap/tidb/pkg/util/stringutil"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/tikv"
 	"github.com/tikv/client-go/v2/tikvrpc"
@@ -368,10 +368,19 @@ func TestForAnalyzeStatus(t *testing.T) {
 }
 
 func TestForServersInfo(t *testing.T) {
+	globalCfg := config.GetGlobalConfig()
+	t.Cleanup(func() {
+		config.StoreGlobalConfig(globalCfg)
+	})
+	newCfg := *globalCfg
+	newCfg.Labels = map[string]string{"dc": "dc1"}
+	config.StoreGlobalConfig(&newCfg)
+
 	store := testkit.CreateMockStore(t)
 	tk := testkit.NewTestKit(t, store)
 	rows := tk.MustQuery("select * from information_schema.TIDB_SERVERS_INFO").Rows()
 	require.Len(t, rows, 1)
+	require.Len(t, rows[0], 8)
 
 	info, err := infosync.GetServerInfo()
 	require.NoError(t, err)
@@ -383,7 +392,7 @@ func TestForServersInfo(t *testing.T) {
 	require.Equal(t, info.Lease, rows[0][4])
 	require.Equal(t, info.Version, rows[0][5])
 	require.Equal(t, info.GitHash, rows[0][6])
-	require.Equal(t, stringutil.BuildStringFromLabels(info.Labels), rows[0][8])
+	require.Equal(t, "dc=dc1", rows[0][7])
 }
 
 func TestTiFlashSystemTableWithTiFlashV620(t *testing.T) {
@@ -628,7 +637,7 @@ func TestColumnTable(t *testing.T) {
 		testkit.RowsWithSep("|",
 			"test|tbl1|col_2"))
 	tk.MustQuery(`select count(*) from information_schema.columns;`).Check(
-		testkit.RowsWithSep("|", "4983"))
+		testkit.RowsWithSep("|", "4982"))
 }
 
 func TestIndexUsageTable(t *testing.T) {
diff --git a/pkg/planner/cardinality/selectivity_test.go b/pkg/planner/cardinality/selectivity_test.go
index 6966bad0de..627a3e084f 100644
--- a/pkg/planner/cardinality/selectivity_test.go
+++ b/pkg/planner/cardinality/selectivity_test.go
@@ -341,6 +341,28 @@ func TestEstimationForUnknownValuesAfterModify(t *testing.T) {
 	require.Truef(t, count > 9, "expected: between 10 to 40, got: %v", count)
 }
 
+func TestNewIndexWithoutStats(t *testing.T) {
+	store, _ := testkit.CreateMockStoreAndDomain(t)
+	testKit := testkit.NewTestKit(t, store)
+	testKit.MustExec("use test")
+	testKit.MustExec("drop table if exists t")
+	testKit.MustExec("create table t(a int, b int, c int, index idxa(a))")
+	testKit.MustExec("set @@tidb_analyze_version=2")
+	testKit.MustExec("set @@global.tidb_enable_auto_analyze='OFF'")
+	testKit.MustExec("insert into t values (1, 1, 1)")
+	testKit.MustExec("insert into t select mod(a,250), mod(a,10), mod(a,100) from (with recursive x as (select 1 as a union all select a + 1 AS a from x where a < 500) select a from x) as subquery")
+	testKit.MustExec("analyze table t")
+	testKit.MustExec("create index idxb on t(b)")
+	// Create index after ANALYZE. SkyLine pruning should ensure that idxa is chosen because it has statistics
+	testKit.MustQuery("explain format='brief' select * from t where a = 5 and b = 5").CheckContain("idxa(a)")
+	testKit.MustExec("analyze table t")
+	// idxa should still win after statistics
+	testKit.MustQuery("explain format='brief' select * from t where a = 5 and b = 5").CheckContain("idxa(a)")
+	testKit.MustExec("create index idxab on t(a, b)")
+	// New index idxab should win due to having the most matching equal predicates - regardless of no statistics
+	testKit.MustQuery("explain format='brief' select * from t where a = 5 and b = 5").CheckContain("idxab(a, b)")
+}
+
 func TestEstimationUniqueKeyEqualConds(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 	testKit := testkit.NewTestKit(t, store)
diff --git a/pkg/server/main_test.go b/pkg/server/main_test.go
index 0d74b129a2..3520e89d83 100644
--- a/pkg/server/main_test.go
+++ b/pkg/server/main_test.go
@@ -72,6 +72,8 @@ func TestMain(m *testing.M) {
 		goleak.IgnoreTopFunction("go.opencensus.io/stats/view.(*worker).start"),
 		goleak.IgnoreTopFunction("go.etcd.io/etcd/client/pkg/v3/logutil.(*MergeLogger).outputLoop"),
 		goleak.IgnoreTopFunction("github.com/go-sql-driver/mysql.(*mysqlConn).startWatcher.func1"),
+		goleak.IgnoreTopFunction("github.com/pingcap/tidb/pkg/util/timeutil.Sleep"),
+		goleak.IgnoreTopFunction("github.com/pingcap/tidb/pkg/ddl/notifier.(*DDLNotifier).start"),
 	}
 
 	callback := func(i int) int {
diff --git a/pkg/statistics/handle/storage/gc_test.go b/pkg/statistics/handle/storage/gc_test.go
index 0a79f254b2..b82e5dceb8 100644
--- a/pkg/statistics/handle/storage/gc_test.go
+++ b/pkg/statistics/handle/storage/gc_test.go
@@ -15,9 +15,12 @@
 package storage_test
 
 import (
+	"context"
 	"testing"
 	"time"
 
+	"github.com/pingcap/failpoint"
+	"github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/sessionctx/variable"
 	statstestutil "github.com/pingcap/tidb/pkg/statistics/handle/ddl/testutil"
 	"github.com/pingcap/tidb/pkg/testkit"
@@ -177,3 +180,27 @@ func TestDeleteAnalyzeJobs(t *testing.T) {
 	rows = testKit.MustQuery("show analyze status").Rows()
 	require.Equal(t, 0, len(rows))
 }
+
+func TestExtremCaseOfGC(t *testing.T) {
+	// This case tests that there's no records in mysql.stats_histograms but this table is not deleted in fact.
+	// We should not delete the record in mysql.stats_meta.
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	testKit := testkit.NewTestKit(t, store)
+	testKit.MustExec("use test")
+	testKit.MustExec("create table t(a int, b int)")
+	testKit.MustExec("insert into t values (1,2),(3,4)")
+	testKit.MustExec("analyze table t")
+	tbl, err := dom.InfoSchema().TableByName(context.TODO(), model.NewCIStr("test"), model.NewCIStr("t"))
+	require.NoError(t, err)
+	tid := tbl.Meta().ID
+	rs := testKit.MustQuery("select * from mysql.stats_meta where table_id = ?", tid)
+	require.Len(t, rs.Rows(), 1)
+	rs = testKit.MustQuery("select * from mysql.stats_histograms where table_id = ?", tid)
+	require.Len(t, rs.Rows(), 0)
+	h := dom.StatsHandle()
+	failpoint.Enable("github.com/pingcap/tidb/pkg/statistics/handle/storage/injectGCStatsLastTSOffset", `return(0)`)
+	h.GCStats(dom.InfoSchema(), time.Second*3)
+	rs = testKit.MustQuery("select * from mysql.stats_meta where table_id = ?", tid)
+	require.Len(t, rs.Rows(), 1)
+	failpoint.Disable("github.com/pingcap/tidb/pkg/statistics/handle/storage/injectGCStatsLastTSOffset")
+}
diff --git a/tests/integrationtest/r/planner/core/casetest/integration.result b/tests/integrationtest/r/planner/core/casetest/integration.result
index ae70cbcea4..2c4455d3e9 100644
--- a/tests/integrationtest/r/planner/core/casetest/integration.result
+++ b/tests/integrationtest/r/planner/core/casetest/integration.result
@@ -1115,7 +1115,7 @@ Level	Code	Message
 Note	1105	[f,f_g] remain after pruning paths for t given Prop{SortItems: [], TaskTp: rootTask}
 explain format = 'verbose' select * from t where f > 3 and g = 5;
 id	estRows	estCost	task	access object	operator info
-IndexLookUp_12	3.33	19551.99	root		
+IndexLookUp_12	3.33	19531.97	root		
 ├─IndexRangeScan_9(Build)	10.00	2035.00	cop[tikv]	table:t, index:g(g)	range:[5,5], keep order:false, stats:pseudo
 └─Selection_11(Probe)	3.33	3269.59	cop[tikv]		gt(planner__core__casetest__integration.t.f, 3)
   └─TableRowIDScan_10	10.00	2770.59	cop[tikv]	table:t	keep order:false, stats:pseudo