feat: cache logical region's metadata

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>
2025-12-26 16:10:02 +00:00 · 2024-10-12 16:16:25 +08:00
605 changed files with 10566 additions and 36434 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 numpy
+      run: pip install pyarrow

    - 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=1500m \
+        --set resources.limits.cpu=1000m \
        --set resources.limits.memory=2Gi \
        --set auth.rbac.create=false \
        --set auth.rbac.token.enabled=false \
--- a/.github/scripts/check-install-script.sh
+++ b/.github/scripts/check-install-script.sh
@@ -1,14 +0,0 @@
-#!/bin/sh
-
-set -e
-
-# Get the latest version of github.com/GreptimeTeam/greptimedb
-VERSION=$(curl -s https://api.github.com/repos/GreptimeTeam/greptimedb/releases/latest | jq -r '.tag_name')
-
-echo "Downloading the latest version: $VERSION"
-
-# Download the install script
-curl -fsSL https://raw.githubusercontent.com/greptimeteam/greptimedb/main/scripts/install.sh | sh -s $VERSION
-
-# Execute the `greptime` command
-./greptime --version
--- 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_migrate_metric_regions", "fuzz_failover_mito_regions", "fuzz_failover_metric_regions"]
+        target: ["fuzz_migrate_mito_regions", "fuzz_failover_mito_regions", "fuzz_failover_metric_regions"]
        mode:
          - name: "Remote WAL"
            minio: true
@@ -449,12 +449,6 @@ 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: |
@@ -694,7 +688,7 @@ jobs:
        with:
          python-version: '3.10'
      - name: Install PyArrow Package
-        run: pip install pyarrow numpy
+        run: pip install pyarrow
      - 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
@@ -22,10 +22,6 @@ jobs:
        uses: actions/checkout@v4
        with:
          fetch-depth: 0
-
-      - name: Check install.sh
-        run: ./.github/scripts/check-install-script.sh
-
      - name: Run sqlness test
        uses: ./.github/actions/sqlness-test
        with:
@@ -96,7 +92,7 @@ jobs:
        with:
          python-version: "3.10"
      - name: Install PyArrow Package
-        run: pip install pyarrow numpy
+        run: pip install pyarrow
      - name: Install WSL distribution
        uses: Vampire/setup-wsl@v2
        with:
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -91,7 +91,7 @@ env:
  # The scheduled version is '${{ env.NEXT_RELEASE_VERSION }}-nightly-YYYYMMDD', like v0.2.0-nigthly-20230313;
  NIGHTLY_RELEASE_PREFIX: nightly
  # Note: The NEXT_RELEASE_VERSION should be modified manually by every formal release.
-  NEXT_RELEASE_VERSION: v0.11.0
+  NEXT_RELEASE_VERSION: v0.10.0

 # Permission reference: https://docs.github.com/en/actions/using-jobs/assigning-permissions-to-jobs
 permissions:
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -17,6 +17,6 @@ repos:
    -    id: fmt
    -    id: clippy
         args: ["--workspace", "--all-targets", "--all-features", "--", "-D", "warnings"]
-         stages: [pre-push]
+         stages: [push]
    -    id: cargo-check
         args: ["--workspace", "--all-targets", "--all-features"]
--- a/Cargo.lock
+++ b/Cargo.lock
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -2,27 +2,24 @@
 members = [
    "src/api",
    "src/auth",
-    "src/cache",
    "src/catalog",
+    "src/cache",
    "src/client",
    "src/cmd",
    "src/common/base",
    "src/common/catalog",
    "src/common/config",
    "src/common/datasource",
-    "src/common/decimal",
    "src/common/error",
    "src/common/frontend",
    "src/common/function",
+    "src/common/macro",
    "src/common/greptimedb-telemetry",
    "src/common/grpc",
    "src/common/grpc-expr",
-    "src/common/macro",
    "src/common/mem-prof",
    "src/common/meta",
-    "src/common/options",
    "src/common/plugins",
-    "src/common/pprof",
    "src/common/procedure",
    "src/common/procedure-test",
    "src/common/query",
@@ -32,6 +29,7 @@ members = [
    "src/common/telemetry",
    "src/common/test-util",
    "src/common/time",
+    "src/common/decimal",
    "src/common/version",
    "src/common/wal",
    "src/datanode",
@@ -39,7 +37,6 @@ members = [
    "src/file-engine",
    "src/flow",
    "src/frontend",
-    "src/index",
    "src/log-store",
    "src/meta-client",
    "src/meta-srv",
@@ -59,6 +56,7 @@ members = [
    "src/sql",
    "src/store-api",
    "src/table",
+    "src/index",
    "tests-fuzz",
    "tests-integration",
    "tests/runner",
@@ -66,7 +64,7 @@ members = [
 resolver = "2"

 [workspace.package]
-version = "0.10.2"
+version = "0.9.3"
 edition = "2021"
 license = "Apache-2.0"

@@ -118,18 +116,17 @@ datafusion-sql = { git = "https://github.com/waynexia/arrow-datafusion.git", rev
 datafusion-substrait = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "7823ef2f63663907edab46af0d51359900f608d6" }
 derive_builder = "0.12"
 dotenv = "0.15"
-etcd-client = "0.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 = "a875e976441188028353f7274a46a7e6e065c5d4" }
-hex = "0.4"
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "0b4f7c8ab06399f6b90e1626e8d5b9697cb33bb9" }
 humantime = "2.1"
 humantime-serde = "1.1"
 itertools = "0.10"
-jsonb = { git = "https://github.com/databendlabs/jsonb.git", rev = "8c8d2fc294a39f3ff08909d60f718639cfba3875", default-features = false }
+jsonb = { git = "https://github.com/datafuselabs/jsonb.git", rev = "46ad50fc71cf75afbf98eec455f7892a6387c1fc", default-features = false }
 lazy_static = "1.4"
-meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "a10facb353b41460eeb98578868ebf19c2084fac" }
+meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd" }
 mockall = "0.11.4"
 moka = "0.12"
 notify = "6.1"
@@ -140,20 +137,17 @@ opentelemetry-proto = { version = "0.5", features = [
    "metrics",
    "trace",
    "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.3", features = ["ser"] }
+promql-parser = { version = "0.4" }
 prost = "0.12"
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.8"
-ratelimit = "0.9"
 regex = "1.8"
-regex-automata = "0.4"
+regex-automata = { version = "0.4" }
 reqwest = { version = "0.12", default-features = false, features = [
    "json",
    "rustls-tls-native-roots",
@@ -171,7 +165,7 @@ schemars = "0.8"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = { version = "1.0", features = ["float_roundtrip"] }
 serde_with = "3"
-shadow-rs = "0.35"
+shadow-rs = "0.31"
 similar-asserts = "1.6.0"
 smallvec = { version = "1", features = ["serde"] }
 snafu = "0.8"
@@ -182,16 +176,13 @@ sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "5
 ] }
 strum = { version = "0.25", features = ["derive"] }
 tempfile = "3"
-tokio = { version = "1.40", features = ["full"] }
+tokio = { version = "1.36", features = ["full"] }
 tokio-postgres = "0.7"
-tokio-stream = "0.1"
+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 = "0.4"
-tracing-appender = "0.2"
-tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "fmt"] }
-typetag = "0.2"
+tower = { version = "0.4" }
 uuid = { version = "1.7", features = ["serde", "v4", "fast-rng"] }
 zstd = "0.13"

@@ -216,9 +207,7 @@ common-grpc-expr = { path = "src/common/grpc-expr" }
 common-macro = { path = "src/common/macro" }
 common-mem-prof = { path = "src/common/mem-prof" }
 common-meta = { path = "src/common/meta" }
-common-options = { path = "src/common/options" }
 common-plugins = { path = "src/common/plugins" }
-common-pprof = { path = "src/common/pprof" }
 common-procedure = { path = "src/common/procedure" }
 common-procedure-test = { path = "src/common/procedure-test" }
 common-query = { path = "src/common/query" }
@@ -264,12 +253,10 @@ 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 = "a10facb353b41460eeb98578868ebf19c2084fac"
+rev = "80eb97c24c88af4dd9a86f8bbaf50e741d4eb8cd"

 [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-10-19-a5c00e85-20241024184445
+DEV_BUILDER_IMAGE_TAG ?= 2024-06-06-5674c14f-20240920110415
 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 & Cost-Effective Time Series Database for Metrics, Logs, and Events</h2>
+<h2 align="center">Unified Time Series Database for Metrics, Logs, and Events</h2>

 <div align="center">
 <h3 align="center">
@@ -48,21 +48,9 @@
 </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 & 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.
+**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.

 ## Why GreptimeDB

--- a/config/config.md
+++ b/config/config.md
@@ -83,7 +83,6 @@
 | `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. |
 | `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 +92,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. 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.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.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`**. |
@@ -109,11 +108,6 @@
 | `storage.sas_token` | String | Unset | The sas token of the azure account.<br/>**It's only used when the storage type is `Azblob`**. |
 | `storage.endpoint` | String | Unset | The endpoint of the S3 service.<br/>**It's only used when the storage type is `S3`, `Oss`, `Gcs` and `Azblob`**. |
 | `storage.region` | String | Unset | The region of the S3 service.<br/>**It's only used when the storage type is `S3`, `Oss`, `Gcs` and `Azblob`**. |
-| `storage.http_client` | -- | -- | The http client options to the storage.<br/>**It's only used when the storage type is `S3`, `Oss`, `Gcs` and `Azblob`**. |
-| `storage.http_client.pool_max_idle_per_host` | Integer | `1024` | The maximum idle connection per host allowed in the pool. |
-| `storage.http_client.connect_timeout` | String | `30s` | The timeout for only the connect phase of a http client. |
-| `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.<br/>Also considered a total deadline. |
-| `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. |
 | `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines. |
 | `region_engine.mito` | -- | -- | The Mito engine options. |
 | `region_engine.mito.num_workers` | Integer | `8` | Number of region workers. |
@@ -121,9 +115,7 @@
 | `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_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.max_background_jobs` | Integer | `4` | Max number of running background jobs |
 | `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`. |
@@ -131,9 +123,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. It is recommended to enable it when using object storage for better performance. |
+| `region_engine.mito.enable_experimental_write_cache` | Bool | `false` | Whether to enable the experimental write cache. |
 | `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 | `1GiB` | Capacity for write cache. If your disk space is sufficient, it is recommended to set it larger. |
+| `region_engine.mito.experimental_write_cache_size` | String | `512MB` | Capacity for write cache. |
 | `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. |
@@ -417,12 +409,11 @@
 | `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. |
 | `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. 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.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.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,11 +428,6 @@
 | `storage.sas_token` | String | Unset | The sas token of the azure account.<br/>**It's only used when the storage type is `Azblob`**. |
 | `storage.endpoint` | String | Unset | The endpoint of the S3 service.<br/>**It's only used when the storage type is `S3`, `Oss`, `Gcs` and `Azblob`**. |
 | `storage.region` | String | Unset | The region of the S3 service.<br/>**It's only used when the storage type is `S3`, `Oss`, `Gcs` and `Azblob`**. |
-| `storage.http_client` | -- | -- | The http client options to the storage.<br/>**It's only used when the storage type is `S3`, `Oss`, `Gcs` and `Azblob`**. |
-| `storage.http_client.pool_max_idle_per_host` | Integer | `1024` | The maximum idle connection per host allowed in the pool. |
-| `storage.http_client.connect_timeout` | String | `30s` | The timeout for only the connect phase of a http client. |
-| `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.<br/>Also considered a total deadline. |
-| `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. |
 | `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines. |
 | `region_engine.mito` | -- | -- | The Mito engine options. |
 | `region_engine.mito.num_workers` | Integer | `8` | Number of region workers. |
@@ -449,9 +435,7 @@
 | `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_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.max_background_jobs` | Integer | `4` | Max number of running background jobs |
 | `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` |
@@ -459,9 +443,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. It is recommended to enable it when using object storage for better performance. |
+| `region_engine.mito.enable_experimental_write_cache` | Bool | `false` | Whether to enable the experimental write cache. |
 | `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 | `1GiB` | Capacity for write cache. If your disk space is sufficient, it is recommended to set it larger. |
+| `region_engine.mito.experimental_write_cache_size` | String | `512MB` | Capacity for write cache. |
 | `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
@@ -213,17 +213,6 @@ create_index = true
 ## **It's only used when the provider is `kafka`**.
 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
-## them as critical errors.
-overwrite_entry_start_id = false
-
 # The Kafka SASL configuration.
 # **It's only used when the provider is `kafka`**.
 # Available SASL mechanisms:
@@ -294,14 +283,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. It is recommended to configure it when using object storage for better performance.
+## Cache configuration for object storage such as 'S3' etc.
 ## The local file cache directory.
 ## @toml2docs:none-default
 cache_path = "/path/local_cache"

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

 ## The S3 bucket name.
 ## **It's only used when the storage type is `S3`, `Oss` and `Gcs`**.
@@ -375,23 +364,6 @@ endpoint = "https://s3.amazonaws.com"
 ## @toml2docs:none-default
 region = "us-west-2"

-## The http client options to the storage.
-## **It's only used when the storage type is `S3`, `Oss`, `Gcs` and `Azblob`**.
-[storage.http_client]
-
-## The maximum idle connection per host allowed in the pool.
-pool_max_idle_per_host = 1024
-
-## The timeout for only the connect phase of a http client.
-connect_timeout = "30s"
-
-## The total request timeout, applied from when the request starts connecting until the response body has finished.
-## Also considered a total deadline.
-timeout = "30s"
-
-## The timeout for idle sockets being kept-alive.
-pool_idle_timeout = "90s"
-
 # Custom storage options
 # [[storage.providers]]
 # name = "S3"
@@ -433,17 +405,8 @@ manifest_checkpoint_distance = 10
 ## Whether to compress manifest and checkpoint file by gzip (default false).
 compress_manifest = false

-## 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
+## Max number of running background jobs
+max_background_jobs = 4

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

-## Whether to enable the experimental write cache. It is recommended to enable it when using object storage for better performance.
+## Whether to enable the experimental write cache.
 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. If your disk space is sufficient, it is recommended to set it larger.
-experimental_write_cache_size = "1GiB"
+## Capacity for write cache.
+experimental_write_cache_size = "512MB"

 ## TTL for write cache.
 ## @toml2docs:none-default
@@ -663,7 +626,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
@@ -237,17 +237,6 @@ backoff_base = 2
 ## **It's only used when the provider is `kafka`**.
 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
-## them as critical errors.
-overwrite_entry_start_id = false
-
 # The Kafka SASL configuration.
 # **It's only used when the provider is `kafka`**.
 # Available SASL mechanisms:
@@ -332,14 +321,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. It is recommended to configure it when using object storage for better performance.
+## Cache configuration for object storage such as 'S3' etc.
 ## The local file cache directory.
 ## @toml2docs:none-default
 cache_path = "/path/local_cache"

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

 ## The S3 bucket name.
 ## **It's only used when the storage type is `S3`, `Oss` and `Gcs`**.
@@ -413,23 +402,6 @@ endpoint = "https://s3.amazonaws.com"
 ## @toml2docs:none-default
 region = "us-west-2"

-## The http client options to the storage.
-## **It's only used when the storage type is `S3`, `Oss`, `Gcs` and `Azblob`**.
-[storage.http_client]
-
-## The maximum idle connection per host allowed in the pool.
-pool_max_idle_per_host = 1024
-
-## The timeout for only the connect phase of a http client.
-connect_timeout = "30s"
-
-## The total request timeout, applied from when the request starts connecting until the response body has finished.
-## Also considered a total deadline.
-timeout = "30s"
-
-## The timeout for idle sockets being kept-alive.
-pool_idle_timeout = "90s"
-
 # Custom storage options
 # [[storage.providers]]
 # name = "S3"
@@ -471,17 +443,8 @@ manifest_checkpoint_distance = 10
 ## Whether to compress manifest and checkpoint file by gzip (default false).
 compress_manifest = false

-## 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
+## Max number of running background jobs
+max_background_jobs = 4

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

-## Whether to enable the experimental write cache. It is recommended to enable it when using object storage for better performance.
+## Whether to enable the experimental write cache.
 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. If your disk space is sufficient, it is recommended to set it larger.
-experimental_write_cache_size = "1GiB"
+## Capacity for write cache.
+experimental_write_cache_size = "512MB"

 ## TTL for write cache.
 ## @toml2docs:none-default
@@ -707,7 +670,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/benchmarks/log/README.md
+++ b/docs/benchmarks/log/README.md
@@ -48,4 +48,4 @@ Please refer to [SQL query](./query.sql) for GreptimeDB and Clickhouse, and [que

 ## Addition
 - You can tune GreptimeDB's configuration to get better performance.
- You can setup GreptimeDB to use S3 as storage, see [here](https://docs.greptime.com/user-guide/deployments/configuration#storage-options).
+- You can setup GreptimeDB to use S3 as storage, see [here](https://docs.greptime.com/user-guide/operations/configuration/#storage-options).
--- 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
@@ -1,16 +0,0 @@
-# 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/docs/how-to/how-to-profile-cpu.md
+++ b/docs/how-to/how-to-profile-cpu.md
@@ -1,5 +1,11 @@
 # Profiling CPU

+## Build GreptimeDB with `pprof` feature
+
+```bash
+cargo build --features=pprof
+```
+
 ## HTTP API
 Sample at 99 Hertz, for 5 seconds, output report in [protobuf format](https://github.com/google/pprof/blob/master/proto/profile.proto).
 ```bash
--- a/docs/how-to/how-to-profile-memory.md
+++ b/docs/how-to/how-to-profile-memory.md
@@ -18,6 +18,12 @@ sudo apt install libjemalloc-dev
 curl https://raw.githubusercontent.com/brendangregg/FlameGraph/master/flamegraph.pl > ./flamegraph.pl
 ```

+### Build GreptimeDB with `mem-prof` feature.
+
+```bash
+cargo build --features=mem-prof
+```
+
 ## Profiling

 Start GreptimeDB instance with environment variables:
--- a/grafana/README.md
+++ b/grafana/README.md
@@ -5,13 +5,6 @@ GreptimeDB's official Grafana dashboard.

 Status notify: we are still working on this config. It's expected to change frequently in the recent days. Please feel free to submit your feedback and/or contribution to this dashboard 🤗

-If you use Helm [chart](https://github.com/GreptimeTeam/helm-charts) to deploy GreptimeDB cluster, you can enable self-monitoring by setting the following values in your Helm chart:
-
- `monitoring.enabled=true`: Deploys a standalone GreptimeDB instance dedicated to monitoring the cluster;
- `grafana.enabled=true`: Deploys Grafana and automatically imports the monitoring dashboard;
-
-The standalone GreptimeDB instance will collect metrics from your cluster and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/nightly/user-guide/deployments/deploy-on-kubernetes/getting-started).
-
 # How to use

 ## `greptimedb.json`
--- a/grafana/greptimedb-cluster.json
+++ b/grafana/greptimedb-cluster.json
--- a/grafana/greptimedb.json
+++ b/grafana/greptimedb.json
@@ -409,39 +409,7 @@
      "fieldConfig": {
        "defaults": {
          "color": {
-            "mode": "palette-classic"
-          },
-          "custom": {
-            "axisBorderShow": false,
-            "axisCenteredZero": false,
-            "axisColorMode": "text",
-            "axisLabel": "",
-            "axisPlacement": "auto",
-            "barAlignment": 0,
-            "drawStyle": "line",
-            "fillOpacity": 0,
-            "gradientMode": "none",
-            "hideFrom": {
-              "legend": false,
-              "tooltip": false,
-              "viz": false
-            },
-            "insertNulls": false,
-            "lineInterpolation": "linear",
-            "lineWidth": 1,
-            "pointSize": 5,
-            "scaleDistribution": {
-              "type": "linear"
-            },
-            "showPoints": "auto",
-            "spanNulls": false,
-            "stacking": {
-              "group": "A",
-              "mode": "none"
-            },
-            "thresholdsStyle": {
-              "mode": "off"
-            }
+            "mode": "thresholds"
          },
          "fieldMinMax": false,
          "mappings": [],
@@ -470,16 +438,18 @@
      },
      "id": 27,
      "options": {
-        "legend": {
-          "calcs": [],
-          "displayMode": "list",
-          "placement": "bottom",
-          "showLegend": true
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": ["lastNotNull"],
+          "fields": "",
+          "values": false
        },
-        "tooltip": {
-          "mode": "single",
-          "sort": "none"
-        }
+        "text": {},
+        "textMode": "auto",
+        "wideLayout": true
      },
      "pluginVersion": "10.2.3",
      "targets": [
@@ -497,7 +467,7 @@
        }
      ],
      "title": "CPU",
-      "type": "timeseries"
+      "type": "stat"
    },
    {
      "datasource": {
@@ -507,39 +477,7 @@
      "fieldConfig": {
        "defaults": {
          "color": {
-            "mode": "palette-classic"
-          },
-          "custom": {
-            "axisBorderShow": false,
-            "axisCenteredZero": false,
-            "axisColorMode": "text",
-            "axisLabel": "",
-            "axisPlacement": "auto",
-            "barAlignment": 0,
-            "drawStyle": "line",
-            "fillOpacity": 0,
-            "gradientMode": "none",
-            "hideFrom": {
-              "legend": false,
-              "tooltip": false,
-              "viz": false
-            },
-            "insertNulls": false,
-            "lineInterpolation": "linear",
-            "lineWidth": 1,
-            "pointSize": 5,
-            "scaleDistribution": {
-              "type": "linear"
-            },
-            "showPoints": "auto",
-            "spanNulls": false,
-            "stacking": {
-              "group": "A",
-              "mode": "none"
-            },
-            "thresholdsStyle": {
-              "mode": "off"
-            }
+            "mode": "thresholds"
          },
          "decimals": 0,
          "fieldMinMax": false,
@@ -565,16 +503,18 @@
      },
      "id": 28,
      "options": {
-        "legend": {
-          "calcs": [],
-          "displayMode": "list",
-          "placement": "bottom",
-          "showLegend": true
+        "colorMode": "value",
+        "graphMode": "area",
+        "justifyMode": "auto",
+        "orientation": "auto",
+        "reduceOptions": {
+          "calcs": ["lastNotNull"],
+          "fields": "",
+          "values": false
        },
-        "tooltip": {
-          "mode": "single",
-          "sort": "none"
-        }
+        "text": {},
+        "textMode": "auto",
+        "wideLayout": true
      },
      "pluginVersion": "10.2.3",
      "targets": [
@@ -592,7 +532,7 @@
        }
      ],
      "title": "Memory",
-      "type": "timeseries"
+      "type": "stat"
    },
    {
      "collapsed": false,
@@ -3395,6 +3335,6 @@
  "timezone": "",
  "title": "GreptimeDB",
  "uid": "e7097237-669b-4f8d-b751-13067afbfb68",
-  "version": 16,
+  "version": 15,
  "weekStart": ""
 }
--- a/rust-toolchain.toml
+++ b/rust-toolchain.toml
@@ -1,2 +1,3 @@
 [toolchain]
-channel = "nightly-2024-10-19"
+channel = "nightly-2024-06-06"
+
--- a/scripts/install.sh
+++ b/scripts/install.sh
@@ -1,4 +1,4 @@
-#!/bin/sh
+#!/usr/bin/env bash

 set -ue

@@ -15,7 +15,7 @@ GITHUB_ORG=GreptimeTeam
 GITHUB_REPO=greptimedb
 BIN=greptime

-get_os_type() {
+function get_os_type() {
  os_type="$(uname -s)"

  case "$os_type" in
@@ -31,7 +31,7 @@ get_os_type() {
  esac
 }

-get_arch_type() {
+function get_arch_type() {
  arch_type="$(uname -m)"

  case "$arch_type" in
@@ -53,7 +53,7 @@ get_arch_type() {
  esac
 }

-download_artifact() {
+function download_artifact() {
  if [ -n "${OS_TYPE}" ] && [ -n "${ARCH_TYPE}" ]; then
    # Use the latest stable released version.
    # GitHub API reference: https://docs.github.com/en/rest/releases/releases?apiVersion=2022-11-28#get-the-latest-release.
--- a/src/api/src/helper.rs
+++ b/src/api/src/helper.rs
@@ -17,11 +17,10 @@ use std::sync::Arc;
 use common_base::BitVec;
 use common_decimal::decimal128::{DECIMAL128_DEFAULT_SCALE, DECIMAL128_MAX_PRECISION};
 use common_decimal::Decimal128;
+use common_time::interval::IntervalUnit;
 use common_time::time::Time;
 use common_time::timestamp::TimeUnit;
-use common_time::{
-    Date, DateTime, IntervalDayTime, IntervalMonthDayNano, IntervalYearMonth, Timestamp,
-};
+use common_time::{Date, DateTime, Interval, Timestamp};
 use datatypes::prelude::{ConcreteDataType, ValueRef};
 use datatypes::scalars::ScalarVector;
 use datatypes::types::{
@@ -36,14 +35,15 @@ use datatypes::vectors::{
    TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, UInt32Vector,
    UInt64Vector, VectorRef,
 };
+use greptime_proto::v1;
 use greptime_proto::v1::column_data_type_extension::TypeExt;
 use greptime_proto::v1::ddl_request::Expr;
 use greptime_proto::v1::greptime_request::Request;
 use greptime_proto::v1::query_request::Query;
 use greptime_proto::v1::value::ValueData;
 use greptime_proto::v1::{
-    self, ColumnDataTypeExtension, DdlRequest, DecimalTypeExtension, JsonTypeExtension,
-    QueryRequest, Row, SemanticType, VectorTypeExtension,
+    ColumnDataTypeExtension, DdlRequest, DecimalTypeExtension, JsonTypeExtension, QueryRequest,
+    Row, SemanticType,
 };
 use paste::paste;
 use snafu::prelude::*;
@@ -115,7 +115,6 @@ 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(),
@@ -149,17 +148,6 @@ impl From<ColumnDataTypeWrapper> for ConcreteDataType {
                    ConcreteDataType::decimal128_default_datatype()
                }
            }
-            ColumnDataType::Vector => {
-                if let Some(TypeExt::VectorType(d)) = datatype_wrapper
-                    .datatype_ext
-                    .as_ref()
-                    .and_then(|datatype_ext| datatype_ext.type_ext.as_ref())
-                {
-                    ConcreteDataType::vector_datatype(d.dim)
-                } else {
-                    ConcreteDataType::vector_default_datatype()
-                }
-            }
        }
    }
 }
@@ -241,15 +229,6 @@ impl ColumnDataTypeWrapper {
            }),
        }
    }
-
-    pub fn vector_datatype(dim: u32) -> Self {
-        ColumnDataTypeWrapper {
-            datatype: ColumnDataType::Vector,
-            datatype_ext: Some(ColumnDataTypeExtension {
-                type_ext: Some(TypeExt::VectorType(VectorTypeExtension { dim })),
-            }),
-        }
-    }
 }

 impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
@@ -268,7 +247,7 @@ impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
            ConcreteDataType::UInt64(_) => ColumnDataType::Uint64,
            ConcreteDataType::Float32(_) => ColumnDataType::Float32,
            ConcreteDataType::Float64(_) => ColumnDataType::Float64,
-            ConcreteDataType::Binary(_) => ColumnDataType::Binary,
+            ConcreteDataType::Binary(_) | ConcreteDataType::Json(_) => ColumnDataType::Binary,
            ConcreteDataType::String(_) => ColumnDataType::String,
            ConcreteDataType::Date(_) => ColumnDataType::Date,
            ConcreteDataType::DateTime(_) => ColumnDataType::Datetime,
@@ -290,8 +269,6 @@ impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
                IntervalType::MonthDayNano(_) => ColumnDataType::IntervalMonthDayNano,
            },
            ConcreteDataType::Decimal128(_) => ColumnDataType::Decimal128,
-            ConcreteDataType::Json(_) => ColumnDataType::Json,
-            ConcreteDataType::Vector(_) => ColumnDataType::Vector,
            ConcreteDataType::Null(_)
            | ConcreteDataType::List(_)
            | ConcreteDataType::Dictionary(_)
@@ -310,17 +287,15 @@ impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
                        })),
                    })
            }
-            ColumnDataType::Json => datatype.as_json().map(|_| ColumnDataTypeExtension {
-                type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())),
-            }),
-            ColumnDataType::Vector => {
-                datatype
-                    .as_vector()
-                    .map(|vector_type| ColumnDataTypeExtension {
-                        type_ext: Some(TypeExt::VectorType(VectorTypeExtension {
-                            dim: vector_type.dim as _,
-                        })),
+            ColumnDataType::Binary => {
+                if datatype == ConcreteDataType::json_datatype() {
+                    // Json is the same as  binary in proto. The extension marks the binary in proto is actually a json.
+                    Some(ColumnDataTypeExtension {
+                        type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())),
                    })
+                } else {
+                    None
+                }
            }
            _ => None,
        };
@@ -441,14 +416,6 @@ 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()
-        },
-        ColumnDataType::Vector => Values {
-            binary_values: Vec::with_capacity(capacity),
-            ..Default::default()
-        },
    }
 }

@@ -489,11 +456,13 @@ pub fn push_vals(column: &mut Column, origin_count: usize, vector: VectorRef) {
            TimeUnit::Microsecond => values.time_microsecond_values.push(val.value()),
            TimeUnit::Nanosecond => values.time_nanosecond_values.push(val.value()),
        },
-        Value::IntervalYearMonth(val) => values.interval_year_month_values.push(val.to_i32()),
-        Value::IntervalDayTime(val) => values.interval_day_time_values.push(val.to_i64()),
-        Value::IntervalMonthDayNano(val) => values
-            .interval_month_day_nano_values
-            .push(convert_month_day_nano_to_pb(val)),
+        Value::Interval(val) => match val.unit() {
+            IntervalUnit::YearMonth => values.interval_year_month_values.push(val.to_i32()),
+            IntervalUnit::DayTime => values.interval_day_time_values.push(val.to_i64()),
+            IntervalUnit::MonthDayNano => values
+                .interval_month_day_nano_values
+                .push(convert_i128_to_interval(val.to_i128())),
+        },
        Value::Decimal128(val) => values.decimal128_values.push(convert_to_pb_decimal128(val)),
        Value::List(_) | Value::Duration(_) => unreachable!(),
    });
@@ -527,24 +496,25 @@ fn ddl_request_type(request: &DdlRequest) -> &'static str {
    match request.expr {
        Some(Expr::CreateDatabase(_)) => "ddl.create_database",
        Some(Expr::CreateTable(_)) => "ddl.create_table",
-        Some(Expr::AlterTable(_)) => "ddl.alter_table",
+        Some(Expr::Alter(_)) => "ddl.alter",
        Some(Expr::DropTable(_)) => "ddl.drop_table",
        Some(Expr::TruncateTable(_)) => "ddl.truncate_table",
        Some(Expr::CreateFlow(_)) => "ddl.create_flow",
        Some(Expr::DropFlow(_)) => "ddl.drop_flow",
        Some(Expr::CreateView(_)) => "ddl.create_view",
        Some(Expr::DropView(_)) => "ddl.drop_view",
-        Some(Expr::AlterDatabase(_)) => "ddl.alter_database",
        None => "ddl.empty",
    }
 }

-/// Converts an interval to google protobuf type [IntervalMonthDayNano].
-pub fn convert_month_day_nano_to_pb(v: IntervalMonthDayNano) -> v1::IntervalMonthDayNano {
+/// Converts an i128 value to google protobuf type [IntervalMonthDayNano].
+pub fn convert_i128_to_interval(v: i128) -> v1::IntervalMonthDayNano {
+    let interval = Interval::from_i128(v);
+    let (months, days, nanoseconds) = interval.to_month_day_nano();
    v1::IntervalMonthDayNano {
-        months: v.months,
-        days: v.days,
-        nanoseconds: v.nanoseconds,
+        months,
+        days,
+        nanoseconds,
    }
 }

@@ -592,15 +562,11 @@ pub fn pb_value_to_value_ref<'a>(
        ValueData::TimeMillisecondValue(t) => ValueRef::Time(Time::new_millisecond(*t)),
        ValueData::TimeMicrosecondValue(t) => ValueRef::Time(Time::new_microsecond(*t)),
        ValueData::TimeNanosecondValue(t) => ValueRef::Time(Time::new_nanosecond(*t)),
-        ValueData::IntervalYearMonthValue(v) => {
-            ValueRef::IntervalYearMonth(IntervalYearMonth::from_i32(*v))
-        }
-        ValueData::IntervalDayTimeValue(v) => {
-            ValueRef::IntervalDayTime(IntervalDayTime::from_i64(*v))
-        }
+        ValueData::IntervalYearMonthValue(v) => ValueRef::Interval(Interval::from_i32(*v)),
+        ValueData::IntervalDayTimeValue(v) => ValueRef::Interval(Interval::from_i64(*v)),
        ValueData::IntervalMonthDayNanoValue(v) => {
-            let interval = IntervalMonthDayNano::new(v.months, v.days, v.nanoseconds);
-            ValueRef::IntervalMonthDayNano(interval)
+            let interval = Interval::from_month_day_nano(v.months, v.days, v.nanoseconds);
+            ValueRef::Interval(interval)
        }
        ValueData::Decimal128Value(v) => {
            // get precision and scale from datatype_extension
@@ -691,7 +657,7 @@ pub fn pb_values_to_vector_ref(data_type: &ConcreteDataType, values: Values) ->
            IntervalType::MonthDayNano(_) => {
                Arc::new(IntervalMonthDayNanoVector::from_iter_values(
                    values.interval_month_day_nano_values.iter().map(|x| {
-                        IntervalMonthDayNano::new(x.months, x.days, x.nanoseconds).to_i128()
+                        Interval::from_month_day_nano(x.months, x.days, x.nanoseconds).to_i128()
                    }),
                ))
            }
@@ -701,7 +667,6 @@ pub fn pb_values_to_vector_ref(data_type: &ConcreteDataType, values: Values) ->
                Decimal128::from_value_precision_scale(x.hi, x.lo, d.precision(), d.scale()).into()
            }),
        )),
-        ConcreteDataType::Vector(_) => Arc::new(BinaryVector::from_vec(values.binary_values)),
        ConcreteDataType::Null(_)
        | ConcreteDataType::List(_)
        | ConcreteDataType::Dictionary(_)
@@ -837,18 +802,18 @@ pub fn pb_values_to_values(data_type: &ConcreteDataType, values: Values) -> Vec<
        ConcreteDataType::Interval(IntervalType::YearMonth(_)) => values
            .interval_year_month_values
            .into_iter()
-            .map(|v| Value::IntervalYearMonth(IntervalYearMonth::from_i32(v)))
+            .map(|v| Value::Interval(Interval::from_i32(v)))
            .collect(),
        ConcreteDataType::Interval(IntervalType::DayTime(_)) => values
            .interval_day_time_values
            .into_iter()
-            .map(|v| Value::IntervalDayTime(IntervalDayTime::from_i64(v)))
+            .map(|v| Value::Interval(Interval::from_i64(v)))
            .collect(),
        ConcreteDataType::Interval(IntervalType::MonthDayNano(_)) => values
            .interval_month_day_nano_values
            .into_iter()
            .map(|v| {
-                Value::IntervalMonthDayNano(IntervalMonthDayNano::new(
+                Value::Interval(Interval::from_month_day_nano(
                    v.months,
                    v.days,
                    v.nanoseconds,
@@ -867,7 +832,6 @@ pub fn pb_values_to_values(data_type: &ConcreteDataType, values: Values) -> Vec<
                ))
            })
            .collect(),
-        ConcreteDataType::Vector(_) => values.binary_values.into_iter().map(|v| v.into()).collect(),
        ConcreteDataType::Null(_)
        | ConcreteDataType::List(_)
        | ConcreteDataType::Dictionary(_)
@@ -892,7 +856,10 @@ pub fn is_column_type_value_eq(
    ColumnDataTypeWrapper::try_new(type_value, type_extension)
        .map(|wrapper| {
            let datatype = ConcreteDataType::from(wrapper);
-            expect_type == &datatype
+            (datatype == *expect_type)
+            // Json type leverage binary type in pb, so this is valid.
+                || (datatype == ConcreteDataType::binary_datatype()
+                    && *expect_type == ConcreteDataType::json_datatype())
        })
        .unwrap_or(false)
 }
@@ -974,16 +941,18 @@ pub fn to_proto_value(value: Value) -> Option<v1::Value> {
                value_data: Some(ValueData::TimeNanosecondValue(v.value())),
            },
        },
-        Value::IntervalYearMonth(v) => v1::Value {
-            value_data: Some(ValueData::IntervalYearMonthValue(v.to_i32())),
-        },
-        Value::IntervalDayTime(v) => v1::Value {
-            value_data: Some(ValueData::IntervalDayTimeValue(v.to_i64())),
-        },
-        Value::IntervalMonthDayNano(v) => v1::Value {
-            value_data: Some(ValueData::IntervalMonthDayNanoValue(
-                convert_month_day_nano_to_pb(v),
-            )),
+        Value::Interval(v) => match v.unit() {
+            IntervalUnit::YearMonth => v1::Value {
+                value_data: Some(ValueData::IntervalYearMonthValue(v.to_i32())),
+            },
+            IntervalUnit::DayTime => v1::Value {
+                value_data: Some(ValueData::IntervalDayTimeValue(v.to_i64())),
+            },
+            IntervalUnit::MonthDayNano => v1::Value {
+                value_data: Some(ValueData::IntervalMonthDayNanoValue(
+                    convert_i128_to_interval(v.to_i128()),
+                )),
+            },
        },
        Value::Decimal128(v) => v1::Value {
            value_data: Some(ValueData::Decimal128Value(convert_to_pb_decimal128(v))),
@@ -1075,11 +1044,13 @@ pub fn value_to_grpc_value(value: Value) -> GrpcValue {
                TimeUnit::Microsecond => ValueData::TimeMicrosecondValue(v.value()),
                TimeUnit::Nanosecond => ValueData::TimeNanosecondValue(v.value()),
            }),
-            Value::IntervalYearMonth(v) => Some(ValueData::IntervalYearMonthValue(v.to_i32())),
-            Value::IntervalDayTime(v) => Some(ValueData::IntervalDayTimeValue(v.to_i64())),
-            Value::IntervalMonthDayNano(v) => Some(ValueData::IntervalMonthDayNanoValue(
-                convert_month_day_nano_to_pb(v),
-            )),
+            Value::Interval(v) => Some(match v.unit() {
+                IntervalUnit::YearMonth => ValueData::IntervalYearMonthValue(v.to_i32()),
+                IntervalUnit::DayTime => ValueData::IntervalDayTimeValue(v.to_i64()),
+                IntervalUnit::MonthDayNano => {
+                    ValueData::IntervalMonthDayNanoValue(convert_i128_to_interval(v.to_i128()))
+                }
+            }),
            Value::Decimal128(v) => Some(ValueData::Decimal128Value(convert_to_pb_decimal128(v))),
            Value::List(_) | Value::Duration(_) => unreachable!(),
        },
@@ -1090,7 +1061,6 @@ pub fn value_to_grpc_value(value: Value) -> GrpcValue {
 mod tests {
    use std::sync::Arc;

-    use common_time::interval::IntervalUnit;
    use datatypes::types::{
        Int32Type, IntervalDayTimeType, IntervalMonthDayNanoType, IntervalYearMonthType,
        TimeMillisecondType, TimeSecondType, TimestampMillisecondType, TimestampSecondType,
@@ -1179,10 +1149,6 @@ mod tests {
        let values = values_with_capacity(ColumnDataType::Decimal128, 2);
        let values = values.decimal128_values;
        assert_eq!(2, values.capacity());
-
-        let values = values_with_capacity(ColumnDataType::Vector, 2);
-        let values = values.binary_values;
-        assert_eq!(2, values.capacity());
    }

    #[test]
@@ -1270,11 +1236,7 @@ mod tests {
        assert_eq!(
            ConcreteDataType::decimal128_datatype(10, 2),
            ColumnDataTypeWrapper::decimal128_datatype(10, 2).into()
-        );
-        assert_eq!(
-            ConcreteDataType::vector_datatype(3),
-            ColumnDataTypeWrapper::vector_datatype(3).into()
-        );
+        )
    }

    #[test]
@@ -1370,10 +1332,6 @@ mod tests {
                .try_into()
                .unwrap()
        );
-        assert_eq!(
-            ColumnDataTypeWrapper::vector_datatype(3),
-            ConcreteDataType::vector_datatype(3).try_into().unwrap()
-        );

        let result: Result<ColumnDataTypeWrapper> = ConcreteDataType::null_datatype().try_into();
        assert!(result.is_err());
@@ -1548,11 +1506,11 @@ mod tests {

    #[test]
    fn test_convert_i128_to_interval() {
-        let i128_val = 3;
-        let interval = convert_month_day_nano_to_pb(IntervalMonthDayNano::from_i128(i128_val));
+        let i128_val = 3000;
+        let interval = convert_i128_to_interval(i128_val);
        assert_eq!(interval.months, 0);
        assert_eq!(interval.days, 0);
-        assert_eq!(interval.nanoseconds, 3);
+        assert_eq!(interval.nanoseconds, 3000);
    }

    #[test]
@@ -1632,9 +1590,9 @@ mod tests {
            },
        );
        let expect = vec![
-            Value::IntervalYearMonth(IntervalYearMonth::new(1_i32)),
-            Value::IntervalYearMonth(IntervalYearMonth::new(2_i32)),
-            Value::IntervalYearMonth(IntervalYearMonth::new(3_i32)),
+            Value::Interval(Interval::from_year_month(1_i32)),
+            Value::Interval(Interval::from_year_month(2_i32)),
+            Value::Interval(Interval::from_year_month(3_i32)),
        ];
        assert_eq!(expect, actual);

@@ -1647,9 +1605,9 @@ mod tests {
            },
        );
        let expect = vec![
-            Value::IntervalDayTime(IntervalDayTime::from_i64(1_i64)),
-            Value::IntervalDayTime(IntervalDayTime::from_i64(2_i64)),
-            Value::IntervalDayTime(IntervalDayTime::from_i64(3_i64)),
+            Value::Interval(Interval::from_i64(1_i64)),
+            Value::Interval(Interval::from_i64(2_i64)),
+            Value::Interval(Interval::from_i64(3_i64)),
        ];
        assert_eq!(expect, actual);

@@ -1678,9 +1636,9 @@ mod tests {
            },
        );
        let expect = vec![
-            Value::IntervalMonthDayNano(IntervalMonthDayNano::new(1, 2, 3)),
-            Value::IntervalMonthDayNano(IntervalMonthDayNano::new(5, 6, 7)),
-            Value::IntervalMonthDayNano(IntervalMonthDayNano::new(9, 10, 11)),
+            Value::Interval(Interval::from_month_day_nano(1, 2, 3)),
+            Value::Interval(Interval::from_month_day_nano(5, 6, 7)),
+            Value::Interval(Interval::from_month_day_nano(9, 10, 11)),
        ];
        assert_eq!(expect, actual);
    }
--- a/src/api/src/v1/column_def.rs
+++ b/src/api/src/v1/column_def.rs
@@ -15,10 +15,8 @@
 use std::collections::HashMap;

 use datatypes::schema::{
-    ColumnDefaultConstraint, ColumnSchema, FulltextAnalyzer, FulltextOptions, COMMENT_KEY,
-    FULLTEXT_KEY, INVERTED_INDEX_KEY,
+    ColumnDefaultConstraint, ColumnSchema, FulltextOptions, COMMENT_KEY, FULLTEXT_KEY,
 };
-use greptime_proto::v1::Analyzer;
 use snafu::ResultExt;

 use crate::error::{self, Result};
@@ -27,8 +25,6 @@ use crate::v1::{ColumnDef, ColumnOptions, SemanticType};

 /// Key used to store fulltext options in gRPC column options.
 const FULLTEXT_GRPC_KEY: &str = "fulltext";
-/// Key used to store inverted index options in gRPC column options.
-const INVERTED_INDEX_GRPC_KEY: &str = "inverted_index";

 /// Tries to construct a `ColumnSchema` from the given  `ColumnDef`.
 pub fn try_as_column_schema(column_def: &ColumnDef) -> Result<ColumnSchema> {
@@ -53,13 +49,10 @@ pub fn try_as_column_schema(column_def: &ColumnDef) -> Result<ColumnSchema> {
    if !column_def.comment.is_empty() {
        metadata.insert(COMMENT_KEY.to_string(), column_def.comment.clone());
    }
-    if let Some(options) = column_def.options.as_ref() {
-        if let Some(fulltext) = options.options.get(FULLTEXT_GRPC_KEY) {
-            metadata.insert(FULLTEXT_KEY.to_string(), fulltext.clone());
-        }
-        if let Some(inverted_index) = options.options.get(INVERTED_INDEX_GRPC_KEY) {
-            metadata.insert(INVERTED_INDEX_KEY.to_string(), inverted_index.clone());
-        }
+    if let Some(options) = column_def.options.as_ref()
+        && let Some(fulltext) = options.options.get(FULLTEXT_GRPC_KEY)
+    {
+        metadata.insert(FULLTEXT_KEY.to_string(), fulltext.to_string());
    }

    ColumnSchema::new(&column_def.name, data_type.into(), column_def.is_nullable)
@@ -77,12 +70,7 @@ pub fn options_from_column_schema(column_schema: &ColumnSchema) -> Option<Column
    if let Some(fulltext) = column_schema.metadata().get(FULLTEXT_KEY) {
        options
            .options
-            .insert(FULLTEXT_GRPC_KEY.to_string(), fulltext.clone());
-    }
-    if let Some(inverted_index) = column_schema.metadata().get(INVERTED_INDEX_KEY) {
-        options
-            .options
-            .insert(INVERTED_INDEX_GRPC_KEY.to_string(), inverted_index.clone());
+            .insert(FULLTEXT_GRPC_KEY.to_string(), fulltext.to_string());
    }

    (!options.options.is_empty()).then_some(options)
@@ -105,14 +93,6 @@ pub fn options_from_fulltext(fulltext: &FulltextOptions) -> Result<Option<Column
    Ok((!options.options.is_empty()).then_some(options))
 }

-/// Tries to construct a `FulltextAnalyzer` from the given analyzer.
-pub fn as_fulltext_option(analyzer: Analyzer) -> FulltextAnalyzer {
-    match analyzer {
-        Analyzer::English => FulltextAnalyzer::English,
-        Analyzer::Chinese => FulltextAnalyzer::Chinese,
-    }
-}
-
 #[cfg(test)]
 mod tests {

@@ -135,13 +115,10 @@ mod tests {
            comment: "test_comment".to_string(),
            datatype_extension: None,
            options: Some(ColumnOptions {
-                options: HashMap::from([
-                    (
-                        FULLTEXT_GRPC_KEY.to_string(),
-                        "{\"enable\":true}".to_string(),
-                    ),
-                    (INVERTED_INDEX_GRPC_KEY.to_string(), "true".to_string()),
-                ]),
+                options: HashMap::from([(
+                    FULLTEXT_GRPC_KEY.to_string(),
+                    "{\"enable\":true}".to_string(),
+                )]),
            }),
        };

@@ -162,7 +139,6 @@ mod tests {
                ..Default::default()
            }
        );
-        assert!(schema.is_inverted_indexed());
    }

    #[test]
@@ -177,17 +153,12 @@ mod tests {
                analyzer: FulltextAnalyzer::English,
                case_sensitive: false,
            })
-            .unwrap()
-            .set_inverted_index(true);
+            .unwrap();
        let options = options_from_column_schema(&schema).unwrap();
        assert_eq!(
            options.options.get(FULLTEXT_GRPC_KEY).unwrap(),
            "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false}"
        );
-        assert_eq!(
-            options.options.get(INVERTED_INDEX_GRPC_KEY).unwrap(),
-            "true"
-        );
    }

    #[test]
--- 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>`",
        })?;
-        match mode {
+        return 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,12 +178,6 @@ 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 },

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

            Error::ViewPlanColumnsChanged { .. } => StatusCode::InvalidArguments,
--- a/src/catalog/src/kvbackend.rs
+++ b/src/catalog/src/kvbackend.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-pub use client::{CachedKvBackend, CachedKvBackendBuilder, MetaKvBackend};
+pub use client::{CachedMetaKvBackend, CachedMetaKvBackendBuilder, MetaKvBackend};

 mod client;
 mod manager;
--- a/src/catalog/src/kvbackend/client.rs
+++ b/src/catalog/src/kvbackend/client.rs
@@ -22,7 +22,6 @@ use common_error::ext::BoxedError;
 use common_meta::cache_invalidator::KvCacheInvalidator;
 use common_meta::error::Error::CacheNotGet;
 use common_meta::error::{CacheNotGetSnafu, Error, ExternalSnafu, GetKvCacheSnafu, Result};
-use common_meta::kv_backend::txn::{Txn, TxnResponse};
 use common_meta::kv_backend::{KvBackend, KvBackendRef, TxnService};
 use common_meta::rpc::store::{
    BatchDeleteRequest, BatchDeleteResponse, BatchGetRequest, BatchGetResponse, BatchPutRequest,
@@ -43,20 +42,20 @@ const DEFAULT_CACHE_MAX_CAPACITY: u64 = 10000;
 const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(10 * 60);
 const DEFAULT_CACHE_TTI: Duration = Duration::from_secs(5 * 60);

-pub struct CachedKvBackendBuilder {
+pub struct CachedMetaKvBackendBuilder {
    cache_max_capacity: Option<u64>,
    cache_ttl: Option<Duration>,
    cache_tti: Option<Duration>,
-    inner: KvBackendRef,
+    meta_client: Arc<MetaClient>,
 }

-impl CachedKvBackendBuilder {
-    pub fn new(inner: KvBackendRef) -> Self {
+impl CachedMetaKvBackendBuilder {
+    pub fn new(meta_client: Arc<MetaClient>) -> Self {
        Self {
            cache_max_capacity: None,
            cache_ttl: None,
            cache_tti: None,
-            inner,
+            meta_client,
        }
    }

@@ -75,7 +74,7 @@ impl CachedKvBackendBuilder {
        self
    }

-    pub fn build(self) -> CachedKvBackend {
+    pub fn build(self) -> CachedMetaKvBackend {
        let cache_max_capacity = self
            .cache_max_capacity
            .unwrap_or(DEFAULT_CACHE_MAX_CAPACITY);
@@ -86,11 +85,14 @@ impl CachedKvBackendBuilder {
            .time_to_live(cache_ttl)
            .time_to_idle(cache_tti)
            .build();
-        let kv_backend = self.inner;
+
+        let kv_backend = Arc::new(MetaKvBackend {
+            client: self.meta_client,
+        });
        let name = format!("CachedKvBackend({})", kv_backend.name());
        let version = AtomicUsize::new(0);

-        CachedKvBackend {
+        CachedMetaKvBackend {
            kv_backend,
            cache,
            name,
@@ -110,29 +112,19 @@ pub type CacheBackend = Cache<Vec<u8>, KeyValue>;
 /// Therefore, it is recommended to use CachedMetaKvBackend to only read metadata related
 /// information. Note: If you read other information, you may read expired data, which depends on
 /// TTL and TTI for cache.
-pub struct CachedKvBackend {
+pub struct CachedMetaKvBackend {
    kv_backend: KvBackendRef,
    cache: CacheBackend,
    name: String,
    version: AtomicUsize,
 }

-#[async_trait::async_trait]
-impl TxnService for CachedKvBackend {
+impl TxnService for CachedMetaKvBackend {
    type Error = Error;
-
-    async fn txn(&self, txn: Txn) -> std::result::Result<TxnResponse, Self::Error> {
-        // TODO(hl): txn of CachedKvBackend simply pass through to inner backend without invalidating caches.
-        self.kv_backend.txn(txn).await
-    }
-
-    fn max_txn_ops(&self) -> usize {
-        self.kv_backend.max_txn_ops()
-    }
 }

 #[async_trait::async_trait]
-impl KvBackend for CachedKvBackend {
+impl KvBackend for CachedMetaKvBackend {
    fn name(&self) -> &str {
        &self.name
    }
@@ -313,7 +305,7 @@ impl KvBackend for CachedKvBackend {
 }

 #[async_trait::async_trait]
-impl KvCacheInvalidator for CachedKvBackend {
+impl KvCacheInvalidator for CachedMetaKvBackend {
    async fn invalidate_key(&self, key: &[u8]) {
        self.create_new_version();
        self.cache.invalidate(key).await;
@@ -321,7 +313,7 @@ impl KvCacheInvalidator for CachedKvBackend {
    }
 }

-impl CachedKvBackend {
+impl CachedMetaKvBackend {
    // only for test
    #[cfg(test)]
    fn wrap(kv_backend: KvBackendRef) -> Self {
@@ -474,7 +466,7 @@ mod tests {
    use common_meta::rpc::KeyValue;
    use dashmap::DashMap;

-    use super::CachedKvBackend;
+    use super::CachedMetaKvBackend;

    #[derive(Default)]
    pub struct SimpleKvBackend {
@@ -548,7 +540,7 @@ mod tests {
    async fn test_cached_kv_backend() {
        let simple_kv = Arc::new(SimpleKvBackend::default());
        let get_execute_times = simple_kv.get_execute_times.clone();
-        let cached_kv = CachedKvBackend::wrap(simple_kv);
+        let cached_kv = CachedMetaKvBackend::wrap(simple_kv);

        add_some_vals(&cached_kv).await;

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

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

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

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

-                let mut table_partitions = partition_manager
-                    .batch_find_table_partitions(&table_ids)
-                    .await
-                    .context(FindPartitionsSnafu)?;
+                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()
+                };

                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,12 +39,9 @@ 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";

@@ -55,12 +52,9 @@ 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 data files size in bytes.
-/// - `index_size`: The sst index files size in bytes.
+/// - `sst_size`: The sst size in bytes.
 /// - `engine`: The engine type.
 /// - `region_role`: The region role.
 ///
@@ -82,12 +76,9 @@ 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),
        ]))
@@ -144,12 +135,9 @@ 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,
 }
@@ -162,12 +150,9 @@ 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),
        }
@@ -192,12 +177,9 @@ 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())),
        ];
@@ -210,12 +192,9 @@ 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()));
    }
@@ -225,12 +204,9 @@ 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/schemata.rs
+++ b/src/catalog/src/system_schema/information_schema/schemata.rs
@@ -180,7 +180,7 @@ impl InformationSchemaSchemataBuilder {
                    .context(TableMetadataManagerSnafu)?
                    // information_schema is not available from this
                    // table_metadata_manager and we return None
-                    .map(|schema_opts| format!("{}", schema_opts.into_inner()))
+                    .map(|schema_opts| format!("{schema_opts}"))
            } else {
                None
            };
--- 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,16 +12,13 @@
 // 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, MITO_ENGINE};
+use common_catalog::consts::INFORMATION_SCHEMA_TABLES_TABLE_ID;
 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;
@@ -34,7 +31,7 @@ use datatypes::vectors::{
 };
 use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
-use store_api::storage::{RegionId, ScanRequest, TableId};
+use store_api::storage::{ScanRequest, TableId};
 use table::metadata::{TableInfo, TableType};

 use super::TABLES;
@@ -42,7 +39,6 @@ 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";
@@ -238,51 +234,17 @@ 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 || table_info.is_physical_table() {
-                        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,
                );
            }
        }
@@ -298,7 +260,6 @@ 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();
@@ -312,9 +273,7 @@ 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)),
        ];
@@ -328,39 +287,21 @@ 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.checksum.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.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() {
-            Ok(RecordBatch::new_empty(self.schema.clone()))
+            RecordBatch::new_empty(self.schema.clone()).context(CreateRecordBatchSnafu)
        } 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,9 +12,6 @@
 // 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};
@@ -43,6 +40,9 @@ 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.workspace = true
+parking_lot = "0.12"
 prometheus.workspace = true
 prost.workspace = true
 query.workspace = true
@@ -45,6 +45,7 @@ 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/client/src/database.rs
+++ b/src/client/src/database.rs
@@ -18,7 +18,7 @@ use api::v1::greptime_database_client::GreptimeDatabaseClient;
 use api::v1::greptime_request::Request;
 use api::v1::query_request::Query;
 use api::v1::{
-    AlterTableExpr, AuthHeader, CreateTableExpr, DdlRequest, GreptimeRequest, InsertRequests,
+    AlterExpr, AuthHeader, CreateTableExpr, DdlRequest, GreptimeRequest, InsertRequests,
    QueryRequest, RequestHeader,
 };
 use arrow_flight::Ticket;
@@ -211,9 +211,9 @@ impl Database {
        .await
    }

-    pub async fn alter(&self, expr: AlterTableExpr) -> Result<Output> {
+    pub async fn alter(&self, expr: AlterExpr) -> Result<Output> {
        self.do_get(Request::Ddl(DdlRequest {
-            expr: Some(DdlExpr::AlterTable(expr)),
+            expr: Some(DdlExpr::Alter(expr)),
        }))
        .await
    }
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -10,7 +10,7 @@ name = "greptime"
 path = "src/bin/greptime.rs"

 [features]
-default = ["python", "servers/pprof", "servers/mem-prof"]
+default = ["python"]
 tokio-console = ["common-telemetry/tokio-console"]
 python = ["frontend/python"]

@@ -33,7 +33,6 @@ common-error.workspace = true
 common-grpc.workspace = true
 common-macro.workspace = true
 common-meta.workspace = true
-common-options.workspace = true
 common-procedure.workspace = true
 common-query.workspace = true
 common-recordbatch.workspace = true
@@ -53,7 +52,6 @@ flow.workspace = true
 frontend = { workspace = true, default-features = false }
 futures.workspace = true
 human-panic = "2.0"
-humantime.workspace = true
 lazy_static.workspace = true
 meta-client.workspace = true
 meta-srv.workspace = true
@@ -80,7 +78,7 @@ table.workspace = true
 tokio.workspace = true
 toml.workspace = true
 tonic.workspace = true
-tracing-appender.workspace = true
+tracing-appender = "0.2"

 [target.'cfg(not(windows))'.dependencies]
 tikv-jemallocator = "0.6"
--- a/src/cmd/src/cli/database.rs
+++ b/src/cmd/src/cli/database.rs
@@ -12,15 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::time::Duration;
-
 use base64::engine::general_purpose;
 use base64::Engine;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
-use humantime::format_duration;
 use serde_json::Value;
-use servers::http::header::constants::GREPTIME_DB_HEADER_TIMEOUT;
-use servers::http::result::greptime_result_v1::GreptimedbV1Response;
+use servers::http::greptime_result_v1::GreptimedbV1Response;
 use servers::http::GreptimeQueryOutput;
 use snafu::ResultExt;

@@ -30,16 +26,10 @@ pub(crate) struct DatabaseClient {
    addr: String,
    catalog: String,
    auth_header: Option<String>,
-    timeout: Duration,
 }

 impl DatabaseClient {
-    pub fn new(
-        addr: String,
-        catalog: String,
-        auth_basic: Option<String>,
-        timeout: Duration,
-    ) -> Self {
+    pub fn new(addr: String, catalog: String, auth_basic: Option<String>) -> Self {
        let auth_header = if let Some(basic) = auth_basic {
            let encoded = general_purpose::STANDARD.encode(basic);
            Some(format!("basic {}", encoded))
@@ -51,7 +41,6 @@ impl DatabaseClient {
            addr,
            catalog,
            auth_header,
-            timeout,
        }
    }

@@ -74,11 +63,6 @@ impl DatabaseClient {
            request = request.header("Authorization", auth);
        }

-        request = request.header(
-            GREPTIME_DB_HEADER_TIMEOUT,
-            format_duration(self.timeout).to_string(),
-        );
-
        let response = request.send().await.with_context(|_| HttpQuerySqlSnafu {
            reason: format!("bad url: {}", url),
        })?;
--- a/src/cmd/src/cli/export.rs
+++ b/src/cmd/src/cli/export.rs
@@ -15,7 +15,6 @@
 use std::collections::HashSet;
 use std::path::PathBuf;
 use std::sync::Arc;
-use std::time::Duration;

 use async_trait::async_trait;
 use clap::{Parser, ValueEnum};
@@ -84,26 +83,14 @@ pub struct ExportCommand {
    /// The basic authentication for connecting to the server
    #[clap(long)]
    auth_basic: Option<String>,
-
-    /// The timeout of invoking the database.
-    ///
-    /// It is used to override the server-side timeout setting.
-    /// The default behavior will disable server-side default timeout(i.e. `0s`).
-    #[clap(long, value_parser = humantime::parse_duration)]
-    timeout: Option<Duration>,
 }

 impl ExportCommand {
    pub async fn build(&self, guard: Vec<WorkerGuard>) -> Result<Instance> {
        let (catalog, schema) = database::split_database(&self.database)?;

-        let database_client = DatabaseClient::new(
-            self.addr.clone(),
-            catalog.clone(),
-            self.auth_basic.clone(),
-            // Treats `None` as `0s` to disable server-side default timeout.
-            self.timeout.unwrap_or_default(),
-        );
+        let database_client =
+            DatabaseClient::new(self.addr.clone(), catalog.clone(), self.auth_basic.clone());

        Ok(Instance::new(
            Box::new(Export {
--- a/src/cmd/src/cli/import.rs
+++ b/src/cmd/src/cli/import.rs
@@ -14,7 +14,6 @@

 use std::path::PathBuf;
 use std::sync::Arc;
-use std::time::Duration;

 use async_trait::async_trait;
 use clap::{Parser, ValueEnum};
@@ -69,25 +68,13 @@ pub struct ImportCommand {
    /// The basic authentication for connecting to the server
    #[clap(long)]
    auth_basic: Option<String>,
-
-    /// The timeout of invoking the database.
-    ///
-    /// It is used to override the server-side timeout setting.
-    /// The default behavior will disable server-side default timeout(i.e. `0s`).
-    #[clap(long, value_parser = humantime::parse_duration)]
-    timeout: Option<Duration>,
 }

 impl ImportCommand {
    pub async fn build(&self, guard: Vec<WorkerGuard>) -> Result<Instance> {
        let (catalog, schema) = database::split_database(&self.database)?;
-        let database_client = DatabaseClient::new(
-            self.addr.clone(),
-            catalog.clone(),
-            self.auth_basic.clone(),
-            // Treats `None` as `0s` to disable server-side default timeout.
-            self.timeout.unwrap_or_default(),
-        );
+        let database_client =
+            DatabaseClient::new(self.addr.clone(), catalog.clone(), self.auth_basic.clone());

        Ok(Instance::new(
            Box::new(Import {
--- a/src/cmd/src/cli/repl.rs
+++ b/src/cmd/src/cli/repl.rs
@@ -21,14 +21,13 @@ use cache::{
    TABLE_ROUTE_CACHE_NAME,
 };
 use catalog::kvbackend::{
-    CachedKvBackend, CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend,
+    CachedMetaKvBackend, CachedMetaKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend,
 };
 use client::{Client, Database, OutputData, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_base::Plugins;
 use common_config::Mode;
 use common_error::ext::ErrorExt;
 use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
-use common_meta::kv_backend::KvBackendRef;
 use common_query::Output;
 use common_recordbatch::RecordBatches;
 use common_telemetry::debug;
@@ -175,7 +174,7 @@ impl Repl {

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

@@ -259,9 +258,8 @@ async fn create_query_engine(meta_addr: &str) -> Result<DatafusionQueryEngine> {
        .context(StartMetaClientSnafu)?;
    let meta_client = Arc::new(meta_client);

-    let cached_meta_backend = Arc::new(
-        CachedKvBackendBuilder::new(Arc::new(MetaKvBackend::new(meta_client.clone()))).build(),
-    );
+    let cached_meta_backend =
+        Arc::new(CachedMetaKvBackendBuilder::new(meta_client.clone()).build());
    let layered_cache_builder = LayeredCacheRegistryBuilder::default().add_cache_registry(
        CacheRegistryBuilder::default()
            .add_cache(cached_meta_backend.clone())
--- a/src/cmd/src/datanode.rs
+++ b/src/cmd/src/datanode.rs
@@ -272,10 +272,9 @@ 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, &plugin_opts, &opts)
+        plugins::setup_datanode_plugins(&mut plugins, &opts)
            .await
            .context(StartDatanodeSnafu)?;

--- a/src/cmd/src/flownode.rs
+++ b/src/cmd/src/flownode.rs
@@ -15,7 +15,7 @@
 use std::sync::Arc;

 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
-use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
+use catalog::kvbackend::{CachedMetaKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
 use clap::Parser;
 use client::client_manager::NodeClients;
 use common_base::Plugins;
@@ -246,12 +246,11 @@ impl StartCommand {
        let cache_tti = meta_config.metadata_cache_tti;

        // TODO(discord9): add helper function to ease the creation of cache registry&such
-        let cached_meta_backend =
-            CachedKvBackendBuilder::new(Arc::new(MetaKvBackend::new(meta_client.clone())))
-                .cache_max_capacity(cache_max_capacity)
-                .cache_ttl(cache_ttl)
-                .cache_tti(cache_tti)
-                .build();
+        let cached_meta_backend = CachedMetaKvBackendBuilder::new(meta_client.clone())
+            .cache_max_capacity(cache_max_capacity)
+            .cache_ttl(cache_ttl)
+            .cache_tti(cache_tti)
+            .build();
        let cached_meta_backend = Arc::new(cached_meta_backend);

        // Builds cache registry
--- a/src/cmd/src/frontend.rs
+++ b/src/cmd/src/frontend.rs
@@ -17,7 +17,7 @@ use std::time::Duration;

 use async_trait::async_trait;
 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
-use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
+use catalog::kvbackend::{CachedMetaKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
 use clap::Parser;
 use client::client_manager::NodeClients;
 use common_base::Plugins;
@@ -266,10 +266,9 @@ 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, &plugin_opts, &opts)
+        plugins::setup_frontend_plugins(&mut plugins, &opts)
            .await
            .context(StartFrontendSnafu)?;

@@ -293,12 +292,11 @@ impl StartCommand {
        .context(MetaClientInitSnafu)?;

        // TODO(discord9): add helper function to ease the creation of cache registry&such
-        let cached_meta_backend =
-            CachedKvBackendBuilder::new(Arc::new(MetaKvBackend::new(meta_client.clone())))
-                .cache_max_capacity(cache_max_capacity)
-                .cache_ttl(cache_ttl)
-                .cache_tti(cache_tti)
-                .build();
+        let cached_meta_backend = CachedMetaKvBackendBuilder::new(meta_client.clone())
+            .cache_max_capacity(cache_max_capacity)
+            .cache_ttl(cache_ttl)
+            .cache_tti(cache_tti)
+            .build();
        let cached_meta_backend = Arc::new(cached_meta_backend);

        // Builds cache registry
@@ -344,8 +342,6 @@ 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);
@@ -476,7 +472,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
@@ -43,31 +43,6 @@ lazy_static::lazy_static! {
        prometheus::register_int_gauge_vec!("greptime_app_version", "app version", &["version", "short_version", "app"]).unwrap();
 }

-/// wait for the close signal, for unix platform it's SIGINT or SIGTERM
-#[cfg(unix)]
-async fn start_wait_for_close_signal() -> std::io::Result<()> {
-    use tokio::signal::unix::{signal, SignalKind};
-    let mut sigint = signal(SignalKind::interrupt())?;
-    let mut sigterm = signal(SignalKind::terminate())?;
-
-    tokio::select! {
-        _ = sigint.recv() => {
-            info!("Received SIGINT, shutting down");
-        }
-        _ = sigterm.recv() => {
-            info!("Received SIGTERM, shutting down");
-        }
-    }
-
-    Ok(())
-}
-
-/// wait for the close signal, for non-unix platform it's ctrl-c
-#[cfg(not(unix))]
-async fn start_wait_for_close_signal() -> std::io::Result<()> {
-    tokio::signal::ctrl_c().await
-}
-
 #[async_trait]
 pub trait App: Send {
    fn name(&self) -> &str;
@@ -94,9 +69,9 @@ pub trait App: Send {
        self.start().await?;

        if self.wait_signal() {
-            if let Err(e) = start_wait_for_close_signal().await {
-                error!(e; "Failed to listen for close signal");
-                // It's unusual to fail to listen for close signal, maybe there's something unexpected in
+            if let Err(e) = tokio::signal::ctrl_c().await {
+                error!(e; "Failed to listen for ctrl-c signal");
+                // It's unusual to fail to listen for ctrl-c signal, maybe there's something unexpected in
                // the underlying system. So we stop the app instead of running nonetheless to let people
                // investigate the issue.
            }
@@ -109,7 +84,6 @@ 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
@@ -48,10 +48,6 @@ impl Instance {
            _guard: guard,
        }
    }
-
-    pub fn get_inner(&self) -> &MetasrvInstance {
-        &self.instance
-    }
 }

 #[async_trait]
@@ -90,14 +86,6 @@ impl Command {
    pub fn load_options(&self, global_options: &GlobalOptions) -> Result<MetasrvOptions> {
        self.subcmd.load_options(global_options)
    }
-
-    pub fn config_file(&self) -> &Option<String> {
-        self.subcmd.config_file()
-    }
-
-    pub fn env_prefix(&self) -> &String {
-        self.subcmd.env_prefix()
-    }
 }

 #[derive(Parser)]
@@ -117,18 +105,6 @@ impl SubCommand {
            SubCommand::Start(cmd) => cmd.load_options(global_options),
        }
    }
-
-    fn config_file(&self) -> &Option<String> {
-        match self {
-            SubCommand::Start(cmd) => &cmd.config_file,
-        }
-    }
-
-    fn env_prefix(&self) -> &String {
-        match self {
-            SubCommand::Start(cmd) => &cmd.env_prefix,
-        }
-    }
 }

 #[derive(Debug, Default, Parser)]
@@ -273,10 +249,9 @@ 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, &plugin_opts, &opts)
+        plugins::setup_metasrv_plugins(&mut plugins, &opts)
            .await
            .context(StartMetaServerSnafu)?;

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

 #[derive(Parser, Default, Debug, Clone)]
@@ -41,8 +40,6 @@ 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,7 +12,6 @@
 // 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};

@@ -251,13 +250,6 @@ 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 {
@@ -348,8 +340,7 @@ pub struct StartCommand {
 }

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

        let guard = common_telemetry::init_global_logging(
@@ -455,16 +445,15 @@ 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, &plugin_opts, &fe_opts)
+        plugins::setup_frontend_plugins(&mut plugins, &fe_opts)
            .await
            .context(StartFrontendSnafu)?;

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

@@ -664,7 +653,7 @@ impl StartCommand {
    }
 }

-pub struct StandaloneInformationExtension {
+struct StandaloneInformationExtension {
    region_server: RegionServer,
    procedure_manager: ProcedureManagerRef,
    start_time_ms: u64,
@@ -736,14 +725,12 @@ impl InformationExtension for StandaloneInformationExtension {
                    id: stat.region_id,
                    rcus: 0,
                    wcus: 0,
-                    approximate_bytes: region_stat.estimated_disk_size(),
+                    approximate_bytes: region_stat.estimated_disk_size() as i64,
                    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<_>>();
@@ -775,7 +762,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,13 +20,13 @@ use common_config::Configurable;
 use common_grpc::channel_manager::{
    DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE, DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
 };
-use common_options::datanode::{ClientOptions, DatanodeClientOptions};
-use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, DEFAULT_OTLP_ENDPOINT};
+use common_telemetry::logging::{LoggingOptions, DEFAULT_OTLP_ENDPOINT};
 use common_wal::config::raft_engine::RaftEngineConfig;
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
 use file_engine::config::EngineConfig;
 use frontend::frontend::FrontendOptions;
+use frontend::service_config::datanode::DatanodeClientOptions;
 use meta_client::MetaClientOptions;
 use meta_srv::metasrv::MetasrvOptions;
 use meta_srv::selector::SelectorType;
@@ -126,11 +126,10 @@ fn test_load_frontend_example_config() {
                tracing_sample_ratio: Some(Default::default()),
                ..Default::default()
            },
-            datanode: DatanodeClientOptions {
-                client: ClientOptions {
+            datanode: frontend::service_config::DatanodeOptions {
+                client: DatanodeClientOptions {
                    connect_timeout: Duration::from_secs(10),
                    tcp_nodelay: true,
-                    ..Default::default()
                },
            },
            export_metrics: ExportMetricsOption {
@@ -160,20 +159,8 @@ 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: DatanodeClientOptions {
-                client: ClientOptions {
-                    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/Cargo.toml
+++ b/src/common/base/Cargo.toml
@@ -16,12 +16,9 @@ common-error.workspace = true
 common-macro.workspace = true
 futures.workspace = true
 paste = "1.0"
-pin-project.workspace = true
 serde = { version = "1.0", features = ["derive"] }
 snafu.workspace = true
-tokio.workspace = true
 zeroize = { version = "1.6", default-features = false, features = ["alloc"] }

 [dev-dependencies]
-common-test-util.workspace = true
 toml.workspace = true
--- a/src/common/base/src/plugins.rs
+++ b/src/common/base/src/plugins.rs
@@ -38,18 +38,6 @@ impl Plugins {
        self.read().get::<T>().cloned()
    }

-    pub fn get_or_insert<T, F>(&self, f: F) -> T
-    where
-        T: 'static + Send + Sync + Clone,
-        F: FnOnce() -> T,
-    {
-        let mut binding = self.write();
-        if !binding.contains::<T>() {
-            binding.insert(f());
-        }
-        binding.get::<T>().cloned().unwrap()
-    }
-
    pub fn map_mut<T: 'static + Send + Sync, F, R>(&self, mapper: F) -> R
    where
        F: FnOnce(Option<&mut T>) -> R,
--- a/src/common/base/src/range_read.rs
+++ b/src/common/base/src/range_read.rs
@@ -12,20 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::future::Future;
 use std::io;
 use std::ops::Range;
-use std::path::Path;
-use std::pin::Pin;
-use std::sync::Arc;
-use std::task::{Context, Poll};

 use async_trait::async_trait;
 use bytes::{BufMut, Bytes};
-use futures::AsyncRead;
-use pin_project::pin_project;
-use tokio::io::{AsyncReadExt as _, AsyncSeekExt as _};
-use tokio::sync::Mutex;
+use futures::{AsyncReadExt, AsyncSeekExt};

 /// `Metadata` contains the metadata of a source.
 pub struct Metadata {
@@ -69,7 +61,7 @@ pub trait RangeReader: Send + Unpin {
 }

 #[async_trait]
-impl<R: ?Sized + RangeReader> RangeReader for &mut R {
+impl<R: RangeReader + Send + Unpin> RangeReader for &mut R {
    async fn metadata(&mut self) -> io::Result<Metadata> {
        (*self).metadata().await
    }
@@ -88,212 +80,26 @@ impl<R: ?Sized + RangeReader> RangeReader for &mut R {
    }
 }

-/// `AsyncReadAdapter` adapts a `RangeReader` to an `AsyncRead`.
-#[pin_project]
-pub struct AsyncReadAdapter<R> {
-    /// The inner `RangeReader`.
-    /// Use `Mutex` to get rid of the borrow checker issue.
-    inner: Arc<Mutex<R>>,
-
-    /// The current position from the view of the reader.
-    position: u64,
-
-    /// The buffer for the read bytes.
-    buffer: Vec<u8>,
-
-    /// The length of the content.
-    content_length: u64,
-
-    /// The future for reading the next bytes.
-    #[pin]
-    read_fut: Option<Pin<Box<dyn Future<Output = io::Result<Bytes>> + Send>>>,
-}
-
-impl<R: RangeReader + 'static> AsyncReadAdapter<R> {
-    pub async fn new(inner: R) -> io::Result<Self> {
-        let mut inner = inner;
-        let metadata = inner.metadata().await?;
-        Ok(AsyncReadAdapter {
-            inner: Arc::new(Mutex::new(inner)),
-            position: 0,
-            buffer: Vec::new(),
-            content_length: metadata.content_length,
-            read_fut: None,
-        })
-    }
-}
-
-/// The maximum size per read for the inner reader in `AsyncReadAdapter`.
-const MAX_SIZE_PER_READ: usize = 8 * 1024 * 1024; // 8MB
-
-impl<R: RangeReader + 'static> AsyncRead for AsyncReadAdapter<R> {
-    fn poll_read(
-        mut self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-        buf: &mut [u8],
-    ) -> Poll<io::Result<usize>> {
-        let mut this = self.as_mut().project();
-
-        if *this.position >= *this.content_length {
-            return Poll::Ready(Ok(0));
-        }
-
-        if !this.buffer.is_empty() {
-            let to_read = this.buffer.len().min(buf.len());
-            buf[..to_read].copy_from_slice(&this.buffer[..to_read]);
-            this.buffer.drain(..to_read);
-            *this.position += to_read as u64;
-            return Poll::Ready(Ok(to_read));
-        }
-
-        if this.read_fut.is_none() {
-            let size = (*this.content_length - *this.position).min(MAX_SIZE_PER_READ as u64);
-            let range = *this.position..(*this.position + size);
-            let inner = this.inner.clone();
-            let fut = async move {
-                let mut inner = inner.lock().await;
-                inner.read(range).await
-            };
-
-            *this.read_fut = Some(Box::pin(fut));
-        }
-
-        match this
-            .read_fut
-            .as_mut()
-            .as_pin_mut()
-            .expect("checked above")
-            .poll(cx)
-        {
-            Poll::Pending => Poll::Pending,
-            Poll::Ready(Ok(bytes)) => {
-                *this.read_fut = None;
-
-                if !bytes.is_empty() {
-                    this.buffer.extend_from_slice(&bytes);
-                    self.poll_read(cx, buf)
-                } else {
-                    Poll::Ready(Ok(0))
-                }
-            }
-            Poll::Ready(Err(e)) => {
-                *this.read_fut = None;
-                Poll::Ready(Err(e))
-            }
-        }
-    }
-}
+/// `RangeReaderAdapter` bridges `RangeReader` and `AsyncRead + AsyncSeek`.
+pub struct RangeReaderAdapter<R>(pub R);

+/// Implements `RangeReader` for a type that implements `AsyncRead + AsyncSeek`.
+///
+/// TODO(zhongzc): It's a temporary solution for porting the codebase from `AsyncRead + AsyncSeek` to `RangeReader`.
+/// Until the codebase is fully ported to `RangeReader`, remove this implementation.
 #[async_trait]
-impl RangeReader for Vec<u8> {
+impl<R: futures::AsyncRead + futures::AsyncSeek + Send + Unpin> RangeReader
+    for RangeReaderAdapter<R>
+{
    async fn metadata(&mut self) -> io::Result<Metadata> {
-        Ok(Metadata {
-            content_length: self.len() as u64,
-        })
+        let content_length = self.0.seek(io::SeekFrom::End(0)).await?;
+        Ok(Metadata { content_length })
    }

-    async fn read(&mut self, mut range: Range<u64>) -> io::Result<Bytes> {
-        range.end = range.end.min(self.len() as u64);
-
-        let bytes = Bytes::copy_from_slice(&self[range.start as usize..range.end as usize]);
-        Ok(bytes)
-    }
-}
-
-/// `FileReader` is a `RangeReader` for reading a file.
-pub struct FileReader {
-    content_length: u64,
-    position: u64,
-    file: tokio::fs::File,
-}
-
-impl FileReader {
-    /// Creates a new `FileReader` for the file at the given path.
-    pub async fn new(path: impl AsRef<Path>) -> io::Result<Self> {
-        let file = tokio::fs::File::open(path).await?;
-        let metadata = file.metadata().await?;
-        Ok(FileReader {
-            content_length: metadata.len(),
-            position: 0,
-            file,
-        })
-    }
-}
-
-#[async_trait]
-impl RangeReader for FileReader {
-    async fn metadata(&mut self) -> io::Result<Metadata> {
-        Ok(Metadata {
-            content_length: self.content_length,
-        })
-    }
-
-    async fn read(&mut self, mut range: Range<u64>) -> io::Result<Bytes> {
-        if range.start != self.position {
-            self.file.seek(io::SeekFrom::Start(range.start)).await?;
-            self.position = range.start;
-        }
-
-        range.end = range.end.min(self.content_length);
-        if range.end <= self.position {
-            return Err(io::Error::new(
-                io::ErrorKind::UnexpectedEof,
-                "Start of range is out of bounds",
-            ));
-        }
-
+    async fn read(&mut self, range: Range<u64>) -> io::Result<Bytes> {
        let mut buf = vec![0; (range.end - range.start) as usize];
-
-        self.file.read_exact(&mut buf).await?;
-        self.position = range.end;
-
+        self.0.seek(io::SeekFrom::Start(range.start)).await?;
+        self.0.read_exact(&mut buf).await?;
        Ok(Bytes::from(buf))
    }
 }
-
-#[cfg(test)]
-mod tests {
-    use common_test_util::temp_dir::create_named_temp_file;
-    use futures::io::AsyncReadExt as _;
-
-    use super::*;
-
-    #[tokio::test]
-    async fn test_async_read_adapter() {
-        let data = b"hello world";
-        let reader = Vec::from(data);
-        let mut adapter = AsyncReadAdapter::new(reader).await.unwrap();
-
-        let mut buf = Vec::new();
-        adapter.read_to_end(&mut buf).await.unwrap();
-        assert_eq!(buf, data);
-    }
-
-    #[tokio::test]
-    async fn test_async_read_adapter_large() {
-        let data = (0..20 * 1024 * 1024).map(|i| i as u8).collect::<Vec<u8>>();
-        let mut adapter = AsyncReadAdapter::new(data.clone()).await.unwrap();
-
-        let mut buf = Vec::new();
-        adapter.read_to_end(&mut buf).await.unwrap();
-        assert_eq!(buf, data);
-    }
-
-    #[tokio::test]
-    async fn test_file_reader() {
-        let file = create_named_temp_file();
-        let path = file.path();
-        let data = b"hello world";
-        tokio::fs::write(path, data).await.unwrap();
-
-        let mut reader = FileReader::new(path).await.unwrap();
-        let metadata = reader.metadata().await.unwrap();
-        assert_eq!(metadata.content_length, data.len() as u64);
-
-        let bytes = reader.read(0..metadata.content_length).await.unwrap();
-        assert_eq!(&*bytes, data);
-
-        let bytes = reader.read(0..5).await.unwrap();
-        assert_eq!(&*bytes, &data[..5]);
-    }
-}
--- a/src/common/base/src/secrets.rs
+++ b/src/common/base/src/secrets.rs
@@ -46,9 +46,8 @@ impl From<String> for SecretString {
    }
 }

-/// Wrapper type for values that contains secrets.
-///
-/// It attempts to limit accidental exposure and ensure secrets are wiped from memory when dropped.
+/// Wrapper type for values that contains secrets, which 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,15 +103,14 @@ 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", "wkt", "geo-types", "dep:geo"]
+geo = ["geohash", "h3o"]

 [dependencies]
 api.workspace = true
@@ -28,17 +28,13 @@ 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
-nalgebra = "0.33"
 num = "0.4"
 num-traits = "0.2"
 once_cell.workspace = true
 paste = "1.0"
-s2 = { version = "0.0.12", optional = true }
 serde.workspace = true
 serde_json.workspace = true
 session.workspace = true
@@ -47,10 +43,8 @@ sql.workspace = true
 statrs = "0.16"
 store-api.workspace = true
 table.workspace = true
-wkt = { version = "0.11", optional = true }

 [dev-dependencies]
-approx = "0.5"
 ron = "0.7"
 serde = { version = "1.0", features = ["derive"] }
 tokio.workspace = true
--- a/src/common/function/src/function_registry.rs
+++ b/src/common/function/src/function_registry.rs
@@ -27,7 +27,6 @@ use crate::scalars::matches::MatchesFunction;
 use crate::scalars::math::MathFunction;
 use crate::scalars::numpy::NumpyFunction;
 use crate::scalars::timestamp::TimestampFunction;
-use crate::scalars::vector::VectorFunction;
 use crate::system::SystemFunction;
 use crate::table::TableFunction;

@@ -121,9 +120,6 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
    // Json related functions
    JsonFunction::register(&function_registry);

-    // Vector related functions
-    VectorFunction::register(&function_registry);
-
    // Geo functions
    #[cfg(feature = "geo")]
    crate::scalars::geo::GeoFunctions::register(&function_registry);
--- a/src/common/function/src/scalars.rs
+++ b/src/common/function/src/scalars.rs
@@ -21,7 +21,6 @@ pub mod json;
 pub mod matches;
 pub mod math;
 pub mod numpy;
-pub mod vector;

 #[cfg(test)]
 pub(crate) mod test;
--- a/src/common/function/src/scalars/aggregate.rs
+++ b/src/common/function/src/scalars/aggregate.rs
@@ -31,6 +31,7 @@ pub use polyval::PolyvalAccumulatorCreator;
 pub use scipy_stats_norm_cdf::ScipyStatsNormCdfAccumulatorCreator;
 pub use scipy_stats_norm_pdf::ScipyStatsNormPdfAccumulatorCreator;

+use super::geo::encoding::JsonPathEncodeFunctionCreator;
 use crate::function_registry::FunctionRegistry;

 /// A function creates `AggregateFunctionCreator`.
@@ -92,11 +93,6 @@ impl AggregateFunctions {
        register_aggr_func!("scipystatsnormcdf", 2, ScipyStatsNormCdfAccumulatorCreator);
        register_aggr_func!("scipystatsnormpdf", 2, ScipyStatsNormPdfAccumulatorCreator);

-        #[cfg(feature = "geo")]
-        register_aggr_func!(
-            "json_encode_path",
-            3,
-            super::geo::encoding::JsonPathEncodeFunctionCreator
-        );
+        register_aggr_func!("json_encode_path", 3, JsonPathEncodeFunctionCreator);
    }
 }
--- a/src/common/function/src/scalars/date/date_add.rs
+++ b/src/common/function/src/scalars/date/date_add.rs
@@ -14,19 +14,18 @@

 use std::fmt;

-use common_query::error::{ArrowComputeSnafu, IntoVectorSnafu, InvalidFuncArgsSnafu, Result};
+use common_query::error::{InvalidFuncArgsSnafu, Result, UnsupportedInputDataTypeSnafu};
 use common_query::prelude::Signature;
-use datatypes::arrow::compute::kernels::numeric;
+use datatypes::data_type::DataType;
 use datatypes::prelude::ConcreteDataType;
-use datatypes::vectors::{Helper, VectorRef};
-use snafu::{ensure, ResultExt};
+use datatypes::value::ValueRef;
+use datatypes::vectors::VectorRef;
+use snafu::ensure;

 use crate::function::{Function, FunctionContext};
 use crate::helper;

-/// A function adds an interval value to Timestamp, Date, and return the result.
-/// The implementation of datetime type is based on Date64 which is incorrect so this function
-/// doesn't support the datetime type.
+/// A function adds an interval value to Timestamp, Date or DateTime, and return the result.
 #[derive(Clone, Debug, Default)]
 pub struct DateAddFunction;

@@ -45,6 +44,7 @@ impl Function for DateAddFunction {
        helper::one_of_sigs2(
            vec![
                ConcreteDataType::date_datatype(),
+                ConcreteDataType::datetime_datatype(),
                ConcreteDataType::timestamp_second_datatype(),
                ConcreteDataType::timestamp_millisecond_datatype(),
                ConcreteDataType::timestamp_microsecond_datatype(),
@@ -69,14 +69,64 @@ impl Function for DateAddFunction {
            }
        );

-        let left = columns[0].to_arrow_array();
-        let right = columns[1].to_arrow_array();
+        let left = &columns[0];
+        let right = &columns[1];

-        let result = numeric::add(&left, &right).context(ArrowComputeSnafu)?;
-        let arrow_type = result.data_type().clone();
-        Helper::try_into_vector(result).context(IntoVectorSnafu {
-            data_type: arrow_type,
-        })
+        let size = left.len();
+        let left_datatype = columns[0].data_type();
+        match left_datatype {
+            ConcreteDataType::Timestamp(_) => {
+                let mut result = left_datatype.create_mutable_vector(size);
+                for i in 0..size {
+                    let ts = left.get(i).as_timestamp();
+                    let interval = right.get(i).as_interval();
+
+                    let new_ts = match (ts, interval) {
+                        (Some(ts), Some(interval)) => ts.add_interval(interval),
+                        _ => ts,
+                    };
+
+                    result.push_value_ref(ValueRef::from(new_ts));
+                }
+
+                Ok(result.to_vector())
+            }
+            ConcreteDataType::Date(_) => {
+                let mut result = left_datatype.create_mutable_vector(size);
+                for i in 0..size {
+                    let date = left.get(i).as_date();
+                    let interval = right.get(i).as_interval();
+                    let new_date = match (date, interval) {
+                        (Some(date), Some(interval)) => date.add_interval(interval),
+                        _ => date,
+                    };
+
+                    result.push_value_ref(ValueRef::from(new_date));
+                }
+
+                Ok(result.to_vector())
+            }
+            ConcreteDataType::DateTime(_) => {
+                let mut result = left_datatype.create_mutable_vector(size);
+                for i in 0..size {
+                    let datetime = left.get(i).as_datetime();
+                    let interval = right.get(i).as_interval();
+                    let new_datetime = match (datetime, interval) {
+                        (Some(datetime), Some(interval)) => datetime.add_interval(interval),
+                        _ => datetime,
+                    };
+
+                    result.push_value_ref(ValueRef::from(new_datetime));
+                }
+
+                Ok(result.to_vector())
+            }
+            _ => UnsupportedInputDataTypeSnafu {
+                function: NAME,
+                datatypes: columns.iter().map(|c| c.data_type()).collect::<Vec<_>>(),
+            }
+            .fail(),
+        }
    }
 }

@@ -94,7 +144,8 @@ mod tests {
    use datatypes::prelude::ConcreteDataType;
    use datatypes::value::Value;
    use datatypes::vectors::{
-        DateVector, IntervalDayTimeVector, IntervalYearMonthVector, TimestampSecondVector,
+        DateTimeVector, DateVector, IntervalDayTimeVector, IntervalYearMonthVector,
+        TimestampSecondVector,
    };

    use super::{DateAddFunction, *};
@@ -117,15 +168,16 @@ mod tests {
            ConcreteDataType::date_datatype(),
            f.return_type(&[ConcreteDataType::date_datatype()]).unwrap()
        );
-        assert!(
-            matches!(f.signature(),
+        assert_eq!(
+            ConcreteDataType::datetime_datatype(),
+            f.return_type(&[ConcreteDataType::datetime_datatype()])
+                .unwrap()
+        );
+        assert!(matches!(f.signature(),
                         Signature {
                             type_signature: TypeSignature::OneOf(sigs),
                             volatility: Volatility::Immutable
-                         } if  sigs.len() == 15),
-            "{:?}",
-            f.signature()
-        );
+                         } if  sigs.len() == 18));
    }

    #[test]
@@ -191,4 +243,36 @@ mod tests {
            }
        }
    }
+
+    #[test]
+    fn test_datetime_date_add() {
+        let f = DateAddFunction;
+
+        let dates = vec![Some(123), None, Some(42), None];
+        // Intervals in months
+        let intervals = vec![1, 2, 3, 1];
+        let results = [Some(2678400123), None, Some(7776000042), None];
+
+        let date_vector = DateTimeVector::from(dates.clone());
+        let interval_vector = IntervalYearMonthVector::from_vec(intervals);
+        let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
+
+        assert_eq!(4, vector.len());
+        for (i, _t) in dates.iter().enumerate() {
+            let v = vector.get(i);
+            let result = results.get(i).unwrap();
+
+            if result.is_none() {
+                assert_eq!(Value::Null, v);
+                continue;
+            }
+            match v {
+                Value::DateTime(date) => {
+                    assert_eq!(date.val(), result.unwrap());
+                }
+                _ => unreachable!(),
+            }
+        }
+    }
 }
--- a/src/common/function/src/scalars/date/date_sub.rs
+++ b/src/common/function/src/scalars/date/date_sub.rs
@@ -14,19 +14,18 @@

 use std::fmt;

-use common_query::error::{ArrowComputeSnafu, IntoVectorSnafu, InvalidFuncArgsSnafu, Result};
+use common_query::error::{InvalidFuncArgsSnafu, Result, UnsupportedInputDataTypeSnafu};
 use common_query::prelude::Signature;
-use datatypes::arrow::compute::kernels::numeric;
+use datatypes::data_type::DataType;
 use datatypes::prelude::ConcreteDataType;
-use datatypes::vectors::{Helper, VectorRef};
-use snafu::{ensure, ResultExt};
+use datatypes::value::ValueRef;
+use datatypes::vectors::VectorRef;
+use snafu::ensure;

 use crate::function::{Function, FunctionContext};
 use crate::helper;

-/// A function subtracts an interval value to Timestamp, Date, and return the result.
-/// The implementation of datetime type is based on Date64 which is incorrect so this function
-/// doesn't support the datetime type.
+/// A function subtracts an interval value to Timestamp, Date or DateTime, and return the result.
 #[derive(Clone, Debug, Default)]
 pub struct DateSubFunction;

@@ -45,6 +44,7 @@ impl Function for DateSubFunction {
        helper::one_of_sigs2(
            vec![
                ConcreteDataType::date_datatype(),
+                ConcreteDataType::datetime_datatype(),
                ConcreteDataType::timestamp_second_datatype(),
                ConcreteDataType::timestamp_millisecond_datatype(),
                ConcreteDataType::timestamp_microsecond_datatype(),
@@ -69,14 +69,65 @@ impl Function for DateSubFunction {
            }
        );

-        let left = columns[0].to_arrow_array();
-        let right = columns[1].to_arrow_array();
+        let left = &columns[0];
+        let right = &columns[1];

-        let result = numeric::sub(&left, &right).context(ArrowComputeSnafu)?;
-        let arrow_type = result.data_type().clone();
-        Helper::try_into_vector(result).context(IntoVectorSnafu {
-            data_type: arrow_type,
-        })
+        let size = left.len();
+        let left_datatype = columns[0].data_type();
+
+        match left_datatype {
+            ConcreteDataType::Timestamp(_) => {
+                let mut result = left_datatype.create_mutable_vector(size);
+                for i in 0..size {
+                    let ts = left.get(i).as_timestamp();
+                    let interval = right.get(i).as_interval();
+
+                    let new_ts = match (ts, interval) {
+                        (Some(ts), Some(interval)) => ts.sub_interval(interval),
+                        _ => ts,
+                    };
+
+                    result.push_value_ref(ValueRef::from(new_ts));
+                }
+
+                Ok(result.to_vector())
+            }
+            ConcreteDataType::Date(_) => {
+                let mut result = left_datatype.create_mutable_vector(size);
+                for i in 0..size {
+                    let date = left.get(i).as_date();
+                    let interval = right.get(i).as_interval();
+                    let new_date = match (date, interval) {
+                        (Some(date), Some(interval)) => date.sub_interval(interval),
+                        _ => date,
+                    };
+
+                    result.push_value_ref(ValueRef::from(new_date));
+                }
+
+                Ok(result.to_vector())
+            }
+            ConcreteDataType::DateTime(_) => {
+                let mut result = left_datatype.create_mutable_vector(size);
+                for i in 0..size {
+                    let datetime = left.get(i).as_datetime();
+                    let interval = right.get(i).as_interval();
+                    let new_datetime = match (datetime, interval) {
+                        (Some(datetime), Some(interval)) => datetime.sub_interval(interval),
+                        _ => datetime,
+                    };
+
+                    result.push_value_ref(ValueRef::from(new_datetime));
+                }
+
+                Ok(result.to_vector())
+            }
+            _ => UnsupportedInputDataTypeSnafu {
+                function: NAME,
+                datatypes: columns.iter().map(|c| c.data_type()).collect::<Vec<_>>(),
+            }
+            .fail(),
+        }
    }
 }

@@ -94,7 +145,8 @@ mod tests {
    use datatypes::prelude::ConcreteDataType;
    use datatypes::value::Value;
    use datatypes::vectors::{
-        DateVector, IntervalDayTimeVector, IntervalYearMonthVector, TimestampSecondVector,
+        DateTimeVector, DateVector, IntervalDayTimeVector, IntervalYearMonthVector,
+        TimestampSecondVector,
    };

    use super::{DateSubFunction, *};
@@ -122,15 +174,11 @@ mod tests {
            f.return_type(&[ConcreteDataType::datetime_datatype()])
                .unwrap()
        );
-        assert!(
-            matches!(f.signature(),
+        assert!(matches!(f.signature(),
                         Signature {
                             type_signature: TypeSignature::OneOf(sigs),
                             volatility: Volatility::Immutable
-                         } if  sigs.len() == 15),
-            "{:?}",
-            f.signature()
-        );
+                         } if  sigs.len() == 18));
    }

    #[test]
@@ -202,4 +250,42 @@ mod tests {
            }
        }
    }
+
+    #[test]
+    fn test_datetime_date_sub() {
+        let f = DateSubFunction;
+        let millis_per_month = 3600 * 24 * 30 * 1000;
+
+        let dates = vec![
+            Some(123 * millis_per_month),
+            None,
+            Some(42 * millis_per_month),
+            None,
+        ];
+        // Intervals in months
+        let intervals = vec![1, 2, 3, 1];
+        let results = [Some(316137600000), None, Some(100915200000), None];
+
+        let date_vector = DateTimeVector::from(dates.clone());
+        let interval_vector = IntervalYearMonthVector::from_vec(intervals);
+        let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
+
+        assert_eq!(4, vector.len());
+        for (i, _t) in dates.iter().enumerate() {
+            let v = vector.get(i);
+            let result = results.get(i).unwrap();
+
+            if result.is_none() {
+                assert_eq!(Value::Null, v);
+                continue;
+            }
+            match v {
+                Value::DateTime(date) => {
+                    assert_eq!(date.val(), result.unwrap());
+                }
+                _ => unreachable!(),
+            }
+        }
+    }
 }
--- a/src/common/function/src/scalars/geo.rs
+++ b/src/common/function/src/scalars/geo.rs
@@ -17,10 +17,8 @@ pub(crate) mod encoding;
 mod geohash;
 mod h3;
 mod helpers;
-mod measure;
-mod relation;
-mod s2;
-mod wkt;
+
+use geohash::{GeohashFunction, GeohashNeighboursFunction};

 use crate::function_registry::FunctionRegistry;

@@ -29,8 +27,8 @@ pub(crate) struct GeoFunctions;
 impl GeoFunctions {
    pub fn register(registry: &FunctionRegistry) {
        // geohash
-        registry.register(Arc::new(geohash::GeohashFunction));
-        registry.register(Arc::new(geohash::GeohashNeighboursFunction));
+        registry.register(Arc::new(GeohashFunction));
+        registry.register(Arc::new(GeohashNeighboursFunction));

        // h3 index
        registry.register(Arc::new(h3::H3LatLngToCell));
@@ -51,35 +49,11 @@ 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));
        registry.register(Arc::new(h3::H3GridDiskDistances));
        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/encoding.rs
+++ b/src/common/function/src/scalars/geo/encoding.rs
@@ -17,7 +17,7 @@ use std::sync::Arc;
 use common_error::ext::{BoxedError, PlainError};
 use common_error::status_code::StatusCode;
 use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
-use common_query::error::{self, InvalidInputStateSnafu, Result};
+use common_query::error::{self, InvalidFuncArgsSnafu, InvalidInputStateSnafu, Result};
 use common_query::logical_plan::accumulator::AggrFuncTypeStore;
 use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
 use common_query::prelude::AccumulatorCreatorFunction;
--- a/src/common/function/src/scalars/geo/h3.rs
+++ b/src/common/function/src/scalars/geo/h3.rs
@@ -16,29 +16,28 @@ use std::str::FromStr;

 use common_error::ext::{BoxedError, PlainError};
 use common_error::status_code::StatusCode;
-use common_query::error::{self, Result};
+use common_query::error::{self, InvalidFuncArgsSnafu, Result};
 use common_query::prelude::{Signature, TypeSignature};
 use datafusion::logical_expr::Volatility;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::scalars::{Scalar, ScalarVectorBuilder};
 use datatypes::value::{ListValue, Value};
 use datatypes::vectors::{
-    BooleanVectorBuilder, Float64VectorBuilder, Int32VectorBuilder, ListVectorBuilder,
-    MutableVector, StringVectorBuilder, UInt64VectorBuilder, UInt8VectorBuilder, VectorRef,
+    BooleanVectorBuilder, Int32VectorBuilder, ListVectorBuilder, MutableVector,
+    StringVectorBuilder, UInt64VectorBuilder, UInt8VectorBuilder, VectorRef,
 };
 use derive_more::Display;
 use h3o::{CellIndex, LatLng, Resolution};
 use once_cell::sync::Lazy;
-use snafu::ResultExt;
+use snafu::{ensure, ResultExt};

-use super::helpers::{ensure_and_coerce, ensure_columns_len, ensure_columns_n};
+use super::helpers::{ensure_columns_len, ensure_columns_n};
 use crate::function::{Function, FunctionContext};

 static CELL_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
    vec![
        ConcreteDataType::int64_datatype(),
        ConcreteDataType::uint64_datatype(),
-        ConcreteDataType::string_datatype(),
    ]
 });

@@ -383,7 +382,15 @@ impl Function for H3CellResolution {
    }

    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure_columns_n!(columns, 1);
+        ensure!(
+            columns.len() == 1,
+            InvalidFuncArgsSnafu {
+                err_msg: format!(
+                    "The length of the args is not correct, expect 1, provided : {}",
+                    columns.len()
+                ),
+            }
+        );

        let cell_vec = &columns[0];
        let size = cell_vec.len();
@@ -953,181 +960,6 @@ 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,
@@ -1150,6 +982,18 @@ fn value_to_resolution(v: Value) -> Result<Resolution> {
        .context(error::ExecuteSnafu)
 }

+macro_rules! ensure_and_coerce {
+    ($compare:expr, $coerce:expr) => {{
+        ensure!(
+            $compare,
+            InvalidFuncArgsSnafu {
+                err_msg: "Argument was outside of acceptable range "
+            }
+        );
+        Ok($coerce)
+    }};
+}
+
 fn value_to_position(v: Value) -> Result<u64> {
    match v {
        Value::Int8(v) => ensure_and_coerce!(v >= 0, v as u64),
@@ -1249,126 +1093,7 @@ 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/helpers.rs
+++ b/src/common/function/src/scalars/geo/helpers.rs
@@ -14,15 +14,15 @@

 macro_rules! ensure_columns_len {
    ($columns:ident) => {
-        snafu::ensure!(
+        ensure!(
            $columns.windows(2).all(|c| c[0].len() == c[1].len()),
-            common_query::error::InvalidFuncArgsSnafu {
+            InvalidFuncArgsSnafu {
                err_msg: "The length of input columns are in different size"
            }
        )
    };
    ($column_a:ident, $column_b:ident, $($column_n:ident),*) => {
-        snafu::ensure!(
+        ensure!(
            {
                let mut result = $column_a.len() == $column_b.len();
                $(
@@ -30,7 +30,7 @@ macro_rules! ensure_columns_len {
                )*
                result
            }
-            common_query::error::InvalidFuncArgsSnafu {
+            InvalidFuncArgsSnafu {
                err_msg: "The length of input columns are in different size"
            }
        )
@@ -41,9 +41,9 @@ pub(super) use ensure_columns_len;

 macro_rules! ensure_columns_n {
    ($columns:ident, $n:literal) => {
-        snafu::ensure!(
+        ensure!(
            $columns.len() == $n,
-            common_query::error::InvalidFuncArgsSnafu {
+            InvalidFuncArgsSnafu {
                err_msg: format!(
                    "The length of arguments is not correct, expect {}, provided : {}",
                    stringify!($n),
@@ -59,17 +59,3 @@ macro_rules! ensure_columns_n {
 }

 pub(super) use ensure_columns_n;
-
-macro_rules! ensure_and_coerce {
-    ($compare:expr, $coerce:expr) => {{
-        snafu::ensure!(
-            $compare,
-            common_query::error::InvalidFuncArgsSnafu {
-                err_msg: "Argument was outside of acceptable range "
-            }
-        );
-        Ok($coerce)
-    }};
-}
-
-pub(super) use ensure_and_coerce;
--- a/src/common/function/src/scalars/geo/measure.rs
+++ b/src/common/function/src/scalars/geo/measure.rs
@@ -1,195 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_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
@@ -1,190 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_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/s2.rs
+++ b/src/common/function/src/scalars/geo/s2.rs
@@ -1,275 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_query::error::{InvalidFuncArgsSnafu, Result};
-use common_query::prelude::{Signature, TypeSignature};
-use datafusion::logical_expr::Volatility;
-use datatypes::prelude::ConcreteDataType;
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::value::Value;
-use datatypes::vectors::{MutableVector, StringVectorBuilder, UInt64VectorBuilder, VectorRef};
-use derive_more::Display;
-use once_cell::sync::Lazy;
-use s2::cellid::{CellID, MAX_LEVEL};
-use s2::latlng::LatLng;
-use snafu::ensure;
-
-use crate::function::{Function, FunctionContext};
-use crate::scalars::geo::helpers::{ensure_and_coerce, ensure_columns_len, ensure_columns_n};
-
-static CELL_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
-    vec![
-        ConcreteDataType::int64_datatype(),
-        ConcreteDataType::uint64_datatype(),
-    ]
-});
-
-static COORDINATE_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
-    vec![
-        ConcreteDataType::float32_datatype(),
-        ConcreteDataType::float64_datatype(),
-    ]
-});
-
-static LEVEL_TYPES: Lazy<Vec<ConcreteDataType>> = Lazy::new(|| {
-    vec![
-        ConcreteDataType::int8_datatype(),
-        ConcreteDataType::int16_datatype(),
-        ConcreteDataType::int32_datatype(),
-        ConcreteDataType::int64_datatype(),
-        ConcreteDataType::uint8_datatype(),
-        ConcreteDataType::uint16_datatype(),
-        ConcreteDataType::uint32_datatype(),
-        ConcreteDataType::uint64_datatype(),
-    ]
-});
-
-/// Function that returns [s2] encoding cellid for a given geospatial coordinate.
-///
-/// [s2]: http://s2geometry.io
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct S2LatLngToCell;
-
-impl Function for S2LatLngToCell {
-    fn name(&self) -> &str {
-        "s2_latlng_to_cell"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        let mut signatures = Vec::with_capacity(COORDINATE_TYPES.len());
-        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 lon_vec = &columns[1];
-
-        let size = lat_vec.len();
-        let mut results = UInt64VectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let lat = lat_vec.get(i).as_f64_lossy();
-            let lon = lon_vec.get(i).as_f64_lossy();
-
-            let result = match (lat, lon) {
-                (Some(lat), Some(lon)) => {
-                    let coord = LatLng::from_degrees(lat, lon);
-                    ensure!(
-                        coord.is_valid(),
-                        InvalidFuncArgsSnafu {
-                            err_msg: "The input coordinates are invalid",
-                        }
-                    );
-                    let cellid = CellID::from(coord);
-                    let encoded: u64 = cellid.0;
-                    Some(encoded)
-                }
-                _ => None,
-            };
-
-            results.push(result);
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-/// Return the level of current s2 cell
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct S2CellLevel;
-
-impl Function for S2CellLevel {
-    fn name(&self) -> &str {
-        "s2_cell_level"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        signature_of_cell()
-    }
-
-    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure_columns_n!(columns, 1);
-
-        let cell_vec = &columns[0];
-        let size = cell_vec.len();
-        let mut results = UInt64VectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let cell = cell_from_value(cell_vec.get(i));
-            let res = cell.map(|cell| cell.level());
-
-            results.push(res);
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-/// Return the string presentation of the cell
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct S2CellToToken;
-
-impl Function for S2CellToToken {
-    fn name(&self) -> &str {
-        "s2_cell_to_token"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::string_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        signature_of_cell()
-    }
-
-    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure_columns_n!(columns, 1);
-
-        let cell_vec = &columns[0];
-        let size = cell_vec.len();
-        let mut results = StringVectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let cell = cell_from_value(cell_vec.get(i));
-            let res = cell.map(|cell| cell.to_token());
-
-            results.push(res.as_deref());
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-/// Return parent at given level of current s2 cell
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct S2CellParent;
-
-impl Function for S2CellParent {
-    fn name(&self) -> &str {
-        "s2_cell_parent"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        signature_of_cell_and_level()
-    }
-
-    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure_columns_n!(columns, 2);
-
-        let cell_vec = &columns[0];
-        let level_vec = &columns[1];
-        let size = cell_vec.len();
-        let mut results = UInt64VectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let cell = cell_from_value(cell_vec.get(i));
-            let level = value_to_level(level_vec.get(i))?;
-            let result = cell.map(|cell| cell.parent(level).0);
-
-            results.push(result);
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-fn signature_of_cell() -> Signature {
-    let mut signatures = Vec::with_capacity(CELL_TYPES.len());
-    for cell_type in CELL_TYPES.as_slice() {
-        signatures.push(TypeSignature::Exact(vec![cell_type.clone()]));
-    }
-
-    Signature::one_of(signatures, Volatility::Stable)
-}
-
-fn signature_of_cell_and_level() -> Signature {
-    let mut signatures = Vec::with_capacity(CELL_TYPES.len() * LEVEL_TYPES.len());
-    for cell_type in CELL_TYPES.as_slice() {
-        for level_type in LEVEL_TYPES.as_slice() {
-            signatures.push(TypeSignature::Exact(vec![
-                cell_type.clone(),
-                level_type.clone(),
-            ]));
-        }
-    }
-    Signature::one_of(signatures, Volatility::Stable)
-}
-
-fn cell_from_value(v: Value) -> Option<CellID> {
-    match v {
-        Value::Int64(v) => Some(CellID(v as u64)),
-        Value::UInt64(v) => Some(CellID(v)),
-        _ => None,
-    }
-}
-
-fn value_to_level(v: Value) -> Result<u64> {
-    match v {
-        Value::Int8(v) => ensure_and_coerce!(v >= 0 && v <= MAX_LEVEL as i8, v as u64),
-        Value::Int16(v) => ensure_and_coerce!(v >= 0 && v <= MAX_LEVEL as i16, v as u64),
-        Value::Int32(v) => ensure_and_coerce!(v >= 0 && v <= MAX_LEVEL as i32, v as u64),
-        Value::Int64(v) => ensure_and_coerce!(v >= 0 && v <= MAX_LEVEL as i64, v as u64),
-        Value::UInt8(v) => ensure_and_coerce!(v <= MAX_LEVEL as u8, v as u64),
-        Value::UInt16(v) => ensure_and_coerce!(v <= MAX_LEVEL as u16, v as u64),
-        Value::UInt32(v) => ensure_and_coerce!(v <= MAX_LEVEL as u32, v as u64),
-        Value::UInt64(v) => ensure_and_coerce!(v <= MAX_LEVEL, v),
-        _ => unreachable!(),
-    }
-}
--- a/src/common/function/src/scalars/geo/wkt.rs
+++ b/src/common/function/src/scalars/geo/wkt.rs
@@ -1,100 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use common_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,7 +16,6 @@ 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;

@@ -50,6 +49,5 @@ 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, TypeSignature};
+use common_query::prelude::Signature;
 use datafusion::logical_expr::Volatility;
 use datatypes::data_type::ConcreteDataType;
 use datatypes::prelude::VectorRef;
@@ -41,24 +41,10 @@ impl Function for JsonPathExistsFunction {
    }

    fn signature(&self) -> Signature {
-        Signature::one_of(
+        Signature::exact(
            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(),
-                ]),
+                ConcreteDataType::json_datatype(),
+                ConcreteDataType::string_datatype(),
            ],
            Volatility::Immutable,
        )
@@ -78,26 +64,25 @@ 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 (jsons.data_type(), paths.data_type()) {
-            (ConcreteDataType::Binary(_), ConcreteDataType::String(_)) => {
+        match datatype {
+            // JSON data type uses binary vector
+            ConcreteDataType::Binary(_) => {
                for i in 0..size {
-                    let result = match (jsons.get_ref(i).as_binary(), paths.get_ref(i).as_string())
-                    {
+                    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) {
                        (Ok(Some(json)), Ok(Some(path))) => {
-                            // 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()
+                            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,
+                            }
                        }
                        _ => None,
                    };
@@ -105,12 +90,6 @@ 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,
@@ -135,8 +114,8 @@ mod tests {
    use std::sync::Arc;

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

    use super::*;

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

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

        let json_strings = [
@@ -182,15 +143,9 @@ 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", "$[0]", "$.a", "null", "null",
-        ];
-        let expected = [false, true, true, false, true, false, false, false];
+        let paths = vec!["$.a.b.c", "$.b", "$.c.a", ".d"];
+        let results = [false, true, true, false];

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

-        // Test for non-nulls.
-        assert_eq!(8, vector.len());
-        for (i, real) in expected.iter().enumerate() {
+        assert_eq!(4, vector.len());
+        for (i, gt) in results.iter().enumerate() {
            let result = vector.get_ref(i);
-            assert!(!result.is_null());
-            let val = result.as_boolean().unwrap().unwrap();
-            assert_eq!(val, *real);
+            let result = result.as_boolean().unwrap().unwrap();
+            assert_eq!(*gt, result);
        }
-
-        // 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
@@ -1,202 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use 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/function/src/scalars/timestamp/greatest.rs
+++ b/src/common/function/src/scalars/timestamp/greatest.rs
@@ -22,12 +22,8 @@ use datafusion::arrow::compute::kernels::cmp::gt;
 use datatypes::arrow::array::AsArray;
 use datatypes::arrow::compute::cast;
 use datatypes::arrow::compute::kernels::zip;
-use datatypes::arrow::datatypes::{
-    DataType as ArrowDataType, Date32Type, Date64Type, TimestampMicrosecondType,
-    TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType,
-};
+use datatypes::arrow::datatypes::{DataType as ArrowDataType, Date32Type};
 use datatypes::prelude::ConcreteDataType;
-use datatypes::types::TimestampType;
 use datatypes::vectors::{Helper, VectorRef};
 use snafu::{ensure, ResultExt};

@@ -38,47 +34,13 @@ pub struct GreatestFunction;

 const NAME: &str = "greatest";

-macro_rules! gt_time_types {
-    ($ty: ident, $columns:expr) => {{
-        let column1 = $columns[0].to_arrow_array();
-        let column2 = $columns[1].to_arrow_array();
-
-        let column1 = column1.as_primitive::<$ty>();
-        let column2 = column2.as_primitive::<$ty>();
-        let boolean_array = gt(&column1, &column2).context(ArrowComputeSnafu)?;
-
-        let result = zip::zip(&boolean_array, &column1, &column2).context(ArrowComputeSnafu)?;
-        Helper::try_into_vector(&result).context(error::FromArrowArraySnafu)
-    }};
-}
-
 impl Function for GreatestFunction {
    fn name(&self) -> &str {
        NAME
    }

-    fn return_type(&self, input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        ensure!(
-            input_types.len() == 2,
-            InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "The length of the args is not correct, expect exactly two, have: {}",
-                    input_types.len()
-                )
-            }
-        );
-
-        match &input_types[0] {
-            ConcreteDataType::String(_) => Ok(ConcreteDataType::datetime_datatype()),
-            ConcreteDataType::Date(_) => Ok(ConcreteDataType::date_datatype()),
-            ConcreteDataType::DateTime(_) => Ok(ConcreteDataType::datetime_datatype()),
-            ConcreteDataType::Timestamp(ts_type) => Ok(ConcreteDataType::Timestamp(*ts_type)),
-            _ => UnsupportedInputDataTypeSnafu {
-                function: NAME,
-                datatypes: input_types,
-            }
-            .fail(),
-        }
+    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
+        Ok(ConcreteDataType::date_datatype())
    }

    fn signature(&self) -> Signature {
@@ -87,11 +49,6 @@ impl Function for GreatestFunction {
            vec![
                ConcreteDataType::string_datatype(),
                ConcreteDataType::date_datatype(),
-                ConcreteDataType::datetime_datatype(),
-                ConcreteDataType::timestamp_nanosecond_datatype(),
-                ConcreteDataType::timestamp_microsecond_datatype(),
-                ConcreteDataType::timestamp_millisecond_datatype(),
-                ConcreteDataType::timestamp_second_datatype(),
            ],
            Volatility::Immutable,
        )
@@ -109,32 +66,27 @@ impl Function for GreatestFunction {
        );
        match columns[0].data_type() {
            ConcreteDataType::String(_) => {
-                // Treats string as `DateTime` type.
-                let column1 = cast(&columns[0].to_arrow_array(), &ArrowDataType::Date64)
+                let column1 = cast(&columns[0].to_arrow_array(), &ArrowDataType::Date32)
                    .context(ArrowComputeSnafu)?;
-                let column1 = column1.as_primitive::<Date64Type>();
-                let column2 = cast(&columns[1].to_arrow_array(), &ArrowDataType::Date64)
+                let column1 = column1.as_primitive::<Date32Type>();
+                let column2 = cast(&columns[1].to_arrow_array(), &ArrowDataType::Date32)
                    .context(ArrowComputeSnafu)?;
-                let column2 = column2.as_primitive::<Date64Type>();
+                let column2 = column2.as_primitive::<Date32Type>();
+                let boolean_array = gt(&column1, &column2).context(ArrowComputeSnafu)?;
+                let result =
+                    zip::zip(&boolean_array, &column1, &column2).context(ArrowComputeSnafu)?;
+                Ok(Helper::try_into_vector(&result).context(error::FromArrowArraySnafu)?)
+            }
+            ConcreteDataType::Date(_) => {
+                let column1 = columns[0].to_arrow_array();
+                let column1 = column1.as_primitive::<Date32Type>();
+                let column2 = columns[1].to_arrow_array();
+                let column2 = column2.as_primitive::<Date32Type>();
                let boolean_array = gt(&column1, &column2).context(ArrowComputeSnafu)?;
                let result =
                    zip::zip(&boolean_array, &column1, &column2).context(ArrowComputeSnafu)?;
                Ok(Helper::try_into_vector(&result).context(error::FromArrowArraySnafu)?)
            }
-            ConcreteDataType::Date(_) => gt_time_types!(Date32Type, columns),
-            ConcreteDataType::DateTime(_) => gt_time_types!(Date64Type, columns),
-            ConcreteDataType::Timestamp(ts_type) => match ts_type {
-                TimestampType::Second(_) => gt_time_types!(TimestampSecondType, columns),
-                TimestampType::Millisecond(_) => {
-                    gt_time_types!(TimestampMillisecondType, columns)
-                }
-                TimestampType::Microsecond(_) => {
-                    gt_time_types!(TimestampMicrosecondType, columns)
-                }
-                TimestampType::Nanosecond(_) => {
-                    gt_time_types!(TimestampNanosecondType, columns)
-                }
-            },
            _ => UnsupportedInputDataTypeSnafu {
                function: NAME,
                datatypes: columns.iter().map(|c| c.data_type()).collect::<Vec<_>>(),
@@ -154,31 +106,19 @@ impl fmt::Display for GreatestFunction {
 mod tests {
    use std::sync::Arc;

-    use common_time::timestamp::TimeUnit;
-    use common_time::{Date, DateTime, Timestamp};
-    use datatypes::types::{
-        DateTimeType, DateType, TimestampMicrosecondType, TimestampMillisecondType,
-        TimestampNanosecondType, TimestampSecondType,
-    };
+    use common_time::Date;
+    use datatypes::prelude::ConcreteDataType;
+    use datatypes::types::DateType;
    use datatypes::value::Value;
-    use datatypes::vectors::{
-        DateTimeVector, DateVector, StringVector, TimestampMicrosecondVector,
-        TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, Vector,
-    };
-    use paste::paste;
+    use datatypes::vectors::{DateVector, StringVector, Vector};

    use super::*;
    #[test]
    fn test_greatest_takes_string_vector() {
        let function = GreatestFunction;
        assert_eq!(
-            function
-                .return_type(&[
-                    ConcreteDataType::string_datatype(),
-                    ConcreteDataType::string_datatype()
-                ])
-                .unwrap(),
-            ConcreteDataType::DateTime(DateTimeType)
+            function.return_type(&[]).unwrap(),
+            ConcreteDataType::Date(DateType)
        );
        let columns = vec![
            Arc::new(StringVector::from(vec![
@@ -192,15 +132,15 @@ mod tests {
        ];

        let result = function.eval(FunctionContext::default(), &columns).unwrap();
-        let result = result.as_any().downcast_ref::<DateTimeVector>().unwrap();
+        let result = result.as_any().downcast_ref::<DateVector>().unwrap();
        assert_eq!(result.len(), 2);
        assert_eq!(
            result.get(0),
-            Value::DateTime(DateTime::from_str("2001-02-01 00:00:00", None).unwrap())
+            Value::Date(Date::from_str_utc("2001-02-01").unwrap())
        );
        assert_eq!(
            result.get(1),
-            Value::DateTime(DateTime::from_str("2012-12-23 00:00:00", None).unwrap())
+            Value::Date(Date::from_str_utc("2012-12-23").unwrap())
        );
    }

@@ -208,15 +148,9 @@ mod tests {
    fn test_greatest_takes_date_vector() {
        let function = GreatestFunction;
        assert_eq!(
-            function
-                .return_type(&[
-                    ConcreteDataType::date_datatype(),
-                    ConcreteDataType::date_datatype()
-                ])
-                .unwrap(),
+            function.return_type(&[]).unwrap(),
            ConcreteDataType::Date(DateType)
        );
-
        let columns = vec![
            Arc::new(DateVector::from_slice(vec![-1, 2])) as _,
            Arc::new(DateVector::from_slice(vec![0, 1])) as _,
@@ -234,81 +168,4 @@ mod tests {
            Value::Date(Date::from_str_utc("1970-01-03").unwrap())
        );
    }
-
-    #[test]
-    fn test_greatest_takes_datetime_vector() {
-        let function = GreatestFunction;
-        assert_eq!(
-            function
-                .return_type(&[
-                    ConcreteDataType::datetime_datatype(),
-                    ConcreteDataType::datetime_datatype()
-                ])
-                .unwrap(),
-            ConcreteDataType::DateTime(DateTimeType)
-        );
-
-        let columns = vec![
-            Arc::new(DateTimeVector::from_slice(vec![-1, 2])) as _,
-            Arc::new(DateTimeVector::from_slice(vec![0, 1])) as _,
-        ];
-
-        let result = function.eval(FunctionContext::default(), &columns).unwrap();
-        let result = result.as_any().downcast_ref::<DateTimeVector>().unwrap();
-        assert_eq!(result.len(), 2);
-        assert_eq!(
-            result.get(0),
-            Value::DateTime(DateTime::from_str("1970-01-01 00:00:00", None).unwrap())
-        );
-        assert_eq!(
-            result.get(1),
-            Value::DateTime(DateTime::from_str("1970-01-01 00:00:00.002", None).unwrap())
-        );
-    }
-
-    macro_rules! test_timestamp {
-        ($type: expr,$unit: ident) => {
-            paste! {
-                #[test]
-                fn [<test_greatest_takes_ $unit:lower _vector>]() {
-                    let function = GreatestFunction;
-                    assert_eq!(
-                        function.return_type(&[$type, $type]).unwrap(),
-                        ConcreteDataType::Timestamp(TimestampType::$unit([<Timestamp $unit Type>]))
-                    );
-
-                    let columns = vec![
-                        Arc::new([<Timestamp $unit Vector>]::from_slice(vec![-1, 2])) as _,
-                        Arc::new([<Timestamp $unit Vector>]::from_slice(vec![0, 1])) as _,
-                    ];
-
-                    let result = function.eval(FunctionContext::default(), &columns).unwrap();
-                    let result = result.as_any().downcast_ref::<[<Timestamp $unit Vector>]>().unwrap();
-                    assert_eq!(result.len(), 2);
-                    assert_eq!(
-                        result.get(0),
-                        Value::Timestamp(Timestamp::new(0, TimeUnit::$unit))
-                    );
-                    assert_eq!(
-                        result.get(1),
-                        Value::Timestamp(Timestamp::new(2, TimeUnit::$unit))
-                    );
-                }
-            }
-        }
-    }
-
-    test_timestamp!(
-        ConcreteDataType::timestamp_nanosecond_datatype(),
-        Nanosecond
-    );
-    test_timestamp!(
-        ConcreteDataType::timestamp_microsecond_datatype(),
-        Microsecond
-    );
-    test_timestamp!(
-        ConcreteDataType::timestamp_millisecond_datatype(),
-        Millisecond
-    );
-    test_timestamp!(ConcreteDataType::timestamp_second_datatype(), Second);
 }
--- a/src/common/function/src/scalars/vector.rs
+++ b/src/common/function/src/scalars/vector.rs
@@ -1,35 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-mod convert;
-mod distance;
-
-use std::sync::Arc;
-
-use crate::function_registry::FunctionRegistry;
-
-pub(crate) struct VectorFunction;
-
-impl VectorFunction {
-    pub fn register(registry: &FunctionRegistry) {
-        // conversion
-        registry.register(Arc::new(convert::ParseVectorFunction));
-        registry.register(Arc::new(convert::VectorToStringFunction));
-
-        // distance
-        registry.register(Arc::new(distance::CosDistanceFunction));
-        registry.register(Arc::new(distance::DotProductFunction));
-        registry.register(Arc::new(distance::L2SqDistanceFunction));
-    }
-}
--- a/src/common/function/src/scalars/vector/convert/parse_vector.rs
+++ b/src/common/function/src/scalars/vector/convert/parse_vector.rs
@@ -1,160 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::fmt::Display;
-
-use common_query::error::{InvalidFuncArgsSnafu, InvalidVectorStringSnafu, Result};
-use common_query::prelude::{Signature, Volatility};
-use datatypes::prelude::ConcreteDataType;
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::types::parse_string_to_vector_type_value;
-use datatypes::vectors::{BinaryVectorBuilder, MutableVector, VectorRef};
-use snafu::{ensure, ResultExt};
-
-use crate::function::{Function, FunctionContext};
-
-const NAME: &str = "parse_vec";
-
-#[derive(Debug, Clone, Default)]
-pub struct ParseVectorFunction;
-
-impl Function for ParseVectorFunction {
-    fn name(&self) -> &str {
-        NAME
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::binary_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::exact(
-            vec![ConcreteDataType::string_datatype()],
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 1,
-            InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "The length of the args is not correct, expect exactly one, have: {}",
-                    columns.len()
-                ),
-            }
-        );
-
-        let column = &columns[0];
-        let size = column.len();
-
-        let mut result = BinaryVectorBuilder::with_capacity(size);
-        for i in 0..size {
-            let value = column.get(i).as_string();
-            if let Some(value) = value {
-                let res = parse_string_to_vector_type_value(&value, None)
-                    .context(InvalidVectorStringSnafu { vec_str: &value })?;
-                result.push(Some(&res));
-            } else {
-                result.push_null();
-            }
-        }
-
-        Ok(result.to_vector())
-    }
-}
-
-impl Display for ParseVectorFunction {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", NAME.to_ascii_uppercase())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use common_base::bytes::Bytes;
-    use datatypes::value::Value;
-    use datatypes::vectors::StringVector;
-
-    use super::*;
-
-    #[test]
-    fn test_parse_vector() {
-        let func = ParseVectorFunction;
-
-        let input = Arc::new(StringVector::from(vec![
-            Some("[1.0,2.0,3.0]".to_string()),
-            Some("[4.0,5.0,6.0]".to_string()),
-            None,
-        ]));
-
-        let result = func.eval(FunctionContext::default(), &[input]).unwrap();
-
-        let result = result.as_ref();
-        assert_eq!(result.len(), 3);
-        assert_eq!(
-            result.get(0),
-            Value::Binary(Bytes::from(
-                [1.0f32, 2.0, 3.0]
-                    .iter()
-                    .flat_map(|e| e.to_le_bytes())
-                    .collect::<Vec<u8>>()
-            ))
-        );
-        assert_eq!(
-            result.get(1),
-            Value::Binary(Bytes::from(
-                [4.0f32, 5.0, 6.0]
-                    .iter()
-                    .flat_map(|e| e.to_le_bytes())
-                    .collect::<Vec<u8>>()
-            ))
-        );
-        assert!(result.get(2).is_null());
-    }
-
-    #[test]
-    fn test_parse_vector_error() {
-        let func = ParseVectorFunction;
-
-        let input = Arc::new(StringVector::from(vec![
-            Some("[1.0,2.0,3.0]".to_string()),
-            Some("[4.0,5.0,6.0]".to_string()),
-            Some("[7.0,8.0,9.0".to_string()),
-        ]));
-
-        let result = func.eval(FunctionContext::default(), &[input]);
-        assert!(result.is_err());
-
-        let input = Arc::new(StringVector::from(vec![
-            Some("[1.0,2.0,3.0]".to_string()),
-            Some("[4.0,5.0,6.0]".to_string()),
-            Some("7.0,8.0,9.0]".to_string()),
-        ]));
-
-        let result = func.eval(FunctionContext::default(), &[input]);
-        assert!(result.is_err());
-
-        let input = Arc::new(StringVector::from(vec![
-            Some("[1.0,2.0,3.0]".to_string()),
-            Some("[4.0,5.0,6.0]".to_string()),
-            Some("[7.0,hello,9.0]".to_string()),
-        ]));
-
-        let result = func.eval(FunctionContext::default(), &[input]);
-        assert!(result.is_err());
-    }
-}
--- a/src/common/function/src/scalars/vector/convert/vector_to_string.rs
+++ b/src/common/function/src/scalars/vector/convert/vector_to_string.rs
@@ -1,139 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::fmt::Display;
-
-use common_query::error::{InvalidFuncArgsSnafu, Result};
-use common_query::prelude::{Signature, Volatility};
-use datatypes::prelude::ConcreteDataType;
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::types::vector_type_value_to_string;
-use datatypes::value::Value;
-use datatypes::vectors::{MutableVector, StringVectorBuilder, VectorRef};
-use snafu::ensure;
-
-use crate::function::{Function, FunctionContext};
-
-const NAME: &str = "vec_to_string";
-
-#[derive(Debug, Clone, Default)]
-pub struct VectorToStringFunction;
-
-impl Function for VectorToStringFunction {
-    fn name(&self) -> &str {
-        NAME
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::string_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::exact(
-            vec![ConcreteDataType::binary_datatype()],
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 1,
-            InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "The length of the args is not correct, expect exactly one, have: {}",
-                    columns.len()
-                ),
-            }
-        );
-
-        let column = &columns[0];
-        let size = column.len();
-
-        let mut result = StringVectorBuilder::with_capacity(size);
-        for i in 0..size {
-            let value = column.get(i);
-            match value {
-                Value::Binary(bytes) => {
-                    let len = bytes.len();
-                    if len % std::mem::size_of::<f32>() != 0 {
-                        return InvalidFuncArgsSnafu {
-                            err_msg: format!("Invalid binary length of vector: {}", len),
-                        }
-                        .fail();
-                    }
-
-                    let dim = len / std::mem::size_of::<f32>();
-                    // Safety: `dim` is calculated from the length of `bytes` and is guaranteed to be valid
-                    let res = vector_type_value_to_string(&bytes, dim as _).unwrap();
-                    result.push(Some(&res));
-                }
-                Value::Null => {
-                    result.push_null();
-                }
-                _ => {
-                    return InvalidFuncArgsSnafu {
-                        err_msg: format!("Invalid value type: {:?}", value.data_type()),
-                    }
-                    .fail();
-                }
-            }
-        }
-
-        Ok(result.to_vector())
-    }
-}
-
-impl Display for VectorToStringFunction {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", NAME.to_ascii_uppercase())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use datatypes::value::Value;
-    use datatypes::vectors::BinaryVectorBuilder;
-
-    use super::*;
-
-    #[test]
-    fn test_vector_to_string() {
-        let func = VectorToStringFunction;
-
-        let mut builder = BinaryVectorBuilder::with_capacity(3);
-        builder.push(Some(
-            [1.0f32, 2.0, 3.0]
-                .iter()
-                .flat_map(|e| e.to_le_bytes())
-                .collect::<Vec<_>>()
-                .as_slice(),
-        ));
-        builder.push(Some(
-            [4.0f32, 5.0, 6.0]
-                .iter()
-                .flat_map(|e| e.to_le_bytes())
-                .collect::<Vec<_>>()
-                .as_slice(),
-        ));
-        builder.push_null();
-        let vector = builder.to_vector();
-
-        let result = func.eval(FunctionContext::default(), &[vector]).unwrap();
-
-        assert_eq!(result.len(), 3);
-        assert_eq!(result.get(0), Value::String("[1,2,3]".to_string().into()));
-        assert_eq!(result.get(1), Value::String("[4,5,6]".to_string().into()));
-        assert_eq!(result.get(2), Value::Null);
-    }
-}
--- a/src/common/function/src/scalars/vector/distance.rs
+++ b/src/common/function/src/scalars/vector/distance.rs
@@ -1,482 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-mod cos;
-mod dot;
-mod l2sq;
-
-use std::borrow::Cow;
-use std::fmt::Display;
-use std::sync::Arc;
-
-use common_query::error::{InvalidFuncArgsSnafu, Result};
-use common_query::prelude::Signature;
-use datatypes::prelude::ConcreteDataType;
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::value::ValueRef;
-use datatypes::vectors::{Float32VectorBuilder, MutableVector, Vector, VectorRef};
-use snafu::ensure;
-
-use crate::function::{Function, FunctionContext};
-use crate::helper;
-
-macro_rules! define_distance_function {
-    ($StructName:ident, $display_name:expr, $similarity_method:path) => {
-
-        /// A function calculates the distance between two vectors.
-
-        #[derive(Debug, Clone, Default)]
-        pub struct $StructName;
-
-        impl Function for $StructName {
-            fn name(&self) -> &str {
-                $display_name
-            }
-
-            fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-                Ok(ConcreteDataType::float32_datatype())
-            }
-
-            fn signature(&self) -> Signature {
-                helper::one_of_sigs2(
-                    vec![
-                        ConcreteDataType::string_datatype(),
-                        ConcreteDataType::binary_datatype(),
-                    ],
-                    vec![
-                        ConcreteDataType::string_datatype(),
-                        ConcreteDataType::binary_datatype(),
-                    ],
-                )
-            }
-
-            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 arg0 = &columns[0];
-                let arg1 = &columns[1];
-
-                let size = arg0.len();
-                let mut result = Float32VectorBuilder::with_capacity(size);
-                if size == 0 {
-                    return Ok(result.to_vector());
-                }
-
-                let arg0_const = parse_if_constant_string(arg0)?;
-                let arg1_const = parse_if_constant_string(arg1)?;
-
-                for i in 0..size {
-                    let vec0 = match arg0_const.as_ref() {
-                        Some(a) => Some(Cow::Borrowed(a.as_slice())),
-                        None => as_vector(arg0.get_ref(i))?,
-                    };
-                    let vec1 = match arg1_const.as_ref() {
-                        Some(b) => Some(Cow::Borrowed(b.as_slice())),
-                        None => as_vector(arg1.get_ref(i))?,
-                    };
-
-                    if let (Some(vec0), Some(vec1)) = (vec0, vec1) {
-                        ensure!(
-                            vec0.len() == vec1.len(),
-                            InvalidFuncArgsSnafu {
-                                err_msg: format!(
-                                    "The length of the vectors must match to calculate distance, have: {} vs {}",
-                                    vec0.len(),
-                                    vec1.len()
-                                ),
-                            }
-                        );
-
-                        // Checked if the length of the vectors match
-                        let d = $similarity_method(vec0.as_ref(), vec1.as_ref());
-                        result.push(Some(d));
-                    } else {
-                        result.push_null();
-                    }
-                }
-
-                return Ok(result.to_vector());
-            }
-        }
-
-        impl Display for $StructName {
-            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-                write!(f, "{}", $display_name.to_ascii_uppercase())
-            }
-        }
-    }
-}
-
-define_distance_function!(CosDistanceFunction, "vec_cos_distance", cos::cos);
-define_distance_function!(L2SqDistanceFunction, "vec_l2sq_distance", l2sq::l2sq);
-define_distance_function!(DotProductFunction, "vec_dot_product", dot::dot);
-
-/// Parse a vector value if the value is a constant string.
-fn parse_if_constant_string(arg: &Arc<dyn Vector>) -> Result<Option<Vec<f32>>> {
-    if !arg.is_const() {
-        return Ok(None);
-    }
-    if arg.data_type() != ConcreteDataType::string_datatype() {
-        return Ok(None);
-    }
-    arg.get_ref(0)
-        .as_string()
-        .unwrap() // Safe: checked if it is a string
-        .map(parse_f32_vector_from_string)
-        .transpose()
-}
-
-/// Convert a value to a vector value.
-/// Supported data types are binary and string.
-fn as_vector(arg: ValueRef<'_>) -> Result<Option<Cow<'_, [f32]>>> {
-    match arg.data_type() {
-        ConcreteDataType::Binary(_) => arg
-            .as_binary()
-            .unwrap() // Safe: checked if it is a binary
-            .map(binary_as_vector)
-            .transpose(),
-        ConcreteDataType::String(_) => arg
-            .as_string()
-            .unwrap() // Safe: checked if it is a string
-            .map(|s| Ok(Cow::Owned(parse_f32_vector_from_string(s)?)))
-            .transpose(),
-        ConcreteDataType::Null(_) => Ok(None),
-        _ => InvalidFuncArgsSnafu {
-            err_msg: format!("Unsupported data type: {:?}", arg.data_type()),
-        }
-        .fail(),
-    }
-}
-
-/// Convert a u8 slice to a vector value.
-fn binary_as_vector(bytes: &[u8]) -> Result<Cow<'_, [f32]>> {
-    if bytes.len() % std::mem::size_of::<f32>() != 0 {
-        return InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid binary length of vector: {}", bytes.len()),
-        }
-        .fail();
-    }
-
-    if cfg!(target_endian = "little") {
-        Ok(unsafe {
-            let vec = std::slice::from_raw_parts(
-                bytes.as_ptr() as *const f32,
-                bytes.len() / std::mem::size_of::<f32>(),
-            );
-            Cow::Borrowed(vec)
-        })
-    } else {
-        let v = bytes
-            .chunks_exact(std::mem::size_of::<f32>())
-            .map(|chunk| f32::from_le_bytes(chunk.try_into().unwrap()))
-            .collect::<Vec<f32>>();
-        Ok(Cow::Owned(v))
-    }
-}
-
-/// Parse a string to a vector value.
-/// Valid inputs are strings like "[1.0, 2.0, 3.0]".
-fn parse_f32_vector_from_string(s: &str) -> Result<Vec<f32>> {
-    let trimmed = s.trim();
-    if !trimmed.starts_with('[') || !trimmed.ends_with(']') {
-        return InvalidFuncArgsSnafu {
-            err_msg: format!(
-                "Failed to parse {s} to Vector value: not properly enclosed in brackets"
-            ),
-        }
-        .fail();
-    }
-    let content = trimmed[1..trimmed.len() - 1].trim();
-    if content.is_empty() {
-        return Ok(Vec::new());
-    }
-
-    content
-        .split(',')
-        .map(|s| s.trim().parse::<f32>())
-        .collect::<std::result::Result<_, _>>()
-        .map_err(|e| {
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Failed to parse {s} to Vector value: {e}"),
-            }
-            .build()
-        })
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use datatypes::vectors::{BinaryVector, ConstantVector, StringVector};
-
-    use super::*;
-
-    #[test]
-    fn test_distance_string_string() {
-        let funcs = [
-            Box::new(CosDistanceFunction {}) as Box<dyn Function>,
-            Box::new(L2SqDistanceFunction {}) as Box<dyn Function>,
-            Box::new(DotProductFunction {}) as Box<dyn Function>,
-        ];
-
-        for func in funcs {
-            let vec1 = Arc::new(StringVector::from(vec![
-                Some("[0.0, 1.0]"),
-                Some("[1.0, 0.0]"),
-                None,
-                Some("[1.0, 0.0]"),
-            ])) as VectorRef;
-            let vec2 = Arc::new(StringVector::from(vec![
-                Some("[0.0, 1.0]"),
-                Some("[0.0, 1.0]"),
-                Some("[0.0, 1.0]"),
-                None,
-            ])) as VectorRef;
-
-            let result = func
-                .eval(FunctionContext::default(), &[vec1.clone(), vec2.clone()])
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(result.get(2).is_null());
-            assert!(result.get(3).is_null());
-
-            let result = func
-                .eval(FunctionContext::default(), &[vec2, vec1])
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(result.get(2).is_null());
-            assert!(result.get(3).is_null());
-        }
-    }
-
-    #[test]
-    fn test_distance_binary_binary() {
-        let funcs = [
-            Box::new(CosDistanceFunction {}) as Box<dyn Function>,
-            Box::new(L2SqDistanceFunction {}) as Box<dyn Function>,
-            Box::new(DotProductFunction {}) as Box<dyn Function>,
-        ];
-
-        for func in funcs {
-            let vec1 = Arc::new(BinaryVector::from(vec![
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                Some(vec![0, 0, 128, 63, 0, 0, 0, 0]),
-                None,
-                Some(vec![0, 0, 128, 63, 0, 0, 0, 0]),
-            ])) as VectorRef;
-            let vec2 = Arc::new(BinaryVector::from(vec![
-                // [0.0, 1.0]
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                None,
-            ])) as VectorRef;
-
-            let result = func
-                .eval(FunctionContext::default(), &[vec1.clone(), vec2.clone()])
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(result.get(2).is_null());
-            assert!(result.get(3).is_null());
-
-            let result = func
-                .eval(FunctionContext::default(), &[vec2, vec1])
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(result.get(2).is_null());
-            assert!(result.get(3).is_null());
-        }
-    }
-
-    #[test]
-    fn test_distance_string_binary() {
-        let funcs = [
-            Box::new(CosDistanceFunction {}) as Box<dyn Function>,
-            Box::new(L2SqDistanceFunction {}) as Box<dyn Function>,
-            Box::new(DotProductFunction {}) as Box<dyn Function>,
-        ];
-
-        for func in funcs {
-            let vec1 = Arc::new(StringVector::from(vec![
-                Some("[0.0, 1.0]"),
-                Some("[1.0, 0.0]"),
-                None,
-                Some("[1.0, 0.0]"),
-            ])) as VectorRef;
-            let vec2 = Arc::new(BinaryVector::from(vec![
-                // [0.0, 1.0]
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                None,
-            ])) as VectorRef;
-
-            let result = func
-                .eval(FunctionContext::default(), &[vec1.clone(), vec2.clone()])
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(result.get(2).is_null());
-            assert!(result.get(3).is_null());
-
-            let result = func
-                .eval(FunctionContext::default(), &[vec2, vec1])
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(result.get(2).is_null());
-            assert!(result.get(3).is_null());
-        }
-    }
-
-    #[test]
-    fn test_distance_const_string() {
-        let funcs = [
-            Box::new(CosDistanceFunction {}) as Box<dyn Function>,
-            Box::new(L2SqDistanceFunction {}) as Box<dyn Function>,
-            Box::new(DotProductFunction {}) as Box<dyn Function>,
-        ];
-
-        for func in funcs {
-            let const_str = Arc::new(ConstantVector::new(
-                Arc::new(StringVector::from(vec!["[0.0, 1.0]"])),
-                4,
-            ));
-
-            let vec1 = Arc::new(StringVector::from(vec![
-                Some("[0.0, 1.0]"),
-                Some("[1.0, 0.0]"),
-                None,
-                Some("[1.0, 0.0]"),
-            ])) as VectorRef;
-            let vec2 = Arc::new(BinaryVector::from(vec![
-                // [0.0, 1.0]
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                Some(vec![0, 0, 0, 0, 0, 0, 128, 63]),
-                None,
-            ])) as VectorRef;
-
-            let result = func
-                .eval(
-                    FunctionContext::default(),
-                    &[const_str.clone(), vec1.clone()],
-                )
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(result.get(2).is_null());
-            assert!(!result.get(3).is_null());
-
-            let result = func
-                .eval(
-                    FunctionContext::default(),
-                    &[vec1.clone(), const_str.clone()],
-                )
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(result.get(2).is_null());
-            assert!(!result.get(3).is_null());
-
-            let result = func
-                .eval(
-                    FunctionContext::default(),
-                    &[const_str.clone(), vec2.clone()],
-                )
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(!result.get(2).is_null());
-            assert!(result.get(3).is_null());
-
-            let result = func
-                .eval(
-                    FunctionContext::default(),
-                    &[vec2.clone(), const_str.clone()],
-                )
-                .unwrap();
-
-            assert!(!result.get(0).is_null());
-            assert!(!result.get(1).is_null());
-            assert!(!result.get(2).is_null());
-            assert!(result.get(3).is_null());
-        }
-    }
-
-    #[test]
-    fn test_invalid_vector_length() {
-        let funcs = [
-            Box::new(CosDistanceFunction {}) as Box<dyn Function>,
-            Box::new(L2SqDistanceFunction {}) as Box<dyn Function>,
-            Box::new(DotProductFunction {}) as Box<dyn Function>,
-        ];
-
-        for func in funcs {
-            let vec1 = Arc::new(StringVector::from(vec!["[1.0]"])) as VectorRef;
-            let vec2 = Arc::new(StringVector::from(vec!["[1.0, 1.0]"])) as VectorRef;
-            let result = func.eval(FunctionContext::default(), &[vec1, vec2]);
-            assert!(result.is_err());
-
-            let vec1 = Arc::new(BinaryVector::from(vec![vec![0, 0, 128, 63]])) as VectorRef;
-            let vec2 =
-                Arc::new(BinaryVector::from(vec![vec![0, 0, 128, 63, 0, 0, 0, 64]])) as VectorRef;
-            let result = func.eval(FunctionContext::default(), &[vec1, vec2]);
-            assert!(result.is_err());
-        }
-    }
-
-    #[test]
-    fn test_parse_vector_from_string() {
-        let result = parse_f32_vector_from_string("[1.0, 2.0, 3.0]").unwrap();
-        assert_eq!(result, vec![1.0, 2.0, 3.0]);
-
-        let result = parse_f32_vector_from_string("[]").unwrap();
-        assert_eq!(result, Vec::<f32>::new());
-
-        let result = parse_f32_vector_from_string("[1.0, a, 3.0]");
-        assert!(result.is_err());
-    }
-
-    #[test]
-    fn test_binary_as_vector() {
-        let bytes = [0, 0, 128, 63];
-        let result = binary_as_vector(&bytes).unwrap();
-        assert_eq!(result.as_ref(), &[1.0]);
-
-        let invalid_bytes = [0, 0, 128];
-        let result = binary_as_vector(&invalid_bytes);
-        assert!(result.is_err());
-    }
-}
--- a/src/common/function/src/scalars/vector/distance/cos.rs
+++ b/src/common/function/src/scalars/vector/distance/cos.rs
@@ -1,87 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use nalgebra::DVectorView;
-
-/// Calculates the cos distance between two vectors.
-///
-/// **Note:** Must ensure that the length of the two vectors are the same.
-pub fn cos(lhs: &[f32], rhs: &[f32]) -> f32 {
-    let lhs_vec = DVectorView::from_slice(lhs, lhs.len());
-    let rhs_vec = DVectorView::from_slice(rhs, rhs.len());
-
-    let dot_product = lhs_vec.dot(&rhs_vec);
-    let lhs_norm = lhs_vec.norm();
-    let rhs_norm = rhs_vec.norm();
-    if dot_product.abs() < f32::EPSILON
-        || lhs_norm.abs() < f32::EPSILON
-        || rhs_norm.abs() < f32::EPSILON
-    {
-        return 1.0;
-    }
-
-    let cos_similar = dot_product / (lhs_norm * rhs_norm);
-    let res = 1.0 - cos_similar;
-    if res.abs() < f32::EPSILON {
-        0.0
-    } else {
-        res
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use approx::assert_relative_eq;
-
-    use super::*;
-
-    #[test]
-    fn test_cos_scalar() {
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 0.0, epsilon = 1e-2);
-
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 0.025, epsilon = 1e-2);
-
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 0.04, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 1.0, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 1.0, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 1.0, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 0.04, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 0.0, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(cos(&lhs, &rhs), 0.0, epsilon = 1e-2);
-    }
-}
--- a/src/common/function/src/scalars/vector/distance/dot.rs
+++ b/src/common/function/src/scalars/vector/distance/dot.rs
@@ -1,71 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use nalgebra::DVectorView;
-
-/// Calculates the dot product between two vectors.
-///
-/// **Note:** Must ensure that the length of the two vectors are the same.
-pub fn dot(lhs: &[f32], rhs: &[f32]) -> f32 {
-    let lhs = DVectorView::from_slice(lhs, lhs.len());
-    let rhs = DVectorView::from_slice(rhs, rhs.len());
-
-    lhs.dot(&rhs)
-}
-
-#[cfg(test)]
-mod tests {
-    use approx::assert_relative_eq;
-
-    use super::*;
-
-    #[test]
-    fn test_dot_scalar() {
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 14.0, epsilon = 1e-2);
-
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 32.0, epsilon = 1e-2);
-
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 50.0, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 0.0, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 0.0, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 0.0, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 50.0, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 122.0, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(dot(&lhs, &rhs), 194.0, epsilon = 1e-2);
-    }
-}
--- a/src/common/function/src/scalars/vector/distance/l2sq.rs
+++ b/src/common/function/src/scalars/vector/distance/l2sq.rs
@@ -1,71 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use nalgebra::DVectorView;
-
-/// Calculates the squared L2 distance between two vectors.
-///
-/// **Note:** Must ensure that the length of the two vectors are the same.
-pub fn l2sq(lhs: &[f32], rhs: &[f32]) -> f32 {
-    let lhs = DVectorView::from_slice(lhs, lhs.len());
-    let rhs = DVectorView::from_slice(rhs, rhs.len());
-
-    (lhs - rhs).norm_squared()
-}
-
-#[cfg(test)]
-mod tests {
-    use approx::assert_relative_eq;
-
-    use super::*;
-
-    #[test]
-    fn test_l2sq_scalar() {
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 0.0, epsilon = 1e-2);
-
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 27.0, epsilon = 1e-2);
-
-        let lhs = vec![1.0, 2.0, 3.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 108.0, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 14.0, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 77.0, epsilon = 1e-2);
-
-        let lhs = vec![0.0, 0.0, 0.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 194.0, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![1.0, 2.0, 3.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 108.0, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![4.0, 5.0, 6.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 27.0, epsilon = 1e-2);
-
-        let lhs = vec![7.0, 8.0, 9.0];
-        let rhs = vec![7.0, 8.0, 9.0];
-        assert_relative_eq!(l2sq(&lhs, &rhs), 0.0, epsilon = 1e-2);
-    }
-}
--- a/src/common/greptimedb-telemetry/src/lib.rs
+++ b/src/common/greptimedb-telemetry/src/lib.rs
@@ -199,7 +199,6 @@ 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`
@@ -325,7 +324,7 @@ mod tests {
            });
            let addr = ([127, 0, 0, 1], port).into();

-            let server = Server::try_bind(&addr).unwrap().serve(make_svc);
+            let server = Server::bind(&addr).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,7 +18,6 @@ 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
@@ -14,30 +14,27 @@

 use api::helper::ColumnDataTypeWrapper;
 use api::v1::add_column_location::LocationType;
-use api::v1::alter_table_expr::Kind;
-use api::v1::column_def::as_fulltext_option;
+use api::v1::alter_expr::Kind;
 use api::v1::{
-    column_def, AddColumnLocation as Location, AlterTableExpr, Analyzer, CreateTableExpr,
-    DropColumns, ModifyColumnTypes, RenameTable, SemanticType,
+    column_def, AddColumnLocation as Location, AlterExpr, ChangeColumnTypes, CreateTableExpr,
+    DropColumns, RenameTable, SemanticType,
 };
 use common_query::AddColumnLocation;
-use datatypes::schema::{ColumnSchema, FulltextOptions, RawSchema};
+use datatypes::schema::{ColumnSchema, RawSchema};
 use snafu::{ensure, OptionExt, ResultExt};
-use store_api::region_request::{SetRegionOption, UnsetRegionOption};
 use table::metadata::TableId;
-use table::requests::{AddColumnRequest, AlterKind, AlterTableRequest, ModifyColumnTypeRequest};
+use table::requests::{AddColumnRequest, AlterKind, AlterTableRequest, ChangeColumnTypeRequest};

 use crate::error::{
-    InvalidColumnDefSnafu, InvalidSetFulltextOptionRequestSnafu, InvalidSetTableOptionRequestSnafu,
-    InvalidUnsetTableOptionRequestSnafu, MissingFieldSnafu, MissingTimestampColumnSnafu, Result,
+    InvalidColumnDefSnafu, MissingFieldSnafu, MissingTimestampColumnSnafu, Result,
    UnknownLocationTypeSnafu,
 };

 const LOCATION_TYPE_FIRST: i32 = LocationType::First as i32;
 const LOCATION_TYPE_AFTER: i32 = LocationType::After as i32;

-/// Convert an [`AlterTableExpr`] to an [`AlterTableRequest`]
-pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<AlterTableRequest> {
+/// Convert an [`AlterExpr`] to an [`AlterTableRequest`]
+pub fn alter_expr_to_request(table_id: TableId, expr: AlterExpr) -> Result<AlterTableRequest> {
    let catalog_name = expr.catalog_name;
    let schema_name = expr.schema_name;
    let kind = expr.kind.context(MissingFieldSnafu { field: "kind" })?;
@@ -68,25 +65,25 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
                columns: add_column_requests,
            }
        }
-        Kind::ModifyColumnTypes(ModifyColumnTypes {
-            modify_column_types,
+        Kind::ChangeColumnTypes(ChangeColumnTypes {
+            change_column_types,
        }) => {
-            let modify_column_type_requests = modify_column_types
+            let change_column_type_requests = change_column_types
                .into_iter()
                .map(|cct| {
                    let target_type =
                        ColumnDataTypeWrapper::new(cct.target_type(), cct.target_type_extension)
                            .into();

-                    Ok(ModifyColumnTypeRequest {
+                    Ok(ChangeColumnTypeRequest {
                        column_name: cct.column_name,
                        target_type,
                    })
                })
                .collect::<Result<Vec<_>>>()?;

-            AlterKind::ModifyColumnTypes {
-                columns: modify_column_type_requests,
+            AlterKind::ChangeColumnTypes {
+                columns: change_column_type_requests,
            }
        }
        Kind::DropColumns(DropColumns { drop_columns }) => AlterKind::DropColumns {
@@ -95,37 +92,6 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
        Kind::RenameTable(RenameTable { new_table_name }) => {
            AlterKind::RenameTable { new_table_name }
        }
-        Kind::SetTableOptions(api::v1::SetTableOptions { table_options }) => {
-            AlterKind::SetTableOptions {
-                options: table_options
-                    .iter()
-                    .map(SetRegionOption::try_from)
-                    .collect::<std::result::Result<Vec<_>, _>>()
-                    .context(InvalidSetTableOptionRequestSnafu)?,
-            }
-        }
-        Kind::UnsetTableOptions(api::v1::UnsetTableOptions { keys }) => {
-            AlterKind::UnsetTableOptions {
-                keys: keys
-                    .iter()
-                    .map(|key| UnsetRegionOption::try_from(key.as_str()))
-                    .collect::<std::result::Result<Vec<_>, _>>()
-                    .context(InvalidUnsetTableOptionRequestSnafu)?,
-            }
-        }
-        Kind::SetColumnFulltext(c) => AlterKind::SetColumnFulltext {
-            column_name: c.column_name,
-            options: FulltextOptions {
-                enable: c.enable,
-                analyzer: as_fulltext_option(
-                    Analyzer::try_from(c.analyzer).context(InvalidSetFulltextOptionRequestSnafu)?,
-                ),
-                case_sensitive: c.case_sensitive,
-            },
-        },
-        Kind::UnsetColumnFulltext(c) => AlterKind::UnsetColumnFulltext {
-            column_name: c.column_name,
-        },
    };

    let request = AlterTableRequest {
@@ -194,7 +160,7 @@ fn parse_location(location: Option<Location>) -> Result<Option<AddColumnLocation
 #[cfg(test)]
 mod tests {
    use api::v1::{
-        AddColumn, AddColumns, ColumnDataType, ColumnDef, DropColumn, ModifyColumnType,
+        AddColumn, AddColumns, ChangeColumnType, ColumnDataType, ColumnDef, DropColumn,
        SemanticType,
    };
    use datatypes::prelude::ConcreteDataType;
@@ -203,7 +169,7 @@ mod tests {

    #[test]
    fn test_alter_expr_to_request() {
-        let expr = AlterTableExpr {
+        let expr = AlterExpr {
            catalog_name: String::default(),
            schema_name: String::default(),
            table_name: "monitor".to_string(),
@@ -244,7 +210,7 @@ mod tests {

    #[test]
    fn test_alter_expr_with_location_to_request() {
-        let expr = AlterTableExpr {
+        let expr = AlterExpr {
            catalog_name: String::default(),
            schema_name: String::default(),
            table_name: "monitor".to_string(),
@@ -320,14 +286,14 @@ mod tests {
    }

    #[test]
-    fn test_modify_column_type_expr() {
-        let expr = AlterTableExpr {
+    fn test_change_column_type_expr() {
+        let expr = AlterExpr {
            catalog_name: "test_catalog".to_string(),
            schema_name: "test_schema".to_string(),
            table_name: "monitor".to_string(),

-            kind: Some(Kind::ModifyColumnTypes(ModifyColumnTypes {
-                modify_column_types: vec![ModifyColumnType {
+            kind: Some(Kind::ChangeColumnTypes(ChangeColumnTypes {
+                change_column_types: vec![ChangeColumnType {
                    column_name: "mem_usage".to_string(),
                    target_type: ColumnDataType::String as i32,
                    target_type_extension: None,
@@ -340,22 +306,22 @@ mod tests {
        assert_eq!(alter_request.schema_name, "test_schema");
        assert_eq!("monitor".to_string(), alter_request.table_name);

-        let mut modify_column_types = match alter_request.alter_kind {
-            AlterKind::ModifyColumnTypes { columns } => columns,
+        let mut change_column_types = match alter_request.alter_kind {
+            AlterKind::ChangeColumnTypes { columns } => columns,
            _ => unreachable!(),
        };

-        let modify_column_type = modify_column_types.pop().unwrap();
-        assert_eq!("mem_usage", modify_column_type.column_name);
+        let change_column_type = change_column_types.pop().unwrap();
+        assert_eq!("mem_usage", change_column_type.column_name);
        assert_eq!(
            ConcreteDataType::string_datatype(),
-            modify_column_type.target_type
+            change_column_type.target_type
        );
    }

    #[test]
    fn test_drop_column_expr() {
-        let expr = AlterTableExpr {
+        let expr = AlterExpr {
            catalog_name: "test_catalog".to_string(),
            schema_name: "test_schema".to_string(),
            table_name: "monitor".to_string(),
--- a/src/common/grpc-expr/src/error.rs
+++ b/src/common/grpc-expr/src/error.rs
@@ -19,7 +19,6 @@ 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))]
@@ -119,26 +118,6 @@ pub enum Error {
        #[snafu(implicit)]
        location: Location,
    },
-
-    #[snafu(display("Invalid set table option request"))]
-    InvalidSetTableOptionRequest {
-        #[snafu(source)]
-        error: MetadataError,
-    },
-
-    #[snafu(display("Invalid unset table option request"))]
-    InvalidUnsetTableOptionRequest {
-        #[snafu(source)]
-        error: MetadataError,
-    },
-
-    #[snafu(display("Invalid set fulltext option request"))]
-    InvalidSetFulltextOptionRequest {
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: prost::DecodeError,
-    },
 }

 pub type Result<T> = std::result::Result<T, Error>;
@@ -162,9 +141,6 @@ impl ErrorExt for Error {
            Error::UnknownColumnDataType { .. } | Error::InvalidFulltextColumnType { .. } => {
                StatusCode::InvalidArguments
            }
-            Error::InvalidSetTableOptionRequest { .. }
-            | Error::InvalidUnsetTableOptionRequest { .. }
-            | Error::InvalidSetFulltextOptionRequest { .. } => StatusCode::InvalidArguments,
        }
    }

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

 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, JsonTypeExtension, SemanticType,
+    ColumnOptions, ColumnSchema, CreateTableExpr, SemanticType,
 };
 use datatypes::schema::Schema;
 use snafu::{ensure, OptionExt, ResultExt};
@@ -26,9 +25,8 @@ use table::metadata::TableId;
 use table::table_reference::TableReference;

 use crate::error::{
-    self, DuplicatedColumnNameSnafu, DuplicatedTimestampColumnSnafu,
-    InvalidFulltextColumnTypeSnafu, MissingTimestampColumnSnafu, Result,
-    UnknownColumnDataTypeSnafu,
+    DuplicatedColumnNameSnafu, DuplicatedTimestampColumnSnafu, InvalidFulltextColumnTypeSnafu,
+    MissingTimestampColumnSnafu, Result, UnknownColumnDataTypeSnafu,
 };
 pub struct ColumnExpr<'a> {
    pub column_name: &'a str,
@@ -74,28 +72,6 @@ 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<'_>,
@@ -148,7 +124,8 @@ pub fn build_create_table_expr(
            _ => {}
        }

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

        ensure!(
            !contains_fulltext(options) || column_type == ColumnDataType::String,
--- a/src/common/grpc/src/select.rs
+++ b/src/common/grpc/src/select.rs
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use api::helper::{convert_month_day_nano_to_pb, convert_to_pb_decimal128};
+use api::helper::{convert_i128_to_interval, convert_to_pb_decimal128};
 use api::v1::column::Values;
 use common_base::BitVec;
 use datatypes::types::{IntervalType, TimeType, TimestampType, WrapperType};
@@ -211,19 +211,13 @@ pub fn values(arrays: &[VectorRef]) -> Result<Values> {
            ConcreteDataType::Interval(IntervalType::MonthDayNano(_)),
            IntervalMonthDayNanoVector,
            interval_month_day_nano_values,
-            |x| { convert_month_day_nano_to_pb(x) }
+            |x| { convert_i128_to_interval(x.into_native()) }
        ),
        (
            ConcreteDataType::Decimal128(_),
            Decimal128Vector,
            decimal128_values,
            |x| { convert_to_pb_decimal128(x) }
-        ),
-        (
-            ConcreteDataType::Vector(_),
-            BinaryVector,
-            binary_values,
-            |x| { x.into() }
        )
    )
 }
--- a/src/common/macro/src/lib.rs
+++ b/src/common/macro/src/lib.rs
@@ -35,9 +35,7 @@ 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]
@@ -46,10 +44,9 @@ 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
@@ -39,7 +39,7 @@ derive_builder.workspace = true
 etcd-client.workspace = true
 futures.workspace = true
 futures-util.workspace = true
-hex.workspace = true
+hex = { version = "0.4" }
 humantime-serde.workspace = true
 itertools.workspace = true
 lazy_static.workspace = true
@@ -60,7 +60,7 @@ table.workspace = true
 tokio.workspace = true
 tokio-postgres = { workspace = true, optional = true }
 tonic.workspace = true
-typetag.workspace = true
+typetag = "0.2"

 [dev-dependencies]
 chrono.workspace = true
--- a/src/common/meta/src/cluster.rs
+++ b/src/common/meta/src/cluster.rs
@@ -55,7 +55,6 @@ 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
@@ -77,22 +77,18 @@ pub struct RegionStat {
    pub rcus: i64,
    /// The write capacity units during this period
    pub wcus: i64,
-    /// Approximate disk bytes of this region, including sst, index, manifest and wal
-    pub approximate_bytes: u64,
+    /// Approximate bytes of this region
+    pub approximate_bytes: i64,
    /// 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 data files in bytes.
+    /// The size of the SST files in bytes.
    pub sst_size: u64,
-    /// The size of the SST index files in bytes.
-    pub index_size: u64,
 }

 impl Stat {
@@ -182,14 +178,12 @@ 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 as u64,
+            approximate_bytes: value.approximate_bytes,
            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.rs
+++ b/src/common/meta/src/ddl.rs
@@ -32,7 +32,6 @@ use crate::rpc::ddl::{SubmitDdlTaskRequest, SubmitDdlTaskResponse};
 use crate::rpc::procedure::{MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse};
 use crate::{ClusterId, DatanodeId};

-pub mod alter_database;
 pub mod alter_logical_tables;
 pub mod alter_table;
 pub mod create_database;
--- a/src/common/meta/src/ddl/alter_database.rs
+++ b/src/common/meta/src/ddl/alter_database.rs
@@ -1,248 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use async_trait::async_trait;
-use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
-use common_procedure::{Context as ProcedureContext, LockKey, Procedure, Status};
-use common_telemetry::tracing::info;
-use serde::{Deserialize, Serialize};
-use snafu::{ensure, ResultExt};
-use strum::AsRefStr;
-
-use super::utils::handle_retry_error;
-use crate::cache_invalidator::Context;
-use crate::ddl::DdlContext;
-use crate::error::{Result, SchemaNotFoundSnafu};
-use crate::instruction::CacheIdent;
-use crate::key::schema_name::{SchemaName, SchemaNameKey, SchemaNameValue};
-use crate::key::DeserializedValueWithBytes;
-use crate::lock_key::{CatalogLock, SchemaLock};
-use crate::rpc::ddl::UnsetDatabaseOption::{self};
-use crate::rpc::ddl::{AlterDatabaseKind, AlterDatabaseTask, SetDatabaseOption};
-use crate::ClusterId;
-
-pub struct AlterDatabaseProcedure {
-    pub context: DdlContext,
-    pub data: AlterDatabaseData,
-}
-
-fn build_new_schema_value(
-    mut value: SchemaNameValue,
-    alter_kind: &AlterDatabaseKind,
-) -> Result<SchemaNameValue> {
-    match alter_kind {
-        AlterDatabaseKind::SetDatabaseOptions(options) => {
-            for option in options.0.iter() {
-                match option {
-                    SetDatabaseOption::Ttl(ttl) => {
-                        if ttl.is_zero() {
-                            value.ttl = None;
-                        } else {
-                            value.ttl = Some(*ttl);
-                        }
-                    }
-                }
-            }
-        }
-        AlterDatabaseKind::UnsetDatabaseOptions(keys) => {
-            for key in keys.0.iter() {
-                match key {
-                    UnsetDatabaseOption::Ttl => value.ttl = None,
-                }
-            }
-        }
-    }
-    Ok(value)
-}
-
-impl AlterDatabaseProcedure {
-    pub const TYPE_NAME: &'static str = "metasrv-procedure::AlterDatabase";
-
-    pub fn new(
-        cluster_id: ClusterId,
-        task: AlterDatabaseTask,
-        context: DdlContext,
-    ) -> Result<Self> {
-        Ok(Self {
-            context,
-            data: AlterDatabaseData::new(task, cluster_id)?,
-        })
-    }
-
-    pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
-        let data = serde_json::from_str(json).context(FromJsonSnafu)?;
-
-        Ok(Self { context, data })
-    }
-
-    pub async fn on_prepare(&mut self) -> Result<Status> {
-        let value = self
-            .context
-            .table_metadata_manager
-            .schema_manager()
-            .get(SchemaNameKey::new(self.data.catalog(), self.data.schema()))
-            .await?;
-
-        ensure!(
-            value.is_some(),
-            SchemaNotFoundSnafu {
-                table_schema: self.data.schema(),
-            }
-        );
-
-        self.data.schema_value = value;
-        self.data.state = AlterDatabaseState::UpdateMetadata;
-
-        Ok(Status::executing(true))
-    }
-
-    pub async fn on_update_metadata(&mut self) -> Result<Status> {
-        let schema_name = SchemaNameKey::new(self.data.catalog(), self.data.schema());
-
-        // Safety: schema_value is not None.
-        let current_schema_value = self.data.schema_value.as_ref().unwrap();
-
-        let new_schema_value = build_new_schema_value(
-            current_schema_value.get_inner_ref().clone(),
-            &self.data.kind,
-        )?;
-
-        self.context
-            .table_metadata_manager
-            .schema_manager()
-            .update(schema_name, current_schema_value, &new_schema_value)
-            .await?;
-
-        info!("Updated database metadata for schema {schema_name}");
-        self.data.state = AlterDatabaseState::InvalidateSchemaCache;
-        Ok(Status::executing(true))
-    }
-
-    pub async fn on_invalidate_schema_cache(&mut self) -> Result<Status> {
-        let cache_invalidator = &self.context.cache_invalidator;
-        cache_invalidator
-            .invalidate(
-                &Context::default(),
-                &[CacheIdent::SchemaName(SchemaName {
-                    catalog_name: self.data.catalog().to_string(),
-                    schema_name: self.data.schema().to_string(),
-                })],
-            )
-            .await?;
-
-        Ok(Status::done())
-    }
-}
-
-#[async_trait]
-impl Procedure for AlterDatabaseProcedure {
-    fn type_name(&self) -> &str {
-        Self::TYPE_NAME
-    }
-
-    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
-        match self.data.state {
-            AlterDatabaseState::Prepare => self.on_prepare().await,
-            AlterDatabaseState::UpdateMetadata => self.on_update_metadata().await,
-            AlterDatabaseState::InvalidateSchemaCache => self.on_invalidate_schema_cache().await,
-        }
-        .map_err(handle_retry_error)
-    }
-
-    fn dump(&self) -> ProcedureResult<String> {
-        serde_json::to_string(&self.data).context(ToJsonSnafu)
-    }
-
-    fn lock_key(&self) -> LockKey {
-        let catalog = self.data.catalog();
-        let schema = self.data.schema();
-
-        let lock_key = vec![
-            CatalogLock::Read(catalog).into(),
-            SchemaLock::write(catalog, schema).into(),
-        ];
-
-        LockKey::new(lock_key)
-    }
-}
-
-#[derive(Debug, Serialize, Deserialize, AsRefStr)]
-enum AlterDatabaseState {
-    Prepare,
-    UpdateMetadata,
-    InvalidateSchemaCache,
-}
-
-/// The data of alter database procedure.
-#[derive(Debug, Serialize, Deserialize)]
-pub struct AlterDatabaseData {
-    cluster_id: ClusterId,
-    state: AlterDatabaseState,
-    kind: AlterDatabaseKind,
-    catalog_name: String,
-    schema_name: String,
-    schema_value: Option<DeserializedValueWithBytes<SchemaNameValue>>,
-}
-
-impl AlterDatabaseData {
-    pub fn new(task: AlterDatabaseTask, cluster_id: ClusterId) -> Result<Self> {
-        Ok(Self {
-            cluster_id,
-            state: AlterDatabaseState::Prepare,
-            kind: AlterDatabaseKind::try_from(task.alter_expr.kind.unwrap())?,
-            catalog_name: task.alter_expr.catalog_name,
-            schema_name: task.alter_expr.schema_name,
-            schema_value: None,
-        })
-    }
-
-    pub fn catalog(&self) -> &str {
-        &self.catalog_name
-    }
-
-    pub fn schema(&self) -> &str {
-        &self.schema_name
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::time::Duration;
-
-    use crate::ddl::alter_database::build_new_schema_value;
-    use crate::key::schema_name::SchemaNameValue;
-    use crate::rpc::ddl::{
-        AlterDatabaseKind, SetDatabaseOption, SetDatabaseOptions, UnsetDatabaseOption,
-        UnsetDatabaseOptions,
-    };
-
-    #[test]
-    fn test_build_new_schema_value() {
-        let set_ttl = AlterDatabaseKind::SetDatabaseOptions(SetDatabaseOptions(vec![
-            SetDatabaseOption::Ttl(Duration::from_secs(10)),
-        ]));
-        let current_schema_value = SchemaNameValue::default();
-        let new_schema_value =
-            build_new_schema_value(current_schema_value.clone(), &set_ttl).unwrap();
-        assert_eq!(new_schema_value.ttl, Some(Duration::from_secs(10)));
-
-        let unset_ttl_alter_kind =
-            AlterDatabaseKind::UnsetDatabaseOptions(UnsetDatabaseOptions(vec![
-                UnsetDatabaseOption::Ttl,
-            ]));
-        let new_schema_value =
-            build_new_schema_value(current_schema_value, &unset_ttl_alter_kind).unwrap();
-        assert_eq!(new_schema_value.ttl, None);
-    }
-}
--- a/src/common/meta/src/ddl/alter_logical_tables/check.rs
+++ b/src/common/meta/src/ddl/alter_logical_tables/check.rs
@@ -14,7 +14,7 @@

 use std::collections::HashSet;

-use api::v1::alter_table_expr::Kind;
+use api::v1::alter_expr::Kind;
 use snafu::{ensure, OptionExt};

 use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
--- a/src/common/meta/src/ddl/alter_logical_tables/region_request.rs
+++ b/src/common/meta/src/ddl/alter_logical_tables/region_request.rs
@@ -13,7 +13,7 @@
 // limitations under the License.

 use api::v1;
-use api::v1::alter_table_expr::Kind;
+use api::v1::alter_expr::Kind;
 use api::v1::region::{
    alter_request, region_request, AddColumn, AddColumns, AlterRequest, AlterRequests,
    RegionColumnDef, RegionRequest, RegionRequestHeader,
--- 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, and we don't need to touch per-region settings.
+        // Updates physical table's metadata
        self.context
            .table_metadata_manager
-            .update_table_info(physical_table_info, None, new_raw_table_info)
+            .update_table_info(physical_table_info, new_raw_table_info)
            .await?;

        Ok(())
--- a/src/common/meta/src/ddl/alter_table.rs
+++ b/src/common/meta/src/ddl/alter_table.rs
@@ -19,7 +19,7 @@ mod update_metadata;

 use std::vec;

-use api::v1::alter_table_expr::Kind;
+use api::v1::alter_expr::Kind;
 use api::v1::RenameTable;
 use async_trait::async_trait;
 use common_error::ext::ErrorExt;
@@ -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, RegionDistribution};
+use crate::key::DeserializedValueWithBytes;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
 use crate::rpc::ddl::AlterTableTask;
-use crate::rpc::router::{find_leader_regions, find_leaders, region_distribution};
+use crate::rpc::router::{find_leader_regions, find_leaders};
 use crate::{metrics, ClusterId};

 /// The alter table procedure
@@ -101,9 +101,6 @@ 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());

@@ -164,14 +161,8 @@ impl AlterTableProcedure {
            self.on_update_metadata_for_rename(new_table_name.to_string(), table_info_value)
                .await?;
        } else {
-            // 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?;
+            self.on_update_metadata_for_alter(new_info.into(), table_info_value)
+                .await?;
        }

        info!("Updated table metadata for table {table_ref}, table_id: {table_id}");
@@ -280,8 +271,6 @@ 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 {
@@ -292,7 +281,6 @@ impl AlterTableData {
            table_id,
            cluster_id,
            table_info_value: None,
-            region_distribution: None,
        }
    }

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