feat: cache logical region's metadata

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

.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

.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 \

.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

.github/workflows/nightly-ci.yml

@@ -92,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:

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.9.5"
+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 = "75c5fb569183bb3d0fa1023df9c2214df722b9b1" }
-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

Makefile

@@ -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

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
 

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`**. |
@@ -116,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`. |
@@ -126,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. |
@@ -412,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`**. |
@@ -439,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` |
@@ -449,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. |

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`**.
@@ -416,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"
@@ -459,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
@@ -646,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"

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"
 

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"

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"

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`**.
@@ -454,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"
@@ -497,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
@@ -690,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"

docker/buildx/centos/Dockerfile

@@ -49,7 +49,6 @@ RUN yum install -y epel-release \
 
 WORKDIR /greptime
 COPY --from=builder /out/target/${OUTPUT_DIR}/greptime /greptime/bin/
-COPY --from=builder /out/target/${OUTPUT_DIR}/build/tikv-jemalloc-sys-*/out/build/bin/jeprof /greptime/bin/
 ENV PATH /greptime/bin/:$PATH
 
 ENTRYPOINT ["greptime"]

docker/buildx/ubuntu/Dockerfile

@@ -57,9 +57,6 @@ RUN python3 -m pip install -r /etc/greptime/requirements.txt
 
 WORKDIR /greptime
 COPY --from=builder /out/target/${OUTPUT_DIR}/greptime /greptime/bin/
-COPY --from=builder /out/target/${OUTPUT_DIR}/build/tikv-jemalloc-sys-*/out/build/bin/jeprof /greptime/bin/
-RUN chmod +x /greptime/bin/jeprof
-
 ENV PATH /greptime/bin/:$PATH
 
 ENTRYPOINT ["greptime"]

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).

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).

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

docs/how-to/how-to-profile-memory.md

@@ -4,18 +4,6 @@ This crate provides an easy approach to dump memory profiling info.
 
 ## Prerequisites
 ### jemalloc
-jeprof is already compiled in the target directory of GreptimeDB. You can find the binary and use it.
-```
-# find jeprof binary
-find . -name 'jeprof'
-
-# add executable permission
-chmod +x <path_to_jeprof>
-```
-The path is usually under `./target/${PROFILE}/build/tikv-jemalloc-sys-${HASH}/out/build/bin/jeprof`.
-
-The default version of jemalloc installed from the package manager may not have the `--collapsed` option.
-You may need to check the whether the `jeprof` version is >= `5.3.0` if you want to install it from the package manager.
 ```bash
 # for macOS
 brew install jemalloc
@@ -30,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:

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": ""
 }

rust-toolchain.toml

@@ -1,2 +1,3 @@
 [toolchain]
-channel = "nightly-2024-10-19"
+channel = "nightly-2024-06-06"
+

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!(),
     });
@@ -538,12 +507,14 @@ fn ddl_request_type(request: &DdlRequest) -> &'static str {
     }
 }
 
-/// 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,
     }
 }
 
@@ -591,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
@@ -690,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()
                     }),
                 ))
             }
@@ -700,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(_)
@@ -836,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,
@@ -866,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(_)
@@ -891,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)
 }
@@ -973,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))),
@@ -1074,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!(),
         },
@@ -1089,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,
@@ -1178,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]
@@ -1269,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]
@@ -1369,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());
@@ -1547,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]
@@ -1631,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);
 
@@ -1646,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);
 
@@ -1677,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);
     }

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]

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(),
-        }
+        };
     }
 }
 

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,

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

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()),
         ];

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";

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());

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)

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;
 

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"

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
@@ -79,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"

src/cmd/src/cli/repl.rs

@@ -174,7 +174,7 @@ impl Repl {
 
             let plan = query_engine
                 .planner()
-                .plan(&stmt, query_ctx.clone())
+                .plan(stmt, query_ctx.clone())
                 .await
                 .context(PlanStatementSnafu)?;
 

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)?;
 

src/cmd/src/frontend.rs

@@ -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)?;
 
@@ -343,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);
@@ -475,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();
 

src/cmd/src/lib.rs

@@ -84,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) {

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)?;
 

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)]

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();
 

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()),

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

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,

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]);
-    }
-}

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`]

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";

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,8 +28,6 @@ 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
@@ -37,7 +35,6 @@ 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
@@ -46,7 +43,6 @@ sql.workspace = true
 statrs = "0.16"
 store-api.workspace = true
 table.workspace = true
-wkt = { version = "0.11", optional = true }
 
 [dev-dependencies]
 ron = "0.7"

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);
     }
 }

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!(),
+            }
+        }
+    }
 }

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!(),
+            }
+        }
+    }
 }

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));
     }
 }

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;

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);
-    }
-}

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;

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())
-    }
-}

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())
-    }
-}

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!(),
-    }
-}

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)
-}

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));
     }
 }

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());
     }
 }

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());
-                }
-            }
-        }
-    }
-}

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();
             });

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]

src/common/grpc-expr/src/alter.rs

@@ -15,20 +15,17 @@
 use api::helper::ColumnDataTypeWrapper;
 use api::v1::add_column_location::LocationType;
 use api::v1::alter_expr::Kind;
-use api::v1::column_def::as_fulltext_option;
 use api::v1::{
-    column_def, AddColumnLocation as Location, AlterExpr, Analyzer, ChangeColumnTypes,
-    CreateTableExpr, DropColumns, 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::ChangeOption;
 use table::metadata::TableId;
 use table::requests::{AddColumnRequest, AlterKind, AlterTableRequest, ChangeColumnTypeRequest};
 
 use crate::error::{
-    InvalidChangeFulltextOptionRequestSnafu, InvalidChangeTableOptionRequestSnafu,
     InvalidColumnDefSnafu, MissingFieldSnafu, MissingTimestampColumnSnafu, Result,
     UnknownLocationTypeSnafu,
 };
@@ -95,26 +92,6 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterExpr) -> Result<Alter
         Kind::RenameTable(RenameTable { new_table_name }) => {
             AlterKind::RenameTable { new_table_name }
         }
-        Kind::ChangeTableOptions(api::v1::ChangeTableOptions {
-            change_table_options,
-        }) => AlterKind::ChangeTableOptions {
-            options: change_table_options
-                .iter()
-                .map(ChangeOption::try_from)
-                .collect::<std::result::Result<Vec<_>, _>>()
-                .context(InvalidChangeTableOptionRequestSnafu)?,
-        },
-        Kind::ChangeColumnFulltext(c) => AlterKind::ChangeColumnFulltext {
-            column_name: c.column_name,
-            options: FulltextOptions {
-                enable: c.enable,
-                analyzer: as_fulltext_option(
-                    Analyzer::try_from(c.analyzer)
-                        .context(InvalidChangeFulltextOptionRequestSnafu)?,
-                ),
-                case_sensitive: c.case_sensitive,
-            },
-        },
     };
 
     let request = AlterTableRequest {

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,20 +118,6 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-
-    #[snafu(display("Invalid change table option request"))]
-    InvalidChangeTableOptionRequest {
-        #[snafu(source)]
-        error: MetadataError,
-    },
-
-    #[snafu(display("Invalid change fulltext option request"))]
-    InvalidChangeFulltextOptionRequest {
-        #[snafu(implicit)]
-        location: Location,
-        #[snafu(source)]
-        error: prost::DecodeError,
-    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -156,8 +141,6 @@ impl ErrorExt for Error {
             Error::UnknownColumnDataType { .. } | Error::InvalidFulltextColumnType { .. } => {
                 StatusCode::InvalidArguments
             }
-            Error::InvalidChangeTableOptionRequest { .. }
-            | Error::InvalidChangeFulltextOptionRequest { .. } => StatusCode::InvalidArguments,
         }
     }
 

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,

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() }
         )
     )
 }

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:

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

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)]

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,
         }
     }
 }

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(())

src/common/meta/src/ddl/alter_table.rs

@@ -43,10 +43,10 @@ use crate::ddl::DdlContext;
 use crate::error::{Error, Result};
 use crate::instruction::CacheIdent;
 use crate::key::table_info::TableInfoValue;
-use crate::key::{DeserializedValueWithBytes, 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,
         }
     }
 

src/common/meta/src/ddl/alter_table/region_request.rs

@@ -106,10 +106,6 @@ fn create_proto_alter_kind(
             })))
         }
         Kind::RenameTable(_) => Ok(None),
-        Kind::ChangeTableOptions(v) => Ok(Some(alter_request::Kind::ChangeTableOptions(v.clone()))),
-        Kind::ChangeColumnFulltext(v) => {
-            Ok(Some(alter_request::Kind::ChangeColumnFulltext(v.clone())))
-        }
     }
 }
 

src/common/meta/src/ddl/alter_table/update_metadata.rs

@@ -20,7 +20,7 @@ use table::requests::AlterKind;
 use crate::ddl::alter_table::AlterTableProcedure;
 use crate::error::{self, Result};
 use crate::key::table_info::TableInfoValue;
-use crate::key::{DeserializedValueWithBytes, RegionDistribution};
+use crate::key::DeserializedValueWithBytes;
 
 impl AlterTableProcedure {
     /// Builds new_meta
@@ -51,10 +51,7 @@ impl AlterTableProcedure {
             AlterKind::RenameTable { new_table_name } => {
                 new_info.name = new_table_name.to_string();
             }
-            AlterKind::DropColumns { .. }
-            | AlterKind::ChangeColumnTypes { .. }
-            | AlterKind::ChangeTableOptions { .. }
-            | AlterKind::ChangeColumnFulltext { .. } => {}
+            AlterKind::DropColumns { .. } | AlterKind::ChangeColumnTypes { .. } => {}
         }
 
         Ok(new_info)
@@ -78,16 +75,11 @@ impl AlterTableProcedure {
     pub(crate) async fn on_update_metadata_for_alter(
         &self,
         new_table_info: RawTableInfo,
-        region_distribution: RegionDistribution,
         current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
     ) -> Result<()> {
         let table_metadata_manager = &self.context.table_metadata_manager;
         table_metadata_manager
-            .update_table_info(
-                current_table_info_value,
-                Some(region_distribution),
-                new_table_info,
-            )
+            .update_table_info(current_table_info_value, new_table_info)
             .await?;
 
         Ok(())

src/common/meta/src/ddl/create_logical_tables/update_metadata.rs

@@ -58,10 +58,10 @@ impl CreateLogicalTablesProcedure {
             &new_table_info.name,
         );
 
-        // Update physical table's metadata and we don't need to touch per-region settings.
+        // Update physical table's metadata
         self.context
             .table_metadata_manager
-            .update_table_info(&physical_table_info, None, new_table_info)
+            .update_table_info(&physical_table_info, new_table_info)
             .await?;
 
         // Invalid physical table cache

src/common/meta/src/ddl/test_util/datanode_handler.rs

@@ -29,10 +29,7 @@ use crate::test_util::MockDatanodeHandler;
 #[async_trait::async_trait]
 impl MockDatanodeHandler for () {
     async fn handle(&self, _peer: &Peer, _request: RegionRequest) -> Result<RegionResponse> {
-        Ok(RegionResponse {
-            affected_rows: 0,
-            extensions: Default::default(),
-        })
+        unreachable!()
     }
 
     async fn handle_query(

src/common/meta/src/ddl/tests/alter_table.rs

@@ -19,14 +19,13 @@ use std::sync::Arc;
 use api::v1::alter_expr::Kind;
 use api::v1::region::{region_request, RegionRequest};
 use api::v1::{
-    AddColumn, AddColumns, AlterExpr, ChangeTableOption, ChangeTableOptions, ColumnDataType,
-    ColumnDef as PbColumnDef, DropColumn, DropColumns, SemanticType,
+    AddColumn, AddColumns, AlterExpr, ColumnDataType, ColumnDef as PbColumnDef, DropColumn,
+    DropColumns, SemanticType,
 };
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use store_api::storage::RegionId;
-use table::requests::TTL_KEY;
 use tokio::sync::mpsc::{self};
 
 use crate::ddl::alter_table::AlterTableProcedure;
@@ -35,7 +34,6 @@ use crate::ddl::test_util::create_table::test_create_table_task;
 use crate::ddl::test_util::datanode_handler::{
     DatanodeWatcher, RequestOutdatedErrorDatanodeHandler,
 };
-use crate::key::datanode_table::DatanodeTableKey;
 use crate::key::table_name::TableNameKey;
 use crate::key::table_route::TableRouteValue;
 use crate::peer::Peer;
@@ -295,21 +293,12 @@ async fn test_on_update_metadata_add_columns() {
     let table_name = "foo";
     let table_id = 1024;
     let task = test_create_table_task(table_name, table_id);
-
-    let region_id = RegionId::new(table_id, 0);
-    let mock_table_routes = vec![RegionRoute {
-        region: Region::new_test(region_id),
-        leader_peer: Some(Peer::default()),
-        follower_peers: vec![],
-        leader_state: None,
-        leader_down_since: None,
-    }];
     // Puts a value to table name key.
     ddl_context
         .table_metadata_manager
         .create_table_metadata(
             task.table_info.clone(),
-            TableRouteValue::physical(mock_table_routes),
+            TableRouteValue::physical(vec![]),
             HashMap::new(),
         )
         .await
@@ -337,7 +326,6 @@ async fn test_on_update_metadata_add_columns() {
     let mut procedure =
         AlterTableProcedure::new(cluster_id, table_id, task, ddl_context.clone()).unwrap();
     procedure.on_prepare().await.unwrap();
-    procedure.submit_alter_region_requests().await.unwrap();
     procedure.on_update_metadata().await.unwrap();
 
     let table_info = ddl_context
@@ -355,76 +343,3 @@ async fn test_on_update_metadata_add_columns() {
         table_info.meta.next_column_id
     );
 }
-
-#[tokio::test]
-async fn test_on_update_table_options() {
-    let node_manager = Arc::new(MockDatanodeManager::new(()));
-    let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
-    let table_name = "foo";
-    let table_id = 1024;
-    let task = test_create_table_task(table_name, table_id);
-
-    let region_id = RegionId::new(table_id, 0);
-    let mock_table_routes = vec![RegionRoute {
-        region: Region::new_test(region_id),
-        leader_peer: Some(Peer::default()),
-        follower_peers: vec![],
-        leader_state: None,
-        leader_down_since: None,
-    }];
-    // Puts a value to table name key.
-    ddl_context
-        .table_metadata_manager
-        .create_table_metadata(
-            task.table_info.clone(),
-            TableRouteValue::physical(mock_table_routes),
-            HashMap::new(),
-        )
-        .await
-        .unwrap();
-
-    let task = AlterTableTask {
-        alter_table: AlterExpr {
-            catalog_name: DEFAULT_CATALOG_NAME.to_string(),
-            schema_name: DEFAULT_SCHEMA_NAME.to_string(),
-            table_name: table_name.to_string(),
-            kind: Some(Kind::ChangeTableOptions(ChangeTableOptions {
-                change_table_options: vec![ChangeTableOption {
-                    key: TTL_KEY.to_string(),
-                    value: "1d".to_string(),
-                }],
-            })),
-        },
-    };
-    let mut procedure =
-        AlterTableProcedure::new(cluster_id, table_id, task, ddl_context.clone()).unwrap();
-    procedure.on_prepare().await.unwrap();
-    procedure.submit_alter_region_requests().await.unwrap();
-    procedure.on_update_metadata().await.unwrap();
-
-    let table_info = ddl_context
-        .table_metadata_manager
-        .table_info_manager()
-        .get(table_id)
-        .await
-        .unwrap()
-        .unwrap()
-        .into_inner()
-        .table_info;
-
-    let datanode_key = DatanodeTableKey::new(0, table_id);
-    let region_info = ddl_context
-        .table_metadata_manager
-        .datanode_table_manager()
-        .get(&datanode_key)
-        .await
-        .unwrap()
-        .unwrap()
-        .region_info;
-
-    assert_eq!(
-        region_info.region_options,
-        HashMap::from(&table_info.meta.options)
-    );
-}

src/common/meta/src/error.rs

@@ -652,18 +652,6 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-
-    #[snafu(display(
-        "Datanode table info not found, table id: {}, datanode id: {}",
-        table_id,
-        datanode_id
-    ))]
-    DatanodeTableInfoNotFound {
-        datanode_id: DatanodeId,
-        table_id: TableId,
-        #[snafu(implicit)]
-        location: Location,
-    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -764,7 +752,6 @@ impl ErrorExt for Error {
             PostgresExecution { .. } => StatusCode::Internal,
             #[cfg(feature = "pg_kvbackend")]
             ConnectPostgres { .. } => StatusCode::Internal,
-            Error::DatanodeTableInfoNotFound { .. } => StatusCode::Internal,
         }
     }
 

src/common/meta/src/key.rs

@@ -91,7 +91,6 @@ pub mod catalog_name;
 pub mod datanode_table;
 pub mod flow;
 pub mod node_address;
-mod schema_metadata_manager;
 pub mod schema_name;
 pub mod table_info;
 pub mod table_name;
@@ -117,7 +116,6 @@ use flow::flow_route::FlowRouteValue;
 use flow::table_flow::TableFlowValue;
 use lazy_static::lazy_static;
 use regex::Regex;
-pub use schema_metadata_manager::{SchemaMetadataManager, SchemaMetadataManagerRef};
 use serde::de::DeserializeOwned;
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, OptionExt, ResultExt};
@@ -135,6 +133,7 @@ use self::flow::flow_name::FlowNameValue;
 use self::schema_name::{SchemaManager, SchemaNameKey, SchemaNameValue};
 use self::table_route::{TableRouteManager, TableRouteValue};
 use self::tombstone::TombstoneManager;
+use crate::ddl::utils::region_storage_path;
 use crate::error::{self, Result, SerdeJsonSnafu};
 use crate::key::node_address::NodeAddressValue;
 use crate::key::table_route::TableRouteKey;
@@ -594,6 +593,8 @@ impl TableMetadataManager {
         table_info.meta.region_numbers = region_numbers;
         let table_id = table_info.ident.table_id;
         let engine = table_info.meta.engine.clone();
+        let region_storage_path =
+            region_storage_path(&table_info.catalog_name, &table_info.schema_name);
 
         // Creates table name.
         let table_name = TableNameKey::new(
@@ -605,7 +606,7 @@ impl TableMetadataManager {
             .table_name_manager()
             .build_create_txn(&table_name, table_id)?;
 
-        let region_options = table_info.to_region_options();
+        let region_options = (&table_info.meta.options).into();
         // Creates table info.
         let table_info_value = TableInfoValue::new(table_info);
         let (create_table_info_txn, on_create_table_info_failure) = self
@@ -624,7 +625,6 @@ impl TableMetadataManager {
         ]);
 
         if let TableRouteValue::Physical(x) = &table_route_value {
-            let region_storage_path = table_info_value.region_storage_path();
             let create_datanode_table_txn = self.datanode_table_manager().build_create_txn(
                 table_id,
                 &engine,
@@ -926,15 +926,13 @@ impl TableMetadataManager {
     }
 
     /// Updates table info and returns an error if different metadata exists.
-    /// And cascade-ly update all redundant table options for each region
-    /// if region_distribution is present.
     pub async fn update_table_info(
         &self,
         current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
-        region_distribution: Option<RegionDistribution>,
         new_table_info: RawTableInfo,
     ) -> Result<()> {
         let table_id = current_table_info_value.table_info.ident.table_id;
+
         let new_table_info_value = current_table_info_value.update(new_table_info);
 
         // Updates table info.
@@ -942,19 +940,8 @@ impl TableMetadataManager {
             .table_info_manager()
             .build_update_txn(table_id, current_table_info_value, &new_table_info_value)?;
 
-        let txn = if let Some(region_distribution) = region_distribution {
-            // region options induced from table info.
-            let new_region_options = new_table_info_value.table_info.to_region_options();
-            let update_datanode_table_options_txn = self
-                .datanode_table_manager
-                .build_update_table_options_txn(table_id, region_distribution, new_region_options)
-                .await?;
-            Txn::merge_all([update_table_info_txn, update_datanode_table_options_txn])
-        } else {
-            update_table_info_txn
-        };
+        let mut r = self.kv_backend.txn(update_table_info_txn).await?;
 
-        let mut r = self.kv_backend.txn(txn).await?;
         // Checks whether metadata was already updated.
         if !r.succeeded {
             let mut set = TxnOpGetResponseSet::from(&mut r.responses);
@@ -1682,12 +1669,12 @@ mod tests {
             DeserializedValueWithBytes::from_inner(TableInfoValue::new(table_info.clone()));
         // should be ok.
         table_metadata_manager
-            .update_table_info(&current_table_info_value, None, new_table_info.clone())
+            .update_table_info(&current_table_info_value, new_table_info.clone())
             .await
             .unwrap();
         // if table info was updated, it should be ok.
         table_metadata_manager
-            .update_table_info(&current_table_info_value, None, new_table_info.clone())
+            .update_table_info(&current_table_info_value, new_table_info.clone())
             .await
             .unwrap();
 
@@ -1709,7 +1696,7 @@ mod tests {
         // if the current_table_info_value is wrong, it should return an error.
         // The ABA problem.
         assert!(table_metadata_manager
-            .update_table_info(&wrong_table_info_value, None, new_table_info)
+            .update_table_info(&wrong_table_info_value, new_table_info)
             .await
             .is_err())
     }

src/common/meta/src/key/catalog_name.rs

@@ -35,7 +35,7 @@ pub struct CatalogNameKey<'a> {
     pub catalog: &'a str,
 }
 
-impl Default for CatalogNameKey<'_> {
+impl<'a> Default for CatalogNameKey<'a> {
     fn default() -> Self {
         Self {
             catalog: DEFAULT_CATALOG_NAME,

src/common/meta/src/key/datanode_table.rs

@@ -23,7 +23,7 @@ use store_api::storage::RegionNumber;
 use table::metadata::TableId;
 
 use super::MetadataKey;
-use crate::error::{DatanodeTableInfoNotFoundSnafu, InvalidMetadataSnafu, Result};
+use crate::error::{InvalidMetadataSnafu, Result};
 use crate::key::{
     MetadataValue, RegionDistribution, DATANODE_TABLE_KEY_PATTERN, DATANODE_TABLE_KEY_PREFIX,
 };
@@ -77,7 +77,7 @@ impl DatanodeTableKey {
     }
 }
 
-impl MetadataKey<'_, DatanodeTableKey> for DatanodeTableKey {
+impl<'a> MetadataKey<'a, DatanodeTableKey> for DatanodeTableKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
@@ -209,49 +209,6 @@ impl DatanodeTableManager {
         Ok(txn)
     }
 
-    /// Builds a transaction to updates the redundant table options (including WAL options)
-    /// for given table id, if provided.
-    ///
-    /// Note that the provided `new_region_options` must be a
-    /// complete set of all options rather than incremental changes.
-    pub(crate) async fn build_update_table_options_txn(
-        &self,
-        table_id: TableId,
-        region_distribution: RegionDistribution,
-        new_region_options: HashMap<String, String>,
-    ) -> Result<Txn> {
-        assert!(!region_distribution.is_empty());
-        // safety: region_distribution must not be empty
-        let (any_datanode, _) = region_distribution.first_key_value().unwrap();
-
-        let mut region_info = self
-            .kv_backend
-            .get(&DatanodeTableKey::new(*any_datanode, table_id).to_bytes())
-            .await
-            .transpose()
-            .context(DatanodeTableInfoNotFoundSnafu {
-                datanode_id: *any_datanode,
-                table_id,
-            })?
-            .and_then(|r| DatanodeTableValue::try_from_raw_value(&r.value))?
-            .region_info;
-        // substitute region options only.
-        region_info.region_options = new_region_options;
-
-        let mut txns = Vec::with_capacity(region_distribution.len());
-
-        for (datanode, regions) in region_distribution.into_iter() {
-            let key = DatanodeTableKey::new(datanode, table_id);
-            let key_bytes = key.to_bytes();
-            let value_bytes = DatanodeTableValue::new(table_id, regions, region_info.clone())
-                .try_as_raw_value()?;
-            txns.push(TxnOp::Put(key_bytes, value_bytes));
-        }
-
-        let txn = Txn::new().and_then(txns);
-        Ok(txn)
-    }
-
     /// Builds the update datanode table transactions. It only executes while the primary keys comparing successes.
     pub(crate) fn build_update_txn(
         &self,

src/common/meta/src/key/flow/flow_info.rs

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

src/common/meta/src/key/flow/flow_name.rs

@@ -46,8 +46,6 @@ lazy_static! {
 /// The layout: `__flow/name/{catalog_name}/{flow_name}`.
 pub struct FlowNameKey<'a>(FlowScoped<FlowNameKeyInner<'a>>);
 
-pub type FlowNameDecodeResult = Result<Option<DeserializedValueWithBytes<FlowNameValue>>>;
-
 #[allow(dead_code)]
 impl<'a> FlowNameKey<'a> {
     /// Returns the [FlowNameKey]
@@ -106,7 +104,7 @@ impl<'a> MetadataKey<'a, FlowNameKeyInner<'a>> for FlowNameKeyInner<'_> {
         .into_bytes()
     }
 
-    fn from_bytes(bytes: &'a [u8]) -> Result<FlowNameKeyInner<'a>> {
+    fn from_bytes(bytes: &'a [u8]) -> Result<FlowNameKeyInner> {
         let key = std::str::from_utf8(bytes).map_err(|e| {
             error::InvalidMetadataSnafu {
                 err_msg: format!(
@@ -225,7 +223,9 @@ impl FlowNameManager {
         flow_id: FlowId,
     ) -> Result<(
         Txn,
-        impl FnOnce(&mut TxnOpGetResponseSet) -> FlowNameDecodeResult,
+        impl FnOnce(
+            &mut TxnOpGetResponseSet,
+        ) -> Result<Option<DeserializedValueWithBytes<FlowNameValue>>>,
     )> {
         let key = FlowNameKey::new(catalog_name, flow_name);
         let raw_key = key.to_bytes();

src/common/meta/src/key/node_address.rs

@@ -52,7 +52,7 @@ impl NodeAddressValue {
     }
 }
 
-impl MetadataKey<'_, NodeAddressKey> for NodeAddressKey {
+impl<'a> MetadataKey<'a, NodeAddressKey> for NodeAddressKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }

src/common/meta/src/key/schema_metadata_manager.rs

@@ -1,122 +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.
-
-//! Schema-level metadata manager.
-
-use std::sync::Arc;
-
-use snafu::OptionExt;
-use store_api::storage::TableId;
-
-use crate::error::TableInfoNotFoundSnafu;
-use crate::key::schema_name::{SchemaManager, SchemaNameKey};
-use crate::key::table_info::{TableInfoManager, TableInfoManagerRef};
-use crate::kv_backend::KvBackendRef;
-use crate::{error, SchemaOptions};
-
-pub type SchemaMetadataManagerRef = Arc<SchemaMetadataManager>;
-
-pub struct SchemaMetadataManager {
-    table_info_manager: TableInfoManagerRef,
-    schema_manager: SchemaManager,
-    #[cfg(any(test, feature = "testing"))]
-    kv_backend: KvBackendRef,
-}
-
-impl SchemaMetadataManager {
-    /// Creates a new database meta
-    #[cfg(not(any(test, feature = "testing")))]
-    pub fn new(kv_backend: KvBackendRef) -> Self {
-        let table_info_manager = Arc::new(TableInfoManager::new(kv_backend.clone()));
-        let schema_manager = SchemaManager::new(kv_backend);
-        Self {
-            table_info_manager,
-            schema_manager,
-        }
-    }
-
-    /// Creates a new database meta
-    #[cfg(any(test, feature = "testing"))]
-    pub fn new(kv_backend: KvBackendRef) -> Self {
-        let table_info_manager = Arc::new(TableInfoManager::new(kv_backend.clone()));
-        let schema_manager = SchemaManager::new(kv_backend.clone());
-        Self {
-            table_info_manager,
-            schema_manager,
-            kv_backend,
-        }
-    }
-
-    /// Gets schema options by table id.
-    pub async fn get_schema_options_by_table_id(
-        &self,
-        table_id: TableId,
-    ) -> error::Result<Option<SchemaOptions>> {
-        let table_info = self
-            .table_info_manager
-            .get(table_id)
-            .await?
-            .with_context(|| TableInfoNotFoundSnafu {
-                table: format!("table id: {}", table_id),
-            })?;
-
-        let key = SchemaNameKey::new(
-            &table_info.table_info.catalog_name,
-            &table_info.table_info.schema_name,
-        );
-        self.schema_manager.get(key).await
-    }
-
-    #[cfg(any(test, feature = "testing"))]
-    pub async fn register_region_table_info(
-        &self,
-        table_id: TableId,
-        table_name: &str,
-        schema_name: &str,
-        catalog_name: &str,
-        schema_value: Option<crate::key::schema_name::SchemaNameValue>,
-    ) {
-        use table::metadata::{RawTableInfo, TableType};
-        let value = crate::key::table_info::TableInfoValue::new(RawTableInfo {
-            ident: Default::default(),
-            name: table_name.to_string(),
-            desc: None,
-            catalog_name: catalog_name.to_string(),
-            schema_name: schema_name.to_string(),
-            meta: Default::default(),
-            table_type: TableType::Base,
-        });
-        let (txn, _) = self
-            .table_info_manager
-            .build_create_txn(table_id, &value)
-            .unwrap();
-        let resp = self.kv_backend.txn(txn).await.unwrap();
-        assert!(resp.succeeded, "Failed to create table metadata");
-        let key = SchemaNameKey {
-            catalog: catalog_name,
-            schema: schema_name,
-        };
-        self.schema_manager
-            .create(key, schema_value, false)
-            .await
-            .expect("Failed to create schema metadata");
-        common_telemetry::info!(
-            "Register table: {}, id: {}, schema: {}, catalog: {}",
-            table_name,
-            table_id,
-            schema_name,
-            catalog_name
-        );
-    }
-}

src/common/meta/src/key/schema_name.rs

@@ -41,7 +41,7 @@ pub struct SchemaNameKey<'a> {
     pub schema: &'a str,
 }
 
-impl Default for SchemaNameKey<'_> {
+impl<'a> Default for SchemaNameKey<'a> {
     fn default() -> Self {
         Self {
             catalog: DEFAULT_CATALOG_NAME,

src/common/meta/src/key/table_info.rs

@@ -23,7 +23,6 @@ use table::table_name::TableName;
 use table::table_reference::TableReference;
 
 use super::TABLE_INFO_KEY_PATTERN;
-use crate::ddl::utils::region_storage_path;
 use crate::error::{InvalidMetadataSnafu, Result};
 use crate::key::txn_helper::TxnOpGetResponseSet;
 use crate::key::{DeserializedValueWithBytes, MetadataKey, MetadataValue, TABLE_INFO_KEY_PREFIX};
@@ -52,7 +51,7 @@ impl Display for TableInfoKey {
     }
 }
 
-impl MetadataKey<'_, TableInfoKey> for TableInfoKey {
+impl<'a> MetadataKey<'a, TableInfoKey> for TableInfoKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
@@ -126,20 +125,13 @@ impl TableInfoValue {
             table_name: self.table_info.name.to_string(),
         }
     }
-
-    /// Builds storage path for all regions in table.
-    pub fn region_storage_path(&self) -> String {
-        region_storage_path(&self.table_info.catalog_name, &self.table_info.schema_name)
-    }
 }
 
 pub type TableInfoManagerRef = Arc<TableInfoManager>;
-
 #[derive(Clone)]
 pub struct TableInfoManager {
     kv_backend: KvBackendRef,
 }
-pub type TableInfoDecodeResult = Result<Option<DeserializedValueWithBytes<TableInfoValue>>>;
 
 impl TableInfoManager {
     pub fn new(kv_backend: KvBackendRef) -> Self {
@@ -153,7 +145,9 @@ impl TableInfoManager {
         table_info_value: &TableInfoValue,
     ) -> Result<(
         Txn,
-        impl FnOnce(&mut TxnOpGetResponseSet) -> TableInfoDecodeResult,
+        impl FnOnce(
+            &mut TxnOpGetResponseSet,
+        ) -> Result<Option<DeserializedValueWithBytes<TableInfoValue>>>,
     )> {
         let key = TableInfoKey::new(table_id);
         let raw_key = key.to_bytes();
@@ -175,7 +169,9 @@ impl TableInfoManager {
         new_table_info_value: &TableInfoValue,
     ) -> Result<(
         Txn,
-        impl FnOnce(&mut TxnOpGetResponseSet) -> TableInfoDecodeResult,
+        impl FnOnce(
+            &mut TxnOpGetResponseSet,
+        ) -> Result<Option<DeserializedValueWithBytes<TableInfoValue>>>,
     )> {
         let key = TableInfoKey::new(table_id);
         let raw_key = key.to_bytes();

src/common/meta/src/key/table_route.rs

@@ -245,7 +245,7 @@ impl LogicalTableRouteValue {
     }
 }
 
-impl MetadataKey<'_, TableRouteKey> for TableRouteKey {
+impl<'a> MetadataKey<'a, TableRouteKey> for TableRouteKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
@@ -472,8 +472,6 @@ pub struct TableRouteStorage {
     kv_backend: KvBackendRef,
 }
 
-pub type TableRouteValueDecodeResult = Result<Option<DeserializedValueWithBytes<TableRouteValue>>>;
-
 impl TableRouteStorage {
     pub fn new(kv_backend: KvBackendRef) -> Self {
         Self { kv_backend }
@@ -487,7 +485,9 @@ impl TableRouteStorage {
         table_route_value: &TableRouteValue,
     ) -> Result<(
         Txn,
-        impl FnOnce(&mut TxnOpGetResponseSet) -> TableRouteValueDecodeResult,
+        impl FnOnce(
+            &mut TxnOpGetResponseSet,
+        ) -> Result<Option<DeserializedValueWithBytes<TableRouteValue>>>,
     )> {
         let key = TableRouteKey::new(table_id);
         let raw_key = key.to_bytes();
@@ -510,7 +510,9 @@ impl TableRouteStorage {
         new_table_route_value: &TableRouteValue,
     ) -> Result<(
         Txn,
-        impl FnOnce(&mut TxnOpGetResponseSet) -> TableRouteValueDecodeResult,
+        impl FnOnce(
+            &mut TxnOpGetResponseSet,
+        ) -> Result<Option<DeserializedValueWithBytes<TableRouteValue>>>,
     )> {
         let key = TableRouteKey::new(table_id);
         let raw_key = key.to_bytes();

src/common/meta/src/key/view_info.rs

@@ -53,7 +53,7 @@ impl Display for ViewInfoKey {
     }
 }
 
-impl MetadataKey<'_, ViewInfoKey> for ViewInfoKey {
+impl<'a> MetadataKey<'a, ViewInfoKey> for ViewInfoKey {
     fn to_bytes(&self) -> Vec<u8> {
         self.to_string().into_bytes()
     }
@@ -139,8 +139,6 @@ pub struct ViewInfoManager {
 
 pub type ViewInfoManagerRef = Arc<ViewInfoManager>;
 
-pub type ViewInfoValueDecodeResult = Result<Option<DeserializedValueWithBytes<ViewInfoValue>>>;
-
 impl ViewInfoManager {
     pub fn new(kv_backend: KvBackendRef) -> Self {
         Self { kv_backend }
@@ -153,7 +151,9 @@ impl ViewInfoManager {
         view_info_value: &ViewInfoValue,
     ) -> Result<(
         Txn,
-        impl FnOnce(&mut TxnOpGetResponseSet) -> ViewInfoValueDecodeResult,
+        impl FnOnce(
+            &mut TxnOpGetResponseSet,
+        ) -> Result<Option<DeserializedValueWithBytes<ViewInfoValue>>>,
     )> {
         let key = ViewInfoKey::new(view_id);
         let raw_key = key.to_bytes();
@@ -175,7 +175,9 @@ impl ViewInfoManager {
         new_view_info_value: &ViewInfoValue,
     ) -> Result<(
         Txn,
-        impl FnOnce(&mut TxnOpGetResponseSet) -> ViewInfoValueDecodeResult,
+        impl FnOnce(
+            &mut TxnOpGetResponseSet,
+        ) -> Result<Option<DeserializedValueWithBytes<ViewInfoValue>>>,
     )> {
         let key = ViewInfoKey::new(view_id);
         let raw_key = key.to_bytes();

src/common/meta/src/leadership_notifier.rs

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::{Arc, Mutex};
+use std::sync::Arc;
 
 use async_trait::async_trait;
-use common_telemetry::{error, info};
+use common_telemetry::error;
 
 use crate::error::Result;
 
@@ -24,8 +24,6 @@ pub type LeadershipChangeNotifierCustomizerRef = Arc<dyn LeadershipChangeNotifie
 /// A trait for customizing the leadership change notifier.
 pub trait LeadershipChangeNotifierCustomizer: Send + Sync {
     fn customize(&self, notifier: &mut LeadershipChangeNotifier);
-
-    fn add_listener(&self, listener: Arc<dyn LeadershipChangeListener>);
 }
 
 /// A trait for handling leadership change events in a distributed system.
@@ -47,31 +45,6 @@ pub struct LeadershipChangeNotifier {
     listeners: Vec<Arc<dyn LeadershipChangeListener>>,
 }
 
-#[derive(Default)]
-pub struct DefaultLeadershipChangeNotifierCustomizer {
-    listeners: Mutex<Vec<Arc<dyn LeadershipChangeListener>>>,
-}
-
-impl DefaultLeadershipChangeNotifierCustomizer {
-    pub fn new() -> Self {
-        Self {
-            listeners: Mutex::new(Vec::new()),
-        }
-    }
-}
-
-impl LeadershipChangeNotifierCustomizer for DefaultLeadershipChangeNotifierCustomizer {
-    fn customize(&self, notifier: &mut LeadershipChangeNotifier) {
-        info!("Customizing leadership change notifier");
-        let listeners = self.listeners.lock().unwrap().clone();
-        notifier.listeners.extend(listeners);
-    }
-
-    fn add_listener(&self, listener: Arc<dyn LeadershipChangeListener>) {
-        self.listeners.lock().unwrap().push(listener);
-    }
-}
-
 impl LeadershipChangeNotifier {
     /// Adds a listener to the notifier.
     pub fn add_listener(&mut self, listener: Arc<dyn LeadershipChangeListener>) {

src/common/meta/src/lib.rs

@@ -54,7 +54,4 @@ pub type DatanodeId = u64;
 // The id of the flownode.
 pub type FlownodeId = u64;
 
-/// Schema options.
-pub type SchemaOptions = key::schema_name::SchemaNameValue;
-
 pub use instruction::RegionIdent;

src/common/meta/src/lock_key.rs

@@ -34,7 +34,7 @@ pub enum CatalogLock<'a> {
     Write(&'a str),
 }
 
-impl Display for CatalogLock<'_> {
+impl<'a> Display for CatalogLock<'a> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let key = match self {
             CatalogLock::Read(s) => s,
@@ -44,7 +44,7 @@ impl Display for CatalogLock<'_> {
     }
 }
 
-impl From<CatalogLock<'_>> for StringKey {
+impl<'a> From<CatalogLock<'a>> for StringKey {
     fn from(value: CatalogLock) -> Self {
         match value {
             CatalogLock::Write(_) => StringKey::Exclusive(value.to_string()),

src/common/meta/src/rpc/router.rs

@@ -289,7 +289,6 @@ pub enum LeaderState {
     ///
     /// - The [`Region`] may be unavailable (e.g., Crashed, Network disconnected).
     /// - The [`Region`] was planned to migrate to another [`Peer`].
-    #[serde(alias = "Downgraded")]
     Downgrading,
 }
 
@@ -517,73 +516,6 @@ mod tests {
         assert_eq!(decoded, region_route);
     }
 
-    #[test]
-    fn test_region_route_compatibility() {
-        let region_route = RegionRoute {
-            region: Region {
-                id: 2.into(),
-                name: "r2".to_string(),
-                partition: None,
-                attrs: BTreeMap::new(),
-            },
-            leader_peer: Some(Peer::new(1, "a1")),
-            follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
-            leader_state: Some(LeaderState::Downgrading),
-            leader_down_since: None,
-        };
-        let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}],"leader_state":"Downgraded","leader_down_since":null}"#;
-        let decoded: RegionRoute = serde_json::from_str(input).unwrap();
-        assert_eq!(decoded, region_route);
-
-        let region_route = RegionRoute {
-            region: Region {
-                id: 2.into(),
-                name: "r2".to_string(),
-                partition: None,
-                attrs: BTreeMap::new(),
-            },
-            leader_peer: Some(Peer::new(1, "a1")),
-            follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
-            leader_state: Some(LeaderState::Downgrading),
-            leader_down_since: None,
-        };
-        let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}],"leader_status":"Downgraded","leader_down_since":null}"#;
-        let decoded: RegionRoute = serde_json::from_str(input).unwrap();
-        assert_eq!(decoded, region_route);
-
-        let region_route = RegionRoute {
-            region: Region {
-                id: 2.into(),
-                name: "r2".to_string(),
-                partition: None,
-                attrs: BTreeMap::new(),
-            },
-            leader_peer: Some(Peer::new(1, "a1")),
-            follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
-            leader_state: Some(LeaderState::Downgrading),
-            leader_down_since: None,
-        };
-        let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}],"leader_state":"Downgrading","leader_down_since":null}"#;
-        let decoded: RegionRoute = serde_json::from_str(input).unwrap();
-        assert_eq!(decoded, region_route);
-
-        let region_route = RegionRoute {
-            region: Region {
-                id: 2.into(),
-                name: "r2".to_string(),
-                partition: None,
-                attrs: BTreeMap::new(),
-            },
-            leader_peer: Some(Peer::new(1, "a1")),
-            follower_peers: vec![Peer::new(2, "a2"), Peer::new(3, "a3")],
-            leader_state: Some(LeaderState::Downgrading),
-            leader_down_since: None,
-        };
-        let input = r#"{"region":{"id":2,"name":"r2","partition":null,"attrs":{}},"leader_peer":{"id":1,"addr":"a1"},"follower_peers":[{"id":2,"addr":"a2"},{"id":3,"addr":"a3"}],"leader_status":"Downgrading","leader_down_since":null}"#;
-        let decoded: RegionRoute = serde_json::from_str(input).unwrap();
-        assert_eq!(decoded, region_route);
-    }
-
     #[test]
     fn test_de_serialize_partition() {
         let p = Partition {

src/common/options/Cargo.toml

@@ -1,13 +0,0 @@
-[package]
-name = "common-options"
-version.workspace = true
-edition.workspace = true
-license.workspace = true
-
-[dependencies]
-common-grpc.workspace = true
-humantime-serde.workspace = true
-serde.workspace = true
-
-[lints]
-workspace = true

src/common/pprof/Cargo.toml

@@ -1,22 +0,0 @@
-[package]
-name = "common-pprof"
-version.workspace = true
-edition.workspace = true
-license.workspace = true
-
-[dependencies]
-common-error.workspace = true
-common-macro.workspace = true
-prost.workspace = true
-snafu.workspace = true
-tokio.workspace = true
-
-[target.'cfg(unix)'.dependencies]
-pprof = { version = "0.13", features = [
-    "flamegraph",
-    "prost-codec",
-    "protobuf",
-] }
-
-[lints]
-workspace = true

src/common/pprof/src/lib.rs

@@ -1,99 +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.
-
-#[cfg(unix)]
-pub mod nix;
-
-pub mod error {
-    use std::any::Any;
-
-    use common_error::ext::ErrorExt;
-    use common_error::status_code::StatusCode;
-    use common_macro::stack_trace_debug;
-    use snafu::{Location, Snafu};
-
-    #[derive(Snafu)]
-    #[stack_trace_debug]
-    #[snafu(visibility(pub(crate)))]
-    pub enum Error {
-        #[cfg(unix)]
-        #[snafu(display("Pprof error"))]
-        Pprof {
-            #[snafu(source)]
-            error: pprof::Error,
-            #[snafu(implicit)]
-            location: Location,
-        },
-
-        #[snafu(display("Pprof is unsupported on this platform"))]
-        Unsupported {
-            #[snafu(implicit)]
-            location: Location,
-        },
-    }
-
-    pub type Result<T> = std::result::Result<T, Error>;
-
-    impl ErrorExt for Error {
-        fn status_code(&self) -> StatusCode {
-            match self {
-                #[cfg(unix)]
-                Error::Pprof { .. } => StatusCode::Unexpected,
-                Error::Unsupported { .. } => StatusCode::Unsupported,
-            }
-        }
-
-        fn as_any(&self) -> &dyn Any {
-            self
-        }
-    }
-}
-
-#[cfg(not(unix))]
-pub mod dummy {
-    use std::time::Duration;
-
-    use crate::error::{Result, UnsupportedSnafu};
-
-    /// Dummpy CPU profiler utility.
-    #[derive(Debug)]
-    pub struct Profiling {}
-
-    impl Profiling {
-        /// Creates a new profiler.
-        pub fn new(_duration: Duration, _frequency: i32) -> Profiling {
-            Profiling {}
-        }
-
-        /// Profiles and returns a generated text.
-        pub async fn dump_text(&self) -> Result<String> {
-            UnsupportedSnafu {}.fail()
-        }
-
-        /// Profiles and returns a generated flamegraph.
-        pub async fn dump_flamegraph(&self) -> Result<Vec<u8>> {
-            UnsupportedSnafu {}.fail()
-        }
-
-        /// Profiles and returns a generated proto.
-        pub async fn dump_proto(&self) -> Result<Vec<u8>> {
-            UnsupportedSnafu {}.fail()
-        }
-    }
-}
-
-#[cfg(not(unix))]
-pub use dummy::Profiling;
-#[cfg(unix)]
-pub use nix::Profiling;

src/common/pprof/src/nix.rs

@@ -1,78 +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::time::Duration;
-
-use pprof::protos::Message;
-use snafu::ResultExt;
-
-use crate::error::{PprofSnafu, Result};
-
-/// CPU profiler utility.
-// Inspired by https://github.com/datafuselabs/databend/blob/67f445e83cd4eceda98f6c1c114858929d564029/src/common/base/src/base/profiling.rs
-#[derive(Debug)]
-pub struct Profiling {
-    /// Sample duration.
-    duration: Duration,
-    /// Sample frequency.
-    frequency: i32,
-}
-
-impl Profiling {
-    /// Creates a new profiler.
-    pub fn new(duration: Duration, frequency: i32) -> Profiling {
-        Profiling {
-            duration,
-            frequency,
-        }
-    }
-
-    /// Profiles and returns a generated pprof report.
-    pub async fn report(&self) -> Result<pprof::Report> {
-        let guard = pprof::ProfilerGuardBuilder::default()
-            .frequency(self.frequency)
-            .blocklist(&["libc", "libgcc", "pthread", "vdso"])
-            .build()
-            .context(PprofSnafu)?;
-        tokio::time::sleep(self.duration).await;
-        guard.report().build().context(PprofSnafu)
-    }
-
-    /// Profiles and returns a generated text.
-    pub async fn dump_text(&self) -> Result<String> {
-        let report = self.report().await?;
-        let text = format!("{report:?}");
-        Ok(text)
-    }
-
-    /// Profiles and returns a generated flamegraph.
-    pub async fn dump_flamegraph(&self) -> Result<Vec<u8>> {
-        let mut body: Vec<u8> = Vec::new();
-
-        let report = self.report().await?;
-        report.flamegraph(&mut body).context(PprofSnafu)?;
-
-        Ok(body)
-    }
-
-    /// Profiles and returns a generated proto.
-    pub async fn dump_proto(&self) -> Result<Vec<u8>> {
-        let report = self.report().await?;
-        // Generate google’s pprof format report.
-        let profile = report.pprof().context(PprofSnafu)?;
-        let body = profile.encode_to_vec();
-
-        Ok(body)
-    }
-}

src/common/procedure/src/store.rs

@@ -297,7 +297,7 @@ struct ParsedKey<'a> {
     key_type: KeyType,
 }
 
-impl fmt::Display for ParsedKey<'_> {
+impl<'a> fmt::Display for ParsedKey<'a> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,