fix(otlp): replace otlp trace attr type from string to jsonb (#4918 )

* chore: minor update * chore: replace otlp trace attr type from string to jsonb * chore: add new util file and remove useless code * chore: add license header * chore: remove unused error * chore: adjust otlp traces column order * chore: update test * chore: minor fix --------- Co-authored-by: shuiyisong <xixing.sys@gmail.com>
fix: json_path_exists null results (#4881 )
2026-01-04 20:32:56 +00:00 · 2024-11-08 06:34:49 +00:00 · 2024-11-08 03:01:45 +00:00 · 2024-11-08 02:49:36 +00:00 · 2024-11-07 13:25:05 +00:00 · 2024-11-07 03:01:12 +00:00
358 changed files with 11267 additions and 2016 deletions
--- a/.github/actions/build-windows-artifacts/action.yml
+++ b/.github/actions/build-windows-artifacts/action.yml
@@ -40,7 +40,7 @@ runs:

    - name: Install PyArrow Package
      shell: pwsh
-      run: pip install pyarrow
+      run: pip install pyarrow numpy

    - name: Install WSL distribution
      uses: Vampire/setup-wsl@v2
--- a/.github/actions/setup-etcd-cluster/action.yml
+++ b/.github/actions/setup-etcd-cluster/action.yml
@@ -18,7 +18,7 @@ runs:
        --set replicaCount=${{ inputs.etcd-replicas }} \
        --set resources.requests.cpu=50m \
        --set resources.requests.memory=128Mi \
-        --set resources.limits.cpu=1000m \
+        --set resources.limits.cpu=1500m \
        --set resources.limits.memory=2Gi \
        --set auth.rbac.create=false \
        --set auth.rbac.token.enabled=false \
--- a/.github/workflows/develop.yml
+++ b/.github/workflows/develop.yml
@@ -436,7 +436,7 @@ jobs:
    timeout-minutes: 60
    strategy:
      matrix:
-        target: ["fuzz_migrate_mito_regions", "fuzz_failover_mito_regions", "fuzz_failover_metric_regions"]
+        target: ["fuzz_migrate_mito_regions", "fuzz_migrate_metric_regions", "fuzz_failover_mito_regions", "fuzz_failover_metric_regions"]
        mode:
          - name: "Remote WAL"
            minio: true
@@ -449,6 +449,12 @@ jobs:
              minio: true
              kafka: false
              values: "with-minio.yaml"
+          - target: "fuzz_migrate_metric_regions"
+            mode:
+              name: "Local WAL"
+              minio: true
+              kafka: false
+              values: "with-minio.yaml"
    steps:
      - name: Remove unused software
        run: |
@@ -688,7 +694,7 @@ jobs:
        with:
          python-version: '3.10'
      - name: Install PyArrow Package
-        run: pip install pyarrow
+        run: pip install pyarrow numpy
      - name: Setup etcd server
        working-directory: tests-integration/fixtures/etcd
        run: docker compose -f docker-compose-standalone.yml up -d --wait
--- a/.github/workflows/nightly-ci.yml
+++ b/.github/workflows/nightly-ci.yml
@@ -92,7 +92,7 @@ jobs:
        with:
          python-version: "3.10"
      - name: Install PyArrow Package
-        run: pip install pyarrow
+        run: pip install pyarrow numpy
      - name: Install WSL distribution
        uses: Vampire/setup-wsl@v2
        with:
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1,6 +1,6 @@
 # This file is automatically @generated by Cargo.
 # It is not intended for manual editing.
-version = 3
+version = 4

 [[package]]
 name = "Inflector"
@@ -200,12 +200,6 @@ version = "1.0.89"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "86fdf8605db99b54d3cd748a44c6d04df638eb5dafb219b135d0149bd0db01f6"

-[[package]]
-name = "anymap"
-version = "1.0.0-beta.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f1f8f5a6f3d50d89e3797d7593a50f96bb2aaa20ca0cc7be1fb673232c91d72"
-
 [[package]]
 name = "anymap2"
 version = "0.13.0"
@@ -1794,7 +1788,6 @@ dependencies = [
 "tokio-stream",
 "tonic 0.11.0",
 "tracing",
- "tracing-subscriber",
 ]

 [[package]]
@@ -1808,6 +1801,17 @@ dependencies = [
 "winapi",
 ]

+[[package]]
+name = "clocksource"
+version = "0.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "129026dd5a8a9592d96916258f3a5379589e513ea5e86aeb0bd2530286e44e9e"
+dependencies = [
+ "libc",
+ "time",
+ "winapi",
+]
+
 [[package]]
 name = "cmake"
 version = "0.1.51"
@@ -2066,6 +2070,8 @@ dependencies = [
 "datafusion",
 "datatypes",
 "derive_more",
+ "geo",
+ "geo-types",
 "geohash",
 "h3o",
 "jsonb",
@@ -2084,6 +2090,7 @@ dependencies = [
 "store-api",
 "table",
 "tokio",
+ "wkt",
 ]

 [[package]]
@@ -2144,6 +2151,7 @@ dependencies = [
 "paste",
 "prost 0.12.6",
 "snafu 0.8.5",
+ "store-api",
 "table",
 ]

@@ -2332,16 +2340,24 @@ name = "common-runtime"
 version = "0.9.5"
 dependencies = [
 "async-trait",
+ "clap 4.5.19",
 "common-error",
 "common-macro",
 "common-telemetry",
+ "futures",
 "lazy_static",
 "num_cpus",
 "once_cell",
+ "parking_lot 0.12.3",
 "paste",
+ "pin-project",
 "prometheus",
+ "rand",
+ "ratelimit",
 "serde",
+ "serde_json",
 "snafu 0.8.5",
+ "tempfile",
 "tokio",
 "tokio-metrics",
 "tokio-metrics-collector",
@@ -3693,6 +3709,16 @@ version = "1.0.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d6ef0072f8a535281e4876be788938b528e9a1d43900b82c2569af7da799125"

+[[package]]
+name = "earcutr"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "79127ed59a85d7687c409e9978547cffb7dc79675355ed22da6b66fd5f6ead01"
+dependencies = [
+ "itertools 0.11.0",
+ "num-traits",
+]
+
 [[package]]
 name = "either"
 version = "1.13.0"
@@ -4001,6 +4027,12 @@ dependencies = [
 "libc",
 ]

+[[package]]
+name = "float_next_after"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8"
+
 [[package]]
 name = "flow"
 version = "0.9.5"
@@ -4425,6 +4457,24 @@ dependencies = [
 "version_check",
 ]

+[[package]]
+name = "geo"
+version = "0.29.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "81d088357a9cc60cec8253b3578f6834b4a3aa20edb55f5d1c030c36d8143f11"
+dependencies = [
+ "earcutr",
+ "float_next_after",
+ "geo-types",
+ "geographiclib-rs",
+ "i_overlay",
+ "log",
+ "num-traits",
+ "robust",
+ "rstar",
+ "spade",
+]
+
 [[package]]
 name = "geo-types"
 version = "0.7.13"
@@ -4433,9 +4483,19 @@ checksum = "9ff16065e5720f376fbced200a5ae0f47ace85fd70b7e54269790281353b6d61"
 dependencies = [
 "approx 0.5.1",
 "num-traits",
+ "rstar",
 "serde",
 ]

+[[package]]
+name = "geographiclib-rs"
+version = "0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e6e5ed84f8089c70234b0a8e0aedb6dc733671612ddc0d37c6066052f9781960"
+dependencies = [
+ "libm",
+]
+
 [[package]]
 name = "geohash"
 version = "0.13.1"
@@ -4518,7 +4578,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=b4d301184eb0d01fd4d1042fcc7c5dfb54f3c1e3#b4d301184eb0d01fd4d1042fcc7c5dfb54f3c1e3"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=255f87a3318ace3f88a67f76995a0e14910983f4#255f87a3318ace3f88a67f76995a0e14910983f4"
 dependencies = [
 "prost 0.12.6",
 "serde",
@@ -4584,6 +4644,15 @@ dependencies = [
 "num-traits",
 ]

+[[package]]
+name = "hash32"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47d60b12902ba28e2730cd37e95b8c9223af2808df9e902d4df49588d1470606"
+dependencies = [
+ "byteorder",
+]
+
 [[package]]
 name = "hashbrown"
 version = "0.12.3"
@@ -4679,6 +4748,16 @@ dependencies = [
 "http 1.1.0",
 ]

+[[package]]
+name = "heapless"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0bfb9eb618601c89945a70e254898da93b13be0388091d42117462b265bb3fad"
+dependencies = [
+ "hash32",
+ "stable_deref_trait",
+]
+
 [[package]]
 name = "heck"
 version = "0.4.1"
@@ -5104,6 +5183,50 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "i_float"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f5fe043aae28ce70bd2f78b2f5f82a3654d63607c82594da4dabb8b6cb81f2b2"
+dependencies = [
+ "serde",
+]
+
+[[package]]
+name = "i_key_sort"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "347c253b4748a1a28baf94c9ce133b6b166f08573157e05afe718812bc599fcd"
+
+[[package]]
+name = "i_overlay"
+version = "1.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a469f68cb8a7cef375b2b0f581faf5859b4b50600438c00d46b71acc25ebbd0c"
+dependencies = [
+ "i_float",
+ "i_key_sort",
+ "i_shape",
+ "i_tree",
+ "rayon",
+]
+
+[[package]]
+name = "i_shape"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1b44852d57a991c7dedaf76c55bc44f677f547ff899a430d29e13efd6133d7d8"
+dependencies = [
+ "i_float",
+ "serde",
+]
+
+[[package]]
+name = "i_tree"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "155181bc97d770181cf9477da51218a19ee92a8e5be642e796661aee2b601139"
+
 [[package]]
 name = "iana-time-zone"
 version = "0.1.61"
@@ -5250,7 +5373,7 @@ dependencies = [
 [[package]]
 name = "influxdb_line_protocol"
 version = "0.1.0"
-source = "git+https://github.com/evenyag/influxdb_iox?branch=feat/line-protocol#10ef0d0b02705ac7518717390939fa3a9bcfcacc"
+source = "git+https://github.com/evenyag/influxdb_iox?branch=feat%2Fline-protocol#10ef0d0b02705ac7518717390939fa3a9bcfcacc"
 dependencies = [
 "bytes",
 "nom",
@@ -5510,8 +5633,8 @@ dependencies = [

 [[package]]
 name = "jsonb"
-version = "0.4.1"
-source = "git+https://github.com/datafuselabs/jsonb.git?rev=46ad50fc71cf75afbf98eec455f7892a6387c1fc#46ad50fc71cf75afbf98eec455f7892a6387c1fc"
+version = "0.4.3"
+source = "git+https://github.com/CookiePieWw/jsonb.git?rev=ed2d4f8575419ed434a4ae09dee18ca900915d9c#ed2d4f8575419ed434a4ae09dee18ca900915d9c"
 dependencies = [
 "byteorder",
 "fast-float",
@@ -5538,6 +5661,19 @@ dependencies = [
 "thiserror",
 ]

+[[package]]
+name = "jsonpath-rust"
+version = "0.7.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69a61b87f6a55cc6c28fed5739dd36b9642321ce63e4a5e4a4715d69106f4a10"
+dependencies = [
+ "pest",
+ "pest_derive",
+ "regex",
+ "serde_json",
+ "thiserror",
+]
+
 [[package]]
 name = "jsonptr"
 version = "0.4.7"
@@ -5648,7 +5784,7 @@ dependencies = [
 "hyper-rustls",
 "hyper-timeout 0.5.1",
 "hyper-util",
- "jsonpath-rust",
+ "jsonpath-rust 0.5.1",
 "k8s-openapi",
 "kube-core",
 "pem 3.0.4",
@@ -6332,6 +6468,7 @@ dependencies = [
 "common-telemetry",
 "datatypes",
 "futures",
+ "futures-util",
 "humantime-serde",
 "meta-srv",
 "rand",
@@ -6408,9 +6545,9 @@ dependencies = [
 [[package]]
 name = "meter-core"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-meter.git?rev=80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd#80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd"
+source = "git+https://github.com/GreptimeTeam/greptime-meter.git?rev=a10facb353b41460eeb98578868ebf19c2084fac#a10facb353b41460eeb98578868ebf19c2084fac"
 dependencies = [
- "anymap",
+ "anymap2",
 "once_cell",
 "parking_lot 0.12.3",
 ]
@@ -6418,7 +6555,7 @@ dependencies = [
 [[package]]
 name = "meter-macros"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-meter.git?rev=80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd#80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd"
+source = "git+https://github.com/GreptimeTeam/greptime-meter.git?rev=a10facb353b41460eeb98578868ebf19c2084fac#a10facb353b41460eeb98578868ebf19c2084fac"
 dependencies = [
 "meter-core",
 ]
@@ -6543,6 +6680,7 @@ dependencies = [
 "common-error",
 "common-function",
 "common-macro",
+ "common-meta",
 "common-procedure-test",
 "common-query",
 "common-recordbatch",
@@ -7583,6 +7721,7 @@ dependencies = [
 "file-engine",
 "futures",
 "futures-util",
+ "jsonb",
 "lazy_static",
 "meta-client",
 "meter-core",
@@ -8185,6 +8324,7 @@ dependencies = [
 "greptime-proto",
 "itertools 0.10.5",
 "jsonb",
+ "jsonpath-rust 0.7.3",
 "lazy_static",
 "moka",
 "once_cell",
@@ -8320,6 +8460,7 @@ dependencies = [
 "datanode",
 "frontend",
 "meta-srv",
+ "serde",
 "snafu 0.8.5",
 ]

@@ -8382,8 +8523,7 @@ checksum = "439ee305def115ba05938db6eb1644ff94165c5ab5e9420d1c1bcedbba909391"
 [[package]]
 name = "pprof"
 version = "0.13.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ef5c97c51bd34c7e742402e216abdeb44d415fbe6ae41d56b114723e953711cb"
+source = "git+https://github.com/GreptimeTeam/pprof-rs?rev=1bd1e21#1bd1e210d8626da3d1e5aff976e6feee994f576d"
 dependencies = [
 "backtrace",
 "cfg-if",
@@ -8613,9 +8753,9 @@ dependencies = [

 [[package]]
 name = "promql-parser"
-version = "0.4.1"
+version = "0.4.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c1ad4a4cfa84ec4aa5831c82e57af0a3faf3f0af83bee13fa1390b2d0a32dc9"
+checksum = "7fe99e6f80a79abccf1e8fb48dd63473a36057e600cc6ea36147c8318698ae6f"
 dependencies = [
 "cfgrammar",
 "chrono",
@@ -8623,6 +8763,8 @@ dependencies = [
 "lrlex",
 "lrpar",
 "regex",
+ "serde",
+ "serde_json",
 ]

 [[package]]
@@ -9016,6 +9158,7 @@ dependencies = [
 "table",
 "tokio",
 "tokio-stream",
+ "uuid",
 ]

 [[package]]
@@ -9195,6 +9338,17 @@ dependencies = [
 "rand",
 ]

+[[package]]
+name = "ratelimit"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6c1bb13e2dcfa2232ac6887157aad8d9b3fe4ca57f7c8d4938ff5ea9be742300"
+dependencies = [
+ "clocksource",
+ "parking_lot 0.12.3",
+ "thiserror",
+]
+
 [[package]]
 name = "raw-cpuid"
 version = "11.2.0"
@@ -9540,6 +9694,12 @@ dependencies = [
 "syn 1.0.109",
 ]

+[[package]]
+name = "robust"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cbf4a6aa5f6d6888f39e980649f3ad6b666acdce1d78e95b8a2cb076e687ae30"
+
 [[package]]
 name = "ron"
 version = "0.7.1"
@@ -9634,6 +9794,17 @@ dependencies = [
 "zstd 0.13.2",
 ]

+[[package]]
+name = "rstar"
+version = "0.12.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "133315eb94c7b1e8d0cb097e5a710d850263372fd028fff18969de708afc7008"
+dependencies = [
+ "heapless",
+ "num-traits",
+ "smallvec",
+]
+
 [[package]]
 name = "rstest"
 version = "0.21.0"
@@ -10912,9 +11083,9 @@ dependencies = [

 [[package]]
 name = "shadow-rs"
-version = "0.31.1"
+version = "0.35.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "02c282402d25101f9c893e9cd7e4cae535fe7db18b81291de973026c219ddf1e"
+checksum = "2311e39772c00391875f40e34d43efef247b23930143a70ca5fbec9505937420"
 dependencies = [
 "const_format",
 "git2",
@@ -11126,6 +11297,18 @@ dependencies = [
 "windows-sys 0.52.0",
 ]

+[[package]]
+name = "spade"
+version = "2.12.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "93f5ef1f863aca7d1d7dda7ccfc36a0a4279bd6d3c375176e5e0712e25cb4889"
+dependencies = [
+ "hashbrown 0.14.5",
+ "num-traits",
+ "robust",
+ "smallvec",
+]
+
 [[package]]
 name = "sparsevec"
 version = "0.2.0"
@@ -11471,6 +11654,7 @@ dependencies = [
 "datatypes",
 "derive_builder 0.12.0",
 "futures",
+ "humantime",
 "serde",
 "serde_json",
 "snafu 0.8.5",
@@ -14118,6 +14302,18 @@ dependencies = [
 "winapi",
 ]

+[[package]]
+name = "wkt"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "54f7f1ff4ea4c18936d6cd26a6fd24f0003af37e951a8e0e8b9e9a2d0bd0a46d"
+dependencies = [
+ "geo-types",
+ "log",
+ "num-traits",
+ "thiserror",
+]
+
 [[package]]
 name = "wyz"
 version = "0.5.1"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -121,13 +121,13 @@ etcd-client = { version = "0.13" }
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "b4d301184eb0d01fd4d1042fcc7c5dfb54f3c1e3" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "255f87a3318ace3f88a67f76995a0e14910983f4" }
 humantime = "2.1"
 humantime-serde = "1.1"
 itertools = "0.10"
-jsonb = { git = "https://github.com/datafuselabs/jsonb.git", rev = "46ad50fc71cf75afbf98eec455f7892a6387c1fc", default-features = false }
+jsonb = { git = "https://github.com/CookiePieWw/jsonb.git", rev = "ed2d4f8575419ed434a4ae09dee18ca900915d9c", default-features = false }
 lazy_static = "1.4"
-meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd" }
+meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "a10facb353b41460eeb98578868ebf19c2084fac" }
 mockall = "0.11.4"
 moka = "0.12"
 notify = "6.1"
@@ -140,14 +140,16 @@ opentelemetry-proto = { version = "0.5", features = [
    "with-serde",
    "logs",
 ] }
+parking_lot = "0.12"
 parquet = { version = "51.0.0", default-features = false, features = ["arrow", "async", "object_store"] }
 paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
-promql-parser = { version = "0.4.1" }
+promql-parser = { version = "0.4.3", features = ["ser"] }
 prost = "0.12"
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.8"
+ratelimit = "0.9"
 regex = "1.8"
 regex-automata = { version = "0.4" }
 reqwest = { version = "0.12", default-features = false, features = [
@@ -167,7 +169,7 @@ schemars = "0.8"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["float_roundtrip"] }
 serde_with = "3"
-shadow-rs = "0.31"
+shadow-rs = "0.35"
 similar-asserts = "1.6.0"
 smallvec = { version = "1", features = ["serde"] }
 snafu = "0.8"
@@ -178,13 +180,16 @@ sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "5
 ] }
 strum = { version = "0.25", features = ["derive"] }
 tempfile = "3"
-tokio = { version = "1.36", features = ["full"] }
+tokio = { version = "1.40", features = ["full"] }
 tokio-postgres = "0.7"
 tokio-stream = { version = "0.1" }
 tokio-util = { version = "0.7", features = ["io-util", "compat"] }
 toml = "0.8.8"
 tonic = { version = "0.11", features = ["tls", "gzip", "zstd"] }
 tower = { version = "0.4" }
+tracing-appender = "0.2"
+tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "fmt"] }
+typetag = "0.2"
 uuid = { version = "1.7", features = ["serde", "v4", "fast-rng"] }
 zstd = "0.13"

@@ -256,10 +261,12 @@ tokio-rustls = { git = "https://github.com/GreptimeTeam/tokio-rustls" }
 # This is commented, since we are not using aws-lc-sys, if we need to use it, we need to uncomment this line or use a release after this commit, or it wouldn't compile with gcc < 8.1
 # see https://github.com/aws/aws-lc-rs/pull/526
 # aws-lc-sys = { git ="https://github.com/aws/aws-lc-rs", rev = "556558441e3494af4b156ae95ebc07ebc2fd38aa" }
+# Apply a fix for pprof for unaligned pointer access
+pprof = { git = "https://github.com/GreptimeTeam/pprof-rs", rev = "1bd1e21" }

 [workspace.dependencies.meter-macros]
 git = "https://github.com/GreptimeTeam/greptime-meter.git"
-rev = "80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd"
+rev = "a10facb353b41460eeb98578868ebf19c2084fac"

 [profile.release]
 debug = 1
--- a/2
+++ b/2
@@ -8,7 +8,7 @@ CARGO_BUILD_OPTS := --locked
 IMAGE_REGISTRY ?= docker.io
 IMAGE_NAMESPACE ?= greptime
 IMAGE_TAG ?= latest
-DEV_BUILDER_IMAGE_TAG ?= 2024-06-06-5674c14f-20240920110415
+DEV_BUILDER_IMAGE_TAG ?= 2024-10-19-a5c00e85-20241024184445
 BUILDX_MULTI_PLATFORM_BUILD ?= false
 BUILDX_BUILDER_NAME ?= gtbuilder
 BASE_IMAGE ?= ubuntu
--- a/README.md
+++ b/README.md
@@ -6,7 +6,7 @@
  </picture>
 </p>

-<h2 align="center">Unified Time Series Database for Metrics, Logs, and Events</h2>
+<h2 align="center">Unified & Cost-Effective Time Series Database for Metrics, Logs, and Events</h2>

 <div align="center">
 <h3 align="center">
@@ -48,9 +48,21 @@
 </a>
 </div>

+- [Introduction](#introduction)
+- [**Features: Why GreptimeDB**](#why-greptimedb)
+- [Architecture](https://docs.greptime.com/contributor-guide/overview/#architecture)
+- [Try it for free](#try-greptimedb)
+- [Getting Started](#getting-started)
+- [Project Status](#project-status)
+- [Join the community](#community)
+  - [Contributing](#contributing)
+- [Extension](#extension )
+- [License](#license)
+- [Acknowledgement](#acknowledgement)
+
 ## Introduction

-**GreptimeDB** is an open-source unified time-series database for **Metrics**, **Logs**, and **Events** (also **Traces** in plan). You can gain real-time insights from Edge to Cloud at any scale.
+**GreptimeDB** is an open-source unified & cost-effective time-series database for **Metrics**, **Logs**, and **Events** (also **Traces** in plan). You can gain real-time insights from Edge to Cloud at Any Scale.

 ## Why GreptimeDB

--- a/config/config.md
+++ b/config/config.md
@@ -83,7 +83,7 @@
 | `wal.backoff_max` | String | `10s` | The maximum backoff delay.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.backoff_base` | Integer | `2` | The exponential backoff rate, i.e. next backoff = base * current backoff.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.backoff_deadline` | String | `5mins` | The deadline of retries.<br/>**It's only used when the provider is `kafka`**. |
-| `wal.overwrite_entry_start_id` | Bool | `false` | Ignore missing entries during read WAL.<br/>**It's only used when the provider is `kafka`**.<br/><br/>This option ensures that when Kafka messages are deleted, the system <br/>can still successfully replay memtable data without throwing an <br/>out-of-range error. <br/>However, enabling this option might lead to unexpected data loss, <br/>as the system will skip over missing entries instead of treating <br/>them as critical errors. |
+| `wal.overwrite_entry_start_id` | Bool | `false` | Ignore missing entries during read WAL.<br/>**It's only used when the provider is `kafka`**.<br/><br/>This option ensures that when Kafka messages are deleted, the system<br/>can still successfully replay memtable data without throwing an<br/>out-of-range error.<br/>However, enabling this option might lead to unexpected data loss,<br/>as the system will skip over missing entries instead of treating<br/>them as critical errors. |
 | `metadata_store` | -- | -- | Metadata storage options. |
 | `metadata_store.file_size` | String | `256MB` | Kv file size in bytes. |
 | `metadata_store.purge_threshold` | String | `4GB` | Kv purge threshold. |
@@ -93,8 +93,8 @@
 | `storage` | -- | -- | The data storage options. |
 | `storage.data_home` | String | `/tmp/greptimedb/` | The working home directory. |
 | `storage.type` | String | `File` | The storage type used to store the data.<br/>- `File`: the data is stored in the local file system.<br/>- `S3`: the data is stored in the S3 object storage.<br/>- `Gcs`: the data is stored in the Google Cloud Storage.<br/>- `Azblob`: the data is stored in the Azure Blob Storage.<br/>- `Oss`: the data is stored in the Aliyun OSS. |
-| `storage.cache_path` | String | Unset | Cache configuration for object storage such as 'S3' etc.<br/>The local file cache directory. |
-| `storage.cache_capacity` | String | Unset | The local file cache capacity in bytes. |
+| `storage.cache_path` | String | Unset | Cache configuration for object storage such as 'S3' etc. It is recommended to configure it when using object storage for better performance.<br/>The local file cache directory. |
+| `storage.cache_capacity` | String | Unset | The local file cache capacity in bytes. If your disk space is sufficient, it is recommended to set it larger. |
 | `storage.bucket` | String | Unset | The S3 bucket name.<br/>**It's only used when the storage type is `S3`, `Oss` and `Gcs`**. |
 | `storage.root` | String | Unset | The S3 data will be stored in the specified prefix, for example, `s3://${bucket}/${root}`.<br/>**It's only used when the storage type is `S3`, `Oss` and `Azblob`**. |
 | `storage.access_key_id` | String | Unset | The access key id of the aws account.<br/>It's **highly recommended** to use AWS IAM roles instead of hardcoding the access key id and secret key.<br/>**It's only used when the storage type is `S3` and `Oss`**. |
@@ -116,7 +116,9 @@
 | `region_engine.mito.worker_request_batch_size` | Integer | `64` | Max batch size for a worker to handle requests. |
 | `region_engine.mito.manifest_checkpoint_distance` | Integer | `10` | Number of meta action updated to trigger a new checkpoint for the manifest. |
 | `region_engine.mito.compress_manifest` | Bool | `false` | Whether to compress manifest and checkpoint file by gzip (default false). |
-| `region_engine.mito.max_background_jobs` | Integer | `4` | Max number of running background jobs |
+| `region_engine.mito.max_background_flushes` | Integer | Auto | Max number of running background flush jobs (default: 1/2 of cpu cores). |
+| `region_engine.mito.max_background_compactions` | Integer | Auto | Max number of running background compaction jobs (default: 1/4 of cpu cores). |
+| `region_engine.mito.max_background_purges` | Integer | Auto | Max number of running background purge jobs (default: number of cpu cores). |
 | `region_engine.mito.auto_flush_interval` | String | `1h` | Interval to auto flush a region if it has not flushed yet. |
 | `region_engine.mito.global_write_buffer_size` | String | Auto | Global write buffer size for all regions. If not set, it's default to 1/8 of OS memory with a max limitation of 1GB. |
 | `region_engine.mito.global_write_buffer_reject_size` | String | Auto | Global write buffer size threshold to reject write requests. If not set, it's default to 2 times of `global_write_buffer_size`. |
@@ -124,9 +126,9 @@
 | `region_engine.mito.vector_cache_size` | String | Auto | Cache size for vectors and arrow arrays. Setting it to 0 to disable the cache.<br/>If not set, it's default to 1/16 of OS memory with a max limitation of 512MB. |
 | `region_engine.mito.page_cache_size` | String | Auto | Cache size for pages of SST row groups. Setting it to 0 to disable the cache.<br/>If not set, it's default to 1/8 of OS memory. |
 | `region_engine.mito.selector_result_cache_size` | String | Auto | Cache size for time series selector (e.g. `last_value()`). Setting it to 0 to disable the cache.<br/>If not set, it's default to 1/16 of OS memory with a max limitation of 512MB. |
-| `region_engine.mito.enable_experimental_write_cache` | Bool | `false` | Whether to enable the experimental write cache. |
+| `region_engine.mito.enable_experimental_write_cache` | Bool | `false` | Whether to enable the experimental write cache. It is recommended to enable it when using object storage for better performance. |
 | `region_engine.mito.experimental_write_cache_path` | String | `""` | File system path for write cache, defaults to `{data_home}/write_cache`. |
-| `region_engine.mito.experimental_write_cache_size` | String | `512MB` | Capacity for write cache. |
+| `region_engine.mito.experimental_write_cache_size` | String | `1GiB` | Capacity for write cache. If your disk space is sufficient, it is recommended to set it larger. |
 | `region_engine.mito.experimental_write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
 | `region_engine.mito.scan_parallelism` | Integer | `0` | Parallelism to scan a region (default: 1/4 of cpu cores).<br/>- `0`: using the default value (1/4 of cpu cores).<br/>- `1`: scan in current thread.<br/>- `n`: scan in parallelism n. |
@@ -410,12 +412,12 @@
 | `wal.backoff_deadline` | String | `5mins` | The deadline of retries.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.create_index` | Bool | `true` | Whether to enable WAL index creation.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.dump_index_interval` | String | `60s` | The interval for dumping WAL indexes.<br/>**It's only used when the provider is `kafka`**. |
-| `wal.overwrite_entry_start_id` | Bool | `false` | Ignore missing entries during read WAL.<br/>**It's only used when the provider is `kafka`**.<br/><br/>This option ensures that when Kafka messages are deleted, the system <br/>can still successfully replay memtable data without throwing an <br/>out-of-range error. <br/>However, enabling this option might lead to unexpected data loss, <br/>as the system will skip over missing entries instead of treating <br/>them as critical errors. |
+| `wal.overwrite_entry_start_id` | Bool | `false` | Ignore missing entries during read WAL.<br/>**It's only used when the provider is `kafka`**.<br/><br/>This option ensures that when Kafka messages are deleted, the system<br/>can still successfully replay memtable data without throwing an<br/>out-of-range error.<br/>However, enabling this option might lead to unexpected data loss,<br/>as the system will skip over missing entries instead of treating<br/>them as critical errors. |
 | `storage` | -- | -- | The data storage options. |
 | `storage.data_home` | String | `/tmp/greptimedb/` | The working home directory. |
 | `storage.type` | String | `File` | The storage type used to store the data.<br/>- `File`: the data is stored in the local file system.<br/>- `S3`: the data is stored in the S3 object storage.<br/>- `Gcs`: the data is stored in the Google Cloud Storage.<br/>- `Azblob`: the data is stored in the Azure Blob Storage.<br/>- `Oss`: the data is stored in the Aliyun OSS. |
-| `storage.cache_path` | String | Unset | Cache configuration for object storage such as 'S3' etc.<br/>The local file cache directory. |
-| `storage.cache_capacity` | String | Unset | The local file cache capacity in bytes. |
+| `storage.cache_path` | String | Unset | Cache configuration for object storage such as 'S3' etc. It is recommended to configure it when using object storage for better performance.<br/>The local file cache directory. |
+| `storage.cache_capacity` | String | Unset | The local file cache capacity in bytes. If your disk space is sufficient, it is recommended to set it larger. |
 | `storage.bucket` | String | Unset | The S3 bucket name.<br/>**It's only used when the storage type is `S3`, `Oss` and `Gcs`**. |
 | `storage.root` | String | Unset | The S3 data will be stored in the specified prefix, for example, `s3://${bucket}/${root}`.<br/>**It's only used when the storage type is `S3`, `Oss` and `Azblob`**. |
 | `storage.access_key_id` | String | Unset | The access key id of the aws account.<br/>It's **highly recommended** to use AWS IAM roles instead of hardcoding the access key id and secret key.<br/>**It's only used when the storage type is `S3` and `Oss`**. |
@@ -437,7 +439,9 @@
 | `region_engine.mito.worker_request_batch_size` | Integer | `64` | Max batch size for a worker to handle requests. |
 | `region_engine.mito.manifest_checkpoint_distance` | Integer | `10` | Number of meta action updated to trigger a new checkpoint for the manifest. |
 | `region_engine.mito.compress_manifest` | Bool | `false` | Whether to compress manifest and checkpoint file by gzip (default false). |
-| `region_engine.mito.max_background_jobs` | Integer | `4` | Max number of running background jobs |
+| `region_engine.mito.max_background_flushes` | Integer | Auto | Max number of running background flush jobs (default: 1/2 of cpu cores). |
+| `region_engine.mito.max_background_compactions` | Integer | Auto | Max number of running background compaction jobs (default: 1/4 of cpu cores). |
+| `region_engine.mito.max_background_purges` | Integer | Auto | Max number of running background purge jobs (default: number of cpu cores). |
 | `region_engine.mito.auto_flush_interval` | String | `1h` | Interval to auto flush a region if it has not flushed yet. |
 | `region_engine.mito.global_write_buffer_size` | String | Auto | Global write buffer size for all regions. If not set, it's default to 1/8 of OS memory with a max limitation of 1GB. |
 | `region_engine.mito.global_write_buffer_reject_size` | String | Auto | Global write buffer size threshold to reject write requests. If not set, it's default to 2 times of `global_write_buffer_size` |
@@ -445,9 +449,9 @@
 | `region_engine.mito.vector_cache_size` | String | Auto | Cache size for vectors and arrow arrays. Setting it to 0 to disable the cache.<br/>If not set, it's default to 1/16 of OS memory with a max limitation of 512MB. |
 | `region_engine.mito.page_cache_size` | String | Auto | Cache size for pages of SST row groups. Setting it to 0 to disable the cache.<br/>If not set, it's default to 1/8 of OS memory. |
 | `region_engine.mito.selector_result_cache_size` | String | Auto | Cache size for time series selector (e.g. `last_value()`). Setting it to 0 to disable the cache.<br/>If not set, it's default to 1/16 of OS memory with a max limitation of 512MB. |
-| `region_engine.mito.enable_experimental_write_cache` | Bool | `false` | Whether to enable the experimental write cache. |
+| `region_engine.mito.enable_experimental_write_cache` | Bool | `false` | Whether to enable the experimental write cache. It is recommended to enable it when using object storage for better performance. |
 | `region_engine.mito.experimental_write_cache_path` | String | `""` | File system path for write cache, defaults to `{data_home}/write_cache`. |
-| `region_engine.mito.experimental_write_cache_size` | String | `512MB` | Capacity for write cache. |
+| `region_engine.mito.experimental_write_cache_size` | String | `1GiB` | Capacity for write cache. If your disk space is sufficient, it is recommended to set it larger. |
 | `region_engine.mito.experimental_write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
 | `region_engine.mito.scan_parallelism` | Integer | `0` | Parallelism to scan a region (default: 1/4 of cpu cores).<br/>- `0`: using the default value (1/4 of cpu cores).<br/>- `1`: scan in current thread.<br/>- `n`: scan in parallelism n. |
--- a/config/datanode.example.toml
+++ b/config/datanode.example.toml
@@ -215,12 +215,12 @@ dump_index_interval = "60s"

 ## Ignore missing entries during read WAL.
 ## **It's only used when the provider is `kafka`**.
-## 
-## This option ensures that when Kafka messages are deleted, the system 
-## can still successfully replay memtable data without throwing an 
-## out-of-range error. 
-## However, enabling this option might lead to unexpected data loss, 
-## as the system will skip over missing entries instead of treating 
+##
+## This option ensures that when Kafka messages are deleted, the system
+## can still successfully replay memtable data without throwing an
+## out-of-range error.
+## However, enabling this option might lead to unexpected data loss,
+## as the system will skip over missing entries instead of treating
 ## them as critical errors.
 overwrite_entry_start_id = false

@@ -294,14 +294,14 @@ data_home = "/tmp/greptimedb/"
 ## - `Oss`: the data is stored in the Aliyun OSS.
 type = "File"

-## Cache configuration for object storage such as 'S3' etc.
+## Cache configuration for object storage such as 'S3' etc. It is recommended to configure it when using object storage for better performance.
 ## The local file cache directory.
 ## @toml2docs:none-default
 cache_path = "/path/local_cache"

-## The local file cache capacity in bytes.
+## The local file cache capacity in bytes. If your disk space is sufficient, it is recommended to set it larger.
 ## @toml2docs:none-default
-cache_capacity = "256MB"
+cache_capacity = "1GiB"

 ## The S3 bucket name.
 ## **It's only used when the storage type is `S3`, `Oss` and `Gcs`**.
@@ -416,8 +416,17 @@ manifest_checkpoint_distance = 10
 ## Whether to compress manifest and checkpoint file by gzip (default false).
 compress_manifest = false

-## Max number of running background jobs
-max_background_jobs = 4
+## Max number of running background flush jobs (default: 1/2 of cpu cores).
+## @toml2docs:none-default="Auto"
+#+ max_background_flushes = 4
+
+## Max number of running background compaction jobs (default: 1/4 of cpu cores).
+## @toml2docs:none-default="Auto"
+#+ max_background_compactions = 2
+
+## Max number of running background purge jobs (default: number of cpu cores).
+## @toml2docs:none-default="Auto"
+#+ max_background_purges = 8

 ## Interval to auto flush a region if it has not flushed yet.
 auto_flush_interval = "1h"
@@ -450,14 +459,14 @@ auto_flush_interval = "1h"
 ## @toml2docs:none-default="Auto"
 #+ selector_result_cache_size = "512MB"

-## Whether to enable the experimental write cache.
+## Whether to enable the experimental write cache. It is recommended to enable it when using object storage for better performance.
 enable_experimental_write_cache = false

 ## File system path for write cache, defaults to `{data_home}/write_cache`.
 experimental_write_cache_path = ""

-## Capacity for write cache.
-experimental_write_cache_size = "512MB"
+## Capacity for write cache. If your disk space is sufficient, it is recommended to set it larger.
+experimental_write_cache_size = "1GiB"

 ## TTL for write cache.
 ## @toml2docs:none-default
@@ -637,7 +646,7 @@ url = ""
 headers = { }

 ## The tracing options. Only effect when compiled with `tokio-console` feature.
-[tracing]
+#+ [tracing]
 ## The tokio console address.
 ## @toml2docs:none-default
-tokio_console_addr = "127.0.0.1"
+#+ tokio_console_addr = "127.0.0.1"
--- a/config/flownode.example.toml
+++ b/config/flownode.example.toml
@@ -101,8 +101,8 @@ threshold = "10s"
 sample_ratio = 1.0

 ## The tracing options. Only effect when compiled with `tokio-console` feature.
-[tracing]
+#+ [tracing]
 ## The tokio console address.
 ## @toml2docs:none-default
-tokio_console_addr = "127.0.0.1"
+#+ tokio_console_addr = "127.0.0.1"

--- a/config/frontend.example.toml
+++ b/config/frontend.example.toml
@@ -231,7 +231,7 @@ url = ""
 headers = { }

 ## The tracing options. Only effect when compiled with `tokio-console` feature.
-[tracing]
+#+ [tracing]
 ## The tokio console address.
 ## @toml2docs:none-default
-tokio_console_addr = "127.0.0.1"
+#+ tokio_console_addr = "127.0.0.1"
--- a/config/metasrv.example.toml
+++ b/config/metasrv.example.toml
@@ -218,7 +218,7 @@ url = ""
 headers = { }

 ## The tracing options. Only effect when compiled with `tokio-console` feature.
-[tracing]
+#+ [tracing]
 ## The tokio console address.
 ## @toml2docs:none-default
-tokio_console_addr = "127.0.0.1"
+#+ tokio_console_addr = "127.0.0.1"
--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -239,12 +239,12 @@ backoff_deadline = "5mins"

 ## Ignore missing entries during read WAL.
 ## **It's only used when the provider is `kafka`**.
-## 
-## This option ensures that when Kafka messages are deleted, the system 
-## can still successfully replay memtable data without throwing an 
-## out-of-range error. 
-## However, enabling this option might lead to unexpected data loss, 
-## as the system will skip over missing entries instead of treating 
+##
+## This option ensures that when Kafka messages are deleted, the system
+## can still successfully replay memtable data without throwing an
+## out-of-range error.
+## However, enabling this option might lead to unexpected data loss,
+## as the system will skip over missing entries instead of treating
 ## them as critical errors.
 overwrite_entry_start_id = false

@@ -332,14 +332,14 @@ data_home = "/tmp/greptimedb/"
 ## - `Oss`: the data is stored in the Aliyun OSS.
 type = "File"

-## Cache configuration for object storage such as 'S3' etc.
+## Cache configuration for object storage such as 'S3' etc. It is recommended to configure it when using object storage for better performance.
 ## The local file cache directory.
 ## @toml2docs:none-default
 cache_path = "/path/local_cache"

-## The local file cache capacity in bytes.
+## The local file cache capacity in bytes. If your disk space is sufficient, it is recommended to set it larger.
 ## @toml2docs:none-default
-cache_capacity = "256MB"
+cache_capacity = "1GiB"

 ## The S3 bucket name.
 ## **It's only used when the storage type is `S3`, `Oss` and `Gcs`**.
@@ -454,8 +454,17 @@ manifest_checkpoint_distance = 10
 ## Whether to compress manifest and checkpoint file by gzip (default false).
 compress_manifest = false

-## Max number of running background jobs
-max_background_jobs = 4
+## Max number of running background flush jobs (default: 1/2 of cpu cores).
+## @toml2docs:none-default="Auto"
+#+ max_background_flushes = 4
+
+## Max number of running background compaction jobs (default: 1/4 of cpu cores).
+## @toml2docs:none-default="Auto"
+#+ max_background_compactions = 2
+
+## Max number of running background purge jobs (default: number of cpu cores).
+## @toml2docs:none-default="Auto"
+#+ max_background_purges = 8

 ## Interval to auto flush a region if it has not flushed yet.
 auto_flush_interval = "1h"
@@ -488,14 +497,14 @@ auto_flush_interval = "1h"
 ## @toml2docs:none-default="Auto"
 #+ selector_result_cache_size = "512MB"

-## Whether to enable the experimental write cache.
+## Whether to enable the experimental write cache. It is recommended to enable it when using object storage for better performance.
 enable_experimental_write_cache = false

 ## File system path for write cache, defaults to `{data_home}/write_cache`.
 experimental_write_cache_path = ""

-## Capacity for write cache.
-experimental_write_cache_size = "512MB"
+## Capacity for write cache. If your disk space is sufficient, it is recommended to set it larger.
+experimental_write_cache_size = "1GiB"

 ## TTL for write cache.
 ## @toml2docs:none-default
@@ -681,7 +690,7 @@ url = ""
 headers = { }

 ## The tracing options. Only effect when compiled with `tokio-console` feature.
-[tracing]
+#+ [tracing]
 ## The tokio console address.
 ## @toml2docs:none-default
-tokio_console_addr = "127.0.0.1"
+#+ tokio_console_addr = "127.0.0.1"
--- a/docs/how-to/how-to-change-log-level-on-the-fly.md
+++ b/docs/how-to/how-to-change-log-level-on-the-fly.md
@@ -0,0 +1,16 @@
+# Change Log Level on the Fly
+
+## HTTP API
+
+example:
+```bash
+curl --data "trace,flow=debug" 127.0.0.1:4000/debug/log_level
+```
+And database will reply with something like:
+```bash
+Log Level changed from Some("info") to "trace,flow=debug"%
+```
+
+The data is a string in the format of `global_level,module1=level1,module2=level2,...` that follow the same rule of `RUST_LOG`. 
+
+The module is the module name of the log, and the level is the log level. The log level can be one of the following: `trace`, `debug`, `info`, `warn`, `error`, `off`(case insensitive).
--- a/rust-toolchain.toml
+++ b/rust-toolchain.toml
@@ -1,3 +1,2 @@
 [toolchain]
-channel = "nightly-2024-06-06"
-
+channel = "nightly-2024-10-19"
--- a/src/api/src/helper.rs
+++ b/src/api/src/helper.rs
@@ -116,6 +116,7 @@ impl From<ColumnDataTypeWrapper> for ConcreteDataType {
                    ConcreteDataType::binary_datatype()
                }
            }
+            ColumnDataType::Json => ConcreteDataType::json_datatype(),
            ColumnDataType::String => ConcreteDataType::string_datatype(),
            ColumnDataType::Date => ConcreteDataType::date_datatype(),
            ColumnDataType::Datetime => ConcreteDataType::datetime_datatype(),
@@ -417,6 +418,10 @@ pub fn values_with_capacity(datatype: ColumnDataType, capacity: usize) -> Values
            decimal128_values: Vec::with_capacity(capacity),
            ..Default::default()
        },
+        ColumnDataType::Json => Values {
+            string_values: Vec::with_capacity(capacity),
+            ..Default::default()
+        },
    }
 }

--- a/src/auth/src/user_provider/static_user_provider.rs
+++ b/src/auth/src/user_provider/static_user_provider.rs
@@ -33,7 +33,7 @@ impl StaticUserProvider {
            value: value.to_string(),
            msg: "StaticUserProviderOption must be in format `<option>:<value>`",
        })?;
-        return match mode {
+        match mode {
            "file" => {
                let users = load_credential_from_file(content)?
                    .context(InvalidConfigSnafu {
@@ -58,7 +58,7 @@ impl StaticUserProvider {
                msg: "StaticUserProviderOption must be in format `file:<path>` or `cmd:<values>`",
            }
                .fail(),
-        };
+        }
    }
 }

--- a/src/catalog/src/error.rs
+++ b/src/catalog/src/error.rs
@@ -178,6 +178,12 @@ pub enum Error {
        location: Location,
    },

+    #[snafu(display("Partition manager not found, it's not expected."))]
+    PartitionManagerNotFound {
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Failed to find table partitions"))]
    FindPartitions { source: partition::error::Error },

@@ -301,6 +307,7 @@ impl ErrorExt for Error {
            | Error::CastManager { .. }
            | Error::Json { .. }
            | Error::GetInformationExtension { .. }
+            | Error::PartitionManagerNotFound { .. }
            | Error::ProcedureIdNotFound { .. } => StatusCode::Unexpected,

            Error::ViewPlanColumnsChanged { .. } => StatusCode::InvalidArguments,
--- a/src/catalog/src/system_schema/information_schema/partitions.rs
+++ b/src/catalog/src/system_schema/information_schema/partitions.rs
@@ -34,15 +34,14 @@ use datatypes::vectors::{
 };
 use futures::{StreamExt, TryStreamExt};
 use partition::manager::PartitionInfo;
-use partition::partition::PartitionDef;
 use snafu::{OptionExt, ResultExt};
-use store_api::storage::{RegionId, ScanRequest, TableId};
+use store_api::storage::{ScanRequest, TableId};
 use table::metadata::{TableInfo, TableType};

 use super::PARTITIONS;
 use crate::error::{
-    CreateRecordBatchSnafu, FindPartitionsSnafu, InternalSnafu, Result,
-    UpgradeWeakCatalogManagerRefSnafu,
+    CreateRecordBatchSnafu, FindPartitionsSnafu, InternalSnafu, PartitionManagerNotFoundSnafu,
+    Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::kvbackend::KvBackendCatalogManager;
 use crate::system_schema::information_schema::{InformationTable, Predicates};
@@ -236,7 +235,8 @@ impl InformationSchemaPartitionsBuilder {
        let partition_manager = catalog_manager
            .as_any()
            .downcast_ref::<KvBackendCatalogManager>()
-            .map(|catalog_manager| catalog_manager.partition_manager());
+            .map(|catalog_manager| catalog_manager.partition_manager())
+            .context(PartitionManagerNotFoundSnafu)?;

        let predicates = Predicates::from_scan_request(&request);

@@ -262,27 +262,10 @@ impl InformationSchemaPartitionsBuilder {
                let table_ids: Vec<TableId> =
                    table_infos.iter().map(|info| info.ident.table_id).collect();

-                let mut table_partitions = if let Some(partition_manager) = &partition_manager {
-                    partition_manager
-                        .batch_find_table_partitions(&table_ids)
-                        .await
-                        .context(FindPartitionsSnafu)?
-                } else {
-                    // Current node must be a standalone instance, contains only one partition by default.
-                    // TODO(dennis): change it when we support multi-regions for standalone.
-                    table_ids
-                        .into_iter()
-                        .map(|table_id| {
-                            (
-                                table_id,
-                                vec![PartitionInfo {
-                                    id: RegionId::new(table_id, 0),
-                                    partition: PartitionDef::new(vec![], vec![]),
-                                }],
-                            )
-                        })
-                        .collect()
-                };
+                let mut table_partitions = partition_manager
+                    .batch_find_table_partitions(&table_ids)
+                    .await
+                    .context(FindPartitionsSnafu)?;

                for table_info in table_infos {
                    let partitions = table_partitions
--- a/src/catalog/src/system_schema/information_schema/region_statistics.rs
+++ b/src/catalog/src/system_schema/information_schema/region_statistics.rs
@@ -39,9 +39,12 @@ use crate::CatalogManager;
 const REGION_ID: &str = "region_id";
 const TABLE_ID: &str = "table_id";
 const REGION_NUMBER: &str = "region_number";
+const REGION_ROWS: &str = "region_rows";
+const DISK_SIZE: &str = "disk_size";
 const MEMTABLE_SIZE: &str = "memtable_size";
 const MANIFEST_SIZE: &str = "manifest_size";
 const SST_SIZE: &str = "sst_size";
+const INDEX_SIZE: &str = "index_size";
 const ENGINE: &str = "engine";
 const REGION_ROLE: &str = "region_role";

@@ -52,9 +55,12 @@ const INIT_CAPACITY: usize = 42;
 /// - `region_id`: The region id.
 /// - `table_id`: The table id.
 /// - `region_number`: The region number.
+/// - `region_rows`: The number of rows in region.
 /// - `memtable_size`: The memtable size in bytes.
+/// - `disk_size`: The approximate disk size in bytes.
 /// - `manifest_size`: The manifest size in bytes.
-/// - `sst_size`: The sst size in bytes.
+/// - `sst_size`: The sst data files size in bytes.
+/// - `index_size`: The sst index files size in bytes.
 /// - `engine`: The engine type.
 /// - `region_role`: The region role.
 ///
@@ -76,9 +82,12 @@ impl InformationSchemaRegionStatistics {
            ColumnSchema::new(REGION_ID, ConcreteDataType::uint64_datatype(), false),
            ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
            ColumnSchema::new(REGION_NUMBER, ConcreteDataType::uint32_datatype(), false),
+            ColumnSchema::new(REGION_ROWS, ConcreteDataType::uint64_datatype(), true),
+            ColumnSchema::new(DISK_SIZE, ConcreteDataType::uint64_datatype(), true),
            ColumnSchema::new(MEMTABLE_SIZE, ConcreteDataType::uint64_datatype(), true),
            ColumnSchema::new(MANIFEST_SIZE, ConcreteDataType::uint64_datatype(), true),
            ColumnSchema::new(SST_SIZE, ConcreteDataType::uint64_datatype(), true),
+            ColumnSchema::new(INDEX_SIZE, ConcreteDataType::uint64_datatype(), true),
            ColumnSchema::new(ENGINE, ConcreteDataType::string_datatype(), true),
            ColumnSchema::new(REGION_ROLE, ConcreteDataType::string_datatype(), true),
        ]))
@@ -135,9 +144,12 @@ struct InformationSchemaRegionStatisticsBuilder {
    region_ids: UInt64VectorBuilder,
    table_ids: UInt32VectorBuilder,
    region_numbers: UInt32VectorBuilder,
+    region_rows: UInt64VectorBuilder,
+    disk_sizes: UInt64VectorBuilder,
    memtable_sizes: UInt64VectorBuilder,
    manifest_sizes: UInt64VectorBuilder,
    sst_sizes: UInt64VectorBuilder,
+    index_sizes: UInt64VectorBuilder,
    engines: StringVectorBuilder,
    region_roles: StringVectorBuilder,
 }
@@ -150,9 +162,12 @@ impl InformationSchemaRegionStatisticsBuilder {
            region_ids: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
            table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
            region_numbers: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
+            region_rows: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
+            disk_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
            memtable_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
            manifest_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
            sst_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
+            index_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
            engines: StringVectorBuilder::with_capacity(INIT_CAPACITY),
            region_roles: StringVectorBuilder::with_capacity(INIT_CAPACITY),
        }
@@ -177,9 +192,12 @@ impl InformationSchemaRegionStatisticsBuilder {
            (REGION_ID, &Value::from(region_stat.id.as_u64())),
            (TABLE_ID, &Value::from(region_stat.id.table_id())),
            (REGION_NUMBER, &Value::from(region_stat.id.region_number())),
+            (REGION_ROWS, &Value::from(region_stat.num_rows)),
+            (DISK_SIZE, &Value::from(region_stat.approximate_bytes)),
            (MEMTABLE_SIZE, &Value::from(region_stat.memtable_size)),
            (MANIFEST_SIZE, &Value::from(region_stat.manifest_size)),
            (SST_SIZE, &Value::from(region_stat.sst_size)),
+            (INDEX_SIZE, &Value::from(region_stat.index_size)),
            (ENGINE, &Value::from(region_stat.engine.as_str())),
            (REGION_ROLE, &Value::from(region_stat.role.to_string())),
        ];
@@ -192,9 +210,12 @@ impl InformationSchemaRegionStatisticsBuilder {
        self.table_ids.push(Some(region_stat.id.table_id()));
        self.region_numbers
            .push(Some(region_stat.id.region_number()));
+        self.region_rows.push(Some(region_stat.num_rows));
+        self.disk_sizes.push(Some(region_stat.approximate_bytes));
        self.memtable_sizes.push(Some(region_stat.memtable_size));
        self.manifest_sizes.push(Some(region_stat.manifest_size));
        self.sst_sizes.push(Some(region_stat.sst_size));
+        self.index_sizes.push(Some(region_stat.index_size));
        self.engines.push(Some(&region_stat.engine));
        self.region_roles.push(Some(&region_stat.role.to_string()));
    }
@@ -204,9 +225,12 @@ impl InformationSchemaRegionStatisticsBuilder {
            Arc::new(self.region_ids.finish()),
            Arc::new(self.table_ids.finish()),
            Arc::new(self.region_numbers.finish()),
+            Arc::new(self.region_rows.finish()),
+            Arc::new(self.disk_sizes.finish()),
            Arc::new(self.memtable_sizes.finish()),
            Arc::new(self.manifest_sizes.finish()),
            Arc::new(self.sst_sizes.finish()),
+            Arc::new(self.index_sizes.finish()),
            Arc::new(self.engines.finish()),
            Arc::new(self.region_roles.finish()),
        ];
--- a/src/catalog/src/system_schema/information_schema/table_names.rs
+++ b/src/catalog/src/system_schema/information_schema/table_names.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-/// All table names in `information_schema`.
+//! All table names in `information_schema`.

 pub const TABLES: &str = "tables";
 pub const COLUMNS: &str = "columns";
--- a/src/catalog/src/system_schema/information_schema/tables.rs
+++ b/src/catalog/src/system_schema/information_schema/tables.rs
@@ -12,13 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::collections::HashSet;
 use std::sync::{Arc, Weak};

 use arrow_schema::SchemaRef as ArrowSchemaRef;
-use common_catalog::consts::INFORMATION_SCHEMA_TABLES_TABLE_ID;
+use common_catalog::consts::{INFORMATION_SCHEMA_TABLES_TABLE_ID, MITO_ENGINE};
 use common_error::ext::BoxedError;
+use common_meta::datanode::RegionStat;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
+use common_telemetry::error;
 use datafusion::execution::TaskContext;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
@@ -31,7 +34,7 @@ use datatypes::vectors::{
 };
 use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
-use store_api::storage::{ScanRequest, TableId};
+use store_api::storage::{RegionId, ScanRequest, TableId};
 use table::metadata::{TableInfo, TableType};

 use super::TABLES;
@@ -39,6 +42,7 @@ use crate::error::{
    CreateRecordBatchSnafu, InternalSnafu, Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::system_schema::information_schema::{InformationTable, Predicates};
+use crate::system_schema::utils;
 use crate::CatalogManager;

 pub const TABLE_CATALOG: &str = "table_catalog";
@@ -234,17 +238,50 @@ impl InformationSchemaTablesBuilder {
            .context(UpgradeWeakCatalogManagerRefSnafu)?;
        let predicates = Predicates::from_scan_request(&request);

+        let information_extension = utils::information_extension(&self.catalog_manager)?;
+
+        // TODO(dennis): `region_stats` API is not stable in distributed cluster because of network issue etc.
+        // But we don't want the statements such as `show tables` fail,
+        // so using `unwrap_or_else` here instead of `?` operator.
+        let region_stats = information_extension
+            .region_stats()
+            .await
+            .map_err(|e| {
+                error!(e; "Failed to call region_stats");
+                e
+            })
+            .unwrap_or_else(|_| vec![]);
+
        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
            let mut stream = catalog_manager.tables(&catalog_name, &schema_name, None);

            while let Some(table) = stream.try_next().await? {
                let table_info = table.table_info();
+
+                // TODO(dennis): make it working for metric engine
+                let table_region_stats = if table_info.meta.engine == MITO_ENGINE {
+                    let region_ids = table_info
+                        .meta
+                        .region_numbers
+                        .iter()
+                        .map(|n| RegionId::new(table_info.ident.table_id, *n))
+                        .collect::<HashSet<_>>();
+
+                    region_stats
+                        .iter()
+                        .filter(|stat| region_ids.contains(&stat.id))
+                        .collect::<Vec<_>>()
+                } else {
+                    vec![]
+                };
+
                self.add_table(
                    &predicates,
                    &catalog_name,
                    &schema_name,
                    table_info,
                    table.table_type(),
+                    &table_region_stats,
                );
            }
        }
@@ -260,6 +297,7 @@ impl InformationSchemaTablesBuilder {
        schema_name: &str,
        table_info: Arc<TableInfo>,
        table_type: TableType,
+        region_stats: &[&RegionStat],
    ) {
        let table_name = table_info.name.as_ref();
        let table_id = table_info.table_id();
@@ -273,7 +311,9 @@ impl InformationSchemaTablesBuilder {

        let row = [
            (TABLE_CATALOG, &Value::from(catalog_name)),
+            (TABLE_ID, &Value::from(table_id)),
            (TABLE_SCHEMA, &Value::from(schema_name)),
+            (ENGINE, &Value::from(engine)),
            (TABLE_NAME, &Value::from(table_name)),
            (TABLE_TYPE, &Value::from(table_type_text)),
        ];
@@ -287,21 +327,39 @@ impl InformationSchemaTablesBuilder {
        self.table_names.push(Some(table_name));
        self.table_types.push(Some(table_type_text));
        self.table_ids.push(Some(table_id));
+
+        let data_length = region_stats.iter().map(|stat| stat.sst_size).sum();
+        let table_rows = region_stats.iter().map(|stat| stat.num_rows).sum();
+        let index_length = region_stats.iter().map(|stat| stat.index_size).sum();
+
+        // It's not precise, but it is acceptable for long-term data storage.
+        let avg_row_length = if table_rows > 0 {
+            let total_data_length = data_length
+                + region_stats
+                    .iter()
+                    .map(|stat| stat.memtable_size)
+                    .sum::<u64>();
+
+            total_data_length / table_rows
+        } else {
+            0
+        };
+
+        self.data_length.push(Some(data_length));
+        self.index_length.push(Some(index_length));
+        self.table_rows.push(Some(table_rows));
+        self.avg_row_length.push(Some(avg_row_length));
+
        // TODO(sunng87): use real data for these fields
-        self.data_length.push(Some(0));
        self.max_data_length.push(Some(0));
-        self.index_length.push(Some(0));
-        self.avg_row_length.push(Some(0));
-        self.max_index_length.push(Some(0));
        self.checksum.push(Some(0));
-        self.table_rows.push(Some(0));
+        self.max_index_length.push(Some(0));
        self.data_free.push(Some(0));
        self.auto_increment.push(Some(0));
        self.row_format.push(Some("Fixed"));
        self.table_collation.push(Some("utf8_bin"));
        self.update_time.push(None);
        self.check_time.push(None);
-
        // use mariadb default table version number here
        self.version.push(Some(11));
        self.table_comment.push(table_info.desc.as_deref());
--- a/src/catalog/src/system_schema/memory_table.rs
+++ b/src/catalog/src/system_schema/memory_table.rs
@@ -74,7 +74,7 @@ impl MemoryTableBuilder {
    /// Construct the `information_schema.{table_name}` virtual table
    pub async fn memory_records(&mut self) -> Result<RecordBatch> {
        if self.columns.is_empty() {
-            RecordBatch::new_empty(self.schema.clone()).context(CreateRecordBatchSnafu)
+            Ok(RecordBatch::new_empty(self.schema.clone()))
        } else {
            RecordBatch::new(self.schema.clone(), std::mem::take(&mut self.columns))
                .context(CreateRecordBatchSnafu)
--- a/src/catalog/src/system_schema/pg_catalog/pg_namespace.rs
+++ b/src/catalog/src/system_schema/pg_catalog/pg_namespace.rs
@@ -12,6 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+//! The `pg_catalog.pg_namespace` table implementation.
+//! namespace is a schema in greptime
+
 pub(super) mod oid_map;

 use std::sync::{Arc, Weak};
@@ -40,9 +43,6 @@ use crate::system_schema::utils::tables::{string_column, u32_column};
 use crate::system_schema::SystemTable;
 use crate::CatalogManager;

-/// The `pg_catalog.pg_namespace` table implementation.
-/// namespace is a schema in greptime
-
 const NSPNAME: &str = "nspname";
 const INIT_CAPACITY: usize = 42;

--- a/src/client/Cargo.toml
+++ b/src/client/Cargo.toml
@@ -28,7 +28,7 @@ enum_dispatch = "0.3"
 futures-util.workspace = true
 lazy_static.workspace = true
 moka = { workspace = true, features = ["future"] }
-parking_lot = "0.12"
+parking_lot.workspace = true
 prometheus.workspace = true
 prost.workspace = true
 query.workspace = true
@@ -45,7 +45,6 @@ common-grpc-expr.workspace = true
 datanode.workspace = true
 derive-new = "0.5"
 tracing = "0.1"
-tracing-subscriber = { version = "0.3", features = ["env-filter"] }

 [dev-dependencies.substrait_proto]
 package = "substrait"
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -78,7 +78,7 @@ table.workspace = true
 tokio.workspace = true
 toml.workspace = true
 tonic.workspace = true
-tracing-appender = "0.2"
+tracing-appender.workspace = true

 [target.'cfg(not(windows))'.dependencies]
 tikv-jemallocator = "0.6"
--- a/src/cmd/src/cli/repl.rs
+++ b/src/cmd/src/cli/repl.rs
@@ -174,7 +174,7 @@ impl Repl {

            let plan = query_engine
                .planner()
-                .plan(stmt, query_ctx.clone())
+                .plan(&stmt, query_ctx.clone())
                .await
                .context(PlanStatementSnafu)?;

--- a/src/cmd/src/datanode.rs
+++ b/src/cmd/src/datanode.rs
@@ -272,9 +272,10 @@ impl StartCommand {
        info!("Datanode start command: {:#?}", self);
        info!("Datanode options: {:#?}", opts);

+        let plugin_opts = opts.plugins;
        let opts = opts.component;
        let mut plugins = Plugins::new();
-        plugins::setup_datanode_plugins(&mut plugins, &opts)
+        plugins::setup_datanode_plugins(&mut plugins, &plugin_opts, &opts)
            .await
            .context(StartDatanodeSnafu)?;

--- a/src/cmd/src/frontend.rs
+++ b/src/cmd/src/frontend.rs
@@ -266,9 +266,10 @@ impl StartCommand {
        info!("Frontend start command: {:#?}", self);
        info!("Frontend options: {:#?}", opts);

+        let plugin_opts = opts.plugins;
        let opts = opts.component;
        let mut plugins = Plugins::new();
-        plugins::setup_frontend_plugins(&mut plugins, &opts)
+        plugins::setup_frontend_plugins(&mut plugins, &plugin_opts, &opts)
            .await
            .context(StartFrontendSnafu)?;

@@ -342,6 +343,8 @@ impl StartCommand {
        // Some queries are expected to take long time.
        let channel_config = ChannelConfig {
            timeout: None,
+            tcp_nodelay: opts.datanode.client.tcp_nodelay,
+            connect_timeout: Some(opts.datanode.client.connect_timeout),
            ..Default::default()
        };
        let client = NodeClients::new(channel_config);
@@ -472,7 +475,7 @@ mod tests {
        };

        let mut plugins = Plugins::new();
-        plugins::setup_frontend_plugins(&mut plugins, &fe_opts)
+        plugins::setup_frontend_plugins(&mut plugins, &[], &fe_opts)
            .await
            .unwrap();

--- a/src/cmd/src/lib.rs
+++ b/src/cmd/src/lib.rs
@@ -84,6 +84,7 @@ pub trait App: Send {
 }

 /// Log the versions of the application, and the arguments passed to the cli.
+///
 /// `version` should be the same as the output of cli "--version";
 /// and the `short_version` is the short version of the codes, often consist of git branch and commit.
 pub fn log_versions(version: &str, short_version: &str, app: &str) {
--- a/src/cmd/src/metasrv.rs
+++ b/src/cmd/src/metasrv.rs
@@ -273,9 +273,10 @@ impl StartCommand {
        info!("Metasrv start command: {:#?}", self);
        info!("Metasrv options: {:#?}", opts);

+        let plugin_opts = opts.plugins;
        let opts = opts.component;
        let mut plugins = Plugins::new();
-        plugins::setup_metasrv_plugins(&mut plugins, &opts)
+        plugins::setup_metasrv_plugins(&mut plugins, &plugin_opts, &opts)
            .await
            .context(StartMetaServerSnafu)?;

--- a/src/cmd/src/options.rs
+++ b/src/cmd/src/options.rs
@@ -15,6 +15,7 @@
 use clap::Parser;
 use common_config::Configurable;
 use common_runtime::global::RuntimeOptions;
+use plugins::PluginOptions;
 use serde::{Deserialize, Serialize};

 #[derive(Parser, Default, Debug, Clone)]
@@ -40,6 +41,8 @@ pub struct GlobalOptions {
 pub struct GreptimeOptions<T> {
    /// The runtime options.
    pub runtime: RuntimeOptions,
+    /// The plugin options.
+    pub plugins: Vec<PluginOptions>,

    /// The options of each component (like Datanode or Standalone) of GreptimeDB.
    #[serde(flatten)]
--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+use std::net::SocketAddr;
 use std::sync::Arc;
 use std::{fs, path};

@@ -250,6 +251,13 @@ pub struct Instance {
    _guard: Vec<WorkerGuard>,
 }

+impl Instance {
+    /// Find the socket addr of a server by its `name`.
+    pub async fn server_addr(&self, name: &str) -> Option<SocketAddr> {
+        self.frontend.server_handlers().addr(name).await
+    }
+}
+
 #[async_trait]
 impl App for Instance {
    fn name(&self) -> &str {
@@ -340,7 +348,8 @@ pub struct StartCommand {
 }

 impl StartCommand {
-    fn load_options(
+    /// Load the GreptimeDB options from various sources (command line, config file or env).
+    pub fn load_options(
        &self,
        global_options: &GlobalOptions,
    ) -> Result<GreptimeOptions<StandaloneOptions>> {
@@ -430,7 +439,8 @@ impl StartCommand {
    #[allow(unreachable_code)]
    #[allow(unused_variables)]
    #[allow(clippy::diverging_sub_expression)]
-    async fn build(&self, opts: GreptimeOptions<StandaloneOptions>) -> Result<Instance> {
+    /// Build GreptimeDB instance with the loaded options.
+    pub async fn build(&self, opts: GreptimeOptions<StandaloneOptions>) -> Result<Instance> {
        common_runtime::init_global_runtimes(&opts.runtime);

        let guard = common_telemetry::init_global_logging(
@@ -445,15 +455,16 @@ impl StartCommand {
        info!("Standalone options: {opts:#?}");

        let mut plugins = Plugins::new();
+        let plugin_opts = opts.plugins;
        let opts = opts.component;
        let fe_opts = opts.frontend_options();
        let dn_opts = opts.datanode_options();

-        plugins::setup_frontend_plugins(&mut plugins, &fe_opts)
+        plugins::setup_frontend_plugins(&mut plugins, &plugin_opts, &fe_opts)
            .await
            .context(StartFrontendSnafu)?;

-        plugins::setup_datanode_plugins(&mut plugins, &dn_opts)
+        plugins::setup_datanode_plugins(&mut plugins, &plugin_opts, &dn_opts)
            .await
            .context(StartDatanodeSnafu)?;

@@ -725,12 +736,14 @@ impl InformationExtension for StandaloneInformationExtension {
                    id: stat.region_id,
                    rcus: 0,
                    wcus: 0,
-                    approximate_bytes: region_stat.estimated_disk_size() as i64,
+                    approximate_bytes: region_stat.estimated_disk_size(),
                    engine: stat.engine,
                    role: RegionRole::from(stat.role).into(),
+                    num_rows: region_stat.num_rows,
                    memtable_size: region_stat.memtable_size,
                    manifest_size: region_stat.manifest_size,
                    sst_size: region_stat.sst_size,
+                    index_size: region_stat.index_size,
                }
            })
            .collect::<Vec<_>>();
@@ -762,7 +775,7 @@ mod tests {
        };

        let mut plugins = Plugins::new();
-        plugins::setup_frontend_plugins(&mut plugins, &fe_opts)
+        plugins::setup_frontend_plugins(&mut plugins, &[], &fe_opts)
            .await
            .unwrap();

--- a/src/cmd/tests/load_config_test.rs
+++ b/src/cmd/tests/load_config_test.rs
@@ -20,7 +20,7 @@ use common_config::Configurable;
 use common_grpc::channel_manager::{
    DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE, DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
 };
-use common_telemetry::logging::{LoggingOptions, DEFAULT_OTLP_ENDPOINT};
+use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, DEFAULT_OTLP_ENDPOINT};
 use common_wal::config::raft_engine::RaftEngineConfig;
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
@@ -159,8 +159,20 @@ fn test_load_metasrv_example_config() {
                level: Some("info".to_string()),
                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
                tracing_sample_ratio: Some(Default::default()),
+                slow_query: SlowQueryOptions {
+                    enable: false,
+                    threshold: Some(Duration::from_secs(10)),
+                    sample_ratio: Some(1.0),
+                },
                ..Default::default()
            },
+            datanode: meta_srv::metasrv::DatanodeOptions {
+                client: meta_srv::metasrv::DatanodeClientOptions {
+                    timeout: Duration::from_secs(10),
+                    connect_timeout: Duration::from_secs(10),
+                    tcp_nodelay: true,
+                },
+            },
            export_metrics: ExportMetricsOption {
                self_import: Some(Default::default()),
                remote_write: Some(Default::default()),
--- a/src/common/base/src/secrets.rs
+++ b/src/common/base/src/secrets.rs
@@ -46,8 +46,9 @@ impl From<String> for SecretString {
    }
 }

-/// Wrapper type for values that contains secrets, which attempts to limit
-/// accidental exposure and ensure secrets are wiped from memory when dropped.
+/// Wrapper type for values that contains secrets.
+///
+/// It attempts to limit accidental exposure and ensure secrets are wiped from memory when dropped.
 /// (e.g. passwords, cryptographic keys, access tokens or other credentials)
 ///
 /// Access to the secret inner value occurs through the [`ExposeSecret`]
--- a/src/common/catalog/src/consts.rs
+++ b/src/common/catalog/src/consts.rs
@@ -103,14 +103,15 @@ pub const INFORMATION_SCHEMA_PROCEDURE_INFO_TABLE_ID: u32 = 34;
 /// id for information_schema.region_statistics
 pub const INFORMATION_SCHEMA_REGION_STATISTICS_TABLE_ID: u32 = 35;

-/// ----- End of information_schema tables -----
+// ----- End of information_schema tables -----

 /// ----- Begin of pg_catalog tables -----
 pub const PG_CATALOG_PG_CLASS_TABLE_ID: u32 = 256;
 pub const PG_CATALOG_PG_TYPE_TABLE_ID: u32 = 257;
 pub const PG_CATALOG_PG_NAMESPACE_TABLE_ID: u32 = 258;

-/// ----- End of pg_catalog tables -----
+// ----- End of pg_catalog tables -----
+
 pub const MITO_ENGINE: &str = "mito";
 pub const MITO2_ENGINE: &str = "mito2";
 pub const METRIC_ENGINE: &str = "metric";
--- a/src/common/function/Cargo.toml
+++ b/src/common/function/Cargo.toml
@@ -9,7 +9,7 @@ workspace = true

 [features]
 default = ["geo"]
-geo = ["geohash", "h3o", "s2"]
+geo = ["geohash", "h3o", "s2", "wkt", "geo-types", "dep:geo"]

 [dependencies]
 api.workspace = true
@@ -28,6 +28,8 @@ common-version.workspace = true
 datafusion.workspace = true
 datatypes.workspace = true
 derive_more = { version = "1", default-features = false, features = ["display"] }
+geo = { version = "0.29", optional = true }
+geo-types = { version = "0.7", optional = true }
 geohash = { version = "0.13", optional = true }
 h3o = { version = "0.6", optional = true }
 jsonb.workspace = true
@@ -44,6 +46,7 @@ sql.workspace = true
 statrs = "0.16"
 store-api.workspace = true
 table.workspace = true
+wkt = { version = "0.11", optional = true }

 [dev-dependencies]
 ron = "0.7"
--- a/src/common/function/src/scalars/geo.rs
+++ b/src/common/function/src/scalars/geo.rs
@@ -17,7 +17,10 @@ pub(crate) mod encoding;
 mod geohash;
 mod h3;
 mod helpers;
+mod measure;
+mod relation;
 mod s2;
+mod wkt;

 use crate::function_registry::FunctionRegistry;

@@ -48,6 +51,7 @@ impl GeoFunctions {
        registry.register(Arc::new(h3::H3CellToChildrenSize));
        registry.register(Arc::new(h3::H3CellToChildPos));
        registry.register(Arc::new(h3::H3ChildPosToCell));
+        registry.register(Arc::new(h3::H3CellContains));

        // h3 grid traversal
        registry.register(Arc::new(h3::H3GridDisk));
@@ -55,10 +59,27 @@ impl GeoFunctions {
        registry.register(Arc::new(h3::H3GridDistance));
        registry.register(Arc::new(h3::H3GridPathCells));

+        // h3 measurement
+        registry.register(Arc::new(h3::H3CellDistanceSphereKm));
+        registry.register(Arc::new(h3::H3CellDistanceEuclideanDegree));
+
        // s2
        registry.register(Arc::new(s2::S2LatLngToCell));
        registry.register(Arc::new(s2::S2CellLevel));
        registry.register(Arc::new(s2::S2CellToToken));
        registry.register(Arc::new(s2::S2CellParent));
+
+        // spatial data type
+        registry.register(Arc::new(wkt::LatLngToPointWkt));
+
+        // spatial relation
+        registry.register(Arc::new(relation::STContains));
+        registry.register(Arc::new(relation::STWithin));
+        registry.register(Arc::new(relation::STIntersects));
+
+        // spatial measure
+        registry.register(Arc::new(measure::STDistance));
+        registry.register(Arc::new(measure::STDistanceSphere));
+        registry.register(Arc::new(measure::STArea));
    }
 }
--- a/src/common/function/src/scalars/geo/h3.rs
+++ b/src/common/function/src/scalars/geo/h3.rs
@@ -23,8 +23,8 @@ use datatypes::prelude::ConcreteDataType;
 use datatypes::scalars::{Scalar, ScalarVectorBuilder};
 use datatypes::value::{ListValue, Value};
 use datatypes::vectors::{
-    BooleanVectorBuilder, Int32VectorBuilder, ListVectorBuilder, MutableVector,
-    StringVectorBuilder, UInt64VectorBuilder, UInt8VectorBuilder, VectorRef,
+    BooleanVectorBuilder, Float64VectorBuilder, Int32VectorBuilder, ListVectorBuilder,
+    MutableVector, StringVectorBuilder, UInt64VectorBuilder, UInt8VectorBuilder, VectorRef,
 };
 use derive_more::Display;
 use h3o::{CellIndex, LatLng, Resolution};
@@ -38,6 +38,7 @@ static CELL_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
    vec![
        ConcreteDataType::int64_datatype(),
        ConcreteDataType::uint64_datatype(),
+        ConcreteDataType::string_datatype(),
    ]
 });

@@ -952,6 +953,181 @@ impl Function for H3GridPathCells {
    }
 }

+/// Tests if cells contains given cells
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct H3CellContains;
+
+impl Function for H3CellContains {
+    fn name(&self) -> &str {
+        "h3_cells_contains"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        let multi_cell_types = vec![
+            ConcreteDataType::list_datatype(ConcreteDataType::int64_datatype()),
+            ConcreteDataType::list_datatype(ConcreteDataType::uint64_datatype()),
+            ConcreteDataType::list_datatype(ConcreteDataType::string_datatype()),
+            ConcreteDataType::string_datatype(),
+        ];
+
+        let mut signatures = Vec::with_capacity(multi_cell_types.len() * CELL_TYPES.len());
+        for multi_cell_type in &multi_cell_types {
+            for cell_type in CELL_TYPES.as_slice() {
+                signatures.push(TypeSignature::Exact(vec![
+                    multi_cell_type.clone(),
+                    cell_type.clone(),
+                ]));
+            }
+        }
+
+        Signature::one_of(signatures, Volatility::Stable)
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let cells_vec = &columns[0];
+        let cell_this_vec = &columns[1];
+
+        let size = cell_this_vec.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let mut result = None;
+            if let (cells, Some(cell_this)) = (
+                cells_from_value(cells_vec.get(i))?,
+                cell_from_value(cell_this_vec.get(i))?,
+            ) {
+                result = Some(false);
+
+                for cell_that in cells.iter() {
+                    // get cell resolution, and find cell_this's parent at
+                    //  this solution, test if cell_that equals the parent
+                    let resolution = cell_that.resolution();
+                    if let Some(cell_this_parent) = cell_this.parent(resolution) {
+                        if cell_this_parent == *cell_that {
+                            result = Some(true);
+                            break;
+                        }
+                    }
+                }
+            }
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Get WGS84 great circle distance of two cell centroid
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct H3CellDistanceSphereKm;
+
+impl Function for H3CellDistanceSphereKm {
+    fn name(&self) -> &str {
+        "h3_distance_sphere_km"
+    }
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        signature_of_double_cells()
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let cell_this_vec = &columns[0];
+        let cell_that_vec = &columns[1];
+        let size = cell_this_vec.len();
+
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let result = match (
+                cell_from_value(cell_this_vec.get(i))?,
+                cell_from_value(cell_that_vec.get(i))?,
+            ) {
+                (Some(cell_this), Some(cell_that)) => {
+                    let centroid_this = LatLng::from(cell_this);
+                    let centroid_that = LatLng::from(cell_that);
+
+                    Some(centroid_this.distance_km(centroid_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Get Euclidean distance of two cell centroid
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct H3CellDistanceEuclideanDegree;
+
+impl H3CellDistanceEuclideanDegree {
+    fn distance(centroid_this: LatLng, centroid_that: LatLng) -> f64 {
+        ((centroid_this.lat() - centroid_that.lat()).powi(2)
+            + (centroid_this.lng() - centroid_that.lng()).powi(2))
+        .sqrt()
+    }
+}
+
+impl Function for H3CellDistanceEuclideanDegree {
+    fn name(&self) -> &str {
+        "h3_distance_degree"
+    }
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        signature_of_double_cells()
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let cell_this_vec = &columns[0];
+        let cell_that_vec = &columns[1];
+        let size = cell_this_vec.len();
+
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let result = match (
+                cell_from_value(cell_this_vec.get(i))?,
+                cell_from_value(cell_that_vec.get(i))?,
+            ) {
+                (Some(cell_this), Some(cell_that)) => {
+                    let centroid_this = LatLng::from(cell_this);
+                    let centroid_that = LatLng::from(cell_that);
+
+                    let dist = Self::distance(centroid_this, centroid_that);
+                    Some(dist)
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
 fn value_to_resolution(v: Value) -> Result<Resolution> {
    let r = match v {
        Value::Int8(v) => v as u8,
@@ -1073,7 +1249,126 @@ fn cell_from_value(v: Value) -> Result<Option<CellIndex>> {
                })
                .context(error::ExecuteSnafu)?,
        ),
+        Value::String(s) => Some(
+            CellIndex::from_str(s.as_utf8())
+                .map_err(|e| {
+                    BoxedError::new(PlainError::new(
+                        format!("H3 error: {}", e),
+                        StatusCode::EngineExecuteQuery,
+                    ))
+                })
+                .context(error::ExecuteSnafu)?,
+        ),
        _ => None,
    };
    Ok(cell)
 }
+
+/// extract cell array from all possible types including:
+/// - int64 list
+/// - uint64 list
+/// - string list
+/// - comma-separated string
+fn cells_from_value(v: Value) -> Result<Vec<CellIndex>> {
+    match v {
+        Value::List(list) => match list.datatype() {
+            ConcreteDataType::Int64(_) => list
+                .items()
+                .iter()
+                .map(|v| {
+                    if let Value::Int64(v) = v {
+                        CellIndex::try_from(*v as u64)
+                            .map_err(|e| {
+                                BoxedError::new(PlainError::new(
+                                    format!("H3 error: {}", e),
+                                    StatusCode::EngineExecuteQuery,
+                                ))
+                            })
+                            .context(error::ExecuteSnafu)
+                    } else {
+                        Err(BoxedError::new(PlainError::new(
+                            "Invalid data type in array".to_string(),
+                            StatusCode::EngineExecuteQuery,
+                        )))
+                        .context(error::ExecuteSnafu)
+                    }
+                })
+                .collect::<Result<Vec<CellIndex>>>(),
+            ConcreteDataType::UInt64(_) => list
+                .items()
+                .iter()
+                .map(|v| {
+                    if let Value::UInt64(v) = v {
+                        CellIndex::try_from(*v)
+                            .map_err(|e| {
+                                BoxedError::new(PlainError::new(
+                                    format!("H3 error: {}", e),
+                                    StatusCode::EngineExecuteQuery,
+                                ))
+                            })
+                            .context(error::ExecuteSnafu)
+                    } else {
+                        Err(BoxedError::new(PlainError::new(
+                            "Invalid data type in array".to_string(),
+                            StatusCode::EngineExecuteQuery,
+                        )))
+                        .context(error::ExecuteSnafu)
+                    }
+                })
+                .collect::<Result<Vec<CellIndex>>>(),
+            ConcreteDataType::String(_) => list
+                .items()
+                .iter()
+                .map(|v| {
+                    if let Value::String(v) = v {
+                        CellIndex::from_str(v.as_utf8().trim())
+                            .map_err(|e| {
+                                BoxedError::new(PlainError::new(
+                                    format!("H3 error: {}", e),
+                                    StatusCode::EngineExecuteQuery,
+                                ))
+                            })
+                            .context(error::ExecuteSnafu)
+                    } else {
+                        Err(BoxedError::new(PlainError::new(
+                            "Invalid data type in array".to_string(),
+                            StatusCode::EngineExecuteQuery,
+                        )))
+                        .context(error::ExecuteSnafu)
+                    }
+                })
+                .collect::<Result<Vec<CellIndex>>>(),
+            _ => Ok(vec![]),
+        },
+        Value::String(csv) => {
+            let str_seq = csv.as_utf8().split(',');
+            str_seq
+                .map(|v| {
+                    CellIndex::from_str(v.trim())
+                        .map_err(|e| {
+                            BoxedError::new(PlainError::new(
+                                format!("H3 error: {}", e),
+                                StatusCode::EngineExecuteQuery,
+                            ))
+                        })
+                        .context(error::ExecuteSnafu)
+                })
+                .collect::<Result<Vec<CellIndex>>>()
+        }
+        _ => Ok(vec![]),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_h3_euclidean_distance() {
+        let point_this = LatLng::new(42.3521, -72.1235).expect("incorrect lat lng");
+        let point_that = LatLng::new(42.45, -72.1260).expect("incorrect lat lng");
+
+        let dist = H3CellDistanceEuclideanDegree::distance(point_this, point_that);
+        assert_eq!(dist, 0.09793191512474639);
+    }
+}
--- a/src/common/function/src/scalars/geo/measure.rs
+++ b/src/common/function/src/scalars/geo/measure.rs
@@ -0,0 +1,195 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_error::ext::{BoxedError, PlainError};
+use common_error::status_code::StatusCode;
+use common_query::error::{self, Result};
+use common_query::prelude::{Signature, TypeSignature};
+use datafusion::logical_expr::Volatility;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{Float64VectorBuilder, MutableVector, VectorRef};
+use derive_more::Display;
+use geo::algorithm::line_measures::metric_spaces::Euclidean;
+use geo::{Area, Distance, Haversine};
+use geo_types::Geometry;
+use snafu::ResultExt;
+
+use super::helpers::{ensure_columns_len, ensure_columns_n};
+use super::wkt::parse_wkt;
+use crate::function::{Function, FunctionContext};
+
+/// Return WGS84(SRID: 4326) euclidean distance between two geometry object, in degree
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STDistance;
+
+impl Function for STDistance {
+    fn name(&self) -> &str {
+        "st_distance"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    Some(Euclidean::distance(&geom_this, &geom_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Return great circle distance between two geometry object, in meters
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STDistanceSphere;
+
+impl Function for STDistanceSphere {
+    fn name(&self) -> &str {
+        "st_distance_sphere_m"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    match (geom_this, geom_that) {
+                        (Geometry::Point(this), Geometry::Point(that)) => {
+                            Some(Haversine::distance(this, that))
+                        }
+                        _ => {
+                            Err(BoxedError::new(PlainError::new(
+                                "Great circle distance between non-point objects are not supported for now.".to_string(),
+                                StatusCode::Unsupported,
+                            ))).context(error::ExecuteSnafu)?
+                        }
+                    }
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Return area of given geometry object
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STArea;
+
+impl Function for STArea {
+    fn name(&self) -> &str {
+        "st_area"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::float64_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 1);
+
+        let wkt_vec = &columns[0];
+
+        let size = wkt_vec.len();
+        let mut results = Float64VectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt = wkt_vec.get(i).as_string();
+
+            let result = if let Some(wkt) = wkt {
+                let geom = parse_wkt(&wkt)?;
+                Some(geom.unsigned_area())
+            } else {
+                None
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
--- a/src/common/function/src/scalars/geo/relation.rs
+++ b/src/common/function/src/scalars/geo/relation.rs
@@ -0,0 +1,190 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_query::error::Result;
+use common_query::prelude::{Signature, TypeSignature};
+use datafusion::logical_expr::Volatility;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{BooleanVectorBuilder, MutableVector, VectorRef};
+use derive_more::Display;
+use geo::algorithm::contains::Contains;
+use geo::algorithm::intersects::Intersects;
+use geo::algorithm::within::Within;
+
+use super::helpers::{ensure_columns_len, ensure_columns_n};
+use super::wkt::parse_wkt;
+use crate::function::{Function, FunctionContext};
+
+/// Test if spatial relationship: contains
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STContains;
+
+impl Function for STContains {
+    fn name(&self) -> &str {
+        "st_contains"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    Some(geom_this.contains(&geom_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Test if spatial relationship: within
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STWithin;
+
+impl Function for STWithin {
+    fn name(&self) -> &str {
+        "st_within"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    Some(geom_this.is_within(&geom_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+/// Test if spatial relationship: within
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct STIntersects;
+
+impl Function for STIntersects {
+    fn name(&self) -> &str {
+        "st_intersects"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::new(
+            TypeSignature::Exact(vec![
+                ConcreteDataType::string_datatype(),
+                ConcreteDataType::string_datatype(),
+            ]),
+            Volatility::Stable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let wkt_this_vec = &columns[0];
+        let wkt_that_vec = &columns[1];
+
+        let size = wkt_this_vec.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let wkt_this = wkt_this_vec.get(i).as_string();
+            let wkt_that = wkt_that_vec.get(i).as_string();
+
+            let result = match (wkt_this, wkt_that) {
+                (Some(wkt_this), Some(wkt_that)) => {
+                    let geom_this = parse_wkt(&wkt_this)?;
+                    let geom_that = parse_wkt(&wkt_that)?;
+
+                    Some(geom_this.intersects(&geom_that))
+                }
+                _ => None,
+            };
+
+            results.push(result);
+        }
+
+        Ok(results.to_vector())
+    }
+}
--- a/src/common/function/src/scalars/geo/wkt.rs
+++ b/src/common/function/src/scalars/geo/wkt.rs
@@ -0,0 +1,100 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use common_error::ext::{BoxedError, PlainError};
+use common_error::status_code::StatusCode;
+use common_query::error::{self, Result};
+use common_query::prelude::{Signature, TypeSignature};
+use datafusion::logical_expr::Volatility;
+use datatypes::prelude::ConcreteDataType;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{MutableVector, StringVectorBuilder, VectorRef};
+use derive_more::Display;
+use geo_types::{Geometry, Point};
+use once_cell::sync::Lazy;
+use snafu::ResultExt;
+use wkt::{ToWkt, TryFromWkt};
+
+use super::helpers::{ensure_columns_len, ensure_columns_n};
+use crate::function::{Function, FunctionContext};
+
+static COORDINATE_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
+    vec![
+        ConcreteDataType::float32_datatype(),
+        ConcreteDataType::float64_datatype(),
+    ]
+});
+
+/// Return WGS84(SRID: 4326) euclidean distance between two geometry object, in degree
+#[derive(Clone, Debug, Default, Display)]
+#[display("{}", self.name())]
+pub struct LatLngToPointWkt;
+
+impl Function for LatLngToPointWkt {
+    fn name(&self) -> &str {
+        "wkt_point_from_latlng"
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::string_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        let mut signatures = Vec::new();
+        for coord_type in COORDINATE_TYPES.as_slice() {
+            signatures.push(TypeSignature::Exact(vec![
+                // latitude
+                coord_type.clone(),
+                // longitude
+                coord_type.clone(),
+            ]));
+        }
+        Signature::one_of(signatures, Volatility::Stable)
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure_columns_n!(columns, 2);
+
+        let lat_vec = &columns[0];
+        let lng_vec = &columns[1];
+
+        let size = lat_vec.len();
+        let mut results = StringVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let lat = lat_vec.get(i).as_f64_lossy();
+            let lng = lng_vec.get(i).as_f64_lossy();
+
+            let result = match (lat, lng) {
+                (Some(lat), Some(lng)) => Some(Point::new(lng, lat).wkt_string()),
+                _ => None,
+            };
+
+            results.push(result.as_deref());
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+pub(super) fn parse_wkt(s: &str) -> Result<Geometry> {
+    Geometry::try_from_wkt_str(s)
+        .map_err(|e| {
+            BoxedError::new(PlainError::new(
+                format!("Fail to parse WKT: {}", e),
+                StatusCode::EngineExecuteQuery,
+            ))
+        })
+        .context(error::ExecuteSnafu)
+}
--- a/src/common/function/src/scalars/json.rs
+++ b/src/common/function/src/scalars/json.rs
@@ -16,6 +16,7 @@ use std::sync::Arc;
 mod json_get;
 mod json_is;
 mod json_path_exists;
+mod json_path_match;
 mod json_to_string;
 mod parse_json;

@@ -49,5 +50,6 @@ impl JsonFunction {
        registry.register(Arc::new(JsonIsObject));

        registry.register(Arc::new(json_path_exists::JsonPathExistsFunction));
+        registry.register(Arc::new(json_path_match::JsonPathMatchFunction));
    }
 }
--- a/src/common/function/src/scalars/json/json_path_exists.rs
+++ b/src/common/function/src/scalars/json/json_path_exists.rs
@@ -15,7 +15,7 @@
 use std::fmt::{self, Display};

 use common_query::error::{InvalidFuncArgsSnafu, Result, UnsupportedInputDataTypeSnafu};
-use common_query::prelude::Signature;
+use common_query::prelude::{Signature, TypeSignature};
 use datafusion::logical_expr::Volatility;
 use datatypes::data_type::ConcreteDataType;
 use datatypes::prelude::VectorRef;
@@ -41,10 +41,24 @@ impl Function for JsonPathExistsFunction {
    }

    fn signature(&self) -> Signature {
-        Signature::exact(
+        Signature::one_of(
            vec![
-                ConcreteDataType::json_datatype(),
-                ConcreteDataType::string_datatype(),
+                TypeSignature::Exact(vec![
+                    ConcreteDataType::json_datatype(),
+                    ConcreteDataType::string_datatype(),
+                ]),
+                TypeSignature::Exact(vec![
+                    ConcreteDataType::null_datatype(),
+                    ConcreteDataType::string_datatype(),
+                ]),
+                TypeSignature::Exact(vec![
+                    ConcreteDataType::json_datatype(),
+                    ConcreteDataType::null_datatype(),
+                ]),
+                TypeSignature::Exact(vec![
+                    ConcreteDataType::null_datatype(),
+                    ConcreteDataType::null_datatype(),
+                ]),
            ],
            Volatility::Immutable,
        )
@@ -64,25 +78,26 @@ impl Function for JsonPathExistsFunction {
        let paths = &columns[1];

        let size = jsons.len();
-        let datatype = jsons.data_type();
        let mut results = BooleanVectorBuilder::with_capacity(size);

-        match datatype {
-            // JSON data type uses binary vector
-            ConcreteDataType::Binary(_) => {
+        match (jsons.data_type(), paths.data_type()) {
+            (ConcreteDataType::Binary(_), ConcreteDataType::String(_)) => {
                for i in 0..size {
-                    let json = jsons.get_ref(i);
-                    let path = paths.get_ref(i);
-
-                    let json = json.as_binary();
-                    let path = path.as_string();
-                    let result = match (json, path) {
+                    let result = match (jsons.get_ref(i).as_binary(), paths.get_ref(i).as_string())
+                    {
                        (Ok(Some(json)), Ok(Some(path))) => {
-                            let json_path = jsonb::jsonpath::parse_json_path(path.as_bytes());
-                            match json_path {
-                                Ok(json_path) => jsonb::path_exists(json, json_path).ok(),
-                                Err(_) => None,
-                            }
+                            // Get `JsonPath`.
+                            let json_path = match jsonb::jsonpath::parse_json_path(path.as_bytes())
+                            {
+                                Ok(json_path) => json_path,
+                                Err(_) => {
+                                    return InvalidFuncArgsSnafu {
+                                        err_msg: format!("Illegal json path: {:?}", path),
+                                    }
+                                    .fail();
+                                }
+                            };
+                            jsonb::path_exists(json, json_path).ok()
                        }
                        _ => None,
                    };
@@ -90,6 +105,12 @@ impl Function for JsonPathExistsFunction {
                    results.push(result);
                }
            }
+
+            // Any null args existence causes the result to be NULL.
+            (ConcreteDataType::Null(_), ConcreteDataType::String(_)) => results.push_nulls(size),
+            (ConcreteDataType::Binary(_), ConcreteDataType::Null(_)) => results.push_nulls(size),
+            (ConcreteDataType::Null(_), ConcreteDataType::Null(_)) => results.push_nulls(size),
+
            _ => {
                return UnsupportedInputDataTypeSnafu {
                    function: NAME,
@@ -114,8 +135,8 @@ mod tests {
    use std::sync::Arc;

    use common_query::prelude::TypeSignature;
-    use datatypes::scalars::ScalarVector;
-    use datatypes::vectors::{BinaryVector, StringVector};
+    use datatypes::prelude::ScalarVector;
+    use datatypes::vectors::{BinaryVector, NullVector, StringVector};

    use super::*;

@@ -133,9 +154,27 @@ mod tests {

        assert!(matches!(json_path_exists.signature(),
                         Signature {
-                             type_signature: TypeSignature::Exact(valid_types),
+                             type_signature: TypeSignature::OneOf(valid_types),
                             volatility: Volatility::Immutable
-                         } if  valid_types == vec![ConcreteDataType::json_datatype(), ConcreteDataType::string_datatype()]
+                         } if valid_types ==
+            vec![
+                TypeSignature::Exact(vec![
+                    ConcreteDataType::json_datatype(),
+                    ConcreteDataType::string_datatype(),
+                ]),
+                TypeSignature::Exact(vec![
+                    ConcreteDataType::null_datatype(),
+                    ConcreteDataType::string_datatype(),
+                ]),
+                TypeSignature::Exact(vec![
+                    ConcreteDataType::json_datatype(),
+                    ConcreteDataType::null_datatype(),
+                ]),
+                TypeSignature::Exact(vec![
+                    ConcreteDataType::null_datatype(),
+                    ConcreteDataType::null_datatype(),
+                ]),
+            ],
        ));

        let json_strings = [
@@ -143,9 +182,15 @@ mod tests {
            r#"{"a": 4, "b": {"c": 6}, "c": 6}"#,
            r#"{"a": 7, "b": 8, "c": {"a": 7}}"#,
            r#"{"a": 7, "b": 8, "c": {"a": 7}}"#,
+            r#"[1, 2, 3]"#,
+            r#"null"#,
+            r#"{"a": 7, "b": 8, "c": {"a": 7}}"#,
+            r#"null"#,
        ];
-        let paths = vec!["$.a.b.c", "$.b", "$.c.a", ".d"];
-        let results = [false, true, true, false];
+        let paths = vec![
+            "$.a.b.c", "$.b", "$.c.a", ".d", "$[0]", "$.a", "null", "null",
+        ];
+        let expected = [false, true, true, false, true, false, false, false];

        let jsonbs = json_strings
            .iter()
@@ -162,11 +207,44 @@ mod tests {
            .eval(FunctionContext::default(), &args)
            .unwrap();

-        assert_eq!(4, vector.len());
-        for (i, gt) in results.iter().enumerate() {
+        // Test for non-nulls.
+        assert_eq!(8, vector.len());
+        for (i, real) in expected.iter().enumerate() {
            let result = vector.get_ref(i);
-            let result = result.as_boolean().unwrap().unwrap();
-            assert_eq!(*gt, result);
+            assert!(!result.is_null());
+            let val = result.as_boolean().unwrap().unwrap();
+            assert_eq!(val, *real);
        }
+
+        // Test for path error.
+        let json_bytes = jsonb::parse_value("{}".as_bytes()).unwrap().to_vec();
+        let json = BinaryVector::from_vec(vec![json_bytes]);
+        let illegal_path = StringVector::from_vec(vec!["$..a"]);
+
+        let args: Vec<VectorRef> = vec![Arc::new(json), Arc::new(illegal_path)];
+        let err = json_path_exists.eval(FunctionContext::default(), &args);
+        assert!(err.is_err());
+
+        // Test for nulls.
+        let json_bytes = jsonb::parse_value("{}".as_bytes()).unwrap().to_vec();
+        let json = BinaryVector::from_vec(vec![json_bytes]);
+        let null_json = NullVector::new(1);
+
+        let path = StringVector::from_vec(vec!["$.a"]);
+        let null_path = NullVector::new(1);
+
+        let args: Vec<VectorRef> = vec![Arc::new(null_json), Arc::new(path)];
+        let result1 = json_path_exists
+            .eval(FunctionContext::default(), &args)
+            .unwrap();
+        let args: Vec<VectorRef> = vec![Arc::new(json), Arc::new(null_path)];
+        let result2 = json_path_exists
+            .eval(FunctionContext::default(), &args)
+            .unwrap();
+
+        assert_eq!(result1.len(), 1);
+        assert!(result1.get_ref(0).is_null());
+        assert_eq!(result2.len(), 1);
+        assert!(result2.get_ref(0).is_null());
    }
 }
--- a/src/common/function/src/scalars/json/json_path_match.rs
+++ b/src/common/function/src/scalars/json/json_path_match.rs
@@ -0,0 +1,202 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt::{self, Display};
+
+use common_query::error::{InvalidFuncArgsSnafu, Result, UnsupportedInputDataTypeSnafu};
+use common_query::prelude::Signature;
+use datafusion::logical_expr::Volatility;
+use datatypes::data_type::ConcreteDataType;
+use datatypes::prelude::VectorRef;
+use datatypes::scalars::ScalarVectorBuilder;
+use datatypes::vectors::{BooleanVectorBuilder, MutableVector};
+use snafu::ensure;
+
+use crate::function::{Function, FunctionContext};
+
+/// Check if the given JSON data match the given JSON path's predicate.
+#[derive(Clone, Debug, Default)]
+pub struct JsonPathMatchFunction;
+
+const NAME: &str = "json_path_match";
+
+impl Function for JsonPathMatchFunction {
+    fn name(&self) -> &str {
+        NAME
+    }
+
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::boolean_datatype())
+    }
+
+    fn signature(&self) -> Signature {
+        Signature::exact(
+            vec![
+                ConcreteDataType::json_datatype(),
+                ConcreteDataType::string_datatype(),
+            ],
+            Volatility::Immutable,
+        )
+    }
+
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+        ensure!(
+            columns.len() == 2,
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect exactly two, have: {}",
+                    columns.len()
+                ),
+            }
+        );
+        let jsons = &columns[0];
+        let paths = &columns[1];
+
+        let size = jsons.len();
+        let mut results = BooleanVectorBuilder::with_capacity(size);
+
+        for i in 0..size {
+            let json = jsons.get_ref(i);
+            let path = paths.get_ref(i);
+
+            match json.data_type() {
+                // JSON data type uses binary vector
+                ConcreteDataType::Binary(_) => {
+                    let json = json.as_binary();
+                    let path = path.as_string();
+                    let result = match (json, path) {
+                        (Ok(Some(json)), Ok(Some(path))) => {
+                            if !jsonb::is_null(json) {
+                                let json_path = jsonb::jsonpath::parse_json_path(path.as_bytes());
+                                match json_path {
+                                    Ok(json_path) => jsonb::path_match(json, json_path).ok(),
+                                    Err(_) => None,
+                                }
+                            } else {
+                                None
+                            }
+                        }
+                        _ => None,
+                    };
+
+                    results.push(result);
+                }
+
+                _ => {
+                    return UnsupportedInputDataTypeSnafu {
+                        function: NAME,
+                        datatypes: columns.iter().map(|c| c.data_type()).collect::<Vec<_>>(),
+                    }
+                    .fail();
+                }
+            }
+        }
+
+        Ok(results.to_vector())
+    }
+}
+
+impl Display for JsonPathMatchFunction {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(f, "JSON_PATH_MATCH")
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_query::prelude::TypeSignature;
+    use datatypes::vectors::{BinaryVector, StringVector};
+
+    use super::*;
+
+    #[test]
+    fn test_json_path_match_function() {
+        let json_path_match = JsonPathMatchFunction;
+
+        assert_eq!("json_path_match", json_path_match.name());
+        assert_eq!(
+            ConcreteDataType::boolean_datatype(),
+            json_path_match
+                .return_type(&[ConcreteDataType::json_datatype()])
+                .unwrap()
+        );
+
+        assert!(matches!(json_path_match.signature(),
+                         Signature {
+                             type_signature: TypeSignature::Exact(valid_types),
+                             volatility: Volatility::Immutable
+                         } if valid_types == vec![ConcreteDataType::json_datatype(), ConcreteDataType::string_datatype()],
+        ));
+
+        let json_strings = [
+            Some(r#"{"a": {"b": 2}, "b": 2, "c": 3}"#.to_string()),
+            Some(r#"{"a": 1, "b": [1,2,3]}"#.to_string()),
+            Some(r#"{"a": 1 ,"b": [1,2,3]}"#.to_string()),
+            Some(r#"[1,2,3]"#.to_string()),
+            Some(r#"{"a":1,"b":[1,2,3]}"#.to_string()),
+            Some(r#"null"#.to_string()),
+            Some(r#"null"#.to_string()),
+        ];
+
+        let paths = vec![
+            Some("$.a.b == 2".to_string()),
+            Some("$.b[1 to last] >= 2".to_string()),
+            Some("$.c > 0".to_string()),
+            Some("$[0 to last] > 0".to_string()),
+            Some(r#"null"#.to_string()),
+            Some("$.c > 0".to_string()),
+            Some(r#"null"#.to_string()),
+        ];
+
+        let results = [
+            Some(true),
+            Some(true),
+            Some(false),
+            Some(true),
+            None,
+            None,
+            None,
+        ];
+
+        let jsonbs = json_strings
+            .into_iter()
+            .map(|s| s.map(|json| jsonb::parse_value(json.as_bytes()).unwrap().to_vec()))
+            .collect::<Vec<_>>();
+
+        let json_vector = BinaryVector::from(jsonbs);
+        let path_vector = StringVector::from(paths);
+        let args: Vec<VectorRef> = vec![Arc::new(json_vector), Arc::new(path_vector)];
+        let vector = json_path_match
+            .eval(FunctionContext::default(), &args)
+            .unwrap();
+
+        assert_eq!(7, vector.len());
+        for (i, expected) in results.iter().enumerate() {
+            let result = vector.get_ref(i);
+
+            match expected {
+                Some(expected_value) => {
+                    assert!(!result.is_null());
+                    let result_value = result.as_boolean().unwrap().unwrap();
+                    assert_eq!(*expected_value, result_value);
+                }
+                None => {
+                    assert!(result.is_null());
+                }
+            }
+        }
+    }
+}
--- a/src/common/greptimedb-telemetry/src/lib.rs
+++ b/src/common/greptimedb-telemetry/src/lib.rs
@@ -199,6 +199,7 @@ pub fn default_get_uuid(working_home: &Option<String>) -> Option<String> {
 }

 /// Report version info to GreptimeDB.
+///
 /// We do not collect any identity-sensitive information.
 /// This task is scheduled to run every 30 minutes.
 /// The task will be disabled default. It can be enabled by setting the build feature `greptimedb-telemetry`
@@ -324,7 +325,7 @@ mod tests {
            });
            let addr = ([127, 0, 0, 1], port).into();

-            let server = Server::bind(&addr).serve(make_svc);
+            let server = Server::try_bind(&addr).unwrap().serve(make_svc);
            let graceful = server.with_graceful_shutdown(async {
                rx.await.ok();
            });
--- a/src/common/grpc-expr/Cargo.toml
+++ b/src/common/grpc-expr/Cargo.toml
@@ -18,6 +18,7 @@ common-time.workspace = true
 datatypes.workspace = true
 prost.workspace = true
 snafu.workspace = true
+store-api.workspace = true
 table.workspace = true

 [dev-dependencies]
--- a/src/common/grpc-expr/src/alter.rs
+++ b/src/common/grpc-expr/src/alter.rs
@@ -22,12 +22,13 @@ use api::v1::{
 use common_query::AddColumnLocation;
 use datatypes::schema::{ColumnSchema, RawSchema};
 use snafu::{ensure, OptionExt, ResultExt};
+use store_api::region_request::ChangeOption;
 use table::metadata::TableId;
 use table::requests::{AddColumnRequest, AlterKind, AlterTableRequest, ChangeColumnTypeRequest};

 use crate::error::{
-    InvalidColumnDefSnafu, MissingFieldSnafu, MissingTimestampColumnSnafu, Result,
-    UnknownLocationTypeSnafu,
+    InvalidChangeTableOptionRequestSnafu, InvalidColumnDefSnafu, MissingFieldSnafu,
+    MissingTimestampColumnSnafu, Result, UnknownLocationTypeSnafu,
 };

 const LOCATION_TYPE_FIRST: i32 = LocationType::First as i32;
@@ -92,6 +93,15 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterExpr) -> Result<Alter
        Kind::RenameTable(RenameTable { new_table_name }) => {
            AlterKind::RenameTable { new_table_name }
        }
+        Kind::ChangeTableOptions(api::v1::ChangeTableOptions {
+            change_table_options,
+        }) => AlterKind::ChangeTableOptions {
+            options: change_table_options
+                .iter()
+                .map(ChangeOption::try_from)
+                .collect::<std::result::Result<Vec<_>, _>>()
+                .context(InvalidChangeTableOptionRequestSnafu)?,
+        },
    };

    let request = AlterTableRequest {
--- a/src/common/grpc-expr/src/error.rs
+++ b/src/common/grpc-expr/src/error.rs
@@ -19,6 +19,7 @@ use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
 use snafu::{Location, Snafu};
+use store_api::metadata::MetadataError;

 #[derive(Snafu)]
 #[snafu(visibility(pub))]
@@ -118,6 +119,12 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
+
+    #[snafu(display("Invalid change table option request"))]
+    InvalidChangeTableOptionRequest {
+        #[snafu(source)]
+        error: MetadataError,
+    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -141,6 +148,7 @@ impl ErrorExt for Error {
            Error::UnknownColumnDataType { .. } | Error::InvalidFulltextColumnType { .. } => {
                StatusCode::InvalidArguments
            }
+            Error::InvalidChangeTableOptionRequest { .. } => StatusCode::InvalidArguments,
        }
    }

--- a/src/common/grpc-expr/src/util.rs
+++ b/src/common/grpc-expr/src/util.rs
@@ -14,10 +14,11 @@

 use std::collections::HashSet;

+use api::v1::column_data_type_extension::TypeExt;
 use api::v1::column_def::contains_fulltext;
 use api::v1::{
    AddColumn, AddColumns, Column, ColumnDataType, ColumnDataTypeExtension, ColumnDef,
-    ColumnOptions, ColumnSchema, CreateTableExpr, SemanticType,
+    ColumnOptions, ColumnSchema, CreateTableExpr, JsonTypeExtension, SemanticType,
 };
 use datatypes::schema::Schema;
 use snafu::{ensure, OptionExt, ResultExt};
@@ -25,8 +26,9 @@ use table::metadata::TableId;
 use table::table_reference::TableReference;

 use crate::error::{
-    DuplicatedColumnNameSnafu, DuplicatedTimestampColumnSnafu, InvalidFulltextColumnTypeSnafu,
-    MissingTimestampColumnSnafu, Result, UnknownColumnDataTypeSnafu,
+    self, DuplicatedColumnNameSnafu, DuplicatedTimestampColumnSnafu,
+    InvalidFulltextColumnTypeSnafu, MissingTimestampColumnSnafu, Result,
+    UnknownColumnDataTypeSnafu,
 };
 pub struct ColumnExpr<'a> {
    pub column_name: &'a str,
@@ -72,6 +74,28 @@ impl<'a> From<&'a ColumnSchema> for ColumnExpr<'a> {
    }
 }

+fn infer_column_datatype(
+    datatype: i32,
+    datatype_extension: &Option<ColumnDataTypeExtension>,
+) -> Result<ColumnDataType> {
+    let column_type =
+        ColumnDataType::try_from(datatype).context(UnknownColumnDataTypeSnafu { datatype })?;
+
+    if matches!(&column_type, ColumnDataType::Binary) {
+        if let Some(ext) = datatype_extension {
+            let type_ext = ext
+                .type_ext
+                .as_ref()
+                .context(error::MissingFieldSnafu { field: "type_ext" })?;
+            if *type_ext == TypeExt::JsonType(JsonTypeExtension::JsonBinary.into()) {
+                return Ok(ColumnDataType::Json);
+            }
+        }
+    }
+
+    Ok(column_type)
+}
+
 pub fn build_create_table_expr(
    table_id: Option<TableId>,
    table_name: &TableReference<'_>,
@@ -124,8 +148,7 @@ pub fn build_create_table_expr(
            _ => {}
        }

-        let column_type =
-            ColumnDataType::try_from(datatype).context(UnknownColumnDataTypeSnafu { datatype })?;
+        let column_type = infer_column_datatype(datatype, datatype_extension)?;

        ensure!(
            !contains_fulltext(options) || column_type == ColumnDataType::String,
--- a/src/common/macro/src/lib.rs
+++ b/src/common/macro/src/lib.rs
@@ -35,7 +35,9 @@ pub fn aggr_func_type_store_derive(input: TokenStream) -> TokenStream {
 }

 /// A struct can be used as a creator for aggregate function if it has been annotated with this
-/// attribute first. This attribute add a necessary field which is intended to store the input
+/// attribute first.
+///
+/// This attribute add a necessary field which is intended to store the input
 /// data's types to the struct.
 /// This attribute is expected to be used along with derive macro [AggrFuncTypeStore].
 #[proc_macro_attribute]
@@ -44,9 +46,10 @@ pub fn as_aggr_func_creator(args: TokenStream, input: TokenStream) -> TokenStrea
 }

 /// Attribute macro to convert an arithimetic function to a range function. The annotated function
-/// should accept servaral arrays as input and return a single value as output. This procedure
-/// macro can works on any number of input parameters. Return type can be either primitive type
-/// or wrapped in `Option`.
+/// should accept servaral arrays as input and return a single value as output.
+///
+/// This procedure macro can works on any number of input parameters. Return type can be either
+/// primitive type or wrapped in `Option`.
 ///
 /// # Example
 /// Take `count_over_time()` in PromQL as an example:
--- a/src/common/meta/Cargo.toml
+++ b/src/common/meta/Cargo.toml
@@ -60,7 +60,7 @@ table.workspace = true
 tokio.workspace = true
 tokio-postgres = { workspace = true, optional = true }
 tonic.workspace = true
-typetag = "0.2"
+typetag.workspace = true

 [dev-dependencies]
 chrono.workspace = true
--- a/src/common/meta/src/cluster.rs
+++ b/src/common/meta/src/cluster.rs
@@ -55,6 +55,7 @@ pub trait ClusterInfo {
 }

 /// The key of [NodeInfo] in the storage. The format is `__meta_cluster_node_info-{cluster_id}-{role}-{node_id}`.
+///
 /// This key cannot be used to describe the `Metasrv` because the `Metasrv` does not have
 /// a `cluster_id`, it serves multiple clusters.
 #[derive(Debug, Clone, Eq, Hash, PartialEq, Serialize, Deserialize)]
--- a/src/common/meta/src/datanode.rs
+++ b/src/common/meta/src/datanode.rs
@@ -78,17 +78,21 @@ pub struct RegionStat {
    /// The write capacity units during this period
    pub wcus: i64,
    /// Approximate bytes of this region
-    pub approximate_bytes: i64,
+    pub approximate_bytes: u64,
    /// The engine name.
    pub engine: String,
    /// The region role.
    pub role: RegionRole,
+    /// The number of rows
+    pub num_rows: u64,
    /// The size of the memtable in bytes.
    pub memtable_size: u64,
    /// The size of the manifest in bytes.
    pub manifest_size: u64,
-    /// The size of the SST files in bytes.
+    /// The size of the SST data files in bytes.
    pub sst_size: u64,
+    /// The size of the SST index files in bytes.
+    pub index_size: u64,
 }

 impl Stat {
@@ -178,12 +182,14 @@ impl From<&api::v1::meta::RegionStat> for RegionStat {
            id: RegionId::from_u64(value.region_id),
            rcus: value.rcus,
            wcus: value.wcus,
-            approximate_bytes: value.approximate_bytes,
+            approximate_bytes: value.approximate_bytes as u64,
            engine: value.engine.to_string(),
            role: RegionRole::from(value.role()),
+            num_rows: region_stat.num_rows,
            memtable_size: region_stat.memtable_size,
            manifest_size: region_stat.manifest_size,
            sst_size: region_stat.sst_size,
+            index_size: region_stat.index_size,
        }
    }
 }
--- a/src/common/meta/src/ddl/alter_logical_tables/update_metadata.rs
+++ b/src/common/meta/src/ddl/alter_logical_tables/update_metadata.rs
@@ -43,10 +43,10 @@ impl AlterLogicalTablesProcedure {
            &self.data.physical_columns,
        );

-        // Updates physical table's metadata
+        // Updates physical table's metadata, and we don't need to touch per-region settings.
        self.context
            .table_metadata_manager
-            .update_table_info(physical_table_info, new_raw_table_info)
+            .update_table_info(physical_table_info, None, new_raw_table_info)
            .await?;

        Ok(())
--- a/src/common/meta/src/ddl/alter_table.rs
+++ b/src/common/meta/src/ddl/alter_table.rs
@@ -43,10 +43,10 @@ use crate::ddl::DdlContext;
 use crate::error::{Error, Result};
 use crate::instruction::CacheIdent;
 use crate::key::table_info::TableInfoValue;
-use crate::key::DeserializedValueWithBytes;
+use crate::key::{DeserializedValueWithBytes, RegionDistribution};
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
 use crate::rpc::ddl::AlterTableTask;
-use crate::rpc::router::{find_leader_regions, find_leaders};
+use crate::rpc::router::{find_leader_regions, find_leaders, region_distribution};
 use crate::{metrics, ClusterId};

 /// The alter table procedure
@@ -101,6 +101,9 @@ impl AlterTableProcedure {
            .get_physical_table_route(table_id)
            .await?;

+        self.data.region_distribution =
+            Some(region_distribution(&physical_table_route.region_routes));
+
        let leaders = find_leaders(&physical_table_route.region_routes);
        let mut alter_region_tasks = Vec::with_capacity(leaders.len());

@@ -161,8 +164,14 @@ impl AlterTableProcedure {
            self.on_update_metadata_for_rename(new_table_name.to_string(), table_info_value)
                .await?;
        } else {
-            self.on_update_metadata_for_alter(new_info.into(), table_info_value)
-                .await?;
+            // region distribution is set in submit_alter_region_requests
+            let region_distribution = self.data.region_distribution.as_ref().unwrap().clone();
+            self.on_update_metadata_for_alter(
+                new_info.into(),
+                region_distribution,
+                table_info_value,
+            )
+            .await?;
        }

        info!("Updated table metadata for table {table_ref}, table_id: {table_id}");
@@ -271,6 +280,8 @@ pub struct AlterTableData {
    table_id: TableId,
    /// Table info value before alteration.
    table_info_value: Option<DeserializedValueWithBytes<TableInfoValue>>,
+    /// Region distribution for table in case we need to update region options.
+    region_distribution: Option<RegionDistribution>,
 }

 impl AlterTableData {
@@ -281,6 +292,7 @@ impl AlterTableData {
            table_id,
            cluster_id,
            table_info_value: None,
+            region_distribution: None,
        }
    }

--- a/src/common/meta/src/ddl/alter_table/region_request.rs
+++ b/src/common/meta/src/ddl/alter_table/region_request.rs
@@ -106,6 +106,7 @@ fn create_proto_alter_kind(
            })))
        }
        Kind::RenameTable(_) => Ok(None),
+        Kind::ChangeTableOptions(v) => Ok(Some(alter_request::Kind::ChangeTableOptions(v.clone()))),
    }
 }

--- a/src/common/meta/src/ddl/alter_table/update_metadata.rs
+++ b/src/common/meta/src/ddl/alter_table/update_metadata.rs
@@ -20,7 +20,7 @@ use table::requests::AlterKind;
 use crate::ddl::alter_table::AlterTableProcedure;
 use crate::error::{self, Result};
 use crate::key::table_info::TableInfoValue;
-use crate::key::DeserializedValueWithBytes;
+use crate::key::{DeserializedValueWithBytes, RegionDistribution};

 impl AlterTableProcedure {
    /// Builds new_meta
@@ -51,7 +51,9 @@ impl AlterTableProcedure {
            AlterKind::RenameTable { new_table_name } => {
                new_info.name = new_table_name.to_string();
            }
-            AlterKind::DropColumns { .. } | AlterKind::ChangeColumnTypes { .. } => {}
+            AlterKind::DropColumns { .. }
+            | AlterKind::ChangeColumnTypes { .. }
+            | AlterKind::ChangeTableOptions { .. } => {}
        }

        Ok(new_info)
@@ -75,11 +77,16 @@ impl AlterTableProcedure {
    pub(crate) async fn on_update_metadata_for_alter(
        &self,
        new_table_info: RawTableInfo,
+        region_distribution: RegionDistribution,
        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
    ) -> Result<()> {
        let table_metadata_manager = &self.context.table_metadata_manager;
        table_metadata_manager
-            .update_table_info(current_table_info_value, new_table_info)
+            .update_table_info(
+                current_table_info_value,
+                Some(region_distribution),
+                new_table_info,
+            )
            .await?;

        Ok(())
--- a/src/common/meta/src/ddl/create_logical_tables/update_metadata.rs
+++ b/src/common/meta/src/ddl/create_logical_tables/update_metadata.rs
@@ -58,10 +58,10 @@ impl CreateLogicalTablesProcedure {
            &new_table_info.name,
        );

-        // Update physical table's metadata
+        // Update physical table's metadata and we don't need to touch per-region settings.
        self.context
            .table_metadata_manager
-            .update_table_info(&physical_table_info, new_table_info)
+            .update_table_info(&physical_table_info, None, new_table_info)
            .await?;

        // Invalid physical table cache
--- a/src/common/meta/src/ddl/test_util/datanode_handler.rs
+++ b/src/common/meta/src/ddl/test_util/datanode_handler.rs
@@ -29,7 +29,10 @@ use crate::test_util::MockDatanodeHandler;
 #[async_trait::async_trait]
 impl MockDatanodeHandler for () {
    async fn handle(&self, _peer: &Peer, _request: RegionRequest) -> Result<RegionResponse> {
-        unreachable!()
+        Ok(RegionResponse {
+            affected_rows: 0,
+            extensions: Default::default(),
+        })
    }

    async fn handle_query(
--- a/src/common/meta/src/ddl/tests/alter_table.rs
+++ b/src/common/meta/src/ddl/tests/alter_table.rs
@@ -19,13 +19,14 @@ use std::sync::Arc;
 use api::v1::alter_expr::Kind;
 use api::v1::region::{region_request, RegionRequest};
 use api::v1::{
-    AddColumn, AddColumns, AlterExpr, ColumnDataType, ColumnDef as PbColumnDef, DropColumn,
-    DropColumns, SemanticType,
+    AddColumn, AddColumns, AlterExpr, ChangeTableOption, ChangeTableOptions, ColumnDataType,
+    ColumnDef as PbColumnDef, DropColumn, DropColumns, SemanticType,
 };
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use store_api::storage::RegionId;
+use table::requests::TTL_KEY;
 use tokio::sync::mpsc::{self};

 use crate::ddl::alter_table::AlterTableProcedure;
@@ -34,6 +35,7 @@ use crate::ddl::test_util::create_table::test_create_table_task;
 use crate::ddl::test_util::datanode_handler::{
    DatanodeWatcher, RequestOutdatedErrorDatanodeHandler,
 };
+use crate::key::datanode_table::DatanodeTableKey;
 use crate::key::table_name::TableNameKey;
 use crate::key::table_route::TableRouteValue;
 use crate::peer::Peer;
@@ -293,12 +295,21 @@ async fn test_on_update_metadata_add_columns() {
    let table_name = "foo";
    let table_id = 1024;
    let task = test_create_table_task(table_name, table_id);
+
+    let region_id = RegionId::new(table_id, 0);
+    let mock_table_routes = vec![RegionRoute {
+        region: Region::new_test(region_id),
+        leader_peer: Some(Peer::default()),
+        follower_peers: vec![],
+        leader_state: None,
+        leader_down_since: None,
+    }];
    // Puts a value to table name key.
    ddl_context
        .table_metadata_manager
        .create_table_metadata(
            task.table_info.clone(),
-            TableRouteValue::physical(vec![]),
+            TableRouteValue::physical(mock_table_routes),
            HashMap::new(),
        )
        .await
@@ -326,6 +337,7 @@ async fn test_on_update_metadata_add_columns() {
    let mut procedure =
        AlterTableProcedure::new(cluster_id, table_id, task, ddl_context.clone()).unwrap();
    procedure.on_prepare().await.unwrap();
+    procedure.submit_alter_region_requests().await.unwrap();
    procedure.on_update_metadata().await.unwrap();

    let table_info = ddl_context
@@ -343,3 +355,76 @@ async fn test_on_update_metadata_add_columns() {
        table_info.meta.next_column_id
    );
 }
+
+#[tokio::test]
+async fn test_on_update_table_options() {
+    let node_manager = Arc::new(MockDatanodeManager::new(()));
+    let ddl_context = new_ddl_context(node_manager);
+    let cluster_id = 1;
+    let table_name = "foo";
+    let table_id = 1024;
+    let task = test_create_table_task(table_name, table_id);
+
+    let region_id = RegionId::new(table_id, 0);
+    let mock_table_routes = vec![RegionRoute {
+        region: Region::new_test(region_id),
+        leader_peer: Some(Peer::default()),
+        follower_peers: vec![],
+        leader_state: None,
+        leader_down_since: None,
+    }];
+    // Puts a value to table name key.
+    ddl_context
+        .table_metadata_manager
+        .create_table_metadata(
+            task.table_info.clone(),
+            TableRouteValue::physical(mock_table_routes),
+            HashMap::new(),
+        )
+        .await
+        .unwrap();
+
+    let task = AlterTableTask {
+        alter_table: AlterExpr {
+            catalog_name: DEFAULT_CATALOG_NAME.to_string(),
+            schema_name: DEFAULT_SCHEMA_NAME.to_string(),
+            table_name: table_name.to_string(),
+            kind: Some(Kind::ChangeTableOptions(ChangeTableOptions {
+                change_table_options: vec![ChangeTableOption {
+                    key: TTL_KEY.to_string(),
+                    value: "1d".to_string(),
+                }],
+            })),
+        },
+    };
+    let mut procedure =
+        AlterTableProcedure::new(cluster_id, table_id, task, ddl_context.clone()).unwrap();
+    procedure.on_prepare().await.unwrap();
+    procedure.submit_alter_region_requests().await.unwrap();
+    procedure.on_update_metadata().await.unwrap();
+
+    let table_info = ddl_context
+        .table_metadata_manager
+        .table_info_manager()
+        .get(table_id)
+        .await
+        .unwrap()
+        .unwrap()
+        .into_inner()
+        .table_info;
+
+    let datanode_key = DatanodeTableKey::new(0, table_id);
+    let region_info = ddl_context
+        .table_metadata_manager
+        .datanode_table_manager()
+        .get(&datanode_key)
+        .await
+        .unwrap()
+        .unwrap()
+        .region_info;
+
+    assert_eq!(
+        region_info.region_options,
+        HashMap::from(&table_info.meta.options)
+    );
+}
--- a/src/common/meta/src/error.rs
+++ b/src/common/meta/src/error.rs
@@ -652,6 +652,18 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
+
+    #[snafu(display(
+        "Datanode table info not found, table id: {}, datanode id: {}",
+        table_id,
+        datanode_id
+    ))]
+    DatanodeTableInfoNotFound {
+        datanode_id: DatanodeId,
+        table_id: TableId,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -752,6 +764,7 @@ impl ErrorExt for Error {
            PostgresExecution { .. } => StatusCode::Internal,
            #[cfg(feature = "pg_kvbackend")]
            ConnectPostgres { .. } => StatusCode::Internal,
+            Error::DatanodeTableInfoNotFound { .. } => StatusCode::Internal,
        }
    }

--- a/src/common/meta/src/key.rs
+++ b/src/common/meta/src/key.rs
@@ -91,6 +91,7 @@ pub mod catalog_name;
 pub mod datanode_table;
 pub mod flow;
 pub mod node_address;
+mod schema_metadata_manager;
 pub mod schema_name;
 pub mod table_info;
 pub mod table_name;
@@ -116,6 +117,7 @@ use flow::flow_route::FlowRouteValue;
 use flow::table_flow::TableFlowValue;
 use lazy_static::lazy_static;
 use regex::Regex;
+pub use schema_metadata_manager::{SchemaMetadataManager, SchemaMetadataManagerRef};
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, OptionExt, ResultExt};
@@ -133,7 +135,6 @@ use self::flow::flow_name::FlowNameValue;
 use self::schema_name::{SchemaManager, SchemaNameKey, SchemaNameValue};
 use self::table_route::{TableRouteManager, TableRouteValue};
 use self::tombstone::TombstoneManager;
-use crate::ddl::utils::region_storage_path;
 use crate::error::{self, Result, SerdeJsonSnafu};
 use crate::key::node_address::NodeAddressValue;
 use crate::key::table_route::TableRouteKey;
@@ -593,8 +594,6 @@ impl TableMetadataManager {
        table_info.meta.region_numbers = region_numbers;
        let table_id = table_info.ident.table_id;
        let engine = table_info.meta.engine.clone();
-        let region_storage_path =
-            region_storage_path(&table_info.catalog_name, &table_info.schema_name);

        // Creates table name.
        let table_name = TableNameKey::new(
@@ -606,7 +605,7 @@ impl TableMetadataManager {
            .table_name_manager()
            .build_create_txn(&table_name, table_id)?;

-        let region_options = (&table_info.meta.options).into();
+        let region_options = table_info.to_region_options();
        // Creates table info.
        let table_info_value = TableInfoValue::new(table_info);
        let (create_table_info_txn, on_create_table_info_failure) = self
@@ -625,6 +624,7 @@ impl TableMetadataManager {
        ]);

        if let TableRouteValue::Physical(x) = &table_route_value {
+            let region_storage_path = table_info_value.region_storage_path();
            let create_datanode_table_txn = self.datanode_table_manager().build_create_txn(
                table_id,
                &engine,
@@ -926,13 +926,15 @@ impl TableMetadataManager {
    }

    /// Updates table info and returns an error if different metadata exists.
+    /// And cascade-ly update all redundant table options for each region
+    /// if region_distribution is present.
    pub async fn update_table_info(
        &self,
        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
+        region_distribution: Option<RegionDistribution>,
        new_table_info: RawTableInfo,
    ) -> Result<()> {
        let table_id = current_table_info_value.table_info.ident.table_id;
-
        let new_table_info_value = current_table_info_value.update(new_table_info);

        // Updates table info.
@@ -940,8 +942,19 @@ impl TableMetadataManager {
            .table_info_manager()
            .build_update_txn(table_id, current_table_info_value, &new_table_info_value)?;

-        let mut r = self.kv_backend.txn(update_table_info_txn).await?;
+        let txn = if let Some(region_distribution) = region_distribution {
+            // region options induced from table info.
+            let new_region_options = new_table_info_value.table_info.to_region_options();
+            let update_datanode_table_options_txn = self
+                .datanode_table_manager
+                .build_update_table_options_txn(table_id, region_distribution, new_region_options)
+                .await?;
+            Txn::merge_all([update_table_info_txn, update_datanode_table_options_txn])
+        } else {
+            update_table_info_txn
+        };

+        let mut r = self.kv_backend.txn(txn).await?;
        // Checks whether metadata was already updated.
        if !r.succeeded {
            let mut set = TxnOpGetResponseSet::from(&mut r.responses);
@@ -1669,12 +1682,12 @@ mod tests {
            DeserializedValueWithBytes::from_inner(TableInfoValue::new(table_info.clone()));
        // should be ok.
        table_metadata_manager
-            .update_table_info(&current_table_info_value, new_table_info.clone())
+            .update_table_info(&current_table_info_value, None, new_table_info.clone())
            .await
            .unwrap();
        // if table info was updated, it should be ok.
        table_metadata_manager
-            .update_table_info(&current_table_info_value, new_table_info.clone())
+            .update_table_info(&current_table_info_value, None, new_table_info.clone())
            .await
            .unwrap();

@@ -1696,7 +1709,7 @@ mod tests {
        // if the current_table_info_value is wrong, it should return an error.
        // The ABA problem.
        assert!(table_metadata_manager
-            .update_table_info(&wrong_table_info_value, new_table_info)
+            .update_table_info(&wrong_table_info_value, None, new_table_info)
            .await
            .is_err())
    }
--- a/src/common/meta/src/key/catalog_name.rs
+++ b/src/common/meta/src/key/catalog_name.rs
@@ -35,7 +35,7 @@ pub struct CatalogNameKey<'a> {
    pub catalog: &'a str,
 }

-impl<'a> Default for CatalogNameKey<'a> {
+impl Default for CatalogNameKey<'_> {
    fn default() -> Self {
        Self {
            catalog: DEFAULT_CATALOG_NAME,
--- a/src/common/meta/src/key/datanode_table.rs
+++ b/src/common/meta/src/key/datanode_table.rs
@@ -23,7 +23,7 @@ use store_api::storage::RegionNumber;
 use table::metadata::TableId;

 use super::MetadataKey;
-use crate::error::{InvalidMetadataSnafu, Result};
+use crate::error::{DatanodeTableInfoNotFoundSnafu, InvalidMetadataSnafu, Result};
 use crate::key::{
    MetadataValue, RegionDistribution, DATANODE_TABLE_KEY_PATTERN, DATANODE_TABLE_KEY_PREFIX,
 };
@@ -77,7 +77,7 @@ impl DatanodeTableKey {
    }
 }

-impl<'a> MetadataKey<'a, DatanodeTableKey> for DatanodeTableKey {
+impl MetadataKey<'_, DatanodeTableKey> for DatanodeTableKey {
    fn to_bytes(&self) -> Vec<u8> {
        self.to_string().into_bytes()
    }
@@ -209,6 +209,49 @@ impl DatanodeTableManager {
        Ok(txn)
    }

+    /// Builds a transaction to updates the redundant table options (including WAL options)
+    /// for given table id, if provided.
+    ///
+    /// Note that the provided `new_region_options` must be a
+    /// complete set of all options rather than incremental changes.
+    pub(crate) async fn build_update_table_options_txn(
+        &self,
+        table_id: TableId,
+        region_distribution: RegionDistribution,
+        new_region_options: HashMap<String, String>,
+    ) -> Result<Txn> {
+        assert!(!region_distribution.is_empty());
+        // safety: region_distribution must not be empty
+        let (any_datanode, _) = region_distribution.first_key_value().unwrap();
+
+        let mut region_info = self
+            .kv_backend
+            .get(&DatanodeTableKey::new(*any_datanode, table_id).to_bytes())
+            .await
+            .transpose()
+            .context(DatanodeTableInfoNotFoundSnafu {
+                datanode_id: *any_datanode,
+                table_id,
+            })?
+            .and_then(|r| DatanodeTableValue::try_from_raw_value(&r.value))?
+            .region_info;
+        // substitute region options only.
+        region_info.region_options = new_region_options;
+
+        let mut txns = Vec::with_capacity(region_distribution.len());
+
+        for (datanode, regions) in region_distribution.into_iter() {
+            let key = DatanodeTableKey::new(datanode, table_id);
+            let key_bytes = key.to_bytes();
+            let value_bytes = DatanodeTableValue::new(table_id, regions, region_info.clone())
+                .try_as_raw_value()?;
+            txns.push(TxnOp::Put(key_bytes, value_bytes));
+        }
+
+        let txn = Txn::new().and_then(txns);
+        Ok(txn)
+    }
+
    /// Builds the update datanode table transactions. It only executes while the primary keys comparing successes.
    pub(crate) fn build_update_txn(
        &self,
--- a/src/common/meta/src/key/flow/flow_info.rs
+++ b/src/common/meta/src/key/flow/flow_info.rs
@@ -42,6 +42,8 @@ lazy_static! {
 /// The layout: `__flow/info/{flow_id}`.
 pub struct FlowInfoKey(FlowScoped<FlowInfoKeyInner>);

+pub type FlowInfoDecodeResult = Result<Option<DeserializedValueWithBytes<FlowInfoValue>>>;
+
 impl<'a> MetadataKey<'a, FlowInfoKey> for FlowInfoKey {
    fn to_bytes(&self) -> Vec<u8> {
        self.0.to_bytes()
@@ -203,9 +205,7 @@ impl FlowInfoManager {
        flow_value: &FlowInfoValue,
    ) -> Result<(
        Txn,
-        impl FnOnce(
-            &mut TxnOpGetResponseSet,
-        ) -> Result<Option<DeserializedValueWithBytes<FlowInfoValue>>>,
+        impl FnOnce(&mut TxnOpGetResponseSet) -> FlowInfoDecodeResult,
    )> {
        let key = FlowInfoKey::new(flow_id).to_bytes();
        let txn = Txn::put_if_not_exists(key.clone(), flow_value.try_as_raw_value()?);
--- a/src/common/meta/src/key/flow/flow_name.rs
+++ b/src/common/meta/src/key/flow/flow_name.rs
@@ -46,6 +46,8 @@ lazy_static! {
 /// The layout: `__flow/name/{catalog_name}/{flow_name}`.
 pub struct FlowNameKey<'a>(FlowScoped<FlowNameKeyInner<'a>>);

+pub type FlowNameDecodeResult = Result<Option<DeserializedValueWithBytes<FlowNameValue>>>;
+
 #[allow(dead_code)]
 impl<'a> FlowNameKey<'a> {
    /// Returns the [FlowNameKey]
@@ -104,7 +106,7 @@ impl<'a> MetadataKey<'a, FlowNameKeyInner<'a>> for FlowNameKeyInner<'_> {
        .into_bytes()
    }

-    fn from_bytes(bytes: &'a [u8]) -> Result<FlowNameKeyInner> {
+    fn from_bytes(bytes: &'a [u8]) -> Result<FlowNameKeyInner<'a>> {
        let key = std::str::from_utf8(bytes).map_err(|e| {
            error::InvalidMetadataSnafu {
                err_msg: format!(
@@ -223,9 +225,7 @@ impl FlowNameManager {
        flow_id: FlowId,
    ) -> Result<(
        Txn,
-        impl FnOnce(
-            &mut TxnOpGetResponseSet,
-        ) -> Result<Option<DeserializedValueWithBytes<FlowNameValue>>>,
+        impl FnOnce(&mut TxnOpGetResponseSet) -> FlowNameDecodeResult,
    )> {
        let key = FlowNameKey::new(catalog_name, flow_name);
        let raw_key = key.to_bytes();
--- a/src/common/meta/src/key/node_address.rs
+++ b/src/common/meta/src/key/node_address.rs
@@ -52,7 +52,7 @@ impl NodeAddressValue {
    }
 }

-impl<'a> MetadataKey<'a, NodeAddressKey> for NodeAddressKey {
+impl MetadataKey<'_, NodeAddressKey> for NodeAddressKey {
    fn to_bytes(&self) -> Vec<u8> {
        self.to_string().into_bytes()
    }
--- a/src/common/meta/src/key/schema_metadata_manager.rs
+++ b/src/common/meta/src/key/schema_metadata_manager.rs
@@ -0,0 +1,122 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Schema-level metadata manager.
+
+use std::sync::Arc;
+
+use snafu::OptionExt;
+use store_api::storage::TableId;
+
+use crate::error::TableInfoNotFoundSnafu;
+use crate::key::schema_name::{SchemaManager, SchemaNameKey};
+use crate::key::table_info::{TableInfoManager, TableInfoManagerRef};
+use crate::kv_backend::KvBackendRef;
+use crate::{error, SchemaOptions};
+
+pub type SchemaMetadataManagerRef = Arc<SchemaMetadataManager>;
+
+pub struct SchemaMetadataManager {
+    table_info_manager: TableInfoManagerRef,
+    schema_manager: SchemaManager,
+    #[cfg(any(test, feature = "testing"))]
+    kv_backend: KvBackendRef,
+}
+
+impl SchemaMetadataManager {
+    /// Creates a new database meta
+    #[cfg(not(any(test, feature = "testing")))]
+    pub fn new(kv_backend: KvBackendRef) -> Self {
+        let table_info_manager = Arc::new(TableInfoManager::new(kv_backend.clone()));
+        let schema_manager = SchemaManager::new(kv_backend);
+        Self {
+            table_info_manager,
+            schema_manager,
+        }
+    }
+
+    /// Creates a new database meta
+    #[cfg(any(test, feature = "testing"))]
+    pub fn new(kv_backend: KvBackendRef) -> Self {
+        let table_info_manager = Arc::new(TableInfoManager::new(kv_backend.clone()));
+        let schema_manager = SchemaManager::new(kv_backend.clone());
+        Self {
+            table_info_manager,
+            schema_manager,
+            kv_backend,
+        }
+    }
+
+    /// Gets schema options by table id.
+    pub async fn get_schema_options_by_table_id(
+        &self,
+        table_id: TableId,
+    ) -> error::Result<Option<SchemaOptions>> {
+        let table_info = self
+            .table_info_manager
+            .get(table_id)
+            .await?
+            .with_context(|| TableInfoNotFoundSnafu {
+                table: format!("table id: {}", table_id),
+            })?;
+
+        let key = SchemaNameKey::new(
+            &table_info.table_info.catalog_name,
+            &table_info.table_info.schema_name,
+        );
+        self.schema_manager.get(key).await
+    }
+
+    #[cfg(any(test, feature = "testing"))]
+    pub async fn register_region_table_info(
+        &self,
+        table_id: TableId,
+        table_name: &str,
+        schema_name: &str,
+        catalog_name: &str,
+        schema_value: Option<crate::key::schema_name::SchemaNameValue>,
+    ) {
+        use table::metadata::{RawTableInfo, TableType};
+        let value = crate::key::table_info::TableInfoValue::new(RawTableInfo {
+            ident: Default::default(),
+            name: table_name.to_string(),
+            desc: None,
+            catalog_name: catalog_name.to_string(),
+            schema_name: schema_name.to_string(),
+            meta: Default::default(),
+            table_type: TableType::Base,
+        });
+        let (txn, _) = self
+            .table_info_manager
+            .build_create_txn(table_id, &value)
+            .unwrap();
+        let resp = self.kv_backend.txn(txn).await.unwrap();
+        assert!(resp.succeeded, "Failed to create table metadata");
+        let key = SchemaNameKey {
+            catalog: catalog_name,
+            schema: schema_name,
+        };
+        self.schema_manager
+            .create(key, schema_value, false)
+            .await
+            .expect("Failed to create schema metadata");
+        common_telemetry::info!(
+            "Register table: {}, id: {}, schema: {}, catalog: {}",
+            table_name,
+            table_id,
+            schema_name,
+            catalog_name
+        );
+    }
+}
--- a/src/common/meta/src/key/schema_name.rs
+++ b/src/common/meta/src/key/schema_name.rs
@@ -41,7 +41,7 @@ pub struct SchemaNameKey<'a> {
    pub schema: &'a str,
 }

-impl<'a> Default for SchemaNameKey<'a> {
+impl Default for SchemaNameKey<'_> {
    fn default() -> Self {
        Self {
            catalog: DEFAULT_CATALOG_NAME,
--- a/src/common/meta/src/key/table_info.rs
+++ b/src/common/meta/src/key/table_info.rs
@@ -23,6 +23,7 @@ use table::table_name::TableName;
 use table::table_reference::TableReference;

 use super::TABLE_INFO_KEY_PATTERN;
+use crate::ddl::utils::region_storage_path;
 use crate::error::{InvalidMetadataSnafu, Result};
 use crate::key::txn_helper::TxnOpGetResponseSet;
 use crate::key::{DeserializedValueWithBytes, MetadataKey, MetadataValue, TABLE_INFO_KEY_PREFIX};
@@ -51,7 +52,7 @@ impl Display for TableInfoKey {
    }
 }

-impl<'a> MetadataKey<'a, TableInfoKey> for TableInfoKey {
+impl MetadataKey<'_, TableInfoKey> for TableInfoKey {
    fn to_bytes(&self) -> Vec<u8> {
        self.to_string().into_bytes()
    }
@@ -125,13 +126,20 @@ impl TableInfoValue {
            table_name: self.table_info.name.to_string(),
        }
    }
+
+    /// Builds storage path for all regions in table.
+    pub fn region_storage_path(&self) -> String {
+        region_storage_path(&self.table_info.catalog_name, &self.table_info.schema_name)
+    }
 }

 pub type TableInfoManagerRef = Arc<TableInfoManager>;
+
 #[derive(Clone)]
 pub struct TableInfoManager {
    kv_backend: KvBackendRef,
 }
+pub type TableInfoDecodeResult = Result<Option<DeserializedValueWithBytes<TableInfoValue>>>;

 impl TableInfoManager {
    pub fn new(kv_backend: KvBackendRef) -> Self {
@@ -145,9 +153,7 @@ impl TableInfoManager {
        table_info_value: &TableInfoValue,
    ) -> Result<(
        Txn,
-        impl FnOnce(
-            &mut TxnOpGetResponseSet,
-        ) -> Result<Option<DeserializedValueWithBytes<TableInfoValue>>>,
+        impl FnOnce(&mut TxnOpGetResponseSet) -> TableInfoDecodeResult,
    )> {
        let key = TableInfoKey::new(table_id);
        let raw_key = key.to_bytes();
@@ -169,9 +175,7 @@ impl TableInfoManager {
        new_table_info_value: &TableInfoValue,
    ) -> Result<(
        Txn,
-        impl FnOnce(
-            &mut TxnOpGetResponseSet,
-        ) -> Result<Option<DeserializedValueWithBytes<TableInfoValue>>>,
+        impl FnOnce(&mut TxnOpGetResponseSet) -> TableInfoDecodeResult,
    )> {
        let key = TableInfoKey::new(table_id);
        let raw_key = key.to_bytes();
--- a/src/common/meta/src/key/table_route.rs
+++ b/src/common/meta/src/key/table_route.rs
@@ -245,7 +245,7 @@ impl LogicalTableRouteValue {
    }
 }

-impl<'a> MetadataKey<'a, TableRouteKey> for TableRouteKey {
+impl MetadataKey<'_, TableRouteKey> for TableRouteKey {
    fn to_bytes(&self) -> Vec<u8> {
        self.to_string().into_bytes()
    }
@@ -472,6 +472,8 @@ pub struct TableRouteStorage {
    kv_backend: KvBackendRef,
 }

+pub type TableRouteValueDecodeResult = Result<Option<DeserializedValueWithBytes<TableRouteValue>>>;
+
 impl TableRouteStorage {
    pub fn new(kv_backend: KvBackendRef) -> Self {
        Self { kv_backend }
@@ -485,9 +487,7 @@ impl TableRouteStorage {
        table_route_value: &TableRouteValue,
    ) -> Result<(
        Txn,
-        impl FnOnce(
-            &mut TxnOpGetResponseSet,
-        ) -> Result<Option<DeserializedValueWithBytes<TableRouteValue>>>,
+        impl FnOnce(&mut TxnOpGetResponseSet) -> TableRouteValueDecodeResult,
    )> {
        let key = TableRouteKey::new(table_id);
        let raw_key = key.to_bytes();
@@ -510,9 +510,7 @@ impl TableRouteStorage {
        new_table_route_value: &TableRouteValue,
    ) -> Result<(
        Txn,
-        impl FnOnce(
-            &mut TxnOpGetResponseSet,
-        ) -> Result<Option<DeserializedValueWithBytes<TableRouteValue>>>,
+        impl FnOnce(&mut TxnOpGetResponseSet) -> TableRouteValueDecodeResult,
    )> {
        let key = TableRouteKey::new(table_id);
        let raw_key = key.to_bytes();
--- a/src/common/meta/src/key/view_info.rs
+++ b/src/common/meta/src/key/view_info.rs
@@ -53,7 +53,7 @@ impl Display for ViewInfoKey {
    }
 }

-impl<'a> MetadataKey<'a, ViewInfoKey> for ViewInfoKey {
+impl MetadataKey<'_, ViewInfoKey> for ViewInfoKey {
    fn to_bytes(&self) -> Vec<u8> {
        self.to_string().into_bytes()
    }
@@ -139,6 +139,8 @@ pub struct ViewInfoManager {

 pub type ViewInfoManagerRef = Arc<ViewInfoManager>;

+pub type ViewInfoValueDecodeResult = Result<Option<DeserializedValueWithBytes<ViewInfoValue>>>;
+
 impl ViewInfoManager {
    pub fn new(kv_backend: KvBackendRef) -> Self {
        Self { kv_backend }
@@ -151,9 +153,7 @@ impl ViewInfoManager {
        view_info_value: &ViewInfoValue,
    ) -> Result<(
        Txn,
-        impl FnOnce(
-            &mut TxnOpGetResponseSet,
-        ) -> Result<Option<DeserializedValueWithBytes<ViewInfoValue>>>,
+        impl FnOnce(&mut TxnOpGetResponseSet) -> ViewInfoValueDecodeResult,
    )> {
        let key = ViewInfoKey::new(view_id);
        let raw_key = key.to_bytes();
@@ -175,9 +175,7 @@ impl ViewInfoManager {
        new_view_info_value: &ViewInfoValue,
    ) -> Result<(
        Txn,
-        impl FnOnce(
-            &mut TxnOpGetResponseSet,
-        ) -> Result<Option<DeserializedValueWithBytes<ViewInfoValue>>>,
+        impl FnOnce(&mut TxnOpGetResponseSet) -> ViewInfoValueDecodeResult,
    )> {
        let key = ViewInfoKey::new(view_id);
        let raw_key = key.to_bytes();
--- a/src/common/meta/src/lib.rs
+++ b/src/common/meta/src/lib.rs
@@ -54,4 +54,7 @@ pub type DatanodeId = u64;
 // The id of the flownode.
 pub type FlownodeId = u64;

+/// Schema options.
+pub type SchemaOptions = key::schema_name::SchemaNameValue;
+
 pub use instruction::RegionIdent;
--- a/src/common/meta/src/lock_key.rs
+++ b/src/common/meta/src/lock_key.rs
@@ -34,7 +34,7 @@ pub enum CatalogLock<'a> {
    Write(&'a str),
 }

-impl<'a> Display for CatalogLock<'a> {
+impl Display for CatalogLock<'_> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let key = match self {
            CatalogLock::Read(s) => s,
@@ -44,7 +44,7 @@ impl<'a> Display for CatalogLock<'a> {
    }
 }

-impl<'a> From<CatalogLock<'a>> for StringKey {
+impl From<CatalogLock<'_>> for StringKey {
    fn from(value: CatalogLock) -> Self {
        match value {
            CatalogLock::Write(_) => StringKey::Exclusive(value.to_string()),
--- a/src/common/procedure/src/store.rs
+++ b/src/common/procedure/src/store.rs
@@ -297,7 +297,7 @@ struct ParsedKey<'a> {
    key_type: KeyType,
 }

-impl<'a> fmt::Display for ParsedKey<'a> {
+impl fmt::Display for ParsedKey<'_> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(
            f,
--- a/src/common/recordbatch/src/recordbatch.rs
+++ b/src/common/recordbatch/src/recordbatch.rs
@@ -17,6 +17,7 @@ use std::slice;
 use std::sync::Arc;

 use datafusion::arrow::util::pretty::pretty_format_batches;
+use datatypes::prelude::DataType;
 use datatypes::schema::SchemaRef;
 use datatypes::value::Value;
 use datatypes::vectors::{Helper, VectorRef};
@@ -58,13 +59,18 @@ impl RecordBatch {
    }

    /// Create an empty [`RecordBatch`] from `schema`.
-    pub fn new_empty(schema: SchemaRef) -> Result<RecordBatch> {
+    pub fn new_empty(schema: SchemaRef) -> RecordBatch {
        let df_record_batch = DfRecordBatch::new_empty(schema.arrow_schema().clone());
-        Ok(RecordBatch {
+        let columns = schema
+            .column_schemas()
+            .iter()
+            .map(|col| col.data_type.create_mutable_vector(0).to_vector())
+            .collect();
+        RecordBatch {
            schema,
-            columns: vec![],
+            columns,
            df_record_batch,
-        })
+        }
    }

    pub fn try_project(&self, indices: &[usize]) -> Result<Self> {
@@ -220,7 +226,7 @@ pub struct RecordBatchRowIterator<'a> {
 }

 impl<'a> RecordBatchRowIterator<'a> {
-    fn new(record_batch: &'a RecordBatch) -> RecordBatchRowIterator {
+    fn new(record_batch: &'a RecordBatch) -> RecordBatchRowIterator<'a> {
        RecordBatchRowIterator {
            record_batch,
            rows: record_batch.df_record_batch.num_rows(),
@@ -230,7 +236,7 @@ impl<'a> RecordBatchRowIterator<'a> {
    }
 }

-impl<'a> Iterator for RecordBatchRowIterator<'a> {
+impl Iterator for RecordBatchRowIterator<'_> {
    type Item = Vec<Value>;

    fn next(&mut self) -> Option<Self::Item> {
--- a/src/common/runtime/Cargo.toml
+++ b/src/common/runtime/Cargo.toml
@@ -4,21 +4,36 @@ version.workspace = true
 edition.workspace = true
 license.workspace = true

+[lib]
+path = "src/lib.rs"
+
+[[bin]]
+name = "common-runtime-bin"
+path = "src/bin.rs"
+
 [lints]
 workspace = true

 [dependencies]
 async-trait.workspace = true
+clap.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
 common-telemetry.workspace = true
+futures.workspace = true
 lazy_static.workspace = true
 num_cpus.workspace = true
 once_cell.workspace = true
+parking_lot.workspace = true
 paste.workspace = true
+pin-project.workspace = true
 prometheus.workspace = true
+rand.workspace = true
+ratelimit.workspace = true
 serde.workspace = true
+serde_json.workspace = true
 snafu.workspace = true
+tempfile.workspace = true
 tokio.workspace = true
 tokio-metrics = "0.3"
 tokio-metrics-collector = { git = "https://github.com/MichaelScofield/tokio-metrics-collector.git", rev = "89d692d5753d28564a7aac73c6ac5aba22243ba0" }
--- a/src/common/runtime/README.md
+++ b/src/common/runtime/README.md
@@ -0,0 +1,60 @@
+# Greptime Runtime
+
+## Run performance test for different priority & workload type
+
+```
+# workspace is at this subcrate
+cargo run --release -- --loop-cnt 500
+```
+
+## Related PRs & issues
+
+- Preliminary support cpu limitation
+  
+  ISSUE: https://github.com/GreptimeTeam/greptimedb/issues/3685
+  
+  PR: https://github.com/GreptimeTeam/greptimedb/pull/4782
+
+## CPU resource constraints (ThrottleableRuntime)
+
+
+To achieve CPU resource constraints, we adopt the concept of rate limiting. When creating a future, we first wrap it with another layer of future to intercept the poll operation during runtime. By using the ratelimit library, we can simply implement a mechanism that allows only a limited number of polls for a batch of tasks under a certain priority within a specific time frame (the current token generation interval is set to 10ms).
+
+The default used runtime can be switched by 
+``` rust
+pub type Runtime = DefaultRuntime;
+```
+in `runtime.rs`.
+
+We tested four type of workload with 5 priorities, whose setup are as follows:
+
+``` rust
+impl Priority {
+    fn ratelimiter_count(&self) -> Result<Option<Ratelimiter>> {
+        let max = 8000;
+        let gen_per_10ms = match self {
+            Priority::VeryLow => Some(2000),
+            Priority::Low => Some(4000),
+            Priority::Middle => Some(6000),
+            Priority::High => Some(8000),
+            Priority::VeryHigh => None,
+        };
+        if let Some(gen_per_10ms) = gen_per_10ms {
+            Ratelimiter::builder(gen_per_10ms, Duration::from_millis(10)) // generate poll count per 10ms
+                .max_tokens(max) // reserved token for batch request
+                .build()
+                .context(BuildRuntimeRateLimiterSnafu)
+                .map(Some)
+        } else {
+            Ok(None)
+        }
+    }
+}
+```
+
+This is the preliminary experimental effect so far:
+
+![](resources/rdme-exp.png)
+
+## TODO
+- Introduce PID to achieve more accurate limitation.
--- a/src/common/runtime/resources/rdme-exp.png
+++ b/src/common/runtime/resources/rdme-exp.png
--- a/src/common/runtime/src/bin.rs
+++ b/src/common/runtime/src/bin.rs
@@ -0,0 +1,205 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use clap::Parser;
+
+#[derive(Debug, Default, Parser)]
+pub struct Command {
+    #[clap(long)]
+    loop_cnt: usize,
+}
+
+fn main() {
+    common_telemetry::init_default_ut_logging();
+    let cmd = Command::parse();
+
+    test_diff_priority_cpu::test_diff_workload_priority(cmd.loop_cnt);
+}
+
+mod test_diff_priority_cpu {
+    use std::path::PathBuf;
+
+    use common_runtime::runtime::{BuilderBuild, Priority, RuntimeTrait};
+    use common_runtime::{Builder, Runtime};
+    use common_telemetry::debug;
+    use tempfile::TempDir;
+
+    fn compute_pi_str(precision: usize) -> String {
+        let mut pi = 0.0;
+        let mut sign = 1.0;
+
+        for i in 0..precision {
+            pi += sign / (2 * i + 1) as f64;
+            sign *= -1.0;
+        }
+
+        pi *= 4.0;
+        format!("{:.prec$}", pi, prec = precision)
+    }
+
+    macro_rules! def_workload_enum {
+        ($($variant:ident),+) => {
+            #[derive(Debug)]
+            enum WorkloadType {
+                $($variant),+
+            }
+
+            /// array of workloads for iteration
+            const WORKLOADS: &'static [WorkloadType] = &[
+                $( WorkloadType::$variant ),+
+            ];
+        };
+    }
+
+    def_workload_enum!(
+        ComputeHeavily,
+        ComputeHeavily2,
+        WriteFile,
+        SpawnBlockingWriteFile
+    );
+
+    async fn workload_compute_heavily() {
+        let prefix = 10;
+
+        for _ in 0..3000 {
+            let _ = compute_pi_str(prefix);
+            tokio::task::yield_now().await;
+        }
+    }
+    async fn workload_compute_heavily2() {
+        let prefix = 30;
+        for _ in 0..2000 {
+            let _ = compute_pi_str(prefix);
+            tokio::task::yield_now().await;
+        }
+    }
+    async fn workload_write_file(_idx: u64, tempdir: PathBuf) {
+        use tokio::io::AsyncWriteExt;
+        let prefix = 50;
+
+        let mut file = tokio::fs::OpenOptions::new()
+            .write(true)
+            .append(true)
+            .create(true)
+            .open(tempdir.join(format!("pi_{}", prefix)))
+            .await
+            .unwrap();
+        for i in 0..200 {
+            let pi = compute_pi_str(prefix);
+
+            if i % 2 == 0 {
+                file.write_all(pi.as_bytes()).await.unwrap();
+            }
+        }
+    }
+    async fn workload_spawn_blocking_write_file(tempdir: PathBuf) {
+        use std::io::Write;
+        let prefix = 100;
+        let mut file = Some(
+            std::fs::OpenOptions::new()
+                .append(true)
+                .create(true)
+                .open(tempdir.join(format!("pi_{}", prefix)))
+                .unwrap(),
+        );
+        for i in 0..100 {
+            let pi = compute_pi_str(prefix);
+            if i % 2 == 0 {
+                let mut file1 = file.take().unwrap();
+                file = Some(
+                    tokio::task::spawn_blocking(move || {
+                        file1.write_all(pi.as_bytes()).unwrap();
+                        file1
+                    })
+                    .await
+                    .unwrap(),
+                );
+            }
+        }
+    }
+
+    pub fn test_diff_workload_priority(loop_cnt: usize) {
+        let tempdir = tempfile::tempdir().unwrap();
+        let priorities = [
+            Priority::VeryLow,
+            Priority::Low,
+            Priority::Middle,
+            Priority::High,
+            Priority::VeryHigh,
+        ];
+        for wl in WORKLOADS {
+            for p in priorities.iter() {
+                let runtime: Runtime = Builder::default()
+                    .runtime_name("test")
+                    .thread_name("test")
+                    .worker_threads(8)
+                    .priority(*p)
+                    .build()
+                    .expect("Fail to create runtime");
+                let runtime2 = runtime.clone();
+                runtime.block_on(test_spec_priority_and_workload(
+                    *p, runtime2, wl, &tempdir, loop_cnt,
+                ));
+            }
+        }
+    }
+
+    async fn test_spec_priority_and_workload(
+        priority: Priority,
+        runtime: Runtime,
+        workload_id: &WorkloadType,
+        tempdir: &TempDir,
+        loop_cnt: usize,
+    ) {
+        tokio::time::sleep(tokio::time::Duration::from_millis(1000)).await;
+        debug!(
+            "testing cpu usage for priority {:?} workload_id {:?}",
+            priority, workload_id,
+        );
+        // start monitor thread
+        let mut tasks = vec![];
+        let start = std::time::Instant::now();
+        for i in 0..loop_cnt {
+            // persist cpu usage in json: {priority}.{workload_id}
+            match *workload_id {
+                WorkloadType::ComputeHeavily => {
+                    tasks.push(runtime.spawn(workload_compute_heavily()));
+                }
+                WorkloadType::ComputeHeavily2 => {
+                    tasks.push(runtime.spawn(workload_compute_heavily2()));
+                }
+                WorkloadType::SpawnBlockingWriteFile => {
+                    tasks.push(runtime.spawn(workload_spawn_blocking_write_file(
+                        tempdir.path().to_path_buf(),
+                    )));
+                }
+                WorkloadType::WriteFile => {
+                    tasks.push(
+                        runtime.spawn(workload_write_file(i as u64, tempdir.path().to_path_buf())),
+                    );
+                }
+            }
+        }
+        for task in tasks {
+            task.await.unwrap();
+        }
+        let elapsed = start.elapsed();
+        debug!(
+            "test cpu usage for priority {:?} workload_id {:?} elapsed {}ms",
+            priority,
+            workload_id,
+            elapsed.as_millis()
+        );
+    }
+}
--- a/src/common/runtime/src/error.rs
+++ b/src/common/runtime/src/error.rs
@@ -33,6 +33,14 @@ pub enum Error {
        location: Location,
    },

+    #[snafu(display("Failed to build runtime rate limiter"))]
+    BuildRuntimeRateLimiter {
+        #[snafu(implicit)]
+        location: Location,
+        #[snafu(source)]
+        error: ratelimit::Error,
+    },
+
    #[snafu(display("Repeated task {} is already started", name))]
    IllegalState {
        name: String,
--- a/src/common/runtime/src/global.rs
+++ b/src/common/runtime/src/global.rs
@@ -21,6 +21,7 @@ use once_cell::sync::Lazy;
 use paste::paste;
 use serde::{Deserialize, Serialize};

+use crate::runtime::{BuilderBuild, RuntimeTrait};
 use crate::{Builder, JoinHandle, Runtime};

 const GLOBAL_WORKERS: usize = 8;
--- a/src/common/runtime/src/lib.rs
+++ b/src/common/runtime/src/lib.rs
@@ -17,6 +17,8 @@ pub mod global;
 mod metrics;
 mod repeated_task;
 pub mod runtime;
+pub mod runtime_default;
+pub mod runtime_throttleable;

 pub use global::{
    block_on_compact, block_on_global, compact_runtime, create_runtime, global_runtime,
--- a/src/common/runtime/src/repeated_task.rs
+++ b/src/common/runtime/src/repeated_task.rs
@@ -23,6 +23,7 @@ use tokio::task::JoinHandle;
 use tokio_util::sync::CancellationToken;

 use crate::error::{IllegalStateSnafu, Result, WaitGcTaskStopSnafu};
+use crate::runtime::RuntimeTrait;
 use crate::Runtime;

 /// Task to execute repeatedly.
--- a/src/common/runtime/src/runtime.rs
+++ b/src/common/runtime/src/runtime.rs
@@ -19,24 +19,20 @@ use std::thread;
 use std::time::Duration;

 use snafu::ResultExt;
-use tokio::runtime::{Builder as RuntimeBuilder, Handle};
+use tokio::runtime::Builder as RuntimeBuilder;
 use tokio::sync::oneshot;
 pub use tokio::task::{JoinError, JoinHandle};

 use crate::error::*;
 use crate::metrics::*;
+use crate::runtime_default::DefaultRuntime;
+use crate::runtime_throttleable::ThrottleableRuntime;
+
+// configurations
+pub type Runtime = DefaultRuntime;

 static RUNTIME_ID: AtomicUsize = AtomicUsize::new(0);

-/// A runtime to run future tasks
-#[derive(Clone, Debug)]
-pub struct Runtime {
-    name: String,
-    handle: Handle,
-    // Used to receive a drop signal when dropper is dropped, inspired by databend
-    _dropper: Arc<Dropper>,
-}
-
 /// Dropping the dropper will cause runtime to shutdown.
 #[derive(Debug)]
 pub struct Dropper {
@@ -50,45 +46,42 @@ impl Drop for Dropper {
    }
 }

-impl Runtime {
-    pub fn builder() -> Builder {
+pub trait RuntimeTrait {
+    /// Get a runtime builder
+    fn builder() -> Builder {
        Builder::default()
    }

    /// Spawn a future and execute it in this thread pool
    ///
    /// Similar to tokio::runtime::Runtime::spawn()
-    pub fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+    fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
    where
        F: Future + Send + 'static,
-        F::Output: Send + 'static,
-    {
-        self.handle.spawn(future)
-    }
+        F::Output: Send + 'static;

    /// Run the provided function on an executor dedicated to blocking
    /// operations.
-    pub fn spawn_blocking<F, R>(&self, func: F) -> JoinHandle<R>
+    fn spawn_blocking<F, R>(&self, func: F) -> JoinHandle<R>
    where
        F: FnOnce() -> R + Send + 'static,
-        R: Send + 'static,
-    {
-        self.handle.spawn_blocking(func)
-    }
+        R: Send + 'static;

    /// Run a future to complete, this is the runtime's entry point
-    pub fn block_on<F: Future>(&self, future: F) -> F::Output {
-        self.handle.block_on(future)
-    }
+    fn block_on<F: Future>(&self, future: F) -> F::Output;

-    pub fn name(&self) -> &str {
-        &self.name
-    }
+    /// Get the name of the runtime
+    fn name(&self) -> &str;
+}
+
+pub trait BuilderBuild<R: RuntimeTrait> {
+    fn build(&mut self) -> Result<R>;
 }

 pub struct Builder {
    runtime_name: String,
    thread_name: String,
+    priority: Priority,
    builder: RuntimeBuilder,
 }

@@ -98,11 +91,17 @@ impl Default for Builder {
            runtime_name: format!("runtime-{}", RUNTIME_ID.fetch_add(1, Ordering::Relaxed)),
            thread_name: "default-worker".to_string(),
            builder: RuntimeBuilder::new_multi_thread(),
+            priority: Priority::VeryHigh,
        }
    }
 }

 impl Builder {
+    pub fn priority(&mut self, priority: Priority) -> &mut Self {
+        self.priority = priority;
+        self
+    }
+
    /// Sets the number of worker threads the Runtime will use.
    ///
    /// This can be any number above 0. The default value is the number of cores available to the system.
@@ -139,8 +138,10 @@ impl Builder {
        self.thread_name = val.into();
        self
    }
+}

-    pub fn build(&mut self) -> Result<Runtime> {
+impl BuilderBuild<DefaultRuntime> for Builder {
+    fn build(&mut self) -> Result<DefaultRuntime> {
        let runtime = self
            .builder
            .enable_all()
@@ -163,18 +164,53 @@ impl Builder {
        #[cfg(tokio_unstable)]
        register_collector(name.clone(), &handle);

-        Ok(Runtime {
-            name,
+        Ok(DefaultRuntime::new(
+            &name,
            handle,
-            _dropper: Arc::new(Dropper {
+            Arc::new(Dropper {
                close: Some(send_stop),
            }),
-        })
+        ))
+    }
+}
+
+impl BuilderBuild<ThrottleableRuntime> for Builder {
+    fn build(&mut self) -> Result<ThrottleableRuntime> {
+        let runtime = self
+            .builder
+            .enable_all()
+            .thread_name(self.thread_name.clone())
+            .on_thread_start(on_thread_start(self.thread_name.clone()))
+            .on_thread_stop(on_thread_stop(self.thread_name.clone()))
+            .on_thread_park(on_thread_park(self.thread_name.clone()))
+            .on_thread_unpark(on_thread_unpark(self.thread_name.clone()))
+            .build()
+            .context(BuildRuntimeSnafu)?;
+
+        let name = self.runtime_name.clone();
+        let handle = runtime.handle().clone();
+        let (send_stop, recv_stop) = oneshot::channel();
+        // Block the runtime to shutdown.
+        let _ = thread::Builder::new()
+            .name(format!("{}-blocker", self.thread_name))
+            .spawn(move || runtime.block_on(recv_stop));
+
+        #[cfg(tokio_unstable)]
+        register_collector(name.clone(), &handle);
+
+        ThrottleableRuntime::new(
+            &name,
+            self.priority,
+            handle,
+            Arc::new(Dropper {
+                close: Some(send_stop),
+            }),
+        )
    }
 }

 #[cfg(tokio_unstable)]
-pub fn register_collector(name: String, handle: &Handle) {
+pub fn register_collector(name: String, handle: &tokio::runtime::Handle) {
    let name = name.replace("-", "_");
    let monitor = tokio_metrics::RuntimeMonitor::new(handle);
    let collector = tokio_metrics_collector::RuntimeCollector::new(monitor, name);
@@ -213,8 +249,18 @@ fn on_thread_unpark(thread_name: String) -> impl Fn() + 'static {
    }
 }

+#[derive(Clone, Copy, Debug, Hash, PartialEq, Eq)]
+pub enum Priority {
+    VeryLow = 0,
+    Low = 1,
+    Middle = 2,
+    High = 3,
+    VeryHigh = 4,
+}
+
 #[cfg(test)]
 mod tests {
+
    use std::sync::Arc;
    use std::thread;
    use std::time::Duration;
@@ -235,12 +281,12 @@ mod tests {

    #[test]
    fn test_metric() {
-        let runtime = Builder::default()
+        let runtime: Runtime = Builder::default()
            .worker_threads(5)
            .thread_name("test_runtime_metric")
            .build()
            .unwrap();
-        // wait threads created
+        // wait threads create
        thread::sleep(Duration::from_millis(50));

        let _handle = runtime.spawn(async {
--- a/src/common/runtime/src/runtime_default.rs
+++ b/src/common/runtime/src/runtime_default.rs
@@ -0,0 +1,77 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::future::Future;
+use std::sync::Arc;
+
+use tokio::runtime::Handle;
+pub use tokio::task::JoinHandle;
+
+use crate::runtime::{Dropper, RuntimeTrait};
+use crate::Builder;
+
+/// A runtime to run future tasks
+#[derive(Clone, Debug)]
+pub struct DefaultRuntime {
+    name: String,
+    handle: Handle,
+    // Used to receive a drop signal when dropper is dropped, inspired by databend
+    _dropper: Arc<Dropper>,
+}
+
+impl DefaultRuntime {
+    pub(crate) fn new(name: &str, handle: Handle, dropper: Arc<Dropper>) -> Self {
+        Self {
+            name: name.to_string(),
+            handle,
+            _dropper: dropper,
+        }
+    }
+}
+
+impl RuntimeTrait for DefaultRuntime {
+    fn builder() -> Builder {
+        Builder::default()
+    }
+
+    /// Spawn a future and execute it in this thread pool
+    ///
+    /// Similar to tokio::runtime::Runtime::spawn()
+    fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        self.handle.spawn(future)
+    }
+
+    /// Run the provided function on an executor dedicated to blocking
+    /// operations.
+    fn spawn_blocking<F, R>(&self, func: F) -> JoinHandle<R>
+    where
+        F: FnOnce() -> R + Send + 'static,
+        R: Send + 'static,
+    {
+        self.handle.spawn_blocking(func)
+    }
+
+    /// Run a future to complete, this is the runtime's entry point
+    fn block_on<F: Future>(&self, future: F) -> F::Output {
+        self.handle.block_on(future)
+    }
+
+    fn name(&self) -> &str {
+        &self.name
+    }
+}
--- a/src/common/runtime/src/runtime_throttleable.rs
+++ b/src/common/runtime/src/runtime_throttleable.rs
@@ -0,0 +1,285 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fmt::Debug;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::task::{Context, Poll};
+use std::time::Duration;
+
+use futures::FutureExt;
+use ratelimit::Ratelimiter;
+use snafu::ResultExt;
+use tokio::runtime::Handle;
+pub use tokio::task::JoinHandle;
+use tokio::time::Sleep;
+
+use crate::error::{BuildRuntimeRateLimiterSnafu, Result};
+use crate::runtime::{Dropper, Priority, RuntimeTrait};
+use crate::Builder;
+
+struct RuntimeRateLimiter {
+    pub ratelimiter: Option<Ratelimiter>,
+}
+
+impl Debug for RuntimeRateLimiter {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RuntimeThrottleShareWithFuture")
+            .field(
+                "ratelimiter_max_tokens",
+                &self.ratelimiter.as_ref().map(|v| v.max_tokens()),
+            )
+            .field(
+                "ratelimiter_refill_amount",
+                &self.ratelimiter.as_ref().map(|v| v.refill_amount()),
+            )
+            .finish()
+    }
+}
+
+/// A runtime to run future tasks
+#[derive(Clone, Debug)]
+pub struct ThrottleableRuntime {
+    name: String,
+    handle: Handle,
+    shared_with_future: Arc<RuntimeRateLimiter>,
+    // Used to receive a drop signal when dropper is dropped, inspired by databend
+    _dropper: Arc<Dropper>,
+}
+
+impl ThrottleableRuntime {
+    pub(crate) fn new(
+        name: &str,
+        priority: Priority,
+        handle: Handle,
+        dropper: Arc<Dropper>,
+    ) -> Result<Self> {
+        Ok(Self {
+            name: name.to_string(),
+            handle,
+            shared_with_future: Arc::new(RuntimeRateLimiter {
+                ratelimiter: priority.ratelimiter_count()?,
+            }),
+            _dropper: dropper,
+        })
+    }
+}
+
+impl RuntimeTrait for ThrottleableRuntime {
+    fn builder() -> Builder {
+        Builder::default()
+    }
+
+    /// Spawn a future and execute it in this thread pool
+    ///
+    /// Similar to tokio::runtime::Runtime::spawn()
+    fn spawn<F>(&self, future: F) -> JoinHandle<F::Output>
+    where
+        F: Future + Send + 'static,
+        F::Output: Send + 'static,
+    {
+        self.handle
+            .spawn(ThrottleFuture::new(self.shared_with_future.clone(), future))
+    }
+
+    /// Run the provided function on an executor dedicated to blocking
+    /// operations.
+    fn spawn_blocking<F, R>(&self, func: F) -> JoinHandle<R>
+    where
+        F: FnOnce() -> R + Send + 'static,
+        R: Send + 'static,
+    {
+        self.handle.spawn_blocking(func)
+    }
+
+    /// Run a future to complete, this is the runtime's entry point
+    fn block_on<F: Future>(&self, future: F) -> F::Output {
+        self.handle.block_on(future)
+    }
+
+    fn name(&self) -> &str {
+        &self.name
+    }
+}
+
+enum State {
+    Pollable,
+    Throttled(Pin<Box<Sleep>>),
+}
+
+impl State {
+    fn unwrap_backoff(&mut self) -> &mut Pin<Box<Sleep>> {
+        match self {
+            State::Throttled(sleep) => sleep,
+            _ => panic!("unwrap_backoff failed"),
+        }
+    }
+}
+
+#[pin_project::pin_project]
+pub struct ThrottleFuture<F: Future + Send + 'static> {
+    #[pin]
+    future: F,
+
+    /// RateLimiter of this future
+    handle: Arc<RuntimeRateLimiter>,
+
+    state: State,
+}
+
+impl<F> ThrottleFuture<F>
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    fn new(handle: Arc<RuntimeRateLimiter>, future: F) -> Self {
+        Self {
+            future,
+            handle,
+            state: State::Pollable,
+        }
+    }
+}
+
+impl<F> Future for ThrottleFuture<F>
+where
+    F: Future + Send + 'static,
+    F::Output: Send + 'static,
+{
+    type Output = F::Output;
+
+    fn poll(self: std::pin::Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        let this = self.project();
+
+        match this.state {
+            State::Pollable => {}
+            State::Throttled(ref mut sleep) => match sleep.poll_unpin(cx) {
+                Poll::Ready(_) => {
+                    *this.state = State::Pollable;
+                }
+                Poll::Pending => return Poll::Pending,
+            },
+        };
+
+        if let Some(ratelimiter) = &this.handle.ratelimiter {
+            if let Err(wait) = ratelimiter.try_wait() {
+                *this.state = State::Throttled(Box::pin(tokio::time::sleep(wait)));
+                match this.state.unwrap_backoff().poll_unpin(cx) {
+                    Poll::Ready(_) => {
+                        *this.state = State::Pollable;
+                    }
+                    Poll::Pending => {
+                        return Poll::Pending;
+                    }
+                }
+            }
+        }
+
+        let poll_res = this.future.poll(cx);
+
+        match poll_res {
+            Poll::Ready(r) => Poll::Ready(r),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+impl Priority {
+    fn ratelimiter_count(&self) -> Result<Option<Ratelimiter>> {
+        let max = 8000;
+        let gen_per_10ms = match self {
+            Priority::VeryLow => Some(2000),
+            Priority::Low => Some(4000),
+            Priority::Middle => Some(6000),
+            Priority::High => Some(8000),
+            Priority::VeryHigh => None,
+        };
+        if let Some(gen_per_10ms) = gen_per_10ms {
+            Ratelimiter::builder(gen_per_10ms, Duration::from_millis(10)) // generate poll count per 10ms
+                .max_tokens(max) // reserved token for batch request
+                .build()
+                .context(BuildRuntimeRateLimiterSnafu)
+                .map(Some)
+        } else {
+            Ok(None)
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+
+    use tokio::fs::File;
+    use tokio::io::AsyncWriteExt;
+    use tokio::time::Duration;
+
+    use super::*;
+    use crate::runtime::BuilderBuild;
+
+    #[tokio::test]
+    async fn test_throttleable_runtime_spawn_simple() {
+        for p in [
+            Priority::VeryLow,
+            Priority::Low,
+            Priority::Middle,
+            Priority::High,
+            Priority::VeryHigh,
+        ] {
+            let runtime: ThrottleableRuntime = Builder::default()
+                .runtime_name("test")
+                .thread_name("test")
+                .worker_threads(8)
+                .priority(p)
+                .build()
+                .expect("Fail to create runtime");
+
+            // Spawn a simple future that returns 42
+            let handle = runtime.spawn(async {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+                42
+            });
+            let result = handle.await.expect("Task panicked");
+            assert_eq!(result, 42);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_throttleable_runtime_spawn_complex() {
+        let tempdir = tempfile::tempdir().unwrap();
+        for p in [
+            Priority::VeryLow,
+            Priority::Low,
+            Priority::Middle,
+            Priority::High,
+            Priority::VeryHigh,
+        ] {
+            let runtime: ThrottleableRuntime = Builder::default()
+                .runtime_name("test")
+                .thread_name("test")
+                .worker_threads(8)
+                .priority(p)
+                .build()
+                .expect("Fail to create runtime");
+            let tempdirpath = tempdir.path().to_path_buf();
+            let handle = runtime.spawn(async move {
+                let mut file = File::create(tempdirpath.join("test.txt")).await.unwrap();
+                file.write_all(b"Hello, world!").await.unwrap();
+                42
+            });
+            let result = handle.await.expect("Task panicked");
+            assert_eq!(result, 42);
+        }
+    }
+}
--- a/src/common/telemetry/Cargo.toml
+++ b/src/common/telemetry/Cargo.toml
@@ -26,13 +26,13 @@ opentelemetry = { version = "0.21.0", default-features = false, features = [
 opentelemetry-otlp = { version = "0.14.0", features = ["tokio"] }
 opentelemetry-semantic-conventions = "0.13.0"
 opentelemetry_sdk = { version = "0.21.0", features = ["rt-tokio"] }
-parking_lot = { version = "0.12" }
+parking_lot.workspace = true
 prometheus.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
 tracing = "0.1"
-tracing-appender = "0.2"
+tracing-appender.workspace = true
 tracing-log = "0.1"
 tracing-opentelemetry = "0.22.0"
-tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "fmt"] }
+tracing-subscriber.workspace = true
--- a/src/datanode/src/alive_keeper.rs
+++ b/src/datanode/src/alive_keeper.rs
@@ -427,7 +427,8 @@ mod test {
        common_telemetry::init_default_ut_logging();
        let mut region_server = mock_region_server();
        let mut engine_env = TestEnv::with_prefix("region-alive-keeper");
-        let engine = Arc::new(engine_env.create_engine(MitoConfig::default()).await);
+        let engine = engine_env.create_engine(MitoConfig::default()).await;
+        let engine = Arc::new(engine);
        region_server.register_engine(engine.clone());

        let alive_keeper = Arc::new(RegionAliveKeeper::new(region_server.clone(), 100));
--- a/src/datanode/src/config.rs
+++ b/src/datanode/src/config.rs
@@ -30,7 +30,7 @@ use servers::heartbeat_options::HeartbeatOptions;
 use servers::http::HttpOptions;
 use servers::Mode;

-pub const DEFAULT_OBJECT_STORE_CACHE_SIZE: ReadableSize = ReadableSize::mb(256);
+pub const DEFAULT_OBJECT_STORE_CACHE_SIZE: ReadableSize = ReadableSize::gb(1);

 /// Default data home in file storage
 const DEFAULT_DATA_HOME: &str = "/tmp/greptimedb";
--- a/src/datanode/src/datanode.rs
+++ b/src/datanode/src/datanode.rs
@@ -23,6 +23,7 @@ use common_base::Plugins;
 use common_error::ext::BoxedError;
 use common_greptimedb_telemetry::GreptimeDBTelemetryTask;
 use common_meta::key::datanode_table::{DatanodeTableManager, DatanodeTableValue};
+use common_meta::key::{SchemaMetadataManager, SchemaMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::wal_options_allocator::prepare_wal_options;
 pub use common_procedure::options::ProcedureConfig;
@@ -207,7 +208,10 @@ impl DatanodeBuilder {
            (Box::new(NoopRegionServerEventListener) as _, None)
        };

-        let region_server = self.new_region_server(region_event_listener).await?;
+        let schema_metadata_manager = Arc::new(SchemaMetadataManager::new(kv_backend.clone()));
+        let region_server = self
+            .new_region_server(schema_metadata_manager, region_event_listener)
+            .await?;

        let datanode_table_manager = DatanodeTableManager::new(kv_backend.clone());
        let table_values = datanode_table_manager
@@ -312,6 +316,7 @@ impl DatanodeBuilder {

    async fn new_region_server(
        &self,
+        schema_metadata_manager: SchemaMetadataManagerRef,
        event_listener: RegionServerEventListenerRef,
    ) -> Result<RegionServer> {
        let opts: &DatanodeOptions = &self.opts;
@@ -340,8 +345,13 @@ impl DatanodeBuilder {
        );

        let object_store_manager = Self::build_object_store_manager(&opts.storage).await?;
-        let engines =
-            Self::build_store_engines(opts, object_store_manager, self.plugins.clone()).await?;
+        let engines = Self::build_store_engines(
+            opts,
+            object_store_manager,
+            schema_metadata_manager,
+            self.plugins.clone(),
+        )
+        .await?;
        for engine in engines {
            region_server.register_engine(engine);
        }
@@ -355,6 +365,7 @@ impl DatanodeBuilder {
    async fn build_store_engines(
        opts: &DatanodeOptions,
        object_store_manager: ObjectStoreManagerRef,
+        schema_metadata_manager: SchemaMetadataManagerRef,
        plugins: Plugins,
    ) -> Result<Vec<RegionEngineRef>> {
        let mut engines = vec![];
@@ -365,6 +376,7 @@ impl DatanodeBuilder {
                        opts,
                        object_store_manager.clone(),
                        config.clone(),
+                        schema_metadata_manager.clone(),
                        plugins.clone(),
                    )
                    .await?;
@@ -390,6 +402,7 @@ impl DatanodeBuilder {
        opts: &DatanodeOptions,
        object_store_manager: ObjectStoreManagerRef,
        config: MitoConfig,
+        schema_metadata_manager: SchemaMetadataManagerRef,
        plugins: Plugins,
    ) -> Result<MitoEngine> {
        let mito_engine = match &opts.wal {
@@ -399,6 +412,7 @@ impl DatanodeBuilder {
                Self::build_raft_engine_log_store(&opts.storage.data_home, raft_engine_config)
                    .await?,
                object_store_manager,
+                schema_metadata_manager,
                plugins,
            )
            .await
@@ -429,6 +443,7 @@ impl DatanodeBuilder {
                    config,
                    Self::build_kafka_log_store(kafka_config, global_index_collector).await?,
                    object_store_manager,
+                    schema_metadata_manager,
                    plugins,
                )
                .await
--- a/src/datanode/src/region_server.rs
+++ b/src/datanode/src/region_server.rs
@@ -1355,7 +1355,7 @@ mod tests {
    }

    #[tokio::test]
-    async fn test_region_server_parallism() {
+    async fn test_region_server_parallelism() {
        let p = RegionServerParallelism::from_opts(2, Duration::from_millis(1)).unwrap();
        let first_query = p.acquire().await;
        assert!(first_query.is_ok());
--- a/src/datanode/src/tests.rs
+++ b/src/datanode/src/tests.rs
@@ -23,6 +23,7 @@ use common_function::function::FunctionRef;
 use common_function::scalars::aggregate::AggregateFunctionMetaRef;
 use common_query::prelude::ScalarUdf;
 use common_query::Output;
+use common_runtime::runtime::{BuilderBuild, RuntimeTrait};
 use common_runtime::Runtime;
 use datafusion_expr::LogicalPlan;
 use query::dataframe::DataFrame;
--- a/src/datatypes/src/error.rs
+++ b/src/datatypes/src/error.rs
@@ -189,6 +189,13 @@ pub enum Error {
        location: Location,
    },

+    #[snafu(display("Invalid JSON text: {}", value))]
+    InvalidJson {
+        value: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Value exceeds the precision {} bound", precision))]
    ValueExceedsPrecision {
        precision: u8,
@@ -222,7 +229,8 @@ impl ErrorExt for Error {
            | DefaultValueType { .. }
            | DuplicateMeta { .. }
            | InvalidTimestampPrecision { .. }
-            | InvalidPrecisionOrScale { .. } => StatusCode::InvalidArguments,
+            | InvalidPrecisionOrScale { .. }
+            | InvalidJson { .. } => StatusCode::InvalidArguments,

            ValueExceedsPrecision { .. }
            | CastType { .. }
--- a/src/datatypes/src/lib.rs
+++ b/src/datatypes/src/lib.rs
@@ -13,6 +13,7 @@
 // limitations under the License.

 #![feature(let_chains)]
+#![feature(assert_matches)]

 pub mod arrow_array;
 pub mod data_type;
--- a/src/datatypes/src/scalars.rs
+++ b/src/datatypes/src/scalars.rs
@@ -203,7 +203,7 @@ impl Scalar for bool {
    }
 }

-impl<'a> ScalarRef<'a> for bool {
+impl ScalarRef<'_> for bool {
    type ScalarType = bool;

    #[inline]
@@ -273,7 +273,7 @@ impl Scalar for Date {
    }
 }

-impl<'a> ScalarRef<'a> for Date {
+impl ScalarRef<'_> for Date {
    type ScalarType = Date;

    fn to_owned_scalar(&self) -> Self::ScalarType {
@@ -294,7 +294,7 @@ impl Scalar for Decimal128 {
    }
 }

-impl<'a> ScalarRef<'a> for Decimal128 {
+impl ScalarRef<'_> for Decimal128 {
    type ScalarType = Decimal128;

    fn to_owned_scalar(&self) -> Self::ScalarType {
@@ -315,7 +315,7 @@ impl Scalar for DateTime {
    }
 }

-impl<'a> ScalarRef<'a> for DateTime {
+impl ScalarRef<'_> for DateTime {
    type ScalarType = DateTime;

    fn to_owned_scalar(&self) -> Self::ScalarType {
--- a/src/datatypes/src/types/cast.rs
+++ b/src/datatypes/src/types/cast.rs
@@ -82,8 +82,8 @@ pub fn cast_with_opt(
    }
 }

-/// Return true if the src_value can be casted to dest_type,
-/// Otherwise, return false.
+/// Return true if the src_value can be casted to dest_type, Otherwise, return false.
+///
 /// Notice: this function does not promise that the `cast_with_opt` will succeed,
 /// it only checks whether the src_value can be casted to dest_type.
 pub fn can_cast_type(src_value: &Value, dest_type: &ConcreteDataType) -> bool {
--- a/src/datatypes/src/types/primitive_type.rs
+++ b/src/datatypes/src/types/primitive_type.rs
@@ -83,9 +83,10 @@ pub trait LogicalPrimitiveType: 'static + Sized {
    fn cast_value_ref(value: ValueRef) -> Result<Option<Self::Wrapper>>;
 }

-/// A new type for [WrapperType], complement the `Ord` feature for it. Wrapping non ordered
-/// primitive types like `f32` and `f64` in `OrdPrimitive` can make them be used in places that
-/// require `Ord`. For example, in `Median` UDAFs.
+/// A new type for [WrapperType], complement the `Ord` feature for it.
+///
+/// Wrapping non ordered primitive types like `f32` and `f64` in `OrdPrimitive`
+/// can make them be used in places that require `Ord`. For example, in `Median` UDAFs.
 #[derive(Debug, Clone, Copy, PartialEq)]
 pub struct OrdPrimitive<T: WrapperType>(pub T);

--- a/Show More
+++ b/Show More