fix: closee issue #6555 return empty result (#6569 )

* fix: closee issue #6555 return empty result Signed-off-by: yihong0618 <zouzou0208@gmail.com> * fix: only start one instance one regrex sqlness test (#6570) Signed-off-by: yihong0618 <zouzou0208@gmail.com> * refactor: refactor partition mod to use PartitionExpr instead of PartitionDef (#6554) * refactor: refactor partition mod to use PartitionExpr instead of PartitionDef Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> * fix snafu Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> * Puts expression into PbPartition Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> * address comments Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> * fix compile Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> * update proto Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> * add serde test Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> * add serde test Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> --------- Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> * fix: address comments Signed-off-by: yihong0618 <zouzou0208@gmail.com> --------- Signed-off-by: yihong0618 <zouzou0208@gmail.com> Signed-off-by: Zhenchi <zhongzc_arch@outlook.com> Co-authored-by: Zhenchi <zhongzc_arch@outlook.com> Signed-off-by: evenyag <realevenyag@gmail.com>
chore: bump version to 0.15.3 (#6580 )
2025-12-23 06:30:05 +00:00 · 2025-07-24 15:00:32 +08:00 · 2025-07-24 11:24:07 +08:00 · 2025-07-23 22:29:14 +08:00 · 2025-07-23 20:54:33 +08:00 · 2025-07-23 20:54:33 +08:00
223 changed files with 11336 additions and 2360 deletions
--- a/.cargo/config.toml
+++ b/.cargo/config.toml
@@ -12,3 +12,6 @@ fetch = true
 checkout = true
 list_files = true
 internal_use_git2 = false
+
+[env]
+CARGO_WORKSPACE_DIR = { value = "", relative = true }
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -211,7 +211,7 @@ checksum = "d301b3b94cb4b2f23d7917810addbbaff90738e0ca2be692bd027e70d7e0330c"

 [[package]]
 name = "api"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "common-base",
 "common-decimal",
@@ -944,7 +944,7 @@ dependencies = [

 [[package]]
 name = "auth"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "async-trait",
@@ -1586,7 +1586,7 @@ dependencies = [

 [[package]]
 name = "cache"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "catalog",
 "common-error",
@@ -1602,6 +1602,17 @@ version = "1.0.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "acbc26382d871df4b7442e3df10a9402bf3cf5e55cbd66f12be38861425f0564"

+[[package]]
+name = "cargo-manifest"
+version = "0.19.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a1d8af896b707212cd0e99c112a78c9497dd32994192a463ed2f7419d29bd8c6"
+dependencies = [
+ "serde",
+ "thiserror 2.0.12",
+ "toml 0.8.19",
+]
+
 [[package]]
 name = "cast"
 version = "0.3.0"
@@ -1610,7 +1621,7 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"

 [[package]]
 name = "catalog"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "arrow 54.2.1",
@@ -1948,7 +1959,7 @@ checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97"

 [[package]]
 name = "cli"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -1993,7 +2004,7 @@ dependencies = [
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "table",
 "tempfile",
 "tokio",
@@ -2002,7 +2013,7 @@ dependencies = [

 [[package]]
 name = "client"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "arc-swap",
@@ -2032,7 +2043,7 @@ dependencies = [
 "rand 0.9.0",
 "serde_json",
 "snafu 0.8.5",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "substrait 0.37.3",
 "tokio",
 "tokio-stream",
@@ -2073,7 +2084,7 @@ dependencies = [

 [[package]]
 name = "cmd"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-trait",
 "auth",
@@ -2134,7 +2145,7 @@ dependencies = [
 "snafu 0.8.5",
 "stat",
 "store-api",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "table",
 "temp-env",
 "tempfile",
@@ -2181,7 +2192,7 @@ checksum = "55b672471b4e9f9e95499ea597ff64941a309b2cdbffcc46f2cc5e2d971fd335"

 [[package]]
 name = "common-base"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "anymap2",
 "async-trait",
@@ -2203,11 +2214,11 @@ dependencies = [

 [[package]]
 name = "common-catalog"
-version = "0.15.0"
+version = "0.15.3"

 [[package]]
 name = "common-config"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "common-base",
 "common-error",
@@ -2232,7 +2243,7 @@ dependencies = [

 [[package]]
 name = "common-datasource"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "arrow 54.2.1",
 "arrow-schema 54.3.1",
@@ -2269,7 +2280,7 @@ dependencies = [

 [[package]]
 name = "common-decimal"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "bigdecimal 0.4.8",
 "common-error",
@@ -2282,7 +2293,7 @@ dependencies = [

 [[package]]
 name = "common-error"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "common-macro",
 "http 1.1.0",
@@ -2293,7 +2304,7 @@ dependencies = [

 [[package]]
 name = "common-frontend"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-trait",
 "common-error",
@@ -2309,7 +2320,7 @@ dependencies = [

 [[package]]
 name = "common-function"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -2362,7 +2373,7 @@ dependencies = [

 [[package]]
 name = "common-greptimedb-telemetry"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-trait",
 "common-runtime",
@@ -2379,7 +2390,7 @@ dependencies = [

 [[package]]
 name = "common-grpc"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "arrow-flight",
@@ -2411,7 +2422,7 @@ dependencies = [

 [[package]]
 name = "common-grpc-expr"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "common-base",
@@ -2430,7 +2441,7 @@ dependencies = [

 [[package]]
 name = "common-macro"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "arc-swap",
 "common-query",
@@ -2444,7 +2455,7 @@ dependencies = [

 [[package]]
 name = "common-mem-prof"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "anyhow",
 "common-error",
@@ -2460,7 +2471,7 @@ dependencies = [

 [[package]]
 name = "common-meta"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "anymap2",
 "api",
@@ -2525,7 +2536,7 @@ dependencies = [

 [[package]]
 name = "common-options"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "common-grpc",
 "humantime-serde",
@@ -2534,11 +2545,11 @@ dependencies = [

 [[package]]
 name = "common-plugins"
-version = "0.15.0"
+version = "0.15.3"

 [[package]]
 name = "common-pprof"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2550,7 +2561,7 @@ dependencies = [

 [[package]]
 name = "common-procedure"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -2577,7 +2588,7 @@ dependencies = [

 [[package]]
 name = "common-procedure-test"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-trait",
 "common-procedure",
@@ -2586,7 +2597,7 @@ dependencies = [

 [[package]]
 name = "common-query"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "async-trait",
@@ -2612,7 +2623,7 @@ dependencies = [

 [[package]]
 name = "common-recordbatch"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "arc-swap",
 "common-error",
@@ -2632,7 +2643,7 @@ dependencies = [

 [[package]]
 name = "common-runtime"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -2662,17 +2673,18 @@ dependencies = [

 [[package]]
 name = "common-session"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "strum 0.27.1",
 ]

 [[package]]
 name = "common-telemetry"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "backtrace",
 "common-error",
+ "common-version",
 "console-subscriber",
 "greptime-proto",
 "humantime-serde",
@@ -2696,7 +2708,7 @@ dependencies = [

 [[package]]
 name = "common-test-util"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "client",
 "common-grpc",
@@ -2709,7 +2721,7 @@ dependencies = [

 [[package]]
 name = "common-time"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "arrow 54.2.1",
 "chrono",
@@ -2727,9 +2739,10 @@ dependencies = [

 [[package]]
 name = "common-version"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "build-data",
+ "cargo-manifest",
 "const_format",
 "serde",
 "shadow-rs",
@@ -2737,7 +2750,7 @@ dependencies = [

 [[package]]
 name = "common-wal"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "common-base",
 "common-error",
@@ -2760,7 +2773,7 @@ dependencies = [

 [[package]]
 name = "common-workload"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "common-telemetry",
@@ -3716,7 +3729,7 @@ dependencies = [

 [[package]]
 name = "datanode"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "arrow-flight",
@@ -3769,7 +3782,7 @@ dependencies = [
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "table",
 "tokio",
 "toml 0.8.19",
@@ -3778,7 +3791,7 @@ dependencies = [

 [[package]]
 name = "datatypes"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "arrow 54.2.1",
 "arrow-array 54.2.1",
@@ -4438,7 +4451,7 @@ checksum = "e8c02a5121d4ea3eb16a80748c74f5549a5665e4c21333c6098f283870fbdea6"

 [[package]]
 name = "file-engine"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "async-trait",
@@ -4575,7 +4588,7 @@ checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8"

 [[package]]
 name = "flow"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "arrow 54.2.1",
@@ -4640,7 +4653,7 @@ dependencies = [
 "sql",
 "store-api",
 "strum 0.27.1",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "table",
 "tokio",
 "tonic 0.12.3",
@@ -4695,10 +4708,11 @@ checksum = "6c2141d6d6c8512188a7891b4b01590a45f6dac67afb4f255c4124dbb86d4eaa"

 [[package]]
 name = "frontend"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "arc-swap",
+ "async-stream",
 "async-trait",
 "auth",
 "bytes",
@@ -4754,7 +4768,7 @@ dependencies = [
 "sqlparser 0.54.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=0cf6c04490d59435ee965edd2078e8855bd8471e)",
 "store-api",
 "strfmt",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "table",
 "tokio",
 "tokio-util",
@@ -5144,7 +5158,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=82fe5c6282f623c185b86f03e898ee8952e50cf9#82fe5c6282f623c185b86f03e898ee8952e50cf9"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=96c733f8472284d3c83a4c011dc6de9cf830c353#96c733f8472284d3c83a4c011dc6de9cf830c353"
 dependencies = [
 "prost 0.13.5",
 "serde",
@@ -5915,7 +5929,7 @@ dependencies = [

 [[package]]
 name = "index"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-trait",
 "asynchronous-codec",
@@ -6800,7 +6814,7 @@ checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"

 [[package]]
 name = "log-query"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "chrono",
 "common-error",
@@ -6812,7 +6826,7 @@ dependencies = [

 [[package]]
 name = "log-store"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -7110,7 +7124,7 @@ dependencies = [

 [[package]]
 name = "meta-client"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "async-trait",
@@ -7138,7 +7152,7 @@ dependencies = [

 [[package]]
 name = "meta-srv"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "async-trait",
@@ -7229,7 +7243,7 @@ dependencies = [

 [[package]]
 name = "metric-engine"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "aquamarine",
@@ -7319,7 +7333,7 @@ dependencies = [

 [[package]]
 name = "mito-codec"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "bytes",
@@ -7342,7 +7356,7 @@ dependencies = [

 [[package]]
 name = "mito2"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "aquamarine",
@@ -8092,7 +8106,7 @@ dependencies = [

 [[package]]
 name = "object-store"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "anyhow",
 "bytes",
@@ -8406,7 +8420,7 @@ dependencies = [

 [[package]]
 name = "operator"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -8461,7 +8475,7 @@ dependencies = [
 "sql",
 "sqlparser 0.54.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=0cf6c04490d59435ee965edd2078e8855bd8471e)",
 "store-api",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "table",
 "tokio",
 "tokio-util",
@@ -8728,7 +8742,7 @@ dependencies = [

 [[package]]
 name = "partition"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "async-trait",
@@ -9016,7 +9030,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"

 [[package]]
 name = "pipeline"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -9159,7 +9173,7 @@ dependencies = [

 [[package]]
 name = "plugins"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "auth",
 "clap 4.5.19",
@@ -9472,7 +9486,7 @@ dependencies = [

 [[package]]
 name = "promql"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "ahash 0.8.11",
 "async-trait",
@@ -9568,7 +9582,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
 dependencies = [
 "heck 0.5.0",
- "itertools 0.14.0",
+ "itertools 0.11.0",
 "log",
 "multimap",
 "once_cell",
@@ -9614,7 +9628,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
 dependencies = [
 "anyhow",
- "itertools 0.14.0",
+ "itertools 0.11.0",
 "proc-macro2",
 "quote",
 "syn 2.0.100",
@@ -9754,7 +9768,7 @@ dependencies = [

 [[package]]
 name = "puffin"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-compression 0.4.13",
 "async-trait",
@@ -9796,7 +9810,7 @@ dependencies = [

 [[package]]
 name = "query"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -9862,7 +9876,7 @@ dependencies = [
 "sqlparser 0.54.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=0cf6c04490d59435ee965edd2078e8855bd8471e)",
 "statrs",
 "store-api",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "table",
 "tokio",
 "tokio-stream",
@@ -11148,7 +11162,7 @@ dependencies = [

 [[package]]
 name = "servers"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -11269,7 +11283,7 @@ dependencies = [

 [[package]]
 name = "session"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "arc-swap",
@@ -11608,7 +11622,7 @@ dependencies = [

 [[package]]
 name = "sql"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "chrono",
@@ -11663,7 +11677,7 @@ dependencies = [

 [[package]]
 name = "sqlness-runner"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -11963,7 +11977,7 @@ dependencies = [

 [[package]]
 name = "stat"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "nix 0.30.1",
 ]
@@ -11989,7 +12003,7 @@ dependencies = [

 [[package]]
 name = "store-api"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "aquamarine",
@@ -12150,7 +12164,7 @@ dependencies = [

 [[package]]
 name = "substrait"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "async-trait",
 "bytes",
@@ -12330,7 +12344,7 @@ dependencies = [

 [[package]]
 name = "table"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "async-trait",
@@ -12591,7 +12605,7 @@ checksum = "3369f5ac52d5eb6ab48c6b4ffdc8efbcad6b89c765749064ba298f2c68a16a76"

 [[package]]
 name = "tests-fuzz"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "arbitrary",
 "async-trait",
@@ -12635,7 +12649,7 @@ dependencies = [

 [[package]]
 name = "tests-integration"
-version = "0.15.0"
+version = "0.15.3"
 dependencies = [
 "api",
 "arrow-flight",
@@ -12702,7 +12716,7 @@ dependencies = [
 "sql",
 "sqlx",
 "store-api",
- "substrait 0.15.0",
+ "substrait 0.15.3",
 "table",
 "tempfile",
 "time",
@@ -13072,6 +13086,7 @@ version = "0.8.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a1ed1f98e3fdc28d6d910e6737ae6ab1a93bf1985935a1193e68f93eeb68d24e"
 dependencies = [
+ "indexmap 2.9.0",
 "serde",
 "serde_spanned",
 "toml_datetime",
@@ -14183,7 +14198,7 @@ version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb"
 dependencies = [
- "windows-sys 0.59.0",
+ "windows-sys 0.48.0",
 ]

 [[package]]
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -71,7 +71,7 @@ members = [
 resolver = "2"

 [workspace.package]
-version = "0.15.0"
+version = "0.15.3"
 edition = "2021"
 license = "Apache-2.0"

@@ -134,7 +134,7 @@ etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "82fe5c6282f623c185b86f03e898ee8952e50cf9" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "96c733f8472284d3c83a4c011dc6de9cf830c353" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
--- a/src/api/src/v1/column_def.rs
+++ b/src/api/src/v1/column_def.rs
@@ -226,18 +226,20 @@ mod tests {
        assert!(options.is_none());

        let mut schema = ColumnSchema::new("test", ConcreteDataType::string_datatype(), true)
-            .with_fulltext_options(FulltextOptions {
-                enable: true,
-                analyzer: FulltextAnalyzer::English,
-                case_sensitive: false,
-                backend: FulltextBackend::Bloom,
-            })
+            .with_fulltext_options(FulltextOptions::new_unchecked(
+                true,
+                FulltextAnalyzer::English,
+                false,
+                FulltextBackend::Bloom,
+                10240,
+                0.01,
+            ))
            .unwrap();
        schema.set_inverted_index(true);
        let options = options_from_column_schema(&schema).unwrap();
        assert_eq!(
            options.options.get(FULLTEXT_GRPC_KEY).unwrap(),
-            "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\"}"
+            "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\",\"granularity\":10240,\"false-positive-rate-in-10000\":100}"
        );
        assert_eq!(
            options.options.get(INVERTED_INDEX_GRPC_KEY).unwrap(),
@@ -247,16 +249,18 @@ mod tests {

    #[test]
    fn test_options_with_fulltext() {
-        let fulltext = FulltextOptions {
-            enable: true,
-            analyzer: FulltextAnalyzer::English,
-            case_sensitive: false,
-            backend: FulltextBackend::Bloom,
-        };
+        let fulltext = FulltextOptions::new_unchecked(
+            true,
+            FulltextAnalyzer::English,
+            false,
+            FulltextBackend::Bloom,
+            10240,
+            0.01,
+        );
        let options = options_from_fulltext(&fulltext).unwrap().unwrap();
        assert_eq!(
            options.options.get(FULLTEXT_GRPC_KEY).unwrap(),
-            "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\"}"
+            "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\",\"granularity\":10240,\"false-positive-rate-in-10000\":100}"
        );
    }

--- a/src/catalog/src/kvbackend/manager.rs
+++ b/src/catalog/src/kvbackend/manager.rs
@@ -28,7 +28,7 @@ use common_meta::cache::{
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::schema_name::SchemaNameKey;
-use common_meta::key::table_info::TableInfoValue;
+use common_meta::key::table_info::{TableInfoManager, TableInfoValue};
 use common_meta::key::table_name::TableNameKey;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
@@ -39,6 +39,7 @@ use moka::sync::Cache;
 use partition::manager::{PartitionRuleManager, PartitionRuleManagerRef};
 use session::context::{Channel, QueryContext};
 use snafu::prelude::*;
+use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
 use table::dist_table::DistTable;
 use table::metadata::TableId;
 use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
@@ -142,6 +143,61 @@ impl KvBackendCatalogManager {
    pub fn procedure_manager(&self) -> Option<ProcedureManagerRef> {
        self.procedure_manager.clone()
    }
+
+    // Override logical table's partition key indices with physical table's.
+    async fn override_logical_table_partition_key_indices(
+        table_route_cache: &TableRouteCacheRef,
+        table_info_manager: &TableInfoManager,
+        table: TableRef,
+    ) -> Result<TableRef> {
+        // If the table is not a metric table, return the table directly.
+        if table.table_info().meta.engine != METRIC_ENGINE_NAME {
+            return Ok(table);
+        }
+
+        if let Some(table_route_value) = table_route_cache
+            .get(table.table_info().table_id())
+            .await
+            .context(TableMetadataManagerSnafu)?
+            && let TableRoute::Logical(logical_route) = &*table_route_value
+            && let Some(physical_table_info_value) = table_info_manager
+                .get(logical_route.physical_table_id())
+                .await
+                .context(TableMetadataManagerSnafu)?
+        {
+            let mut new_table_info = (*table.table_info()).clone();
+
+            // Remap partition key indices from physical table to logical table
+            new_table_info.meta.partition_key_indices = physical_table_info_value
+                .table_info
+                .meta
+                .partition_key_indices
+                .iter()
+                .filter_map(|&physical_index| {
+                    // Get the column name from the physical table using the physical index
+                    physical_table_info_value
+                        .table_info
+                        .meta
+                        .schema
+                        .column_schemas
+                        .get(physical_index)
+                        .and_then(|physical_column| {
+                            // Find the corresponding index in the logical table schema
+                            new_table_info
+                                .meta
+                                .schema
+                                .column_index_by_name(physical_column.name.as_str())
+                        })
+                })
+                .collect();
+
+            let new_table = DistTable::table(Arc::new(new_table_info));
+
+            return Ok(new_table);
+        }
+
+        Ok(table)
+    }
 }

 #[async_trait::async_trait]
@@ -268,10 +324,7 @@ impl CatalogManager for KvBackendCatalogManager {
        let table_cache: TableCacheRef = self.cache_registry.get().context(CacheNotFoundSnafu {
            name: "table_cache",
        })?;
-        let table_route_cache: TableRouteCacheRef =
-            self.cache_registry.get().context(CacheNotFoundSnafu {
-                name: "table_route_cache",
-            })?;
+
        let table = table_cache
            .get_by_ref(&TableName {
                catalog_name: catalog_name.to_string(),
@@ -281,55 +334,18 @@ impl CatalogManager for KvBackendCatalogManager {
            .await
            .context(GetTableCacheSnafu)?;

-        // Override logical table's partition key indices with physical table's.
-        if let Some(table) = &table
-            && let Some(table_route_value) = table_route_cache
-                .get(table.table_info().table_id())
-                .await
-                .context(TableMetadataManagerSnafu)?
-            && let TableRoute::Logical(logical_route) = &*table_route_value
-            && let Some(physical_table_info_value) = self
-                .table_metadata_manager
-                .table_info_manager()
-                .get(logical_route.physical_table_id())
-                .await
-                .context(TableMetadataManagerSnafu)?
-        {
-            let mut new_table_info = (*table.table_info()).clone();
-            // Gather all column names from the logical table
-            let logical_column_names: std::collections::HashSet<_> = new_table_info
-                .meta
-                .schema
-                .column_schemas()
-                .iter()
-                .map(|col| &col.name)
-                .collect();
-
-            // Only preserve partition key indices where the corresponding columns exist in logical table
-            new_table_info.meta.partition_key_indices = physical_table_info_value
-                .table_info
-                .meta
-                .partition_key_indices
-                .iter()
-                .filter(|&&index| {
-                    if let Some(physical_column) = physical_table_info_value
-                        .table_info
-                        .meta
-                        .schema
-                        .column_schemas
-                        .get(index)
-                    {
-                        logical_column_names.contains(&physical_column.name)
-                    } else {
-                        false
-                    }
-                })
-                .cloned()
-                .collect();
-
-            let new_table = DistTable::table(Arc::new(new_table_info));
-
-            return Ok(Some(new_table));
+        if let Some(table) = table {
+            let table_route_cache: TableRouteCacheRef =
+                self.cache_registry.get().context(CacheNotFoundSnafu {
+                    name: "table_route_cache",
+                })?;
+            return Self::override_logical_table_partition_key_indices(
+                &table_route_cache,
+                self.table_metadata_manager.table_info_manager(),
+                table,
+            )
+            .await
+            .map(Some);
        }

        if channel == Channel::Postgres {
@@ -342,7 +358,7 @@ impl CatalogManager for KvBackendCatalogManager {
            }
        }

-        Ok(table)
+        Ok(None)
    }

    async fn tables_by_ids(
@@ -394,8 +410,20 @@ impl CatalogManager for KvBackendCatalogManager {
        let catalog = catalog.to_string();
        let schema = schema.to_string();
        let semaphore = Arc::new(Semaphore::new(CONCURRENCY));
+        let table_route_cache: Result<TableRouteCacheRef> =
+            self.cache_registry.get().context(CacheNotFoundSnafu {
+                name: "table_route_cache",
+            });

        common_runtime::spawn_global(async move {
+            let table_route_cache = match table_route_cache {
+                Ok(table_route_cache) => table_route_cache,
+                Err(e) => {
+                    let _ = tx.send(Err(e)).await;
+                    return;
+                }
+            };
+
            let table_id_stream = metadata_manager
                .table_name_manager()
                .tables(&catalog, &schema)
@@ -422,6 +450,7 @@ impl CatalogManager for KvBackendCatalogManager {
                let metadata_manager = metadata_manager.clone();
                let tx = tx.clone();
                let semaphore = semaphore.clone();
+                let table_route_cache = table_route_cache.clone();
                common_runtime::spawn_global(async move {
                    // we don't explicitly close the semaphore so just ignore the potential error.
                    let _ = semaphore.acquire().await;
@@ -439,6 +468,16 @@ impl CatalogManager for KvBackendCatalogManager {
                    };

                    for table in table_info_values.into_values().map(build_table) {
+                        let table = if let Ok(table) = table {
+                            Self::override_logical_table_partition_key_indices(
+                                &table_route_cache,
+                                metadata_manager.table_info_manager(),
+                                table,
+                            )
+                            .await
+                        } else {
+                            table
+                        };
                        if tx.send(table).await.is_err() {
                            return;
                        }
--- a/src/catalog/src/process_manager.rs
+++ b/src/catalog/src/process_manager.rs
@@ -21,7 +21,7 @@ use std::sync::{Arc, RwLock};
 use api::v1::frontend::{KillProcessRequest, ListProcessRequest, ProcessInfo};
 use common_base::cancellation::CancellationHandle;
 use common_frontend::selector::{FrontendSelector, MetaClientSelector};
-use common_telemetry::{debug, info};
+use common_telemetry::{debug, info, warn};
 use common_time::util::current_time_millis;
 use meta_client::MetaClientRef;
 use snafu::{ensure, OptionExt, ResultExt};
@@ -141,14 +141,20 @@ impl ProcessManager {
                .await
                .context(error::InvokeFrontendSnafu)?;
            for mut f in frontends {
-                processes.extend(
-                    f.list_process(ListProcessRequest {
+                let result = f
+                    .list_process(ListProcessRequest {
                        catalog: catalog.unwrap_or_default().to_string(),
                    })
                    .await
-                    .context(error::InvokeFrontendSnafu)?
-                    .processes,
-                );
+                    .context(error::InvokeFrontendSnafu);
+                match result {
+                    Ok(resp) => {
+                        processes.extend(resp.processes);
+                    }
+                    Err(e) => {
+                        warn!(e; "Skipping failing node: {:?}", f)
+                    }
+                }
            }
        }
        processes.extend(self.local_processes(catalog)?);
--- a/src/client/src/database.rs
+++ b/src/client/src/database.rs
@@ -211,12 +211,18 @@ impl Database {
                        retries += 1;
                        warn!("Retrying {} times with error = {:?}", retries, err);
                        continue;
+                    } else {
+                        error!(
+                            err; "Failed to send request to grpc handle, retries = {}, not retryable error, aborting",
+                            retries
+                        );
+                        return Err(err.into());
                    }
                }
                (Err(err), false) => {
                    error!(
-                        "Failed to send request to grpc handle after {} retries, error = {:?}",
-                        retries, err
+                        err; "Failed to send request to grpc handle after {} retries",
+                        retries,
                    );
                    return Err(err.into());
                }
--- a/src/client/src/region.rs
+++ b/src/client/src/region.rs
@@ -163,19 +163,70 @@ impl RegionRequester {
            let _span = tracing_context.attach(common_telemetry::tracing::info_span!(
                "poll_flight_data_stream"
            ));
-            while let Some(flight_message) = flight_message_stream.next().await {
-                let flight_message = flight_message
-                    .map_err(BoxedError::new)
-                    .context(ExternalSnafu)?;
+
+            let mut buffered_message: Option<FlightMessage> = None;
+            let mut stream_ended = false;
+
+            while !stream_ended {
+                // get the next message from the buffered message or read from the flight message stream
+                let flight_message_item = if let Some(msg) = buffered_message.take() {
+                    Some(Ok(msg))
+                } else {
+                    flight_message_stream.next().await
+                };
+
+                let flight_message = match flight_message_item {
+                    Some(Ok(message)) => message,
+                    Some(Err(e)) => {
+                        yield Err(BoxedError::new(e)).context(ExternalSnafu);
+                        break;
+                    }
+                    None => break,
+                };

                match flight_message {
                    FlightMessage::RecordBatch(record_batch) => {
-                        yield RecordBatch::try_from_df_record_batch(
+                        let result_to_yield = RecordBatch::try_from_df_record_batch(
                            schema_cloned.clone(),
                            record_batch,
-                        )
+                        );
+
+                        // get the next message from the stream. normally it should be a metrics message.
+                        if let Some(next_flight_message_result) = flight_message_stream.next().await
+                        {
+                            match next_flight_message_result {
+                                Ok(FlightMessage::Metrics(s)) => {
+                                    let m = serde_json::from_str(&s).ok().map(Arc::new);
+                                    metrics_ref.swap(m);
+                                }
+                                Ok(FlightMessage::RecordBatch(rb)) => {
+                                    // for some reason it's not a metrics message, so we need to buffer this record batch
+                                    // and yield it in the next iteration.
+                                    buffered_message = Some(FlightMessage::RecordBatch(rb));
+                                }
+                                Ok(_) => {
+                                    yield IllegalFlightMessagesSnafu {
+                                        reason: "A RecordBatch message can only be succeeded by a Metrics message or another RecordBatch message"
+                                    }
+                                    .fail()
+                                    .map_err(BoxedError::new)
+                                    .context(ExternalSnafu);
+                                    break;
+                                }
+                                Err(e) => {
+                                    yield Err(BoxedError::new(e)).context(ExternalSnafu);
+                                    break;
+                                }
+                            }
+                        } else {
+                            // the stream has ended
+                            stream_ended = true;
+                        }
+
+                        yield result_to_yield;
                    }
                    FlightMessage::Metrics(s) => {
+                        // just a branch in case of some metrics message comes after other things.
                        let m = serde_json::from_str(&s).ok().map(Arc::new);
                        metrics_ref.swap(m);
                        break;
--- a/src/cmd/src/bin/greptime.rs
+++ b/src/cmd/src/bin/greptime.rs
@@ -20,11 +20,11 @@ use cmd::error::{InitTlsProviderSnafu, Result};
 use cmd::options::GlobalOptions;
 use cmd::{cli, datanode, flownode, frontend, metasrv, standalone, App};
 use common_base::Plugins;
-use common_version::version;
+use common_version::{verbose_version, version};
 use servers::install_ring_crypto_provider;

 #[derive(Parser)]
-#[command(name = "greptime", author, version, long_version = version(), about)]
+#[command(name = "greptime", author, version, long_version = verbose_version(), about)]
 #[command(propagate_version = true)]
 pub(crate) struct Command {
    #[clap(subcommand)]
@@ -143,10 +143,8 @@ async fn start(cli: Command) -> Result<()> {
 }

 fn setup_human_panic() {
-    human_panic::setup_panic!(
-        human_panic::Metadata::new("GreptimeDB", env!("CARGO_PKG_VERSION"))
-            .homepage("https://github.com/GreptimeTeam/greptimedb/discussions")
-    );
+    human_panic::setup_panic!(human_panic::Metadata::new("GreptimeDB", version())
+        .homepage("https://github.com/GreptimeTeam/greptimedb/discussions"));

    common_telemetry::set_panic_hook();
 }
--- a/src/cmd/src/datanode/builder.rs
+++ b/src/cmd/src/datanode/builder.rs
@@ -19,7 +19,7 @@ use catalog::kvbackend::MetaKvBackend;
 use common_base::Plugins;
 use common_meta::cache::LayeredCacheRegistryBuilder;
 use common_telemetry::info;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use datanode::datanode::DatanodeBuilder;
 use datanode::service::DatanodeServiceBuilder;
 use meta_client::MetaClientType;
@@ -67,7 +67,7 @@ impl InstanceBuilder {
            None,
        );

-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
        create_resource_limit_metrics(APP_NAME);

        plugins::setup_datanode_plugins(plugins, &opts.plugins, dn_opts)
--- a/src/cmd/src/flownode.rs
+++ b/src/cmd/src/flownode.rs
@@ -32,7 +32,7 @@ use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::TableMetadataManager;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use flow::{
    get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder,
    FrontendClient, FrontendInvoker,
@@ -279,7 +279,7 @@ impl StartCommand {
            None,
        );

-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
        create_resource_limit_metrics(APP_NAME);

        info!("Flownode start command: {:#?}", self);
--- a/src/cmd/src/frontend.rs
+++ b/src/cmd/src/frontend.rs
@@ -33,7 +33,7 @@ use common_meta::heartbeat::handler::HandlerGroupExecutor;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
 use common_time::timezone::set_default_timezone;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use frontend::frontend::Frontend;
 use frontend::heartbeat::HeartbeatTask;
 use frontend::instance::builder::FrontendBuilder;
@@ -282,7 +282,7 @@ impl StartCommand {
            opts.component.slow_query.as_ref(),
        );

-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
        create_resource_limit_metrics(APP_NAME);

        info!("Frontend start command: {:#?}", self);
--- a/src/cmd/src/lib.rs
+++ b/src/cmd/src/lib.rs
@@ -112,7 +112,7 @@ pub trait App: Send {
 pub fn log_versions(version: &str, short_version: &str, app: &str) {
    // Report app version as gauge.
    APP_VERSION
-        .with_label_values(&[env!("CARGO_PKG_VERSION"), short_version, app])
+        .with_label_values(&[common_version::version(), short_version, app])
        .inc();

    // Log version and argument flags.
--- a/src/cmd/src/metasrv.rs
+++ b/src/cmd/src/metasrv.rs
@@ -22,7 +22,7 @@ use common_base::Plugins;
 use common_config::Configurable;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use meta_srv::bootstrap::MetasrvInstance;
 use meta_srv::metasrv::BackendImpl;
 use snafu::ResultExt;
@@ -320,7 +320,7 @@ impl StartCommand {
            None,
        );

-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
        create_resource_limit_metrics(APP_NAME);

        info!("Metasrv start command: {:#?}", self);
--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -30,20 +30,16 @@ use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID};
 use common_config::{metadata_store_dir, Configurable, KvBackendConfig};
 use common_error::ext::BoxedError;
 use common_meta::cache::LayeredCacheRegistryBuilder;
-use common_meta::cache_invalidator::CacheInvalidatorRef;
 use common_meta::cluster::{NodeInfo, NodeStatus};
 use common_meta::datanode::RegionStat;
-use common_meta::ddl::flow_meta::{FlowMetadataAllocator, FlowMetadataAllocatorRef};
-use common_meta::ddl::table_meta::{TableMetadataAllocator, TableMetadataAllocatorRef};
+use common_meta::ddl::flow_meta::FlowMetadataAllocator;
+use common_meta::ddl::table_meta::TableMetadataAllocator;
 use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl, ProcedureExecutorRef};
 use common_meta::ddl_manager::DdlManager;
-#[cfg(feature = "enterprise")]
-use common_meta::ddl_manager::TriggerDdlManagerRef;
 use common_meta::key::flow::flow_state::FlowStat;
-use common_meta::key::flow::{FlowMetadataManager, FlowMetadataManagerRef};
+use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
-use common_meta::node_manager::NodeManagerRef;
 use common_meta::peer::Peer;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::region_registry::LeaderRegionRegistry;
@@ -55,7 +51,7 @@ use common_telemetry::logging::{
    LoggingOptions, SlowQueryOptions, TracingOptions, DEFAULT_LOGGING_DIR,
 };
 use common_time::timezone::set_default_timezone;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
 use datanode::datanode::{Datanode, DatanodeBuilder};
@@ -470,7 +466,7 @@ impl StartCommand {
            opts.component.slow_query.as_ref(),
        );

-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
        create_resource_limit_metrics(APP_NAME);

        info!("Standalone start command: {:#?}", self);
@@ -594,28 +590,36 @@ impl StartCommand {
            .await
            .context(error::BuildWalOptionsAllocatorSnafu)?;
        let wal_options_allocator = Arc::new(wal_options_allocator);
-        let table_meta_allocator = Arc::new(TableMetadataAllocator::new(
+        let table_metadata_allocator = Arc::new(TableMetadataAllocator::new(
            table_id_sequence,
            wal_options_allocator.clone(),
        ));
-        let flow_meta_allocator = Arc::new(FlowMetadataAllocator::with_noop_peer_allocator(
+        let flow_metadata_allocator = Arc::new(FlowMetadataAllocator::with_noop_peer_allocator(
            flow_id_sequence,
        ));

+        let ddl_context = DdlContext {
+            node_manager: node_manager.clone(),
+            cache_invalidator: layered_cache_registry.clone(),
+            memory_region_keeper: Arc::new(MemoryRegionKeeper::default()),
+            leader_region_registry: Arc::new(LeaderRegionRegistry::default()),
+            table_metadata_manager: table_metadata_manager.clone(),
+            table_metadata_allocator: table_metadata_allocator.clone(),
+            flow_metadata_manager: flow_metadata_manager.clone(),
+            flow_metadata_allocator: flow_metadata_allocator.clone(),
+            region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
+        };
+        let procedure_manager_c = procedure_manager.clone();
+
+        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager_c, true)
+            .context(error::InitDdlManagerSnafu)?;
        #[cfg(feature = "enterprise")]
-        let trigger_ddl_manager: Option<TriggerDdlManagerRef> = plugins.get();
-        let ddl_task_executor = Self::create_ddl_task_executor(
-            procedure_manager.clone(),
-            node_manager.clone(),
-            layered_cache_registry.clone(),
-            table_metadata_manager,
-            table_meta_allocator,
-            flow_metadata_manager,
-            flow_meta_allocator,
-            #[cfg(feature = "enterprise")]
-            trigger_ddl_manager,
-        )
-        .await?;
+        let ddl_manager = {
+            let trigger_ddl_manager: Option<common_meta::ddl_manager::TriggerDdlManagerRef> =
+                plugins.get();
+            ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
+        };
+        let ddl_task_executor: ProcedureExecutorRef = Arc::new(ddl_manager);

        let fe_instance = FrontendBuilder::new(
            fe_opts.clone(),
@@ -679,41 +683,6 @@ impl StartCommand {
        })
    }

-    #[allow(clippy::too_many_arguments)]
-    pub async fn create_ddl_task_executor(
-        procedure_manager: ProcedureManagerRef,
-        node_manager: NodeManagerRef,
-        cache_invalidator: CacheInvalidatorRef,
-        table_metadata_manager: TableMetadataManagerRef,
-        table_metadata_allocator: TableMetadataAllocatorRef,
-        flow_metadata_manager: FlowMetadataManagerRef,
-        flow_metadata_allocator: FlowMetadataAllocatorRef,
-        #[cfg(feature = "enterprise")] trigger_ddl_manager: Option<TriggerDdlManagerRef>,
-    ) -> Result<ProcedureExecutorRef> {
-        let procedure_executor: ProcedureExecutorRef = Arc::new(
-            DdlManager::try_new(
-                DdlContext {
-                    node_manager,
-                    cache_invalidator,
-                    memory_region_keeper: Arc::new(MemoryRegionKeeper::default()),
-                    leader_region_registry: Arc::new(LeaderRegionRegistry::default()),
-                    table_metadata_manager,
-                    table_metadata_allocator,
-                    flow_metadata_manager,
-                    flow_metadata_allocator,
-                    region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
-                },
-                procedure_manager,
-                true,
-                #[cfg(feature = "enterprise")]
-                trigger_ddl_manager,
-            )
-            .context(error::InitDdlManagerSnafu)?,
-        );
-
-        Ok(procedure_executor)
-    }
-
    pub async fn create_table_metadata_manager(
        kv_backend: KvBackendRef,
    ) -> Result<TableMetadataManagerRef> {
--- a/src/common/frontend/src/selector.rs
+++ b/src/common/frontend/src/selector.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::fmt::Debug;
 use std::time::Duration;

 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
@@ -30,7 +31,7 @@ use crate::error::{MetaSnafu, Result};
 pub type FrontendClientPtr = Box<dyn FrontendClient>;

 #[async_trait::async_trait]
-pub trait FrontendClient: Send {
+pub trait FrontendClient: Send + Debug {
    async fn list_process(&mut self, req: ListProcessRequest) -> Result<ListProcessResponse>;

    async fn kill_process(&mut self, req: KillProcessRequest) -> Result<KillProcessResponse>;
--- a/src/common/function/src/system/pg_catalog/version.rs
+++ b/src/common/function/src/system/pg_catalog/version.rs
@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::fmt;
 use std::sync::Arc;
-use std::{env, fmt};

 use common_query::error::Result;
 use common_query::prelude::{Signature, Volatility};
@@ -47,7 +47,7 @@ impl Function for PGVersionFunction {
    fn eval(&self, _func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
        let result = StringVector::from(vec![format!(
            "PostgreSQL 16.3 GreptimeDB {}",
-            env!("CARGO_PKG_VERSION")
+            common_version::version()
        )]);
        Ok(Arc::new(result))
    }
--- a/src/common/function/src/system/version.rs
+++ b/src/common/function/src/system/version.rs
@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::fmt;
 use std::sync::Arc;
-use std::{env, fmt};

 use common_query::error::Result;
 use common_query::prelude::{Signature, Volatility};
@@ -52,13 +52,13 @@ impl Function for VersionFunction {
                    "{}-greptimedb-{}",
                    std::env::var("GREPTIMEDB_MYSQL_SERVER_VERSION")
                        .unwrap_or_else(|_| "8.4.2".to_string()),
-                    env!("CARGO_PKG_VERSION")
+                    common_version::version()
                )
            }
            Channel::Postgres => {
-                format!("16.3-greptimedb-{}", env!("CARGO_PKG_VERSION"))
+                format!("16.3-greptimedb-{}", common_version::version())
            }
-            _ => env!("CARGO_PKG_VERSION").to_string(),
+            _ => common_version::version().to_string(),
        };
        let result = StringVector::from(vec![version]);
        Ok(Arc::new(result))
--- a/src/common/grpc-expr/src/alter.rs
+++ b/src/common/grpc-expr/src/alter.rs
@@ -34,7 +34,7 @@ use table::requests::{
 };

 use crate::error::{
-    InvalidColumnDefSnafu, InvalidSetFulltextOptionRequestSnafu,
+    InvalidColumnDefSnafu, InvalidIndexOptionSnafu, InvalidSetFulltextOptionRequestSnafu,
    InvalidSetSkippingIndexOptionRequestSnafu, InvalidSetTableOptionRequestSnafu,
    InvalidUnsetTableOptionRequestSnafu, MissingAlterIndexOptionSnafu, MissingFieldSnafu,
    MissingTimestampColumnSnafu, Result, UnknownLocationTypeSnafu,
@@ -126,18 +126,21 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
                api::v1::set_index::Options::Fulltext(f) => AlterKind::SetIndex {
                    options: SetIndexOptions::Fulltext {
                        column_name: f.column_name.clone(),
-                        options: FulltextOptions {
-                            enable: f.enable,
-                            analyzer: as_fulltext_option_analyzer(
+                        options: FulltextOptions::new(
+                            f.enable,
+                            as_fulltext_option_analyzer(
                                Analyzer::try_from(f.analyzer)
                                    .context(InvalidSetFulltextOptionRequestSnafu)?,
                            ),
-                            case_sensitive: f.case_sensitive,
-                            backend: as_fulltext_option_backend(
+                            f.case_sensitive,
+                            as_fulltext_option_backend(
                                PbFulltextBackend::try_from(f.backend)
                                    .context(InvalidSetFulltextOptionRequestSnafu)?,
                            ),
-                        },
+                            f.granularity as u32,
+                            f.false_positive_rate,
+                        )
+                        .context(InvalidIndexOptionSnafu)?,
                    },
                },
                api::v1::set_index::Options::Inverted(i) => AlterKind::SetIndex {
@@ -148,13 +151,15 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
                api::v1::set_index::Options::Skipping(s) => AlterKind::SetIndex {
                    options: SetIndexOptions::Skipping {
                        column_name: s.column_name,
-                        options: SkippingIndexOptions {
-                            granularity: s.granularity as u32,
-                            index_type: as_skipping_index_type(
+                        options: SkippingIndexOptions::new(
+                            s.granularity as u32,
+                            s.false_positive_rate,
+                            as_skipping_index_type(
                                PbSkippingIndexType::try_from(s.skipping_index_type)
                                    .context(InvalidSetSkippingIndexOptionRequestSnafu)?,
                            ),
-                        },
+                        )
+                        .context(InvalidIndexOptionSnafu)?,
                    },
                },
            },
--- a/src/common/grpc-expr/src/error.rs
+++ b/src/common/grpc-expr/src/error.rs
@@ -153,6 +153,14 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
+
+    #[snafu(display("Invalid index option"))]
+    InvalidIndexOption {
+        #[snafu(implicit)]
+        location: Location,
+        #[snafu(source)]
+        error: datatypes::error::Error,
+    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -180,7 +188,8 @@ impl ErrorExt for Error {
            | Error::InvalidUnsetTableOptionRequest { .. }
            | Error::InvalidSetFulltextOptionRequest { .. }
            | Error::InvalidSetSkippingIndexOptionRequest { .. }
-            | Error::MissingAlterIndexOption { .. } => StatusCode::InvalidArguments,
+            | Error::MissingAlterIndexOption { .. }
+            | Error::InvalidIndexOption { .. } => StatusCode::InvalidArguments,
        }
    }

--- a/src/common/meta/src/cache/flow/table_flownode.rs
+++ b/src/common/meta/src/cache/flow/table_flownode.rs
@@ -15,6 +15,7 @@
 use std::collections::HashMap;
 use std::sync::Arc;

+use common_telemetry::info;
 use futures::future::BoxFuture;
 use moka::future::Cache;
 use moka::ops::compute::Op;
@@ -89,6 +90,12 @@ fn init_factory(table_flow_manager: TableFlowManagerRef) -> Initializer<TableId,
                // we have a corresponding cache invalidation mechanism to invalidate `(Key, EmptyHashSet)`.
                .map(Arc::new)
                .map(Some)
+                .inspect(|set| {
+                    info!(
+                        "Initialized table_flownode cache for table_id: {}, set: {:?}",
+                        table_id, set
+                    );
+                })
        })
    })
 }
@@ -167,6 +174,13 @@ fn invalidator<'a>(
        match ident {
            CacheIdent::CreateFlow(create_flow) => handle_create_flow(cache, create_flow).await,
            CacheIdent::DropFlow(drop_flow) => handle_drop_flow(cache, drop_flow).await,
+            CacheIdent::FlowNodeAddressChange(node_id) => {
+                info!(
+                    "Invalidate flow node cache for node_id in table_flownode: {}",
+                    node_id
+                );
+                cache.invalidate_all();
+            }
            _ => {}
        }
        Ok(())
@@ -174,7 +188,10 @@ fn invalidator<'a>(
 }

 fn filter(ident: &CacheIdent) -> bool {
-    matches!(ident, CacheIdent::CreateFlow(_) | CacheIdent::DropFlow(_))
+    matches!(
+        ident,
+        CacheIdent::CreateFlow(_) | CacheIdent::DropFlow(_) | CacheIdent::FlowNodeAddressChange(_)
+    )
 }

 #[cfg(test)]
--- a/src/common/meta/src/cache_invalidator.rs
+++ b/src/common/meta/src/cache_invalidator.rs
@@ -22,6 +22,7 @@ use crate::key::flow::flow_name::FlowNameKey;
 use crate::key::flow::flow_route::FlowRouteKey;
 use crate::key::flow::flownode_flow::FlownodeFlowKey;
 use crate::key::flow::table_flow::TableFlowKey;
+use crate::key::node_address::NodeAddressKey;
 use crate::key::schema_name::SchemaNameKey;
 use crate::key::table_info::TableInfoKey;
 use crate::key::table_name::TableNameKey;
@@ -53,6 +54,10 @@ pub struct Context {
 #[async_trait::async_trait]
 pub trait CacheInvalidator: Send + Sync {
    async fn invalidate(&self, ctx: &Context, caches: &[CacheIdent]) -> Result<()>;
+
+    fn name(&self) -> &'static str {
+        std::any::type_name::<Self>()
+    }
 }

 pub type CacheInvalidatorRef = Arc<dyn CacheInvalidator>;
@@ -137,6 +142,13 @@ where
                    let key = FlowInfoKey::new(*flow_id);
                    self.invalidate_key(&key.to_bytes()).await;
                }
+                CacheIdent::FlowNodeAddressChange(node_id) => {
+                    // other caches doesn't need to be invalidated
+                    // since this is only for flownode address change not id change
+                    common_telemetry::info!("Invalidate flow node cache for node_id: {}", node_id);
+                    let key = NodeAddressKey::with_flownode(*node_id);
+                    self.invalidate_key(&key.to_bytes()).await;
+                }
            }
        }
        Ok(())
--- a/src/common/meta/src/ddl_manager.rs
+++ b/src/common/meta/src/ddl_manager.rs
@@ -125,13 +125,12 @@ impl DdlManager {
        ddl_context: DdlContext,
        procedure_manager: ProcedureManagerRef,
        register_loaders: bool,
-        #[cfg(feature = "enterprise")] trigger_ddl_manager: Option<TriggerDdlManagerRef>,
    ) -> Result<Self> {
        let manager = Self {
            ddl_context,
            procedure_manager,
            #[cfg(feature = "enterprise")]
-            trigger_ddl_manager,
+            trigger_ddl_manager: None,
        };
        if register_loaders {
            manager.register_loaders()?;
@@ -139,6 +138,15 @@ impl DdlManager {
        Ok(manager)
    }

+    #[cfg(feature = "enterprise")]
+    pub fn with_trigger_ddl_manager(
+        mut self,
+        trigger_ddl_manager: Option<TriggerDdlManagerRef>,
+    ) -> Self {
+        self.trigger_ddl_manager = trigger_ddl_manager;
+        self
+    }
+
    /// Returns the [TableMetadataManagerRef].
    pub fn table_metadata_manager(&self) -> &TableMetadataManagerRef {
        &self.ddl_context.table_metadata_manager
@@ -964,8 +972,6 @@ mod tests {
            },
            procedure_manager.clone(),
            true,
-            #[cfg(feature = "enterprise")]
-            None,
        );

        let expected_loaders = vec![
--- a/src/common/meta/src/instruction.rs
+++ b/src/common/meta/src/instruction.rs
@@ -174,6 +174,8 @@ pub struct UpgradeRegion {
 /// The identifier of cache.
 pub enum CacheIdent {
    FlowId(FlowId),
+    /// Indicate change of address of flownode.
+    FlowNodeAddressChange(u64),
    FlowName(FlowName),
    TableId(TableId),
    TableName(TableName),
--- a/src/common/meta/src/key/tombstone.rs
+++ b/src/common/meta/src/key/tombstone.rs
@@ -14,13 +14,14 @@

 use std::collections::HashMap;

+use common_telemetry::debug;
 use snafu::ensure;

 use crate::error::{self, Result};
 use crate::key::txn_helper::TxnOpGetResponseSet;
 use crate::kv_backend::txn::{Compare, CompareOp, Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
-use crate::rpc::store::BatchGetRequest;
+use crate::rpc::store::{BatchDeleteRequest, BatchGetRequest};

 /// [TombstoneManager] provides the ability to:
 /// - logically delete values
@@ -28,6 +29,9 @@ use crate::rpc::store::BatchGetRequest;
 pub struct TombstoneManager {
    kv_backend: KvBackendRef,
    tombstone_prefix: String,
+    // Only used for testing.
+    #[cfg(test)]
+    max_txn_ops: Option<usize>,
 }

 const TOMBSTONE_PREFIX: &str = "__tombstone/";
@@ -35,10 +39,7 @@ const TOMBSTONE_PREFIX: &str = "__tombstone/";
 impl TombstoneManager {
    /// Returns [TombstoneManager].
    pub fn new(kv_backend: KvBackendRef) -> Self {
-        Self {
-            kv_backend,
-            tombstone_prefix: TOMBSTONE_PREFIX.to_string(),
-        }
+        Self::new_with_prefix(kv_backend, TOMBSTONE_PREFIX)
    }

    /// Returns [TombstoneManager] with a custom tombstone prefix.
@@ -46,6 +47,8 @@ impl TombstoneManager {
        Self {
            kv_backend,
            tombstone_prefix: prefix.to_string(),
+            #[cfg(test)]
+            max_txn_ops: None,
        }
    }

@@ -53,6 +56,11 @@ impl TombstoneManager {
        [self.tombstone_prefix.as_bytes(), key].concat()
    }

+    #[cfg(test)]
+    pub fn set_max_txn_ops(&mut self, max_txn_ops: usize) {
+        self.max_txn_ops = Some(max_txn_ops);
+    }
+
    /// Moves value to `dest_key`.
    ///
    /// Puts `value` to `dest_key` if the value of `src_key` equals `value`.
@@ -83,7 +91,11 @@ impl TombstoneManager {
        ensure!(
            keys.len() == dest_keys.len(),
            error::UnexpectedSnafu {
-                err_msg: "The length of keys does not match the length of dest_keys."
+                err_msg: format!(
+                    "The length of keys({}) does not match the length of dest_keys({}).",
+                    keys.len(),
+                    dest_keys.len()
+                ),
            }
        );
        // The key -> dest key mapping.
@@ -136,19 +148,45 @@ impl TombstoneManager {
        .fail()
    }

+    fn max_txn_ops(&self) -> usize {
+        #[cfg(test)]
+        if let Some(max_txn_ops) = self.max_txn_ops {
+            return max_txn_ops;
+        }
+        self.kv_backend.max_txn_ops()
+    }
+
    /// Moves values to `dest_key`.
    ///
    /// Returns the number of keys that were moved.
    async fn move_values(&self, keys: Vec<Vec<u8>>, dest_keys: Vec<Vec<u8>>) -> Result<usize> {
-        let chunk_size = self.kv_backend.max_txn_ops() / 2;
-        if keys.len() > chunk_size {
-            let keys_chunks = keys.chunks(chunk_size).collect::<Vec<_>>();
-            let dest_keys_chunks = keys.chunks(chunk_size).collect::<Vec<_>>();
-            for (keys, dest_keys) in keys_chunks.into_iter().zip(dest_keys_chunks) {
-                self.move_values_inner(keys, dest_keys).await?;
+        ensure!(
+            keys.len() == dest_keys.len(),
+            error::UnexpectedSnafu {
+                err_msg: format!(
+                    "The length of keys({}) does not match the length of dest_keys({}).",
+                    keys.len(),
+                    dest_keys.len()
+                ),
            }
-
-            Ok(keys.len())
+        );
+        if keys.is_empty() {
+            return Ok(0);
+        }
+        let chunk_size = self.max_txn_ops() / 2;
+        if keys.len() > chunk_size {
+            debug!(
+                "Moving values with multiple chunks, keys len: {}, chunk_size: {}",
+                keys.len(),
+                chunk_size
+            );
+            let mut moved_keys = 0;
+            let keys_chunks = keys.chunks(chunk_size).collect::<Vec<_>>();
+            let dest_keys_chunks = dest_keys.chunks(chunk_size).collect::<Vec<_>>();
+            for (keys, dest_keys) in keys_chunks.into_iter().zip(dest_keys_chunks) {
+                moved_keys += self.move_values_inner(keys, dest_keys).await?;
+            }
+            Ok(moved_keys)
        } else {
            self.move_values_inner(&keys, &dest_keys).await
        }
@@ -196,15 +234,18 @@ impl TombstoneManager {
    ///
    /// Returns the number of keys that were deleted.
    pub async fn delete(&self, keys: Vec<Vec<u8>>) -> Result<usize> {
-        let operations = keys
+        let keys = keys
            .iter()
-            .map(|key| TxnOp::Delete(self.to_tombstone(key)))
+            .map(|key| self.to_tombstone(key))
            .collect::<Vec<_>>();

-        let txn = Txn::new().and_then(operations);
-        // Always success.
-        let _ = self.kv_backend.txn(txn).await?;
-        Ok(keys.len())
+        let num_keys = keys.len();
+        let _ = self
+            .kv_backend
+            .batch_delete(BatchDeleteRequest::new().with_keys(keys))
+            .await?;
+
+        Ok(num_keys)
    }
 }

@@ -392,16 +433,73 @@ mod tests {
            .into_iter()
            .map(|kv| (kv.key, kv.dest_key))
            .unzip();
-        tombstone_manager
+        let moved_keys = tombstone_manager
            .move_values(keys.clone(), dest_keys.clone())
            .await
            .unwrap();
+        assert_eq!(kvs.len(), moved_keys);
        check_moved_values(kv_backend.clone(), &move_values).await;
        // Moves again
-        tombstone_manager
+        let moved_keys = tombstone_manager
            .move_values(keys.clone(), dest_keys.clone())
            .await
            .unwrap();
+        assert_eq!(0, moved_keys);
+        check_moved_values(kv_backend.clone(), &move_values).await;
+    }
+
+    #[tokio::test]
+    async fn test_move_values_with_max_txn_ops() {
+        common_telemetry::init_default_ut_logging();
+        let kv_backend = Arc::new(MemoryKvBackend::default());
+        let mut tombstone_manager = TombstoneManager::new(kv_backend.clone());
+        tombstone_manager.set_max_txn_ops(4);
+        let kvs = HashMap::from([
+            (b"bar".to_vec(), b"baz".to_vec()),
+            (b"foo".to_vec(), b"hi".to_vec()),
+            (b"baz".to_vec(), b"hello".to_vec()),
+            (b"qux".to_vec(), b"world".to_vec()),
+            (b"quux".to_vec(), b"world".to_vec()),
+            (b"quuux".to_vec(), b"world".to_vec()),
+            (b"quuuux".to_vec(), b"world".to_vec()),
+            (b"quuuuux".to_vec(), b"world".to_vec()),
+            (b"quuuuuux".to_vec(), b"world".to_vec()),
+        ]);
+        for (key, value) in &kvs {
+            kv_backend
+                .put(
+                    PutRequest::new()
+                        .with_key(key.clone())
+                        .with_value(value.clone()),
+                )
+                .await
+                .unwrap();
+        }
+        let move_values = kvs
+            .iter()
+            .map(|(key, value)| MoveValue {
+                key: key.clone(),
+                dest_key: tombstone_manager.to_tombstone(key),
+                value: value.clone(),
+            })
+            .collect::<Vec<_>>();
+        let (keys, dest_keys): (Vec<_>, Vec<_>) = move_values
+            .clone()
+            .into_iter()
+            .map(|kv| (kv.key, kv.dest_key))
+            .unzip();
+        let moved_keys = tombstone_manager
+            .move_values(keys.clone(), dest_keys.clone())
+            .await
+            .unwrap();
+        assert_eq!(kvs.len(), moved_keys);
+        check_moved_values(kv_backend.clone(), &move_values).await;
+        // Moves again
+        let moved_keys = tombstone_manager
+            .move_values(keys.clone(), dest_keys.clone())
+            .await
+            .unwrap();
+        assert_eq!(0, moved_keys);
        check_moved_values(kv_backend.clone(), &move_values).await;
    }

@@ -439,17 +537,19 @@ mod tests {
            .unzip();
        keys.push(b"non-exists".to_vec());
        dest_keys.push(b"hi/non-exists".to_vec());
-        tombstone_manager
+        let moved_keys = tombstone_manager
            .move_values(keys.clone(), dest_keys.clone())
            .await
            .unwrap();
        check_moved_values(kv_backend.clone(), &move_values).await;
+        assert_eq!(3, moved_keys);
        // Moves again
-        tombstone_manager
+        let moved_keys = tombstone_manager
            .move_values(keys.clone(), dest_keys.clone())
            .await
            .unwrap();
        check_moved_values(kv_backend.clone(), &move_values).await;
+        assert_eq!(0, moved_keys);
    }

    #[tokio::test]
@@ -490,10 +590,11 @@ mod tests {
            .into_iter()
            .map(|kv| (kv.key, kv.dest_key))
            .unzip();
-        tombstone_manager
+        let moved_keys = tombstone_manager
            .move_values(keys, dest_keys)
            .await
            .unwrap();
+        assert_eq!(kvs.len(), moved_keys);
    }

    #[tokio::test]
@@ -571,4 +672,24 @@ mod tests {
            .unwrap();
        check_moved_values(kv_backend.clone(), &move_values).await;
    }
+
+    #[tokio::test]
+    async fn test_move_values_with_different_lengths() {
+        let kv_backend = Arc::new(MemoryKvBackend::default());
+        let tombstone_manager = TombstoneManager::new(kv_backend.clone());
+
+        let keys = vec![b"bar".to_vec(), b"foo".to_vec()];
+        let dest_keys = vec![b"bar".to_vec(), b"foo".to_vec(), b"baz".to_vec()];
+
+        let err = tombstone_manager
+            .move_values(keys, dest_keys)
+            .await
+            .unwrap_err();
+        assert!(err
+            .to_string()
+            .contains("The length of keys(2) does not match the length of dest_keys(3)."),);
+
+        let moved_keys = tombstone_manager.move_values(vec![], vec![]).await.unwrap();
+        assert_eq!(0, moved_keys);
+    }
 }
--- a/src/common/recordbatch/src/adapter.rs
+++ b/src/common/recordbatch/src/adapter.rs
@@ -222,6 +222,7 @@ pub struct RecordBatchStreamAdapter {
 enum Metrics {
    Unavailable,
    Unresolved(Arc<dyn ExecutionPlan>),
+    PartialResolved(Arc<dyn ExecutionPlan>, RecordBatchMetrics),
    Resolved(RecordBatchMetrics),
 }

@@ -275,7 +276,9 @@ impl RecordBatchStream for RecordBatchStreamAdapter {

    fn metrics(&self) -> Option<RecordBatchMetrics> {
        match &self.metrics_2 {
-            Metrics::Resolved(metrics) => Some(metrics.clone()),
+            Metrics::Resolved(metrics) | Metrics::PartialResolved(_, metrics) => {
+                Some(metrics.clone())
+            }
            Metrics::Unavailable | Metrics::Unresolved(_) => None,
        }
    }
@@ -299,13 +302,25 @@ impl Stream for RecordBatchStreamAdapter {
            Poll::Pending => Poll::Pending,
            Poll::Ready(Some(df_record_batch)) => {
                let df_record_batch = df_record_batch?;
+                if let Metrics::Unresolved(df_plan) | Metrics::PartialResolved(df_plan, _) =
+                    &self.metrics_2
+                {
+                    let mut metric_collector = MetricCollector::new(self.explain_verbose);
+                    accept(df_plan.as_ref(), &mut metric_collector).unwrap();
+                    self.metrics_2 = Metrics::PartialResolved(
+                        df_plan.clone(),
+                        metric_collector.record_batch_metrics,
+                    );
+                }
                Poll::Ready(Some(RecordBatch::try_from_df_record_batch(
                    self.schema(),
                    df_record_batch,
                )))
            }
            Poll::Ready(None) => {
-                if let Metrics::Unresolved(df_plan) = &self.metrics_2 {
+                if let Metrics::Unresolved(df_plan) | Metrics::PartialResolved(df_plan, _) =
+                    &self.metrics_2
+                {
                    let mut metric_collector = MetricCollector::new(self.explain_verbose);
                    accept(df_plan.as_ref(), &mut metric_collector).unwrap();
                    self.metrics_2 = Metrics::Resolved(metric_collector.record_batch_metrics);
--- a/src/common/recordbatch/src/error.rs
+++ b/src/common/recordbatch/src/error.rs
@@ -178,8 +178,6 @@ pub enum Error {
    StreamTimeout {
        #[snafu(implicit)]
        location: Location,
-        #[snafu(source)]
-        error: tokio::time::error::Elapsed,
    },

    #[snafu(display("RecordBatch slice index overflow: {visit_index} > {size}"))]
--- a/src/common/telemetry/Cargo.toml
+++ b/src/common/telemetry/Cargo.toml
@@ -14,6 +14,7 @@ workspace = true
 [dependencies]
 backtrace = "0.3"
 common-error.workspace = true
+common-version.workspace = true
 console-subscriber = { version = "0.1", optional = true }
 greptime-proto.workspace = true
 humantime-serde.workspace = true
--- a/src/common/telemetry/src/logging.rs
+++ b/src/common/telemetry/src/logging.rs
@@ -384,7 +384,7 @@ pub fn init_global_logging(
                        resource::SERVICE_INSTANCE_ID,
                        node_id.unwrap_or("none".to_string()),
                    ),
-                    KeyValue::new(resource::SERVICE_VERSION, env!("CARGO_PKG_VERSION")),
+                    KeyValue::new(resource::SERVICE_VERSION, common_version::version()),
                    KeyValue::new(resource::PROCESS_PID, std::process::id().to_string()),
                ]));

--- a/src/common/version/Cargo.toml
+++ b/src/common/version/Cargo.toml
@@ -17,4 +17,5 @@ shadow-rs.workspace = true

 [build-dependencies]
 build-data = "0.2"
+cargo-manifest = "0.19"
 shadow-rs.workspace = true
--- a/src/common/version/build.rs
+++ b/src/common/version/build.rs
@@ -14,8 +14,10 @@

 use std::collections::BTreeSet;
 use std::env;
+use std::path::PathBuf;

 use build_data::{format_timestamp, get_source_time};
+use cargo_manifest::Manifest;
 use shadow_rs::{BuildPattern, ShadowBuilder, CARGO_METADATA, CARGO_TREE};

 fn main() -> shadow_rs::SdResult<()> {
@@ -33,6 +35,24 @@ fn main() -> shadow_rs::SdResult<()> {
    // solve the problem where the "CARGO_MANIFEST_DIR" is not what we want when this repo is
    // made as a submodule in another repo.
    let src_path = env::var("CARGO_WORKSPACE_DIR").or_else(|_| env::var("CARGO_MANIFEST_DIR"))?;
+
+    let manifest = Manifest::from_path(PathBuf::from(&src_path).join("Cargo.toml"))
+        .expect("Failed to parse Cargo.toml");
+    if let Some(product_version) = manifest.workspace.as_ref().and_then(|w| {
+        w.metadata.as_ref().and_then(|m| {
+            m.get("greptime")
+                .and_then(|g| g.get("product_version").and_then(|v| v.as_str()))
+        })
+    }) {
+        println!(
+            "cargo:rustc-env=GREPTIME_PRODUCT_VERSION={}",
+            product_version
+        );
+    } else {
+        let version = env::var("CARGO_PKG_VERSION").unwrap();
+        println!("cargo:rustc-env=GREPTIME_PRODUCT_VERSION={}", version,);
+    }
+
    let out_path = env::var("OUT_DIR")?;

    let _ = ShadowBuilder::builder()
--- a/src/common/version/src/lib.rs
+++ b/src/common/version/src/lib.rs
@@ -105,13 +105,17 @@ pub const fn build_info() -> BuildInfo {
        build_time: env!("BUILD_TIMESTAMP"),
        rustc: build::RUST_VERSION,
        target: build::BUILD_TARGET,
-        version: build::PKG_VERSION,
+        version: env!("GREPTIME_PRODUCT_VERSION"),
    }
 }

 const BUILD_INFO: BuildInfo = build_info();

 pub const fn version() -> &'static str {
+    BUILD_INFO.version
+}
+
+pub const fn verbose_version() -> &'static str {
    const_format::formatcp!(
        "\nbranch: {}\ncommit: {}\nclean: {}\nversion: {}",
        BUILD_INFO.branch,
--- a/src/datanode/src/alive_keeper.rs
+++ b/src/datanode/src/alive_keeper.rs
@@ -475,7 +475,7 @@ mod test {
    async fn region_alive_keeper() {
        common_telemetry::init_default_ut_logging();
        let mut region_server = mock_region_server();
-        let mut engine_env = TestEnv::with_prefix("region-alive-keeper");
+        let mut engine_env = TestEnv::with_prefix("region-alive-keeper").await;
        let engine = engine_env.create_engine(MitoConfig::default()).await;
        let engine = Arc::new(engine);
        region_server.register_engine(engine.clone());
--- a/src/datanode/src/heartbeat/handler.rs
+++ b/src/datanode/src/heartbeat/handler.rs
@@ -278,7 +278,7 @@ mod tests {
        let mut region_server = mock_region_server();
        let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());

-        let mut engine_env = TestEnv::with_prefix("close-region");
+        let mut engine_env = TestEnv::with_prefix("close-region").await;
        let engine = engine_env.create_engine(MitoConfig::default()).await;
        region_server.register_engine(Arc::new(engine));
        let region_id = RegionId::new(1024, 1);
@@ -326,7 +326,7 @@ mod tests {
        let mut region_server = mock_region_server();
        let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());

-        let mut engine_env = TestEnv::with_prefix("open-region");
+        let mut engine_env = TestEnv::with_prefix("open-region").await;
        let engine = engine_env.create_engine(MitoConfig::default()).await;
        region_server.register_engine(Arc::new(engine));
        let region_id = RegionId::new(1024, 1);
@@ -374,7 +374,7 @@ mod tests {
        let mut region_server = mock_region_server();
        let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());

-        let mut engine_env = TestEnv::with_prefix("open-not-exists-region");
+        let mut engine_env = TestEnv::with_prefix("open-not-exists-region").await;
        let engine = engine_env.create_engine(MitoConfig::default()).await;
        region_server.register_engine(Arc::new(engine));
        let region_id = RegionId::new(1024, 1);
@@ -406,7 +406,7 @@ mod tests {
        let mut region_server = mock_region_server();
        let heartbeat_handler = RegionHeartbeatResponseHandler::new(region_server.clone());

-        let mut engine_env = TestEnv::with_prefix("downgrade-region");
+        let mut engine_env = TestEnv::with_prefix("downgrade-region").await;
        let engine = engine_env.create_engine(MitoConfig::default()).await;
        region_server.register_engine(Arc::new(engine));
        let region_id = RegionId::new(1024, 1);
--- a/src/datanode/src/metrics.rs
+++ b/src/datanode/src/metrics.rs
@@ -27,14 +27,14 @@ lazy_static! {
    pub static ref HANDLE_REGION_REQUEST_ELAPSED: HistogramVec = register_histogram_vec!(
        "greptime_datanode_handle_region_request_elapsed",
        "datanode handle region request elapsed",
-        &[REGION_ID, REGION_REQUEST_TYPE]
+        &[REGION_REQUEST_TYPE]
    )
    .unwrap();
    /// The number of rows in region request received by region server, labeled with request type.
    pub static ref REGION_CHANGED_ROW_COUNT: IntCounterVec = register_int_counter_vec!(
        "greptime_datanode_region_changed_row_count",
        "datanode region changed row count",
-        &[REGION_ID, REGION_REQUEST_TYPE]
+        &[REGION_REQUEST_TYPE]
    )
    .unwrap();
    /// The elapsed time since the last received heartbeat.
--- a/src/datanode/src/region_server.rs
+++ b/src/datanode/src/region_server.rs
@@ -51,7 +51,7 @@ use servers::error::{self as servers_error, ExecuteGrpcRequestSnafu, Result as S
 use servers::grpc::flight::{FlightCraft, FlightRecordBatchStream, TonicStream};
 use servers::grpc::region_server::RegionServerHandler;
 use servers::grpc::FlightCompression;
-use session::context::{QueryContextBuilder, QueryContextRef};
+use session::context::{QueryContext, QueryContextBuilder, QueryContextRef};
 use snafu::{ensure, OptionExt, ResultExt};
 use store_api::metric_engine_consts::{
    FILE_ENGINE_NAME, LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME,
@@ -194,6 +194,7 @@ impl RegionServer {
    pub async fn handle_remote_read(
        &self,
        request: api::v1::region::QueryRequest,
+        query_ctx: QueryContextRef,
    ) -> Result<SendableRecordBatchStream> {
        let _permit = if let Some(p) = &self.inner.parallelism {
            Some(p.acquire().await?)
@@ -201,12 +202,6 @@ impl RegionServer {
            None
        };

-        let query_ctx: QueryContextRef = request
-            .header
-            .as_ref()
-            .map(|h| Arc::new(h.into()))
-            .unwrap_or_else(|| Arc::new(QueryContextBuilder::default().build()));
-
        let region_id = RegionId::from_u64(request.region_id);
        let provider = self.table_provider(region_id, Some(&query_ctx)).await?;
        let catalog_list = Arc::new(DummyCatalogList::with_table_provider(provider));
@@ -214,7 +209,7 @@ impl RegionServer {
        let decoder = self
            .inner
            .query_engine
-            .engine_context(query_ctx)
+            .engine_context(query_ctx.clone())
            .new_plan_decoder()
            .context(NewPlanDecoderSnafu)?;

@@ -224,11 +219,14 @@ impl RegionServer {
            .context(DecodeLogicalPlanSnafu)?;

        self.inner
-            .handle_read(QueryRequest {
-                header: request.header,
-                region_id,
-                plan,
-            })
+            .handle_read(
+                QueryRequest {
+                    header: request.header,
+                    region_id,
+                    plan,
+                },
+                query_ctx,
+            )
            .await
    }

@@ -243,6 +241,7 @@ impl RegionServer {
        let ctx: Option<session::context::QueryContext> = request.header.as_ref().map(|h| h.into());

        let provider = self.table_provider(request.region_id, ctx.as_ref()).await?;
+        let query_ctx = Arc::new(ctx.unwrap_or_else(|| QueryContextBuilder::default().build()));

        struct RegionDataSourceInjector {
            source: Arc<dyn TableSource>,
@@ -271,7 +270,7 @@ impl RegionServer {
            .data;

        self.inner
-            .handle_read(QueryRequest { plan, ..request })
+            .handle_read(QueryRequest { plan, ..request }, query_ctx)
            .await
    }

@@ -536,9 +535,14 @@ impl FlightCraft for RegionServer {
            .as_ref()
            .map(|h| TracingContext::from_w3c(&h.tracing_context))
            .unwrap_or_default();
+        let query_ctx = request
+            .header
+            .as_ref()
+            .map(|h| Arc::new(QueryContext::from(h)))
+            .unwrap_or(QueryContext::arc());

        let result = self
-            .handle_remote_read(request)
+            .handle_remote_read(request, query_ctx.clone())
            .trace(tracing_context.attach(info_span!("RegionServer::handle_read")))
            .await?;

@@ -546,6 +550,7 @@ impl FlightCraft for RegionServer {
            result,
            tracing_context,
            self.flight_compression,
+            query_ctx,
        ));
        Ok(Response::new(stream))
    }
@@ -915,9 +920,8 @@ impl RegionServerInner {
        request: RegionRequest,
    ) -> Result<RegionResponse> {
        let request_type = request.request_type();
-        let region_id_str = region_id.to_string();
        let _timer = crate::metrics::HANDLE_REGION_REQUEST_ELAPSED
-            .with_label_values(&[&region_id_str, request_type])
+            .with_label_values(&[request_type])
            .start_timer();

        let region_change = match &request {
@@ -957,7 +961,7 @@ impl RegionServerInner {
                // Update metrics
                if matches!(region_change, RegionChange::Ingest) {
                    crate::metrics::REGION_CHANGED_ROW_COUNT
-                        .with_label_values(&[&region_id_str, request_type])
+                        .with_label_values(&[request_type])
                        .inc_by(result.affected_rows as u64);
                }
                // Sets corresponding region status to ready.
@@ -1124,16 +1128,13 @@ impl RegionServerInner {
        Ok(())
    }

-    pub async fn handle_read(&self, request: QueryRequest) -> Result<SendableRecordBatchStream> {
+    pub async fn handle_read(
+        &self,
+        request: QueryRequest,
+        query_ctx: QueryContextRef,
+    ) -> Result<SendableRecordBatchStream> {
        // TODO(ruihang): add metrics and set trace id

-        // Build query context from gRPC header
-        let query_ctx: QueryContextRef = request
-            .header
-            .as_ref()
-            .map(|h| Arc::new(h.into()))
-            .unwrap_or_else(|| QueryContextBuilder::default().build().into());
-
        let result = self
            .query_engine
            .execute(request.plan, query_ctx)
--- a/src/datatypes/src/schema.rs
+++ b/src/datatypes/src/schema.rs
@@ -31,9 +31,10 @@ pub use crate::schema::column_schema::{
    ColumnSchema, FulltextAnalyzer, FulltextBackend, FulltextOptions, Metadata,
    SkippingIndexOptions, SkippingIndexType, COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE,
    COLUMN_FULLTEXT_OPT_KEY_ANALYZER, COLUMN_FULLTEXT_OPT_KEY_BACKEND,
-    COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY,
-    COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE, COMMENT_KEY, FULLTEXT_KEY, INVERTED_INDEX_KEY,
-    SKIPPING_INDEX_KEY, TIME_INDEX_KEY,
+    COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE, COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE,
+    COLUMN_FULLTEXT_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE,
+    COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY, COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE, COMMENT_KEY,
+    FULLTEXT_KEY, INVERTED_INDEX_KEY, SKIPPING_INDEX_KEY, TIME_INDEX_KEY,
 };
 pub use crate::schema::constraint::ColumnDefaultConstraint;
 pub use crate::schema::raw::RawSchema;
--- a/src/datatypes/src/schema/column_schema.rs
+++ b/src/datatypes/src/schema/column_schema.rs
@@ -47,13 +47,18 @@ pub const COLUMN_FULLTEXT_CHANGE_OPT_KEY_ENABLE: &str = "enable";
 pub const COLUMN_FULLTEXT_OPT_KEY_ANALYZER: &str = "analyzer";
 pub const COLUMN_FULLTEXT_OPT_KEY_CASE_SENSITIVE: &str = "case_sensitive";
 pub const COLUMN_FULLTEXT_OPT_KEY_BACKEND: &str = "backend";
+pub const COLUMN_FULLTEXT_OPT_KEY_GRANULARITY: &str = "granularity";
+pub const COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE: &str = "false_positive_rate";

 /// Keys used in SKIPPING index options
 pub const COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY: &str = "granularity";
+pub const COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE: &str = "false_positive_rate";
 pub const COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE: &str = "type";

 pub const DEFAULT_GRANULARITY: u32 = 10240;

+pub const DEFAULT_FALSE_POSITIVE_RATE: f64 = 0.01;
+
 /// Schema of a column, used as an immutable struct.
 #[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct ColumnSchema {
@@ -504,7 +509,7 @@ impl TryFrom<&ColumnSchema> for Field {
 }

 /// Fulltext options for a column.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default, Visit, VisitMut)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Visit, VisitMut)]
 #[serde(rename_all = "kebab-case")]
 pub struct FulltextOptions {
    /// Whether the fulltext index is enabled.
@@ -518,6 +523,92 @@ pub struct FulltextOptions {
    /// The fulltext backend to use.
    #[serde(default)]
    pub backend: FulltextBackend,
+    /// The granularity of the fulltext index (for bloom backend only)
+    #[serde(default = "fulltext_options_default_granularity")]
+    pub granularity: u32,
+    /// The false positive rate of the fulltext index (for bloom backend only)
+    #[serde(default = "index_options_default_false_positive_rate_in_10000")]
+    pub false_positive_rate_in_10000: u32,
+}
+
+fn fulltext_options_default_granularity() -> u32 {
+    DEFAULT_GRANULARITY
+}
+
+fn index_options_default_false_positive_rate_in_10000() -> u32 {
+    (DEFAULT_FALSE_POSITIVE_RATE * 10000.0) as u32
+}
+
+impl FulltextOptions {
+    /// Creates a new fulltext options.
+    pub fn new(
+        enable: bool,
+        analyzer: FulltextAnalyzer,
+        case_sensitive: bool,
+        backend: FulltextBackend,
+        granularity: u32,
+        false_positive_rate: f64,
+    ) -> Result<Self> {
+        ensure!(
+            0.0 < false_positive_rate && false_positive_rate <= 1.0,
+            error::InvalidFulltextOptionSnafu {
+                msg: format!(
+                    "Invalid false positive rate: {false_positive_rate}, expected: 0.0 < rate <= 1.0"
+                ),
+            }
+        );
+        ensure!(
+            granularity > 0,
+            error::InvalidFulltextOptionSnafu {
+                msg: format!("Invalid granularity: {granularity}, expected: positive integer"),
+            }
+        );
+        Ok(Self::new_unchecked(
+            enable,
+            analyzer,
+            case_sensitive,
+            backend,
+            granularity,
+            false_positive_rate,
+        ))
+    }
+
+    /// Creates a new fulltext options without checking `false_positive_rate` and `granularity`.
+    pub fn new_unchecked(
+        enable: bool,
+        analyzer: FulltextAnalyzer,
+        case_sensitive: bool,
+        backend: FulltextBackend,
+        granularity: u32,
+        false_positive_rate: f64,
+    ) -> Self {
+        Self {
+            enable,
+            analyzer,
+            case_sensitive,
+            backend,
+            granularity,
+            false_positive_rate_in_10000: (false_positive_rate * 10000.0) as u32,
+        }
+    }
+
+    /// Gets the false positive rate.
+    pub fn false_positive_rate(&self) -> f64 {
+        self.false_positive_rate_in_10000 as f64 / 10000.0
+    }
+}
+
+impl Default for FulltextOptions {
+    fn default() -> Self {
+        Self::new_unchecked(
+            false,
+            FulltextAnalyzer::default(),
+            false,
+            FulltextBackend::default(),
+            DEFAULT_GRANULARITY,
+            DEFAULT_FALSE_POSITIVE_RATE,
+        )
+    }
 }

 impl fmt::Display for FulltextOptions {
@@ -527,6 +618,10 @@ impl fmt::Display for FulltextOptions {
            write!(f, ", analyzer={}", self.analyzer)?;
            write!(f, ", case_sensitive={}", self.case_sensitive)?;
            write!(f, ", backend={}", self.backend)?;
+            if self.backend == FulltextBackend::Bloom {
+                write!(f, ", granularity={}", self.granularity)?;
+                write!(f, ", false_positive_rate={}", self.false_positive_rate())?;
+            }
        }
        Ok(())
    }
@@ -611,6 +706,45 @@ impl TryFrom<HashMap<String, String>> for FulltextOptions {
            }
        }

+        if fulltext_options.backend == FulltextBackend::Bloom {
+            // Parse granularity with default value 10240
+            let granularity = match options.get(COLUMN_FULLTEXT_OPT_KEY_GRANULARITY) {
+                Some(value) => value
+                    .parse::<u32>()
+                    .ok()
+                    .filter(|&v| v > 0)
+                    .ok_or_else(|| {
+                        error::InvalidFulltextOptionSnafu {
+                            msg: format!(
+                                "Invalid granularity: {value}, expected: positive integer"
+                            ),
+                        }
+                        .build()
+                    })?,
+                None => DEFAULT_GRANULARITY,
+            };
+            fulltext_options.granularity = granularity;
+
+            // Parse false positive rate with default value 0.01
+            let false_positive_rate = match options.get(COLUMN_FULLTEXT_OPT_KEY_FALSE_POSITIVE_RATE)
+            {
+                Some(value) => value
+                    .parse::<f64>()
+                    .ok()
+                    .filter(|&v| v > 0.0 && v <= 1.0)
+                    .ok_or_else(|| {
+                        error::InvalidFulltextOptionSnafu {
+                            msg: format!(
+                                "Invalid false positive rate: {value}, expected: 0.0 < rate <= 1.0"
+                            ),
+                        }
+                        .build()
+                    })?,
+                None => DEFAULT_FALSE_POSITIVE_RATE,
+            };
+            fulltext_options.false_positive_rate_in_10000 = (false_positive_rate * 10000.0) as u32;
+        }
+
        Ok(fulltext_options)
    }
 }
@@ -638,23 +772,73 @@ impl fmt::Display for FulltextAnalyzer {
 pub struct SkippingIndexOptions {
    /// The granularity of the skip index.
    pub granularity: u32,
+    /// The false positive rate of the skip index (in ten-thousandths, e.g., 100 = 1%).
+    #[serde(default = "index_options_default_false_positive_rate_in_10000")]
+    pub false_positive_rate_in_10000: u32,
    /// The type of the skip index.
    #[serde(default)]
    pub index_type: SkippingIndexType,
 }

+impl SkippingIndexOptions {
+    /// Creates a new skipping index options without checking `false_positive_rate` and `granularity`.
+    pub fn new_unchecked(
+        granularity: u32,
+        false_positive_rate: f64,
+        index_type: SkippingIndexType,
+    ) -> Self {
+        Self {
+            granularity,
+            false_positive_rate_in_10000: (false_positive_rate * 10000.0) as u32,
+            index_type,
+        }
+    }
+
+    /// Creates a new skipping index options.
+    pub fn new(
+        granularity: u32,
+        false_positive_rate: f64,
+        index_type: SkippingIndexType,
+    ) -> Result<Self> {
+        ensure!(
+            0.0 < false_positive_rate && false_positive_rate <= 1.0,
+            error::InvalidSkippingIndexOptionSnafu {
+                msg: format!("Invalid false positive rate: {false_positive_rate}, expected: 0.0 < rate <= 1.0"),
+            }
+        );
+        ensure!(
+            granularity > 0,
+            error::InvalidSkippingIndexOptionSnafu {
+                msg: format!("Invalid granularity: {granularity}, expected: positive integer"),
+            }
+        );
+        Ok(Self::new_unchecked(
+            granularity,
+            false_positive_rate,
+            index_type,
+        ))
+    }
+
+    /// Gets the false positive rate.
+    pub fn false_positive_rate(&self) -> f64 {
+        self.false_positive_rate_in_10000 as f64 / 10000.0
+    }
+}
+
 impl Default for SkippingIndexOptions {
    fn default() -> Self {
-        Self {
-            granularity: DEFAULT_GRANULARITY,
-            index_type: SkippingIndexType::default(),
-        }
+        Self::new_unchecked(
+            DEFAULT_GRANULARITY,
+            DEFAULT_FALSE_POSITIVE_RATE,
+            SkippingIndexType::default(),
+        )
    }
 }

 impl fmt::Display for SkippingIndexOptions {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "granularity={}", self.granularity)?;
+        write!(f, ", false_positive_rate={}", self.false_positive_rate())?;
        write!(f, ", index_type={}", self.index_type)?;
        Ok(())
    }
@@ -681,15 +865,37 @@ impl TryFrom<HashMap<String, String>> for SkippingIndexOptions {
    fn try_from(options: HashMap<String, String>) -> Result<Self> {
        // Parse granularity with default value 1
        let granularity = match options.get(COLUMN_SKIPPING_INDEX_OPT_KEY_GRANULARITY) {
-            Some(value) => value.parse::<u32>().map_err(|_| {
-                error::InvalidSkippingIndexOptionSnafu {
-                    msg: format!("Invalid granularity: {value}, expected: positive integer"),
-                }
-                .build()
-            })?,
+            Some(value) => value
+                .parse::<u32>()
+                .ok()
+                .filter(|&v| v > 0)
+                .ok_or_else(|| {
+                    error::InvalidSkippingIndexOptionSnafu {
+                        msg: format!("Invalid granularity: {value}, expected: positive integer"),
+                    }
+                    .build()
+                })?,
            None => DEFAULT_GRANULARITY,
        };

+        // Parse false positive rate with default value 100
+        let false_positive_rate =
+            match options.get(COLUMN_SKIPPING_INDEX_OPT_KEY_FALSE_POSITIVE_RATE) {
+                Some(value) => value
+                    .parse::<f64>()
+                    .ok()
+                    .filter(|&v| v > 0.0 && v <= 1.0)
+                    .ok_or_else(|| {
+                        error::InvalidSkippingIndexOptionSnafu {
+                            msg: format!(
+                                "Invalid false positive rate: {value}, expected: 0.0 < rate <= 1.0"
+                            ),
+                        }
+                        .build()
+                    })?,
+                None => DEFAULT_FALSE_POSITIVE_RATE,
+            };
+
        // Parse index type with default value BloomFilter
        let index_type = match options.get(COLUMN_SKIPPING_INDEX_OPT_KEY_TYPE) {
            Some(typ) => match typ.to_ascii_uppercase().as_str() {
@@ -704,10 +910,11 @@ impl TryFrom<HashMap<String, String>> for SkippingIndexOptions {
            None => SkippingIndexType::default(),
        };

-        Ok(SkippingIndexOptions {
+        Ok(SkippingIndexOptions::new_unchecked(
            granularity,
+            false_positive_rate,
            index_type,
-        })
+        ))
    }
 }

@@ -973,4 +1180,59 @@ mod tests {
        assert!(column_schema.default_constraint.is_none());
        assert!(column_schema.metadata.is_empty());
    }
+
+    #[test]
+    fn test_skipping_index_options_deserialization() {
+        let original_options = "{\"granularity\":1024,\"false-positive-rate-in-10000\":10,\"index-type\":\"BloomFilter\"}";
+        let options = serde_json::from_str::<SkippingIndexOptions>(original_options).unwrap();
+        assert_eq!(1024, options.granularity);
+        assert_eq!(SkippingIndexType::BloomFilter, options.index_type);
+        assert_eq!(0.001, options.false_positive_rate());
+
+        let options_str = serde_json::to_string(&options).unwrap();
+        assert_eq!(options_str, original_options);
+    }
+
+    #[test]
+    fn test_skipping_index_options_deserialization_v0_14_to_v0_15() {
+        let options = "{\"granularity\":10240,\"index-type\":\"BloomFilter\"}";
+        let options = serde_json::from_str::<SkippingIndexOptions>(options).unwrap();
+        assert_eq!(10240, options.granularity);
+        assert_eq!(SkippingIndexType::BloomFilter, options.index_type);
+        assert_eq!(DEFAULT_FALSE_POSITIVE_RATE, options.false_positive_rate());
+
+        let options_str = serde_json::to_string(&options).unwrap();
+        assert_eq!(options_str, "{\"granularity\":10240,\"false-positive-rate-in-10000\":100,\"index-type\":\"BloomFilter\"}");
+    }
+
+    #[test]
+    fn test_fulltext_options_deserialization() {
+        let original_options = "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\",\"granularity\":1024,\"false-positive-rate-in-10000\":10}";
+        let options = serde_json::from_str::<FulltextOptions>(original_options).unwrap();
+        assert!(!options.case_sensitive);
+        assert!(options.enable);
+        assert_eq!(FulltextBackend::Bloom, options.backend);
+        assert_eq!(FulltextAnalyzer::default(), options.analyzer);
+        assert_eq!(1024, options.granularity);
+        assert_eq!(0.001, options.false_positive_rate());
+
+        let options_str = serde_json::to_string(&options).unwrap();
+        assert_eq!(options_str, original_options);
+    }
+
+    #[test]
+    fn test_fulltext_options_deserialization_v0_14_to_v0_15() {
+        // 0.14 to 0.15
+        let options = "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\"}";
+        let options = serde_json::from_str::<FulltextOptions>(options).unwrap();
+        assert!(!options.case_sensitive);
+        assert!(options.enable);
+        assert_eq!(FulltextBackend::Bloom, options.backend);
+        assert_eq!(FulltextAnalyzer::default(), options.analyzer);
+        assert_eq!(DEFAULT_GRANULARITY, options.granularity);
+        assert_eq!(DEFAULT_FALSE_POSITIVE_RATE, options.false_positive_rate());
+
+        let options_str = serde_json::to_string(&options).unwrap();
+        assert_eq!(options_str, "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\",\"granularity\":10240,\"false-positive-rate-in-10000\":100}");
+    }
 }
--- a/src/datatypes/src/timestamp.rs
+++ b/src/datatypes/src/timestamp.rs
@@ -12,6 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use arrow_array::{
+    ArrayRef, PrimitiveArray, TimestampMicrosecondArray, TimestampMillisecondArray,
+    TimestampNanosecondArray, TimestampSecondArray,
+};
+use arrow_schema::DataType;
 use common_time::timestamp::TimeUnit;
 use common_time::Timestamp;
 use paste::paste;
@@ -138,6 +143,41 @@ define_timestamp_with_unit!(Millisecond);
 define_timestamp_with_unit!(Microsecond);
 define_timestamp_with_unit!(Nanosecond);

+pub fn timestamp_array_to_primitive(
+    ts_array: &ArrayRef,
+) -> Option<(
+    PrimitiveArray<arrow_array::types::Int64Type>,
+    arrow::datatypes::TimeUnit,
+)> {
+    let DataType::Timestamp(unit, _) = ts_array.data_type() else {
+        return None;
+    };
+
+    let ts_primitive = match unit {
+        arrow_schema::TimeUnit::Second => ts_array
+            .as_any()
+            .downcast_ref::<TimestampSecondArray>()
+            .unwrap()
+            .reinterpret_cast::<arrow_array::types::Int64Type>(),
+        arrow_schema::TimeUnit::Millisecond => ts_array
+            .as_any()
+            .downcast_ref::<TimestampMillisecondArray>()
+            .unwrap()
+            .reinterpret_cast::<arrow_array::types::Int64Type>(),
+        arrow_schema::TimeUnit::Microsecond => ts_array
+            .as_any()
+            .downcast_ref::<TimestampMicrosecondArray>()
+            .unwrap()
+            .reinterpret_cast::<arrow_array::types::Int64Type>(),
+        arrow_schema::TimeUnit::Nanosecond => ts_array
+            .as_any()
+            .downcast_ref::<TimestampNanosecondArray>()
+            .unwrap()
+            .reinterpret_cast::<arrow_array::types::Int64Type>(),
+    };
+    Some((ts_primitive, *unit))
+}
+
 #[cfg(test)]
 mod tests {
    use common_time::timezone::set_default_timezone;
--- a/src/flow/src/adapter.rs
+++ b/src/flow/src/adapter.rs
@@ -899,7 +899,7 @@ impl StreamingEngine {
        let rows_send = self.run_available(true).await?;
        let row = self.send_writeback_requests().await?;
        debug!(
-            "Done to flush flow_id={:?} with {} input rows flushed, {} rows sended and {} output rows flushed",
+            "Done to flush flow_id={:?} with {} input rows flushed, {} rows sent and {} output rows flushed",
            flow_id, flushed_input_rows, rows_send, row
        );
        Ok(row)
--- a/src/flow/src/batching_mode/engine.rs
+++ b/src/flow/src/batching_mode/engine.rs
@@ -14,7 +14,7 @@

 //! Batching mode engine

-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 use std::sync::Arc;

 use api::v1::flow::{DirtyWindowRequests, FlowResponse};
@@ -142,7 +142,7 @@ impl BatchingEngine {

            let handle: JoinHandle<Result<(), Error>> = tokio::spawn(async move {
                let src_table_names = &task.config.source_table_names;
-                let mut all_dirty_windows = vec![];
+                let mut all_dirty_windows = HashSet::new();
                for src_table_name in src_table_names {
                    if let Some((timestamps, unit)) = group_by_table_name.get(src_table_name) {
                        let Some(expr) = &task.config.time_window_expr else {
@@ -155,7 +155,7 @@ impl BatchingEngine {
                                .context(UnexpectedSnafu {
                                    reason: "Failed to eval start value",
                                })?;
-                            all_dirty_windows.push(align_start);
+                            all_dirty_windows.insert(align_start);
                        }
                    }
                }
--- a/src/flow/src/batching_mode/time_window.rs
+++ b/src/flow/src/batching_mode/time_window.rs
@@ -50,7 +50,8 @@ use snafu::{ensure, OptionExt, ResultExt};

 use crate::adapter::util::from_proto_to_data_type;
 use crate::error::{
-    ArrowSnafu, DatafusionSnafu, DatatypesSnafu, ExternalSnafu, PlanSnafu, UnexpectedSnafu,
+    ArrowSnafu, DatafusionSnafu, DatatypesSnafu, ExternalSnafu, PlanSnafu, TimeSnafu,
+    UnexpectedSnafu,
 };
 use crate::expr::error::DataTypeSnafu;
 use crate::Error;
@@ -74,6 +75,7 @@ pub struct TimeWindowExpr {
    logical_expr: Expr,
    df_schema: DFSchema,
    eval_time_window_size: Option<std::time::Duration>,
+    eval_time_original: Option<Timestamp>,
 }

 impl std::fmt::Display for TimeWindowExpr {
@@ -106,10 +108,11 @@ impl TimeWindowExpr {
            logical_expr: expr.clone(),
            df_schema: df_schema.clone(),
            eval_time_window_size: None,
+            eval_time_original: None,
        };
        let test_ts = DEFAULT_TEST_TIMESTAMP;
-        let (l, u) = zelf.eval(test_ts)?;
-        let time_window_size = match (l, u) {
+        let (lower, upper) = zelf.eval(test_ts)?;
+        let time_window_size = match (lower, upper) {
            (Some(l), Some(u)) => u.sub(&l).map(|r| r.to_std()).transpose().map_err(|_| {
                UnexpectedSnafu {
                    reason: format!(
@@ -121,13 +124,59 @@ impl TimeWindowExpr {
            _ => None,
        };
        zelf.eval_time_window_size = time_window_size;
+        zelf.eval_time_original = lower;
+
        Ok(zelf)
    }

+    /// TODO(discord9): add `eval_batch` too
    pub fn eval(
        &self,
        current: Timestamp,
    ) -> Result<(Option<Timestamp>, Option<Timestamp>), Error> {
+        fn compute_distance(time_diff_ns: i64, stride_ns: i64) -> i64 {
+            if stride_ns == 0 {
+                return time_diff_ns;
+            }
+            // a - (a % n) impl ceil to nearest n * stride
+            let time_delta = time_diff_ns - (time_diff_ns % stride_ns);
+
+            if time_diff_ns < 0 && time_delta != time_diff_ns {
+                // The origin is later than the source timestamp, round down to the previous bin
+
+                time_delta - stride_ns
+            } else {
+                time_delta
+            }
+        }
+
+        // FAST PATH: if we have eval_time_original and eval_time_window_size,
+        // we can compute the bounds directly
+        if let (Some(original), Some(window_size)) =
+            (self.eval_time_original, self.eval_time_window_size)
+        {
+            // date_bin align current to lower bound
+            let time_diff_ns = current.sub(&original).and_then(|s|s.num_nanoseconds()).with_context(||UnexpectedSnafu {
+                reason: format!(
+                    "Failed to compute time difference between current {current:?} and original {original:?}"
+                ),
+            })?;
+
+            let window_size_ns = window_size.as_nanos() as i64;
+
+            let distance_ns = compute_distance(time_diff_ns, window_size_ns);
+
+            let lower_bound = if distance_ns >= 0 {
+                original.add_duration(std::time::Duration::from_nanos(distance_ns as u64))
+            } else {
+                original.sub_duration(std::time::Duration::from_nanos((-distance_ns) as u64))
+            }
+            .context(TimeSnafu)?;
+            let upper_bound = lower_bound.add_duration(window_size).context(TimeSnafu)?;
+
+            return Ok((Some(lower_bound), Some(upper_bound)));
+        }
+
        let lower_bound =
            calc_expr_time_window_lower_bound(&self.phy_expr, &self.df_schema, current)?;
        let upper_bound =
--- a/src/frontend/Cargo.toml
+++ b/src/frontend/Cargo.toml
@@ -14,6 +14,7 @@ workspace = true
 [dependencies]
 api.workspace = true
 arc-swap = "1.0"
+async-stream.workspace = true
 async-trait.workspace = true
 auth.workspace = true
 bytes.workspace = true
--- a/src/frontend/src/error.rs
+++ b/src/frontend/src/error.rs
@@ -363,6 +363,12 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
+
+    #[snafu(display("Canceling statement due to statement timeout"))]
+    StatementTimeout {
+        #[snafu(implicit)]
+        location: Location,
+    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -443,6 +449,8 @@ impl ErrorExt for Error {
            Error::DataFusion { error, .. } => datafusion_status_code::<Self>(error, None),

            Error::Cancelled { .. } => StatusCode::Cancelled,
+
+            Error::StatementTimeout { .. } => StatusCode::Cancelled,
        }
    }

--- a/src/frontend/src/instance.rs
+++ b/src/frontend/src/instance.rs
@@ -25,9 +25,11 @@ mod promql;
 mod region_query;
 pub mod standalone;

+use std::pin::Pin;
 use std::sync::Arc;
-use std::time::SystemTime;
+use std::time::{Duration, SystemTime};

+use async_stream::stream;
 use async_trait::async_trait;
 use auth::{PermissionChecker, PermissionCheckerRef, PermissionReq};
 use catalog::process_manager::ProcessManagerRef;
@@ -44,8 +46,11 @@ use common_procedure::local::{LocalManager, ManagerConfig};
 use common_procedure::options::ProcedureConfig;
 use common_procedure::ProcedureManagerRef;
 use common_query::Output;
+use common_recordbatch::error::StreamTimeoutSnafu;
+use common_recordbatch::RecordBatchStreamWrapper;
 use common_telemetry::{debug, error, info, tracing};
 use datafusion_expr::LogicalPlan;
+use futures::{Stream, StreamExt};
 use log_store::raft_engine::RaftEngineBackend;
 use operator::delete::DeleterRef;
 use operator::insert::InserterRef;
@@ -65,20 +70,21 @@ use servers::interceptor::{
 };
 use servers::prometheus_handler::PrometheusHandler;
 use servers::query_handler::sql::SqlQueryHandler;
-use session::context::QueryContextRef;
+use session::context::{Channel, QueryContextRef};
 use session::table_name::table_idents_to_full_name;
 use snafu::prelude::*;
 use sql::dialect::Dialect;
 use sql::parser::{ParseOptions, ParserContext};
 use sql::statements::copy::{CopyDatabase, CopyTable};
 use sql::statements::statement::Statement;
+use sql::statements::tql::Tql;
 use sqlparser::ast::ObjectName;
 pub use standalone::StandaloneDatanodeManager;

 use crate::error::{
    self, Error, ExecLogicalPlanSnafu, ExecutePromqlSnafu, ExternalSnafu, InvalidSqlSnafu,
    ParseSqlSnafu, PermissionSnafu, PlanStatementSnafu, Result, SqlExecInterceptedSnafu,
-    TableOperationSnafu,
+    StatementTimeoutSnafu, TableOperationSnafu,
 };
 use crate::limiter::LimiterRef;
 use crate::slow_query_recorder::SlowQueryRecorder;
@@ -188,56 +194,7 @@ impl Instance {
            Some(query_ctx.process_id()),
        );

-        let query_fut = async {
-            match stmt {
-                Statement::Query(_) | Statement::Explain(_) | Statement::Delete(_) => {
-                    // TODO: remove this when format is supported in datafusion
-                    if let Statement::Explain(explain) = &stmt {
-                        if let Some(format) = explain.format() {
-                            query_ctx.set_explain_format(format.to_string());
-                        }
-                    }
-
-                    let stmt = QueryStatement::Sql(stmt);
-                    let plan = self
-                        .statement_executor
-                        .plan(&stmt, query_ctx.clone())
-                        .await?;
-
-                    let QueryStatement::Sql(stmt) = stmt else {
-                        unreachable!()
-                    };
-                    query_interceptor.pre_execute(&stmt, Some(&plan), query_ctx.clone())?;
-                    self.statement_executor
-                        .exec_plan(plan, query_ctx)
-                        .await
-                        .context(TableOperationSnafu)
-                }
-                Statement::Tql(tql) => {
-                    let plan = self
-                        .statement_executor
-                        .plan_tql(tql.clone(), &query_ctx)
-                        .await?;
-
-                    query_interceptor.pre_execute(
-                        &Statement::Tql(tql),
-                        Some(&plan),
-                        query_ctx.clone(),
-                    )?;
-                    self.statement_executor
-                        .exec_plan(plan, query_ctx)
-                        .await
-                        .context(TableOperationSnafu)
-                }
-                _ => {
-                    query_interceptor.pre_execute(&stmt, None, query_ctx.clone())?;
-                    self.statement_executor
-                        .execute_sql(stmt, query_ctx)
-                        .await
-                        .context(TableOperationSnafu)
-                }
-            }
-        };
+        let query_fut = self.exec_statement_with_timeout(stmt, query_ctx, query_interceptor);

        CancellableFuture::new(query_fut, ticket.cancellation_handle.clone())
            .await
@@ -254,6 +211,153 @@ impl Instance {
                Output { data, meta }
            })
    }
+
+    async fn exec_statement_with_timeout(
+        &self,
+        stmt: Statement,
+        query_ctx: QueryContextRef,
+        query_interceptor: Option<&SqlQueryInterceptorRef<Error>>,
+    ) -> Result<Output> {
+        let timeout = derive_timeout(&stmt, &query_ctx);
+        match timeout {
+            Some(timeout) => {
+                let start = tokio::time::Instant::now();
+                let output = tokio::time::timeout(
+                    timeout,
+                    self.exec_statement(stmt, query_ctx, query_interceptor),
+                )
+                .await
+                .map_err(|_| StatementTimeoutSnafu.build())??;
+                // compute remaining timeout
+                let remaining_timeout = timeout.checked_sub(start.elapsed()).unwrap_or_default();
+                attach_timeout(output, remaining_timeout)
+            }
+            None => {
+                self.exec_statement(stmt, query_ctx, query_interceptor)
+                    .await
+            }
+        }
+    }
+
+    async fn exec_statement(
+        &self,
+        stmt: Statement,
+        query_ctx: QueryContextRef,
+        query_interceptor: Option<&SqlQueryInterceptorRef<Error>>,
+    ) -> Result<Output> {
+        match stmt {
+            Statement::Query(_) | Statement::Explain(_) | Statement::Delete(_) => {
+                // TODO: remove this when format is supported in datafusion
+                if let Statement::Explain(explain) = &stmt {
+                    if let Some(format) = explain.format() {
+                        query_ctx.set_explain_format(format.to_string());
+                    }
+                }
+
+                self.plan_and_exec_sql(stmt, &query_ctx, query_interceptor)
+                    .await
+            }
+            Statement::Tql(tql) => {
+                self.plan_and_exec_tql(&query_ctx, query_interceptor, tql)
+                    .await
+            }
+            _ => {
+                query_interceptor.pre_execute(&stmt, None, query_ctx.clone())?;
+                self.statement_executor
+                    .execute_sql(stmt, query_ctx)
+                    .await
+                    .context(TableOperationSnafu)
+            }
+        }
+    }
+
+    async fn plan_and_exec_sql(
+        &self,
+        stmt: Statement,
+        query_ctx: &QueryContextRef,
+        query_interceptor: Option<&SqlQueryInterceptorRef<Error>>,
+    ) -> Result<Output> {
+        let stmt = QueryStatement::Sql(stmt);
+        let plan = self
+            .statement_executor
+            .plan(&stmt, query_ctx.clone())
+            .await?;
+        let QueryStatement::Sql(stmt) = stmt else {
+            unreachable!()
+        };
+        query_interceptor.pre_execute(&stmt, Some(&plan), query_ctx.clone())?;
+        self.statement_executor
+            .exec_plan(plan, query_ctx.clone())
+            .await
+            .context(TableOperationSnafu)
+    }
+
+    async fn plan_and_exec_tql(
+        &self,
+        query_ctx: &QueryContextRef,
+        query_interceptor: Option<&SqlQueryInterceptorRef<Error>>,
+        tql: Tql,
+    ) -> Result<Output> {
+        let plan = self
+            .statement_executor
+            .plan_tql(tql.clone(), query_ctx)
+            .await?;
+        query_interceptor.pre_execute(&Statement::Tql(tql), Some(&plan), query_ctx.clone())?;
+        self.statement_executor
+            .exec_plan(plan, query_ctx.clone())
+            .await
+            .context(TableOperationSnafu)
+    }
+}
+
+/// If the relevant variables are set, the timeout is enforced for all PostgreSQL statements.
+/// For MySQL, it applies only to read-only statements.
+fn derive_timeout(stmt: &Statement, query_ctx: &QueryContextRef) -> Option<Duration> {
+    let query_timeout = query_ctx.query_timeout()?;
+    if query_timeout.is_zero() {
+        return None;
+    }
+    match query_ctx.channel() {
+        Channel::Mysql if stmt.is_readonly() => Some(query_timeout),
+        Channel::Postgres => Some(query_timeout),
+        _ => None,
+    }
+}
+
+fn attach_timeout(output: Output, mut timeout: Duration) -> Result<Output> {
+    if timeout.is_zero() {
+        return StatementTimeoutSnafu.fail();
+    }
+
+    let output = match output.data {
+        OutputData::AffectedRows(_) | OutputData::RecordBatches(_) => output,
+        OutputData::Stream(mut stream) => {
+            let schema = stream.schema();
+            let s = Box::pin(stream! {
+                let mut start = tokio::time::Instant::now();
+                while let Some(item) = tokio::time::timeout(timeout, stream.next()).await.map_err(|_| StreamTimeoutSnafu.build())? {
+                    yield item;
+
+                    let now = tokio::time::Instant::now();
+                    timeout = timeout.checked_sub(now - start).unwrap_or(Duration::ZERO);
+                    start = now;
+                    // tokio::time::timeout may not return an error immediately when timeout is 0.
+                    if timeout.is_zero() {
+                        StreamTimeoutSnafu.fail()?;
+                    }
+                }
+            }) as Pin<Box<dyn Stream<Item = _> + Send>>;
+            let stream = RecordBatchStreamWrapper {
+                schema,
+                stream: s,
+                output_ordering: None,
+                metrics: Default::default(),
+            };
+            Output::new(OutputData::Stream(Box::pin(stream)), output.meta)
+        }
+    };
+
+    Ok(output)
 }

 #[async_trait]
@@ -276,6 +380,13 @@ impl SqlQueryHandler for Instance {
            .and_then(|stmts| query_interceptor.post_parsing(stmts, query_ctx.clone()))
        {
            Ok(stmts) => {
+                if stmts.is_empty() {
+                    return vec![InvalidSqlSnafu {
+                        err_msg: "empty statements",
+                    }
+                    .fail()];
+                }
+
                let mut results = Vec::with_capacity(stmts.len());
                for stmt in stmts {
                    if let Err(e) = checker
--- a/src/index/src/bloom_filter/applier.rs
+++ b/src/index/src/bloom_filter/applier.rs
@@ -218,6 +218,7 @@ mod tests {
        let mut writer = Cursor::new(Vec::new());
        let mut creator = BloomFilterCreator::new(
            4,
+            0.01,
            Arc::new(MockExternalTempFileProvider::new()),
            Arc::new(AtomicUsize::new(0)),
            None,
--- a/src/index/src/bloom_filter/creator.rs
+++ b/src/index/src/bloom_filter/creator.rs
@@ -30,9 +30,6 @@ use crate::bloom_filter::SEED;
 use crate::external_provider::ExternalTempFileProvider;
 use crate::Bytes;

-/// The false positive rate of the Bloom filter.
-pub const FALSE_POSITIVE_RATE: f64 = 0.01;
-
 /// `BloomFilterCreator` is responsible for creating and managing bloom filters
 /// for a set of elements. It divides the rows into segments and creates
 /// bloom filters for each segment.
@@ -79,6 +76,7 @@ impl BloomFilterCreator {
    /// `rows_per_segment` <= 0
    pub fn new(
        rows_per_segment: usize,
+        false_positive_rate: f64,
        intermediate_provider: Arc<dyn ExternalTempFileProvider>,
        global_memory_usage: Arc<AtomicUsize>,
        global_memory_usage_threshold: Option<usize>,
@@ -95,6 +93,7 @@ impl BloomFilterCreator {
            cur_seg_distinct_elems_mem_usage: 0,
            global_memory_usage: global_memory_usage.clone(),
            finalized_bloom_filters: FinalizedBloomFilterStorage::new(
+                false_positive_rate,
                intermediate_provider,
                global_memory_usage,
                global_memory_usage_threshold,
@@ -263,6 +262,7 @@ mod tests {
        let mut writer = Cursor::new(Vec::new());
        let mut creator = BloomFilterCreator::new(
            2,
+            0.01,
            Arc::new(MockExternalTempFileProvider::new()),
            Arc::new(AtomicUsize::new(0)),
            None,
@@ -337,6 +337,7 @@ mod tests {
        let mut writer = Cursor::new(Vec::new());
        let mut creator: BloomFilterCreator = BloomFilterCreator::new(
            2,
+            0.01,
            Arc::new(MockExternalTempFileProvider::new()),
            Arc::new(AtomicUsize::new(0)),
            None,
@@ -418,6 +419,7 @@ mod tests {
        let mut writer = Cursor::new(Vec::new());
        let mut creator = BloomFilterCreator::new(
            2,
+            0.01,
            Arc::new(MockExternalTempFileProvider::new()),
            Arc::new(AtomicUsize::new(0)),
            None,
--- a/src/index/src/bloom_filter/creator/finalize_segment.rs
+++ b/src/index/src/bloom_filter/creator/finalize_segment.rs
@@ -23,7 +23,7 @@ use futures::{stream, AsyncWriteExt, Stream};
 use snafu::ResultExt;

 use crate::bloom_filter::creator::intermediate_codec::IntermediateBloomFilterCodecV1;
-use crate::bloom_filter::creator::{FALSE_POSITIVE_RATE, SEED};
+use crate::bloom_filter::creator::SEED;
 use crate::bloom_filter::error::{IntermediateSnafu, IoSnafu, Result};
 use crate::external_provider::ExternalTempFileProvider;
 use crate::Bytes;
@@ -33,6 +33,9 @@ const MIN_MEMORY_USAGE_THRESHOLD: usize = 1024 * 1024; // 1MB

 /// Storage for finalized Bloom filters.
 pub struct FinalizedBloomFilterStorage {
+    /// The false positive rate of the Bloom filter.
+    false_positive_rate: f64,
+
    /// Indices of the segments in the sequence of finalized Bloom filters.
    segment_indices: Vec<usize>,

@@ -65,12 +68,14 @@ pub struct FinalizedBloomFilterStorage {
 impl FinalizedBloomFilterStorage {
    /// Creates a new `FinalizedBloomFilterStorage`.
    pub fn new(
+        false_positive_rate: f64,
        intermediate_provider: Arc<dyn ExternalTempFileProvider>,
        global_memory_usage: Arc<AtomicUsize>,
        global_memory_usage_threshold: Option<usize>,
    ) -> Self {
        let external_prefix = format!("intm-bloom-filters-{}", uuid::Uuid::new_v4());
        Self {
+            false_positive_rate,
            segment_indices: Vec::new(),
            in_memory: Vec::new(),
            intermediate_file_id_counter: 0,
@@ -96,7 +101,7 @@ impl FinalizedBloomFilterStorage {
        elems: impl IntoIterator<Item = Bytes>,
        element_count: usize,
    ) -> Result<()> {
-        let mut bf = BloomFilter::with_false_pos(FALSE_POSITIVE_RATE)
+        let mut bf = BloomFilter::with_false_pos(self.false_positive_rate)
            .seed(&SEED)
            .expected_items(element_count);
        for elem in elems.into_iter() {
@@ -284,6 +289,7 @@ mod tests {
        let global_memory_usage_threshold = Some(1024 * 1024); // 1MB
        let provider = Arc::new(mock_provider);
        let mut storage = FinalizedBloomFilterStorage::new(
+            0.01,
            provider,
            global_memory_usage.clone(),
            global_memory_usage_threshold,
@@ -340,6 +346,7 @@ mod tests {
        let global_memory_usage_threshold = Some(1024 * 1024); // 1MB
        let provider = Arc::new(mock_provider);
        let mut storage = FinalizedBloomFilterStorage::new(
+            0.01,
            provider,
            global_memory_usage.clone(),
            global_memory_usage_threshold,
--- a/src/index/src/bloom_filter/reader.rs
+++ b/src/index/src/bloom_filter/reader.rs
@@ -222,6 +222,7 @@ mod tests {
        let mut writer = Cursor::new(vec![]);
        let mut creator = BloomFilterCreator::new(
            2,
+            0.01,
            Arc::new(MockExternalTempFileProvider::new()),
            Arc::new(AtomicUsize::new(0)),
            None,
--- a/src/index/src/fulltext_index/create/bloom_filter.rs
+++ b/src/index/src/fulltext_index/create/bloom_filter.rs
@@ -45,6 +45,7 @@ impl BloomFilterFulltextIndexCreator {
    pub fn new(
        config: Config,
        rows_per_segment: usize,
+        false_positive_rate: f64,
        intermediate_provider: Arc<dyn ExternalTempFileProvider>,
        global_memory_usage: Arc<AtomicUsize>,
        global_memory_usage_threshold: Option<usize>,
@@ -57,6 +58,7 @@ impl BloomFilterFulltextIndexCreator {

        let inner = BloomFilterCreator::new(
            rows_per_segment,
+            false_positive_rate,
            intermediate_provider,
            global_memory_usage,
            global_memory_usage_threshold,
--- a/src/meta-srv/src/handler/remap_flow_peer_handler.rs
+++ b/src/meta-srv/src/handler/remap_flow_peer_handler.rs
@@ -13,6 +13,7 @@
 // limitations under the License.

 use api::v1::meta::{HeartbeatRequest, Peer, Role};
+use common_meta::instruction::CacheIdent;
 use common_meta::key::node_address::{NodeAddressKey, NodeAddressValue};
 use common_meta::key::{MetadataKey, MetadataValue};
 use common_meta::rpc::store::PutRequest;
@@ -80,7 +81,19 @@ async fn rewrite_node_address(ctx: &mut Context, peer: &Peer) {
        match ctx.leader_cached_kv_backend.put(put).await {
            Ok(_) => {
                info!("Successfully updated flow `NodeAddressValue`: {:?}", peer);
-                // TODO(discord): broadcast invalidating cache to all frontends
+                // broadcast invalidating cache to all frontends
+                let cache_idents = vec![CacheIdent::FlowNodeAddressChange(peer.id)];
+                info!(
+                    "Invalidate flow node cache for new address with cache idents: {:?}",
+                    cache_idents
+                );
+                if let Err(e) = ctx
+                    .cache_invalidator
+                    .invalidate(&Default::default(), &cache_idents)
+                    .await
+                {
+                    error!(e; "Failed to invalidate {} `NodeAddressKey` cache, peer: {:?}", cache_idents.len(), peer);
+                }
            }
            Err(e) => {
                error!(e; "Failed to update flow `NodeAddressValue`: {:?}", peer);
--- a/src/meta-srv/src/metasrv/builder.rs
+++ b/src/meta-srv/src/metasrv/builder.rs
@@ -353,30 +353,28 @@ impl MetasrvBuilder {

        let leader_region_registry = Arc::new(LeaderRegionRegistry::default());

+        let ddl_context = DdlContext {
+            node_manager,
+            cache_invalidator: cache_invalidator.clone(),
+            memory_region_keeper: memory_region_keeper.clone(),
+            leader_region_registry: leader_region_registry.clone(),
+            table_metadata_manager: table_metadata_manager.clone(),
+            table_metadata_allocator: table_metadata_allocator.clone(),
+            flow_metadata_manager: flow_metadata_manager.clone(),
+            flow_metadata_allocator: flow_metadata_allocator.clone(),
+            region_failure_detector_controller,
+        };
+        let procedure_manager_c = procedure_manager.clone();
+        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager_c, true)
+            .context(error::InitDdlManagerSnafu)?;
        #[cfg(feature = "enterprise")]
-        let trigger_ddl_manager = plugins
-            .as_ref()
-            .and_then(|plugins| plugins.get::<common_meta::ddl_manager::TriggerDdlManagerRef>());
-        let ddl_manager = Arc::new(
-            DdlManager::try_new(
-                DdlContext {
-                    node_manager,
-                    cache_invalidator: cache_invalidator.clone(),
-                    memory_region_keeper: memory_region_keeper.clone(),
-                    leader_region_registry: leader_region_registry.clone(),
-                    table_metadata_manager: table_metadata_manager.clone(),
-                    table_metadata_allocator: table_metadata_allocator.clone(),
-                    flow_metadata_manager: flow_metadata_manager.clone(),
-                    flow_metadata_allocator: flow_metadata_allocator.clone(),
-                    region_failure_detector_controller,
-                },
-                procedure_manager.clone(),
-                true,
-                #[cfg(feature = "enterprise")]
-                trigger_ddl_manager,
-            )
-            .context(error::InitDdlManagerSnafu)?,
-        );
+        let ddl_manager = {
+            let trigger_ddl_manager = plugins.as_ref().and_then(|plugins| {
+                plugins.get::<common_meta::ddl_manager::TriggerDdlManagerRef>()
+            });
+            ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
+        };
+        let ddl_manager = Arc::new(ddl_manager);

        // remote WAL prune ticker and manager
        let wal_prune_ticker = if is_remote_wal && options.wal.enable_active_wal_pruning() {
--- a/src/metric-engine/src/data_region.rs
+++ b/src/metric-engine/src/data_region.rs
@@ -145,12 +145,19 @@ impl DataRegion {
                    IndexOptions::Inverted => {
                        c.column_schema.set_inverted_index(true);
                    }
-                    IndexOptions::Skipping { granularity } => {
+                    IndexOptions::Skipping {
+                        granularity,
+                        false_positive_rate,
+                    } => {
                        c.column_schema
-                            .set_skipping_options(&SkippingIndexOptions {
-                                granularity,
-                                index_type: SkippingIndexType::BloomFilter,
-                            })
+                            .set_skipping_options(
+                                &SkippingIndexOptions::new(
+                                    granularity,
+                                    false_positive_rate,
+                                    SkippingIndexType::BloomFilter,
+                                )
+                                .context(SetSkippingIndexOptionSnafu)?,
+                            )
                            .context(SetSkippingIndexOptionSnafu)?;
                    }
                }
--- a/src/metric-engine/src/engine.rs
+++ b/src/metric-engine/src/engine.rs
@@ -473,8 +473,9 @@ struct MetricEngineInner {
 mod test {
    use std::collections::HashMap;

+    use common_telemetry::info;
    use store_api::metric_engine_consts::PHYSICAL_TABLE_METADATA_KEY;
-    use store_api::region_request::{RegionCloseRequest, RegionOpenRequest};
+    use store_api::region_request::{RegionCloseRequest, RegionFlushRequest, RegionOpenRequest};

    use super::*;
    use crate::test_util::TestEnv;
@@ -559,4 +560,90 @@ mod test {
        assert!(env.metric().region_statistic(logical_region_id).is_none());
        assert!(env.metric().region_statistic(physical_region_id).is_some());
    }
+
+    #[tokio::test]
+    async fn test_open_region_failure() {
+        let env = TestEnv::new().await;
+        env.init_metric_region().await;
+        let physical_region_id = env.default_physical_region_id();
+
+        let metric_engine = env.metric();
+        metric_engine
+            .handle_request(
+                physical_region_id,
+                RegionRequest::Flush(RegionFlushRequest {
+                    row_group_size: None,
+                }),
+            )
+            .await
+            .unwrap();
+
+        let path = format!("{}/metadata/", env.default_region_dir());
+        let object_store = env.get_object_store().unwrap();
+        let list = object_store.list(&path).await.unwrap();
+        // Delete parquet files in metadata region
+        for entry in list {
+            if entry.metadata().is_dir() {
+                continue;
+            }
+            if entry.name().ends_with("parquet") {
+                info!("deleting {}", entry.path());
+                object_store.delete(entry.path()).await.unwrap();
+            }
+        }
+
+        let physical_region_option = [(PHYSICAL_TABLE_METADATA_KEY.to_string(), String::new())]
+            .into_iter()
+            .collect();
+        let open_request = RegionOpenRequest {
+            engine: METRIC_ENGINE_NAME.to_string(),
+            region_dir: env.default_region_dir(),
+            options: physical_region_option,
+            skip_wal_replay: false,
+        };
+        // Opening an already opened region should succeed.
+        // Since the region is already open, no metadata recovery operations will be performed.
+        metric_engine
+            .handle_request(physical_region_id, RegionRequest::Open(open_request))
+            .await
+            .unwrap();
+
+        // Close the region
+        metric_engine
+            .handle_request(
+                physical_region_id,
+                RegionRequest::Close(RegionCloseRequest {}),
+            )
+            .await
+            .unwrap();
+
+        // Try to reopen region.
+        let physical_region_option = [(PHYSICAL_TABLE_METADATA_KEY.to_string(), String::new())]
+            .into_iter()
+            .collect();
+        let open_request = RegionOpenRequest {
+            engine: METRIC_ENGINE_NAME.to_string(),
+            region_dir: env.default_region_dir(),
+            options: physical_region_option,
+            skip_wal_replay: false,
+        };
+        let err = metric_engine
+            .handle_request(physical_region_id, RegionRequest::Open(open_request))
+            .await
+            .unwrap_err();
+        // Failed to open region because of missing parquet files.
+        assert_eq!(err.status_code(), StatusCode::StorageUnavailable);
+
+        let mito_engine = metric_engine.mito();
+        let data_region_id = utils::to_data_region_id(physical_region_id);
+        let metadata_region_id = utils::to_metadata_region_id(physical_region_id);
+        // The metadata/data region should be closed.
+        let err = mito_engine.get_metadata(data_region_id).await.unwrap_err();
+        assert_eq!(err.status_code(), StatusCode::RegionNotFound);
+        let err = mito_engine
+            .get_metadata(metadata_region_id)
+            .await
+            .unwrap_err();
+        assert_eq!(err.status_code(), StatusCode::RegionNotFound);
+    }
 }
--- a/src/metric-engine/src/engine/close.rs
+++ b/src/metric-engine/src/engine/close.rs
@@ -59,7 +59,7 @@ impl MetricEngineInner {
        }
    }

-    async fn close_physical_region(&self, region_id: RegionId) -> Result<AffectedRows> {
+    pub(crate) async fn close_physical_region(&self, region_id: RegionId) -> Result<AffectedRows> {
        let data_region_id = utils::to_data_region_id(region_id);
        let metadata_region_id = utils::to_metadata_region_id(region_id);

--- a/src/metric-engine/src/engine/create.rs
+++ b/src/metric-engine/src/engine/create.rs
@@ -55,6 +55,7 @@ use crate::utils::{
 };

 const DEFAULT_TABLE_ID_SKIPPING_INDEX_GRANULARITY: u32 = 1024;
+const DEFAULT_TABLE_ID_SKIPPING_INDEX_FALSE_POSITIVE_RATE: f64 = 0.01;

 impl MetricEngineInner {
    pub async fn create_regions(
@@ -542,10 +543,11 @@ impl MetricEngineInner {
                ConcreteDataType::uint32_datatype(),
                false,
            )
-            .with_skipping_options(SkippingIndexOptions {
-                granularity: DEFAULT_TABLE_ID_SKIPPING_INDEX_GRANULARITY,
-                index_type: datatypes::schema::SkippingIndexType::BloomFilter,
-            })
+            .with_skipping_options(SkippingIndexOptions::new_unchecked(
+                DEFAULT_TABLE_ID_SKIPPING_INDEX_GRANULARITY,
+                DEFAULT_TABLE_ID_SKIPPING_INDEX_FALSE_POSITIVE_RATE,
+                datatypes::schema::SkippingIndexType::BloomFilter,
+            ))
            .unwrap(),
        };
        let tsid_col = ColumnMetadata {
--- a/src/metric-engine/src/engine/open.rs
+++ b/src/metric-engine/src/engine/open.rs
@@ -17,7 +17,7 @@
 use api::region::RegionResponse;
 use api::v1::SemanticType;
 use common_error::ext::BoxedError;
-use common_telemetry::info;
+use common_telemetry::{error, info, warn};
 use datafusion::common::HashMap;
 use mito2::engine::MITO_ENGINE_NAME;
 use object_store::util::join_dir;
@@ -94,6 +94,21 @@ impl MetricEngineInner {
        Ok(responses)
    }

+    // If the metadata region is opened with a stale manifest,
+    // the metric engine may fail to recover logical tables from the metadata region,
+    // as the manifest could reference files that have already been deleted
+    // due to compaction operations performed by the region leader.
+    async fn close_physical_region_on_recovery_failure(&self, physical_region_id: RegionId) {
+        info!(
+            "Closing metadata region {} and data region {} on metadata recovery failure",
+            utils::to_metadata_region_id(physical_region_id),
+            utils::to_data_region_id(physical_region_id)
+        );
+        if let Err(err) = self.close_physical_region(physical_region_id).await {
+            error!(err; "Failed to close physical region {}", physical_region_id);
+        }
+    }
+
    async fn open_physical_region_with_results(
        &self,
        metadata_region_result: Option<std::result::Result<RegionResponse, BoxedError>>,
@@ -119,8 +134,14 @@ impl MetricEngineInner {
                region_type: "data",
            })?;

-        self.recover_states(physical_region_id, physical_region_options)
-            .await?;
+        if let Err(err) = self
+            .recover_states(physical_region_id, physical_region_options)
+            .await
+        {
+            self.close_physical_region_on_recovery_failure(physical_region_id)
+                .await;
+            return Err(err);
+        }
        Ok(data_region_response)
    }

@@ -139,11 +160,31 @@ impl MetricEngineInner {
        request: RegionOpenRequest,
    ) -> Result<AffectedRows> {
        if request.is_physical_table() {
+            if self
+                .state
+                .read()
+                .unwrap()
+                .physical_region_states()
+                .get(&region_id)
+                .is_some()
+            {
+                warn!(
+                    "The physical region {} is already open, ignore the open request",
+                    region_id
+                );
+                return Ok(0);
+            }
            // open physical region and recover states
            let physical_region_options = PhysicalRegionOptions::try_from(&request.options)?;
            self.open_physical_region(region_id, request).await?;
-            self.recover_states(region_id, physical_region_options)
-                .await?;
+            if let Err(err) = self
+                .recover_states(region_id, physical_region_options)
+                .await
+            {
+                self.close_physical_region_on_recovery_failure(region_id)
+                    .await;
+                return Err(err);
+            }

            Ok(0)
        } else {
--- a/src/metric-engine/src/engine/options.rs
+++ b/src/metric-engine/src/engine/options.rs
@@ -17,6 +17,8 @@
 use std::collections::HashMap;

 use store_api::metric_engine_consts::{
+    METRIC_ENGINE_INDEX_SKIPPING_INDEX_FALSE_POSITIVE_RATE_OPTION,
+    METRIC_ENGINE_INDEX_SKIPPING_INDEX_FALSE_POSITIVE_RATE_OPTION_DEFAULT,
    METRIC_ENGINE_INDEX_SKIPPING_INDEX_GRANULARITY_OPTION,
    METRIC_ENGINE_INDEX_SKIPPING_INDEX_GRANULARITY_OPTION_DEFAULT, METRIC_ENGINE_INDEX_TYPE_OPTION,
 };
@@ -31,19 +33,20 @@ use crate::error::{Error, ParseRegionOptionsSnafu, Result};
 const SEG_ROW_COUNT_FOR_DATA_REGION: u32 = 256;

 /// Physical region options.
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, PartialEq)]
 pub struct PhysicalRegionOptions {
    pub index: IndexOptions,
 }

 /// Index options for auto created columns
-#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Default, PartialEq)]
 pub enum IndexOptions {
    #[default]
    None,
    Inverted,
    Skipping {
        granularity: u32,
+        false_positive_rate: f64,
    },
 }

@@ -54,6 +57,7 @@ pub fn set_data_region_options(
 ) {
    options.remove(METRIC_ENGINE_INDEX_TYPE_OPTION);
    options.remove(METRIC_ENGINE_INDEX_SKIPPING_INDEX_GRANULARITY_OPTION);
+    options.remove(METRIC_ENGINE_INDEX_SKIPPING_INDEX_FALSE_POSITIVE_RATE_OPTION);
    options.insert(
        "index.inverted_index.segment_row_count".to_string(),
        SEG_ROW_COUNT_FOR_DATA_REGION.to_string(),
@@ -93,7 +97,23 @@ impl TryFrom<&HashMap<String, String>> for PhysicalRegionOptions {
                            })
                        },
                    )?;
-                Ok(IndexOptions::Skipping { granularity })
+                let false_positive_rate = value
+                    .get(METRIC_ENGINE_INDEX_SKIPPING_INDEX_FALSE_POSITIVE_RATE_OPTION)
+                    .map_or(
+                        Ok(METRIC_ENGINE_INDEX_SKIPPING_INDEX_FALSE_POSITIVE_RATE_OPTION_DEFAULT),
+                        |f| {
+                            f.parse().ok().filter(|f| *f > 0.0 && *f <= 1.0).ok_or(
+                                ParseRegionOptionsSnafu {
+                                    reason: format!("Invalid false positive rate: {}", f),
+                                }
+                                .build(),
+                            )
+                        },
+                    )?;
+                Ok(IndexOptions::Skipping {
+                    granularity,
+                    false_positive_rate,
+                })
            }
            Some(index_type) => ParseRegionOptionsSnafu {
                reason: format!("Invalid index type: {}", index_type),
@@ -121,11 +141,16 @@ mod tests {
            METRIC_ENGINE_INDEX_SKIPPING_INDEX_GRANULARITY_OPTION.to_string(),
            "102400".to_string(),
        );
+        options.insert(
+            METRIC_ENGINE_INDEX_SKIPPING_INDEX_FALSE_POSITIVE_RATE_OPTION.to_string(),
+            "0.01".to_string(),
+        );
        set_data_region_options(&mut options, false);

        for key in [
            METRIC_ENGINE_INDEX_TYPE_OPTION,
            METRIC_ENGINE_INDEX_SKIPPING_INDEX_GRANULARITY_OPTION,
+            METRIC_ENGINE_INDEX_SKIPPING_INDEX_FALSE_POSITIVE_RATE_OPTION,
        ] {
            assert_eq!(options.get(key), None);
        }
@@ -154,11 +179,16 @@ mod tests {
            METRIC_ENGINE_INDEX_SKIPPING_INDEX_GRANULARITY_OPTION.to_string(),
            "102400".to_string(),
        );
+        options.insert(
+            METRIC_ENGINE_INDEX_SKIPPING_INDEX_FALSE_POSITIVE_RATE_OPTION.to_string(),
+            "0.01".to_string(),
+        );
        let physical_region_options = PhysicalRegionOptions::try_from(&options).unwrap();
        assert_eq!(
            physical_region_options.index,
            IndexOptions::Skipping {
-                granularity: 102400
+                granularity: 102400,
+                false_positive_rate: 0.01,
            }
        );
    }
--- a/src/metric-engine/src/test_util.rs
+++ b/src/metric-engine/src/test_util.rs
@@ -23,6 +23,7 @@ use mito2::config::MitoConfig;
 use mito2::engine::MitoEngine;
 use mito2::test_util::TestEnv as MitoTestEnv;
 use object_store::util::join_dir;
+use object_store::ObjectStore;
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::{
    LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME, PHYSICAL_TABLE_METADATA_KEY,
@@ -59,7 +60,7 @@ impl TestEnv {

    /// Returns a new env with specific `prefix` and `config` for test.
    pub async fn with_prefix_and_config(prefix: &str, config: EngineConfig) -> Self {
-        let mut mito_env = MitoTestEnv::with_prefix(prefix);
+        let mut mito_env = MitoTestEnv::with_prefix(prefix).await;
        let mito = mito_env.create_engine(MitoConfig::default()).await;
        let metric = MetricEngine::try_new(mito.clone(), config).unwrap();
        Self {
@@ -74,6 +75,10 @@ impl TestEnv {
        join_dir(&env_root, "data")
    }

+    pub fn get_object_store(&self) -> Option<ObjectStore> {
+        self.mito_env.get_object_store()
+    }
+
    /// Returns a reference to the engine.
    pub fn mito(&self) -> MitoEngine {
        self.mito.clone()
--- a/src/mito2/src/cache/index/inverted_index.rs
+++ b/src/mito2/src/cache/index/inverted_index.rs
@@ -245,7 +245,7 @@ mod test {
        let blob = create_inverted_index_blob().await;

        // Init a test range reader in local fs.
-        let mut env = TestEnv::new();
+        let mut env = TestEnv::new().await;
        let file_size = blob.len() as u64;
        let store = env.init_object_store_manager();
        let temp_path = "data";
--- a/src/mito2/src/cache/index/result_cache.rs
+++ b/src/mito2/src/cache/index/result_cache.rs
@@ -31,6 +31,11 @@ use crate::sst::parquet::row_selection::RowGroupSelection;
 const INDEX_RESULT_TYPE: &str = "index_result";

 /// Cache for storing index query results.
+///
+/// The `RowGroupSelection` is a collection of row groups that match the predicate.
+///
+/// Row groups can be partially searched. Row groups that not contained in `RowGroupSelection` are not searched.
+/// User can retrieve the partial results and handle uncontained row groups required by the predicate subsequently.
 pub struct IndexResultCache {
    cache: Cache<(PredicateKey, FileId), Arc<RowGroupSelection>>,
 }
@@ -64,6 +69,8 @@ impl IndexResultCache {
    }

    /// Puts a query result into the cache.
+    ///
+    /// Allow user to put a partial result (not containing all row groups) into the cache.
    pub fn put(&self, key: PredicateKey, file_id: FileId, result: Arc<RowGroupSelection>) {
        let key = (key, file_id);
        let size = Self::index_result_cache_weight(&key, &result);
@@ -74,6 +81,9 @@ impl IndexResultCache {
    }

    /// Gets a query result from the cache.
+    ///
+    /// Note: the returned `RowGroupSelection` only contains the row groups that are searched.
+    ///       Caller should handle the uncontained row groups required by the predicate subsequently.
    pub fn get(&self, key: &PredicateKey, file_id: FileId) -> Option<Arc<RowGroupSelection>> {
        let res = self.cache.get(&(key.clone(), file_id));
        if res.is_some() {
--- a/src/mito2/src/cache/write_cache.rs
+++ b/src/mito2/src/cache/write_cache.rs
@@ -449,7 +449,7 @@ mod tests {
    async fn test_write_and_upload_sst() {
        // TODO(QuenKar): maybe find a way to create some object server for testing,
        // and now just use local file system to mock.
-        let mut env = TestEnv::new();
+        let mut env = TestEnv::new().await;
        let mock_store = env.init_object_store_manager();
        let path_provider = RegionFilePathFactory::new("test".to_string());

@@ -537,7 +537,7 @@ mod tests {
    #[tokio::test]
    async fn test_read_metadata_from_write_cache() {
        common_telemetry::init_default_ut_logging();
-        let mut env = TestEnv::new();
+        let mut env = TestEnv::new().await;
        let data_home = env.data_home().display().to_string();
        let mock_store = env.init_object_store_manager();

@@ -606,7 +606,7 @@ mod tests {
    #[tokio::test]
    async fn test_write_cache_clean_tmp_files() {
        common_telemetry::init_default_ut_logging();
-        let mut env = TestEnv::new();
+        let mut env = TestEnv::new().await;
        let data_home = env.data_home().display().to_string();
        let mock_store = env.init_object_store_manager();

--- a/src/mito2/src/compaction.rs
+++ b/src/mito2/src/compaction.rs
@@ -62,7 +62,7 @@ use crate::read::BoxedBatchReader;
 use crate::region::options::MergeMode;
 use crate::region::version::VersionControlRef;
 use crate::region::ManifestContextRef;
-use crate::request::{OptionOutputTx, OutputTx, WorkerRequest};
+use crate::request::{OptionOutputTx, OutputTx, WorkerRequestWithTime};
 use crate::schedule::remote_job_scheduler::{
    CompactionJob, DefaultNotifier, RemoteJob, RemoteJobSchedulerRef,
 };
@@ -77,7 +77,7 @@ pub struct CompactionRequest {
    pub(crate) current_version: CompactionVersion,
    pub(crate) access_layer: AccessLayerRef,
    /// Sender to send notification to the region worker.
-    pub(crate) request_sender: mpsc::Sender<WorkerRequest>,
+    pub(crate) request_sender: mpsc::Sender<WorkerRequestWithTime>,
    /// Waiters of the compaction request.
    pub(crate) waiters: Vec<OutputTx>,
    /// Start time of compaction task.
@@ -101,7 +101,7 @@ pub(crate) struct CompactionScheduler {
    /// Compacting regions.
    region_status: HashMap<RegionId, CompactionStatus>,
    /// Request sender of the worker that this scheduler belongs to.
-    request_sender: Sender<WorkerRequest>,
+    request_sender: Sender<WorkerRequestWithTime>,
    cache_manager: CacheManagerRef,
    engine_config: Arc<MitoConfig>,
    listener: WorkerListener,
@@ -112,7 +112,7 @@ pub(crate) struct CompactionScheduler {
 impl CompactionScheduler {
    pub(crate) fn new(
        scheduler: SchedulerRef,
-        request_sender: Sender<WorkerRequest>,
+        request_sender: Sender<WorkerRequestWithTime>,
        cache_manager: CacheManagerRef,
        engine_config: Arc<MitoConfig>,
        listener: WorkerListener,
@@ -559,7 +559,7 @@ impl CompactionStatus {
    #[allow(clippy::too_many_arguments)]
    fn new_compaction_request(
        &mut self,
-        request_sender: Sender<WorkerRequest>,
+        request_sender: Sender<WorkerRequestWithTime>,
        mut waiter: OptionOutputTx,
        engine_config: Arc<MitoConfig>,
        cache_manager: CacheManagerRef,
--- a/src/mito2/src/compaction/task.rs
+++ b/src/mito2/src/compaction/task.rs
@@ -27,6 +27,7 @@ use crate::manifest::action::RegionEdit;
 use crate::metrics::{COMPACTION_FAILURE_COUNT, COMPACTION_STAGE_ELAPSED};
 use crate::request::{
    BackgroundNotify, CompactionFailed, CompactionFinished, OutputTx, WorkerRequest,
+    WorkerRequestWithTime,
 };
 use crate::worker::WorkerListener;
 use crate::{error, metrics};
@@ -37,7 +38,7 @@ pub const MAX_PARALLEL_COMPACTION: usize = 1;
 pub(crate) struct CompactionTaskImpl {
    pub compaction_region: CompactionRegion,
    /// Request sender to notify the worker.
-    pub(crate) request_sender: mpsc::Sender<WorkerRequest>,
+    pub(crate) request_sender: mpsc::Sender<WorkerRequestWithTime>,
    /// Senders that are used to notify waiters waiting for pending compaction tasks.
    pub waiters: Vec<OutputTx>,
    /// Start time of compaction task
@@ -135,7 +136,11 @@ impl CompactionTaskImpl {

    /// Notifies region worker to handle post-compaction tasks.
    async fn send_to_worker(&self, request: WorkerRequest) {
-        if let Err(e) = self.request_sender.send(request).await {
+        if let Err(e) = self
+            .request_sender
+            .send(WorkerRequestWithTime::new(request))
+            .await
+        {
            error!(
                "Failed to notify compaction job status for region {}, request: {:?}",
                self.compaction_region.region_id, e.0
--- a/src/mito2/src/engine/alter_test.rs
+++ b/src/mito2/src/engine/alter_test.rs
@@ -84,12 +84,14 @@ fn alter_column_fulltext_options() -> RegionAlterRequest {
        kind: AlterKind::SetIndex {
            options: ApiSetIndexOptions::Fulltext {
                column_name: "tag_0".to_string(),
-                options: FulltextOptions {
-                    enable: true,
-                    analyzer: FulltextAnalyzer::English,
-                    case_sensitive: false,
-                    backend: FulltextBackend::Bloom,
-                },
+                options: FulltextOptions::new_unchecked(
+                    true,
+                    FulltextAnalyzer::English,
+                    false,
+                    FulltextBackend::Bloom,
+                    1000,
+                    0.01,
+                ),
            },
        },
    }
@@ -115,7 +117,7 @@ fn check_region_version(
 async fn test_alter_region() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -211,7 +213,7 @@ fn build_rows_for_tags(

 #[tokio::test]
 async fn test_put_after_alter() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;
    let region_id = RegionId::new(1, 1);
    let request = CreateRequestBuilder::new().build();
@@ -316,7 +318,7 @@ async fn test_put_after_alter() {
 async fn test_alter_region_retry() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -374,7 +376,7 @@ async fn test_alter_region_retry() {
 async fn test_alter_on_flushing() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let listener = Arc::new(AlterFlushListener::default());
    let engine = env
        .create_engine_with(MitoConfig::default(), None, Some(listener.clone()))
@@ -478,7 +480,7 @@ async fn test_alter_on_flushing() {
 async fn test_alter_column_fulltext_options() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let listener = Arc::new(AlterFlushListener::default());
    let engine = env
        .create_engine_with(MitoConfig::default(), None, Some(listener.clone()))
@@ -553,12 +555,14 @@ async fn test_alter_column_fulltext_options() {
    // Wait for the write job.
    alter_job.await.unwrap();

-    let expect_fulltext_options = FulltextOptions {
-        enable: true,
-        analyzer: FulltextAnalyzer::English,
-        case_sensitive: false,
-        backend: FulltextBackend::Bloom,
-    };
+    let expect_fulltext_options = FulltextOptions::new_unchecked(
+        true,
+        FulltextAnalyzer::English,
+        false,
+        FulltextBackend::Bloom,
+        1000,
+        0.01,
+    );
    let check_fulltext_options = |engine: &MitoEngine, expected: &FulltextOptions| {
        let current_fulltext_options = engine
            .get_region(region_id)
@@ -597,7 +601,7 @@ async fn test_alter_column_fulltext_options() {
 async fn test_alter_column_set_inverted_index() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let listener = Arc::new(AlterFlushListener::default());
    let engine = env
        .create_engine_with(MitoConfig::default(), None, Some(listener.clone()))
@@ -707,7 +711,7 @@ async fn test_alter_column_set_inverted_index() {
 async fn test_alter_region_ttl_options() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let listener = Arc::new(AlterFlushListener::default());
    let engine = env
        .create_engine_with(MitoConfig::default(), None, Some(listener.clone()))
@@ -757,7 +761,7 @@ async fn test_alter_region_ttl_options() {
 async fn test_write_stall_on_altering() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let listener = Arc::new(NotifyRegionChangeResultListener::default());
    let engine = env
        .create_engine_with(MitoConfig::default(), None, Some(listener.clone()))
--- a/src/mito2/src/engine/append_mode_test.rs
+++ b/src/mito2/src/engine/append_mode_test.rs
@@ -31,7 +31,7 @@ use crate::test_util::{
 async fn test_append_mode_write_query() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -89,7 +89,7 @@ async fn test_append_mode_write_query() {

 #[tokio::test]
 async fn test_append_mode_compaction() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env
        .create_engine(MitoConfig {
            ..Default::default()
--- a/src/mito2/src/engine/basic_test.rs
+++ b/src/mito2/src/engine/basic_test.rs
@@ -42,7 +42,7 @@ use crate::test_util::{

 #[tokio::test]
 async fn test_engine_new_stop() {
-    let mut env = TestEnv::with_prefix("engine-stop");
+    let mut env = TestEnv::with_prefix("engine-stop").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -69,7 +69,7 @@ async fn test_engine_new_stop() {

 #[tokio::test]
 async fn test_write_to_region() {
-    let mut env = TestEnv::with_prefix("write-to-region");
+    let mut env = TestEnv::with_prefix("write-to-region").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -97,7 +97,9 @@ async fn test_region_replay(factory: Option<LogStoreFactory>) {
    let Some(factory) = factory else {
        return;
    };
-    let mut env = TestEnv::with_prefix("region-replay").with_log_store_factory(factory.clone());
+    let mut env = TestEnv::with_prefix("region-replay")
+        .await
+        .with_log_store_factory(factory.clone());
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -173,7 +175,7 @@ async fn test_region_replay(factory: Option<LogStoreFactory>) {

 #[tokio::test]
 async fn test_write_query_region() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -207,7 +209,7 @@ async fn test_write_query_region() {

 #[tokio::test]
 async fn test_different_order() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -268,7 +270,7 @@ async fn test_different_order() {

 #[tokio::test]
 async fn test_different_order_and_type() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -332,7 +334,7 @@ async fn test_different_order_and_type() {
 async fn test_put_delete() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -384,7 +386,7 @@ async fn test_put_delete() {

 #[tokio::test]
 async fn test_delete_not_null_fields() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -433,7 +435,7 @@ async fn test_delete_not_null_fields() {

 #[tokio::test]
 async fn test_put_overwrite() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -493,7 +495,7 @@ async fn test_put_overwrite() {

 #[tokio::test]
 async fn test_absent_and_invalid_columns() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -541,7 +543,7 @@ async fn test_absent_and_invalid_columns() {

 #[tokio::test]
 async fn test_region_usage() {
-    let mut env = TestEnv::with_prefix("region_usage");
+    let mut env = TestEnv::with_prefix("region_usage").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -595,7 +597,7 @@ async fn test_region_usage() {
 async fn test_engine_with_write_cache() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let path = env.data_home().to_str().unwrap().to_string();
    let mito_config = MitoConfig::default().enable_write_cache(path, ReadableSize::mb(512), None);
    let engine = env.create_engine(mito_config).await;
@@ -635,7 +637,7 @@ async fn test_engine_with_write_cache() {

 #[tokio::test]
 async fn test_cache_null_primary_key() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env
        .create_engine(MitoConfig {
            vector_cache_size: ReadableSize::mb(32),
--- a/src/mito2/src/engine/batch_open_test.rs
+++ b/src/mito2/src/engine/batch_open_test.rs
@@ -39,8 +39,9 @@ async fn test_batch_open(factory: Option<LogStoreFactory>) {
    let Some(factory) = factory else {
        return;
    };
-    let mut env =
-        TestEnv::with_prefix("open-batch-regions").with_log_store_factory(factory.clone());
+    let mut env = TestEnv::with_prefix("open-batch-regions")
+        .await
+        .with_log_store_factory(factory.clone());
    let engine = env.create_engine(MitoConfig::default()).await;
    let topic = prepare_test_for_kafka_log_store(&factory).await;

@@ -160,8 +161,9 @@ async fn test_batch_open_err(factory: Option<LogStoreFactory>) {
    let Some(factory) = factory else {
        return;
    };
-    let mut env =
-        TestEnv::with_prefix("open-batch-regions-err").with_log_store_factory(factory.clone());
+    let mut env = TestEnv::with_prefix("open-batch-regions-err")
+        .await
+        .with_log_store_factory(factory.clone());
    let engine = env.create_engine(MitoConfig::default()).await;
    let topic = prepare_test_for_kafka_log_store(&factory).await;
    let mut options = HashMap::new();
--- a/src/mito2/src/engine/catchup_test.rs
+++ b/src/mito2/src/engine/catchup_test.rs
@@ -57,7 +57,9 @@ async fn test_catchup_with_last_entry_id(factory: Option<LogStoreFactory>) {
        return;
    };

-    let mut env = TestEnv::with_prefix("last_entry_id").with_log_store_factory(factory.clone());
+    let mut env = TestEnv::with_prefix("last_entry_id")
+        .await
+        .with_log_store_factory(factory.clone());
    let topic = prepare_test_for_kafka_log_store(&factory).await;
    let leader_engine = env.create_engine(MitoConfig::default()).await;
    let follower_engine = env.create_follower_engine(MitoConfig::default()).await;
@@ -175,8 +177,9 @@ async fn test_catchup_with_incorrect_last_entry_id(factory: Option<LogStoreFacto
        return;
    };

-    let mut env =
-        TestEnv::with_prefix("incorrect_last_entry_id").with_log_store_factory(factory.clone());
+    let mut env = TestEnv::with_prefix("incorrect_last_entry_id")
+        .await
+        .with_log_store_factory(factory.clone());
    let topic = prepare_test_for_kafka_log_store(&factory).await;
    let leader_engine = env.create_engine(MitoConfig::default()).await;
    let follower_engine = env.create_follower_engine(MitoConfig::default()).await;
@@ -277,8 +280,9 @@ async fn test_catchup_without_last_entry_id(factory: Option<LogStoreFactory>) {
        return;
    };

-    let mut env =
-        TestEnv::with_prefix("without_last_entry_id").with_log_store_factory(factory.clone());
+    let mut env = TestEnv::with_prefix("without_last_entry_id")
+        .await
+        .with_log_store_factory(factory.clone());
    let topic = prepare_test_for_kafka_log_store(&factory).await;
    let leader_engine = env.create_engine(MitoConfig::default()).await;
    let follower_engine = env.create_follower_engine(MitoConfig::default()).await;
@@ -380,8 +384,9 @@ async fn test_catchup_with_manifest_update(factory: Option<LogStoreFactory>) {
        return;
    };

-    let mut env =
-        TestEnv::with_prefix("without_manifest_update").with_log_store_factory(factory.clone());
+    let mut env = TestEnv::with_prefix("without_manifest_update")
+        .await
+        .with_log_store_factory(factory.clone());
    let topic = prepare_test_for_kafka_log_store(&factory).await;
    let leader_engine = env.create_engine(MitoConfig::default()).await;
    let follower_engine = env.create_follower_engine(MitoConfig::default()).await;
@@ -545,7 +550,9 @@ async fn test_local_catchup(factory: Option<LogStoreFactory>) {
        return;
    };

-    let mut env = TestEnv::with_prefix("local_catchup").with_log_store_factory(factory.clone());
+    let mut env = TestEnv::with_prefix("local_catchup")
+        .await
+        .with_log_store_factory(factory.clone());
    let leader_engine = env.create_engine(MitoConfig::default()).await;
    let Some(LogStoreImpl::RaftEngine(log_store)) = env.get_log_store() else {
        unreachable!()
@@ -686,7 +693,7 @@ async fn test_local_catchup(factory: Option<LogStoreFactory>) {

 #[tokio::test]
 async fn test_catchup_not_exist() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let non_exist_region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/close_test.rs
+++ b/src/mito2/src/engine/close_test.rs
@@ -21,7 +21,7 @@ use crate::test_util::{CreateRequestBuilder, TestEnv};

 #[tokio::test]
 async fn test_engine_close_region() {
-    let mut env = TestEnv::with_prefix("close");
+    let mut env = TestEnv::with_prefix("close").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/compaction_test.rs
+++ b/src/mito2/src/engine/compaction_test.rs
@@ -136,7 +136,7 @@ async fn collect_stream_ts(stream: SendableRecordBatchStream) -> Vec<i64> {
 #[tokio::test]
 async fn test_compaction_region() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -202,7 +202,7 @@ async fn test_compaction_region() {
 #[tokio::test]
 async fn test_infer_compaction_time_window() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -341,7 +341,7 @@ async fn test_infer_compaction_time_window() {
 #[tokio::test]
 async fn test_compaction_overlapping_files() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -402,7 +402,7 @@ async fn test_compaction_overlapping_files() {
 #[tokio::test]
 async fn test_compaction_region_with_overlapping() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;
    let region_id = RegionId::new(1, 1);

@@ -450,7 +450,7 @@ async fn test_compaction_region_with_overlapping() {
 #[tokio::test]
 async fn test_compaction_region_with_overlapping_delete_all() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -506,7 +506,7 @@ async fn test_compaction_region_with_overlapping_delete_all() {
 #[tokio::test]
 async fn test_readonly_during_compaction() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let listener = Arc::new(CompactionListener::default());
    let engine = env
        .create_engine_with(
@@ -590,7 +590,7 @@ async fn test_readonly_during_compaction() {
 #[tokio::test]
 async fn test_compaction_update_time_window() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -686,7 +686,7 @@ async fn test_compaction_update_time_window() {
 #[tokio::test]
 async fn test_change_region_compaction_window() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -811,7 +811,7 @@ async fn test_change_region_compaction_window() {
 #[tokio::test]
 async fn test_open_overwrite_compaction_window() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/create_test.rs
+++ b/src/mito2/src/engine/create_test.rs
@@ -26,7 +26,7 @@ use crate::test_util::{build_rows, put_rows, rows_schema, CreateRequestBuilder,

 #[tokio::test]
 async fn test_engine_create_new_region() {
-    let mut env = TestEnv::with_prefix("new-region");
+    let mut env = TestEnv::with_prefix("new-region").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -41,7 +41,7 @@ async fn test_engine_create_new_region() {

 #[tokio::test]
 async fn test_engine_create_existing_region() {
-    let mut env = TestEnv::with_prefix("create-existing");
+    let mut env = TestEnv::with_prefix("create-existing").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -61,7 +61,7 @@ async fn test_engine_create_existing_region() {
 #[tokio::test]
 async fn test_engine_create_close_create_region() {
    // This test will trigger create_or_open function.
-    let mut env = TestEnv::with_prefix("create-close-create");
+    let mut env = TestEnv::with_prefix("create-close-create").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -91,7 +91,7 @@ async fn test_engine_create_close_create_region() {

 #[tokio::test]
 async fn test_engine_create_with_different_id() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -110,7 +110,7 @@ async fn test_engine_create_with_different_id() {

 #[tokio::test]
 async fn test_engine_create_with_different_schema() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -130,7 +130,7 @@ async fn test_engine_create_with_different_schema() {

 #[tokio::test]
 async fn test_engine_create_with_different_primary_key() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -150,7 +150,7 @@ async fn test_engine_create_with_different_primary_key() {

 #[tokio::test]
 async fn test_engine_create_with_options() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -172,7 +172,7 @@ async fn test_engine_create_with_options() {

 #[tokio::test]
 async fn test_engine_create_with_custom_store() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env
        .create_engine_with_multiple_object_stores(MitoConfig::default(), None, None, &["Gcs"])
        .await;
@@ -204,7 +204,7 @@ async fn test_engine_create_with_custom_store() {

 #[tokio::test]
 async fn test_engine_create_with_memtable_opts() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/drop_test.rs
+++ b/src/mito2/src/engine/drop_test.rs
@@ -35,7 +35,7 @@ use crate::worker::DROPPING_MARKER_FILE;
 async fn test_engine_drop_region() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::with_prefix("drop");
+    let mut env = TestEnv::with_prefix("drop").await;
    let listener = Arc::new(DropListener::new(Duration::from_millis(100)));
    let engine = env
        .create_engine_with(MitoConfig::default(), None, Some(listener.clone()))
@@ -143,7 +143,7 @@ async fn test_engine_drop_region_for_custom_store() {
        put_rows(engine, region_id, rows).await;
        flush_region(engine, region_id, None).await;
    }
-    let mut env = TestEnv::with_prefix("drop");
+    let mut env = TestEnv::with_prefix("drop").await;
    let listener = Arc::new(DropListener::new(Duration::from_millis(100)));
    let engine = env
        .create_engine_with_multiple_object_stores(
--- a/src/mito2/src/engine/edit_region_test.rs
+++ b/src/mito2/src/engine/edit_region_test.rs
@@ -33,7 +33,7 @@ use crate::test_util::{CreateRequestBuilder, TestEnv};

 #[tokio::test]
 async fn test_edit_region_schedule_compaction() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;

    struct EditRegionListener {
        tx: Mutex<Option<oneshot::Sender<RegionId>>>,
@@ -122,7 +122,7 @@ async fn test_edit_region_schedule_compaction() {

 #[tokio::test]
 async fn test_edit_region_fill_cache() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;

    struct EditRegionListener {
        tx: Mutex<Option<oneshot::Sender<FileId>>>,
@@ -241,7 +241,7 @@ async fn test_edit_region_concurrently() {
        }
    }

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env
        .create_engine(MitoConfig {
            // Suppress the compaction to not impede the speed of this kinda stress testing.
--- a/src/mito2/src/engine/filter_deleted_test.rs
+++ b/src/mito2/src/engine/filter_deleted_test.rs
@@ -28,7 +28,7 @@ use crate::test_util::{
 async fn test_scan_without_filtering_deleted() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/flush_test.rs
+++ b/src/mito2/src/engine/flush_test.rs
@@ -41,7 +41,7 @@ use crate::worker::MAX_INITIAL_CHECK_DELAY_SECS;

 #[tokio::test]
 async fn test_manual_flush() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -91,7 +91,7 @@ async fn test_manual_flush() {

 #[tokio::test]
 async fn test_flush_engine() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let write_buffer_manager = Arc::new(MockWriteBufferManager::default());
    let listener = Arc::new(FlushListener::default());
    let engine = env
@@ -161,7 +161,7 @@ async fn test_flush_engine() {

 #[tokio::test]
 async fn test_write_stall() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let write_buffer_manager = Arc::new(MockWriteBufferManager::default());
    let listener = Arc::new(StallListener::default());
    let engine = env
@@ -236,7 +236,7 @@ async fn test_write_stall() {

 #[tokio::test]
 async fn test_flush_empty() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let write_buffer_manager = Arc::new(MockWriteBufferManager::default());
    let engine = env
        .create_engine_with(
@@ -289,7 +289,7 @@ async fn test_flush_reopen_region(factory: Option<LogStoreFactory>) {
        return;
    };

-    let mut env = TestEnv::new().with_log_store_factory(factory.clone());
+    let mut env = TestEnv::new().await.with_log_store_factory(factory.clone());
    let engine = env.create_engine(MitoConfig::default()).await;
    let region_id = RegionId::new(1, 1);
    env.get_schema_metadata_manager()
@@ -396,7 +396,7 @@ impl MockTimeProvider {

 #[tokio::test]
 async fn test_auto_flush_engine() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let write_buffer_manager = Arc::new(MockWriteBufferManager::default());
    let listener = Arc::new(FlushListener::default());
    let now = current_time_millis();
@@ -467,7 +467,7 @@ async fn test_auto_flush_engine() {

 #[tokio::test]
 async fn test_flush_workers() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let write_buffer_manager = Arc::new(MockWriteBufferManager::default());
    let listener = Arc::new(FlushListener::default());
    let engine = env
@@ -554,7 +554,7 @@ async fn test_update_topic_latest_entry_id(factory: Option<LogStoreFactory>) {
    let write_buffer_manager = Arc::new(MockWriteBufferManager::default());
    let listener = Arc::new(FlushListener::default());

-    let mut env = TestEnv::new().with_log_store_factory(factory.clone());
+    let mut env = TestEnv::new().await.with_log_store_factory(factory.clone());
    let engine = env
        .create_engine_with(
            MitoConfig::default(),
--- a/src/mito2/src/engine/merge_mode_test.rs
+++ b/src/mito2/src/engine/merge_mode_test.rs
@@ -31,7 +31,7 @@ use crate::test_util::{
 async fn test_merge_mode_write_query() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -89,7 +89,7 @@ async fn test_merge_mode_write_query() {
 async fn test_merge_mode_compaction() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env
        .create_engine(MitoConfig {
            ..Default::default()
--- a/src/mito2/src/engine/open_test.rs
+++ b/src/mito2/src/engine/open_test.rs
@@ -36,7 +36,7 @@ use crate::test_util::{

 #[tokio::test]
 async fn test_engine_open_empty() {
-    let mut env = TestEnv::with_prefix("open-empty");
+    let mut env = TestEnv::with_prefix("open-empty").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -63,7 +63,7 @@ async fn test_engine_open_empty() {

 #[tokio::test]
 async fn test_engine_open_existing() {
-    let mut env = TestEnv::with_prefix("open-exiting");
+    let mut env = TestEnv::with_prefix("open-exiting").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -90,7 +90,7 @@ async fn test_engine_open_existing() {

 #[tokio::test]
 async fn test_engine_reopen_region() {
-    let mut env = TestEnv::with_prefix("reopen-region");
+    let mut env = TestEnv::with_prefix("reopen-region").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -107,7 +107,7 @@ async fn test_engine_reopen_region() {

 #[tokio::test]
 async fn test_engine_open_readonly() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -150,7 +150,7 @@ async fn test_engine_open_readonly() {

 #[tokio::test]
 async fn test_engine_region_open_with_options() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -190,7 +190,7 @@ async fn test_engine_region_open_with_options() {

 #[tokio::test]
 async fn test_engine_region_open_with_custom_store() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env
        .create_engine_with_multiple_object_stores(MitoConfig::default(), None, None, &["Gcs"])
        .await;
@@ -244,7 +244,7 @@ async fn test_engine_region_open_with_custom_store() {

 #[tokio::test]
 async fn test_open_region_skip_wal_replay() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -344,7 +344,7 @@ async fn test_open_region_skip_wal_replay() {

 #[tokio::test]
 async fn test_open_region_wait_for_opening_region_ok() {
-    let mut env = TestEnv::with_prefix("wait-for-opening-region-ok");
+    let mut env = TestEnv::with_prefix("wait-for-opening-region-ok").await;
    let engine = env.create_engine(MitoConfig::default()).await;
    let region_id = RegionId::new(1, 1);
    let worker = engine.inner.workers.worker(region_id);
@@ -383,7 +383,7 @@ async fn test_open_region_wait_for_opening_region_ok() {

 #[tokio::test]
 async fn test_open_region_wait_for_opening_region_err() {
-    let mut env = TestEnv::with_prefix("wait-for-opening-region-err");
+    let mut env = TestEnv::with_prefix("wait-for-opening-region-err").await;
    let engine = env.create_engine(MitoConfig::default()).await;
    let region_id = RegionId::new(1, 1);
    let worker = engine.inner.workers.worker(region_id);
@@ -428,7 +428,7 @@ async fn test_open_region_wait_for_opening_region_err() {

 #[tokio::test]
 async fn test_open_compaction_region() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let mut mito_config = MitoConfig::default();
    mito_config
        .sanitize(&env.data_home().display().to_string())
--- a/src/mito2/src/engine/parallel_test.rs
+++ b/src/mito2/src/engine/parallel_test.rs
@@ -73,7 +73,7 @@ async fn scan_in_parallel(

 #[tokio::test]
 async fn test_parallel_scan() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/projection_test.rs
+++ b/src/mito2/src/engine/projection_test.rs
@@ -53,7 +53,7 @@ fn build_rows_multi_tags_fields(

 #[tokio::test]
 async fn test_scan_projection() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/prune_test.rs
+++ b/src/mito2/src/engine/prune_test.rs
@@ -27,7 +27,7 @@ use crate::test_util::{
 };

 async fn check_prune_row_groups(exprs: Vec<Expr>, expected: &str) {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -147,7 +147,7 @@ fn time_range_expr(start_sec: i64, end_sec: i64) -> Expr {

 #[tokio::test]
 async fn test_prune_memtable() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -221,7 +221,7 @@ async fn test_prune_memtable() {

 #[tokio::test]
 async fn test_prune_memtable_complex_expr() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -274,7 +274,7 @@ async fn test_prune_memtable_complex_expr() {

 #[tokio::test]
 async fn test_mem_range_prune() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/row_selector_test.rs
+++ b/src/mito2/src/engine/row_selector_test.rs
@@ -25,7 +25,7 @@ use crate::test_util::{
 };

 async fn test_last_row(append_mode: bool) {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;
    let region_id = RegionId::new(1, 1);

--- a/src/mito2/src/engine/scan_test.rs
+++ b/src/mito2/src/engine/scan_test.rs
@@ -27,7 +27,7 @@ use crate::test_util::{CreateRequestBuilder, TestEnv};

 #[tokio::test]
 async fn test_scan_with_min_sst_sequence() {
-    let mut env = TestEnv::with_prefix("test_scan_with_min_sst_sequence");
+    let mut env = TestEnv::with_prefix("test_scan_with_min_sst_sequence").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
@@ -153,7 +153,7 @@ async fn test_scan_with_min_sst_sequence() {

 #[tokio::test]
 async fn test_series_scan() {
-    let mut env = TestEnv::with_prefix("test_series_scan");
+    let mut env = TestEnv::with_prefix("test_series_scan").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/set_role_state_test.rs
+++ b/src/mito2/src/engine/set_role_state_test.rs
@@ -32,7 +32,7 @@ async fn test_set_role_state_gracefully() {
        SettableRegionRoleState::DowngradingLeader,
    ];
    for settable_role_state in settable_role_states {
-        let mut env = TestEnv::new();
+        let mut env = TestEnv::new().await;
        let engine = env.create_engine(MitoConfig::default()).await;

        let region_id = RegionId::new(1, 1);
@@ -101,7 +101,7 @@ async fn test_set_role_state_gracefully() {

 #[tokio::test]
 async fn test_set_role_state_gracefully_not_exist() {
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let non_exist_region_id = RegionId::new(1, 1);
@@ -116,7 +116,7 @@ async fn test_set_role_state_gracefully_not_exist() {

 #[tokio::test]
 async fn test_write_downgrading_region() {
-    let mut env = TestEnv::with_prefix("write-to-downgrading-region");
+    let mut env = TestEnv::with_prefix("write-to-downgrading-region").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/sync_test.rs
+++ b/src/mito2/src/engine/sync_test.rs
@@ -71,7 +71,7 @@ async fn scan_check(
 #[tokio::test]
 async fn test_sync_after_flush_region() {
    common_telemetry::init_default_ut_logging();
-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;
    let region_id = RegionId::new(1, 1);
    env.get_schema_metadata_manager()
@@ -163,7 +163,7 @@ async fn test_sync_after_flush_region() {
 async fn test_sync_after_alter_region() {
    common_telemetry::init_default_ut_logging();

-    let mut env = TestEnv::new();
+    let mut env = TestEnv::new().await;
    let engine = env.create_engine(MitoConfig::default()).await;

    let region_id = RegionId::new(1, 1);
--- a/src/mito2/src/engine/truncate_test.rs
+++ b/src/mito2/src/engine/truncate_test.rs
@@ -33,7 +33,7 @@ use crate::test_util::{

 #[tokio::test]
 async fn test_engine_truncate_region_basic() {
-    let mut env = TestEnv::with_prefix("truncate-basic");
+    let mut env = TestEnv::with_prefix("truncate-basic").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    // Create the region.
@@ -83,7 +83,7 @@ async fn test_engine_truncate_region_basic() {

 #[tokio::test]
 async fn test_engine_put_data_after_truncate() {
-    let mut env = TestEnv::with_prefix("truncate-put");
+    let mut env = TestEnv::with_prefix("truncate-put").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    // Create the region.
@@ -146,7 +146,7 @@ async fn test_engine_put_data_after_truncate() {

 #[tokio::test]
 async fn test_engine_truncate_after_flush() {
-    let mut env = TestEnv::with_prefix("truncate-flush");
+    let mut env = TestEnv::with_prefix("truncate-flush").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    // Create the region.
@@ -223,7 +223,7 @@ async fn test_engine_truncate_after_flush() {

 #[tokio::test]
 async fn test_engine_truncate_reopen() {
-    let mut env = TestEnv::with_prefix("truncate-reopen");
+    let mut env = TestEnv::with_prefix("truncate-reopen").await;
    let engine = env.create_engine(MitoConfig::default()).await;

    // Create the region.
@@ -282,7 +282,7 @@ async fn test_engine_truncate_reopen() {
 #[tokio::test]
 async fn test_engine_truncate_during_flush() {
    init_default_ut_logging();
-    let mut env = TestEnv::with_prefix("truncate-during-flush");
+    let mut env = TestEnv::with_prefix("truncate-during-flush").await;
    let write_buffer_manager = Arc::new(MockWriteBufferManager::default());
    let listener = Arc::new(FlushTruncateListener::default());
    let engine = env
--- a/src/mito2/src/error.rs
+++ b/src/mito2/src/error.rs
@@ -1020,6 +1020,18 @@ pub enum Error {
        location: Location,
        source: mito_codec::error::Error,
    },
+
+    #[snafu(display(
+        "Inconsistent timestamp column length, expect: {}, actual: {}",
+        expected,
+        actual
+    ))]
+    InconsistentTimestampLength {
+        expected: usize,
+        actual: usize,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }

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

            Encode { source, .. } | Decode { source, .. } => source.status_code(),
+
+            InconsistentTimestampLength { .. } => StatusCode::InvalidArguments,
        }
    }

--- a/src/mito2/src/flush.rs
+++ b/src/mito2/src/flush.rs
@@ -42,7 +42,7 @@ use crate::region::version::{VersionControlData, VersionControlRef};
 use crate::region::{ManifestContextRef, RegionLeaderState};
 use crate::request::{
    BackgroundNotify, FlushFailed, FlushFinished, OptionOutputTx, OutputTx, SenderBulkRequest,
-    SenderDdlRequest, SenderWriteRequest, WorkerRequest,
+    SenderDdlRequest, SenderWriteRequest, WorkerRequest, WorkerRequestWithTime,
 };
 use crate::schedule::scheduler::{Job, SchedulerRef};
 use crate::sst::file::FileMeta;
@@ -223,7 +223,7 @@ pub(crate) struct RegionFlushTask {
    /// Flush result senders.
    pub(crate) senders: Vec<OutputTx>,
    /// Request sender to notify the worker.
-    pub(crate) request_sender: mpsc::Sender<WorkerRequest>,
+    pub(crate) request_sender: mpsc::Sender<WorkerRequestWithTime>,

    pub(crate) access_layer: AccessLayerRef,
    pub(crate) listener: WorkerListener,
@@ -441,7 +441,11 @@ impl RegionFlushTask {

    /// Notify flush job status.
    async fn send_worker_request(&self, request: WorkerRequest) {
-        if let Err(e) = self.request_sender.send(request).await {
+        if let Err(e) = self
+            .request_sender
+            .send(WorkerRequestWithTime::new(request))
+            .await
+        {
            error!(
                "Failed to notify flush job status for region {}, request: {:?}",
                self.region_id, e.0
--- a/src/mito2/src/manifest/manager.rs
+++ b/src/mito2/src/manifest/manager.rs
@@ -594,7 +594,7 @@ mod test {
    #[tokio::test]
    async fn create_manifest_manager() {
        let metadata = Arc::new(basic_region_metadata());
-        let env = TestEnv::new();
+        let env = TestEnv::new().await;
        let manager = env
            .create_manifest_manager(CompressionType::Uncompressed, 10, Some(metadata.clone()))
            .await
@@ -606,7 +606,7 @@ mod test {

    #[tokio::test]
    async fn open_manifest_manager() {
-        let env = TestEnv::new();
+        let env = TestEnv::new().await;
        // Try to opens an empty manifest.
        assert!(env
            .create_manifest_manager(CompressionType::Uncompressed, 10, None)
@@ -637,7 +637,7 @@ mod test {
    #[tokio::test]
    async fn region_change_add_column() {
        let metadata = Arc::new(basic_region_metadata());
-        let env = TestEnv::new();
+        let env = TestEnv::new().await;
        let mut manager = env
            .create_manifest_manager(CompressionType::Uncompressed, 10, Some(metadata.clone()))
            .await
@@ -696,7 +696,7 @@ mod test {
        let metadata = Arc::new(basic_region_metadata());
        let data_home = create_temp_dir("");
        let data_home_path = data_home.path().to_str().unwrap().to_string();
-        let env = TestEnv::with_data_home(data_home);
+        let env = TestEnv::with_data_home(data_home).await;

        let manifest_dir = format!("{}/manifest", data_home_path);

--- a/src/mito2/src/manifest/tests/checkpoint.rs
+++ b/src/mito2/src/manifest/tests/checkpoint.rs
@@ -34,7 +34,7 @@ async fn build_manager(
    compress_type: CompressionType,
 ) -> (TestEnv, RegionManifestManager) {
    let metadata = Arc::new(basic_region_metadata());
-    let env = TestEnv::new();
+    let env = TestEnv::new().await;
    let manager = env
        .create_manifest_manager(compress_type, checkpoint_distance, Some(metadata.clone()))
        .await
--- a/src/mito2/src/memtable/bulk/part.rs
+++ b/src/mito2/src/memtable/bulk/part.rs
@@ -126,7 +126,12 @@ impl From<&BulkPart> for BulkWalEntry {

 impl BulkPart {
    pub(crate) fn estimated_size(&self) -> usize {
-        self.batch.get_array_memory_size()
+        self.batch
+            .columns()
+            .iter()
+            // If can not get slice memory size, assume 0 here.
+            .map(|c| c.to_data().get_slice_memory_size().unwrap_or(0))
+            .sum()
    }

    /// Converts [BulkPart] to [Mutation] for fallback `write_bulk` implementation.
--- a/src/mito2/src/metrics.rs
+++ b/src/mito2/src/metrics.rs
@@ -94,12 +94,7 @@ lazy_static! {


    // ------ Write related metrics
-    /// Number of stalled write requests in each worker.
-    pub static ref WRITE_STALL_TOTAL: IntGaugeVec = register_int_gauge_vec!(
-            "greptime_mito_write_stall_total",
-            "mito stalled write request in each worker",
-            &[WORKER_LABEL]
-        ).unwrap();
+    //
    /// Counter of rejected write requests.
    pub static ref WRITE_REJECT_TOTAL: IntCounter =
        register_int_counter!("greptime_mito_write_reject_total", "mito write reject total").unwrap();
@@ -402,6 +397,7 @@ lazy_static! {

 }

+// Use another block to avoid reaching the recursion limit.
 lazy_static! {
    /// Counter for compaction input file size.
    pub static ref COMPACTION_INPUT_BYTES: Counter = register_counter!(
@@ -426,6 +422,27 @@ lazy_static! {
        "greptime_mito_memtable_field_builder_count",
        "active field builder count in TimeSeriesMemtable",
        ).unwrap();
+
+    /// Number of stalling write requests in each worker.
+    pub static ref WRITE_STALLING: IntGaugeVec = register_int_gauge_vec!(
+            "greptime_mito_write_stalling_count",
+            "mito stalled write request in each worker",
+            &[WORKER_LABEL]
+        ).unwrap();
+    /// Total number of stalled write requests.
+    pub static ref WRITE_STALL_TOTAL: IntCounter = register_int_counter!(
+        "greptime_mito_write_stall_total",
+        "Total number of stalled write requests"
+    ).unwrap();
+    /// Time waiting for requests to be handled by the region worker.
+    pub static ref REQUEST_WAIT_TIME: HistogramVec = register_histogram_vec!(
+            "greptime_mito_request_wait_time",
+            "mito request wait time before being handled by region worker",
+            &[WORKER_LABEL],
+            // 0.001 ~ 10000
+            exponential_buckets(0.001, 10.0, 8).unwrap(),
+        )
+        .unwrap();
 }

 /// Stager notifier to collect metrics.
--- a/src/mito2/src/request.rs
+++ b/src/mito2/src/request.rs
@@ -542,6 +542,22 @@ pub(crate) struct SenderBulkRequest {
    pub(crate) region_metadata: RegionMetadataRef,
 }

+/// Request sent to a worker with timestamp
+#[derive(Debug)]
+pub(crate) struct WorkerRequestWithTime {
+    pub(crate) request: WorkerRequest,
+    pub(crate) created_at: Instant,
+}
+
+impl WorkerRequestWithTime {
+    pub(crate) fn new(request: WorkerRequest) -> Self {
+        Self {
+            request,
+            created_at: Instant::now(),
+        }
+    }
+}
+
 /// Request sent to a worker
 #[derive(Debug)]
 pub(crate) enum WorkerRequest {
--- a/src/mito2/src/schedule/remote_job_scheduler.rs
+++ b/src/mito2/src/schedule/remote_job_scheduler.rs
@@ -30,6 +30,7 @@ use crate::manifest::action::RegionEdit;
 use crate::metrics::{COMPACTION_FAILURE_COUNT, INFLIGHT_COMPACTION_COUNT};
 use crate::request::{
    BackgroundNotify, CompactionFailed, CompactionFinished, OutputTx, WorkerRequest,
+    WorkerRequestWithTime,
 };

 pub type RemoteJobSchedulerRef = Arc<dyn RemoteJobScheduler>;
@@ -130,7 +131,7 @@ pub struct CompactionJobResult {
 /// DefaultNotifier is a default implementation of Notifier that sends WorkerRequest to the mito engine.
 pub(crate) struct DefaultNotifier {
    /// The sender to send WorkerRequest to the mito engine. This is used to notify the mito engine when a remote job is completed.
-    pub(crate) request_sender: Sender<WorkerRequest>,
+    pub(crate) request_sender: Sender<WorkerRequestWithTime>,
 }

 impl DefaultNotifier {
@@ -173,10 +174,10 @@ impl Notifier for DefaultNotifier {

                if let Err(e) = self
                    .request_sender
-                    .send(WorkerRequest::Background {
+                    .send(WorkerRequestWithTime::new(WorkerRequest::Background {
                        region_id: result.region_id,
                        notify,
-                    })
+                    }))
                    .await
                {
                    error!(
--- a/src/mito2/src/sst/index.rs
+++ b/src/mito2/src/sst/index.rs
@@ -45,8 +45,6 @@ pub(crate) const TYPE_INVERTED_INDEX: &str = "inverted_index";
 pub(crate) const TYPE_FULLTEXT_INDEX: &str = "fulltext_index";
 pub(crate) const TYPE_BLOOM_FILTER_INDEX: &str = "bloom_filter_index";

-const DEFAULT_FULLTEXT_BLOOM_ROW_GRANULARITY: usize = 8096;
-
 /// Output of the index creation.
 #[derive(Debug, Clone, Default)]
 pub struct IndexOutput {
@@ -293,7 +291,6 @@ impl IndexerBuilderImpl {
            &self.intermediate_manager,
            &self.metadata,
            self.fulltext_index_config.compress,
-            DEFAULT_FULLTEXT_BLOOM_ROW_GRANULARITY,
            mem_limit,
        )
        .await;
@@ -455,10 +452,11 @@ mod tests {
        if with_skipping_bloom {
            let column_schema =
                ColumnSchema::new("bloom", ConcreteDataType::string_datatype(), false)
-                    .with_skipping_options(SkippingIndexOptions {
-                        granularity: 42,
-                        index_type: SkippingIndexType::BloomFilter,
-                    })
+                    .with_skipping_options(SkippingIndexOptions::new_unchecked(
+                        42,
+                        0.01,
+                        SkippingIndexType::BloomFilter,
+                    ))
                    .unwrap();

            let column = ColumnMetadata {
--- a/src/mito2/src/sst/index/bloom_filter/applier.rs
+++ b/src/mito2/src/sst/index/bloom_filter/applier.rs
@@ -128,6 +128,9 @@ impl BloomFilterIndexApplier {
    /// list of row group ranges that match the predicates.
    ///
    /// The `row_groups` iterator provides the row group lengths and whether to search in the row group.
+    ///
+    /// Row group id existing in the returned result means that the row group is searched.
+    /// Empty ranges means that the row group is searched but no rows are found.
    pub async fn apply(
        &self,
        file_id: FileId,
@@ -195,7 +198,6 @@ impl BloomFilterIndexApplier {
                range.end -= start;
            }
        }
-        output.retain(|(_, ranges)| !ranges.is_empty());

        Ok(output)
    }
@@ -386,6 +388,9 @@ mod tests {
                    .apply(file_id, None, row_groups.into_iter())
                    .await
                    .unwrap()
+                    .into_iter()
+                    .filter(|(_, ranges)| !ranges.is_empty())
+                    .collect()
            })
        }
    }
--- a/src/mito2/src/sst/index/bloom_filter/creator.rs
+++ b/src/mito2/src/sst/index/bloom_filter/creator.rs
@@ -97,6 +97,7 @@ impl BloomFilterIndexer {

            let creator = BloomFilterCreator::new(
                options.granularity as _,
+                options.false_positive_rate(),
                temp_file_provider.clone(),
                global_memory_usage.clone(),
                memory_usage_threshold,
@@ -408,10 +409,11 @@ pub(crate) mod tests {
                    ConcreteDataType::string_datatype(),
                    false,
                )
-                .with_skipping_options(SkippingIndexOptions {
-                    index_type: SkippingIndexType::BloomFilter,
-                    granularity: 2,
-                })
+                .with_skipping_options(SkippingIndexOptions::new_unchecked(
+                    2,
+                    0.01,
+                    SkippingIndexType::BloomFilter,
+                ))
                .unwrap(),
                semantic_type: SemanticType::Tag,
                column_id: 1,
@@ -431,10 +433,11 @@ pub(crate) mod tests {
                    ConcreteDataType::uint64_datatype(),
                    false,
                )
-                .with_skipping_options(SkippingIndexOptions {
-                    index_type: SkippingIndexType::BloomFilter,
-                    granularity: 4,
-                })
+                .with_skipping_options(SkippingIndexOptions::new_unchecked(
+                    4,
+                    0.01,
+                    SkippingIndexType::BloomFilter,
+                ))
                .unwrap(),
                semantic_type: SemanticType::Field,
                column_id: 3,
--- a/src/mito2/src/sst/index/fulltext_index/applier.rs
+++ b/src/mito2/src/sst/index/fulltext_index/applier.rs
@@ -231,6 +231,9 @@ impl FulltextIndexApplier {
 impl FulltextIndexApplier {
    /// Applies coarse-grained fulltext index to the specified SST file.
    /// Returns (row group id -> ranges) that match the queries.
+    ///
+    /// Row group id existing in the returned result means that the row group is searched.
+    /// Empty ranges means that the row group is searched but no rows are found.
    pub async fn apply_coarse(
        &self,
        file_id: FileId,
@@ -367,7 +370,7 @@ impl FulltextIndexApplier {
    /// Adjusts the coarse output. Makes the output ranges based on row group start.
    fn adjust_coarse_output(
        input: Vec<(usize, Range<usize>)>,
-        output: &mut Vec<(usize, Vec<Range<usize>>)>,
+        output: &mut [(usize, Vec<Range<usize>>)],
    ) {
        // adjust ranges to be based on row group
        for ((_, output), (_, input)) in output.iter_mut().zip(input) {
@@ -377,7 +380,6 @@ impl FulltextIndexApplier {
                range.end -= start;
            }
        }
-        output.retain(|(_, ranges)| !ranges.is_empty());
    }

    /// Converts terms to predicates.
@@ -463,7 +465,6 @@ impl IndexSource {
    /// Returns the blob with the specified key from local cache or remote store.
    ///
    /// Returns `None` if the blob is not found.
-    #[allow(unused)]
    async fn blob(
        &self,
        file_id: FileId,
--- a/src/mito2/src/sst/index/fulltext_index/creator.rs
+++ b/src/mito2/src/sst/index/fulltext_index/creator.rs
@@ -60,7 +60,6 @@ impl FulltextIndexer {
        intermediate_manager: &IntermediateManager,
        metadata: &RegionMetadataRef,
        compress: bool,
-        bloom_row_granularity: usize,
        mem_limit: usize,
    ) -> Result<Option<Self>> {
        let mut creators = HashMap::new();
@@ -106,7 +105,8 @@ impl FulltextIndexer {
                    let global_memory_usage = Arc::new(AtomicUsize::new(0));
                    let creator = BloomFilterFulltextIndexCreator::new(
                        config,
-                        bloom_row_granularity,
+                        options.granularity as _,
+                        options.false_positive_rate(),
                        temp_file_provider,
                        global_memory_usage,
                        Some(mem_limit),
@@ -400,12 +400,14 @@ mod tests {
                    ConcreteDataType::string_datatype(),
                    true,
                )
-                .with_fulltext_options(FulltextOptions {
-                    enable: true,
-                    analyzer: FulltextAnalyzer::English,
-                    case_sensitive: true,
-                    backend: backend.clone(),
-                })
+                .with_fulltext_options(FulltextOptions::new_unchecked(
+                    true,
+                    FulltextAnalyzer::English,
+                    true,
+                    backend.clone(),
+                    1,
+                    0.01,
+                ))
                .unwrap(),
                semantic_type: SemanticType::Field,
                column_id: 1,
@@ -416,12 +418,14 @@ mod tests {
                    ConcreteDataType::string_datatype(),
                    true,
                )
-                .with_fulltext_options(FulltextOptions {
-                    enable: true,
-                    analyzer: FulltextAnalyzer::English,
-                    case_sensitive: false,
-                    backend: backend.clone(),
-                })
+                .with_fulltext_options(FulltextOptions::new_unchecked(
+                    true,
+                    FulltextAnalyzer::English,
+                    false,
+                    backend.clone(),
+                    1,
+                    0.01,
+                ))
                .unwrap(),
                semantic_type: SemanticType::Field,
                column_id: 2,
@@ -432,12 +436,14 @@ mod tests {
                    ConcreteDataType::string_datatype(),
                    true,
                )
-                .with_fulltext_options(FulltextOptions {
-                    enable: true,
-                    analyzer: FulltextAnalyzer::Chinese,
-                    case_sensitive: false,
-                    backend: backend.clone(),
-                })
+                .with_fulltext_options(FulltextOptions::new_unchecked(
+                    true,
+                    FulltextAnalyzer::Chinese,
+                    false,
+                    backend.clone(),
+                    1,
+                    0.01,
+                ))
                .unwrap(),
                semantic_type: SemanticType::Field,
                column_id: 3,
@@ -547,7 +553,6 @@ mod tests {
            &intm_mgr,
            &region_metadata,
            true,
-            1,
            1024,
        )
        .await
@@ -625,6 +630,7 @@ mod tests {
                            .unwrap();
                        resp.map(|r| {
                            r.into_iter()
+                                .filter(|(_, ranges)| !ranges.is_empty())
                                .map(|(row_group_id, _)| row_group_id as RowId)
                                .collect()
                        })
--- a/src/mito2/src/sst/index/indexer/finish.rs
+++ b/src/mito2/src/sst/index/indexer/finish.rs
@@ -14,10 +14,8 @@

 use common_telemetry::{debug, warn};
 use puffin::puffin_manager::{PuffinManager, PuffinWriter};
+use store_api::storage::ColumnId;

-use crate::sst::index::bloom_filter::creator::BloomFilterIndexer;
-use crate::sst::index::fulltext_index::creator::FulltextIndexer;
-use crate::sst::index::inverted_index::creator::InvertedIndexer;
 use crate::sst::index::puffin_manager::SstPuffinWriter;
 use crate::sst::index::statistics::{ByteCount, RowCount};
 use crate::sst::index::{
@@ -113,13 +111,14 @@ impl Indexer {
            return true;
        };

+        let column_ids = indexer.column_ids().collect();
        let err = match indexer.finish(puffin_writer).await {
            Ok((row_count, byte_count)) => {
                self.fill_inverted_index_output(
                    &mut index_output.inverted_index,
                    row_count,
                    byte_count,
-                    &indexer,
+                    column_ids,
                );
                return true;
            }
@@ -150,13 +149,14 @@ impl Indexer {
            return true;
        };

+        let column_ids = indexer.column_ids().collect();
        let err = match indexer.finish(puffin_writer).await {
            Ok((row_count, byte_count)) => {
                self.fill_fulltext_index_output(
                    &mut index_output.fulltext_index,
                    row_count,
                    byte_count,
-                    &indexer,
+                    column_ids,
                );
                return true;
            }
@@ -187,13 +187,14 @@ impl Indexer {
            return true;
        };

+        let column_ids = indexer.column_ids().collect();
        let err = match indexer.finish(puffin_writer).await {
            Ok((row_count, byte_count)) => {
                self.fill_bloom_filter_output(
                    &mut index_output.bloom_filter,
                    row_count,
                    byte_count,
-                    &indexer,
+                    column_ids,
                );
                return true;
            }
@@ -220,16 +221,16 @@ impl Indexer {
        output: &mut InvertedIndexOutput,
        row_count: RowCount,
        byte_count: ByteCount,
-        indexer: &InvertedIndexer,
+        column_ids: Vec<ColumnId>,
    ) {
        debug!(
-            "Inverted index created, region_id: {}, file_id: {}, written_bytes: {}, written_rows: {}",
-            self.region_id, self.file_id, byte_count, row_count
+            "Inverted index created, region_id: {}, file_id: {}, written_bytes: {}, written_rows: {}, columns: {:?}",
+            self.region_id, self.file_id, byte_count, row_count, column_ids
        );

        output.index_size = byte_count;
        output.row_count = row_count;
-        output.columns = indexer.column_ids().collect();
+        output.columns = column_ids;
    }

    fn fill_fulltext_index_output(
@@ -237,16 +238,16 @@ impl Indexer {
        output: &mut FulltextIndexOutput,
        row_count: RowCount,
        byte_count: ByteCount,
-        indexer: &FulltextIndexer,
+        column_ids: Vec<ColumnId>,
    ) {
        debug!(
-            "Full-text index created, region_id: {}, file_id: {}, written_bytes: {}, written_rows: {}",
-            self.region_id, self.file_id, byte_count, row_count
+            "Full-text index created, region_id: {}, file_id: {}, written_bytes: {}, written_rows: {}, columns: {:?}",
+            self.region_id, self.file_id, byte_count, row_count, column_ids
        );

        output.index_size = byte_count;
        output.row_count = row_count;
-        output.columns = indexer.column_ids().collect();
+        output.columns = column_ids;
    }

    fn fill_bloom_filter_output(
@@ -254,15 +255,15 @@ impl Indexer {
        output: &mut BloomFilterOutput,
        row_count: RowCount,
        byte_count: ByteCount,
-        indexer: &BloomFilterIndexer,
+        column_ids: Vec<ColumnId>,
    ) {
        debug!(
-            "Bloom filter created, region_id: {}, file_id: {}, written_bytes: {}, written_rows: {}",
-            self.region_id, self.file_id, byte_count, row_count
+            "Bloom filter created, region_id: {}, file_id: {}, written_bytes: {}, written_rows: {}, columns: {:?}",
+            self.region_id, self.file_id, byte_count, row_count, column_ids
        );

        output.index_size = byte_count;
        output.row_count = row_count;
-        output.columns = indexer.column_ids().collect();
+        output.columns = column_ids;
    }
 }
--- a/Show More
+++ b/Show More