test: build a failing example with LogicalPlanBuilder

Signed-off-by: discord9 <discord9@163.com>
test: reproduce the panic, still no clue why
2025-12-27 16:32:54 +00:00 · 2025-08-12 15:19:47 +08:00 · 2025-08-11 19:37:48 +08:00 · 2025-08-07 19:55:32 +08:00 · 2025-08-07 06:17:52 +00:00 · 2025-08-07 06:16:23 +00:00
34 changed files with 1612 additions and 422 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -217,7 +217,7 @@ checksum = "d301b3b94cb4b2f23d7917810addbbaff90738e0ca2be692bd027e70d7e0330c"

 [[package]]
 name = "api"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "common-base",
 "common-decimal",
@@ -950,7 +950,7 @@ dependencies = [

 [[package]]
 name = "auth"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-trait",
@@ -1617,7 +1617,7 @@ dependencies = [

 [[package]]
 name = "cache"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "catalog",
 "common-error",
@@ -1652,7 +1652,7 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"

 [[package]]
 name = "catalog"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "arrow 54.2.1",
@@ -1992,7 +1992,7 @@ checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97"

 [[package]]
 name = "cli"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -2036,7 +2036,7 @@ dependencies = [
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "table",
 "tempfile",
 "tokio",
@@ -2045,7 +2045,7 @@ dependencies = [

 [[package]]
 name = "client"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "arc-swap",
@@ -2075,7 +2075,7 @@ dependencies = [
 "rand 0.9.0",
 "serde_json",
 "snafu 0.8.5",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "substrait 0.37.3",
 "tokio",
 "tokio-stream",
@@ -2116,7 +2116,7 @@ dependencies = [

 [[package]]
 name = "cmd"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-trait",
 "auth",
@@ -2178,7 +2178,7 @@ dependencies = [
 "snafu 0.8.5",
 "stat",
 "store-api",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "table",
 "temp-env",
 "tempfile",
@@ -2225,7 +2225,7 @@ checksum = "55b672471b4e9f9e95499ea597ff64941a309b2cdbffcc46f2cc5e2d971fd335"

 [[package]]
 name = "common-base"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "anymap2",
 "async-trait",
@@ -2247,11 +2247,11 @@ dependencies = [

 [[package]]
 name = "common-catalog"
-version = "0.16.0"
+version = "0.17.0"

 [[package]]
 name = "common-config"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "common-base",
 "common-error",
@@ -2277,7 +2277,7 @@ dependencies = [

 [[package]]
 name = "common-datasource"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "arrow 54.2.1",
 "arrow-schema 54.3.1",
@@ -2314,7 +2314,7 @@ dependencies = [

 [[package]]
 name = "common-decimal"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "bigdecimal 0.4.8",
 "common-error",
@@ -2327,7 +2327,7 @@ dependencies = [

 [[package]]
 name = "common-error"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "common-macro",
 "http 1.1.0",
@@ -2338,7 +2338,7 @@ dependencies = [

 [[package]]
 name = "common-event-recorder"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-trait",
@@ -2357,7 +2357,7 @@ dependencies = [

 [[package]]
 name = "common-frontend"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-trait",
 "common-error",
@@ -2374,7 +2374,7 @@ dependencies = [

 [[package]]
 name = "common-function"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -2433,7 +2433,7 @@ dependencies = [

 [[package]]
 name = "common-greptimedb-telemetry"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-trait",
 "common-runtime",
@@ -2450,7 +2450,7 @@ dependencies = [

 [[package]]
 name = "common-grpc"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -2483,7 +2483,7 @@ dependencies = [

 [[package]]
 name = "common-grpc-expr"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "common-base",
@@ -2503,7 +2503,7 @@ dependencies = [

 [[package]]
 name = "common-macro"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "arc-swap",
 "common-query",
@@ -2517,7 +2517,7 @@ dependencies = [

 [[package]]
 name = "common-mem-prof"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "anyhow",
 "common-error",
@@ -2533,7 +2533,7 @@ dependencies = [

 [[package]]
 name = "common-meta"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "anymap2",
 "api",
@@ -2604,7 +2604,7 @@ dependencies = [

 [[package]]
 name = "common-options"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "common-grpc",
 "humantime-serde",
@@ -2613,11 +2613,11 @@ dependencies = [

 [[package]]
 name = "common-plugins"
-version = "0.16.0"
+version = "0.17.0"

 [[package]]
 name = "common-pprof"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2629,7 +2629,7 @@ dependencies = [

 [[package]]
 name = "common-procedure"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-stream",
@@ -2658,7 +2658,7 @@ dependencies = [

 [[package]]
 name = "common-procedure-test"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-trait",
 "common-procedure",
@@ -2668,7 +2668,7 @@ dependencies = [

 [[package]]
 name = "common-query"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-trait",
@@ -2694,7 +2694,7 @@ dependencies = [

 [[package]]
 name = "common-recordbatch"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "arc-swap",
 "common-error",
@@ -2715,7 +2715,7 @@ dependencies = [

 [[package]]
 name = "common-runtime"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -2745,14 +2745,14 @@ dependencies = [

 [[package]]
 name = "common-session"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "strum 0.27.1",
 ]

 [[package]]
 name = "common-sql"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "common-base",
 "common-datasource",
@@ -2771,7 +2771,7 @@ dependencies = [

 [[package]]
 name = "common-telemetry"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "backtrace",
 "common-error",
@@ -2799,7 +2799,7 @@ dependencies = [

 [[package]]
 name = "common-test-util"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "client",
 "common-grpc",
@@ -2812,7 +2812,7 @@ dependencies = [

 [[package]]
 name = "common-time"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "arrow 54.2.1",
 "chrono",
@@ -2830,7 +2830,7 @@ dependencies = [

 [[package]]
 name = "common-version"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "build-data",
 "cargo-manifest",
@@ -2841,7 +2841,7 @@ dependencies = [

 [[package]]
 name = "common-wal"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "common-base",
 "common-error",
@@ -2864,7 +2864,7 @@ dependencies = [

 [[package]]
 name = "common-workload"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "common-telemetry",
@@ -3863,7 +3863,7 @@ dependencies = [

 [[package]]
 name = "datanode"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -3917,7 +3917,7 @@ dependencies = [
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "table",
 "tokio",
 "toml 0.8.19",
@@ -3927,7 +3927,7 @@ dependencies = [

 [[package]]
 name = "datatypes"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -4602,7 +4602,7 @@ checksum = "e8c02a5121d4ea3eb16a80748c74f5549a5665e4c21333c6098f283870fbdea6"

 [[package]]
 name = "file-engine"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-trait",
@@ -4739,7 +4739,7 @@ checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8"

 [[package]]
 name = "flow"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "arrow 54.2.1",
@@ -4806,7 +4806,7 @@ dependencies = [
 "sql",
 "store-api",
 "strum 0.27.1",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "table",
 "tokio",
 "tonic 0.12.3",
@@ -4861,7 +4861,7 @@ checksum = "6c2141d6d6c8512188a7891b4b01590a45f6dac67afb4f255c4124dbb86d4eaa"

 [[package]]
 name = "frontend"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "arc-swap",
@@ -4922,7 +4922,7 @@ dependencies = [
 "sqlparser 0.54.0-greptime",
 "store-api",
 "strfmt",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "table",
 "tokio",
 "tokio-util",
@@ -6107,7 +6107,7 @@ dependencies = [

 [[package]]
 name = "index"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-trait",
 "asynchronous-codec",
@@ -7033,7 +7033,7 @@ checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"

 [[package]]
 name = "log-query"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "chrono",
 "common-error",
@@ -7045,7 +7045,7 @@ dependencies = [

 [[package]]
 name = "log-store"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -7287,6 +7287,12 @@ version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "490cc448043f947bae3cbee9c203358d62dbee0db12107a74be5c30ccfd09771"

+[[package]]
+name = "md5"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae960838283323069879657ca3de837e9f7bbb4c7bf6ea7f1b290d5e9476d2e0"
+
 [[package]]
 name = "measure_time"
 version = "0.9.0"
@@ -7342,7 +7348,7 @@ dependencies = [

 [[package]]
 name = "meta-client"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-trait",
@@ -7370,7 +7376,7 @@ dependencies = [

 [[package]]
 name = "meta-srv"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-trait",
@@ -7468,7 +7474,7 @@ dependencies = [

 [[package]]
 name = "metric-engine"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -7493,6 +7499,7 @@ dependencies = [
 "lazy_static",
 "mito-codec",
 "mito2",
+ "moka",
 "mur3",
 "object-store",
 "prometheus",
@@ -7560,7 +7567,7 @@ dependencies = [

 [[package]]
 name = "mito-codec"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "bytes",
@@ -7583,7 +7590,7 @@ dependencies = [

 [[package]]
 name = "mito2"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -8336,7 +8343,7 @@ dependencies = [

 [[package]]
 name = "object-store"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "anyhow",
 "bytes",
@@ -8348,7 +8355,7 @@ dependencies = [
 "futures",
 "humantime-serde",
 "lazy_static",
- "md5",
+ "md5 0.7.0",
 "moka",
 "opendal",
 "prometheus",
@@ -8672,7 +8679,7 @@ dependencies = [

 [[package]]
 name = "operator"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -8728,7 +8735,7 @@ dependencies = [
 "sql",
 "sqlparser 0.54.0-greptime",
 "store-api",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "table",
 "tokio",
 "tokio-util",
@@ -8987,7 +8994,7 @@ dependencies = [

 [[package]]
 name = "partition"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-trait",
@@ -9181,8 +9188,9 @@ dependencies = [

 [[package]]
 name = "pgwire"
-version = "0.30.2"
-source = "git+https://github.com/sunng87/pgwire?rev=127573d997228cfb70c7699881c568eae8131270#127573d997228cfb70c7699881c568eae8131270"
+version = "0.32.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "017b8b74f9e8c7aff0087d4ef2b91676a5509e8928100d6a3510fd472210feb5"
 dependencies = [
 "async-trait",
 "bytes",
@@ -9191,7 +9199,7 @@ dependencies = [
 "futures",
 "hex",
 "lazy-regex",
- "md5",
+ "md5 0.8.0",
 "postgres-types",
 "rand 0.9.0",
 "ring",
@@ -9314,7 +9322,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"

 [[package]]
 name = "pipeline"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -9458,7 +9466,7 @@ dependencies = [

 [[package]]
 name = "plugins"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "auth",
 "clap 4.5.19",
@@ -9771,7 +9779,7 @@ dependencies = [

 [[package]]
 name = "promql"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "ahash 0.8.11",
 "async-trait",
@@ -10054,7 +10062,7 @@ dependencies = [

 [[package]]
 name = "puffin"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-compression 0.4.13",
 "async-trait",
@@ -10096,7 +10104,7 @@ dependencies = [

 [[package]]
 name = "query"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -10162,7 +10170,7 @@ dependencies = [
 "sqlparser 0.54.0-greptime",
 "statrs",
 "store-api",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "table",
 "tokio",
 "tokio-stream",
@@ -11515,7 +11523,7 @@ dependencies = [

 [[package]]
 name = "servers"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -11639,7 +11647,7 @@ dependencies = [

 [[package]]
 name = "session"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -11979,7 +11987,7 @@ dependencies = [

 [[package]]
 name = "sql"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "chrono",
@@ -12036,7 +12044,7 @@ dependencies = [

 [[package]]
 name = "sqlness-runner"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -12336,7 +12344,7 @@ dependencies = [

 [[package]]
 name = "stat"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "nix 0.30.1",
 ]
@@ -12362,7 +12370,7 @@ dependencies = [

 [[package]]
 name = "store-api"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -12525,7 +12533,7 @@ dependencies = [

 [[package]]
 name = "substrait"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "async-trait",
 "bytes",
@@ -12726,7 +12734,7 @@ dependencies = [

 [[package]]
 name = "table"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "async-trait",
@@ -12995,7 +13003,7 @@ checksum = "3369f5ac52d5eb6ab48c6b4ffdc8efbcad6b89c765749064ba298f2c68a16a76"

 [[package]]
 name = "tests-fuzz"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "arbitrary",
 "async-trait",
@@ -13039,7 +13047,7 @@ dependencies = [

 [[package]]
 name = "tests-integration"
-version = "0.16.0"
+version = "0.17.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -13109,7 +13117,7 @@ dependencies = [
 "sql",
 "sqlx",
 "store-api",
- "substrait 0.16.0",
+ "substrait 0.17.0",
 "table",
 "tempfile",
 "time",
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -73,7 +73,7 @@ members = [
 resolver = "2"

 [workspace.package]
-version = "0.16.0"
+version = "0.17.0"
 edition = "2021"
 license = "Apache-2.0"

--- a/docs/rfcs/2025-07-23-global-gc-worker.md
+++ b/docs/rfcs/2025-07-23-global-gc-worker.md
@@ -0,0 +1,157 @@
+---
+Feature Name: "global-gc-worker"
+Tracking Issue: https://github.com/GreptimeTeam/greptimedb/issues/6571
+Date: 2025-07-23
+Author: "discord9 <discord9@163.com>"
+---
+
+# Global GC Worker
+
+## Summary
+
+This RFC proposes the integration of a garbage collection (GC) mechanism within the Compaction process. This mechanism aims to manage and remove stale files that are no longer actively used by any system component, thereby reclaiming storage space.
+
+## Motivation
+
+With the introduction of features such as table repartitioning, a substantial number of Parquet files can become obsolete. Furthermore, failures during manifest updates may result in orphaned files that are never referenced by the system. Therefore, a periodic garbage collection mechanism is essential to reclaim storage space by systematically removing these unused files.
+
+## Details
+
+### Overview
+
+The garbage collection process will be integrated directly into the Compaction process. Upon the completion of a Compaction for a given region, the GC worker will be automatically triggered. Its primary function will be to identify and subsequently delete obsolete files that have persisted beyond their designated retention period. This integration ensures that garbage collection is performed in close conjunction with data lifecycle management, effectively leveraging the compaction process's inherent knowledge of file states.
+
+This design prioritizes correctness and safety by explicitly linking GC execution to a well-defined operational boundary: the successful completion of a compaction cycle.
+
+### Terminology
+
+- **Unused File**: Refers to a file present in the storage directory that has never been formally recorded in any manifest. A common scenario for this includes cases where a new SST file is successfully written to storage, but the subsequent update to the manifest fails, leaving the file unreferenced.
+- **Obsolete File**: Denotes a file that was previously recorded in a manifest but has since been explicitly marked for removal. This typically occurs following operations such as data repartitioning or compaction.
+
+### GC Worker Process
+
+The GC worker operates as an integral part of the Compaction process. Once a Compaction for a specific region is completed, the GC worker is automatically triggered. Executing this process on a `datanode` is preferred to eliminate the overhead associated with having to set object storage configurations in the `metasrv`.
+
+The detailed process is as follows:
+
+1.  **Invocation**: Upon the successful completion of a Compaction for a region, the GC worker is invoked.
+2.  **Manifest Reading**: The worker reads the region's primary manifest to obtain a comprehensive list of all files marked as obsolete. Concurrently, it reads any temporary manifests generated by long-running queries to identify files that are currently in active use, thereby preventing their premature deletion.
+3.  **Lingering Time Check (Obsolete Files)**: For each identified obsolete file, the GC worker evaluates its "lingering time." Which is the time passed after it had been removed from manifest.
+4.  **Deletion Marking (Obsolete Files)**: Files that have exceeded their maximum configurable lingering time and are not referenced by any active temporary manifests are marked for deletion.
+5.  **Lingering Time (Unused Files)**: Unused files (those never recorded in any manifest) are also subject to a configurable maximum lingering time before they are eligible for deletion.
+
+Following flowchart illustrates the GC worker's process:
+
+```mermaid
+flowchart TD
+    A[Compaction Completed] --> B[Trigger GC Worker]
+    B --> C[Scan Region Manifest]
+    C --> D[Identify File Types]
+    D --> E[Unused Files<br/>Never recorded in manifest]
+    D --> F[Obsolete Files<br/>Previously in manifest<br/>but marked for removal]
+    E --> G[Check Lingering Time]
+    F --> G
+    G --> H{File exceeds<br/>configured lingering time?}
+    H -->|No| I[Skip deletion]
+    H -->|Yes| J[Check Temporary Manifest]
+    J --> K{File in use by<br/>active queries?}
+    K -->|Yes| L[Retain file<br/>Wait for next GC cycle]
+    K -->|No| M[Safely delete file]
+    I --> N[End GC cycle]
+    L --> N
+    M --> O[Update Manifest]
+    O --> N
+    N --> P[Wait for next Compaction]
+    P --> A
+    style A fill:#e1f5fe
+    style B fill:#f3e5f5
+    style M fill:#e8f5e8
+    style L fill:#fff3e0
+```
+
+#### Handling Obsolete Files
+
+An obsolete file is permanently deleted only if two conditions are met:
+1. The time elapsed since its removal from the manifest (its obsolescence timestamp) exceeds a configurable threshold.
+2. It is not currently referenced by any active temporary manifests.
+
+
+#### Handling Unused Files
+
+With the integration of the GC worker into the Compaction process, the risk of accidentally deleting newly created SST files that have not yet been recorded in the manifest is significantly mitigated. Consequently, the concept of "Unused Files" as a distinct category primarily susceptible to accidental deletion is largely resolved. Any files that are genuinely "unused" (i.e., never referenced by any manifest, including temporary ones) can be safely deleted after a configurable maximum lingering time.
+
+For debugging and auditing purposes, a comprehensive list of recently deleted files can be maintained.
+
+### Ensuring Read Consistency
+
+To prevent the GC worker from inadvertently deleting files that are actively being utilized by long-running analytical queries, a robust protection mechanism is introduced. This mechanism relies on temporary manifests that are actively kept "alive" by the queries using them.
+
+When a long-running query is detected (e.g., by a slow query recorder), it will write a temporary manifest to the region's manifest directory. This manifest lists all files required for the query. However, simply creating this file is not enough, as a query runner might crash, leaving the temporary manifest orphaned and preventing garbage collection indefinitely.
+
+To address this, the following "heartbeat" mechanism is implemented:
+1.  **Periodic Updates**: The process executing the long-running query is responsible for periodically updating the modification timestamp of its temporary manifest file (i.e., "touching" the file). This serves as a heartbeat, signaling that the query is still active.
+2.  **GC Worker Verification**: When the GC worker runs, it scans for temporary manifests. For each one it finds, it checks the file's last modification time.
+3.  **Stale File Handling**: If a temporary manifest's last modification time is older than a configurable threshold, the GC worker considers it stale (left over from a crashed or terminated query). The GC worker will then delete this stale temporary manifest. Files that were protected only by this stale manifest are no longer shielded from garbage collection.
+
+This approach ensures that only files for genuinely active queries are protected. The lifecycle of the temporary manifest is managed dynamically: it is created when a long query starts, kept alive through periodic updates, and is either deleted by the query upon normal completion or automatically cleaned up by the GC worker if the query terminates unexpectedly.
+
+This mechanism may be too complex to implement at once. We can consider a two-phased approach:
+1.  **Phase 1 (Simple Time-Based Deletion)**: Initially, implement a simpler GC strategy that deletes obsolete files based solely on a configurable lingering time. This provides a baseline for space reclamation without the complexity of temporary manifests.
+2.  **Phase 2 (Consistency-Aware GC)**: Based on the practical effectiveness and observed issues from Phase 1, we can then decide whether to implement the full temporary manifest and heartbeat mechanism to handle long-running queries. This iterative approach allows for a quicker initial implementation while gathering real-world data to justify the need for a more complex solution.
+
+## Drawbacks
+
+- **Dependency on Compaction Frequency**: The integration of the GC worker with Compaction means that GC cycles are directly tied to the frequency of compactions. In environments with infrequent compaction operations, obsolete files may accumulate for extended periods before being reclaimed, potentially leading to increased storage consumption.
+- **Race Condition with Long-Running Queries**: A potential race condition exists if a long-running query initiates but haven't write its temporary manifest in time, while a compaction process simultaneously begins and marks files used by that query as obsolete. This scenario could lead to the premature deletion of files still required by the active query. To mitigate this, the threshold time for writing a temporary manifest should be significantly shorter than the lingering time configured for obsolete files, ensuring that next GC worker runs do not delete files that are now referenced by a temporary manifest if the query is still running.
+Also the read replica shouldn't be later in manifest version for more than the lingering time of obsolete files, otherwise it might ref to files that are already deleted by the GC worker.
+- need to upload tmp manifest to object storage, which may introduce additional complexity and potential performance overhead. But since long-running queries are typically not frequent, the performance impact is expected to be minimal.
+
+
+## Conclusion and Rationale
+
+This section summarizes the key aspects and trade-offs of the proposed integrated GC worker, highlighting its advantages and potential challenges.
+
+| Aspect | Current Proposal (Integrated GC) |
+| :--- | :--- |
+| **Implementation Complexity** | **Medium**. Requires careful integration with the compaction process and the slow query recorder for temporary manifest management. |
+| **Reliability** | **High**. Integration with compaction and leveraging temporary manifests from long-running queries significantly mitigates the risk of incorrect deletion. Accurate management of lingering times for obsolete files and prevention of accidental deletion of newly created SSTs enhance data safety. |
+| **Performance Overhead** | **Low to Medium**. The GC worker runs post-compaction, minimizing direct impact on write paths. Overhead from temporary manifest management by the slow query recorder is expected to be acceptable for long-running queries. |
+| **Impact on Other Components** | **Moderate**. Requires modifications to the compaction process to trigger GC and the slow query recorder to manage temporary manifests. This introduces some coupling but enhances overall data safety. |
+| **Deletion Strategy** | **State- and Time-Based**. Obsolete files are deleted based on a configurable lingering time, which is paused if the file is referenced by a temporary manifest. Unused files (never in a manifest) are also subject to a lingering time. |
+
+## Unresolved Questions and Future Work
+
+This section outlines key areas requiring further discussion and defines potential avenues for future development.
+
+*   **Slow Query Recorder Implementation**: Detailed specifications for modify slow query recorder's implementation and its precise interaction mechanisms with temporary manifests are needed.
+*   **Configurable Lingering Times**: Establish and make configurable the specific lingering times for both obsolete and unused files to optimize storage reclamation and data availability.
+
+## Alternatives
+
+### 1. Standalone GC Service
+
+Instead of integrating the GC worker directly into the Compaction process, a standalone GC service could be implemented. This service would operate independently, periodically scanning the storage for obsolete and unused files based on manifest information and predefined retention policies.
+
+**Pros:**
+*   **Decoupling**: Separates GC logic from compaction, allowing independent scaling and deployment.
+*   **Flexibility**: Can be configured to run at different frequencies and with different strategies than compaction.
+
+**Cons:**
+*   **Increased Complexity**: Requires a separate service to manage, monitor, and coordinate with other components.
+*   **Potential for Redundancy**: May duplicate some file scanning logic already present in compaction.
+*   **Consistency Challenges**: Ensuring read consistency would require more complex coordination mechanisms between the standalone GC service and active queries, potentially involving a distributed lock manager or a more sophisticated temporary manifest system.
+
+This alternative could be implemented in the future if the integrated GC worker proves insufficient or if there is a need for more advanced GC strategies.
+
+### 2. Manifest-Driven Deletion (No Lingering Time)
+
+This alternative would involve immediate deletion of files once they are removed from the manifest, without a lingering time.
+
+**Pros:**
+*   **Simplicity**: Simplifies the GC logic by removing the need for lingering time management.
+*   **Immediate Space Reclamation**: Storage space is reclaimed as soon as files are marked for deletion.
+
+**Cons:**
+*   **Increased Risk of Data Loss**: Higher risk of deleting files still in use by long-running queries or other processes if not perfectly synchronized.
+*   **Complex Read Consistency**: Requires extremely robust and immediate mechanisms to ensure that no active queries are referencing files marked for deletion, potentially leading to performance bottlenecks or complex error handling.
+*   **Debugging Challenges**: Difficult to debug issues related to premature file deletion due to the immediate nature of the operation.
--- a/src/common/meta/src/error.rs
+++ b/src/common/meta/src/error.rs
@@ -987,6 +987,39 @@ pub enum Error {
        table_name: String,
        table_id: TableId,
    },
+
+    #[snafu(display(
+        "Column not found in column metadata, column_name: {}, column_id: {}",
+        column_name,
+        column_id
+    ))]
+    ColumnNotFound { column_name: String, column_id: u32 },
+
+    #[snafu(display(
+        "Column id mismatch, column_name: {}, expected column_id: {}, actual column_id: {}",
+        column_name,
+        expected_column_id,
+        actual_column_id
+    ))]
+    ColumnIdMismatch {
+        column_name: String,
+        expected_column_id: u32,
+        actual_column_id: u32,
+    },
+
+    #[snafu(display(
+        "Timestamp column mismatch, expected column_name: {}, expected column_id: {}, actual column_name: {}, actual column_id: {}",
+        expected_column_name,
+        expected_column_id,
+        actual_column_name,
+        actual_column_id,
+    ))]
+    TimestampMismatch {
+        expected_column_name: String,
+        expected_column_id: u32,
+        actual_column_name: String,
+        actual_column_id: u32,
+    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -1012,7 +1045,10 @@ impl ErrorExt for Error {
            | MissingColumnIds { .. }
            | MissingColumnInColumnMetadata { .. }
            | MismatchColumnId { .. }
-            | ColumnMetadataConflicts { .. } => StatusCode::Unexpected,
+            | ColumnMetadataConflicts { .. }
+            | ColumnNotFound { .. }
+            | ColumnIdMismatch { .. }
+            | TimestampMismatch { .. } => StatusCode::Unexpected,

            Unsupported { .. } => StatusCode::Unsupported,
            WriteObject { .. } | ReadObject { .. } => StatusCode::StorageUnavailable,
--- a/src/common/meta/src/metrics.rs
+++ b/src/common/meta/src/metrics.rs
@@ -15,6 +15,13 @@
 use lazy_static::lazy_static;
 use prometheus::*;

+pub const TABLE_TYPE_PHYSICAL: &str = "physical";
+pub const TABLE_TYPE_LOGICAL: &str = "logical";
+pub const ERROR_TYPE_RETRYABLE: &str = "retryable";
+pub const ERROR_TYPE_EXTERNAL: &str = "external";
+pub const STATS_TYPE_NO_REGION_METADATA: &str = "no_region_metadata";
+pub const STATS_TYPE_REGION_NOT_OPEN: &str = "region_not_open";
+
 lazy_static! {
    pub static ref METRIC_META_TXN_REQUEST: HistogramVec = register_histogram_vec!(
        "greptime_meta_txn_request",
@@ -114,4 +121,39 @@ lazy_static! {
        &["backend", "result", "op", "type"]
    )
    .unwrap();
+    pub static ref METRIC_META_RECONCILIATION_LIST_REGION_METADATA_DURATION: HistogramVec =
+        register_histogram_vec!(
+            "greptime_meta_reconciliation_list_region_metadata_duration",
+            "reconciliation list region metadata duration",
+            &["table_type"]
+        )
+        .unwrap();
+    pub static ref METRIC_META_RECONCILIATION_RESOLVED_COLUMN_METADATA: IntCounterVec =
+        register_int_counter_vec!(
+            "greptime_meta_reconciliation_resolved_column_metadata",
+            "reconciliation resolved column metadata",
+            &["strategy"]
+        )
+        .unwrap();
+    pub static ref METRIC_META_RECONCILIATION_STATS: IntCounterVec =
+        register_int_counter_vec!(
+            "greptime_meta_reconciliation_stats",
+            "reconciliation stats",
+            &["procedure_name", "table_type", "type"]
+        )
+        .unwrap();
+    pub static ref METRIC_META_RECONCILIATION_PROCEDURE: HistogramVec =
+        register_histogram_vec!(
+            "greptime_meta_reconciliation_procedure",
+            "reconcile table procedure",
+            &["procedure_name", "step"]
+        )
+        .unwrap();
+    pub static ref METRIC_META_RECONCILIATION_PROCEDURE_ERROR: IntCounterVec =
+        register_int_counter_vec!(
+            "greptime_meta_reconciliation_procedure_error",
+            "reconciliation procedure error",
+            &["procedure_name", "step", "error_type"]
+        )
+        .unwrap();
 }
--- a/src/common/meta/src/reconciliation/reconcile_catalog.rs
+++ b/src/common/meta/src/reconciliation/reconcile_catalog.rs
@@ -14,10 +14,11 @@

 use std::any::Any;
 use std::fmt::Debug;
+use std::time::Instant;

 use common_procedure::error::FromJsonSnafu;
 use common_procedure::{
-    Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure, ProcedureId,
+    Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure,
    Result as ProcedureResult, Status,
 };
 use futures::stream::BoxStream;
@@ -28,11 +29,13 @@ use crate::cache_invalidator::CacheInvalidatorRef;
 use crate::error::Result;
 use crate::key::TableMetadataManagerRef;
 use crate::lock_key::CatalogLock;
+use crate::metrics;
 use crate::node_manager::NodeManagerRef;
 use crate::reconciliation::reconcile_catalog::start::ReconcileCatalogStart;
-use crate::reconciliation::reconcile_database::utils::wait_for_inflight_subprocedures;
 use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy;
-use crate::reconciliation::utils::Context;
+use crate::reconciliation::utils::{
+    wait_for_inflight_subprocedures, Context, ReconcileCatalogMetrics, SubprocedureMeta,
+};

 pub(crate) mod end;
 pub(crate) mod reconcile_databases;
@@ -61,13 +64,15 @@ impl ReconcileCatalogContext {
        &mut self,
        procedure_ctx: &ProcedureContext,
    ) -> Result<()> {
-        if let Some(procedure_id) = self.volatile_ctx.inflight_subprocedure {
-            wait_for_inflight_subprocedures(
+        if let Some(subprocedure) = self.volatile_ctx.inflight_subprocedure.take() {
+            let subprocedures = [subprocedure];
+            let result = wait_for_inflight_subprocedures(
                procedure_ctx,
-                &[procedure_id],
+                &subprocedures,
                self.persistent_ctx.fast_fail,
            )
            .await?;
+            self.volatile_ctx.metrics += result.into();
        }
        Ok(())
    }
@@ -97,12 +102,26 @@ impl PersistentContext {
    }
 }

-#[derive(Default)]
 pub(crate) struct VolatileContext {
    /// Stores the stream of catalogs.
    schemas: Option<BoxStream<'static, Result<String>>>,
    /// Stores the inflight subprocedure.
-    inflight_subprocedure: Option<ProcedureId>,
+    inflight_subprocedure: Option<SubprocedureMeta>,
+    /// Stores the metrics of reconciling catalog.
+    metrics: ReconcileCatalogMetrics,
+    /// The start time of the reconciliation.
+    start_time: Instant,
+}
+
+impl Default for VolatileContext {
+    fn default() -> Self {
+        Self {
+            schemas: None,
+            inflight_subprocedure: None,
+            metrics: Default::default(),
+            start_time: Instant::now(),
+        }
+    }
 }

 pub struct ReconcileCatalogProcedure {
@@ -158,6 +177,11 @@ impl Procedure for ReconcileCatalogProcedure {
    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
        let state = &mut self.state;

+        let procedure_name = Self::TYPE_NAME;
+        let step = state.name();
+        let _timer = metrics::METRIC_META_RECONCILIATION_PROCEDURE
+            .with_label_values(&[procedure_name, step])
+            .start_timer();
        match state.next(&mut self.context, _ctx).await {
            Ok((next, status)) => {
                *state = next;
@@ -165,8 +189,14 @@ impl Procedure for ReconcileCatalogProcedure {
            }
            Err(e) => {
                if e.is_retry_later() {
+                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
+                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
+                        .inc();
                    Err(ProcedureError::retry_later(e))
                } else {
+                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
+                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_EXTERNAL])
+                        .inc();
                    Err(ProcedureError::external(e))
                }
            }
--- a/src/common/meta/src/reconciliation/reconcile_catalog/end.rs
+++ b/src/common/meta/src/reconciliation/reconcile_catalog/end.rs
@@ -15,6 +15,7 @@
 use std::any::Any;

 use common_procedure::{Context as ProcedureContext, Status};
+use common_telemetry::info;
 use serde::{Deserialize, Serialize};

 use crate::error::Result;
@@ -28,9 +29,16 @@ pub(crate) struct ReconcileCatalogEnd;
 impl State for ReconcileCatalogEnd {
    async fn next(
        &mut self,
-        _ctx: &mut ReconcileCatalogContext,
-        _procedure_ctx: &ProcedureContext,
+        ctx: &mut ReconcileCatalogContext,
+        procedure_ctx: &ProcedureContext,
    ) -> Result<(Box<dyn State>, Status)> {
+        info!(
+            "Catalog reconciliation completed. catalog: {}, procedure_id: {}, metrics: {}, elapsed: {:?}",
+            ctx.persistent_ctx.catalog,
+            procedure_ctx.procedure_id,
+            ctx.volatile_ctx.metrics,
+            ctx.volatile_ctx.start_time.elapsed()
+        );
        Ok((Box::new(ReconcileCatalogEnd), Status::done()))
    }

--- a/src/common/meta/src/reconciliation/reconcile_catalog/reconcile_databases.rs
+++ b/src/common/meta/src/reconciliation/reconcile_catalog/reconcile_databases.rs
@@ -23,7 +23,7 @@ use crate::error::Result;
 use crate::reconciliation::reconcile_catalog::end::ReconcileCatalogEnd;
 use crate::reconciliation::reconcile_catalog::{ReconcileCatalogContext, State};
 use crate::reconciliation::reconcile_database::ReconcileDatabaseProcedure;
-use crate::reconciliation::utils::Context;
+use crate::reconciliation::utils::{Context, SubprocedureMeta};

 #[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct ReconcileDatabases;
@@ -83,13 +83,18 @@ impl ReconcileDatabases {
        let procedure = ReconcileDatabaseProcedure::new(
            context,
            ctx.persistent_ctx.catalog.clone(),
-            schema,
+            schema.clone(),
            ctx.persistent_ctx.fast_fail,
            ctx.persistent_ctx.parallelism,
            ctx.persistent_ctx.resolve_strategy,
            true,
        );
        let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
+        ctx.volatile_ctx.inflight_subprocedure = Some(SubprocedureMeta::new_reconcile_database(
+            procedure_with_id.id,
+            ctx.persistent_ctx.catalog.clone(),
+            schema,
+        ));

        Ok((
            Box::new(ReconcileDatabases),
--- a/src/common/meta/src/reconciliation/reconcile_database.rs
+++ b/src/common/meta/src/reconciliation/reconcile_database.rs
@@ -16,16 +16,16 @@ pub(crate) mod end;
 pub(crate) mod reconcile_logical_tables;
 pub(crate) mod reconcile_tables;
 pub(crate) mod start;
-pub(crate) mod utils;

 use std::any::Any;
 use std::collections::HashMap;
 use std::fmt::Debug;
+use std::time::Instant;

 use async_trait::async_trait;
 use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
-    Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure, ProcedureId,
+    Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure,
    Result as ProcedureResult, Status,
 };
 use futures::stream::BoxStream;
@@ -39,12 +39,13 @@ use crate::error::Result;
 use crate::key::table_name::TableNameValue;
 use crate::key::TableMetadataManagerRef;
 use crate::lock_key::{CatalogLock, SchemaLock};
+use crate::metrics;
 use crate::node_manager::NodeManagerRef;
 use crate::reconciliation::reconcile_database::start::ReconcileDatabaseStart;
-use crate::reconciliation::reconcile_database::utils::wait_for_inflight_subprocedures;
 use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy;
-use crate::reconciliation::utils::Context;
-
+use crate::reconciliation::utils::{
+    wait_for_inflight_subprocedures, Context, ReconcileDatabaseMetrics, SubprocedureMeta,
+};
 pub(crate) const DEFAULT_PARALLELISM: usize = 64;

 pub(crate) struct ReconcileDatabaseContext {
@@ -66,22 +67,32 @@ impl ReconcileDatabaseContext {
        }
    }

+    /// Waits for inflight subprocedures to complete.
    pub(crate) async fn wait_for_inflight_subprocedures(
        &mut self,
        procedure_ctx: &ProcedureContext,
    ) -> Result<()> {
        if !self.volatile_ctx.inflight_subprocedures.is_empty() {
-            wait_for_inflight_subprocedures(
+            let result = wait_for_inflight_subprocedures(
                procedure_ctx,
                &self.volatile_ctx.inflight_subprocedures,
                self.persistent_ctx.fail_fast,
            )
            .await?;
+
+            // Collects result into metrics
+            let metrics = result.into();
            self.volatile_ctx.inflight_subprocedures.clear();
+            self.volatile_ctx.metrics += metrics;
        }

        Ok(())
    }
+
+    /// Returns the immutable metrics.
+    pub(crate) fn metrics(&self) -> &ReconcileDatabaseMetrics {
+        &self.volatile_ctx.metrics
+    }
 }

 #[derive(Debug, Serialize, Deserialize)]
@@ -114,7 +125,6 @@ impl PersistentContext {
    }
 }

-#[derive(Default)]
 pub(crate) struct VolatileContext {
    /// Stores pending physical tables.
    pending_tables: Vec<(TableId, TableName)>,
@@ -124,9 +134,26 @@ pub(crate) struct VolatileContext {
    /// - Value: Vector of (TableId, TableName) tuples representing logical tables belonging to the physical table.
    pending_logical_tables: HashMap<TableId, Vec<(TableId, TableName)>>,
    /// Stores inflight subprocedures.
-    inflight_subprocedures: Vec<ProcedureId>,
+    inflight_subprocedures: Vec<SubprocedureMeta>,
    /// Stores the stream of tables.
    tables: Option<BoxStream<'static, Result<(String, TableNameValue)>>>,
+    /// The metrics of reconciling database.
+    metrics: ReconcileDatabaseMetrics,
+    /// The start time of the reconciliation.
+    start_time: Instant,
+}
+
+impl Default for VolatileContext {
+    fn default() -> Self {
+        Self {
+            pending_tables: vec![],
+            pending_logical_tables: HashMap::new(),
+            inflight_subprocedures: vec![],
+            tables: None,
+            metrics: ReconcileDatabaseMetrics::default(),
+            start_time: Instant::now(),
+        }
+    }
 }

 pub struct ReconcileDatabaseProcedure {
@@ -190,6 +217,11 @@ impl Procedure for ReconcileDatabaseProcedure {
    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
        let state = &mut self.state;

+        let procedure_name = Self::TYPE_NAME;
+        let step = state.name();
+        let _timer = metrics::METRIC_META_RECONCILIATION_PROCEDURE
+            .with_label_values(&[procedure_name, step])
+            .start_timer();
        match state.next(&mut self.context, _ctx).await {
            Ok((next, status)) => {
                *state = next;
@@ -197,8 +229,14 @@ impl Procedure for ReconcileDatabaseProcedure {
            }
            Err(e) => {
                if e.is_retry_later() {
+                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
+                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
+                        .inc();
                    Err(ProcedureError::retry_later(e))
                } else {
+                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
+                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_EXTERNAL])
+                        .inc();
                    Err(ProcedureError::external(e))
                }
            }
--- a/src/common/meta/src/reconciliation/reconcile_database/end.rs
+++ b/src/common/meta/src/reconciliation/reconcile_database/end.rs
@@ -15,6 +15,7 @@
 use std::any::Any;

 use common_procedure::{Context as ProcedureContext, Status};
+use common_telemetry::info;
 use serde::{Deserialize, Serialize};

 use crate::error::Result;
@@ -28,9 +29,17 @@ pub(crate) struct ReconcileDatabaseEnd;
 impl State for ReconcileDatabaseEnd {
    async fn next(
        &mut self,
-        _ctx: &mut ReconcileDatabaseContext,
-        _procedure_ctx: &ProcedureContext,
+        ctx: &mut ReconcileDatabaseContext,
+        procedure_ctx: &ProcedureContext,
    ) -> Result<(Box<dyn State>, Status)> {
+        info!(
+            "Database reconciliation completed. schema: {}, catalog: {}, procedure_id: {}, metrics: {}, elapsed: {:?}",
+            ctx.persistent_ctx.schema,
+            ctx.persistent_ctx.catalog,
+            procedure_ctx.procedure_id,
+            ctx.metrics(),
+            ctx.volatile_ctx.start_time.elapsed(),
+        );
        Ok((Box::new(ReconcileDatabaseEnd), Status::done()))
    }

--- a/src/common/meta/src/reconciliation/reconcile_database/reconcile_logical_tables.rs
+++ b/src/common/meta/src/reconciliation/reconcile_database/reconcile_logical_tables.rs
@@ -29,7 +29,7 @@ use crate::key::table_route::TableRouteValue;
 use crate::reconciliation::reconcile_database::end::ReconcileDatabaseEnd;
 use crate::reconciliation::reconcile_database::{ReconcileDatabaseContext, State};
 use crate::reconciliation::reconcile_logical_tables::ReconcileLogicalTablesProcedure;
-use crate::reconciliation::utils::Context;
+use crate::reconciliation::utils::{Context, SubprocedureMeta};

 #[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct ReconcileLogicalTables;
@@ -128,13 +128,12 @@ impl State for ReconcileLogicalTables {
 impl ReconcileLogicalTables {
    fn schedule_reconcile_logical_tables(
        ctx: &mut ReconcileDatabaseContext,
-        buffer: &mut Vec<ProcedureWithId>,
+        buffer: &mut Vec<(ProcedureWithId, SubprocedureMeta)>,
    ) -> Result<(Box<dyn State>, Status)> {
-        let procedures = std::mem::take(buffer);
-        ctx.volatile_ctx
-            .inflight_subprocedures
-            .extend(procedures.iter().map(|p| p.id));
+        let buffer = std::mem::take(buffer);
+        let (procedures, meta): (Vec<_>, Vec<_>) = buffer.into_iter().unzip();

+        ctx.volatile_ctx.inflight_subprocedures.extend(meta);
        Ok((
            Box::new(ReconcileLogicalTables),
            Status::suspended(procedures, false),
@@ -142,7 +141,7 @@ impl ReconcileLogicalTables {
    }

    fn should_schedule_reconcile_logical_tables(
-        buffer: &[ProcedureWithId],
+        buffer: &[(ProcedureWithId, SubprocedureMeta)],
        parallelism: usize,
    ) -> bool {
        buffer.len() >= parallelism
@@ -152,7 +151,7 @@ impl ReconcileLogicalTables {
        ctx: &Context,
        pending_logical_tables: &mut HashMap<TableId, Vec<(TableId, TableName)>>,
        parallelism: usize,
-    ) -> Result<Option<ProcedureWithId>> {
+    ) -> Result<Option<(ProcedureWithId, SubprocedureMeta)>> {
        let mut physical_table_id = None;
        for (table_id, tables) in pending_logical_tables.iter() {
            if tables.len() >= parallelism {
@@ -176,7 +175,7 @@ impl ReconcileLogicalTables {
    async fn build_remaining_procedures(
        ctx: &Context,
        pending_logical_tables: &mut HashMap<TableId, Vec<(TableId, TableName)>>,
-        pending_procedures: &mut Vec<ProcedureWithId>,
+        pending_procedures: &mut Vec<(ProcedureWithId, SubprocedureMeta)>,
        parallelism: usize,
    ) -> Result<()> {
        if pending_logical_tables.is_empty() {
@@ -203,7 +202,7 @@ impl ReconcileLogicalTables {
        ctx: &Context,
        physical_table_id: TableId,
        logical_tables: Vec<(TableId, TableName)>,
-    ) -> Result<ProcedureWithId> {
+    ) -> Result<(ProcedureWithId, SubprocedureMeta)> {
        let table_info = ctx
            .table_metadata_manager
            .table_info_manager()
@@ -217,12 +216,18 @@ impl ReconcileLogicalTables {
        let procedure = ReconcileLogicalTablesProcedure::new(
            ctx.clone(),
            physical_table_id,
-            physical_table_name,
-            logical_tables,
+            physical_table_name.clone(),
+            logical_tables.clone(),
            true,
        );
-
-        Ok(ProcedureWithId::with_random_id(Box::new(procedure)))
+        let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
+        let subprocedure_meta = SubprocedureMeta::new_logical_table(
+            procedure_with_id.id,
+            physical_table_id,
+            physical_table_name,
+            logical_tables,
+        );
+        Ok((procedure_with_id, subprocedure_meta))
    }

    fn enqueue_logical_table(
--- a/src/common/meta/src/reconciliation/reconcile_database/reconcile_tables.rs
+++ b/src/common/meta/src/reconciliation/reconcile_database/reconcile_tables.rs
@@ -27,7 +27,7 @@ use crate::key::table_route::TableRouteValue;
 use crate::reconciliation::reconcile_database::reconcile_logical_tables::ReconcileLogicalTables;
 use crate::reconciliation::reconcile_database::{ReconcileDatabaseContext, State};
 use crate::reconciliation::reconcile_table::ReconcileTableProcedure;
-use crate::reconciliation::utils::Context;
+use crate::reconciliation::utils::{Context, SubprocedureMeta};

 #[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct ReconcileTables;
@@ -104,14 +104,14 @@ impl ReconcileTables {
        ctx: &mut ReconcileDatabaseContext,
    ) -> Result<(Box<dyn State>, Status)> {
        let tables = std::mem::take(&mut ctx.volatile_ctx.pending_tables);
-        let subprocedures = Self::build_reconcile_table_procedures(ctx, tables);
-        ctx.volatile_ctx
-            .inflight_subprocedures
-            .extend(subprocedures.iter().map(|p| p.id));
-
+        let (procedures, meta): (Vec<_>, Vec<_>) =
+            Self::build_reconcile_table_procedures(ctx, tables)
+                .into_iter()
+                .unzip();
+        ctx.volatile_ctx.inflight_subprocedures.extend(meta);
        Ok((
            Box::new(ReconcileTables),
-            Status::suspended(subprocedures, false),
+            Status::suspended(procedures, false),
        ))
    }

@@ -125,7 +125,7 @@ impl ReconcileTables {
    fn build_reconcile_table_procedures(
        ctx: &ReconcileDatabaseContext,
        tables: Vec<(TableId, TableName)>,
-    ) -> Vec<ProcedureWithId> {
+    ) -> Vec<(ProcedureWithId, SubprocedureMeta)> {
        let mut procedures = Vec::with_capacity(tables.len());
        for (table_id, table_name) in tables {
            let context = Context {
@@ -141,11 +141,13 @@ impl ReconcileTables {
                true,
            );
            let procedure = ProcedureWithId::with_random_id(Box::new(procedure));
+            let meta =
+                SubprocedureMeta::new_physical_table(procedure.id, table_id, table_name.clone());
            info!(
                "Reconcile table: {}, table_id: {}, procedure_id: {}",
                table_name, table_id, procedure.id
            );
-            procedures.push(procedure)
+            procedures.push((procedure, meta));
        }

        procedures
--- a/src/common/meta/src/reconciliation/reconcile_database/start.rs
+++ b/src/common/meta/src/reconciliation/reconcile_database/start.rs
@@ -33,7 +33,7 @@ impl State for ReconcileDatabaseStart {
    async fn next(
        &mut self,
        ctx: &mut ReconcileDatabaseContext,
-        _procedure_ctx: &ProcedureContext,
+        procedure_ctx: &ProcedureContext,
    ) -> Result<(Box<dyn State>, Status)> {
        let exists = ctx
            .table_metadata_manager
@@ -51,8 +51,8 @@ impl State for ReconcileDatabaseStart {
            },
        );
        info!(
-            "Reconcile database: {}, catalog: {}",
-            ctx.persistent_ctx.schema, ctx.persistent_ctx.catalog
+            "Reconcile database: {}, catalog: {}, procedure_id: {}",
+            ctx.persistent_ctx.schema, ctx.persistent_ctx.catalog, procedure_ctx.procedure_id,
        );
        Ok((Box::new(ReconcileTables), Status::executing(true)))
    }
--- a/src/common/meta/src/reconciliation/reconcile_database/utils.rs
+++ b/src/common/meta/src/reconciliation/reconcile_database/utils.rs
@@ -1,79 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// 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.
-
-use common_procedure::{watcher, Context as ProcedureContext, ProcedureId};
-use common_telemetry::{error, info, warn};
-use futures::future::{join_all, try_join_all};
-use snafu::{OptionExt, ResultExt};
-
-use crate::error::{
-    ProcedureStateReceiverNotFoundSnafu, ProcedureStateReceiverSnafu, Result, WaitProcedureSnafu,
-};
-
-/// Wait for inflight subprocedures.
-///
-/// If `fail_fast` is true, the function will return an error if any subprocedure fails.
-/// Otherwise, the function will continue waiting for all subprocedures to complete.
-pub(crate) async fn wait_for_inflight_subprocedures(
-    procedure_ctx: &ProcedureContext,
-    subprocedures: &[ProcedureId],
-    fail_fast: bool,
-) -> Result<()> {
-    let mut receivers = Vec::with_capacity(subprocedures.len());
-    for procedure_id in subprocedures {
-        let receiver = procedure_ctx
-            .provider
-            .procedure_state_receiver(*procedure_id)
-            .await
-            .context(ProcedureStateReceiverSnafu {
-                procedure_id: *procedure_id,
-            })?
-            .context(ProcedureStateReceiverNotFoundSnafu {
-                procedure_id: *procedure_id,
-            })?;
-        receivers.push(receiver);
-    }
-
-    let mut tasks = Vec::with_capacity(receivers.len());
-    for receiver in receivers.iter_mut() {
-        let fut = watcher::wait(receiver);
-        tasks.push(fut);
-    }
-
-    if fail_fast {
-        try_join_all(tasks).await.context(WaitProcedureSnafu)?;
-    } else {
-        let mut failed = 0;
-        let total = tasks.len();
-        for result in join_all(tasks).await {
-            if let Err(e) = result {
-                error!(e; "inflight subprocedure, procedure_id: {}", procedure_ctx.procedure_id);
-                failed += 1;
-            }
-        }
-        if failed > 0 {
-            warn!(
-                "{} inflight subprocedures failed, total: {}, procedure_id: {}",
-                failed, total, procedure_ctx.procedure_id
-            );
-        } else {
-            info!(
-                "{} inflight subprocedures completed, procedure_id: {}",
-                total, procedure_ctx.procedure_id
-            );
-        }
-    }
-
-    Ok(())
-}
--- a/src/common/meta/src/reconciliation/reconcile_logical_tables.rs
+++ b/src/common/meta/src/reconciliation/reconcile_logical_tables.rs
@@ -40,15 +40,17 @@ use crate::key::table_info::TableInfoValue;
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
+use crate::metrics;
 use crate::node_manager::NodeManagerRef;
 use crate::reconciliation::reconcile_logical_tables::reconciliation_start::ReconciliationStart;
-use crate::reconciliation::utils::Context;
+use crate::reconciliation::utils::{Context, ReconcileLogicalTableMetrics};

 pub struct ReconcileLogicalTablesContext {
    pub node_manager: NodeManagerRef,
    pub table_metadata_manager: TableMetadataManagerRef,
    pub cache_invalidator: CacheInvalidatorRef,
    pub persistent_ctx: PersistentContext,
+    pub volatile_ctx: VolatileContext,
 }

 impl ReconcileLogicalTablesContext {
@@ -59,16 +61,29 @@ impl ReconcileLogicalTablesContext {
            table_metadata_manager: ctx.table_metadata_manager,
            cache_invalidator: ctx.cache_invalidator,
            persistent_ctx,
+            volatile_ctx: VolatileContext::default(),
        }
    }

+    /// Returns the physical table name.
    pub(crate) fn table_name(&self) -> &TableName {
        &self.persistent_ctx.table_name
    }

+    /// Returns the physical table id.
    pub(crate) fn table_id(&self) -> TableId {
        self.persistent_ctx.table_id
    }
+
+    /// Returns a mutable reference to the metrics.
+    pub(crate) fn mut_metrics(&mut self) -> &mut ReconcileLogicalTableMetrics {
+        &mut self.volatile_ctx.metrics
+    }
+
+    /// Returns a reference to the metrics.
+    pub(crate) fn metrics(&self) -> &ReconcileLogicalTableMetrics {
+        &self.volatile_ctx.metrics
+    }
 }

 #[derive(Debug, Serialize, Deserialize)]
@@ -120,6 +135,11 @@ impl PersistentContext {
    }
 }

+#[derive(Default)]
+pub(crate) struct VolatileContext {
+    pub(crate) metrics: ReconcileLogicalTableMetrics,
+}
+
 pub struct ReconcileLogicalTablesProcedure {
    pub context: ReconcileLogicalTablesContext,
    state: Box<dyn State>,
@@ -173,6 +193,11 @@ impl Procedure for ReconcileLogicalTablesProcedure {
    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
        let state = &mut self.state;

+        let procedure_name = Self::TYPE_NAME;
+        let step = state.name();
+        let _timer = metrics::METRIC_META_RECONCILIATION_PROCEDURE
+            .with_label_values(&[procedure_name, step])
+            .start_timer();
        match state.next(&mut self.context, _ctx).await {
            Ok((next, status)) => {
                *state = next;
@@ -180,8 +205,14 @@ impl Procedure for ReconcileLogicalTablesProcedure {
            }
            Err(e) => {
                if e.is_retry_later() {
+                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
+                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
+                        .inc();
                    Err(ProcedureError::retry_later(e))
                } else {
+                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
+                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_EXTERNAL])
+                        .inc();
                    Err(ProcedureError::external(e))
                }
            }
--- a/src/common/meta/src/reconciliation/reconcile_logical_tables/reconciliation_end.rs
+++ b/src/common/meta/src/reconciliation/reconcile_logical_tables/reconciliation_end.rs
@@ -15,6 +15,7 @@
 use std::any::Any;

 use common_procedure::{Context as ProcedureContext, Status};
+use common_telemetry::info;
 use serde::{Deserialize, Serialize};

 use crate::error::Result;
@@ -28,9 +29,21 @@ pub struct ReconciliationEnd;
 impl State for ReconciliationEnd {
    async fn next(
        &mut self,
-        _ctx: &mut ReconcileLogicalTablesContext,
-        _procedure_ctx: &ProcedureContext,
+        ctx: &mut ReconcileLogicalTablesContext,
+        procedure_ctx: &ProcedureContext,
    ) -> Result<(Box<dyn State>, Status)> {
+        let table_id = ctx.table_id();
+        let table_name = ctx.table_name();
+        let metrics = ctx.metrics();
+
+        info!(
+            "Logical tables reconciliation completed. logical tables: {:?}, physical_table_id: {}, table_name: {}, procedure_id: {}, metrics: {}",
+            ctx.persistent_ctx.logical_table_ids,
+            table_id,
+            table_name,
+            procedure_ctx.procedure_id,
+            metrics
+        );
        Ok((Box::new(ReconciliationEnd), Status::done()))
    }

--- a/src/common/meta/src/reconciliation/reconcile_logical_tables/reconciliation_start.rs
+++ b/src/common/meta/src/reconciliation/reconcile_logical_tables/reconciliation_start.rs
@@ -25,8 +25,11 @@ use crate::ddl::utils::region_metadata_lister::RegionMetadataLister;
 use crate::ddl::utils::table_id::get_all_table_ids_by_names;
 use crate::ddl::utils::table_info::all_logical_table_routes_have_same_physical_id;
 use crate::error::{self, Result};
+use crate::metrics;
 use crate::reconciliation::reconcile_logical_tables::resolve_table_metadatas::ResolveTableMetadatas;
-use crate::reconciliation::reconcile_logical_tables::{ReconcileLogicalTablesContext, State};
+use crate::reconciliation::reconcile_logical_tables::{
+    ReconcileLogicalTablesContext, ReconcileLogicalTablesProcedure, State,
+};
 use crate::reconciliation::utils::check_column_metadatas_consistent;

 /// The start state of the reconciliation procedure.
@@ -39,7 +42,7 @@ impl State for ReconciliationStart {
    async fn next(
        &mut self,
        ctx: &mut ReconcileLogicalTablesContext,
-        _procedure_ctx: &ProcedureContext,
+        procedure_ctx: &ProcedureContext,
    ) -> Result<(Box<dyn State>, Status)> {
        let table_id = ctx.table_id();
        let table_name = ctx.table_name();
@@ -58,35 +61,48 @@ impl State for ReconciliationStart {
            }
        );

-        info!(
-            "Starting reconciliation for logical table: table_id: {}, table_name: {}",
-            table_id, table_name
-        );
-
        let region_metadata_lister = RegionMetadataLister::new(ctx.node_manager.clone());
-        let region_metadatas = region_metadata_lister
-            .list(physical_table_id, &physical_table_route.region_routes)
-            .await?;
+        let region_metadatas = {
+            let _timer = metrics::METRIC_META_RECONCILIATION_LIST_REGION_METADATA_DURATION
+                .with_label_values(&[metrics::TABLE_TYPE_PHYSICAL])
+                .start_timer();
+            region_metadata_lister
+                .list(physical_table_id, &physical_table_route.region_routes)
+                .await?
+        };
+
+        ensure!(!region_metadatas.is_empty(), {
+            metrics::METRIC_META_RECONCILIATION_STATS
+                .with_label_values(&[
+                    ReconcileLogicalTablesProcedure::TYPE_NAME,
+                    metrics::TABLE_TYPE_PHYSICAL,
+                    metrics::STATS_TYPE_NO_REGION_METADATA,
+                ])
+                .inc();

-        ensure!(
-            !region_metadatas.is_empty(),
            error::UnexpectedSnafu {
                err_msg: format!(
-                    "No region metadata found for table: {}, table_id: {}",
+                    "No region metadata found for physical table: {}, table_id: {}",
                    table_name, table_id
                ),
            }
-        );
+        });

-        if region_metadatas.iter().any(|r| r.is_none()) {
-            return error::UnexpectedSnafu {
+        ensure!(region_metadatas.iter().all(|r| r.is_some()), {
+            metrics::METRIC_META_RECONCILIATION_STATS
+                .with_label_values(&[
+                    ReconcileLogicalTablesProcedure::TYPE_NAME,
+                    metrics::TABLE_TYPE_PHYSICAL,
+                    metrics::STATS_TYPE_REGION_NOT_OPEN,
+                ])
+                .inc();
+            error::UnexpectedSnafu {
                err_msg: format!(
-                    "Some regions of the physical table are not open. Table: {}, table_id: {}",
+                    "Some regions of the physical table are not open. physical table: {}, table_id: {}",
                    table_name, table_id
                ),
            }
-            .fail();
-        }
+        });

        // Safety: checked above
        let region_metadatas = region_metadatas
@@ -96,14 +112,13 @@ impl State for ReconciliationStart {
        let _region_metadata = check_column_metadatas_consistent(&region_metadatas).context(
            error::UnexpectedSnafu {
                err_msg: format!(
-                    "Column metadatas are not consistent for table: {}, table_id: {}",
+                    "Column metadatas are not consistent for physical table: {}, table_id: {}",
                    table_name, table_id
                ),
            },
        )?;

        // TODO(weny): ensure all columns in region metadata can be found in table info.
-
        // Validates the logical tables.
        Self::validate_schema(&ctx.persistent_ctx.logical_tables)?;
        let table_refs = ctx
@@ -119,6 +134,12 @@ impl State for ReconciliationStart {
        .await?;
        Self::validate_logical_table_routes(ctx, &table_ids).await?;

+        let table_name = ctx.table_name();
+        info!(
+            "Starting reconciliation for logical tables: {:?}, physical_table_id: {}, table_name: {}, procedure_id: {}",
+            table_ids, table_id, table_name, procedure_ctx.procedure_id
+        );
+
        ctx.persistent_ctx.physical_table_route = Some(physical_table_route);
        ctx.persistent_ctx.logical_table_ids = table_ids;
        Ok((Box::new(ResolveTableMetadatas), Status::executing(true)))
--- a/src/common/meta/src/reconciliation/reconcile_logical_tables/resolve_table_metadatas.rs
+++ b/src/common/meta/src/reconciliation/reconcile_logical_tables/resolve_table_metadatas.rs
@@ -22,8 +22,11 @@ use snafu::ensure;
 use crate::ddl::utils::region_metadata_lister::RegionMetadataLister;
 use crate::ddl::utils::table_info::get_all_table_info_values_by_table_ids;
 use crate::error::{self, Result};
+use crate::metrics;
 use crate::reconciliation::reconcile_logical_tables::reconcile_regions::ReconcileRegions;
-use crate::reconciliation::reconcile_logical_tables::{ReconcileLogicalTablesContext, State};
+use crate::reconciliation::reconcile_logical_tables::{
+    ReconcileLogicalTablesContext, ReconcileLogicalTablesProcedure, State,
+};
 use crate::reconciliation::utils::{
    check_column_metadatas_consistent, need_update_logical_table_info,
 };
@@ -65,22 +68,38 @@ impl State for ResolveTableMetadatas {
            .unwrap()
            .region_routes;
        let region_metadata_lister = RegionMetadataLister::new(ctx.node_manager.clone());
+        let mut metadata_consistent_count = 0;
+        let mut metadata_inconsistent_count = 0;
+        let mut create_tables_count = 0;
        for (table_id, table_info_value) in table_ids.iter().zip(table_info_values.iter()) {
-            let region_metadatas = region_metadata_lister
-                .list(*table_id, region_routes)
-                .await?;
+            let region_metadatas = {
+                let _timer = metrics::METRIC_META_RECONCILIATION_LIST_REGION_METADATA_DURATION
+                    .with_label_values(&[metrics::TABLE_TYPE_LOGICAL])
+                    .start_timer();
+                region_metadata_lister
+                    .list(*table_id, region_routes)
+                    .await?
+            };
+
+            ensure!(!region_metadatas.is_empty(), {
+                metrics::METRIC_META_RECONCILIATION_STATS
+                    .with_label_values(&[
+                        ReconcileLogicalTablesProcedure::TYPE_NAME,
+                        metrics::TABLE_TYPE_LOGICAL,
+                        metrics::STATS_TYPE_NO_REGION_METADATA,
+                    ])
+                    .inc();

-            ensure!(
-                !region_metadatas.is_empty(),
                error::UnexpectedSnafu {
                    err_msg: format!(
                        "No region metadata found for table: {}, table_id: {}",
                        table_info_value.table_info.name, table_id
                    ),
                }
-            );
+            });

            if region_metadatas.iter().any(|r| r.is_none()) {
+                create_tables_count += 1;
                create_tables.push((*table_id, table_info_value.table_info.clone()));
                continue;
            }
@@ -91,10 +110,12 @@ impl State for ResolveTableMetadatas {
                .map(|r| r.unwrap())
                .collect::<Vec<_>>();
            if let Some(column_metadatas) = check_column_metadatas_consistent(&region_metadatas) {
+                metadata_consistent_count += 1;
                if need_update_logical_table_info(&table_info_value.table_info, &column_metadatas) {
                    update_table_infos.push((*table_id, column_metadatas));
                }
            } else {
+                metadata_inconsistent_count += 1;
                // If the logical regions have inconsistent column metadatas, it won't affect read and write.
                // It's safe to continue if the column metadatas of the logical table are inconsistent.
                warn!(
@@ -121,6 +142,11 @@ impl State for ResolveTableMetadatas {
        );
        ctx.persistent_ctx.update_table_infos = update_table_infos;
        ctx.persistent_ctx.create_tables = create_tables;
+        // Update metrics.
+        let metrics = ctx.mut_metrics();
+        metrics.column_metadata_consistent_count = metadata_consistent_count;
+        metrics.column_metadata_inconsistent_count = metadata_inconsistent_count;
+        metrics.create_tables_count = create_tables_count;
        Ok((Box::new(ReconcileRegions), Status::executing(true)))
    }

--- a/src/common/meta/src/reconciliation/reconcile_logical_tables/update_table_infos.rs
+++ b/src/common/meta/src/reconciliation/reconcile_logical_tables/update_table_infos.rs
@@ -96,6 +96,7 @@ impl State for UpdateTableInfos {
        let table_id = ctx.table_id();
        let table_name = ctx.table_name();

+        let updated_table_info_num = table_info_values_to_update.len();
        batch_update_table_info_values(&ctx.table_metadata_manager, table_info_values_to_update)
            .await?;

@@ -122,6 +123,9 @@ impl State for UpdateTableInfos {
            .await?;

        ctx.persistent_ctx.update_table_infos.clear();
+        // Update metrics.
+        let metrics = ctx.mut_metrics();
+        metrics.update_table_info_count = updated_table_info_num;
        Ok((Box::new(ReconciliationEnd), Status::executing(false)))
    }

--- a/src/common/meta/src/reconciliation/reconcile_table.rs
+++ b/src/common/meta/src/reconciliation/reconcile_table.rs
@@ -40,10 +40,13 @@ use crate::key::table_info::TableInfoValue;
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
 use crate::lock_key::{CatalogLock, SchemaLock, TableNameLock};
+use crate::metrics;
 use crate::node_manager::NodeManagerRef;
 use crate::reconciliation::reconcile_table::reconciliation_start::ReconciliationStart;
 use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy;
-use crate::reconciliation::utils::{build_table_meta_from_column_metadatas, Context};
+use crate::reconciliation::utils::{
+    build_table_meta_from_column_metadatas, Context, ReconcileTableMetrics,
+};

 pub struct ReconcileTableContext {
    pub node_manager: NodeManagerRef,
@@ -65,13 +68,46 @@ impl ReconcileTableContext {
        }
    }

+    /// Returns the physical table name.
    pub(crate) fn table_name(&self) -> &TableName {
        &self.persistent_ctx.table_name
    }

+    /// Returns the physical table id.
    pub(crate) fn table_id(&self) -> TableId {
        self.persistent_ctx.table_id
    }
+
+    /// Builds a [`RawTableMeta`] from the provided [`ColumnMetadata`]s.
+    pub(crate) fn build_table_meta(
+        &self,
+        column_metadatas: &[ColumnMetadata],
+    ) -> Result<RawTableMeta> {
+        // Safety: The table info value is set in `ReconciliationStart` state.
+        let table_info_value = self.persistent_ctx.table_info_value.as_ref().unwrap();
+        let table_id = self.table_id();
+        let table_ref = self.table_name().table_ref();
+        let name_to_ids = table_info_value.table_info.name_to_ids();
+        let table_meta = build_table_meta_from_column_metadatas(
+            table_id,
+            table_ref,
+            &table_info_value.table_info.meta,
+            name_to_ids,
+            column_metadatas,
+        )?;
+
+        Ok(table_meta)
+    }
+
+    /// Returns a mutable reference to the metrics.
+    pub(crate) fn mut_metrics(&mut self) -> &mut ReconcileTableMetrics {
+        &mut self.volatile_ctx.metrics
+    }
+
+    /// Returns a reference to the metrics.
+    pub(crate) fn metrics(&self) -> &ReconcileTableMetrics {
+        &self.volatile_ctx.metrics
+    }
 }

 #[derive(Debug, Serialize, Deserialize)]
@@ -110,29 +146,7 @@ impl PersistentContext {
 #[derive(Default)]
 pub(crate) struct VolatileContext {
    pub(crate) table_meta: Option<RawTableMeta>,
-}
-
-impl ReconcileTableContext {
-    /// Builds a [`RawTableMeta`] from the provided [`ColumnMetadata`]s.
-    pub(crate) fn build_table_meta(
-        &self,
-        column_metadatas: &[ColumnMetadata],
-    ) -> Result<RawTableMeta> {
-        // Safety: The table info value is set in `ReconciliationStart` state.
-        let table_info_value = self.persistent_ctx.table_info_value.as_ref().unwrap();
-        let table_id = self.table_id();
-        let table_ref = self.table_name().table_ref();
-        let name_to_ids = table_info_value.table_info.name_to_ids();
-        let table_meta = build_table_meta_from_column_metadatas(
-            table_id,
-            table_ref,
-            &table_info_value.table_info.meta,
-            name_to_ids,
-            column_metadatas,
-        )?;
-
-        Ok(table_meta)
-    }
+    pub(crate) metrics: ReconcileTableMetrics,
 }

 pub struct ReconcileTableProcedure {
@@ -191,6 +205,11 @@ impl Procedure for ReconcileTableProcedure {
    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
        let state = &mut self.state;

+        let procedure_name = Self::TYPE_NAME;
+        let step = state.name();
+        let _timer = metrics::METRIC_META_RECONCILIATION_PROCEDURE
+            .with_label_values(&[procedure_name, step])
+            .start_timer();
        match state.next(&mut self.context, _ctx).await {
            Ok((next, status)) => {
                *state = next;
@@ -198,8 +217,14 @@ impl Procedure for ReconcileTableProcedure {
            }
            Err(e) => {
                if e.is_retry_later() {
+                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
+                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_RETRYABLE])
+                        .inc();
                    Err(ProcedureError::retry_later(e))
                } else {
+                    metrics::METRIC_META_RECONCILIATION_PROCEDURE_ERROR
+                        .with_label_values(&[procedure_name, step, metrics::ERROR_TYPE_EXTERNAL])
+                        .inc();
                    Err(ProcedureError::external(e))
                }
            }
--- a/src/common/meta/src/reconciliation/reconcile_table/reconciliation_end.rs
+++ b/src/common/meta/src/reconciliation/reconcile_table/reconciliation_end.rs
@@ -15,6 +15,7 @@
 use std::any::Any;

 use common_procedure::{Context as ProcedureContext, Status};
+use common_telemetry::info;
 use serde::{Deserialize, Serialize};
 use tonic::async_trait;

@@ -31,9 +32,18 @@ pub struct ReconciliationEnd;
 impl State for ReconciliationEnd {
    async fn next(
        &mut self,
-        _ctx: &mut ReconcileTableContext,
-        _procedure_ctx: &ProcedureContext,
+        ctx: &mut ReconcileTableContext,
+        procedure_ctx: &ProcedureContext,
    ) -> Result<(Box<dyn State>, Status)> {
+        let table_id = ctx.table_id();
+        let table_name = ctx.table_name();
+        let metrics = ctx.metrics();
+
+        info!(
+            "Physical table reconciliation completed. table_name: {}, table_id: {}, procedure_id: {}, metrics: {}",
+            table_name, table_id, procedure_ctx.procedure_id, metrics
+        );
+
        Ok((Box::new(ReconciliationEnd), Status::done()))
    }

--- a/src/common/meta/src/reconciliation/reconcile_table/reconciliation_start.rs
+++ b/src/common/meta/src/reconciliation/reconcile_table/reconciliation_start.rs
@@ -20,9 +20,12 @@ use serde::{Deserialize, Serialize};
 use snafu::ensure;

 use crate::ddl::utils::region_metadata_lister::RegionMetadataLister;
-use crate::error::{self, Result, UnexpectedSnafu};
+use crate::error::{self, Result};
+use crate::metrics::{self};
 use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveColumnMetadata;
-use crate::reconciliation::reconcile_table::{ReconcileTableContext, State};
+use crate::reconciliation::reconcile_table::{
+    ReconcileTableContext, ReconcileTableProcedure, State,
+};

 /// The start state of the reconciliation procedure.
 ///
@@ -40,7 +43,7 @@ impl State for ReconciliationStart {
    async fn next(
        &mut self,
        ctx: &mut ReconcileTableContext,
-        _procedure_ctx: &ProcedureContext,
+        procedure_ctx: &ProcedureContext,
    ) -> Result<(Box<dyn State>, Status)> {
        let table_id = ctx.table_id();
        let table_name = ctx.table_name();
@@ -60,33 +63,56 @@ impl State for ReconciliationStart {
            }
        );

-        info!("Reconciling table: {}, table_id: {}", table_name, table_id);
+        info!(
+            "Reconciling table: {}, table_id: {}, procedure_id: {}",
+            table_name, table_id, procedure_ctx.procedure_id
+        );
        // TODO(weny): Repairs the table route if needed.
        let region_metadata_lister = RegionMetadataLister::new(ctx.node_manager.clone());
-        // Always list region metadatas for the physical table.
-        let region_metadatas = region_metadata_lister
-            .list(physical_table_id, &physical_table_route.region_routes)
-            .await?;

-        ensure!(
-            !region_metadatas.is_empty(),
+        let region_metadatas = {
+            let _timer = metrics::METRIC_META_RECONCILIATION_LIST_REGION_METADATA_DURATION
+                .with_label_values(&[metrics::TABLE_TYPE_PHYSICAL])
+                .start_timer();
+            // Always list region metadatas for the physical table.
+            region_metadata_lister
+                .list(physical_table_id, &physical_table_route.region_routes)
+                .await?
+        };
+
+        ensure!(!region_metadatas.is_empty(), {
+            metrics::METRIC_META_RECONCILIATION_STATS
+                .with_label_values(&[
+                    ReconcileTableProcedure::TYPE_NAME,
+                    metrics::TABLE_TYPE_PHYSICAL,
+                    metrics::STATS_TYPE_NO_REGION_METADATA,
+                ])
+                .inc();
+
            error::UnexpectedSnafu {
                err_msg: format!(
                    "No region metadata found for table: {}, table_id: {}",
                    table_name, table_id
                ),
            }
-        );
+        });

-        if region_metadatas.iter().any(|r| r.is_none()) {
-            return UnexpectedSnafu {
+        ensure!(region_metadatas.iter().all(|r| r.is_some()), {
+            metrics::METRIC_META_RECONCILIATION_STATS
+                .with_label_values(&[
+                    ReconcileTableProcedure::TYPE_NAME,
+                    metrics::TABLE_TYPE_PHYSICAL,
+                    metrics::STATS_TYPE_REGION_NOT_OPEN,
+                ])
+                .inc();
+
+            error::UnexpectedSnafu {
                err_msg: format!(
                    "Some regions are not opened, table: {}, table_id: {}",
                    table_name, table_id
                ),
            }
-            .fail();
-        }
+        });

        // Persist the physical table route.
        // TODO(weny): refetch the physical table route if repair is needed.
--- a/src/common/meta/src/reconciliation/reconcile_table/resolve_column_metadata.rs
+++ b/src/common/meta/src/reconciliation/reconcile_table/resolve_column_metadata.rs
@@ -20,6 +20,7 @@ use common_telemetry::info;
 use serde::{Deserialize, Serialize};
 use snafu::OptionExt;
 use store_api::metadata::RegionMetadata;
+use strum::AsRefStr;

 use crate::error::{self, MissingColumnIdsSnafu, Result};
 use crate::reconciliation::reconcile_table::reconcile_regions::ReconcileRegions;
@@ -28,10 +29,11 @@ use crate::reconciliation::reconcile_table::{ReconcileTableContext, State};
 use crate::reconciliation::utils::{
    build_column_metadata_from_table_info, check_column_metadatas_consistent,
    resolve_column_metadatas_with_latest, resolve_column_metadatas_with_metasrv,
+    ResolveColumnMetadataResult,
 };

 /// Strategy for resolving column metadata inconsistencies.
-#[derive(Debug, Serialize, Deserialize, Clone, Copy, Default)]
+#[derive(Debug, Serialize, Deserialize, Clone, Copy, Default, AsRefStr)]
 pub enum ResolveStrategy {
    #[default]
    /// Trusts the latest column metadata from datanode.
@@ -98,6 +100,10 @@ impl State for ResolveColumnMetadata {
                "Column metadatas are consistent for table: {}, table_id: {}.",
                table_name, table_id
            );
+
+            // Update metrics.
+            ctx.mut_metrics().resolve_column_metadata_result =
+                Some(ResolveColumnMetadataResult::Consistent);
            return Ok((
                Box::new(UpdateTableInfo::new(table_info_value, column_metadatas)),
                Status::executing(false),
@@ -119,6 +125,11 @@ impl State for ResolveColumnMetadata {

                let region_ids =
                    resolve_column_metadatas_with_metasrv(&column_metadata, &self.region_metadata)?;
+
+                // Update metrics.
+                let metrics = ctx.mut_metrics();
+                metrics.resolve_column_metadata_result =
+                    Some(ResolveColumnMetadataResult::Inconsistent(self.strategy));
                Ok((
                    Box::new(ReconcileRegions::new(column_metadata, region_ids)),
                    Status::executing(true),
@@ -127,16 +138,29 @@ impl State for ResolveColumnMetadata {
            ResolveStrategy::UseLatest => {
                let (column_metadatas, region_ids) =
                    resolve_column_metadatas_with_latest(&self.region_metadata)?;
+
+                // Update metrics.
+                let metrics = ctx.mut_metrics();
+                metrics.resolve_column_metadata_result =
+                    Some(ResolveColumnMetadataResult::Inconsistent(self.strategy));
                Ok((
                    Box::new(ReconcileRegions::new(column_metadatas, region_ids)),
                    Status::executing(true),
                ))
            }
-            ResolveStrategy::AbortOnConflict => error::ColumnMetadataConflictsSnafu {
-                table_name: table_name.to_string(),
-                table_id,
+            ResolveStrategy::AbortOnConflict => {
+                let table_name = table_name.to_string();
+
+                // Update metrics.
+                let metrics = ctx.mut_metrics();
+                metrics.resolve_column_metadata_result =
+                    Some(ResolveColumnMetadataResult::Inconsistent(self.strategy));
+                error::ColumnMetadataConflictsSnafu {
+                    table_name,
+                    table_id,
+                }
+                .fail()
            }
-            .fail(),
        }
    }

--- a/src/common/meta/src/reconciliation/reconcile_table/update_table_info.rs
+++ b/src/common/meta/src/reconciliation/reconcile_table/update_table_info.rs
@@ -116,6 +116,9 @@ impl State for UpdateTableInfo {
                ],
            )
            .await?;
+        // Update metrics.
+        let metrics = ctx.mut_metrics();
+        metrics.update_table_info = true;

        Ok((Box::new(ReconciliationEnd), Status::executing(true)))
    }
--- a/src/common/meta/src/reconciliation/utils.rs
+++ b/src/common/meta/src/reconciliation/utils.rs
@@ -13,23 +13,35 @@
 // limitations under the License.

 use std::collections::{HashMap, HashSet};
-use std::fmt;
+use std::fmt::{self, Display};
+use std::ops::AddAssign;
+use std::time::Instant;

 use api::v1::SemanticType;
-use common_telemetry::warn;
+use common_procedure::{watcher, Context as ProcedureContext, ProcedureId};
+use common_telemetry::{error, warn};
 use datatypes::schema::ColumnSchema;
-use snafu::{ensure, OptionExt};
+use futures::future::{join_all, try_join_all};
+use snafu::{ensure, OptionExt, ResultExt};
 use store_api::metadata::{ColumnMetadata, RegionMetadata};
 use store_api::storage::{RegionId, TableId};
 use table::metadata::{RawTableInfo, RawTableMeta};
+use table::table_name::TableName;
 use table::table_reference::TableReference;

 use crate::cache_invalidator::CacheInvalidatorRef;
 use crate::error::{
-    MismatchColumnIdSnafu, MissingColumnInColumnMetadataSnafu, Result, UnexpectedSnafu,
+    ColumnIdMismatchSnafu, ColumnNotFoundSnafu, MismatchColumnIdSnafu,
+    MissingColumnInColumnMetadataSnafu, ProcedureStateReceiverNotFoundSnafu,
+    ProcedureStateReceiverSnafu, Result, TimestampMismatchSnafu, UnexpectedSnafu,
+    WaitProcedureSnafu,
 };
 use crate::key::TableMetadataManagerRef;
+use crate::metrics;
 use crate::node_manager::NodeManagerRef;
+use crate::reconciliation::reconcile_logical_tables::ReconcileLogicalTablesProcedure;
+use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy;
+use crate::reconciliation::reconcile_table::ReconcileTableProcedure;

 #[derive(Debug, PartialEq, Eq)]
 pub(crate) struct PartialRegionMetadata<'a> {
@@ -48,20 +60,6 @@ impl<'a> From<&'a RegionMetadata> for PartialRegionMetadata<'a> {
    }
 }

-/// A display wrapper for [`ColumnMetadata`] that formats the column metadata in a more readable way.
-struct ColumnMetadataDisplay<'a>(pub &'a ColumnMetadata);
-
-impl<'a> fmt::Debug for ColumnMetadataDisplay<'a> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        let col = self.0;
-        write!(
-            f,
-            "Column {{ name: {}, id: {}, semantic_type: {:?}, data_type: {:?} }}",
-            col.column_schema.name, col.column_id, col.semantic_type, col.column_schema.data_type,
-        )
-    }
-}
-
 /// Checks if the column metadatas are consistent.
 ///
 /// The column metadatas are consistent if:
@@ -110,21 +108,7 @@ pub(crate) fn resolve_column_metadatas_with_metasrv(
    let mut regions_ids = vec![];
    for region_metadata in region_metadatas {
        if region_metadata.column_metadatas != column_metadatas {
-            let is_invariant_preserved = check_column_metadata_invariants(
-                column_metadatas,
-                &region_metadata.column_metadatas,
-            );
-            ensure!(
-                is_invariant_preserved,
-                UnexpectedSnafu {
-                    err_msg: format!(
-                        "Column metadata invariants violated for region {}. Resolved column metadata: {:?}, region column metadata: {:?}",
-                        region_metadata.region_id,
-                        column_metadatas.iter().map(ColumnMetadataDisplay).collect::<Vec<_>>(),
-                        region_metadata.column_metadatas.iter().map(ColumnMetadataDisplay).collect::<Vec<_>>(),
-                    )
-                }
-            );
+            check_column_metadata_invariants(column_metadatas, &region_metadata.column_metadatas)?;
            regions_ids.push(region_metadata.region_id);
        }
    }
@@ -163,21 +147,10 @@ pub(crate) fn resolve_column_metadatas_with_latest(
    let mut region_ids = vec![];
    for region_metadata in region_metadatas {
        if PartialRegionMetadata::from(region_metadata) != latest_column_metadatas {
-            let is_invariant_preserved = check_column_metadata_invariants(
+            check_column_metadata_invariants(
                &latest_region_metadata.column_metadatas,
                &region_metadata.column_metadatas,
-            );
-            ensure!(
-                is_invariant_preserved,
-                UnexpectedSnafu {
-                    err_msg: format!(
-                        "Column metadata invariants violated for region {}. Resolved column metadata: {:?}, region column metadata: {:?}",
-                        region_metadata.region_id,
-                        latest_column_metadatas.column_metadatas.iter().map(ColumnMetadataDisplay).collect::<Vec<_>>(),
-                        region_metadata.column_metadatas.iter().map(ColumnMetadataDisplay).collect::<Vec<_>>()
-                    )
-                }
-            );
+            )?;
            region_ids.push(region_metadata.region_id);
        }
    }
@@ -239,7 +212,7 @@ pub(crate) fn build_column_metadata_from_table_info(
 pub(crate) fn check_column_metadata_invariants(
    new_column_metadatas: &[ColumnMetadata],
    column_metadatas: &[ColumnMetadata],
-) -> bool {
+) -> Result<()> {
    let new_primary_keys = new_column_metadatas
        .iter()
        .filter(|c| c.semantic_type == SemanticType::Tag)
@@ -252,22 +225,50 @@ pub(crate) fn check_column_metadata_invariants(
        .map(|c| (c.column_schema.name.as_str(), c.column_id));

    for (name, id) in old_primary_keys {
-        if new_primary_keys.get(name) != Some(&id) {
-            return false;
-        }
+        let column_id = new_primary_keys
+            .get(name)
+            .cloned()
+            .context(ColumnNotFoundSnafu {
+                column_name: name,
+                column_id: id,
+            })?;
+
+        ensure!(
+            column_id == id,
+            ColumnIdMismatchSnafu {
+                column_name: name,
+                expected_column_id: id,
+                actual_column_id: column_id,
+            }
+        );
    }

    let new_ts_column = new_column_metadatas
        .iter()
        .find(|c| c.semantic_type == SemanticType::Timestamp)
-        .map(|c| (c.column_schema.name.as_str(), c.column_id));
+        .map(|c| (c.column_schema.name.as_str(), c.column_id))
+        .context(UnexpectedSnafu {
+            err_msg: "Timestamp column not found in new column metadata",
+        })?;

    let old_ts_column = column_metadatas
        .iter()
        .find(|c| c.semantic_type == SemanticType::Timestamp)
-        .map(|c| (c.column_schema.name.as_str(), c.column_id));
+        .map(|c| (c.column_schema.name.as_str(), c.column_id))
+        .context(UnexpectedSnafu {
+            err_msg: "Timestamp column not found in column metadata",
+        })?;
+    ensure!(
+        new_ts_column == old_ts_column,
+        TimestampMismatchSnafu {
+            expected_column_name: old_ts_column.0,
+            expected_column_id: old_ts_column.1,
+            actual_column_name: new_ts_column.0,
+            actual_column_id: new_ts_column.1,
+        }
+    );

-    new_ts_column == old_ts_column
+    Ok(())
 }

 /// Builds a [`RawTableMeta`] from the provided [`ColumnMetadata`]s.
@@ -406,6 +407,88 @@ pub(crate) fn need_update_logical_table_info(
    table_info.meta.schema.column_schemas.len() != column_metadatas.len()
 }

+/// The result of waiting for inflight subprocedures.
+pub struct PartialSuccessResult<'a> {
+    pub failed_procedures: Vec<&'a SubprocedureMeta>,
+    pub success_procedures: Vec<&'a SubprocedureMeta>,
+}
+
+/// The result of waiting for inflight subprocedures.
+pub enum WaitForInflightSubproceduresResult<'a> {
+    Success(Vec<&'a SubprocedureMeta>),
+    PartialSuccess(PartialSuccessResult<'a>),
+}
+
+/// Wait for inflight subprocedures.
+///
+/// If `fail_fast` is true, the function will return an error if any subprocedure fails.
+/// Otherwise, the function will continue waiting for all subprocedures to complete.
+pub(crate) async fn wait_for_inflight_subprocedures<'a>(
+    procedure_ctx: &ProcedureContext,
+    subprocedures: &'a [SubprocedureMeta],
+    fail_fast: bool,
+) -> Result<WaitForInflightSubproceduresResult<'a>> {
+    let mut receivers = Vec::with_capacity(subprocedures.len());
+    for subprocedure in subprocedures {
+        let procedure_id = subprocedure.procedure_id();
+        let receiver = procedure_ctx
+            .provider
+            .procedure_state_receiver(procedure_id)
+            .await
+            .context(ProcedureStateReceiverSnafu { procedure_id })?
+            .context(ProcedureStateReceiverNotFoundSnafu { procedure_id })?;
+        receivers.push((receiver, subprocedure));
+    }
+
+    let mut tasks = Vec::with_capacity(receivers.len());
+    for (receiver, subprocedure) in receivers.iter_mut() {
+        tasks.push(async move {
+            watcher::wait(receiver).await.inspect_err(|e| {
+                error!(e; "inflight subprocedure failed, parent procedure_id: {}, procedure: {}", procedure_ctx.procedure_id, subprocedure);
+            })
+        });
+    }
+
+    if fail_fast {
+        try_join_all(tasks).await.context(WaitProcedureSnafu)?;
+        return Ok(WaitForInflightSubproceduresResult::Success(
+            subprocedures.iter().collect(),
+        ));
+    }
+
+    // If fail_fast is false, we need to wait for all subprocedures to complete.
+    let results = join_all(tasks).await;
+    let failed_procedures_num = results.iter().filter(|r| r.is_err()).count();
+    if failed_procedures_num == 0 {
+        return Ok(WaitForInflightSubproceduresResult::Success(
+            subprocedures.iter().collect(),
+        ));
+    }
+    warn!(
+        "{} inflight subprocedures failed, total: {}, parent procedure_id: {}",
+        failed_procedures_num,
+        subprocedures.len(),
+        procedure_ctx.procedure_id
+    );
+
+    let mut failed_procedures = Vec::with_capacity(failed_procedures_num);
+    let mut success_procedures = Vec::with_capacity(subprocedures.len() - failed_procedures_num);
+    for (result, subprocedure) in results.into_iter().zip(subprocedures) {
+        if result.is_err() {
+            failed_procedures.push(subprocedure);
+        } else {
+            success_procedures.push(subprocedure);
+        }
+    }
+
+    Ok(WaitForInflightSubproceduresResult::PartialSuccess(
+        PartialSuccessResult {
+            failed_procedures,
+            success_procedures,
+        },
+    ))
+}
+
 #[derive(Clone)]
 pub struct Context {
    pub node_manager: NodeManagerRef,
@@ -413,6 +496,446 @@ pub struct Context {
    pub cache_invalidator: CacheInvalidatorRef,
 }

+/// Metadata for an inflight physical table subprocedure.
+pub struct PhysicalTableMeta {
+    pub procedure_id: ProcedureId,
+    pub table_id: TableId,
+    pub table_name: TableName,
+}
+
+/// Metadata for an inflight logical table subprocedure.
+pub struct LogicalTableMeta {
+    pub procedure_id: ProcedureId,
+    pub physical_table_id: TableId,
+    pub physical_table_name: TableName,
+    pub logical_tables: Vec<(TableId, TableName)>,
+}
+
+/// Metadata for an inflight database subprocedure.
+pub struct ReconcileDatabaseMeta {
+    pub procedure_id: ProcedureId,
+    pub catalog: String,
+    pub schema: String,
+}
+
+/// The inflight subprocedure metadata.
+pub enum SubprocedureMeta {
+    PhysicalTable(PhysicalTableMeta),
+    LogicalTable(LogicalTableMeta),
+    Database(ReconcileDatabaseMeta),
+}
+
+impl Display for SubprocedureMeta {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            SubprocedureMeta::PhysicalTable(meta) => {
+                write!(
+                    f,
+                    "ReconcilePhysicalTable(procedure_id: {}, table_id: {}, table_name: {})",
+                    meta.procedure_id, meta.table_id, meta.table_name
+                )
+            }
+            SubprocedureMeta::LogicalTable(meta) => {
+                write!(
+                    f,
+                    "ReconcileLogicalTable(procedure_id: {}, physical_table_id: {}, physical_table_name: {}, logical_tables: {:?})",
+                    meta.procedure_id, meta.physical_table_id, meta.physical_table_name, meta.logical_tables
+                )
+            }
+            SubprocedureMeta::Database(meta) => {
+                write!(
+                    f,
+                    "ReconcileDatabase(procedure_id: {}, catalog: {}, schema: {})",
+                    meta.procedure_id, meta.catalog, meta.schema
+                )
+            }
+        }
+    }
+}
+
+impl SubprocedureMeta {
+    /// Creates a new logical table subprocedure metadata.
+    pub fn new_logical_table(
+        procedure_id: ProcedureId,
+        physical_table_id: TableId,
+        physical_table_name: TableName,
+        logical_tables: Vec<(TableId, TableName)>,
+    ) -> Self {
+        Self::LogicalTable(LogicalTableMeta {
+            procedure_id,
+            physical_table_id,
+            physical_table_name,
+            logical_tables,
+        })
+    }
+
+    /// Creates a new physical table subprocedure metadata.
+    pub fn new_physical_table(
+        procedure_id: ProcedureId,
+        table_id: TableId,
+        table_name: TableName,
+    ) -> Self {
+        Self::PhysicalTable(PhysicalTableMeta {
+            procedure_id,
+            table_id,
+            table_name,
+        })
+    }
+
+    /// Creates a new reconcile database subprocedure metadata.
+    pub fn new_reconcile_database(
+        procedure_id: ProcedureId,
+        catalog: String,
+        schema: String,
+    ) -> Self {
+        Self::Database(ReconcileDatabaseMeta {
+            procedure_id,
+            catalog,
+            schema,
+        })
+    }
+
+    /// Returns the procedure id of the subprocedure.
+    pub fn procedure_id(&self) -> ProcedureId {
+        match self {
+            SubprocedureMeta::PhysicalTable(meta) => meta.procedure_id,
+            SubprocedureMeta::LogicalTable(meta) => meta.procedure_id,
+            SubprocedureMeta::Database(meta) => meta.procedure_id,
+        }
+    }
+
+    /// Returns the number of tables will be reconciled.
+    pub fn table_num(&self) -> usize {
+        match self {
+            SubprocedureMeta::PhysicalTable(_) => 1,
+            SubprocedureMeta::LogicalTable(meta) => meta.logical_tables.len(),
+            SubprocedureMeta::Database(_) => 0,
+        }
+    }
+
+    /// Returns the number of databases will be reconciled.
+    pub fn database_num(&self) -> usize {
+        match self {
+            SubprocedureMeta::Database(_) => 1,
+            _ => 0,
+        }
+    }
+}
+
+/// The metrics of reconciling catalog.
+#[derive(Clone, Default)]
+pub struct ReconcileCatalogMetrics {
+    pub succeeded_databases: usize,
+    pub failed_databases: usize,
+}
+
+impl AddAssign for ReconcileCatalogMetrics {
+    fn add_assign(&mut self, other: Self) {
+        self.succeeded_databases += other.succeeded_databases;
+        self.failed_databases += other.failed_databases;
+    }
+}
+
+impl Display for ReconcileCatalogMetrics {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "succeeded_databases: {}, failed_databases: {}",
+            self.succeeded_databases, self.failed_databases
+        )
+    }
+}
+
+impl From<WaitForInflightSubproceduresResult<'_>> for ReconcileCatalogMetrics {
+    fn from(result: WaitForInflightSubproceduresResult<'_>) -> Self {
+        match result {
+            WaitForInflightSubproceduresResult::Success(subprocedures) => ReconcileCatalogMetrics {
+                succeeded_databases: subprocedures.len(),
+                failed_databases: 0,
+            },
+            WaitForInflightSubproceduresResult::PartialSuccess(PartialSuccessResult {
+                failed_procedures,
+                success_procedures,
+            }) => {
+                let succeeded_databases = success_procedures
+                    .iter()
+                    .map(|subprocedure| subprocedure.database_num())
+                    .sum();
+                let failed_databases = failed_procedures
+                    .iter()
+                    .map(|subprocedure| subprocedure.database_num())
+                    .sum();
+                ReconcileCatalogMetrics {
+                    succeeded_databases,
+                    failed_databases,
+                }
+            }
+        }
+    }
+}
+
+/// The metrics of reconciling database.
+#[derive(Clone, Default)]
+pub struct ReconcileDatabaseMetrics {
+    pub succeeded_tables: usize,
+    pub failed_tables: usize,
+    pub succeeded_procedures: usize,
+    pub failed_procedures: usize,
+}
+
+impl Display for ReconcileDatabaseMetrics {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "succeeded_tables: {}, failed_tables: {}, succeeded_procedures: {}, failed_procedures: {}", self.succeeded_tables, self.failed_tables, self.succeeded_procedures, self.failed_procedures)
+    }
+}
+
+impl AddAssign for ReconcileDatabaseMetrics {
+    fn add_assign(&mut self, other: Self) {
+        self.succeeded_tables += other.succeeded_tables;
+        self.failed_tables += other.failed_tables;
+        self.succeeded_procedures += other.succeeded_procedures;
+        self.failed_procedures += other.failed_procedures;
+    }
+}
+
+impl From<WaitForInflightSubproceduresResult<'_>> for ReconcileDatabaseMetrics {
+    fn from(result: WaitForInflightSubproceduresResult<'_>) -> Self {
+        match result {
+            WaitForInflightSubproceduresResult::Success(subprocedures) => {
+                let table_num = subprocedures
+                    .iter()
+                    .map(|subprocedure| subprocedure.table_num())
+                    .sum();
+                ReconcileDatabaseMetrics {
+                    succeeded_procedures: subprocedures.len(),
+                    failed_procedures: 0,
+                    succeeded_tables: table_num,
+                    failed_tables: 0,
+                }
+            }
+            WaitForInflightSubproceduresResult::PartialSuccess(PartialSuccessResult {
+                failed_procedures,
+                success_procedures,
+            }) => {
+                let succeeded_tables = success_procedures
+                    .iter()
+                    .map(|subprocedure| subprocedure.table_num())
+                    .sum();
+                let failed_tables = failed_procedures
+                    .iter()
+                    .map(|subprocedure| subprocedure.table_num())
+                    .sum();
+                ReconcileDatabaseMetrics {
+                    succeeded_procedures: success_procedures.len(),
+                    failed_procedures: failed_procedures.len(),
+                    succeeded_tables,
+                    failed_tables,
+                }
+            }
+        }
+    }
+}
+
+/// The metrics of reconciling logical tables.
+#[derive(Clone)]
+pub struct ReconcileLogicalTableMetrics {
+    pub start_time: Instant,
+    pub update_table_info_count: usize,
+    pub create_tables_count: usize,
+    pub column_metadata_consistent_count: usize,
+    pub column_metadata_inconsistent_count: usize,
+}
+
+impl Default for ReconcileLogicalTableMetrics {
+    fn default() -> Self {
+        Self {
+            start_time: Instant::now(),
+            update_table_info_count: 0,
+            create_tables_count: 0,
+            column_metadata_consistent_count: 0,
+            column_metadata_inconsistent_count: 0,
+        }
+    }
+}
+
+const CREATE_TABLES: &str = "create_tables";
+const UPDATE_TABLE_INFO: &str = "update_table_info";
+const COLUMN_METADATA_CONSISTENT: &str = "column_metadata_consistent";
+const COLUMN_METADATA_INCONSISTENT: &str = "column_metadata_inconsistent";
+
+impl ReconcileLogicalTableMetrics {
+    /// The total number of tables that have been reconciled.
+    pub fn total_table_count(&self) -> usize {
+        self.create_tables_count
+            + self.column_metadata_consistent_count
+            + self.column_metadata_inconsistent_count
+    }
+}
+
+impl Drop for ReconcileLogicalTableMetrics {
+    fn drop(&mut self) {
+        let procedure_name = ReconcileLogicalTablesProcedure::TYPE_NAME;
+        metrics::METRIC_META_RECONCILIATION_STATS
+            .with_label_values(&[procedure_name, metrics::TABLE_TYPE_LOGICAL, CREATE_TABLES])
+            .inc_by(self.create_tables_count as u64);
+        metrics::METRIC_META_RECONCILIATION_STATS
+            .with_label_values(&[
+                procedure_name,
+                metrics::TABLE_TYPE_LOGICAL,
+                UPDATE_TABLE_INFO,
+            ])
+            .inc_by(self.update_table_info_count as u64);
+        metrics::METRIC_META_RECONCILIATION_STATS
+            .with_label_values(&[
+                procedure_name,
+                metrics::TABLE_TYPE_LOGICAL,
+                COLUMN_METADATA_CONSISTENT,
+            ])
+            .inc_by(self.column_metadata_consistent_count as u64);
+        metrics::METRIC_META_RECONCILIATION_STATS
+            .with_label_values(&[
+                procedure_name,
+                metrics::TABLE_TYPE_LOGICAL,
+                COLUMN_METADATA_INCONSISTENT,
+            ])
+            .inc_by(self.column_metadata_inconsistent_count as u64);
+    }
+}
+
+impl Display for ReconcileLogicalTableMetrics {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let elapsed = self.start_time.elapsed();
+        if self.create_tables_count > 0 {
+            write!(f, "create_tables_count: {}, ", self.create_tables_count)?;
+        }
+        if self.update_table_info_count > 0 {
+            write!(
+                f,
+                "update_table_info_count: {}, ",
+                self.update_table_info_count
+            )?;
+        }
+        if self.column_metadata_consistent_count > 0 {
+            write!(
+                f,
+                "column_metadata_consistent_count: {}, ",
+                self.column_metadata_consistent_count
+            )?;
+        }
+        if self.column_metadata_inconsistent_count > 0 {
+            write!(
+                f,
+                "column_metadata_inconsistent_count: {}, ",
+                self.column_metadata_inconsistent_count
+            )?;
+        }
+
+        write!(
+            f,
+            "total_table_count: {}, elapsed: {:?}",
+            self.total_table_count(),
+            elapsed
+        )
+    }
+}
+
+/// The result of resolving column metadata.
+#[derive(Clone, Copy)]
+pub enum ResolveColumnMetadataResult {
+    Consistent,
+    Inconsistent(ResolveStrategy),
+}
+
+impl Display for ResolveColumnMetadataResult {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ResolveColumnMetadataResult::Consistent => write!(f, "Consistent"),
+            ResolveColumnMetadataResult::Inconsistent(strategy) => {
+                let strategy_str = strategy.as_ref();
+                write!(f, "Inconsistent({})", strategy_str)
+            }
+        }
+    }
+}
+
+/// The metrics of reconciling physical tables.
+#[derive(Clone)]
+pub struct ReconcileTableMetrics {
+    /// The start time of the reconciliation.
+    pub start_time: Instant,
+    /// The result of resolving column metadata.
+    pub resolve_column_metadata_result: Option<ResolveColumnMetadataResult>,
+    /// Whether the table info has been updated.
+    pub update_table_info: bool,
+}
+
+impl Drop for ReconcileTableMetrics {
+    fn drop(&mut self) {
+        if let Some(resolve_column_metadata_result) = self.resolve_column_metadata_result {
+            match resolve_column_metadata_result {
+                ResolveColumnMetadataResult::Consistent => {
+                    metrics::METRIC_META_RECONCILIATION_STATS
+                        .with_label_values(&[
+                            ReconcileTableProcedure::TYPE_NAME,
+                            metrics::TABLE_TYPE_PHYSICAL,
+                            COLUMN_METADATA_CONSISTENT,
+                        ])
+                        .inc();
+                }
+                ResolveColumnMetadataResult::Inconsistent(strategy) => {
+                    metrics::METRIC_META_RECONCILIATION_STATS
+                        .with_label_values(&[
+                            ReconcileTableProcedure::TYPE_NAME,
+                            metrics::TABLE_TYPE_PHYSICAL,
+                            COLUMN_METADATA_INCONSISTENT,
+                        ])
+                        .inc();
+                    metrics::METRIC_META_RECONCILIATION_RESOLVED_COLUMN_METADATA
+                        .with_label_values(&[strategy.as_ref()])
+                        .inc();
+                }
+            }
+        }
+        if self.update_table_info {
+            metrics::METRIC_META_RECONCILIATION_STATS
+                .with_label_values(&[
+                    ReconcileTableProcedure::TYPE_NAME,
+                    metrics::TABLE_TYPE_PHYSICAL,
+                    UPDATE_TABLE_INFO,
+                ])
+                .inc();
+        }
+    }
+}
+
+impl Default for ReconcileTableMetrics {
+    fn default() -> Self {
+        Self {
+            start_time: Instant::now(),
+            resolve_column_metadata_result: None,
+            update_table_info: false,
+        }
+    }
+}
+
+impl Display for ReconcileTableMetrics {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let elapsed = self.start_time.elapsed();
+        if let Some(resolve_column_metadata_result) = self.resolve_column_metadata_result {
+            write!(
+                f,
+                "resolve_column_metadata_result: {}, ",
+                resolve_column_metadata_result
+            )?;
+        }
+        write!(
+            f,
+            "update_table_info: {}, elapsed: {:?}",
+            self.update_table_info, elapsed
+        )
+    }
+}
+
 #[cfg(test)]
 mod tests {
    use std::assert_matches::assert_matches;
@@ -665,10 +1188,7 @@ mod tests {
            semantic_type: SemanticType::Field,
            column_id: 3,
        });
-        assert!(check_column_metadata_invariants(
-            &new_column_metadatas,
-            &column_metadatas
-        ));
+        check_column_metadata_invariants(&new_column_metadatas, &column_metadatas).unwrap();
    }

    #[test]
@@ -676,18 +1196,12 @@ mod tests {
        let column_metadatas = new_test_column_metadatas();
        let mut new_column_metadatas = column_metadatas.clone();
        new_column_metadatas.retain(|c| c.semantic_type != SemanticType::Timestamp);
-        assert!(!check_column_metadata_invariants(
-            &new_column_metadatas,
-            &column_metadatas
-        ));
+        check_column_metadata_invariants(&new_column_metadatas, &column_metadatas).unwrap_err();

        let column_metadatas = new_test_column_metadatas();
        let mut new_column_metadatas = column_metadatas.clone();
        new_column_metadatas.retain(|c| c.semantic_type != SemanticType::Tag);
-        assert!(!check_column_metadata_invariants(
-            &new_column_metadatas,
-            &column_metadatas
-        ));
+        check_column_metadata_invariants(&new_column_metadatas, &column_metadatas).unwrap_err();
    }

    #[test]
@@ -700,10 +1214,7 @@ mod tests {
        {
            col.column_id = 100;
        }
-        assert!(!check_column_metadata_invariants(
-            &new_column_metadatas,
-            &column_metadatas
-        ));
+        check_column_metadata_invariants(&new_column_metadatas, &column_metadatas).unwrap_err();

        let column_metadatas = new_test_column_metadatas();
        let mut new_column_metadatas = column_metadatas.clone();
@@ -713,10 +1224,7 @@ mod tests {
        {
            col.column_id = 100;
        }
-        assert!(!check_column_metadata_invariants(
-            &new_column_metadatas,
-            &column_metadatas
-        ));
+        check_column_metadata_invariants(&new_column_metadatas, &column_metadatas).unwrap_err();
    }

    #[test]
--- a/src/datanode/src/datanode.rs
+++ b/src/datanode/src/datanode.rs
@@ -669,16 +669,24 @@ async fn open_all_regions(
            ignore_nonexistent_region,
        )
        .await?;
-    ensure!(
-        open_regions.len() == num_regions,
-        error::UnexpectedSnafu {
-            violated: format!(
-                "Expected to open {} of regions, only {} of regions has opened",
-                num_regions,
-                open_regions.len()
-            )
-        }
-    );
+    if !ignore_nonexistent_region {
+        ensure!(
+            open_regions.len() == num_regions,
+            error::UnexpectedSnafu {
+                violated: format!(
+                    "Expected to open {} of regions, only {} of regions has opened",
+                    num_regions,
+                    open_regions.len()
+                )
+            }
+        );
+    } else if open_regions.len() != num_regions {
+        warn!(
+            "ignore nonexistent region, expected to open {} of regions, only {} of regions has opened",
+            num_regions,
+            open_regions.len()
+        );
+    }

    for region_id in open_regions {
        if open_with_writable {
@@ -719,16 +727,24 @@ async fn open_all_regions(
            )
            .await?;

-        ensure!(
-            open_regions.len() == num_regions,
-            error::UnexpectedSnafu {
-                violated: format!(
-                    "Expected to open {} of follower regions, only {} of regions has opened",
-                    num_regions,
-                    open_regions.len()
-                )
-            }
-        );
+        if !ignore_nonexistent_region {
+            ensure!(
+                open_regions.len() == num_regions,
+                error::UnexpectedSnafu {
+                    violated: format!(
+                        "Expected to open {} of follower regions, only {} of regions has opened",
+                        num_regions,
+                        open_regions.len()
+                    )
+                }
+            );
+        } else if open_regions.len() != num_regions {
+            warn!(
+                "ignore nonexistent region, expected to open {} of follower regions, only {} of regions has opened",
+                num_regions,
+                open_regions.len()
+            );
+        }
    }

    info!("all regions are opened");
--- a/src/metric-engine/Cargo.toml
+++ b/src/metric-engine/Cargo.toml
@@ -29,6 +29,7 @@ itertools.workspace = true
 lazy_static = "1.4"
 mito-codec.workspace = true
 mito2.workspace = true
+moka.workspace = true
 mur3 = "0.1"
 object-store.workspace = true
 prometheus.workspace = true
--- a/src/metric-engine/src/error.rs
+++ b/src/metric-engine/src/error.rs
@@ -13,6 +13,7 @@
 // limitations under the License.

 use std::any::Any;
+use std::sync::Arc;

 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
@@ -304,6 +305,13 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
+
+    #[snafu(display("Get value from cache"))]
+    CacheGet {
+        source: Arc<Error>,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }

 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -362,6 +370,8 @@ impl ErrorExt for Error {
            StartRepeatedTask { source, .. } => source.status_code(),

            MetricManifestInfo { .. } => StatusCode::Internal,
+
+            CacheGet { source, .. } => source.status_code(),
        }
    }

--- a/src/metric-engine/src/metadata_region.rs
+++ b/src/metric-engine/src/metadata_region.rs
@@ -13,19 +13,23 @@
 // limitations under the License.

 use std::collections::hash_map::Entry;
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 use std::sync::Arc;
+use std::time::Duration;

 use api::v1::value::ValueData;
 use api::v1::{ColumnDataType, ColumnSchema, Row, Rows, SemanticType, Value};
 use async_stream::try_stream;
 use base64::engine::general_purpose::STANDARD_NO_PAD;
 use base64::Engine;
+use common_base::readable_size::ReadableSize;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
 use datafusion::prelude::{col, lit};
 use futures_util::stream::BoxStream;
 use futures_util::TryStreamExt;
 use mito2::engine::MitoEngine;
+use moka::future::Cache;
+use moka::policy::EvictionPolicy;
 use snafu::{OptionExt, ResultExt};
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::{
@@ -39,9 +43,9 @@ use store_api::storage::{RegionId, ScanRequest};
 use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock};

 use crate::error::{
-    CollectRecordBatchStreamSnafu, DecodeColumnValueSnafu, DeserializeColumnMetadataSnafu,
-    LogicalRegionNotFoundSnafu, MitoReadOperationSnafu, MitoWriteOperationSnafu,
-    ParseRegionIdSnafu, Result,
+    CacheGetSnafu, CollectRecordBatchStreamSnafu, DecodeColumnValueSnafu,
+    DeserializeColumnMetadataSnafu, LogicalRegionNotFoundSnafu, MitoReadOperationSnafu,
+    MitoWriteOperationSnafu, ParseRegionIdSnafu, Result,
 };
 use crate::utils;

@@ -62,6 +66,11 @@ const COLUMN_PREFIX: &str = "__column_";
 /// itself.
 pub struct MetadataRegion {
    pub(crate) mito: MitoEngine,
+    /// The cache for contents(key-value pairs) of region metadata.
+    ///
+    /// The cache should be invalidated when any new values are put into the metadata region or any
+    /// values are deleted from the metadata region.
+    cache: Cache<RegionId, RegionMetadataCacheEntry>,
    /// Logical lock for operations that need to be serialized. Like update & read region columns.
    ///
    /// Region entry will be registered on creating and opening logical region, and deregistered on
@@ -69,10 +78,30 @@ pub struct MetadataRegion {
    logical_region_lock: RwLock<HashMap<RegionId, Arc<RwLock<()>>>>,
 }

+#[derive(Clone)]
+struct RegionMetadataCacheEntry {
+    key_values: Arc<BTreeMap<String, String>>,
+    size: usize,
+}
+
+/// The max size of the region metadata cache.
+const MAX_CACHE_SIZE: u64 = ReadableSize::mb(128).as_bytes();
+/// The TTL of the region metadata cache.
+const CACHE_TTL: Duration = Duration::from_secs(5 * 60);
+
 impl MetadataRegion {
    pub fn new(mito: MitoEngine) -> Self {
+        let cache = Cache::builder()
+            .max_capacity(MAX_CACHE_SIZE)
+            // Use the LRU eviction policy to minimize frequent mito scans.
+            // Recently accessed items are retained longer in the cache.
+            .eviction_policy(EvictionPolicy::lru())
+            .time_to_live(CACHE_TTL)
+            .weigher(|_, v: &RegionMetadataCacheEntry| v.size as u32)
+            .build();
        Self {
            mito,
+            cache,
            logical_region_lock: RwLock::new(HashMap::new()),
        }
    }
@@ -351,21 +380,60 @@ impl MetadataRegion {
        }
    }

-    pub async fn get_all_with_prefix(
-        &self,
-        region_id: RegionId,
-        prefix: &str,
-    ) -> Result<HashMap<String, String>> {
-        let scan_req = MetadataRegion::build_prefix_read_request(prefix, false);
+    fn build_read_request() -> ScanRequest {
+        let projection = vec![
+            METADATA_SCHEMA_KEY_COLUMN_INDEX,
+            METADATA_SCHEMA_VALUE_COLUMN_INDEX,
+        ];
+        ScanRequest {
+            projection: Some(projection),
+            ..Default::default()
+        }
+    }
+
+    async fn load_all(&self, metadata_region_id: RegionId) -> Result<RegionMetadataCacheEntry> {
+        let scan_req = MetadataRegion::build_read_request();
        let record_batch_stream = self
            .mito
-            .scan_to_stream(region_id, scan_req)
+            .scan_to_stream(metadata_region_id, scan_req)
            .await
            .context(MitoReadOperationSnafu)?;

-        decode_batch_stream(record_batch_stream, decode_record_batch_to_key_and_value)
-            .try_collect::<HashMap<_, _>>()
+        let kv = decode_batch_stream(record_batch_stream, decode_record_batch_to_key_and_value)
+            .try_collect::<BTreeMap<_, _>>()
+            .await?;
+        let mut size = 0;
+        for (k, v) in kv.iter() {
+            size += k.len();
+            size += v.len();
+        }
+        let kv = Arc::new(kv);
+        Ok(RegionMetadataCacheEntry {
+            key_values: kv,
+            size,
+        })
+    }
+
+    async fn get_all_with_prefix(
+        &self,
+        metadata_region_id: RegionId,
+        prefix: &str,
+    ) -> Result<HashMap<String, String>> {
+        let region_metadata = self
+            .cache
+            .try_get_with(metadata_region_id, self.load_all(metadata_region_id))
            .await
+            .context(CacheGetSnafu)?;
+
+        let range = region_metadata.key_values.range(prefix.to_string()..);
+        let mut result = HashMap::new();
+        for (k, v) in range {
+            if !k.starts_with(prefix) {
+                break;
+            }
+            result.insert(k.to_string(), v.to_string());
+        }
+        Ok(result)
    }

    pub async fn get_all_key_with_prefix(
@@ -387,15 +455,18 @@ impl MetadataRegion {

    /// Delete the given keys. For performance consideration, this method
    /// doesn't check if those keys exist or not.
-    async fn delete(&self, region_id: RegionId, keys: &[String]) -> Result<()> {
+    async fn delete(&self, metadata_region_id: RegionId, keys: &[String]) -> Result<()> {
        let delete_request = Self::build_delete_request(keys);
        self.mito
            .handle_request(
-                region_id,
+                metadata_region_id,
                store_api::region_request::RegionRequest::Delete(delete_request),
            )
            .await
            .context(MitoWriteOperationSnafu)?;
+        // Invalidates the region metadata cache if any values are deleted from the metadata region.
+        self.cache.invalidate(&metadata_region_id).await;
+
        Ok(())
    }

@@ -485,7 +556,7 @@ impl MetadataRegion {
        write_region_id: bool,
        logical_regions: impl Iterator<Item = (RegionId, HashMap<&str, &ColumnMetadata>)>,
    ) -> Result<()> {
-        let region_id = utils::to_metadata_region_id(physical_region_id);
+        let metadata_region_id = utils::to_metadata_region_id(physical_region_id);
        let iter = logical_regions
            .into_iter()
            .flat_map(|(logical_region_id, column_metadatas)| {
@@ -512,11 +583,13 @@ impl MetadataRegion {
        let put_request = MetadataRegion::build_put_request_from_iter(iter.into_iter());
        self.mito
            .handle_request(
-                region_id,
+                metadata_region_id,
                store_api::region_request::RegionRequest::Put(put_request),
            )
            .await
            .context(MitoWriteOperationSnafu)?;
+        // Invalidates the region metadata cache if any new values are put into the metadata region.
+        self.cache.invalidate(&metadata_region_id).await;

        Ok(())
    }
--- a/src/mito2/src/compaction/compactor.rs
+++ b/src/mito2/src/compaction/compactor.rs
@@ -90,7 +90,8 @@ pub struct CompactionRegion {
    pub(crate) engine_config: Arc<MitoConfig>,
    pub(crate) region_metadata: RegionMetadataRef,
    pub(crate) cache_manager: CacheManagerRef,
-    pub(crate) access_layer: AccessLayerRef,
+    /// Access layer to get the table path and path type.
+    pub access_layer: AccessLayerRef,
    pub(crate) manifest_ctx: Arc<ManifestContext>,
    pub(crate) current_version: CompactionVersion,
    pub(crate) file_purger: Option<Arc<LocalFilePurger>>,
--- a/src/query/src/promql/planner.rs
+++ b/src/query/src/promql/planner.rs
@@ -23,6 +23,7 @@ use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_function::function::FunctionContext;
 use common_query::prelude::GREPTIME_VALUE;
+use common_telemetry::debug;
 use datafusion::common::DFSchemaRef;
 use datafusion::datasource::DefaultTableSource;
 use datafusion::functions_aggregate::average::avg_udaf;
@@ -661,10 +662,30 @@ impl PromPlanner {
                    }
                    Ok(binary_expr)
                };
+
+                fn optimize(plan: &LogicalPlan) -> LogicalPlan {
+                    use datafusion_optimizer::OptimizerRule;
+                    let new_plan =
+                        datafusion::optimizer::optimize_projections::OptimizeProjections::new()
+                            .rewrite(
+                                plan.clone(),
+                                &datafusion::optimizer::OptimizerContext::default(),
+                            )
+                            .unwrap()
+                            .data;
+                    if new_plan != *plan {
+                        debug!(
+                            "Optimized projection plan: {new_plan:#?}\n From old plan: {plan:#?}"
+                        );
+                    }
+                    new_plan
+                }
+
                if is_comparison_op && !should_return_bool {
                    self.filter_on_field_column(join_plan, bin_expr_builder)
                } else {
                    self.projection_for_each_field_column(join_plan, bin_expr_builder)
+                        .map(|p| optimize(&p))
                }
            }
        }
@@ -3289,6 +3310,8 @@ mod test {
    use common_base::Plugins;
    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
    use common_query::test_util::DummyDecoder;
+    use datafusion::functions_aggregate::count::count;
+    use datafusion_optimizer::OptimizerContext;
    use datatypes::prelude::ConcreteDataType;
    use datatypes::schema::{ColumnSchema, Schema};
    use promql_parser::label::Labels;
@@ -4909,6 +4932,132 @@ Filter: up.field_0 IS NOT NULL [timestamp:Timestamp(Millisecond, None), field_0:
        assert_eq!(plan.display_indent_schema().to_string(), expected);
    }

+    #[tokio::test]
+    async fn test_nested_aggr_not_exists_table_label() {
+        let mut eval_stmt = EvalStmt {
+            expr: PromExpr::NumberLiteral(NumberLiteral { val: 1.0 }),
+            start: UNIX_EPOCH,
+            end: UNIX_EPOCH
+                .checked_add(Duration::from_secs(100_000))
+                .unwrap(),
+            interval: Duration::from_secs(5),
+            lookback_delta: Duration::from_secs(1),
+        };
+        let case = r#"count(count(node_cpu_seconds_total)) / node_load5"#;
+
+        let prom_expr = parser::parse(case).unwrap();
+        eval_stmt.expr = prom_expr;
+        let table_provider = build_test_table_provider_with_fields(
+            &[(DEFAULT_SCHEMA_NAME.to_string(), "metric_exists".to_string())],
+            &["job"],
+        )
+        .await;
+
+        let plan =
+            PromPlanner::stmt_to_plan(table_provider, &eval_stmt, &build_query_engine_state())
+                .await
+                .unwrap();
+        let expected = r#"Projection: lhs.time, lhs.count(count(.value)) / rhs.value AS lhs.count(count(.value)) / rhs.value [time:Timestamp(Millisecond, None), lhs.count(count(.value)) / rhs.value:Float64;N]
+  Inner Join: lhs.time = rhs.time [time:Timestamp(Millisecond, None), count(.value):Int64, count(count(.value)):Int64, time:Timestamp(Millisecond, None), value:Float64;N]
+    SubqueryAlias: lhs [time:Timestamp(Millisecond, None), count(.value):Int64, count(count(.value)):Int64]
+      Sort: .time ASC NULLS LAST [time:Timestamp(Millisecond, None), count(.value):Int64, count(count(.value)):Int64]
+        Aggregate: groupBy=[[.time, count(.value)]], aggr=[[count(count(.value))]] [time:Timestamp(Millisecond, None), count(.value):Int64, count(count(.value)):Int64]
+          Sort: .time ASC NULLS LAST [time:Timestamp(Millisecond, None), count(.value):Int64]
+            Aggregate: groupBy=[[.time]], aggr=[[count(.value)]] [time:Timestamp(Millisecond, None), count(.value):Int64]
+              EmptyMetric: range=[0..-1], interval=[5000] [time:Timestamp(Millisecond, None), value:Float64;N]
+    SubqueryAlias: rhs [time:Timestamp(Millisecond, None), value:Float64;N]
+      EmptyMetric: range=[0..-1], interval=[5000] [time:Timestamp(Millisecond, None), value:Float64;N]"#;
+
+        let rhs = LogicalPlanBuilder::from(LogicalPlan::Extension(Extension {
+            node: Arc::new(
+                EmptyMetric::new(
+                    0,
+                    -1,
+                    5000,
+                    "time".to_string(),
+                    "value".to_string(),
+                    Some(lit(0.0f64)),
+                )
+                .unwrap(),
+            ),
+        }))
+        .alias("rhs")
+        .unwrap()
+        .build()
+        .unwrap();
+
+        let full = LogicalPlanBuilder::from(LogicalPlan::Extension(Extension {
+            node: Arc::new(
+                EmptyMetric::new(
+                    0,
+                    -1,
+                    5000,
+                    "time".to_string(),
+                    "value".to_string(),
+                    Some(lit(0.0f64)),
+                )
+                .unwrap(),
+            ),
+        }))
+        .aggregate(
+            vec![col(Column::new(Some(""), "time"))],
+            vec![count(col(Column::new(Some(""), "value")))],
+        )
+        .unwrap()
+        .sort(vec![SortExpr::new(
+            col(Column::new(Some(""), "time")),
+            true,
+            false,
+        )])
+        .unwrap()
+        .aggregate(
+            vec![col(Column::new(Some(""), "time"))],
+            vec![count(col("count(.value)"))],
+        )
+        .unwrap()
+        .sort(vec![SortExpr::new(
+            col(Column::new(Some(""), "time")),
+            true,
+            false,
+        )])
+        .unwrap()
+        .alias("lhs")
+        .unwrap()
+        .project(vec![
+            col("lhs.time"),
+            col(Column::new(Some("lhs"), "count(count(.value))")),
+        ])
+        .unwrap()
+        .join(
+            rhs,
+            JoinType::Inner,
+            (
+                vec![Column::new(Some("lhs"), "time")],
+                vec![Column::new(Some("rhs"), "time")],
+            ),
+            None,
+        )
+        .unwrap()
+        .build()
+        .unwrap();
+
+        dbg!(&full);
+        {
+            let optimizer = datafusion_optimizer::Optimizer::new();
+            let optimized_full_plan = optimizer
+                .optimize(full, &OptimizerContext::default(), |_, _| {})
+                .unwrap();
+        }
+        return;
+
+        assert_eq!(plan.display_indent_schema().to_string(), expected);
+        let optimizer = datafusion_optimizer::Optimizer::new();
+        let optimized_plan = optimizer
+            .optimize(plan, &OptimizerContext::default(), |_, _| {})
+            .unwrap();
+        println!("{}", optimized_plan.display_indent_schema().to_string());
+    }
+
    #[tokio::test]
    async fn test_histogram_quantile_missing_le_column() {
        let mut eval_stmt = EvalStmt {
--- a/src/servers/Cargo.toml
+++ b/src/servers/Cargo.toml
@@ -89,10 +89,7 @@ opensrv-mysql = { git = "https://github.com/datafuselabs/opensrv", rev = "a1fb4d
 opentelemetry-proto.workspace = true
 otel-arrow-rust.workspace = true
 parking_lot.workspace = true
-#pgwire = { version = "0.30", default-features = false, features = ["server-api-ring"] }
-pgwire = { git = "https://github.com/sunng87/pgwire", rev = "127573d997228cfb70c7699881c568eae8131270", default-features = false, features = [
-    "server-api-ring",
-] }
+pgwire = { version = "0.32", default-features = false, features = ["server-api-ring"] }
 pin-project = "1.0"
 pipeline.workspace = true
 postgres-types = { version = "0.2", features = ["with-chrono-0_4", "with-serde_json-1"] }
--- a/src/servers/src/postgres.rs
+++ b/src/servers/src/postgres.rs
@@ -32,9 +32,9 @@ use std::sync::Arc;

 use ::auth::UserProviderRef;
 use derive_builder::Builder;
-use pgwire::api::auth::ServerParameterProvider;
-use pgwire::api::copy::NoopCopyHandler;
-use pgwire::api::{ClientInfo, PgWireServerHandlers};
+use pgwire::api::auth::{ServerParameterProvider, StartupHandler};
+use pgwire::api::query::{ExtendedQueryHandler, SimpleQueryHandler};
+use pgwire::api::{ClientInfo, ErrorHandler, PgWireServerHandlers};
 pub use server::PostgresServer;
 use session::context::Channel;
 use session::Session;
@@ -92,29 +92,19 @@ pub(crate) struct MakePostgresServerHandler {
 pub(crate) struct PostgresServerHandler(Arc<PostgresServerHandlerInner>);

 impl PgWireServerHandlers for PostgresServerHandler {
-    type StartupHandler = PostgresServerHandlerInner;
-    type SimpleQueryHandler = PostgresServerHandlerInner;
-    type ExtendedQueryHandler = PostgresServerHandlerInner;
-    type CopyHandler = NoopCopyHandler;
-    type ErrorHandler = PostgresServerHandlerInner;
-
-    fn simple_query_handler(&self) -> Arc<Self::SimpleQueryHandler> {
+    fn simple_query_handler(&self) -> Arc<impl SimpleQueryHandler> {
        self.0.clone()
    }

-    fn extended_query_handler(&self) -> Arc<Self::ExtendedQueryHandler> {
+    fn extended_query_handler(&self) -> Arc<impl ExtendedQueryHandler> {
        self.0.clone()
    }

-    fn startup_handler(&self) -> Arc<Self::StartupHandler> {
+    fn startup_handler(&self) -> Arc<impl StartupHandler> {
        self.0.clone()
    }

-    fn copy_handler(&self) -> Arc<Self::CopyHandler> {
-        Arc::new(NoopCopyHandler)
-    }
-
-    fn error_handler(&self) -> Arc<Self::ErrorHandler> {
+    fn error_handler(&self) -> Arc<impl ErrorHandler> {
        self.0.clone()
    }
 }
--- a/src/servers/src/postgres/auth_handler.rs
+++ b/src/servers/src/postgres/auth_handler.rs
@@ -24,7 +24,7 @@ use pgwire::api::auth::StartupHandler;
 use pgwire::api::{auth, ClientInfo, PgWireConnectionState};
 use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};
 use pgwire::messages::response::ErrorResponse;
-use pgwire::messages::startup::Authentication;
+use pgwire::messages::startup::{Authentication, SecretKey};
 use pgwire::messages::{PgWireBackendMessage, PgWireFrontendMessage};
 use session::Session;
 use snafu::IntoError;
@@ -127,7 +127,8 @@ where

    // pass generated process id and secret key to client, this information will
    // be sent to postgres client for query cancellation.
-    client.set_pid_and_secret_key(session.process_id() as i32, rand::random::<i32>());
+    // use all 0 before we actually supported query cancellation
+    client.set_pid_and_secret_key(0, SecretKey::I32(0));
    // set userinfo outside
 }