chore: rm unwrap

Signed-off-by: discord9 <discord9@163.com>

.cargo/config.toml

@@ -12,3 +12,6 @@ fetch = true
 checkout = true
 list_files = true
 internal_use_git2 = false
+
+[env]
+CARGO_WORKSPACE_DIR = { value = "", relative = true }

.github/labeler.yaml

@@ -0,0 +1,15 @@
+ci:
+  - changed-files:
+      - any-glob-to-any-file: .github/**
+
+docker:
+  - changed-files:
+      - any-glob-to-any-file: docker/**
+
+documentation:
+  - changed-files:
+      - any-glob-to-any-file: docs/**
+
+dashboard:
+  - changed-files:
+      - any-glob-to-any-file: grafana/**

.github/scripts/check-version.sh

@@ -0,0 +1,42 @@
+#!/bin/bash
+
+# Get current version
+CURRENT_VERSION=$1
+if [ -z "$CURRENT_VERSION" ]; then
+  echo "Error: Failed to get current version"
+  exit 1
+fi
+
+# Get the latest version from GitHub Releases
+API_RESPONSE=$(curl -s "https://api.github.com/repos/GreptimeTeam/greptimedb/releases/latest")
+
+if [ -z "$API_RESPONSE" ] || [ "$(echo "$API_RESPONSE" | jq -r '.message')" = "Not Found" ]; then
+  echo "Error: Failed to fetch latest version from GitHub"
+  exit 1
+fi
+
+# Get the latest version
+LATEST_VERSION=$(echo "$API_RESPONSE" | jq -r '.tag_name')
+
+if [ -z "$LATEST_VERSION" ] || [ "$LATEST_VERSION" = "null" ]; then
+  echo "Error: No valid version found in GitHub releases"
+  exit 1
+fi
+
+# Cleaned up version number format (removed possible 'v' prefix and -nightly suffix)
+CLEAN_CURRENT=$(echo "$CURRENT_VERSION" | sed 's/^v//' | sed 's/-nightly-.*//')
+CLEAN_LATEST=$(echo "$LATEST_VERSION" | sed 's/^v//' | sed 's/-nightly-.*//')
+
+echo "Current version: $CLEAN_CURRENT"
+echo "Latest release version: $CLEAN_LATEST"
+
+# Use sort -V to compare versions
+HIGHER_VERSION=$(printf "%s\n%s" "$CLEAN_CURRENT" "$CLEAN_LATEST" | sort -V | tail -n1)
+
+if [ "$HIGHER_VERSION" = "$CLEAN_CURRENT" ]; then
+  echo "Current version ($CLEAN_CURRENT) is NEWER than or EQUAL to latest ($CLEAN_LATEST)"
+  echo "should-push-latest-tag=true" >> $GITHUB_OUTPUT
+else
+  echo "Current version ($CLEAN_CURRENT) is OLDER than latest ($CLEAN_LATEST)"
+  echo "should-push-latest-tag=false" >> $GITHUB_OUTPUT
+fi

.github/workflows/pr-labeling.yaml

@@ -0,0 +1,42 @@
+name: 'PR Labeling'
+
+on:
+  pull_request_target:
+    types:
+      - opened
+      - synchronize
+      - reopened
+
+permissions:
+  contents: read
+  pull-requests: write
+  issues: write
+
+jobs:
+  labeler:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout sources
+        uses: actions/checkout@v4
+
+      - uses: actions/labeler@v5
+        with:
+          configuration-path: ".github/labeler.yaml"
+          repo-token: "${{ secrets.GITHUB_TOKEN }}"
+
+  size-label:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: pascalgn/size-label-action@v0.5.5
+        env:
+          GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
+        with:
+          sizes: >
+            {
+              "0": "XS",
+              "100": "S",
+              "300": "M",
+              "1000": "L",
+              "1500": "XL",
+              "2000": "XXL"
+            }

.github/workflows/release.yml

@@ -110,6 +110,8 @@ jobs:
 
       # The 'version' use as the global tag name of the release workflow.
       version: ${{ steps.create-version.outputs.version }}
+
+      should-push-latest-tag: ${{ steps.check-version.outputs.should-push-latest-tag }}
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -135,6 +137,11 @@ jobs:
           GITHUB_REF_NAME: ${{ github.ref_name }}
           NIGHTLY_RELEASE_PREFIX: ${{ env.NIGHTLY_RELEASE_PREFIX }}
 
+      - name: Check version
+        id: check-version
+        run: |
+          ./.github/scripts/check-version.sh "${{ steps.create-version.outputs.version }}"
+
       - name: Allocate linux-amd64 runner
         if: ${{ inputs.build_linux_amd64_artifacts || github.event_name == 'push' || github.event_name == 'schedule' }}
         uses: ./.github/actions/start-runner
@@ -314,7 +321,7 @@ jobs:
           image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
           image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
           version: ${{ needs.allocate-runners.outputs.version }}
-          push-latest-tag: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
+          push-latest-tag: ${{ needs.allocate-runners.outputs.should-push-latest-tag == 'true' && github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
 
       - name: Set build image result
         id: set-build-image-result
@@ -332,7 +339,7 @@ jobs:
       build-windows-artifacts,
       release-images-to-dockerhub,
     ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-latest-16-cores
     # When we push to ACR, it's easy to fail due to some unknown network issues.
     # However, we don't want to fail the whole workflow because of this.
     # The ACR have daily sync with DockerHub, so don't worry about the image not being updated.
@@ -361,7 +368,7 @@ jobs:
           dev-mode: false
           upload-to-s3: true
           update-version-info: true
-          push-latest-tag: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
+          push-latest-tag: ${{ needs.allocate-runners.outputs.should-push-latest-tag == 'true' && github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
 
   publish-github-release:
     name: Create GitHub release and upload artifacts

.github/workflows/semantic-pull-request.yml

@@ -11,17 +11,17 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
+permissions:
+  issues: write
+  contents: write
+  pull-requests: write
+
 jobs:
   check:
     runs-on: ubuntu-latest
-    permissions:
-      pull-requests: write  # Add permissions to modify PRs
-      issues: write
     timeout-minutes: 10
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - name: Check Pull Request
         working-directory: cyborg

Cargo.toml

@@ -30,6 +30,7 @@ members = [
     "src/common/recordbatch",
     "src/common/runtime",
     "src/common/session",
+    "src/common/sql",
     "src/common/stat",
     "src/common/substrait",
     "src/common/telemetry",
@@ -71,11 +72,13 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.15.0"
+version = "0.16.0"
 edition = "2021"
 license = "Apache-2.0"
 
 [workspace.lints]
+clippy.print_stdout = "warn"
+clippy.print_stderr = "warn"
 clippy.dbg_macro = "warn"
 clippy.implicit_clone = "warn"
 clippy.result_large_err = "allow"
@@ -121,6 +124,7 @@ datafusion = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "
 datafusion-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
 datafusion-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
 datafusion-functions = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
+datafusion-functions-aggregate-common = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
 datafusion-optimizer = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
 datafusion-physical-expr = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
 datafusion-physical-plan = { git = "https://github.com/waynexia/arrow-datafusion.git", rev = "12c0381babd52c681043957e9d6ee083a03f7646" }
@@ -130,11 +134,12 @@ deadpool = "0.12"
 deadpool-postgres = "0.14"
 derive_builder = "0.20"
 dotenv = "0.15"
+either = "1.15"
 etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "17971523673f4fbc982510d3c9d6647ff642e16f" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "ceb1af4fa9309ce65bda0367db7b384df2bb4d4f" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -167,9 +172,7 @@ parquet = { version = "54.2", default-features = false, features = ["arrow", "as
 paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
-promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", rev = "0410e8b459dda7cb222ce9596f8bf3971bd07bd2", features = [
+promql-parser = { version = "0.6", features = ["ser"] }
-    "ser",
-] }
 prost = { version = "0.13", features = ["no-recursion-limit"] }
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.9"
@@ -258,6 +261,7 @@ common-query = { path = "src/common/query" }
 common-recordbatch = { path = "src/common/recordbatch" }
 common-runtime = { path = "src/common/runtime" }
 common-session = { path = "src/common/session" }
+common-sql = { path = "src/common/sql" }
 common-telemetry = { path = "src/common/telemetry" }
 common-test-util = { path = "src/common/test-util" }
 common-time = { path = "src/common/time" }

README.md

@@ -75,9 +75,9 @@
 | --------- | ----------- |
 | [Unified Observability Data](https://docs.greptime.com/user-guide/concepts/why-greptimedb) | Store metrics, logs, and traces as timestamped, contextual wide events. Query via [SQL](https://docs.greptime.com/user-guide/query-data/sql), [PromQL](https://docs.greptime.com/user-guide/query-data/promql), and [streaming](https://docs.greptime.com/user-guide/flow-computation/overview). |
 | [High Performance & Cost Effective](https://docs.greptime.com/user-guide/manage-data/data-index) | Written in Rust, with a distributed query engine, [rich indexing](https://docs.greptime.com/user-guide/manage-data/data-index), and optimized columnar storage, delivering sub-second responses at PB scale. |
-| [Cloud-Native Architecture](https://docs.greptime.com/user-guide/concepts/architecture) | Designed for [Kubernetes](https://docs.greptime.com/user-guide/deployments/deploy-on-kubernetes/greptimedb-operator-management), with compute/storage separation, native object storage (AWS S3, Azure Blob, etc.) and seamless cross-cloud access. |
+| [Cloud-Native Architecture](https://docs.greptime.com/user-guide/concepts/architecture) | Designed for [Kubernetes](https://docs.greptime.com/user-guide/deployments-administration/deploy-on-kubernetes/greptimedb-operator-management), with compute/storage separation, native object storage (AWS S3, Azure Blob, etc.) and seamless cross-cloud access. |
 | [Developer-Friendly](https://docs.greptime.com/user-guide/protocols/overview) | Access via SQL/PromQL interfaces, REST API, MySQL/PostgreSQL protocols, and popular ingestion [protocols](https://docs.greptime.com/user-guide/protocols/overview). |
-| [Flexible Deployment](https://docs.greptime.com/user-guide/deployments/overview) | Deploy anywhere: edge (including ARM/[Android](https://docs.greptime.com/user-guide/deployments/run-on-android)) or cloud, with unified APIs and efficient data sync. |
+| [Flexible Deployment](https://docs.greptime.com/user-guide/deployments-administration/overview) | Deploy anywhere: edge (including ARM/[Android](https://docs.greptime.com/user-guide/deployments-administration/run-on-android)) or cloud, with unified APIs and efficient data sync. |
 
 Learn more in [Why GreptimeDB](https://docs.greptime.com/user-guide/concepts/why-greptimedb) and [Observability 2.0 and the Database for It](https://greptime.com/blogs/2025-04-25-greptimedb-observability2-new-database).
 
@@ -189,7 +189,8 @@ We invite you to engage and contribute!
 - [Official Website](https://greptime.com/)
 - [Blog](https://greptime.com/blogs/)
 - [LinkedIn](https://www.linkedin.com/company/greptime/)
-- [Twitter](https://twitter.com/greptime)
+- [X (Twitter)](https://X.com/greptime)
+- [YouTube](https://www.youtube.com/@greptime)
 
 ## License
 

config/config.md

@@ -123,6 +123,7 @@
 | `storage.http_client.connect_timeout` | String | `30s` | The timeout for only the connect phase of a http client. |
 | `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.<br/>Also considered a total deadline. |
 | `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. |
+| `storage.http_client.skip_ssl_validation` | Bool | `false` | To skip the ssl verification<br/>**Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks. |
 | `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines. |
 | `region_engine.mito` | -- | -- | The Mito engine options. |
 | `region_engine.mito.num_workers` | Integer | `8` | Number of region workers. |
@@ -184,10 +185,11 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
+| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `slow_query` | -- | -- | The slow query log options. |
@@ -287,10 +289,11 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
+| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `slow_query` | -- | -- | The slow query log options. |
@@ -322,6 +325,7 @@
 | `selector` | String | `round_robin` | Datanode selector type.<br/>- `round_robin` (default value)<br/>- `lease_based`<br/>- `load_based`<br/>For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector". |
 | `use_memory_store` | Bool | `false` | Store data in memory. |
 | `enable_region_failover` | Bool | `false` | Whether to enable region failover.<br/>This feature is only available on GreptimeDB running on cluster mode and<br/>- Using Remote WAL<br/>- Using shared storage (e.g., s3). |
+| `region_failure_detector_initialization_delay` | String | `10m` | The delay before starting region failure detection.<br/>This delay helps prevent Metasrv from triggering unnecessary region failovers before all Datanodes are fully started.<br/>Especially useful when the cluster is not deployed with GreptimeDB Operator and maintenance mode is not enabled. |
 | `allow_region_failover_on_local_wal` | Bool | `false` | Whether to allow region failover on local WAL.<br/>**This option is not recommended to be set to true, because it may lead to data loss during failover.** |
 | `node_max_idle_time` | String | `24hours` | Max allowed idle time before removing node info from metasrv memory. |
 | `enable_telemetry` | Bool | `true` | Whether to enable greptimedb telemetry. Enabled by default. |
@@ -369,10 +373,11 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
+| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `export_metrics` | -- | -- | The metasrv can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
@@ -431,8 +436,8 @@
 | `wal.provider` | String | `raft_engine` | The provider of the WAL.<br/>- `raft_engine`: the wal is stored in the local file system by raft-engine.<br/>- `kafka`: it's remote wal that data is stored in Kafka. |
 | `wal.dir` | String | Unset | The directory to store the WAL files.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.file_size` | String | `128MB` | The size of the WAL segment file.<br/>**It's only used when the provider is `raft_engine`**. |
-| `wal.purge_threshold` | String | `1GB` | The threshold of the WAL size to trigger a flush.<br/>**It's only used when the provider is `raft_engine`**. |
+| `wal.purge_threshold` | String | `1GB` | The threshold of the WAL size to trigger a purge.<br/>**It's only used when the provider is `raft_engine`**. |
-| `wal.purge_interval` | String | `1m` | The interval to trigger a flush.<br/>**It's only used when the provider is `raft_engine`**. |
+| `wal.purge_interval` | String | `1m` | The interval to trigger a purge.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.read_batch_size` | Integer | `128` | The read batch size.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.sync_write` | Bool | `false` | Whether to use sync write.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.enable_log_recycle` | Bool | `true` | Whether to reuse logically truncated log files.<br/>**It's only used when the provider is `raft_engine`**. |
@@ -471,6 +476,7 @@
 | `storage.http_client.connect_timeout` | String | `30s` | The timeout for only the connect phase of a http client. |
 | `storage.http_client.timeout` | String | `30s` | The total request timeout, applied from when the request starts connecting until the response body has finished.<br/>Also considered a total deadline. |
 | `storage.http_client.pool_idle_timeout` | String | `90s` | The timeout for idle sockets being kept-alive. |
+| `storage.http_client.skip_ssl_validation` | Bool | `false` | To skip the ssl verification<br/>**Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks. |
 | `[[region_engine]]` | -- | -- | The region engine options. You can configure multiple region engines. |
 | `region_engine.mito` | -- | -- | The Mito engine options. |
 | `region_engine.mito.num_workers` | Integer | `8` | Number of region workers. |
@@ -532,10 +538,11 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
+| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `export_metrics` | -- | -- | The datanode can export its metrics and send to Prometheus compatible service (e.g. `greptimedb` itself) from remote-write API.<br/>This is only used for `greptimedb` to export its own metrics internally. It's different from prometheus scrape. |
@@ -582,11 +589,14 @@
 | `logging.dir` | String | `./greptimedb_data/logs` | The directory to store the log files. If set to empty, logs will not be written to files. |
 | `logging.level` | String | Unset | The log level. Can be `info`/`debug`/`warn`/`error`. |
 | `logging.enable_otlp_tracing` | Bool | `false` | Enable OTLP tracing. |
-| `logging.otlp_endpoint` | String | `http://localhost:4317` | The OTLP tracing endpoint. |
+| `logging.otlp_endpoint` | String | `http://localhost:4318` | The OTLP tracing endpoint. |
 | `logging.append_stdout` | Bool | `true` | Whether to append logs to stdout. |
 | `logging.log_format` | String | `text` | The log format. Can be `text`/`json`. |
 | `logging.max_log_files` | Integer | `720` | The maximum amount of log files. |
+| `logging.otlp_export_protocol` | String | `http` | The OTLP tracing export protocol. Can be `grpc`/`http`. |
 | `logging.tracing_sample_ratio` | -- | -- | The percentage of tracing will be sampled and exported.<br/>Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.<br/>ratio > 1 are treated as 1. Fractions < 0 are treated as 0 |
 | `logging.tracing_sample_ratio.default_ratio` | Float | `1.0` | -- |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |
 | `tracing.tokio_console_addr` | String | Unset | The tokio console address. |
+| `query` | -- | -- | -- |
+| `query.parallelism` | Integer | `1` | Parallelism of the query engine for query sent by flownode.<br/>Default to 1, so it won't use too much cpu or memory |

config/datanode.example.toml

@@ -129,11 +129,11 @@ dir = "./greptimedb_data/wal"
 ## **It's only used when the provider is `raft_engine`**.
 file_size = "128MB"
 
-## The threshold of the WAL size to trigger a flush.
+## The threshold of the WAL size to trigger a purge.
 ## **It's only used when the provider is `raft_engine`**.
 purge_threshold = "1GB"
 
-## The interval to trigger a flush.
+## The interval to trigger a purge.
 ## **It's only used when the provider is `raft_engine`**.
 purge_interval = "1m"
 
@@ -367,6 +367,10 @@ timeout = "30s"
 ## The timeout for idle sockets being kept-alive.
 pool_idle_timeout = "90s"
 
+## To skip the ssl verification
+## **Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks.
+skip_ssl_validation = false
+
 # Custom storage options
 # [[storage.providers]]
 # name = "S3"
@@ -625,7 +629,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4317"
+otlp_endpoint = "http://localhost:4318"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -636,6 +640,9 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720
 
+## The OTLP tracing export protocol. Can be `grpc`/`http`.
+otlp_export_protocol = "http"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0

config/flownode.example.toml

@@ -83,7 +83,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4317"
+otlp_endpoint = "http://localhost:4318"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -94,6 +94,9 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720
 
+## The OTLP tracing export protocol. Can be `grpc`/`http`.
+otlp_export_protocol = "http"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0
@@ -105,3 +108,8 @@ default_ratio = 1.0
 ## The tokio console address.
 ## @toml2docs:none-default
 #+ tokio_console_addr = "127.0.0.1"
+
+[query]
+## Parallelism of the query engine for query sent by flownode.
+## Default to 1, so it won't use too much cpu or memory
+parallelism = 1

config/frontend.example.toml

@@ -218,7 +218,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4317"
+otlp_endpoint = "http://localhost:4318"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -229,6 +229,9 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720
 
+## The OTLP tracing export protocol. Can be `grpc`/`http`.
+otlp_export_protocol = "http"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0

config/metasrv.example.toml

@@ -43,6 +43,11 @@ use_memory_store = false
 ## - Using shared storage (e.g., s3).
 enable_region_failover = false
 
+## The delay before starting region failure detection.
+## This delay helps prevent Metasrv from triggering unnecessary region failovers before all Datanodes are fully started.
+## Especially useful when the cluster is not deployed with GreptimeDB Operator and maintenance mode is not enabled.
+region_failure_detector_initialization_delay = '10m'
+
 ## Whether to allow region failover on local WAL.
 ## **This option is not recommended to be set to true, because it may lead to data loss during failover.**
 allow_region_failover_on_local_wal = false
@@ -220,7 +225,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4317"
+otlp_endpoint = "http://localhost:4318"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -231,6 +236,9 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720
 
+## The OTLP tracing export protocol. Can be `grpc`/`http`.
+otlp_export_protocol = "http"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0

config/standalone.example.toml

@@ -458,6 +458,10 @@ timeout = "30s"
 ## The timeout for idle sockets being kept-alive.
 pool_idle_timeout = "90s"
 
+## To skip the ssl verification
+## **Security Notice**: Setting `skip_ssl_validation = true` disables certificate verification, making connections vulnerable to man-in-the-middle attacks. Only use this in development or trusted private networks.
+skip_ssl_validation = false
+
 # Custom storage options
 # [[storage.providers]]
 # name = "S3"
@@ -716,7 +720,7 @@ level = "info"
 enable_otlp_tracing = false
 
 ## The OTLP tracing endpoint.
-otlp_endpoint = "http://localhost:4317"
+otlp_endpoint = "http://localhost:4318"
 
 ## Whether to append logs to stdout.
 append_stdout = true
@@ -727,6 +731,9 @@ log_format = "text"
 ## The maximum amount of log files.
 max_log_files = 720
 
+## The OTLP tracing export protocol. Can be `grpc`/`http`.
+otlp_export_protocol = "http"
+
 ## The percentage of tracing will be sampled and exported.
 ## Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1.
 ## ratio > 1 are treated as 1. Fractions < 0 are treated as 0

cyborg/bin/follow-up-docs-issue.ts

@@ -55,12 +55,25 @@ async function main() {
         await client.rest.issues.addLabels({
             owner, repo, issue_number: number, labels: [labelDocsRequired],
         })
+
+        // Get available assignees for the docs repo
+        const assigneesResponse = await docsClient.rest.issues.listAssignees({
+            owner: 'GreptimeTeam',
+            repo: 'docs',
+        })
+        const validAssignees = assigneesResponse.data.map(assignee => assignee.login)
+        core.info(`Available assignees: ${validAssignees.join(', ')}`)
+
+        // Check if the actor is a valid assignee, otherwise fallback to fengjiachun
+        const assignee = validAssignees.includes(actor) ? actor : 'fengjiachun'
+        core.info(`Assigning issue to: ${assignee}`)
+
         await docsClient.rest.issues.create({
             owner: 'GreptimeTeam',
             repo: 'docs',
             title: `Update docs for ${title}`,
             body: `A document change request is generated from ${html_url}`,
-            assignee: actor,
+            assignee: assignee,
         }).then((res) => {
             core.info(`Created issue ${res.data}`)
         })

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/deployments-administration/configuration#storage-options).

grafana/README.md

@@ -83,7 +83,7 @@ If you use the [Helm Chart](https://github.com/GreptimeTeam/helm-charts) to depl
 - `monitoring.enabled=true`: Deploys a standalone GreptimeDB instance dedicated to monitoring the cluster;
 - `grafana.enabled=true`: Deploys Grafana and automatically imports the monitoring dashboard;
 
-The standalone GreptimeDB instance will collect metrics from your cluster, and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/user-guide/deployments-administration/deploy-on-kubernetes/getting-started).
+The standalone GreptimeDB instance will collect metrics from your cluster, and the dashboard will be available in the Grafana UI. For detailed deployment instructions, please refer to our [Kubernetes deployment guide](https://docs.greptime.com/user-guide/deployments-administration-administration/deploy-on-kubernetes/getting-started).
 
 ### Self-host Prometheus and import dashboards manually
 

grafana/dashboards/metrics/cluster/dashboard.md

@@ -70,6 +70,7 @@
 | Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
 | Compaction Input/Output Bytes | `sum by(instance, pod) (greptime_mito_compaction_input_bytes)`<br/>`sum by(instance, pod) (greptime_mito_compaction_output_bytes)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-input` |
 | Region Worker Handle Bulk Insert Requests | `histogram_quantile(0.95, sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_bucket[$__rate_interval])))`<br/>`sum by(instance, stage, pod) (rate(greptime_region_worker_handle_write_sum[$__rate_interval]))/sum by(instance, stage, pod) (rate(greptime_region_worker_handle_write_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to handle bulk insert region requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
+| Active Series and Field Builders Count | `sum by(instance, pod) (greptime_mito_memtable_active_series_count)`<br/>`sum by(instance, pod) (greptime_mito_memtable_field_builder_count)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]-series` |
 | Region Worker Convert Requests | `histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))`<br/>`sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to decode requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |

grafana/dashboards/metrics/cluster/dashboard.yaml

@@ -612,6 +612,21 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
+        - title: Active Series and Field Builders Count
+          type: timeseries
+          description: Compaction oinput output bytes
+          unit: none
+          queries:
+            - expr: sum by(instance, pod) (greptime_mito_memtable_active_series_count)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-series'
+            - expr: sum by(instance, pod) (greptime_mito_memtable_field_builder_count)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-field_builders'
         - title: Region Worker Convert Requests
           type: timeseries
           description: Per-stage elapsed time for region worker to decode requests.

grafana/dashboards/metrics/standalone/dashboard.md

@@ -70,6 +70,7 @@
 | Inflight Flush | `greptime_mito_inflight_flush_count` | `timeseries` | Ongoing flush task count | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]` |
 | Compaction Input/Output Bytes | `sum by(instance, pod) (greptime_mito_compaction_input_bytes)`<br/>`sum by(instance, pod) (greptime_mito_compaction_output_bytes)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `bytes` | `[{{instance}}]-[{{pod}}]-input` |
 | Region Worker Handle Bulk Insert Requests | `histogram_quantile(0.95, sum by(le,instance, stage, pod) (rate(greptime_region_worker_handle_write_bucket[$__rate_interval])))`<br/>`sum by(instance, stage, pod) (rate(greptime_region_worker_handle_write_sum[$__rate_interval]))/sum by(instance, stage, pod) (rate(greptime_region_worker_handle_write_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to handle bulk insert region requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
+| Active Series and Field Builders Count | `sum by(instance, pod) (greptime_mito_memtable_active_series_count)`<br/>`sum by(instance, pod) (greptime_mito_memtable_field_builder_count)` | `timeseries` | Compaction oinput output bytes | `prometheus` | `none` | `[{{instance}}]-[{{pod}}]-series` |
 | Region Worker Convert Requests | `histogram_quantile(0.95, sum by(le, instance, stage, pod) (rate(greptime_datanode_convert_region_request_bucket[$__rate_interval])))`<br/>`sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_sum[$__rate_interval]))/sum by(le,instance, stage, pod) (rate(greptime_datanode_convert_region_request_count[$__rate_interval]))` | `timeseries` | Per-stage elapsed time for region worker to decode requests. | `prometheus` | `s` | `[{{instance}}]-[{{pod}}]-[{{stage}}]-P95` |
 # OpenDAL
 | Title | Query | Type | Description | Datasource | Unit | Legend Format |

grafana/dashboards/metrics/standalone/dashboard.yaml

@@ -612,6 +612,21 @@ groups:
                 type: prometheus
                 uid: ${metrics}
               legendFormat: '[{{instance}}]-[{{pod}}]-[{{stage}}]-AVG'
+        - title: Active Series and Field Builders Count
+          type: timeseries
+          description: Compaction oinput output bytes
+          unit: none
+          queries:
+            - expr: sum by(instance, pod) (greptime_mito_memtable_active_series_count)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-series'
+            - expr: sum by(instance, pod) (greptime_mito_memtable_field_builder_count)
+              datasource:
+                type: prometheus
+                uid: ${metrics}
+              legendFormat: '[{{instance}}]-[{{pod}}]-field_builders'
         - title: Region Worker Convert Requests
           type: timeseries
           description: Per-stage elapsed time for region worker to decode requests.

licenserc.toml

@@ -31,8 +31,10 @@ excludes = [
     "src/operator/src/expr_helper/trigger.rs",
     "src/sql/src/statements/create/trigger.rs",
     "src/sql/src/statements/show/trigger.rs",
+    "src/sql/src/statements/drop/trigger.rs",
     "src/sql/src/parsers/create_parser/trigger.rs",
     "src/sql/src/parsers/show_parser/trigger.rs",
+    "src/mito2/src/extension.rs",
 ]
 
 [properties]

src/api/src/region.rs

@@ -22,6 +22,7 @@ use greptime_proto::v1::region::RegionResponse as RegionResponseV1;
 pub struct RegionResponse {
     pub affected_rows: AffectedRows,
     pub extensions: HashMap<String, Vec<u8>>,
+    pub metadata: Vec<u8>,
 }
 
 impl RegionResponse {
@@ -29,6 +30,7 @@ impl RegionResponse {
         Self {
             affected_rows: region_response.affected_rows as _,
             extensions: region_response.extensions,
+            metadata: region_response.metadata,
         }
     }
 
@@ -37,6 +39,16 @@ impl RegionResponse {
         Self {
             affected_rows,
             extensions: Default::default(),
+            metadata: Vec::new(),
+        }
+    }
+
+    /// Creates one response with metadata.
+    pub fn from_metadata(metadata: Vec<u8>) -> Self {
+        Self {
+            affected_rows: 0,
+            extensions: Default::default(),
+            metadata,
         }
     }
 }

src/api/src/v1/column_def.rs

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

src/catalog/Cargo.toml

@@ -5,6 +5,7 @@ edition.workspace = true
 license.workspace = true
 
 [features]
+enterprise = []
 testing = []
 
 [lints]

src/catalog/src/kvbackend.rs

@@ -14,9 +14,11 @@
 
 pub use client::{CachedKvBackend, CachedKvBackendBuilder, MetaKvBackend};
 
+mod builder;
 mod client;
 mod manager;
 mod table_cache;
 
+pub use builder::KvBackendCatalogManagerBuilder;
 pub use manager::KvBackendCatalogManager;
 pub use table_cache::{new_table_cache, TableCache, TableCacheRef};

src/catalog/src/kvbackend/builder.rs

@@ -0,0 +1,131 @@
+// 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::sync::Arc;
+
+use common_catalog::consts::DEFAULT_CATALOG_NAME;
+use common_meta::cache::LayeredCacheRegistryRef;
+use common_meta::key::flow::FlowMetadataManager;
+use common_meta::key::TableMetadataManager;
+use common_meta::kv_backend::KvBackendRef;
+use common_procedure::ProcedureManagerRef;
+use moka::sync::Cache;
+use partition::manager::PartitionRuleManager;
+
+#[cfg(feature = "enterprise")]
+use crate::information_schema::InformationSchemaTableFactoryRef;
+use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
+use crate::kvbackend::manager::{SystemCatalog, CATALOG_CACHE_MAX_CAPACITY};
+use crate::kvbackend::KvBackendCatalogManager;
+use crate::process_manager::ProcessManagerRef;
+use crate::system_schema::pg_catalog::PGCatalogProvider;
+
+pub struct KvBackendCatalogManagerBuilder {
+    information_extension: InformationExtensionRef,
+    backend: KvBackendRef,
+    cache_registry: LayeredCacheRegistryRef,
+    procedure_manager: Option<ProcedureManagerRef>,
+    process_manager: Option<ProcessManagerRef>,
+    #[cfg(feature = "enterprise")]
+    extra_information_table_factories:
+        std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
+}
+
+impl KvBackendCatalogManagerBuilder {
+    pub fn new(
+        information_extension: InformationExtensionRef,
+        backend: KvBackendRef,
+        cache_registry: LayeredCacheRegistryRef,
+    ) -> Self {
+        Self {
+            information_extension,
+            backend,
+            cache_registry,
+            procedure_manager: None,
+            process_manager: None,
+            #[cfg(feature = "enterprise")]
+            extra_information_table_factories: std::collections::HashMap::new(),
+        }
+    }
+
+    pub fn with_procedure_manager(mut self, procedure_manager: ProcedureManagerRef) -> Self {
+        self.procedure_manager = Some(procedure_manager);
+        self
+    }
+
+    pub fn with_process_manager(mut self, process_manager: ProcessManagerRef) -> Self {
+        self.process_manager = Some(process_manager);
+        self
+    }
+
+    /// Sets the extra information tables.
+    #[cfg(feature = "enterprise")]
+    pub fn with_extra_information_table_factories(
+        mut self,
+        factories: std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
+    ) -> Self {
+        self.extra_information_table_factories = factories;
+        self
+    }
+
+    pub fn build(self) -> Arc<KvBackendCatalogManager> {
+        let Self {
+            information_extension,
+            backend,
+            cache_registry,
+            procedure_manager,
+            process_manager,
+            #[cfg(feature = "enterprise")]
+            extra_information_table_factories,
+        } = self;
+        Arc::new_cyclic(|me| KvBackendCatalogManager {
+            information_extension,
+            partition_manager: Arc::new(PartitionRuleManager::new(
+                backend.clone(),
+                cache_registry
+                    .get()
+                    .expect("Failed to get table_route_cache"),
+            )),
+            table_metadata_manager: Arc::new(TableMetadataManager::new(backend.clone())),
+            system_catalog: SystemCatalog {
+                catalog_manager: me.clone(),
+                catalog_cache: Cache::new(CATALOG_CACHE_MAX_CAPACITY),
+                pg_catalog_cache: Cache::new(CATALOG_CACHE_MAX_CAPACITY),
+                information_schema_provider: {
+                    let provider = InformationSchemaProvider::new(
+                        DEFAULT_CATALOG_NAME.to_string(),
+                        me.clone(),
+                        Arc::new(FlowMetadataManager::new(backend.clone())),
+                        process_manager.clone(),
+                        backend.clone(),
+                    );
+                    #[cfg(feature = "enterprise")]
+                    let provider = provider
+                        .with_extra_table_factories(extra_information_table_factories.clone());
+                    Arc::new(provider)
+                },
+                pg_catalog_provider: Arc::new(PGCatalogProvider::new(
+                    DEFAULT_CATALOG_NAME.to_string(),
+                    me.clone(),
+                )),
+                backend,
+                process_manager,
+                #[cfg(feature = "enterprise")]
+                extra_information_table_factories,
+            },
+            cache_registry,
+            procedure_manager,
+        })
+    }
+}

src/catalog/src/kvbackend/manager.rs

@@ -22,21 +22,24 @@ use common_catalog::consts::{
     PG_CATALOG_NAME,
 };
 use common_error::ext::BoxedError;
-use common_meta::cache::{LayeredCacheRegistryRef, ViewInfoCacheRef};
+use common_meta::cache::{
+    LayeredCacheRegistryRef, TableRoute, TableRouteCacheRef, ViewInfoCacheRef,
+};
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::schema_name::SchemaNameKey;
-use common_meta::key::table_info::TableInfoValue;
+use common_meta::key::table_info::{TableInfoManager, TableInfoValue};
 use common_meta::key::table_name::TableNameKey;
-use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
+use common_meta::key::TableMetadataManagerRef;
 use common_meta::kv_backend::KvBackendRef;
 use common_procedure::ProcedureManagerRef;
 use futures_util::stream::BoxStream;
 use futures_util::{StreamExt, TryStreamExt};
 use moka::sync::Cache;
-use partition::manager::{PartitionRuleManager, PartitionRuleManagerRef};
+use partition::manager::PartitionRuleManagerRef;
 use session::context::{Channel, QueryContext};
 use snafu::prelude::*;
+use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
 use table::dist_table::DistTable;
 use table::metadata::TableId;
 use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
@@ -49,6 +52,8 @@ use crate::error::{
     CacheNotFoundSnafu, GetTableCacheSnafu, InvalidTableInfoInCatalogSnafu, ListCatalogsSnafu,
     ListSchemasSnafu, ListTablesSnafu, Result, TableMetadataManagerSnafu,
 };
+#[cfg(feature = "enterprise")]
+use crate::information_schema::InformationSchemaTableFactoryRef;
 use crate::information_schema::{InformationExtensionRef, InformationSchemaProvider};
 use crate::kvbackend::TableCacheRef;
 use crate::process_manager::ProcessManagerRef;
@@ -64,60 +69,22 @@ use crate::CatalogManager;
 #[derive(Clone)]
 pub struct KvBackendCatalogManager {
     /// Provides the extension methods for the `information_schema` tables
-    information_extension: InformationExtensionRef,
+    pub(super) information_extension: InformationExtensionRef,
     /// Manages partition rules.
-    partition_manager: PartitionRuleManagerRef,
+    pub(super) partition_manager: PartitionRuleManagerRef,
     /// Manages table metadata.
-    table_metadata_manager: TableMetadataManagerRef,
+    pub(super) table_metadata_manager: TableMetadataManagerRef,
     /// A sub-CatalogManager that handles system tables
-    system_catalog: SystemCatalog,
+    pub(super) system_catalog: SystemCatalog,
     /// Cache registry for all caches.
-    cache_registry: LayeredCacheRegistryRef,
+    pub(super) cache_registry: LayeredCacheRegistryRef,
     /// Only available in `Standalone` mode.
-    procedure_manager: Option<ProcedureManagerRef>,
+    pub(super) procedure_manager: Option<ProcedureManagerRef>,
 }
 
-const CATALOG_CACHE_MAX_CAPACITY: u64 = 128;
+pub(super) const CATALOG_CACHE_MAX_CAPACITY: u64 = 128;
 
 impl KvBackendCatalogManager {
-    pub fn new(
-        information_extension: InformationExtensionRef,
-        backend: KvBackendRef,
-        cache_registry: LayeredCacheRegistryRef,
-        procedure_manager: Option<ProcedureManagerRef>,
-        process_manager: Option<ProcessManagerRef>,
-    ) -> Arc<Self> {
-        Arc::new_cyclic(|me| Self {
-            information_extension,
-            partition_manager: Arc::new(PartitionRuleManager::new(
-                backend.clone(),
-                cache_registry
-                    .get()
-                    .expect("Failed to get table_route_cache"),
-            )),
-            table_metadata_manager: Arc::new(TableMetadataManager::new(backend.clone())),
-            system_catalog: SystemCatalog {
-                catalog_manager: me.clone(),
-                catalog_cache: Cache::new(CATALOG_CACHE_MAX_CAPACITY),
-                pg_catalog_cache: Cache::new(CATALOG_CACHE_MAX_CAPACITY),
-                information_schema_provider: Arc::new(InformationSchemaProvider::new(
-                    DEFAULT_CATALOG_NAME.to_string(),
-                    me.clone(),
-                    Arc::new(FlowMetadataManager::new(backend.clone())),
-                    process_manager.clone(),
-                )),
-                pg_catalog_provider: Arc::new(PGCatalogProvider::new(
-                    DEFAULT_CATALOG_NAME.to_string(),
-                    me.clone(),
-                )),
-                backend,
-                process_manager,
-            },
-            cache_registry,
-            procedure_manager,
-        })
-    }
-
     pub fn view_info_cache(&self) -> Result<ViewInfoCacheRef> {
         self.cache_registry.get().context(CacheNotFoundSnafu {
             name: "view_info_cache",
@@ -140,6 +107,61 @@ impl KvBackendCatalogManager {
     pub fn procedure_manager(&self) -> Option<ProcedureManagerRef> {
         self.procedure_manager.clone()
     }
+
+    // Override logical table's partition key indices with physical table's.
+    async fn override_logical_table_partition_key_indices(
+        table_route_cache: &TableRouteCacheRef,
+        table_info_manager: &TableInfoManager,
+        table: TableRef,
+    ) -> Result<TableRef> {
+        // If the table is not a metric table, return the table directly.
+        if table.table_info().meta.engine != METRIC_ENGINE_NAME {
+            return Ok(table);
+        }
+
+        if let Some(table_route_value) = table_route_cache
+            .get(table.table_info().table_id())
+            .await
+            .context(TableMetadataManagerSnafu)?
+            && let TableRoute::Logical(logical_route) = &*table_route_value
+            && let Some(physical_table_info_value) = table_info_manager
+                .get(logical_route.physical_table_id())
+                .await
+                .context(TableMetadataManagerSnafu)?
+        {
+            let mut new_table_info = (*table.table_info()).clone();
+
+            // Remap partition key indices from physical table to logical table
+            new_table_info.meta.partition_key_indices = physical_table_info_value
+                .table_info
+                .meta
+                .partition_key_indices
+                .iter()
+                .filter_map(|&physical_index| {
+                    // Get the column name from the physical table using the physical index
+                    physical_table_info_value
+                        .table_info
+                        .meta
+                        .schema
+                        .column_schemas
+                        .get(physical_index)
+                        .and_then(|physical_column| {
+                            // Find the corresponding index in the logical table schema
+                            new_table_info
+                                .meta
+                                .schema
+                                .column_index_by_name(physical_column.name.as_str())
+                        })
+                })
+                .collect();
+
+            let new_table = DistTable::table(Arc::new(new_table_info));
+
+            return Ok(new_table);
+        }
+
+        Ok(table)
+    }
 }
 
 #[async_trait::async_trait]
@@ -266,16 +288,28 @@ impl CatalogManager for KvBackendCatalogManager {
         let table_cache: TableCacheRef = self.cache_registry.get().context(CacheNotFoundSnafu {
             name: "table_cache",
         })?;
-        if let Some(table) = table_cache
+
+        let table = table_cache
             .get_by_ref(&TableName {
                 catalog_name: catalog_name.to_string(),
                 schema_name: schema_name.to_string(),
                 table_name: table_name.to_string(),
             })
             .await
-            .context(GetTableCacheSnafu)?
+            .context(GetTableCacheSnafu)?;
-        {
+
-            return Ok(Some(table));
+        if let Some(table) = table {
+            let table_route_cache: TableRouteCacheRef =
+                self.cache_registry.get().context(CacheNotFoundSnafu {
+                    name: "table_route_cache",
+                })?;
+            return Self::override_logical_table_partition_key_indices(
+                &table_route_cache,
+                self.table_metadata_manager.table_info_manager(),
+                table,
+            )
+            .await
+            .map(Some);
         }
 
         if channel == Channel::Postgres {
@@ -288,7 +322,7 @@ impl CatalogManager for KvBackendCatalogManager {
             }
         }
 
-        return Ok(None);
+        Ok(None)
     }
 
     async fn tables_by_ids(
@@ -340,8 +374,20 @@ impl CatalogManager for KvBackendCatalogManager {
         let catalog = catalog.to_string();
         let schema = schema.to_string();
         let semaphore = Arc::new(Semaphore::new(CONCURRENCY));
+        let table_route_cache: Result<TableRouteCacheRef> =
+            self.cache_registry.get().context(CacheNotFoundSnafu {
+                name: "table_route_cache",
+            });
 
         common_runtime::spawn_global(async move {
+            let table_route_cache = match table_route_cache {
+                Ok(table_route_cache) => table_route_cache,
+                Err(e) => {
+                    let _ = tx.send(Err(e)).await;
+                    return;
+                }
+            };
+
             let table_id_stream = metadata_manager
                 .table_name_manager()
                 .tables(&catalog, &schema)
@@ -368,6 +414,7 @@ impl CatalogManager for KvBackendCatalogManager {
                 let metadata_manager = metadata_manager.clone();
                 let tx = tx.clone();
                 let semaphore = semaphore.clone();
+                let table_route_cache = table_route_cache.clone();
                 common_runtime::spawn_global(async move {
                     // we don't explicitly close the semaphore so just ignore the potential error.
                     let _ = semaphore.acquire().await;
@@ -385,6 +432,16 @@ impl CatalogManager for KvBackendCatalogManager {
                     };
 
                     for table in table_info_values.into_values().map(build_table) {
+                        let table = if let Ok(table) = table {
+                            Self::override_logical_table_partition_key_indices(
+                                &table_route_cache,
+                                metadata_manager.table_info_manager(),
+                                table,
+                            )
+                            .await
+                        } else {
+                            table
+                        };
                         if tx.send(table).await.is_err() {
                             return;
                         }
@@ -414,16 +471,19 @@ fn build_table(table_info_value: TableInfoValue) -> Result<TableRef> {
 /// - information_schema.{tables}
 /// - pg_catalog.{tables}
 #[derive(Clone)]
-struct SystemCatalog {
+pub(super) struct SystemCatalog {
-    catalog_manager: Weak<KvBackendCatalogManager>,
+    pub(super) catalog_manager: Weak<KvBackendCatalogManager>,
-    catalog_cache: Cache<String, Arc<InformationSchemaProvider>>,
+    pub(super) catalog_cache: Cache<String, Arc<InformationSchemaProvider>>,
-    pg_catalog_cache: Cache<String, Arc<PGCatalogProvider>>,
+    pub(super) pg_catalog_cache: Cache<String, Arc<PGCatalogProvider>>,
 
     // system_schema_provider for default catalog
-    information_schema_provider: Arc<InformationSchemaProvider>,
+    pub(super) information_schema_provider: Arc<InformationSchemaProvider>,
-    pg_catalog_provider: Arc<PGCatalogProvider>,
+    pub(super) pg_catalog_provider: Arc<PGCatalogProvider>,
-    backend: KvBackendRef,
+    pub(super) backend: KvBackendRef,
-    process_manager: Option<ProcessManagerRef>,
+    pub(super) process_manager: Option<ProcessManagerRef>,
+    #[cfg(feature = "enterprise")]
+    pub(super) extra_information_table_factories:
+        std::collections::HashMap<String, InformationSchemaTableFactoryRef>,
 }
 
 impl SystemCatalog {
@@ -487,12 +547,17 @@ impl SystemCatalog {
         if schema == INFORMATION_SCHEMA_NAME {
             let information_schema_provider =
                 self.catalog_cache.get_with_by_ref(catalog, move || {
-                    Arc::new(InformationSchemaProvider::new(
+                    let provider = InformationSchemaProvider::new(
                         catalog.to_string(),
                         self.catalog_manager.clone(),
                         Arc::new(FlowMetadataManager::new(self.backend.clone())),
                         self.process_manager.clone(),
-                    ))
+                        self.backend.clone(),
+                    );
+                    #[cfg(feature = "enterprise")]
+                    let provider = provider
+                        .with_extra_table_factories(self.extra_information_table_factories.clone());
+                    Arc::new(provider)
                 });
             information_schema_provider.table(table_name)
         } else if schema == PG_CATALOG_NAME && channel == Channel::Postgres {

src/catalog/src/lib.rs

@@ -14,6 +14,7 @@
 
 #![feature(assert_matches)]
 #![feature(try_blocks)]
+#![feature(let_chains)]
 
 use std::any::Any;
 use std::fmt::{Debug, Formatter};

src/catalog/src/memory/manager.rs

@@ -352,11 +352,13 @@ impl MemoryCatalogManager {
     }
 
     fn create_catalog_entry(self: &Arc<Self>, catalog: String) -> SchemaEntries {
+        let backend = Arc::new(MemoryKvBackend::new());
         let information_schema_provider = InformationSchemaProvider::new(
             catalog,
             Arc::downgrade(self) as Weak<dyn CatalogManager>,
-            Arc::new(FlowMetadataManager::new(Arc::new(MemoryKvBackend::new()))),
+            Arc::new(FlowMetadataManager::new(backend.clone())),
             None, // we don't need ProcessManager on regions server.
+            backend,
         );
         let information_schema = information_schema_provider.tables().clone();
 

src/catalog/src/process_manager.rs

@@ -15,13 +15,13 @@
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::fmt::{Debug, Formatter};
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
 
 use api::v1::frontend::{KillProcessRequest, ListProcessRequest, ProcessInfo};
 use common_base::cancellation::CancellationHandle;
 use common_frontend::selector::{FrontendSelector, MetaClientSelector};
-use common_telemetry::{debug, info};
+use common_telemetry::{debug, info, warn};
 use common_time::util::current_time_millis;
 use meta_client::MetaClientRef;
 use snafu::{ensure, OptionExt, ResultExt};
@@ -29,6 +29,7 @@ use snafu::{ensure, OptionExt, ResultExt};
 use crate::error;
 use crate::metrics::{PROCESS_KILL_COUNT, PROCESS_LIST_COUNT};
 
+pub type ProcessId = u32;
 pub type ProcessManagerRef = Arc<ProcessManager>;
 
 /// Query process manager.
@@ -36,9 +37,9 @@ pub struct ProcessManager {
     /// Local frontend server address,
     server_addr: String,
     /// Next process id for local queries.
-    next_id: AtomicU64,
+    next_id: AtomicU32,
     /// Running process per catalog.
-    catalogs: RwLock<HashMap<String, HashMap<u64, CancellableProcess>>>,
+    catalogs: RwLock<HashMap<String, HashMap<ProcessId, CancellableProcess>>>,
     /// Frontend selector to locate frontend nodes.
     frontend_selector: Option<MetaClientSelector>,
 }
@@ -65,9 +66,9 @@ impl ProcessManager {
         schemas: Vec<String>,
         query: String,
         client: String,
-        id: Option<u64>,
+        query_id: Option<ProcessId>,
     ) -> Ticket {
-        let id = id.unwrap_or_else(|| self.next_id.fetch_add(1, Ordering::Relaxed));
+        let id = query_id.unwrap_or_else(|| self.next_id.fetch_add(1, Ordering::Relaxed));
         let process = ProcessInfo {
             id,
             catalog: catalog.clone(),
@@ -96,12 +97,12 @@ impl ProcessManager {
     }
 
     /// Generates the next process id.
-    pub fn next_id(&self) -> u64 {
+    pub fn next_id(&self) -> u32 {
         self.next_id.fetch_add(1, Ordering::Relaxed)
     }
 
     /// De-register a query from process list.
-    pub fn deregister_query(&self, catalog: String, id: u64) {
+    pub fn deregister_query(&self, catalog: String, id: ProcessId) {
         if let Entry::Occupied(mut o) = self.catalogs.write().unwrap().entry(catalog) {
             let process = o.get_mut().remove(&id);
             debug!("Deregister process: {:?}", process);
@@ -140,14 +141,20 @@ impl ProcessManager {
                 .await
                 .context(error::InvokeFrontendSnafu)?;
             for mut f in frontends {
-                processes.extend(
+                let result = f
-                    f.list_process(ListProcessRequest {
+                    .list_process(ListProcessRequest {
                         catalog: catalog.unwrap_or_default().to_string(),
                     })
                     .await
-                    .context(error::InvokeFrontendSnafu)?
+                    .context(error::InvokeFrontendSnafu);
-                    .processes,
+                match result {
-                );
+                    Ok(resp) => {
+                        processes.extend(resp.processes);
+                    }
+                    Err(e) => {
+                        warn!(e; "Skipping failing node: {:?}", f)
+                    }
+                }
             }
         }
         processes.extend(self.local_processes(catalog)?);
@@ -159,26 +166,10 @@ impl ProcessManager {
         &self,
         server_addr: String,
         catalog: String,
-        id: u64,
+        id: ProcessId,
     ) -> error::Result<bool> {
         if server_addr == self.server_addr {
-            if let Some(catalogs) = self.catalogs.write().unwrap().get_mut(&catalog) {
+            self.kill_local_process(catalog, id).await
-                if let Some(process) = catalogs.remove(&id) {
-                    process.handle.cancel();
-                    info!(
-                        "Killed process, catalog: {}, id: {:?}",
-                        process.process.catalog, process.process.id
-                    );
-                    PROCESS_KILL_COUNT.with_label_values(&[&catalog]).inc();
-                    Ok(true)
-                } else {
-                    debug!("Failed to kill process, id not found: {}", id);
-                    Ok(false)
-                }
-            } else {
-                debug!("Failed to kill process, catalog not found: {}", catalog);
-                Ok(false)
-            }
         } else {
             let mut nodes = self
                 .frontend_selector
@@ -204,12 +195,33 @@ impl ProcessManager {
             Ok(true)
         }
     }
+
+    /// Kills local query with provided catalog and id.
+    pub async fn kill_local_process(&self, catalog: String, id: ProcessId) -> error::Result<bool> {
+        if let Some(catalogs) = self.catalogs.write().unwrap().get_mut(&catalog) {
+            if let Some(process) = catalogs.remove(&id) {
+                process.handle.cancel();
+                info!(
+                    "Killed process, catalog: {}, id: {:?}",
+                    process.process.catalog, process.process.id
+                );
+                PROCESS_KILL_COUNT.with_label_values(&[&catalog]).inc();
+                Ok(true)
+            } else {
+                debug!("Failed to kill process, id not found: {}", id);
+                Ok(false)
+            }
+        } else {
+            debug!("Failed to kill process, catalog not found: {}", catalog);
+            Ok(false)
+        }
+    }
 }
 
 pub struct Ticket {
     pub(crate) catalog: String,
     pub(crate) manager: ProcessManagerRef,
-    pub(crate) id: u64,
+    pub(crate) id: ProcessId,
     pub cancellation_handle: Arc<CancellationHandle>,
 }
 
@@ -323,7 +335,7 @@ mod tests {
         assert_eq!(running_processes.len(), 2);
 
         // Verify both processes are present
-        let ids: Vec<u64> = running_processes.iter().map(|p| p.id).collect();
+        let ids: Vec<u32> = running_processes.iter().map(|p| p.id).collect();
         assert!(ids.contains(&ticket1.id));
         assert!(ids.contains(&ticket2.id));
     }

src/catalog/src/system_schema.rs

@@ -15,7 +15,7 @@
 pub mod information_schema;
 mod memory_table;
 pub mod pg_catalog;
-mod predicate;
+pub mod predicate;
 mod utils;
 
 use std::collections::HashMap;
@@ -96,7 +96,7 @@ trait SystemSchemaProviderInner {
     }
 }
 
-pub(crate) trait SystemTable {
+pub trait SystemTable {
     fn table_id(&self) -> TableId;
 
     fn table_name(&self) -> &'static str;
@@ -110,7 +110,7 @@ pub(crate) trait SystemTable {
     }
 }
 
-pub(crate) type SystemTableRef = Arc<dyn SystemTable + Send + Sync>;
+pub type SystemTableRef = Arc<dyn SystemTable + Send + Sync>;
 
 struct SystemTableDataSource {
     table: SystemTableRef,

src/catalog/src/system_schema/information_schema.rs

@@ -19,7 +19,7 @@ mod information_memory_table;
 pub mod key_column_usage;
 mod partitions;
 mod procedure_info;
-mod process_list;
+pub mod process_list;
 pub mod region_peers;
 mod region_statistics;
 mod runtime_metrics;
@@ -38,6 +38,7 @@ use common_meta::cluster::NodeInfo;
 use common_meta::datanode::RegionStat;
 use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
+use common_meta::kv_backend::KvBackendRef;
 use common_procedure::ProcedureInfo;
 use common_recordbatch::SendableRecordBatchStream;
 use datatypes::schema::SchemaRef;
@@ -112,6 +113,25 @@ macro_rules! setup_memory_table {
     };
 }
 
+#[cfg(feature = "enterprise")]
+pub struct MakeInformationTableRequest {
+    pub catalog_name: String,
+    pub catalog_manager: Weak<dyn CatalogManager>,
+    pub kv_backend: KvBackendRef,
+}
+
+/// A factory trait for making information schema tables.
+///
+/// This trait allows for extensibility of the information schema by providing
+/// a way to dynamically create custom information schema tables.
+#[cfg(feature = "enterprise")]
+pub trait InformationSchemaTableFactory {
+    fn make_information_table(&self, req: MakeInformationTableRequest) -> SystemTableRef;
+}
+
+#[cfg(feature = "enterprise")]
+pub type InformationSchemaTableFactoryRef = Arc<dyn InformationSchemaTableFactory + Send + Sync>;
+
 /// The `information_schema` tables info provider.
 pub struct InformationSchemaProvider {
     catalog_name: String,
@@ -119,6 +139,10 @@ pub struct InformationSchemaProvider {
     process_manager: Option<ProcessManagerRef>,
     flow_metadata_manager: Arc<FlowMetadataManager>,
     tables: HashMap<String, TableRef>,
+    #[allow(dead_code)]
+    kv_backend: KvBackendRef,
+    #[cfg(feature = "enterprise")]
+    extra_table_factories: HashMap<String, InformationSchemaTableFactoryRef>,
 }
 
 impl SystemSchemaProvider for InformationSchemaProvider {
@@ -128,6 +152,7 @@ impl SystemSchemaProvider for InformationSchemaProvider {
         &self.tables
     }
 }
+
 impl SystemSchemaProviderInner for InformationSchemaProvider {
     fn catalog_name(&self) -> &str {
         &self.catalog_name
@@ -215,7 +240,22 @@ impl SystemSchemaProviderInner for InformationSchemaProvider {
                 .process_manager
                 .as_ref()
                 .map(|p| Arc::new(InformationSchemaProcessList::new(p.clone())) as _),
-            _ => None,
+            table_name => {
+                #[cfg(feature = "enterprise")]
+                return self.extra_table_factories.get(table_name).map(|factory| {
+                    let req = MakeInformationTableRequest {
+                        catalog_name: self.catalog_name.clone(),
+                        catalog_manager: self.catalog_manager.clone(),
+                        kv_backend: self.kv_backend.clone(),
+                    };
+                    factory.make_information_table(req)
+                });
+                #[cfg(not(feature = "enterprise"))]
+                {
+                    let _ = table_name;
+                    None
+                }
+            }
         }
     }
 }
@@ -226,6 +266,7 @@ impl InformationSchemaProvider {
         catalog_manager: Weak<dyn CatalogManager>,
         flow_metadata_manager: Arc<FlowMetadataManager>,
         process_manager: Option<ProcessManagerRef>,
+        kv_backend: KvBackendRef,
     ) -> Self {
         let mut provider = Self {
             catalog_name,
@@ -233,6 +274,9 @@ impl InformationSchemaProvider {
             flow_metadata_manager,
             process_manager,
             tables: HashMap::new(),
+            kv_backend,
+            #[cfg(feature = "enterprise")]
+            extra_table_factories: HashMap::new(),
         };
 
         provider.build_tables();
@@ -240,6 +284,16 @@ impl InformationSchemaProvider {
         provider
     }
 
+    #[cfg(feature = "enterprise")]
+    pub(crate) fn with_extra_table_factories(
+        mut self,
+        factories: HashMap<String, InformationSchemaTableFactoryRef>,
+    ) -> Self {
+        self.extra_table_factories = factories;
+        self.build_tables();
+        self
+    }
+
     fn build_tables(&mut self) {
         let mut tables = HashMap::new();
 
@@ -290,16 +344,19 @@ impl InformationSchemaProvider {
         if let Some(process_list) = self.build_table(PROCESS_LIST) {
             tables.insert(PROCESS_LIST.to_string(), process_list);
         }
+        #[cfg(feature = "enterprise")]
+        for name in self.extra_table_factories.keys() {
+            tables.insert(name.to_string(), self.build_table(name).expect(name));
+        }
         // Add memory tables
         for name in MEMORY_TABLES.iter() {
             tables.insert((*name).to_string(), self.build_table(name).expect(name));
         }
-
         self.tables = tables;
     }
 }
 
-trait InformationTable {
+pub trait InformationTable {
     fn table_id(&self) -> TableId;
 
     fn table_name(&self) -> &'static str;

src/catalog/src/system_schema/information_schema/process_list.rs

@@ -39,14 +39,14 @@ use crate::process_manager::ProcessManagerRef;
 use crate::system_schema::information_schema::InformationTable;
 
 /// Column names of `information_schema.process_list`
-const ID: &str = "id";
+pub const ID: &str = "id";
-const CATALOG: &str = "catalog";
+pub const CATALOG: &str = "catalog";
-const SCHEMAS: &str = "schemas";
+pub const SCHEMAS: &str = "schemas";
-const QUERY: &str = "query";
+pub const QUERY: &str = "query";
-const CLIENT: &str = "client";
+pub const CLIENT: &str = "client";
-const FRONTEND: &str = "frontend";
+pub const FRONTEND: &str = "frontend";
-const START_TIMESTAMP: &str = "start_timestamp";
+pub const START_TIMESTAMP: &str = "start_timestamp";
-const ELAPSED_TIME: &str = "elapsed_time";
+pub const ELAPSED_TIME: &str = "elapsed_time";
 
 /// `information_schema.process_list` table implementation that tracks running
 /// queries in current cluster.

src/catalog/src/system_schema/information_schema/table_names.rs

@@ -48,3 +48,4 @@ pub const FLOWS: &str = "flows";
 pub const PROCEDURE_INFO: &str = "procedure_info";
 pub const REGION_STATISTICS: &str = "region_statistics";
 pub const PROCESS_LIST: &str = "process_list";
+pub const TRIGGER_LIST: &str = "trigger_list";

src/catalog/src/table_source.rs

@@ -207,6 +207,7 @@ mod tests {
     use session::context::QueryContext;
 
     use super::*;
+    use crate::kvbackend::KvBackendCatalogManagerBuilder;
     use crate::memory::MemoryCatalogManager;
 
     #[test]
@@ -323,13 +324,13 @@ mod tests {
             .build(),
         );
 
-        let catalog_manager = KvBackendCatalogManager::new(
+        let catalog_manager = KvBackendCatalogManagerBuilder::new(
             Arc::new(NoopInformationExtension),
             backend.clone(),
             layered_cache_registry,
-            None,
+        )
-            None,
+        .build();
-        );
+
         let table_metadata_manager = TableMetadataManager::new(backend);
         let mut view_info = common_meta::key::test_utils::new_test_table_info(1024, vec![]);
         view_info.table_type = TableType::View;

src/cli/Cargo.toml

@@ -16,6 +16,7 @@ mysql_kvbackend = ["common-meta/mysql_kvbackend", "meta-srv/mysql_kvbackend"]
 workspace = true
 
 [dependencies]
+async-stream.workspace = true
 async-trait.workspace = true
 auth.workspace = true
 base64.workspace = true
@@ -42,7 +43,6 @@ common-time.workspace = true
 common-version.workspace = true
 common-wal.workspace = true
 datatypes.workspace = true
-either = "1.8"
 etcd-client.workspace = true
 futures.workspace = true
 humantime.workspace = true
@@ -50,6 +50,7 @@ meta-client.workspace = true
 meta-srv.workspace = true
 nu-ansi-term = "0.46"
 object-store.workspace = true
+operator.workspace = true
 query.workspace = true
 rand.workspace = true
 reqwest.workspace = true
@@ -65,6 +66,7 @@ tokio.workspace = true
 tracing-appender.workspace = true
 
 [dev-dependencies]
+common-meta = { workspace = true, features = ["testing"] }
 common-version.workspace = true
 serde.workspace = true
 tempfile.workspace = true

src/cli/src/bench.rs

@@ -160,6 +160,7 @@ fn create_table_info(table_id: TableId, table_name: TableName) -> RawTableInfo {
         options: Default::default(),
         region_numbers: (1..=100).collect(),
         partition_key_indices: vec![],
+        column_ids: vec![],
     };
 
     RawTableInfo {

src/cli/src/error.rs

@@ -17,8 +17,10 @@ use std::any::Any;
 use common_error::ext::{BoxedError, ErrorExt};
 use common_error::status_code::StatusCode;
 use common_macro::stack_trace_debug;
+use common_meta::peer::Peer;
 use object_store::Error as ObjectStoreError;
 use snafu::{Location, Snafu};
+use store_api::storage::TableId;
 
 #[derive(Snafu)]
 #[snafu(visibility(pub))]
@@ -73,6 +75,20 @@ pub enum Error {
         source: common_meta::error::Error,
     },
 
+    #[snafu(display("Failed to get table metadata"))]
+    TableMetadata {
+        #[snafu(implicit)]
+        location: Location,
+        source: common_meta::error::Error,
+    },
+
+    #[snafu(display("Unexpected error: {}", msg))]
+    Unexpected {
+        msg: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Missing config, msg: {}", msg))]
     MissingConfig {
         msg: String,
@@ -222,6 +238,13 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Table not found: {table_id}"))]
+    TableNotFound {
+        table_id: TableId,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("OpenDAL operator failed"))]
     OpenDal {
         #[snafu(implicit)]
@@ -267,6 +290,29 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Failed to init backend"))]
+    InitBackend {
+        #[snafu(implicit)]
+        location: Location,
+        #[snafu(source)]
+        error: ObjectStoreError,
+    },
+
+    #[snafu(display("Covert column schemas to defs failed"))]
+    CovertColumnSchemasToDefs {
+        #[snafu(implicit)]
+        location: Location,
+        source: operator::error::Error,
+    },
+
+    #[snafu(display("Failed to send request to datanode: {}", peer))]
+    SendRequestToDatanode {
+        peer: Peer,
+        #[snafu(implicit)]
+        location: Location,
+        source: common_meta::error::Error,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -274,9 +320,9 @@ pub type Result<T> = std::result::Result<T, Error>;
 impl ErrorExt for Error {
     fn status_code(&self) -> StatusCode {
         match self {
-            Error::InitMetadata { source, .. } | Error::InitDdlManager { source, .. } => {
+            Error::InitMetadata { source, .. }
-                source.status_code()
+            | Error::InitDdlManager { source, .. }
-            }
+            | Error::TableMetadata { source, .. } => source.status_code(),
 
             Error::MissingConfig { .. }
             | Error::LoadLayeredConfig { .. }
@@ -290,6 +336,9 @@ impl ErrorExt for Error {
             | Error::InvalidArguments { .. }
             | Error::ParseProxyOpts { .. } => StatusCode::InvalidArguments,
 
+            Error::CovertColumnSchemasToDefs { source, .. } => source.status_code(),
+            Error::SendRequestToDatanode { source, .. } => source.status_code(),
+
             Error::StartProcedureManager { source, .. }
             | Error::StopProcedureManager { source, .. } => source.status_code(),
             Error::StartWalOptionsAllocator { source, .. } => source.status_code(),
@@ -297,6 +346,7 @@ impl ErrorExt for Error {
             Error::ParseSql { source, .. } | Error::PlanStatement { source, .. } => {
                 source.status_code()
             }
+            Error::Unexpected { .. } => StatusCode::Unexpected,
 
             Error::SerdeJson { .. }
             | Error::FileIo { .. }
@@ -305,7 +355,7 @@ impl ErrorExt for Error {
             | Error::BuildClient { .. } => StatusCode::Unexpected,
 
             Error::Other { source, .. } => source.status_code(),
-            Error::OpenDal { .. } => StatusCode::Internal,
+            Error::OpenDal { .. } | Error::InitBackend { .. } => StatusCode::Internal,
             Error::S3ConfigNotSet { .. }
             | Error::OutputDirNotSet { .. }
             | Error::EmptyStoreAddrs { .. } => StatusCode::InvalidArguments,
@@ -314,6 +364,7 @@ impl ErrorExt for Error {
 
             Error::CacheRequired { .. } | Error::BuildCacheRegistry { .. } => StatusCode::Internal,
             Error::MetaClientInit { source, .. } => source.status_code(),
+            Error::TableNotFound { .. } => StatusCode::TableNotFound,
             Error::SchemaNotFound { .. } => StatusCode::DatabaseNotFound,
         }
     }

src/cli/src/lib.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![allow(clippy::print_stdout)]
 mod bench;
 mod data;
 mod database;

src/cli/src/metadata.rs

@@ -14,29 +14,39 @@
 
 mod common;
 mod control;
+mod repair;
 mod snapshot;
+mod utils;
 
 use clap::Subcommand;
 use common_error::ext::BoxedError;
 
-use crate::metadata::control::ControlCommand;
+use crate::metadata::control::{DelCommand, GetCommand};
+use crate::metadata::repair::RepairLogicalTablesCommand;
 use crate::metadata::snapshot::SnapshotCommand;
 use crate::Tool;
 
-/// Command for managing metadata operations, including saving metadata snapshots and restoring metadata from snapshots.
+/// Command for managing metadata operations,
+/// including saving and restoring metadata snapshots,
+/// controlling metadata operations, and diagnosing and repairing metadata.
 #[derive(Subcommand)]
 pub enum MetadataCommand {
     #[clap(subcommand)]
     Snapshot(SnapshotCommand),
     #[clap(subcommand)]
-    Control(ControlCommand),
+    Get(GetCommand),
+    #[clap(subcommand)]
+    Del(DelCommand),
+    RepairLogicalTables(RepairLogicalTablesCommand),
 }
 
 impl MetadataCommand {
     pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
         match self {
             MetadataCommand::Snapshot(cmd) => cmd.build().await,
-            MetadataCommand::Control(cmd) => cmd.build().await,
+            MetadataCommand::RepairLogicalTables(cmd) => cmd.build().await,
+            MetadataCommand::Get(cmd) => cmd.build().await,
+            MetadataCommand::Del(cmd) => cmd.build().await,
         }
     }
 }

src/cli/src/metadata/control.rs

@@ -12,27 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod del;
 mod get;
+#[cfg(test)]
+mod test_utils;
 mod utils;
 
-use clap::Subcommand;
+pub(crate) use del::DelCommand;
-use common_error::ext::BoxedError;
+pub(crate) use get::GetCommand;
-use get::GetCommand;
-
-use crate::Tool;
-
-/// Subcommand for metadata control.
-#[derive(Subcommand)]
-pub enum ControlCommand {
-    /// Get the metadata from the metasrv.
-    #[clap(subcommand)]
-    Get(GetCommand),
-}
-
-impl ControlCommand {
-    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
-        match self {
-            ControlCommand::Get(cmd) => cmd.build().await,
-        }
-    }
-}

src/cli/src/metadata/control/del.rs

@@ -0,0 +1,42 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod key;
+mod table;
+
+use clap::Subcommand;
+use common_error::ext::BoxedError;
+
+use crate::metadata::control::del::key::DelKeyCommand;
+use crate::metadata::control::del::table::DelTableCommand;
+use crate::Tool;
+
+/// The prefix of the tombstone keys.
+pub(crate) const CLI_TOMBSTONE_PREFIX: &str = "__cli_tombstone/";
+
+/// Subcommand for deleting metadata from the metadata store.
+#[derive(Subcommand)]
+pub enum DelCommand {
+    Key(DelKeyCommand),
+    Table(DelTableCommand),
+}
+
+impl DelCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        match self {
+            DelCommand::Key(cmd) => cmd.build().await,
+            DelCommand::Table(cmd) => cmd.build().await,
+        }
+    }
+}

src/cli/src/metadata/control/del/key.rs

@@ -0,0 +1,132 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use async_trait::async_trait;
+use clap::Parser;
+use common_error::ext::BoxedError;
+use common_meta::key::tombstone::TombstoneManager;
+use common_meta::kv_backend::KvBackendRef;
+use common_meta::rpc::store::RangeRequest;
+
+use crate::metadata::common::StoreConfig;
+use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
+use crate::Tool;
+
+/// Delete key-value pairs logically from the metadata store.
+#[derive(Debug, Default, Parser)]
+pub struct DelKeyCommand {
+    /// The key to delete from the metadata store.
+    key: String,
+
+    /// Delete key-value pairs with the given prefix.
+    #[clap(long)]
+    prefix: bool,
+
+    #[clap(flatten)]
+    store: StoreConfig,
+}
+
+impl DelKeyCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        let kv_backend = self.store.build().await?;
+        Ok(Box::new(DelKeyTool {
+            key: self.key.to_string(),
+            prefix: self.prefix,
+            key_deleter: KeyDeleter::new(kv_backend),
+        }))
+    }
+}
+
+struct KeyDeleter {
+    kv_backend: KvBackendRef,
+    tombstone_manager: TombstoneManager,
+}
+
+impl KeyDeleter {
+    fn new(kv_backend: KvBackendRef) -> Self {
+        Self {
+            kv_backend: kv_backend.clone(),
+            tombstone_manager: TombstoneManager::new_with_prefix(kv_backend, CLI_TOMBSTONE_PREFIX),
+        }
+    }
+
+    async fn delete(&self, key: &str, prefix: bool) -> Result<usize, BoxedError> {
+        let mut req = RangeRequest::default().with_keys_only();
+        if prefix {
+            req = req.with_prefix(key.as_bytes());
+        } else {
+            req = req.with_key(key.as_bytes());
+        }
+        let resp = self.kv_backend.range(req).await.map_err(BoxedError::new)?;
+        let keys = resp.kvs.iter().map(|kv| kv.key.clone()).collect::<Vec<_>>();
+        self.tombstone_manager
+            .create(keys)
+            .await
+            .map_err(BoxedError::new)
+    }
+}
+
+struct DelKeyTool {
+    key: String,
+    prefix: bool,
+    key_deleter: KeyDeleter,
+}
+
+#[async_trait]
+impl Tool for DelKeyTool {
+    async fn do_work(&self) -> Result<(), BoxedError> {
+        let deleted = self.key_deleter.delete(&self.key, self.prefix).await?;
+        // Print the number of deleted keys.
+        println!("{}", deleted);
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_meta::kv_backend::chroot::ChrootKvBackend;
+    use common_meta::kv_backend::memory::MemoryKvBackend;
+    use common_meta::kv_backend::{KvBackend, KvBackendRef};
+    use common_meta::rpc::store::RangeRequest;
+
+    use crate::metadata::control::del::key::KeyDeleter;
+    use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
+    use crate::metadata::control::test_utils::put_key;
+
+    #[tokio::test]
+    async fn test_delete_keys() {
+        let kv_backend = Arc::new(MemoryKvBackend::new()) as KvBackendRef;
+        let key_deleter = KeyDeleter::new(kv_backend.clone());
+        put_key(&kv_backend, "foo", "bar").await;
+        put_key(&kv_backend, "foo/bar", "baz").await;
+        put_key(&kv_backend, "foo/baz", "qux").await;
+        let deleted = key_deleter.delete("foo", true).await.unwrap();
+        assert_eq!(deleted, 3);
+        let deleted = key_deleter.delete("foo/bar", false).await.unwrap();
+        assert_eq!(deleted, 0);
+
+        let chroot = ChrootKvBackend::new(CLI_TOMBSTONE_PREFIX.as_bytes().to_vec(), kv_backend);
+        let req = RangeRequest::default().with_prefix(b"foo");
+        let resp = chroot.range(req).await.unwrap();
+        assert_eq!(resp.kvs.len(), 3);
+        assert_eq!(resp.kvs[0].key, b"foo");
+        assert_eq!(resp.kvs[0].value, b"bar");
+        assert_eq!(resp.kvs[1].key, b"foo/bar");
+        assert_eq!(resp.kvs[1].value, b"baz");
+        assert_eq!(resp.kvs[2].key, b"foo/baz");
+        assert_eq!(resp.kvs[2].value, b"qux");
+    }
+}

src/cli/src/metadata/control/del/table.rs

@@ -0,0 +1,235 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use async_trait::async_trait;
+use clap::Parser;
+use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
+use common_catalog::format_full_table_name;
+use common_error::ext::BoxedError;
+use common_meta::ddl::utils::get_region_wal_options;
+use common_meta::key::table_name::TableNameManager;
+use common_meta::key::TableMetadataManager;
+use common_meta::kv_backend::KvBackendRef;
+use store_api::storage::TableId;
+
+use crate::error::{InvalidArgumentsSnafu, TableNotFoundSnafu};
+use crate::metadata::common::StoreConfig;
+use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
+use crate::metadata::control::utils::get_table_id_by_name;
+use crate::Tool;
+
+/// Delete table metadata logically from the metadata store.
+#[derive(Debug, Default, Parser)]
+pub struct DelTableCommand {
+    /// The table id to delete from the metadata store.
+    #[clap(long)]
+    table_id: Option<u32>,
+
+    /// The table name to delete from the metadata store.
+    #[clap(long)]
+    table_name: Option<String>,
+
+    /// The schema name of the table.
+    #[clap(long, default_value = DEFAULT_SCHEMA_NAME)]
+    schema_name: String,
+
+    /// The catalog name of the table.
+    #[clap(long, default_value = DEFAULT_CATALOG_NAME)]
+    catalog_name: String,
+
+    #[clap(flatten)]
+    store: StoreConfig,
+}
+
+impl DelTableCommand {
+    fn validate(&self) -> Result<(), BoxedError> {
+        if matches!(
+            (&self.table_id, &self.table_name),
+            (Some(_), Some(_)) | (None, None)
+        ) {
+            return Err(BoxedError::new(
+                InvalidArgumentsSnafu {
+                    msg: "You must specify either --table-id or --table-name.",
+                }
+                .build(),
+            ));
+        }
+        Ok(())
+    }
+}
+
+impl DelTableCommand {
+    pub async fn build(&self) -> Result<Box<dyn Tool>, BoxedError> {
+        self.validate()?;
+        let kv_backend = self.store.build().await?;
+        Ok(Box::new(DelTableTool {
+            table_id: self.table_id,
+            table_name: self.table_name.clone(),
+            schema_name: self.schema_name.clone(),
+            catalog_name: self.catalog_name.clone(),
+            table_name_manager: TableNameManager::new(kv_backend.clone()),
+            table_metadata_deleter: TableMetadataDeleter::new(kv_backend),
+        }))
+    }
+}
+
+struct DelTableTool {
+    table_id: Option<u32>,
+    table_name: Option<String>,
+    schema_name: String,
+    catalog_name: String,
+    table_name_manager: TableNameManager,
+    table_metadata_deleter: TableMetadataDeleter,
+}
+
+#[async_trait]
+impl Tool for DelTableTool {
+    async fn do_work(&self) -> Result<(), BoxedError> {
+        let table_id = if let Some(table_name) = &self.table_name {
+            let catalog_name = &self.catalog_name;
+            let schema_name = &self.schema_name;
+
+            let Some(table_id) = get_table_id_by_name(
+                &self.table_name_manager,
+                catalog_name,
+                schema_name,
+                table_name,
+            )
+            .await?
+            else {
+                println!(
+                    "Table({}) not found",
+                    format_full_table_name(catalog_name, schema_name, table_name)
+                );
+                return Ok(());
+            };
+            table_id
+        } else {
+            // Safety: we have validated that table_id or table_name is not None
+            self.table_id.unwrap()
+        };
+        self.table_metadata_deleter.delete(table_id).await?;
+        println!("Table({}) deleted", table_id);
+
+        Ok(())
+    }
+}
+
+struct TableMetadataDeleter {
+    table_metadata_manager: TableMetadataManager,
+}
+
+impl TableMetadataDeleter {
+    fn new(kv_backend: KvBackendRef) -> Self {
+        Self {
+            table_metadata_manager: TableMetadataManager::new_with_custom_tombstone_prefix(
+                kv_backend,
+                CLI_TOMBSTONE_PREFIX,
+            ),
+        }
+    }
+
+    async fn delete(&self, table_id: TableId) -> Result<(), BoxedError> {
+        let (table_info, table_route) = self
+            .table_metadata_manager
+            .get_full_table_info(table_id)
+            .await
+            .map_err(BoxedError::new)?;
+        let Some(table_info) = table_info else {
+            return Err(BoxedError::new(TableNotFoundSnafu { table_id }.build()));
+        };
+        let Some(table_route) = table_route else {
+            return Err(BoxedError::new(TableNotFoundSnafu { table_id }.build()));
+        };
+        let physical_table_id = self
+            .table_metadata_manager
+            .table_route_manager()
+            .get_physical_table_id(table_id)
+            .await
+            .map_err(BoxedError::new)?;
+
+        let table_name = table_info.table_name();
+        let region_wal_options = get_region_wal_options(
+            &self.table_metadata_manager,
+            &table_route,
+            physical_table_id,
+        )
+        .await
+        .map_err(BoxedError::new)?;
+
+        self.table_metadata_manager
+            .delete_table_metadata(table_id, &table_name, &table_route, &region_wal_options)
+            .await
+            .map_err(BoxedError::new)?;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::HashMap;
+    use std::sync::Arc;
+
+    use common_error::ext::ErrorExt;
+    use common_error::status_code::StatusCode;
+    use common_meta::key::table_route::TableRouteValue;
+    use common_meta::key::TableMetadataManager;
+    use common_meta::kv_backend::chroot::ChrootKvBackend;
+    use common_meta::kv_backend::memory::MemoryKvBackend;
+    use common_meta::kv_backend::{KvBackend, KvBackendRef};
+    use common_meta::rpc::store::RangeRequest;
+
+    use crate::metadata::control::del::table::TableMetadataDeleter;
+    use crate::metadata::control::del::CLI_TOMBSTONE_PREFIX;
+    use crate::metadata::control::test_utils::prepare_physical_table_metadata;
+
+    #[tokio::test]
+    async fn test_delete_table_not_found() {
+        let kv_backend = Arc::new(MemoryKvBackend::new()) as KvBackendRef;
+
+        let table_metadata_deleter = TableMetadataDeleter::new(kv_backend);
+        let table_id = 1;
+        let err = table_metadata_deleter.delete(table_id).await.unwrap_err();
+        assert_eq!(err.status_code(), StatusCode::TableNotFound);
+    }
+
+    #[tokio::test]
+    async fn test_delete_table_metadata() {
+        let kv_backend = Arc::new(MemoryKvBackend::new());
+        let table_metadata_manager = TableMetadataManager::new(kv_backend.clone());
+        let table_id = 1024;
+        let (table_info, table_route) = prepare_physical_table_metadata("my_table", table_id).await;
+        table_metadata_manager
+            .create_table_metadata(
+                table_info,
+                TableRouteValue::Physical(table_route),
+                HashMap::new(),
+            )
+            .await
+            .unwrap();
+
+        let total_keys = kv_backend.len();
+        assert!(total_keys > 0);
+
+        let table_metadata_deleter = TableMetadataDeleter::new(kv_backend.clone());
+        table_metadata_deleter.delete(table_id).await.unwrap();
+
+        // Check the tombstone keys are deleted
+        let chroot =
+            ChrootKvBackend::new(CLI_TOMBSTONE_PREFIX.as_bytes().to_vec(), kv_backend.clone());
+        let req = RangeRequest::default().with_range(vec![0], vec![0]);
+        let resp = chroot.range(req).await.unwrap();
+        assert_eq!(resp.kvs.len(), total_keys);
+    }
+}

src/cli/src/metadata/control/get.rs

@@ -20,7 +20,6 @@ use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_catalog::format_full_table_name;
 use common_error::ext::BoxedError;
 use common_meta::key::table_info::TableInfoKey;
-use common_meta::key::table_name::TableNameKey;
 use common_meta::key::table_route::TableRouteKey;
 use common_meta::key::TableMetadataManager;
 use common_meta::kv_backend::KvBackendRef;
@@ -30,10 +29,10 @@ use futures::TryStreamExt;
 
 use crate::error::InvalidArgumentsSnafu;
 use crate::metadata::common::StoreConfig;
-use crate::metadata::control::utils::{decode_key_value, json_fromatter};
+use crate::metadata::control::utils::{decode_key_value, get_table_id_by_name, json_fromatter};
 use crate::Tool;
 
-/// Subcommand for get command.
+/// Getting metadata from metadata store.
 #[derive(Subcommand)]
 pub enum GetCommand {
     Key(GetKeyCommand),
@@ -52,7 +51,7 @@ impl GetCommand {
 /// Get key-value pairs from the metadata store.
 #[derive(Debug, Default, Parser)]
 pub struct GetKeyCommand {
-    /// The key to get from the metadata store. If empty, returns all key-value pairs.
+    /// The key to get from the metadata store.
     #[clap(default_value = "")]
     key: String,
 
@@ -130,8 +129,12 @@ pub struct GetTableCommand {
     table_name: Option<String>,
 
     /// The schema name of the table.
-    #[clap(long)]
+    #[clap(long, default_value = DEFAULT_SCHEMA_NAME)]
-    schema_name: Option<String>,
+    schema_name: String,
+
+    /// The catalog name of the table.
+    #[clap(long, default_value = DEFAULT_CATALOG_NAME)]
+    catalog_name: String,
 
     /// Pretty print the output.
     #[clap(long, default_value = "false")]
@@ -143,7 +146,10 @@ pub struct GetTableCommand {
 
 impl GetTableCommand {
     pub fn validate(&self) -> Result<(), BoxedError> {
-        if self.table_id.is_none() && self.table_name.is_none() {
+        if matches!(
+            (&self.table_id, &self.table_name),
+            (Some(_), Some(_)) | (None, None)
+        ) {
             return Err(BoxedError::new(
                 InvalidArgumentsSnafu {
                     msg: "You must specify either --table-id or --table-name.",
@@ -159,7 +165,8 @@ struct GetTableTool {
     kvbackend: KvBackendRef,
     table_id: Option<u32>,
     table_name: Option<String>,
-    schema_name: Option<String>,
+    schema_name: String,
+    catalog_name: String,
     pretty: bool,
 }
 
@@ -172,23 +179,20 @@ impl Tool for GetTableTool {
         let table_route_manager = table_metadata_manager.table_route_manager();
 
         let table_id = if let Some(table_name) = &self.table_name {
-            let catalog = DEFAULT_CATALOG_NAME.to_string();
+            let catalog_name = &self.catalog_name;
-            let schema_name = self
+            let schema_name = &self.schema_name;
-                .schema_name
-                .clone()
-                .unwrap_or_else(|| DEFAULT_SCHEMA_NAME.to_string());
-            let key = TableNameKey::new(&catalog, &schema_name, table_name);
 
-            let Some(table_name) = table_name_manager.get(key).await.map_err(BoxedError::new)?
+            let Some(table_id) =
+                get_table_id_by_name(table_name_manager, catalog_name, schema_name, table_name)
+                    .await?
             else {
                 println!(
                     "Table({}) not found",
-                    format_full_table_name(&catalog, &schema_name, table_name)
+                    format_full_table_name(catalog_name, schema_name, table_name)
                 );
                 return Ok(());
             };
-
+            table_id
-            table_name.table_id()
         } else {
             // Safety: we have validated that table_id or table_name is not None
             self.table_id.unwrap()
@@ -236,6 +240,7 @@ impl GetTableCommand {
             table_id: self.table_id,
             table_name: self.table_name.clone(),
             schema_name: self.schema_name.clone(),
+            catalog_name: self.catalog_name.clone(),
             pretty: self.pretty,
         }))
     }

src/cli/src/metadata/control/test_utils.rs

@@ -0,0 +1,51 @@
+// 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_meta::ddl::test_util::test_create_physical_table_task;
+use common_meta::key::table_route::PhysicalTableRouteValue;
+use common_meta::kv_backend::KvBackendRef;
+use common_meta::peer::Peer;
+use common_meta::rpc::router::{Region, RegionRoute};
+use common_meta::rpc::store::PutRequest;
+use store_api::storage::{RegionId, TableId};
+use table::metadata::RawTableInfo;
+
+/// Puts a key-value pair into the kv backend.
+pub async fn put_key(kv_backend: &KvBackendRef, key: &str, value: &str) {
+    let put_req = PutRequest::new()
+        .with_key(key.as_bytes())
+        .with_value(value.as_bytes());
+    kv_backend.put(put_req).await.unwrap();
+}
+
+/// Prepares the physical table metadata for testing.
+///
+/// Returns the table info and the table route.
+pub async fn prepare_physical_table_metadata(
+    table_name: &str,
+    table_id: TableId,
+) -> (RawTableInfo, PhysicalTableRouteValue) {
+    let mut create_physical_table_task = test_create_physical_table_task(table_name);
+    let table_route = PhysicalTableRouteValue::new(vec![RegionRoute {
+        region: Region {
+            id: RegionId::new(table_id, 1),
+            ..Default::default()
+        },
+        leader_peer: Some(Peer::empty(1)),
+        ..Default::default()
+    }]);
+    create_physical_table_task.set_table_id(table_id);
+
+    (create_physical_table_task.table_info, table_route)
+}

src/cli/src/metadata/control/utils.rs

@@ -12,9 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use common_error::ext::BoxedError;
 use common_meta::error::Result as CommonMetaResult;
+use common_meta::key::table_name::{TableNameKey, TableNameManager};
 use common_meta::rpc::KeyValue;
 use serde::Serialize;
+use store_api::storage::TableId;
 
 /// Decodes a key-value pair into a string.
 pub fn decode_key_value(kv: KeyValue) -> CommonMetaResult<(String, String)> {
@@ -34,3 +37,21 @@ where
         serde_json::to_string(value).unwrap()
     }
 }
+
+/// Gets the table id by table name.
+pub async fn get_table_id_by_name(
+    table_name_manager: &TableNameManager,
+    catalog_name: &str,
+    schema_name: &str,
+    table_name: &str,
+) -> Result<Option<TableId>, BoxedError> {
+    let table_name_key = TableNameKey::new(catalog_name, schema_name, table_name);
+    let Some(table_name_value) = table_name_manager
+        .get(table_name_key)
+        .await
+        .map_err(BoxedError::new)?
+    else {
+        return Ok(None);
+    };
+    Ok(Some(table_name_value.table_id()))
+}

src/cli/src/metadata/repair.rs

@@ -0,0 +1,368 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+mod alter_table;
+mod create_table;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use clap::Parser;
+use client::api::v1::CreateTableExpr;
+use client::client_manager::NodeClients;
+use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
+use common_error::ext::{BoxedError, ErrorExt};
+use common_error::status_code::StatusCode;
+use common_grpc::channel_manager::ChannelConfig;
+use common_meta::error::Error as CommonMetaError;
+use common_meta::key::TableMetadataManager;
+use common_meta::kv_backend::KvBackendRef;
+use common_meta::node_manager::NodeManagerRef;
+use common_meta::peer::Peer;
+use common_meta::rpc::router::{find_leaders, RegionRoute};
+use common_telemetry::{error, info, warn};
+use futures::TryStreamExt;
+use snafu::{ensure, ResultExt};
+use store_api::storage::TableId;
+
+use crate::error::{
+    InvalidArgumentsSnafu, Result, SendRequestToDatanodeSnafu, TableMetadataSnafu, UnexpectedSnafu,
+};
+use crate::metadata::common::StoreConfig;
+use crate::metadata::utils::{FullTableMetadata, IteratorInput, TableMetadataIterator};
+use crate::Tool;
+
+/// Repair metadata of logical tables.
+#[derive(Debug, Default, Parser)]
+pub struct RepairLogicalTablesCommand {
+    /// The names of the tables to repair.
+    #[clap(long, value_delimiter = ',', alias = "table-name")]
+    table_names: Vec<String>,
+
+    /// The id of the table to repair.
+    #[clap(long, value_delimiter = ',', alias = "table-id")]
+    table_ids: Vec<TableId>,
+
+    /// The schema of the tables to repair.
+    #[clap(long, default_value = DEFAULT_SCHEMA_NAME)]
+    schema_name: String,
+
+    /// The catalog of the tables to repair.
+    #[clap(long, default_value = DEFAULT_CATALOG_NAME)]
+    catalog_name: String,
+
+    /// Whether to fail fast if any repair operation fails.
+    #[clap(long)]
+    fail_fast: bool,
+
+    #[clap(flatten)]
+    store: StoreConfig,
+
+    /// The timeout for the client to operate the datanode.
+    #[clap(long, default_value_t = 30)]
+    client_timeout_secs: u64,
+
+    /// The timeout for the client to connect to the datanode.
+    #[clap(long, default_value_t = 3)]
+    client_connect_timeout_secs: u64,
+}
+
+impl RepairLogicalTablesCommand {
+    fn validate(&self) -> Result<()> {
+        ensure!(
+            !self.table_names.is_empty() || !self.table_ids.is_empty(),
+            InvalidArgumentsSnafu {
+                msg: "You must specify --table-names or --table-ids.",
+            }
+        );
+        Ok(())
+    }
+}
+
+impl RepairLogicalTablesCommand {
+    pub async fn build(&self) -> std::result::Result<Box<dyn Tool>, BoxedError> {
+        self.validate().map_err(BoxedError::new)?;
+        let kv_backend = self.store.build().await?;
+        let node_client_channel_config = ChannelConfig::new()
+            .timeout(Duration::from_secs(self.client_timeout_secs))
+            .connect_timeout(Duration::from_secs(self.client_connect_timeout_secs));
+        let node_manager = Arc::new(NodeClients::new(node_client_channel_config));
+
+        Ok(Box::new(RepairTool {
+            table_names: self.table_names.clone(),
+            table_ids: self.table_ids.clone(),
+            schema_name: self.schema_name.clone(),
+            catalog_name: self.catalog_name.clone(),
+            fail_fast: self.fail_fast,
+            kv_backend,
+            node_manager,
+        }))
+    }
+}
+
+struct RepairTool {
+    table_names: Vec<String>,
+    table_ids: Vec<TableId>,
+    schema_name: String,
+    catalog_name: String,
+    fail_fast: bool,
+    kv_backend: KvBackendRef,
+    node_manager: NodeManagerRef,
+}
+
+#[async_trait]
+impl Tool for RepairTool {
+    async fn do_work(&self) -> std::result::Result<(), BoxedError> {
+        self.repair_tables().await.map_err(BoxedError::new)
+    }
+}
+
+impl RepairTool {
+    fn generate_iterator_input(&self) -> Result<IteratorInput> {
+        if !self.table_names.is_empty() {
+            let table_names = &self.table_names;
+            let catalog = &self.catalog_name;
+            let schema_name = &self.schema_name;
+
+            let table_names = table_names
+                .iter()
+                .map(|table_name| {
+                    (
+                        catalog.to_string(),
+                        schema_name.to_string(),
+                        table_name.to_string(),
+                    )
+                })
+                .collect::<Vec<_>>();
+            return Ok(IteratorInput::new_table_names(table_names));
+        } else if !self.table_ids.is_empty() {
+            return Ok(IteratorInput::new_table_ids(self.table_ids.clone()));
+        };
+
+        InvalidArgumentsSnafu {
+            msg: "You must specify --table-names or --table-id.",
+        }
+        .fail()
+    }
+
+    async fn repair_tables(&self) -> Result<()> {
+        let input = self.generate_iterator_input()?;
+        let mut table_metadata_iterator =
+            Box::pin(TableMetadataIterator::new(self.kv_backend.clone(), input).into_stream());
+        let table_metadata_manager = TableMetadataManager::new(self.kv_backend.clone());
+
+        let mut skipped_table = 0;
+        let mut success_table = 0;
+        while let Some(full_table_metadata) = table_metadata_iterator.try_next().await? {
+            let full_table_name = full_table_metadata.full_table_name();
+            if !full_table_metadata.is_metric_engine() {
+                warn!(
+                    "Skipping repair for non-metric engine table: {}",
+                    full_table_name
+                );
+                skipped_table += 1;
+                continue;
+            }
+
+            if full_table_metadata.is_physical_table() {
+                warn!("Skipping repair for physical table: {}", full_table_name);
+                skipped_table += 1;
+                continue;
+            }
+
+            let (physical_table_id, physical_table_route) = table_metadata_manager
+                .table_route_manager()
+                .get_physical_table_route(full_table_metadata.table_id)
+                .await
+                .context(TableMetadataSnafu)?;
+
+            if let Err(err) = self
+                .repair_table(
+                    &full_table_metadata,
+                    physical_table_id,
+                    &physical_table_route.region_routes,
+                )
+                .await
+            {
+                error!(
+                    err;
+                    "Failed to repair table: {}, skipped table: {}",
+                    full_table_name,
+                    skipped_table,
+                );
+
+                if self.fail_fast {
+                    return Err(err);
+                }
+            } else {
+                success_table += 1;
+            }
+        }
+
+        info!(
+            "Repair logical tables result: {} tables repaired, {} tables skipped",
+            success_table, skipped_table
+        );
+
+        Ok(())
+    }
+
+    async fn alter_table_on_datanodes(
+        &self,
+        full_table_metadata: &FullTableMetadata,
+        physical_region_routes: &[RegionRoute],
+    ) -> Result<Vec<(Peer, CommonMetaError)>> {
+        let logical_table_id = full_table_metadata.table_id;
+        let alter_table_expr = alter_table::generate_alter_table_expr_for_all_columns(
+            &full_table_metadata.table_info,
+        )?;
+        let node_manager = self.node_manager.clone();
+
+        let mut failed_peers = Vec::new();
+        info!(
+            "Sending alter table requests to all datanodes for table: {}, number of regions:{}.",
+            full_table_metadata.full_table_name(),
+            physical_region_routes.len()
+        );
+        let leaders = find_leaders(physical_region_routes);
+        for peer in &leaders {
+            let alter_table_request = alter_table::make_alter_region_request_for_peer(
+                logical_table_id,
+                &alter_table_expr,
+                peer,
+                physical_region_routes,
+            )?;
+            let datanode = node_manager.datanode(peer).await;
+            if let Err(err) = datanode.handle(alter_table_request).await {
+                failed_peers.push((peer.clone(), err));
+            }
+        }
+
+        Ok(failed_peers)
+    }
+
+    async fn create_table_on_datanode(
+        &self,
+        create_table_expr: &CreateTableExpr,
+        logical_table_id: TableId,
+        physical_table_id: TableId,
+        peer: &Peer,
+        physical_region_routes: &[RegionRoute],
+    ) -> Result<()> {
+        let node_manager = self.node_manager.clone();
+        let datanode = node_manager.datanode(peer).await;
+        let create_table_request = create_table::make_create_region_request_for_peer(
+            logical_table_id,
+            physical_table_id,
+            create_table_expr,
+            peer,
+            physical_region_routes,
+        )?;
+
+        datanode
+            .handle(create_table_request)
+            .await
+            .with_context(|_| SendRequestToDatanodeSnafu { peer: peer.clone() })?;
+
+        Ok(())
+    }
+
+    async fn repair_table(
+        &self,
+        full_table_metadata: &FullTableMetadata,
+        physical_table_id: TableId,
+        physical_region_routes: &[RegionRoute],
+    ) -> Result<()> {
+        let full_table_name = full_table_metadata.full_table_name();
+        // First we sends alter table requests to all datanodes with all columns.
+        let failed_peers = self
+            .alter_table_on_datanodes(full_table_metadata, physical_region_routes)
+            .await?;
+
+        if failed_peers.is_empty() {
+            info!(
+                "All alter table requests sent successfully for table: {}",
+                full_table_name
+            );
+            return Ok(());
+        }
+        warn!(
+            "Sending alter table requests to datanodes for table: {} failed for the datanodes: {:?}",
+            full_table_name,
+            failed_peers.iter().map(|(peer, _)| peer.id).collect::<Vec<_>>()
+        );
+
+        let create_table_expr =
+            create_table::generate_create_table_expr(&full_table_metadata.table_info)?;
+
+        let mut errors = Vec::new();
+        for (peer, err) in failed_peers {
+            if err.status_code() != StatusCode::RegionNotFound {
+                error!(
+                    err;
+                    "Sending alter table requests to datanode: {} for table: {} failed",
+                    peer.id,
+                    full_table_name,
+                );
+                continue;
+            }
+            info!(
+                "Region not found for table: {}, datanode: {}, trying to create the logical table on that datanode",
+                full_table_name,
+                peer.id
+            );
+
+            // If the alter table request fails for any datanode, we attempt to create the table on that datanode
+            // as a fallback mechanism to ensure table consistency across the cluster.
+            if let Err(err) = self
+                .create_table_on_datanode(
+                    &create_table_expr,
+                    full_table_metadata.table_id,
+                    physical_table_id,
+                    &peer,
+                    physical_region_routes,
+                )
+                .await
+            {
+                error!(
+                    err;
+                    "Failed to create table on datanode: {} for table: {}",
+                    peer.id, full_table_name
+                );
+                errors.push(err);
+                if self.fail_fast {
+                    break;
+                }
+            } else {
+                info!(
+                    "Created table on datanode: {} for table: {}",
+                    peer.id, full_table_name
+                );
+            }
+        }
+
+        if !errors.is_empty() {
+            return UnexpectedSnafu {
+                msg: format!(
+                    "Failed to create table on datanodes for table: {}",
+                    full_table_name,
+                ),
+            }
+            .fail();
+        }
+
+        Ok(())
+    }
+}

src/cli/src/metadata/repair/alter_table.rs

@@ -0,0 +1,84 @@
+// 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 client::api::v1::alter_table_expr::Kind;
+use client::api::v1::region::{region_request, AlterRequests, RegionRequest, RegionRequestHeader};
+use client::api::v1::{AddColumn, AddColumns, AlterTableExpr};
+use common_meta::ddl::alter_logical_tables::make_alter_region_request;
+use common_meta::peer::Peer;
+use common_meta::rpc::router::{find_leader_regions, RegionRoute};
+use operator::expr_helper::column_schemas_to_defs;
+use snafu::ResultExt;
+use store_api::storage::{RegionId, TableId};
+use table::metadata::RawTableInfo;
+
+use crate::error::{CovertColumnSchemasToDefsSnafu, Result};
+
+/// Generates alter table expression for all columns.
+pub fn generate_alter_table_expr_for_all_columns(
+    table_info: &RawTableInfo,
+) -> Result<AlterTableExpr> {
+    let schema = &table_info.meta.schema;
+
+    let mut alter_table_expr = AlterTableExpr {
+        catalog_name: table_info.catalog_name.to_string(),
+        schema_name: table_info.schema_name.to_string(),
+        table_name: table_info.name.to_string(),
+        ..Default::default()
+    };
+
+    let primary_keys = table_info
+        .meta
+        .primary_key_indices
+        .iter()
+        .map(|i| schema.column_schemas[*i].name.clone())
+        .collect::<Vec<_>>();
+
+    let add_columns = column_schemas_to_defs(schema.column_schemas.clone(), &primary_keys)
+        .context(CovertColumnSchemasToDefsSnafu)?;
+
+    alter_table_expr.kind = Some(Kind::AddColumns(AddColumns {
+        add_columns: add_columns
+            .into_iter()
+            .map(|col| AddColumn {
+                column_def: Some(col),
+                location: None,
+                add_if_not_exists: true,
+            })
+            .collect(),
+    }));
+
+    Ok(alter_table_expr)
+}
+
+/// Makes an alter region request for a peer.
+pub fn make_alter_region_request_for_peer(
+    logical_table_id: TableId,
+    alter_table_expr: &AlterTableExpr,
+    peer: &Peer,
+    region_routes: &[RegionRoute],
+) -> Result<RegionRequest> {
+    let regions_on_this_peer = find_leader_regions(region_routes, peer);
+    let mut requests = Vec::with_capacity(regions_on_this_peer.len());
+    for region_number in &regions_on_this_peer {
+        let region_id = RegionId::new(logical_table_id, *region_number);
+        let request = make_alter_region_request(region_id, alter_table_expr);
+        requests.push(request);
+    }
+
+    Ok(RegionRequest {
+        header: Some(RegionRequestHeader::default()),
+        body: Some(region_request::Body::Alters(AlterRequests { requests })),
+    })
+}

src/cli/src/metadata/repair/create_table.rs

@@ -0,0 +1,89 @@
+// 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::collections::HashMap;
+
+use client::api::v1::region::{region_request, CreateRequests, RegionRequest, RegionRequestHeader};
+use client::api::v1::CreateTableExpr;
+use common_meta::ddl::create_logical_tables::create_region_request_builder;
+use common_meta::ddl::utils::region_storage_path;
+use common_meta::peer::Peer;
+use common_meta::rpc::router::{find_leader_regions, RegionRoute};
+use operator::expr_helper::column_schemas_to_defs;
+use snafu::ResultExt;
+use store_api::storage::{RegionId, TableId};
+use table::metadata::RawTableInfo;
+
+use crate::error::{CovertColumnSchemasToDefsSnafu, Result};
+
+/// Generates a `CreateTableExpr` from a `RawTableInfo`.
+pub fn generate_create_table_expr(table_info: &RawTableInfo) -> Result<CreateTableExpr> {
+    let schema = &table_info.meta.schema;
+    let primary_keys = table_info
+        .meta
+        .primary_key_indices
+        .iter()
+        .map(|i| schema.column_schemas[*i].name.clone())
+        .collect::<Vec<_>>();
+
+    let timestamp_index = schema.timestamp_index.as_ref().unwrap();
+    let time_index = schema.column_schemas[*timestamp_index].name.clone();
+    let column_defs = column_schemas_to_defs(schema.column_schemas.clone(), &primary_keys)
+        .context(CovertColumnSchemasToDefsSnafu)?;
+    let table_options = HashMap::from(&table_info.meta.options);
+
+    Ok(CreateTableExpr {
+        catalog_name: table_info.catalog_name.to_string(),
+        schema_name: table_info.schema_name.to_string(),
+        table_name: table_info.name.to_string(),
+        desc: String::default(),
+        column_defs,
+        time_index,
+        primary_keys,
+        create_if_not_exists: true,
+        table_options,
+        table_id: None,
+        engine: table_info.meta.engine.to_string(),
+    })
+}
+
+/// Makes a create region request for a peer.
+pub fn make_create_region_request_for_peer(
+    logical_table_id: TableId,
+    physical_table_id: TableId,
+    create_table_expr: &CreateTableExpr,
+    peer: &Peer,
+    region_routes: &[RegionRoute],
+) -> Result<RegionRequest> {
+    let regions_on_this_peer = find_leader_regions(region_routes, peer);
+    let mut requests = Vec::with_capacity(regions_on_this_peer.len());
+    let request_builder =
+        create_region_request_builder(create_table_expr, physical_table_id).unwrap();
+
+    let catalog = &create_table_expr.catalog_name;
+    let schema = &create_table_expr.schema_name;
+    let storage_path = region_storage_path(catalog, schema);
+
+    for region_number in &regions_on_this_peer {
+        let region_id = RegionId::new(logical_table_id, *region_number);
+        let region_request =
+            request_builder.build_one(region_id, storage_path.clone(), &HashMap::new());
+        requests.push(region_request);
+    }
+
+    Ok(RegionRequest {
+        header: Some(RegionRequestHeader::default()),
+        body: Some(region_request::Body::Creates(CreateRequests { requests })),
+    })
+}

src/cli/src/metadata/snapshot.rs

@@ -301,7 +301,6 @@ struct MetaInfoTool {
 
 #[async_trait]
 impl Tool for MetaInfoTool {
-    #[allow(clippy::print_stdout)]
     async fn do_work(&self) -> std::result::Result<(), BoxedError> {
         let result = MetadataSnapshotManager::info(
             &self.inner,

src/cli/src/metadata/utils.rs

@@ -0,0 +1,178 @@
+// 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::collections::VecDeque;
+
+use async_stream::try_stream;
+use common_catalog::consts::METRIC_ENGINE;
+use common_catalog::format_full_table_name;
+use common_meta::key::table_name::TableNameKey;
+use common_meta::key::table_route::TableRouteValue;
+use common_meta::key::TableMetadataManager;
+use common_meta::kv_backend::KvBackendRef;
+use futures::Stream;
+use snafu::{OptionExt, ResultExt};
+use store_api::storage::TableId;
+use table::metadata::RawTableInfo;
+
+use crate::error::{Result, TableMetadataSnafu, UnexpectedSnafu};
+
+/// The input for the iterator.
+pub enum IteratorInput {
+    TableIds(VecDeque<TableId>),
+    TableNames(VecDeque<(String, String, String)>),
+}
+
+impl IteratorInput {
+    /// Creates a new iterator input from a list of table ids.
+    pub fn new_table_ids(table_ids: Vec<TableId>) -> Self {
+        Self::TableIds(table_ids.into())
+    }
+
+    /// Creates a new iterator input from a list of table names.
+    pub fn new_table_names(table_names: Vec<(String, String, String)>) -> Self {
+        Self::TableNames(table_names.into())
+    }
+}
+
+/// An iterator for retrieving table metadata from the metadata store.
+///
+/// This struct provides functionality to iterate over table metadata based on
+/// either [`TableId`] and their associated regions or fully qualified table names.
+pub struct TableMetadataIterator {
+    input: IteratorInput,
+    table_metadata_manager: TableMetadataManager,
+}
+
+/// The full table metadata.
+pub struct FullTableMetadata {
+    pub table_id: TableId,
+    pub table_info: RawTableInfo,
+    pub table_route: TableRouteValue,
+}
+
+impl FullTableMetadata {
+    /// Returns true if it's [TableRouteValue::Physical].
+    pub fn is_physical_table(&self) -> bool {
+        self.table_route.is_physical()
+    }
+
+    /// Returns true if it's a metric engine table.
+    pub fn is_metric_engine(&self) -> bool {
+        self.table_info.meta.engine == METRIC_ENGINE
+    }
+
+    /// Returns the full table name.
+    pub fn full_table_name(&self) -> String {
+        format_full_table_name(
+            &self.table_info.catalog_name,
+            &self.table_info.schema_name,
+            &self.table_info.name,
+        )
+    }
+}
+
+impl TableMetadataIterator {
+    pub fn new(kvbackend: KvBackendRef, input: IteratorInput) -> Self {
+        let table_metadata_manager = TableMetadataManager::new(kvbackend);
+        Self {
+            input,
+            table_metadata_manager,
+        }
+    }
+
+    /// Returns the next table metadata.
+    ///
+    /// This method handles two types of inputs:
+    /// - TableIds: Returns metadata for a specific [`TableId`].
+    /// - TableNames: Returns metadata for a table identified by its full name (catalog.schema.table).
+    ///
+    /// Returns `None` when there are no more tables to process.
+    pub async fn next(&mut self) -> Result<Option<FullTableMetadata>> {
+        match &mut self.input {
+            IteratorInput::TableIds(table_ids) => {
+                if let Some(table_id) = table_ids.pop_front() {
+                    let full_table_metadata = self.get_table_metadata(table_id).await?;
+                    return Ok(Some(full_table_metadata));
+                }
+            }
+
+            IteratorInput::TableNames(table_names) => {
+                if let Some(full_table_name) = table_names.pop_front() {
+                    let table_id = self.get_table_id_by_name(full_table_name).await?;
+                    let full_table_metadata = self.get_table_metadata(table_id).await?;
+                    return Ok(Some(full_table_metadata));
+                }
+            }
+        }
+
+        Ok(None)
+    }
+
+    /// Converts the iterator into a stream of table metadata.
+    pub fn into_stream(mut self) -> impl Stream<Item = Result<FullTableMetadata>> {
+        try_stream!({
+            while let Some(full_table_metadata) = self.next().await? {
+                yield full_table_metadata;
+            }
+        })
+    }
+
+    async fn get_table_id_by_name(
+        &mut self,
+        (catalog_name, schema_name, table_name): (String, String, String),
+    ) -> Result<TableId> {
+        let key = TableNameKey::new(&catalog_name, &schema_name, &table_name);
+        let table_id = self
+            .table_metadata_manager
+            .table_name_manager()
+            .get(key)
+            .await
+            .context(TableMetadataSnafu)?
+            .with_context(|| UnexpectedSnafu {
+                msg: format!(
+                    "Table not found: {}",
+                    format_full_table_name(&catalog_name, &schema_name, &table_name)
+                ),
+            })?
+            .table_id();
+        Ok(table_id)
+    }
+
+    async fn get_table_metadata(&mut self, table_id: TableId) -> Result<FullTableMetadata> {
+        let (table_info, table_route) = self
+            .table_metadata_manager
+            .get_full_table_info(table_id)
+            .await
+            .context(TableMetadataSnafu)?;
+
+        let table_info = table_info
+            .with_context(|| UnexpectedSnafu {
+                msg: format!("Table info not found for table id: {table_id}"),
+            })?
+            .into_inner()
+            .table_info;
+        let table_route = table_route
+            .with_context(|| UnexpectedSnafu {
+                msg: format!("Table route not found for table id: {table_id}"),
+            })?
+            .into_inner();
+
+        Ok(FullTableMetadata {
+            table_id,
+            table_info,
+            table_route,
+        })
+    }
+}

src/client/src/database.rs

@@ -31,7 +31,7 @@ use base64::prelude::BASE64_STANDARD;
 use base64::Engine;
 use common_catalog::build_db_string;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
-use common_error::ext::{BoxedError, ErrorExt};
+use common_error::ext::BoxedError;
 use common_grpc::flight::do_put::DoPutResponse;
 use common_grpc::flight::{FlightDecoder, FlightMessage};
 use common_query::Output;
@@ -48,7 +48,7 @@ use tonic::transport::Channel;
 
 use crate::error::{
     ConvertFlightDataSnafu, Error, FlightGetSnafu, IllegalFlightMessagesSnafu,
-    InvalidTonicMetadataValueSnafu, ServerSnafu,
+    InvalidTonicMetadataValueSnafu,
 };
 use crate::{error, from_grpc_response, Client, Result};
 
@@ -196,12 +196,22 @@ impl Database {
 
     /// Retry if connection fails, max_retries is the max number of retries, so the total wait time
     /// is `max_retries * GRPC_CONN_TIMEOUT`
-    pub async fn handle_with_retry(&self, request: Request, max_retries: u32) -> Result<u32> {
+    pub async fn handle_with_retry(
+        &self,
+        request: Request,
+        max_retries: u32,
+        hints: &[(&str, &str)],
+    ) -> Result<u32> {
         let mut client = make_database_client(&self.client)?.inner;
         let mut retries = 0;
+
         let request = self.to_rpc_request(request);
+
         loop {
-            let raw_response = client.handle(request.clone()).await;
+            let mut tonic_request = tonic::Request::new(request.clone());
+            let metadata = tonic_request.metadata_mut();
+            Self::put_hints(metadata, hints)?;
+            let raw_response = client.handle(tonic_request).await;
             match (raw_response, retries < max_retries) {
                 (Ok(resp), _) => return from_grpc_response(resp.into_inner()),
                 (Err(err), true) => {
@@ -292,21 +302,16 @@ impl Database {
         let response = client.mut_inner().do_get(request).await.or_else(|e| {
             let tonic_code = e.code();
             let e: Error = e.into();
-            let code = e.status_code();
-            let msg = e.to_string();
-            let error =
-                Err(BoxedError::new(ServerSnafu { code, msg }.build())).with_context(|_| {
-                    FlightGetSnafu {
-                        addr: client.addr().to_string(),
-                        tonic_code,
-                    }
-                });
             error!(
                 "Failed to do Flight get, addr: {}, code: {}, source: {:?}",
                 client.addr(),
                 tonic_code,
-                error
+                e
             );
+            let error = Err(BoxedError::new(e)).with_context(|_| FlightGetSnafu {
+                addr: client.addr().to_string(),
+                tonic_code,
+            });
             error
         })?;
 
@@ -436,8 +441,11 @@ mod tests {
 
     use api::v1::auth_header::AuthScheme;
     use api::v1::{AuthHeader, Basic};
+    use common_error::status_code::StatusCode;
+    use tonic::{Code, Status};
 
     use super::*;
+    use crate::error::TonicSnafu;
 
     #[test]
     fn test_flight_ctx() {
@@ -460,4 +468,19 @@ mod tests {
             })
         )
     }
+
+    #[test]
+    fn test_from_tonic_status() {
+        let expected = TonicSnafu {
+            code: StatusCode::Internal,
+            msg: "blabla".to_string(),
+            tonic_code: Code::Internal,
+        }
+        .build();
+
+        let status = Status::new(Code::Internal, "blabla");
+        let actual: Error = status.into();
+
+        assert_eq!(expected.to_string(), actual.to_string());
+    }
 }

src/client/src/error.rs

@@ -14,13 +14,13 @@
 
 use std::any::Any;
 
+use common_error::define_from_tonic_status;
 use common_error::ext::{BoxedError, ErrorExt};
-use common_error::status_code::{convert_tonic_code_to_status_code, StatusCode};
+use common_error::status_code::StatusCode;
-use common_error::{GREPTIME_DB_HEADER_ERROR_CODE, GREPTIME_DB_HEADER_ERROR_MSG};
 use common_macro::stack_trace_debug;
 use snafu::{location, Location, Snafu};
 use tonic::metadata::errors::InvalidMetadataValue;
-use tonic::{Code, Status};
+use tonic::Code;
 
 #[derive(Snafu)]
 #[snafu(visibility(pub))]
@@ -124,6 +124,15 @@ pub enum Error {
         location: Location,
         source: datatypes::error::Error,
     },
+
+    #[snafu(display("{}", msg))]
+    Tonic {
+        code: StatusCode,
+        msg: String,
+        tonic_code: Code,
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -135,7 +144,7 @@ impl ErrorExt for Error {
             | Error::MissingField { .. }
             | Error::IllegalDatabaseResponse { .. } => StatusCode::Internal,
 
-            Error::Server { code, .. } => *code,
+            Error::Server { code, .. } | Error::Tonic { code, .. } => *code,
             Error::FlightGet { source, .. }
             | Error::RegionServer { source, .. }
             | Error::FlowServer { source, .. } => source.status_code(),
@@ -153,34 +162,7 @@ impl ErrorExt for Error {
     }
 }
 
-impl From<Status> for Error {
+define_from_tonic_status!(Error, Tonic);
-    fn from(e: Status) -> Self {
-        fn get_metadata_value(e: &Status, key: &str) -> Option<String> {
-            e.metadata()
-                .get(key)
-                .and_then(|v| String::from_utf8(v.as_bytes().to_vec()).ok())
-        }
-
-        let code = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_CODE).and_then(|s| {
-            if let Ok(code) = s.parse::<u32>() {
-                StatusCode::from_u32(code)
-            } else {
-                None
-            }
-        });
-        let tonic_code = e.code();
-        let code = code.unwrap_or_else(|| convert_tonic_code_to_status_code(tonic_code));
-
-        let msg = get_metadata_value(&e, GREPTIME_DB_HEADER_ERROR_MSG)
-            .unwrap_or_else(|| e.message().to_string());
-
-        Self::Server {
-            code,
-            msg,
-            location: location!(),
-        }
-    }
-}
 
 impl Error {
     pub fn should_retry(&self) -> bool {

src/client/src/region.rs

@@ -21,7 +21,7 @@ use arc_swap::ArcSwapOption;
 use arrow_flight::Ticket;
 use async_stream::stream;
 use async_trait::async_trait;
-use common_error::ext::{BoxedError, ErrorExt};
+use common_error::ext::BoxedError;
 use common_error::status_code::StatusCode;
 use common_grpc::flight::{FlightDecoder, FlightMessage};
 use common_meta::error::{self as meta_error, Result as MetaResult};
@@ -107,24 +107,18 @@ impl RegionRequester {
             .mut_inner()
             .do_get(ticket)
             .await
-            .map_err(|e| {
+            .or_else(|e| {
                 let tonic_code = e.code();
                 let e: error::Error = e.into();
-                let code = e.status_code();
-                let msg = e.to_string();
-                let error = ServerSnafu { code, msg }
-                    .fail::<()>()
-                    .map_err(BoxedError::new)
-                    .with_context(|_| FlightGetSnafu {
-                        tonic_code,
-                        addr: flight_client.addr().to_string(),
-                    })
-                    .unwrap_err();
                 error!(
                     e; "Failed to do Flight get, addr: {}, code: {}",
                     flight_client.addr(),
                     tonic_code
                 );
+                let error = Err(BoxedError::new(e)).with_context(|_| FlightGetSnafu {
+                    addr: flight_client.addr().to_string(),
+                    tonic_code,
+                });
                 error
             })?;
 

src/cmd/Cargo.toml

@@ -16,7 +16,7 @@ default = [
     "meta-srv/pg_kvbackend",
     "meta-srv/mysql_kvbackend",
 ]
-enterprise = ["common-meta/enterprise", "frontend/enterprise", "meta-srv/enterprise"]
+enterprise = ["common-meta/enterprise", "frontend/enterprise", "meta-srv/enterprise", "catalog/enterprise"]
 tokio-console = ["common-telemetry/tokio-console"]
 
 [lints]
@@ -52,7 +52,6 @@ common-version.workspace = true
 common-wal.workspace = true
 datanode.workspace = true
 datatypes.workspace = true
-either = "1.8"
 etcd-client.workspace = true
 file-engine.workspace = true
 flow.workspace = true
@@ -67,6 +66,7 @@ metric-engine.workspace = true
 mito2.workspace = true
 moka.workspace = true
 nu-ansi-term = "0.46"
+object-store.workspace = true
 plugins.workspace = true
 prometheus.workspace = true
 prost.workspace = true

src/cmd/src/bin/greptime.rs

@@ -20,11 +20,11 @@ use cmd::error::{InitTlsProviderSnafu, Result};
 use cmd::options::GlobalOptions;
 use cmd::{cli, datanode, flownode, frontend, metasrv, standalone, App};
 use common_base::Plugins;
-use common_version::version;
+use common_version::{verbose_version, version};
 use servers::install_ring_crypto_provider;
 
 #[derive(Parser)]
-#[command(name = "greptime", author, version, long_version = version(), about)]
+#[command(name = "greptime", author, version, long_version = verbose_version(), about)]
 #[command(propagate_version = true)]
 pub(crate) struct Command {
     #[clap(subcommand)]
@@ -143,10 +143,8 @@ async fn start(cli: Command) -> Result<()> {
 }
 
 fn setup_human_panic() {
-    human_panic::setup_panic!(
+    human_panic::setup_panic!(human_panic::Metadata::new("GreptimeDB", version())
-        human_panic::Metadata::new("GreptimeDB", env!("CARGO_PKG_VERSION"))
+        .homepage("https://github.com/GreptimeTeam/greptimedb/discussions"));
-            .homepage("https://github.com/GreptimeTeam/greptimedb/discussions")
-    );
 
     common_telemetry::set_panic_hook();
 }

src/cmd/src/datanode.rs

@@ -280,7 +280,7 @@ mod tests {
 
     use common_config::ENV_VAR_SEP;
     use common_test_util::temp_dir::create_named_temp_file;
-    use datanode::config::{FileConfig, GcsConfig, ObjectStoreConfig, S3Config};
+    use object_store::config::{FileConfig, GcsConfig, ObjectStoreConfig, S3Config};
     use servers::heartbeat_options::HeartbeatOptions;
 
     use super::*;

src/cmd/src/datanode/builder.rs

@@ -19,7 +19,7 @@ use catalog::kvbackend::MetaKvBackend;
 use common_base::Plugins;
 use common_meta::cache::LayeredCacheRegistryBuilder;
 use common_telemetry::info;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use datanode::datanode::DatanodeBuilder;
 use datanode::service::DatanodeServiceBuilder;
 use meta_client::MetaClientType;
@@ -67,7 +67,7 @@ impl InstanceBuilder {
             None,
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         plugins::setup_datanode_plugins(plugins, &opts.plugins, dn_opts)
@@ -93,6 +93,7 @@ impl InstanceBuilder {
             MetaClientType::Datanode { member_id },
             meta_client_options,
             Some(&plugins),
+            None,
         )
         .await
         .context(MetaClientInitSnafu)?;

src/cmd/src/flownode.rs

@@ -18,7 +18,7 @@ use std::time::Duration;
 
 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
 use catalog::information_extension::DistributedInformationExtension;
-use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
+use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManagerBuilder, MetaKvBackend};
 use clap::Parser;
 use client::client_manager::NodeClients;
 use common_base::Plugins;
@@ -32,7 +32,7 @@ use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::TableMetadataManager;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use flow::{
     get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder,
     FrontendClient, FrontendInvoker,
@@ -55,14 +55,32 @@ type FlownodeOptions = GreptimeOptions<flow::FlownodeOptions>;
 pub struct Instance {
     flownode: FlownodeInstance,
 
+    // The components of flownode, which make it easier to expand based
+    // on the components.
+    #[cfg(feature = "enterprise")]
+    components: Components,
+
     // Keep the logging guard to prevent the worker from being dropped.
     _guard: Vec<WorkerGuard>,
 }
 
+#[cfg(feature = "enterprise")]
+pub struct Components {
+    pub catalog_manager: catalog::CatalogManagerRef,
+    pub fe_client: Arc<FrontendClient>,
+    pub kv_backend: common_meta::kv_backend::KvBackendRef,
+}
+
 impl Instance {
-    pub fn new(flownode: FlownodeInstance, guard: Vec<WorkerGuard>) -> Self {
+    pub fn new(
+        flownode: FlownodeInstance,
+        #[cfg(feature = "enterprise")] components: Components,
+        guard: Vec<WorkerGuard>,
+    ) -> Self {
         Self {
             flownode,
+            #[cfg(feature = "enterprise")]
+            components,
             _guard: guard,
         }
     }
@@ -75,6 +93,11 @@ impl Instance {
     pub fn flownode_mut(&mut self) -> &mut FlownodeInstance {
         &mut self.flownode
     }
+
+    #[cfg(feature = "enterprise")]
+    pub fn components(&self) -> &Components {
+        &self.components
+    }
 }
 
 #[async_trait::async_trait]
@@ -256,7 +279,7 @@ impl StartCommand {
             None,
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Flownode start command: {:#?}", self);
@@ -283,6 +306,7 @@ impl StartCommand {
             MetaClientType::Flownode { member_id },
             meta_config,
             None,
+            None,
         )
         .await
         .context(MetaClientInitSnafu)?;
@@ -318,13 +342,12 @@ impl StartCommand {
 
         let information_extension =
             Arc::new(DistributedInformationExtension::new(meta_client.clone()));
-        let catalog_manager = KvBackendCatalogManager::new(
+        let catalog_manager = KvBackendCatalogManagerBuilder::new(
             information_extension,
             cached_meta_backend.clone(),
             layered_cache_registry.clone(),
-            None,
+        )
-            None,
+        .build();
-        );
 
         let table_metadata_manager =
             Arc::new(TableMetadataManager::new(cached_meta_backend.clone()));
@@ -347,21 +370,25 @@ impl StartCommand {
 
         let flow_metadata_manager = Arc::new(FlowMetadataManager::new(cached_meta_backend.clone()));
         let flow_auth_header = get_flow_auth_options(&opts).context(StartFlownodeSnafu)?;
-        let frontend_client =
+        let frontend_client = FrontendClient::from_meta_client(
-            FrontendClient::from_meta_client(meta_client.clone(), flow_auth_header);
+            meta_client.clone(),
+            flow_auth_header,
+            opts.query.clone(),
+        );
+        let frontend_client = Arc::new(frontend_client);
         let flownode_builder = FlownodeBuilder::new(
             opts.clone(),
             plugins,
             table_metadata_manager,
             catalog_manager.clone(),
             flow_metadata_manager,
-            Arc::new(frontend_client),
+            frontend_client.clone(),
         )
         .with_heartbeat_task(heartbeat_task);
 
         let mut flownode = flownode_builder.build().await.context(StartFlownodeSnafu)?;
         let services = FlownodeServiceBuilder::new(&opts)
-            .with_grpc_server(flownode.flownode_server().clone())
+            .with_default_grpc_server(flownode.flownode_server())
             .enable_http_service()
             .build()
             .context(StartFlownodeSnafu)?;
@@ -393,6 +420,16 @@ impl StartCommand {
             .set_frontend_invoker(invoker)
             .await;
 
-        Ok(Instance::new(flownode, guard))
+        #[cfg(feature = "enterprise")]
+        let components = Components {
+            catalog_manager: catalog_manager.clone(),
+            fe_client: frontend_client,
+            kv_backend: cached_meta_backend,
+        };
+
+        #[cfg(not(feature = "enterprise"))]
+        return Ok(Instance::new(flownode, guard));
+        #[cfg(feature = "enterprise")]
+        Ok(Instance::new(flownode, components, guard))
     }
 }

src/cmd/src/frontend.rs

@@ -19,7 +19,7 @@ use std::time::Duration;
 use async_trait::async_trait;
 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
 use catalog::information_extension::DistributedInformationExtension;
-use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManager, MetaKvBackend};
+use catalog::kvbackend::{CachedKvBackendBuilder, KvBackendCatalogManagerBuilder, MetaKvBackend};
 use catalog::process_manager::ProcessManager;
 use clap::Parser;
 use client::client_manager::NodeClients;
@@ -33,7 +33,7 @@ use common_meta::heartbeat::handler::HandlerGroupExecutor;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
 use common_time::timezone::set_default_timezone;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use frontend::frontend::Frontend;
 use frontend::heartbeat::HeartbeatTask;
 use frontend::instance::builder::FrontendBuilder;
@@ -102,7 +102,7 @@ impl App for Instance {
 #[derive(Parser)]
 pub struct Command {
     #[clap(subcommand)]
-    subcmd: SubCommand,
+    pub subcmd: SubCommand,
 }
 
 impl Command {
@@ -116,7 +116,7 @@ impl Command {
 }
 
 #[derive(Parser)]
-enum SubCommand {
+pub enum SubCommand {
     Start(StartCommand),
 }
 
@@ -153,7 +153,7 @@ pub struct StartCommand {
     #[clap(long)]
     postgres_addr: Option<String>,
     #[clap(short, long)]
-    config_file: Option<String>,
+    pub config_file: Option<String>,
     #[clap(short, long)]
     influxdb_enable: Option<bool>,
     #[clap(long, value_delimiter = ',', num_args = 1..)]
@@ -169,7 +169,7 @@ pub struct StartCommand {
     #[clap(long)]
     disable_dashboard: Option<bool>,
     #[clap(long, default_value = "GREPTIMEDB_FRONTEND")]
-    env_prefix: String,
+    pub env_prefix: String,
 }
 
 impl StartCommand {
@@ -282,7 +282,7 @@ impl StartCommand {
             opts.component.slow_query.as_ref(),
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Frontend start command: {:#?}", self);
@@ -313,6 +313,7 @@ impl StartCommand {
             MetaClientType::Frontend,
             meta_client_options,
             Some(&plugins),
+            None,
         )
         .await
         .context(error::MetaClientInitSnafu)?;
@@ -349,13 +350,20 @@ impl StartCommand {
             addrs::resolve_addr(&opts.grpc.bind_addr, Some(&opts.grpc.server_addr)),
             Some(meta_client.clone()),
         ));
-        let catalog_manager = KvBackendCatalogManager::new(
+
+        let builder = KvBackendCatalogManagerBuilder::new(
             information_extension,
             cached_meta_backend.clone(),
             layered_cache_registry.clone(),
-            None,
+        )
-            Some(process_manager.clone()),
+        .with_process_manager(process_manager.clone());
-        );
+        #[cfg(feature = "enterprise")]
+        let builder = if let Some(factories) = plugins.get() {
+            builder.with_extra_information_table_factories(factories)
+        } else {
+            builder
+        };
+        let catalog_manager = builder.build();
 
         let executor = HandlerGroupExecutor::new(vec![
             Arc::new(ParseMailboxMessageHandler),

src/cmd/src/lib.rs

@@ -112,7 +112,7 @@ pub trait App: Send {
 pub fn log_versions(version: &str, short_version: &str, app: &str) {
     // Report app version as gauge.
     APP_VERSION
-        .with_label_values(&[env!("CARGO_PKG_VERSION"), short_version, app])
+        .with_label_values(&[common_version::version(), short_version, app])
         .inc();
 
     // Log version and argument flags.

src/cmd/src/metasrv.rs

@@ -22,7 +22,7 @@ use common_base::Plugins;
 use common_config::Configurable;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use meta_srv::bootstrap::MetasrvInstance;
 use meta_srv::metasrv::BackendImpl;
 use snafu::ResultExt;
@@ -54,6 +54,10 @@ impl Instance {
     pub fn get_inner(&self) -> &MetasrvInstance {
         &self.instance
     }
+
+    pub fn mut_inner(&mut self) -> &mut MetasrvInstance {
+        &mut self.instance
+    }
 }
 
 #[async_trait]
@@ -320,7 +324,7 @@ impl StartCommand {
             None,
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Metasrv start command: {:#?}", self);
@@ -336,12 +340,12 @@ impl StartCommand {
             .await
             .context(StartMetaServerSnafu)?;
 
-        let builder = meta_srv::bootstrap::metasrv_builder(&opts, plugins.clone(), None)
+        let builder = meta_srv::bootstrap::metasrv_builder(&opts, plugins, None)
             .await
             .context(error::BuildMetaServerSnafu)?;
         let metasrv = builder.build().await.context(error::BuildMetaServerSnafu)?;
 
-        let instance = MetasrvInstance::new(opts, plugins, metasrv)
+        let instance = MetasrvInstance::new(metasrv)
             .await
             .context(error::BuildMetaServerSnafu)?;
 

src/cmd/src/standalone.rs

@@ -20,7 +20,7 @@ use std::{fs, path};
 use async_trait::async_trait;
 use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
 use catalog::information_schema::InformationExtension;
-use catalog::kvbackend::KvBackendCatalogManager;
+use catalog::kvbackend::KvBackendCatalogManagerBuilder;
 use catalog::process_manager::ProcessManager;
 use clap::Parser;
 use client::api::v1::meta::RegionRole;
@@ -30,20 +30,16 @@ use common_catalog::consts::{MIN_USER_FLOW_ID, MIN_USER_TABLE_ID};
 use common_config::{metadata_store_dir, Configurable, KvBackendConfig};
 use common_error::ext::BoxedError;
 use common_meta::cache::LayeredCacheRegistryBuilder;
-use common_meta::cache_invalidator::CacheInvalidatorRef;
 use common_meta::cluster::{NodeInfo, NodeStatus};
 use common_meta::datanode::RegionStat;
-use common_meta::ddl::flow_meta::{FlowMetadataAllocator, FlowMetadataAllocatorRef};
+use common_meta::ddl::flow_meta::FlowMetadataAllocator;
-use common_meta::ddl::table_meta::{TableMetadataAllocator, TableMetadataAllocatorRef};
+use common_meta::ddl::table_meta::TableMetadataAllocator;
 use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl, ProcedureExecutorRef};
 use common_meta::ddl_manager::DdlManager;
-#[cfg(feature = "enterprise")]
-use common_meta::ddl_manager::TriggerDdlManagerRef;
 use common_meta::key::flow::flow_state::FlowStat;
-use common_meta::key::flow::{FlowMetadataManager, FlowMetadataManagerRef};
+use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
-use common_meta::node_manager::NodeManagerRef;
 use common_meta::peer::Peer;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::region_registry::LeaderRegionRegistry;
@@ -55,7 +51,7 @@ use common_telemetry::logging::{
     LoggingOptions, SlowQueryOptions, TracingOptions, DEFAULT_LOGGING_DIR,
 };
 use common_time::timezone::set_default_timezone;
-use common_version::{short_version, version};
+use common_version::{short_version, verbose_version};
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
 use datanode::datanode::{Datanode, DatanodeBuilder};
@@ -261,15 +257,34 @@ pub struct Instance {
     flownode: FlownodeInstance,
     procedure_manager: ProcedureManagerRef,
     wal_options_allocator: WalOptionsAllocatorRef,
+
+    // The components of standalone, which make it easier to expand based
+    // on the components.
+    #[cfg(feature = "enterprise")]
+    components: Components,
+
     // Keep the logging guard to prevent the worker from being dropped.
     _guard: Vec<WorkerGuard>,
 }
 
+#[cfg(feature = "enterprise")]
+pub struct Components {
+    pub plugins: Plugins,
+    pub kv_backend: KvBackendRef,
+    pub frontend_client: Arc<FrontendClient>,
+    pub catalog_manager: catalog::CatalogManagerRef,
+}
+
 impl Instance {
     /// Find the socket addr of a server by its `name`.
     pub fn server_addr(&self, name: &str) -> Option<SocketAddr> {
         self.frontend.server_handlers().addr(name)
     }
+
+    #[cfg(feature = "enterprise")]
+    pub fn components(&self) -> &Components {
+        &self.components
+    }
 }
 
 #[async_trait]
@@ -470,7 +485,7 @@ impl StartCommand {
             opts.component.slow_query.as_ref(),
         );
 
-        log_versions(version(), short_version(), APP_NAME);
+        log_versions(verbose_version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Standalone start command: {:#?}", self);
@@ -529,13 +544,20 @@ impl StartCommand {
         ));
 
         let process_manager = Arc::new(ProcessManager::new(opts.grpc.server_addr.clone(), None));
-        let catalog_manager = KvBackendCatalogManager::new(
+        let builder = KvBackendCatalogManagerBuilder::new(
             information_extension.clone(),
             kv_backend.clone(),
             layered_cache_registry.clone(),
-            Some(procedure_manager.clone()),
+        )
-            Some(process_manager.clone()),
+        .with_procedure_manager(procedure_manager.clone())
-        );
+        .with_process_manager(process_manager.clone());
+        #[cfg(feature = "enterprise")]
+        let builder = if let Some(factories) = plugins.get() {
+            builder.with_extra_information_table_factories(factories)
+        } else {
+            builder
+        };
+        let catalog_manager = builder.build();
 
         let table_metadata_manager =
             Self::create_table_metadata_manager(kv_backend.clone()).await?;
@@ -549,14 +571,15 @@ impl StartCommand {
         // for standalone not use grpc, but get a handler to frontend grpc client without
         // actually make a connection
         let (frontend_client, frontend_instance_handler) =
-            FrontendClient::from_empty_grpc_handler();
+            FrontendClient::from_empty_grpc_handler(opts.query.clone());
+        let frontend_client = Arc::new(frontend_client);
         let flow_builder = FlownodeBuilder::new(
             flownode_options,
             plugins.clone(),
             table_metadata_manager.clone(),
             catalog_manager.clone(),
             flow_metadata_manager.clone(),
-            Arc::new(frontend_client.clone()),
+            frontend_client.clone(),
         );
         let flownode = flow_builder
             .build()
@@ -594,28 +617,36 @@ impl StartCommand {
             .await
             .context(error::BuildWalOptionsAllocatorSnafu)?;
         let wal_options_allocator = Arc::new(wal_options_allocator);
-        let table_meta_allocator = Arc::new(TableMetadataAllocator::new(
+        let table_metadata_allocator = Arc::new(TableMetadataAllocator::new(
             table_id_sequence,
             wal_options_allocator.clone(),
         ));
-        let flow_meta_allocator = Arc::new(FlowMetadataAllocator::with_noop_peer_allocator(
+        let flow_metadata_allocator = Arc::new(FlowMetadataAllocator::with_noop_peer_allocator(
             flow_id_sequence,
         ));
 
+        let ddl_context = DdlContext {
+            node_manager: node_manager.clone(),
+            cache_invalidator: layered_cache_registry.clone(),
+            memory_region_keeper: Arc::new(MemoryRegionKeeper::default()),
+            leader_region_registry: Arc::new(LeaderRegionRegistry::default()),
+            table_metadata_manager: table_metadata_manager.clone(),
+            table_metadata_allocator: table_metadata_allocator.clone(),
+            flow_metadata_manager: flow_metadata_manager.clone(),
+            flow_metadata_allocator: flow_metadata_allocator.clone(),
+            region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
+        };
+        let procedure_manager_c = procedure_manager.clone();
+
+        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager_c, true)
+            .context(error::InitDdlManagerSnafu)?;
         #[cfg(feature = "enterprise")]
-        let trigger_ddl_manager: Option<TriggerDdlManagerRef> = plugins.get();
+        let ddl_manager = {
-        let ddl_task_executor = Self::create_ddl_task_executor(
+            let trigger_ddl_manager: Option<common_meta::ddl_manager::TriggerDdlManagerRef> =
-            procedure_manager.clone(),
+                plugins.get();
-            node_manager.clone(),
+            ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
-            layered_cache_registry.clone(),
+        };
-            table_metadata_manager,
+        let ddl_task_executor: ProcedureExecutorRef = Arc::new(ddl_manager);
-            table_meta_allocator,
-            flow_metadata_manager,
-            flow_meta_allocator,
-            #[cfg(feature = "enterprise")]
-            trigger_ddl_manager,
-        )
-        .await?;
 
         let fe_instance = FrontendBuilder::new(
             fe_opts.clone(),
@@ -658,7 +689,7 @@ impl StartCommand {
         let export_metrics_task = ExportMetricsTask::try_new(&opts.export_metrics, Some(&plugins))
             .context(error::ServersSnafu)?;
 
-        let servers = Services::new(opts, fe_instance.clone(), plugins)
+        let servers = Services::new(opts, fe_instance.clone(), plugins.clone())
             .build()
             .context(error::StartFrontendSnafu)?;
 
@@ -669,51 +700,26 @@ impl StartCommand {
             export_metrics_task,
         };
 
+        #[cfg(feature = "enterprise")]
+        let components = Components {
+            plugins,
+            kv_backend,
+            frontend_client,
+            catalog_manager,
+        };
+
         Ok(Instance {
             datanode,
             frontend,
             flownode,
             procedure_manager,
             wal_options_allocator,
+            #[cfg(feature = "enterprise")]
+            components,
             _guard: guard,
         })
     }
 
-    #[allow(clippy::too_many_arguments)]
-    pub async fn create_ddl_task_executor(
-        procedure_manager: ProcedureManagerRef,
-        node_manager: NodeManagerRef,
-        cache_invalidator: CacheInvalidatorRef,
-        table_metadata_manager: TableMetadataManagerRef,
-        table_metadata_allocator: TableMetadataAllocatorRef,
-        flow_metadata_manager: FlowMetadataManagerRef,
-        flow_metadata_allocator: FlowMetadataAllocatorRef,
-        #[cfg(feature = "enterprise")] trigger_ddl_manager: Option<TriggerDdlManagerRef>,
-    ) -> Result<ProcedureExecutorRef> {
-        let procedure_executor: ProcedureExecutorRef = Arc::new(
-            DdlManager::try_new(
-                DdlContext {
-                    node_manager,
-                    cache_invalidator,
-                    memory_region_keeper: Arc::new(MemoryRegionKeeper::default()),
-                    leader_region_registry: Arc::new(LeaderRegionRegistry::default()),
-                    table_metadata_manager,
-                    table_metadata_allocator,
-                    flow_metadata_manager,
-                    flow_metadata_allocator,
-                    region_failure_detector_controller: Arc::new(NoopRegionFailureDetectorControl),
-                },
-                procedure_manager,
-                true,
-                #[cfg(feature = "enterprise")]
-                trigger_ddl_manager,
-            )
-            .context(error::InitDdlManagerSnafu)?,
-        );
-
-        Ok(procedure_executor)
-    }
-
     pub async fn create_table_metadata_manager(
         kv_backend: KvBackendRef,
     ) -> Result<TableMetadataManagerRef> {
@@ -849,7 +855,7 @@ mod tests {
     use common_config::ENV_VAR_SEP;
     use common_test_util::temp_dir::create_named_temp_file;
     use common_wal::config::DatanodeWalConfig;
-    use datanode::config::{FileConfig, GcsConfig};
+    use object_store::config::{FileConfig, GcsConfig};
 
     use super::*;
     use crate::options::GlobalOptions;
@@ -968,15 +974,15 @@ mod tests {
 
         assert!(matches!(
             &dn_opts.storage.store,
-            datanode::config::ObjectStoreConfig::File(FileConfig { .. })
+            object_store::config::ObjectStoreConfig::File(FileConfig { .. })
         ));
         assert_eq!(dn_opts.storage.providers.len(), 2);
         assert!(matches!(
             dn_opts.storage.providers[0],
-            datanode::config::ObjectStoreConfig::Gcs(GcsConfig { .. })
+            object_store::config::ObjectStoreConfig::Gcs(GcsConfig { .. })
         ));
         match &dn_opts.storage.providers[1] {
-            datanode::config::ObjectStoreConfig::S3(s3_config) => {
+            object_store::config::ObjectStoreConfig::S3(s3_config) => {
                 assert_eq!(
                     "SecretBox<alloc::string::String>([REDACTED])".to_string(),
                     format!("{:?}", s3_config.access_key_id)

src/cmd/tests/load_config_test.rs

@@ -18,17 +18,19 @@ use cmd::options::GreptimeOptions;
 use cmd::standalone::StandaloneOptions;
 use common_config::{Configurable, DEFAULT_DATA_HOME};
 use common_options::datanode::{ClientOptions, DatanodeClientOptions};
-use common_telemetry::logging::{LoggingOptions, DEFAULT_LOGGING_DIR, DEFAULT_OTLP_ENDPOINT};
+use common_telemetry::logging::{LoggingOptions, DEFAULT_LOGGING_DIR, DEFAULT_OTLP_HTTP_ENDPOINT};
 use common_wal::config::raft_engine::RaftEngineConfig;
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
 use file_engine::config::EngineConfig as FileEngineConfig;
+use flow::FlownodeOptions;
 use frontend::frontend::FrontendOptions;
 use meta_client::MetaClientOptions;
 use meta_srv::metasrv::MetasrvOptions;
 use meta_srv::selector::SelectorType;
 use metric_engine::config::EngineConfig as MetricEngineConfig;
 use mito2::config::MitoConfig;
+use query::options::QueryOptions;
 use servers::export_metrics::ExportMetricsOption;
 use servers::grpc::GrpcOptions;
 use servers::http::HttpOptions;
@@ -81,7 +83,7 @@ fn test_load_datanode_example_config() {
             logging: LoggingOptions {
                 level: Some("info".to_string()),
                 dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
-                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
                 tracing_sample_ratio: Some(Default::default()),
                 ..Default::default()
             },
@@ -124,7 +126,7 @@ fn test_load_frontend_example_config() {
             logging: LoggingOptions {
                 level: Some("info".to_string()),
                 dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
-                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
                 tracing_sample_ratio: Some(Default::default()),
                 ..Default::default()
             },
@@ -172,7 +174,7 @@ fn test_load_metasrv_example_config() {
             logging: LoggingOptions {
                 dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
                 level: Some("info".to_string()),
-                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
                 tracing_sample_ratio: Some(Default::default()),
                 ..Default::default()
             },
@@ -195,6 +197,57 @@ fn test_load_metasrv_example_config() {
     similar_asserts::assert_eq!(options, expected);
 }
 
+#[test]
+fn test_load_flownode_example_config() {
+    let example_config = common_test_util::find_workspace_path("config/flownode.example.toml");
+    let options =
+        GreptimeOptions::<FlownodeOptions>::load_layered_options(example_config.to_str(), "")
+            .unwrap();
+    let expected = GreptimeOptions::<FlownodeOptions> {
+        component: FlownodeOptions {
+            node_id: Some(14),
+            flow: Default::default(),
+            grpc: GrpcOptions {
+                bind_addr: "127.0.0.1:6800".to_string(),
+                server_addr: "127.0.0.1:6800".to_string(),
+                runtime_size: 2,
+                ..Default::default()
+            },
+            logging: LoggingOptions {
+                dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
+                level: Some("info".to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
+                otlp_export_protocol: Some(common_telemetry::logging::OtlpExportProtocol::Http),
+                tracing_sample_ratio: Some(Default::default()),
+                ..Default::default()
+            },
+            tracing: Default::default(),
+            heartbeat: Default::default(),
+            // flownode deliberately use a slower query parallelism
+            // to avoid overwhelming the frontend with too many queries
+            query: QueryOptions { parallelism: 1 },
+            meta_client: Some(MetaClientOptions {
+                metasrv_addrs: vec!["127.0.0.1:3002".to_string()],
+                timeout: Duration::from_secs(3),
+                heartbeat_timeout: Duration::from_millis(500),
+                ddl_timeout: Duration::from_secs(10),
+                connect_timeout: Duration::from_secs(1),
+                tcp_nodelay: true,
+                metadata_cache_max_capacity: 100000,
+                metadata_cache_ttl: Duration::from_secs(600),
+                metadata_cache_tti: Duration::from_secs(300),
+            }),
+            http: HttpOptions {
+                addr: "127.0.0.1:4000".to_string(),
+                ..Default::default()
+            },
+            user_provider: None,
+        },
+        ..Default::default()
+    };
+    similar_asserts::assert_eq!(options, expected);
+}
+
 #[test]
 fn test_load_standalone_example_config() {
     let example_config = common_test_util::find_workspace_path("config/standalone.example.toml");
@@ -229,7 +282,7 @@ fn test_load_standalone_example_config() {
             logging: LoggingOptions {
                 level: Some("info".to_string()),
                 dir: format!("{}/{}", DEFAULT_DATA_HOME, DEFAULT_LOGGING_DIR),
-                otlp_endpoint: Some(DEFAULT_OTLP_ENDPOINT.to_string()),
+                otlp_endpoint: Some(DEFAULT_OTLP_HTTP_ENDPOINT.to_string()),
                 tracing_sample_ratio: Some(Default::default()),
                 ..Default::default()
             },

src/common/catalog/src/consts.rs

@@ -78,7 +78,7 @@ pub const INFORMATION_SCHEMA_ROUTINES_TABLE_ID: u32 = 21;
 pub const INFORMATION_SCHEMA_SCHEMA_PRIVILEGES_TABLE_ID: u32 = 22;
 /// id for information_schema.TABLE_PRIVILEGES
 pub const INFORMATION_SCHEMA_TABLE_PRIVILEGES_TABLE_ID: u32 = 23;
-/// id for information_schema.TRIGGERS
+/// id for information_schema.TRIGGERS (for mysql)
 pub const INFORMATION_SCHEMA_TRIGGERS_TABLE_ID: u32 = 24;
 /// id for information_schema.GLOBAL_STATUS
 pub const INFORMATION_SCHEMA_GLOBAL_STATUS_TABLE_ID: u32 = 25;
@@ -104,6 +104,8 @@ pub const INFORMATION_SCHEMA_PROCEDURE_INFO_TABLE_ID: u32 = 34;
 pub const INFORMATION_SCHEMA_REGION_STATISTICS_TABLE_ID: u32 = 35;
 /// id for information_schema.process_list
 pub const INFORMATION_SCHEMA_PROCESS_LIST_TABLE_ID: u32 = 36;
+/// id for information_schema.trigger_list (for greptimedb trigger)
+pub const INFORMATION_SCHEMA_TRIGGER_TABLE_ID: u32 = 37;
 
 // ----- End of information_schema tables -----
 

src/common/config/Cargo.toml

@@ -14,6 +14,7 @@ common-macro.workspace = true
 config.workspace = true
 humantime-serde.workspace = true
 num_cpus.workspace = true
+object-store.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 serde_with.workspace = true

src/common/config/src/config.rs

@@ -106,7 +106,7 @@ mod tests {
     use common_telemetry::logging::LoggingOptions;
     use common_test_util::temp_dir::create_named_temp_file;
     use common_wal::config::DatanodeWalConfig;
-    use datanode::config::{ObjectStoreConfig, StorageConfig};
+    use datanode::config::StorageConfig;
     use meta_client::MetaClientOptions;
     use serde::{Deserialize, Serialize};
 
@@ -212,7 +212,7 @@ mod tests {
 
                 // Check the configs from environment variables.
                 match &opts.storage.store {
-                    ObjectStoreConfig::S3(s3_config) => {
+                    object_store::config::ObjectStoreConfig::S3(s3_config) => {
                         assert_eq!(s3_config.bucket, "mybucket".to_string());
                     }
                     _ => panic!("unexpected store type"),

src/common/error/src/status_code.rs

@@ -119,6 +119,11 @@ pub enum StatusCode {
     FlowAlreadyExists = 8000,
     FlowNotFound = 8001,
     // ====== End of flow related status code =====
+
+    // ====== Begin of trigger related status code =====
+    TriggerAlreadyExists = 9000,
+    TriggerNotFound = 9001,
+    // ====== End of trigger related status code =====
 }
 
 impl StatusCode {
@@ -155,6 +160,8 @@ impl StatusCode {
             | StatusCode::RegionNotFound
             | StatusCode::FlowAlreadyExists
             | StatusCode::FlowNotFound
+            | StatusCode::TriggerAlreadyExists
+            | StatusCode::TriggerNotFound
             | StatusCode::RegionReadonly
             | StatusCode::TableColumnNotFound
             | StatusCode::TableColumnExists
@@ -198,6 +205,8 @@ impl StatusCode {
             | StatusCode::PlanQuery
             | StatusCode::FlowAlreadyExists
             | StatusCode::FlowNotFound
+            | StatusCode::TriggerAlreadyExists
+            | StatusCode::TriggerNotFound
             | StatusCode::RegionNotReady
             | StatusCode::RegionBusy
             | StatusCode::RegionReadonly
@@ -230,6 +239,48 @@ impl fmt::Display for StatusCode {
     }
 }
 
+#[macro_export]
+macro_rules! define_from_tonic_status {
+    ($Error: ty, $Variant: ident) => {
+        impl From<tonic::Status> for $Error {
+            fn from(e: tonic::Status) -> Self {
+                use snafu::location;
+
+                fn metadata_value(e: &tonic::Status, key: &str) -> Option<String> {
+                    e.metadata()
+                        .get(key)
+                        .and_then(|v| String::from_utf8(v.as_bytes().to_vec()).ok())
+                }
+
+                let code = metadata_value(&e, $crate::GREPTIME_DB_HEADER_ERROR_CODE)
+                    .and_then(|s| {
+                        if let Ok(code) = s.parse::<u32>() {
+                            StatusCode::from_u32(code)
+                        } else {
+                            None
+                        }
+                    })
+                    .unwrap_or_else(|| match e.code() {
+                        tonic::Code::Cancelled => StatusCode::Cancelled,
+                        tonic::Code::DeadlineExceeded => StatusCode::DeadlineExceeded,
+                        _ => StatusCode::Internal,
+                    });
+
+                let msg = metadata_value(&e, $crate::GREPTIME_DB_HEADER_ERROR_MSG)
+                    .unwrap_or_else(|| e.message().to_string());
+
+                // TODO(LFC): Make the error variant defined automatically.
+                Self::$Variant {
+                    code,
+                    msg,
+                    tonic_code: e.code(),
+                    location: location!(),
+                }
+            }
+        }
+    };
+}
+
 #[macro_export]
 macro_rules! define_into_tonic_status {
     ($Error: ty) => {
@@ -281,12 +332,14 @@ pub fn status_to_tonic_code(status_code: StatusCode) -> Code {
         | StatusCode::TableColumnExists
         | StatusCode::RegionAlreadyExists
         | StatusCode::DatabaseAlreadyExists
+        | StatusCode::TriggerAlreadyExists
         | StatusCode::FlowAlreadyExists => Code::AlreadyExists,
         StatusCode::TableNotFound
         | StatusCode::RegionNotFound
         | StatusCode::TableColumnNotFound
         | StatusCode::DatabaseNotFound
         | StatusCode::UserNotFound
+        | StatusCode::TriggerNotFound
         | StatusCode::FlowNotFound => Code::NotFound,
         StatusCode::TableUnavailable
         | StatusCode::StorageUnavailable
@@ -304,15 +357,6 @@ pub fn status_to_tonic_code(status_code: StatusCode) -> Code {
     }
 }
 
-/// Converts tonic [Code] to [StatusCode].
-pub fn convert_tonic_code_to_status_code(code: Code) -> StatusCode {
-    match code {
-        Code::Cancelled => StatusCode::Cancelled,
-        Code::DeadlineExceeded => StatusCode::DeadlineExceeded,
-        _ => StatusCode::Internal,
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use strum::IntoEnumIterator;

src/common/frontend/src/lib.rs

@@ -23,7 +23,7 @@ pub mod selector;
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct DisplayProcessId {
     pub server_addr: String,
-    pub id: u64,
+    pub id: u32,
 }
 
 impl Display for DisplayProcessId {
@@ -44,7 +44,7 @@ impl TryFrom<&str> for DisplayProcessId {
         let id = split
             .next()
             .context(error::ParseProcessIdSnafu { s: value })?;
-        let id = u64::from_str(id)
+        let id = u32::from_str(id)
             .ok()
             .context(error::ParseProcessIdSnafu { s: value })?;
         Ok(DisplayProcessId { server_addr, id })

src/common/frontend/src/selector.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt::Debug;
 use std::time::Duration;
 
 use common_grpc::channel_manager::{ChannelConfig, ChannelManager};
@@ -30,7 +31,7 @@ use crate::error::{MetaSnafu, Result};
 pub type FrontendClientPtr = Box<dyn FrontendClient>;
 
 #[async_trait::async_trait]
-pub trait FrontendClient: Send {
+pub trait FrontendClient: Send + Debug {
     async fn list_process(&mut self, req: ListProcessRequest) -> Result<ListProcessResponse>;
 
     async fn kill_process(&mut self, req: KillProcessRequest) -> Result<KillProcessResponse>;

src/common/function/Cargo.toml

@@ -33,6 +33,7 @@ common-version.workspace = true
 datafusion.workspace = true
 datafusion-common.workspace = true
 datafusion-expr.workspace = true
+datafusion-functions-aggregate-common.workspace = true
 datatypes.workspace = true
 derive_more = { version = "1", default-features = false, features = ["display"] }
 geo = { version = "0.29", optional = true }

src/common/function/src/aggrs.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub mod approximate;
+pub mod count_hash;
 #[cfg(feature = "geo")]
 pub mod geo;
 pub mod vector;

src/common/function/src/aggrs/approximate.rs

@@ -14,8 +14,8 @@
 
 use crate::function_registry::FunctionRegistry;
 
-pub(crate) mod hll;
+pub mod hll;
-mod uddsketch;
+pub mod uddsketch;
 
 pub(crate) struct ApproximateFunction;
 

src/common/function/src/aggrs/count_hash.rs

@@ -0,0 +1,647 @@
+// 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.
+
+//! `CountHash` / `count_hash` is a hash-based approximate distinct count function.
+//!
+//! It is a variant of `CountDistinct` that uses a hash function to approximate the
+//! distinct count.
+//! It is designed to be more efficient than `CountDistinct` for large datasets,
+//! but it is not as accurate, as the hash value may be collision.
+
+use std::collections::HashSet;
+use std::fmt::Debug;
+use std::sync::Arc;
+
+use ahash::RandomState;
+use datafusion_common::cast::as_list_array;
+use datafusion_common::error::Result;
+use datafusion_common::hash_utils::create_hashes;
+use datafusion_common::utils::SingleRowListArrayBuilder;
+use datafusion_common::{internal_err, not_impl_err, ScalarValue};
+use datafusion_expr::function::{AccumulatorArgs, StateFieldsArgs};
+use datafusion_expr::utils::{format_state_name, AggregateOrderSensitivity};
+use datafusion_expr::{
+    Accumulator, AggregateUDF, AggregateUDFImpl, EmitTo, GroupsAccumulator, ReversedUDAF,
+    SetMonotonicity, Signature, TypeSignature, Volatility,
+};
+use datafusion_functions_aggregate_common::aggregate::groups_accumulator::nulls::filtered_null_mask;
+use datatypes::arrow;
+use datatypes::arrow::array::{
+    Array, ArrayRef, AsArray, BooleanArray, Int64Array, ListArray, UInt64Array,
+};
+use datatypes::arrow::buffer::{OffsetBuffer, ScalarBuffer};
+use datatypes::arrow::datatypes::{DataType, Field};
+
+use crate::function_registry::FunctionRegistry;
+
+type HashValueType = u64;
+
+// read from /dev/urandom 4047821dc6144e4b2abddf23ad4171126a52eeecd26eff2191cf673b965a7875
+const RANDOM_SEED_0: u64 = 0x4047821dc6144e4b;
+const RANDOM_SEED_1: u64 = 0x2abddf23ad417112;
+const RANDOM_SEED_2: u64 = 0x6a52eeecd26eff21;
+const RANDOM_SEED_3: u64 = 0x91cf673b965a7875;
+
+impl CountHash {
+    pub fn register(registry: &FunctionRegistry) {
+        registry.register_aggr(CountHash::udf_impl());
+    }
+
+    pub fn udf_impl() -> AggregateUDF {
+        AggregateUDF::new_from_impl(CountHash {
+            signature: Signature::one_of(
+                vec![TypeSignature::VariadicAny, TypeSignature::Nullary],
+                Volatility::Immutable,
+            ),
+        })
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct CountHash {
+    signature: Signature,
+}
+
+impl AggregateUDFImpl for CountHash {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn name(&self) -> &str {
+        "count_hash"
+    }
+
+    fn signature(&self) -> &Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, _arg_types: &[DataType]) -> Result<DataType> {
+        Ok(DataType::Int64)
+    }
+
+    fn is_nullable(&self) -> bool {
+        false
+    }
+
+    fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<Field>> {
+        Ok(vec![Field::new_list(
+            format_state_name(args.name, "count_hash"),
+            Field::new_list_field(DataType::UInt64, true),
+            // For count_hash accumulator, null list item stands for an
+            // empty value set (i.e., all NULL value so far for that group).
+            true,
+        )])
+    }
+
+    fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
+        if acc_args.exprs.len() > 1 {
+            return not_impl_err!("count_hash with multiple arguments");
+        }
+
+        Ok(Box::new(CountHashAccumulator {
+            values: HashSet::default(),
+            random_state: RandomState::with_seeds(
+                RANDOM_SEED_0,
+                RANDOM_SEED_1,
+                RANDOM_SEED_2,
+                RANDOM_SEED_3,
+            ),
+            batch_hashes: vec![],
+        }))
+    }
+
+    fn aliases(&self) -> &[String] {
+        &[]
+    }
+
+    fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool {
+        true
+    }
+
+    fn create_groups_accumulator(
+        &self,
+        args: AccumulatorArgs,
+    ) -> Result<Box<dyn GroupsAccumulator>> {
+        if args.exprs.len() > 1 {
+            return not_impl_err!("count_hash with multiple arguments");
+        }
+
+        Ok(Box::new(CountHashGroupAccumulator::new()))
+    }
+
+    fn reverse_expr(&self) -> ReversedUDAF {
+        ReversedUDAF::Identical
+    }
+
+    fn order_sensitivity(&self) -> AggregateOrderSensitivity {
+        AggregateOrderSensitivity::Insensitive
+    }
+
+    fn default_value(&self, _data_type: &DataType) -> Result<ScalarValue> {
+        Ok(ScalarValue::Int64(Some(0)))
+    }
+
+    fn set_monotonicity(&self, _data_type: &DataType) -> SetMonotonicity {
+        SetMonotonicity::Increasing
+    }
+}
+
+/// GroupsAccumulator for `count_hash` aggregate function
+#[derive(Debug)]
+pub struct CountHashGroupAccumulator {
+    /// One HashSet per group to track distinct values
+    distinct_sets: Vec<HashSet<HashValueType, RandomState>>,
+    random_state: RandomState,
+    batch_hashes: Vec<HashValueType>,
+}
+
+impl Default for CountHashGroupAccumulator {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl CountHashGroupAccumulator {
+    pub fn new() -> Self {
+        Self {
+            distinct_sets: vec![],
+            random_state: RandomState::with_seeds(
+                RANDOM_SEED_0,
+                RANDOM_SEED_1,
+                RANDOM_SEED_2,
+                RANDOM_SEED_3,
+            ),
+            batch_hashes: vec![],
+        }
+    }
+
+    fn ensure_sets(&mut self, total_num_groups: usize) {
+        if self.distinct_sets.len() < total_num_groups {
+            self.distinct_sets
+                .resize_with(total_num_groups, HashSet::default);
+        }
+    }
+}
+
+impl GroupsAccumulator for CountHashGroupAccumulator {
+    fn update_batch(
+        &mut self,
+        values: &[ArrayRef],
+        group_indices: &[usize],
+        opt_filter: Option<&BooleanArray>,
+        total_num_groups: usize,
+    ) -> Result<()> {
+        assert_eq!(values.len(), 1, "count_hash expects a single argument");
+        self.ensure_sets(total_num_groups);
+
+        let array = &values[0];
+        self.batch_hashes.clear();
+        self.batch_hashes.resize(array.len(), 0);
+        let hashes = create_hashes(
+            &[ArrayRef::clone(array)],
+            &self.random_state,
+            &mut self.batch_hashes,
+        )?;
+
+        // Use a pattern similar to accumulate_indices to process rows
+        // that are not null and pass the filter
+        let nulls = array.logical_nulls();
+
+        match (nulls.as_ref(), opt_filter) {
+            (None, None) => {
+                // No nulls, no filter - process all rows
+                for (row_idx, &group_idx) in group_indices.iter().enumerate() {
+                    self.distinct_sets[group_idx].insert(hashes[row_idx]);
+                }
+            }
+            (Some(nulls), None) => {
+                // Has nulls, no filter
+                for (row_idx, (&group_idx, is_valid)) in
+                    group_indices.iter().zip(nulls.iter()).enumerate()
+                {
+                    if is_valid {
+                        self.distinct_sets[group_idx].insert(hashes[row_idx]);
+                    }
+                }
+            }
+            (None, Some(filter)) => {
+                // No nulls, has filter
+                for (row_idx, (&group_idx, filter_value)) in
+                    group_indices.iter().zip(filter.iter()).enumerate()
+                {
+                    if let Some(true) = filter_value {
+                        self.distinct_sets[group_idx].insert(hashes[row_idx]);
+                    }
+                }
+            }
+            (Some(nulls), Some(filter)) => {
+                // Has nulls and filter
+                let iter = filter
+                    .iter()
+                    .zip(group_indices.iter())
+                    .zip(nulls.iter())
+                    .enumerate();
+
+                for (row_idx, ((filter_value, &group_idx), is_valid)) in iter {
+                    if is_valid && filter_value == Some(true) {
+                        self.distinct_sets[group_idx].insert(hashes[row_idx]);
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    fn evaluate(&mut self, emit_to: EmitTo) -> Result<ArrayRef> {
+        let distinct_sets: Vec<HashSet<u64, RandomState>> =
+            emit_to.take_needed(&mut self.distinct_sets);
+
+        let counts = distinct_sets
+            .iter()
+            .map(|set| set.len() as i64)
+            .collect::<Vec<_>>();
+        Ok(Arc::new(Int64Array::from(counts)))
+    }
+
+    fn merge_batch(
+        &mut self,
+        values: &[ArrayRef],
+        group_indices: &[usize],
+        _opt_filter: Option<&BooleanArray>,
+        total_num_groups: usize,
+    ) -> Result<()> {
+        assert_eq!(
+            values.len(),
+            1,
+            "count_hash merge expects a single state array"
+        );
+        self.ensure_sets(total_num_groups);
+
+        let list_array = as_list_array(&values[0])?;
+
+        // For each group in the incoming batch
+        for (i, &group_idx) in group_indices.iter().enumerate() {
+            if i < list_array.len() {
+                let inner_array = list_array.value(i);
+                let inner_array = inner_array.as_any().downcast_ref::<UInt64Array>().unwrap();
+                // Add each value to our set for this group
+                for j in 0..inner_array.len() {
+                    if !inner_array.is_null(j) {
+                        self.distinct_sets[group_idx].insert(inner_array.value(j));
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    fn state(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
+        let distinct_sets: Vec<HashSet<u64, RandomState>> =
+            emit_to.take_needed(&mut self.distinct_sets);
+
+        let mut offsets = Vec::with_capacity(distinct_sets.len() + 1);
+        offsets.push(0);
+        let mut curr_len = 0i32;
+
+        let mut value_iter = distinct_sets
+            .into_iter()
+            .flat_map(|set| {
+                // build offset
+                curr_len += set.len() as i32;
+                offsets.push(curr_len);
+                // convert into iter
+                set.into_iter()
+            })
+            .peekable();
+        let data_array: ArrayRef = if value_iter.peek().is_none() {
+            arrow::array::new_empty_array(&DataType::UInt64) as _
+        } else {
+            Arc::new(UInt64Array::from_iter_values(value_iter))
+        };
+        let offset_buffer = OffsetBuffer::new(ScalarBuffer::from(offsets));
+
+        let list_array = ListArray::new(
+            Arc::new(Field::new_list_field(DataType::UInt64, true)),
+            offset_buffer,
+            data_array,
+            None,
+        );
+
+        Ok(vec![Arc::new(list_array) as _])
+    }
+
+    fn convert_to_state(
+        &self,
+        values: &[ArrayRef],
+        opt_filter: Option<&BooleanArray>,
+    ) -> Result<Vec<ArrayRef>> {
+        // For a single hash value per row, create a list array with that value
+        assert_eq!(values.len(), 1, "count_hash expects a single argument");
+        let values = ArrayRef::clone(&values[0]);
+
+        let offsets = OffsetBuffer::new(ScalarBuffer::from_iter(0..values.len() as i32 + 1));
+        let nulls = filtered_null_mask(opt_filter, &values);
+        let list_array = ListArray::new(
+            Arc::new(Field::new_list_field(DataType::UInt64, true)),
+            offsets,
+            values,
+            nulls,
+        );
+
+        Ok(vec![Arc::new(list_array)])
+    }
+
+    fn supports_convert_to_state(&self) -> bool {
+        true
+    }
+
+    fn size(&self) -> usize {
+        // Base size of the struct
+        let mut size = size_of::<Self>();
+
+        // Size of the vector holding the HashSets
+        size += size_of::<Vec<HashSet<HashValueType, RandomState>>>()
+            + self.distinct_sets.capacity() * size_of::<HashSet<HashValueType, RandomState>>();
+
+        // Estimate HashSet contents size more efficiently
+        // Instead of iterating through all values which is expensive, use an approximation
+        for set in &self.distinct_sets {
+            // Base size of the HashSet
+            size += set.capacity() * size_of::<HashValueType>();
+        }
+
+        size
+    }
+}
+
+#[derive(Debug)]
+struct CountHashAccumulator {
+    values: HashSet<HashValueType, RandomState>,
+    random_state: RandomState,
+    batch_hashes: Vec<HashValueType>,
+}
+
+impl CountHashAccumulator {
+    // calculating the size for fixed length values, taking first batch size *
+    // number of batches.
+    fn fixed_size(&self) -> usize {
+        size_of_val(self) + (size_of::<HashValueType>() * self.values.capacity())
+    }
+}
+
+impl Accumulator for CountHashAccumulator {
+    /// Returns the distinct values seen so far as (one element) ListArray.
+    fn state(&mut self) -> Result<Vec<ScalarValue>> {
+        let values = self.values.iter().cloned().collect::<Vec<_>>();
+        let arr = Arc::new(UInt64Array::from(values)) as _;
+        let list_scalar = SingleRowListArrayBuilder::new(arr).build_list_scalar();
+        Ok(vec![list_scalar])
+    }
+
+    fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+        if values.is_empty() {
+            return Ok(());
+        }
+
+        let arr = &values[0];
+        if arr.data_type() == &DataType::Null {
+            return Ok(());
+        }
+
+        self.batch_hashes.clear();
+        self.batch_hashes.resize(arr.len(), 0);
+        let hashes = create_hashes(
+            &[ArrayRef::clone(arr)],
+            &self.random_state,
+            &mut self.batch_hashes,
+        )?;
+        for hash in hashes.as_slice() {
+            self.values.insert(*hash);
+        }
+        Ok(())
+    }
+
+    /// Merges multiple sets of distinct values into the current set.
+    ///
+    /// The input to this function is a `ListArray` with **multiple** rows,
+    /// where each row contains the values from a partial aggregate's phase (e.g.
+    /// the result of calling `Self::state` on multiple accumulators).
+    fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+        if states.is_empty() {
+            return Ok(());
+        }
+        assert_eq!(states.len(), 1, "array_agg states must be singleton!");
+        let array = &states[0];
+        let list_array = array.as_list::<i32>();
+        for inner_array in list_array.iter() {
+            let Some(inner_array) = inner_array else {
+                return internal_err!(
+                    "Intermediate results of count_hash should always be non null"
+                );
+            };
+            let hash_array = inner_array.as_any().downcast_ref::<UInt64Array>().unwrap();
+            for i in 0..hash_array.len() {
+                self.values.insert(hash_array.value(i));
+            }
+        }
+        Ok(())
+    }
+
+    fn evaluate(&mut self) -> Result<ScalarValue> {
+        Ok(ScalarValue::Int64(Some(self.values.len() as i64)))
+    }
+
+    fn size(&self) -> usize {
+        self.fixed_size()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use datatypes::arrow::array::{Array, BooleanArray, Int32Array, Int64Array};
+
+    use super::*;
+
+    fn create_test_accumulator() -> CountHashAccumulator {
+        CountHashAccumulator {
+            values: HashSet::default(),
+            random_state: RandomState::with_seeds(
+                RANDOM_SEED_0,
+                RANDOM_SEED_1,
+                RANDOM_SEED_2,
+                RANDOM_SEED_3,
+            ),
+            batch_hashes: vec![],
+        }
+    }
+
+    #[test]
+    fn test_count_hash_accumulator() -> Result<()> {
+        let mut acc = create_test_accumulator();
+
+        // Test with some data
+        let array = Arc::new(Int32Array::from(vec![
+            Some(1),
+            Some(2),
+            Some(3),
+            Some(1),
+            Some(2),
+            None,
+        ])) as ArrayRef;
+        acc.update_batch(&[array])?;
+        let result = acc.evaluate()?;
+        assert_eq!(result, ScalarValue::Int64(Some(4)));
+
+        // Test with empty data
+        let mut acc = create_test_accumulator();
+        let array = Arc::new(Int32Array::from(vec![] as Vec<Option<i32>>)) as ArrayRef;
+        acc.update_batch(&[array])?;
+        let result = acc.evaluate()?;
+        assert_eq!(result, ScalarValue::Int64(Some(0)));
+
+        // Test with only nulls
+        let mut acc = create_test_accumulator();
+        let array = Arc::new(Int32Array::from(vec![None, None, None])) as ArrayRef;
+        acc.update_batch(&[array])?;
+        let result = acc.evaluate()?;
+        assert_eq!(result, ScalarValue::Int64(Some(1)));
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_count_hash_accumulator_merge() -> Result<()> {
+        // Accumulator 1
+        let mut acc1 = create_test_accumulator();
+        let array1 = Arc::new(Int32Array::from(vec![Some(1), Some(2), Some(3)])) as ArrayRef;
+        acc1.update_batch(&[array1])?;
+        let state1 = acc1.state()?;
+
+        // Accumulator 2
+        let mut acc2 = create_test_accumulator();
+        let array2 = Arc::new(Int32Array::from(vec![Some(3), Some(4), Some(5)])) as ArrayRef;
+        acc2.update_batch(&[array2])?;
+        let state2 = acc2.state()?;
+
+        // Merge state1 and state2 into a new accumulator
+        let mut acc_merged = create_test_accumulator();
+        let state_array1 = state1[0].to_array()?;
+        let state_array2 = state2[0].to_array()?;
+
+        acc_merged.merge_batch(&[state_array1])?;
+        acc_merged.merge_batch(&[state_array2])?;
+
+        let result = acc_merged.evaluate()?;
+        // Distinct values are {1, 2, 3, 4, 5}, so count is 5
+        assert_eq!(result, ScalarValue::Int64(Some(5)));
+
+        Ok(())
+    }
+
+    fn create_test_group_accumulator() -> CountHashGroupAccumulator {
+        CountHashGroupAccumulator::new()
+    }
+
+    #[test]
+    fn test_count_hash_group_accumulator() -> Result<()> {
+        let mut acc = create_test_group_accumulator();
+        let values = Arc::new(Int32Array::from(vec![1, 2, 1, 3, 2, 4, 5])) as ArrayRef;
+        let group_indices = vec![0, 1, 0, 0, 1, 2, 0];
+        let total_num_groups = 3;
+
+        acc.update_batch(&[values], &group_indices, None, total_num_groups)?;
+
+        let result_array = acc.evaluate(EmitTo::All)?;
+        let result = result_array.as_any().downcast_ref::<Int64Array>().unwrap();
+
+        // Group 0: {1, 3, 5} -> 3
+        // Group 1: {2} -> 1
+        // Group 2: {4} -> 1
+        assert_eq!(result.value(0), 3);
+        assert_eq!(result.value(1), 1);
+        assert_eq!(result.value(2), 1);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_count_hash_group_accumulator_with_filter() -> Result<()> {
+        let mut acc = create_test_group_accumulator();
+        let values = Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5, 6])) as ArrayRef;
+        let group_indices = vec![0, 0, 1, 1, 2, 2];
+        let filter = BooleanArray::from(vec![true, false, true, true, false, true]);
+        let total_num_groups = 3;
+
+        acc.update_batch(&[values], &group_indices, Some(&filter), total_num_groups)?;
+
+        let result_array = acc.evaluate(EmitTo::All)?;
+        let result = result_array.as_any().downcast_ref::<Int64Array>().unwrap();
+
+        // Group 0: {1} (2 is filtered out) -> 1
+        // Group 1: {3, 4} -> 2
+        // Group 2: {6} (5 is filtered out) -> 1
+        assert_eq!(result.value(0), 1);
+        assert_eq!(result.value(1), 2);
+        assert_eq!(result.value(2), 1);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_count_hash_group_accumulator_merge() -> Result<()> {
+        // Accumulator 1
+        let mut acc1 = create_test_group_accumulator();
+        let values1 = Arc::new(Int32Array::from(vec![1, 2, 3, 4])) as ArrayRef;
+        let group_indices1 = vec![0, 0, 1, 1];
+        acc1.update_batch(&[values1], &group_indices1, None, 2)?;
+        // acc1 state: group 0 -> {1, 2}, group 1 -> {3, 4}
+        let state1 = acc1.state(EmitTo::All)?;
+
+        // Accumulator 2
+        let mut acc2 = create_test_group_accumulator();
+        let values2 = Arc::new(Int32Array::from(vec![5, 6, 1, 3])) as ArrayRef;
+        // Merge into different group indices
+        let group_indices2 = vec![2, 2, 0, 1];
+        acc2.update_batch(&[values2], &group_indices2, None, 3)?;
+        // acc2 state: group 0 -> {1}, group 1 -> {3}, group 2 -> {5, 6}
+
+        // Merge state from acc1 into acc2
+        // We will merge acc1's group 0 into acc2's group 0
+        // and acc1's group 1 into acc2's group 2
+        let merge_group_indices = vec![0, 2];
+        acc2.merge_batch(&state1, &merge_group_indices, None, 3)?;
+
+        let result_array = acc2.evaluate(EmitTo::All)?;
+        let result = result_array.as_any().downcast_ref::<Int64Array>().unwrap();
+
+        // Final state of acc2:
+        // Group 0: {1} U {1, 2} -> {1, 2}, count = 2
+        // Group 1: {3}, count = 1
+        // Group 2: {5, 6} U {3, 4} -> {3, 4, 5, 6}, count = 4
+        assert_eq!(result.value(0), 2);
+        assert_eq!(result.value(1), 1);
+        assert_eq!(result.value(2), 4);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_size() {
+        let acc = create_test_group_accumulator();
+        // Just test it doesn't crash and returns a value.
+        assert!(acc.size() > 0);
+    }
+}

src/common/function/src/function_registry.rs

@@ -21,6 +21,7 @@ use once_cell::sync::Lazy;
 
 use crate::admin::AdminFunction;
 use crate::aggrs::approximate::ApproximateFunction;
+use crate::aggrs::count_hash::CountHash;
 use crate::aggrs::vector::VectorFunction as VectorAggrFunction;
 use crate::function::{AsyncFunctionRef, Function, FunctionRef};
 use crate::function_factory::ScalarFunctionFactory;
@@ -144,6 +145,9 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
     // Approximate functions
     ApproximateFunction::register(&function_registry);
 
+    // CountHash function
+    CountHash::register(&function_registry);
+
     Arc::new(function_registry)
 });
 

src/common/function/src/scalars/math.rs

@@ -14,7 +14,6 @@
 
 pub mod clamp;
 mod modulo;
-mod pow;
 mod rate;
 
 use std::fmt;
@@ -26,7 +25,6 @@ use datafusion::error::DataFusionError;
 use datafusion::logical_expr::Volatility;
 use datatypes::prelude::ConcreteDataType;
 use datatypes::vectors::VectorRef;
-pub use pow::PowFunction;
 pub use rate::RateFunction;
 use snafu::ResultExt;
 
@@ -39,7 +37,6 @@ pub(crate) struct MathFunction;
 impl MathFunction {
     pub fn register(registry: &FunctionRegistry) {
         registry.register_scalar(ModuloFunction);
-        registry.register_scalar(PowFunction);
         registry.register_scalar(RateFunction);
         registry.register_scalar(RangeFunction);
         registry.register_scalar(ClampFunction);

src/common/function/src/scalars/math/pow.rs

@@ -1,120 +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;
-use std::sync::Arc;
-
-use common_query::error::Result;
-use common_query::prelude::{Signature, Volatility};
-use datatypes::data_type::DataType;
-use datatypes::prelude::ConcreteDataType;
-use datatypes::types::LogicalPrimitiveType;
-use datatypes::vectors::VectorRef;
-use datatypes::with_match_primitive_type_id;
-use num::traits::Pow;
-use num_traits::AsPrimitive;
-
-use crate::function::{Function, FunctionContext};
-use crate::scalars::expression::{scalar_binary_op, EvalContext};
-
-#[derive(Clone, Debug, Default)]
-pub struct PowFunction;
-
-impl Function for PowFunction {
-    fn name(&self) -> &str {
-        "pow"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::float64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::uniform(2, ConcreteDataType::numerics(), Volatility::Immutable)
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        with_match_primitive_type_id!(columns[0].data_type().logical_type_id(), |$S| {
-            with_match_primitive_type_id!(columns[1].data_type().logical_type_id(), |$T| {
-                let col = scalar_binary_op::<<$S as LogicalPrimitiveType>::Native, <$T as LogicalPrimitiveType>::Native, f64, _>(&columns[0], &columns[1], scalar_pow, &mut EvalContext::default())?;
-                Ok(Arc::new(col))
-            },{
-                unreachable!()
-            })
-        },{
-            unreachable!()
-        })
-    }
-}
-
-#[inline]
-fn scalar_pow<S, T>(value: Option<S>, base: Option<T>, _ctx: &mut EvalContext) -> Option<f64>
-where
-    S: AsPrimitive<f64>,
-    T: AsPrimitive<f64>,
-{
-    match (value, base) {
-        (Some(value), Some(base)) => Some(value.as_().pow(base.as_())),
-        _ => None,
-    }
-}
-
-impl fmt::Display for PowFunction {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "POW")
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use common_query::prelude::TypeSignature;
-    use datatypes::value::Value;
-    use datatypes::vectors::{Float32Vector, Int8Vector};
-
-    use super::*;
-    use crate::function::FunctionContext;
-    #[test]
-    fn test_pow_function() {
-        let pow = PowFunction;
-
-        assert_eq!("pow", pow.name());
-        assert_eq!(
-            ConcreteDataType::float64_datatype(),
-            pow.return_type(&[]).unwrap()
-        );
-
-        assert!(matches!(pow.signature(),
-                         Signature {
-                             type_signature: TypeSignature::Uniform(2, valid_types),
-                             volatility: Volatility::Immutable
-                         } if  valid_types == ConcreteDataType::numerics()
-        ));
-
-        let values = vec![1.0, 2.0, 3.0];
-        let bases = vec![0i8, -1i8, 3i8];
-
-        let args: Vec<VectorRef> = vec![
-            Arc::new(Float32Vector::from_vec(values.clone())),
-            Arc::new(Int8Vector::from_vec(bases.clone())),
-        ];
-
-        let vector = pow.eval(&FunctionContext::default(), &args).unwrap();
-        assert_eq!(3, vector.len());
-
-        for i in 0..3 {
-            let p: f64 = (values[i] as f64).pow(bases[i] as f64);
-            assert!(matches!(vector.get(i), Value::Float64(v) if v == p));
-        }
-    }
-}

src/common/function/src/system/database.rs

@@ -18,7 +18,7 @@ use std::sync::Arc;
 use common_query::error::Result;
 use common_query::prelude::{Signature, Volatility};
 use datatypes::prelude::{ConcreteDataType, ScalarVector};
-use datatypes::vectors::{StringVector, UInt64Vector, VectorRef};
+use datatypes::vectors::{StringVector, UInt32Vector, VectorRef};
 use derive_more::Display;
 
 use crate::function::{Function, FunctionContext};
@@ -144,7 +144,7 @@ impl Function for PgBackendPidFunction {
     fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
         let pid = func_ctx.query_ctx.process_id();
 
-        Ok(Arc::new(UInt64Vector::from_slice([pid])) as _)
+        Ok(Arc::new(UInt32Vector::from_slice([pid])) as _)
     }
 }
 
@@ -164,7 +164,7 @@ impl Function for ConnectionIdFunction {
     fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
         let pid = func_ctx.query_ctx.process_id();
 
-        Ok(Arc::new(UInt64Vector::from_slice([pid])) as _)
+        Ok(Arc::new(UInt32Vector::from_slice([pid])) as _)
     }
 }
 

src/common/function/src/system/pg_catalog/version.rs

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

src/common/function/src/system/version.rs

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::fmt;
 use std::sync::Arc;
-use std::{env, fmt};
 
 use common_query::error::Result;
 use common_query::prelude::{Signature, Volatility};
@@ -52,13 +52,13 @@ impl Function for VersionFunction {
                     "{}-greptimedb-{}",
                     std::env::var("GREPTIMEDB_MYSQL_SERVER_VERSION")
                         .unwrap_or_else(|_| "8.4.2".to_string()),
-                    env!("CARGO_PKG_VERSION")
+                    common_version::version()
                 )
             }
             Channel::Postgres => {
-                format!("16.3-greptimedb-{}", env!("CARGO_PKG_VERSION"))
+                format!("16.3-greptimedb-{}", common_version::version())
             }
-            _ => env!("CARGO_PKG_VERSION").to_string(),
+            _ => common_version::version().to_string(),
         };
         let result = StringVector::from(vec![version]);
         Ok(Arc::new(result))

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

@@ -34,7 +34,7 @@ use table::requests::{
 };
 
 use crate::error::{
-    InvalidColumnDefSnafu, InvalidSetFulltextOptionRequestSnafu,
+    InvalidColumnDefSnafu, InvalidIndexOptionSnafu, InvalidSetFulltextOptionRequestSnafu,
     InvalidSetSkippingIndexOptionRequestSnafu, InvalidSetTableOptionRequestSnafu,
     InvalidUnsetTableOptionRequestSnafu, MissingAlterIndexOptionSnafu, MissingFieldSnafu,
     MissingTimestampColumnSnafu, Result, UnknownLocationTypeSnafu,
@@ -126,18 +126,21 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
                 api::v1::set_index::Options::Fulltext(f) => AlterKind::SetIndex {
                     options: SetIndexOptions::Fulltext {
                         column_name: f.column_name.clone(),
-                        options: FulltextOptions {
+                        options: FulltextOptions::new(
-                            enable: f.enable,
+                            f.enable,
-                            analyzer: as_fulltext_option_analyzer(
+                            as_fulltext_option_analyzer(
                                 Analyzer::try_from(f.analyzer)
                                     .context(InvalidSetFulltextOptionRequestSnafu)?,
                             ),
-                            case_sensitive: f.case_sensitive,
+                            f.case_sensitive,
-                            backend: as_fulltext_option_backend(
+                            as_fulltext_option_backend(
                                 PbFulltextBackend::try_from(f.backend)
                                     .context(InvalidSetFulltextOptionRequestSnafu)?,
                             ),
-                        },
+                            f.granularity as u32,
+                            f.false_positive_rate,
+                        )
+                        .context(InvalidIndexOptionSnafu)?,
                     },
                 },
                 api::v1::set_index::Options::Inverted(i) => AlterKind::SetIndex {
@@ -148,13 +151,15 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
                 api::v1::set_index::Options::Skipping(s) => AlterKind::SetIndex {
                     options: SetIndexOptions::Skipping {
                         column_name: s.column_name,
-                        options: SkippingIndexOptions {
+                        options: SkippingIndexOptions::new(
-                            granularity: s.granularity as u32,
+                            s.granularity as u32,
-                            index_type: as_skipping_index_type(
+                            s.false_positive_rate,
+                            as_skipping_index_type(
                                 PbSkippingIndexType::try_from(s.skipping_index_type)
                                     .context(InvalidSetSkippingIndexOptionRequestSnafu)?,
                             ),
-                        },
+                        )
+                        .context(InvalidIndexOptionSnafu)?,
                     },
                 },
             },
@@ -180,6 +185,22 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
             },
             None => return MissingAlterIndexOptionSnafu.fail(),
         },
+        Kind::DropDefaults(o) => {
+            let names = o
+                .drop_defaults
+                .into_iter()
+                .map(|col| {
+                    ensure!(
+                        !col.column_name.is_empty(),
+                        MissingFieldSnafu {
+                            field: "column_name"
+                        }
+                    );
+                    Ok(col.column_name)
+                })
+                .collect::<Result<Vec<_>>>()?;
+            AlterKind::DropDefaults { names }
+        }
     };
 
     let request = AlterTableRequest {

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

@@ -153,6 +153,14 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("Invalid index option"))]
+    InvalidIndexOption {
+        #[snafu(implicit)]
+        location: Location,
+        #[snafu(source)]
+        error: datatypes::error::Error,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -180,7 +188,8 @@ impl ErrorExt for Error {
             | Error::InvalidUnsetTableOptionRequest { .. }
             | Error::InvalidSetFulltextOptionRequest { .. }
             | Error::InvalidSetSkippingIndexOptionRequest { .. }
-            | Error::MissingAlterIndexOption { .. } => StatusCode::InvalidArguments,
+            | Error::MissingAlterIndexOption { .. }
+            | Error::InvalidIndexOption { .. } => StatusCode::InvalidArguments,
         }
     }
 

src/common/grpc/src/channel_manager.rs

@@ -201,8 +201,8 @@ impl ChannelManager {
             "http"
         };
 
-        let mut endpoint =
+        let mut endpoint = Endpoint::new(format!("{http_prefix}://{addr}"))
-            Endpoint::new(format!("{http_prefix}://{addr}")).context(CreateChannelSnafu)?;
+            .context(CreateChannelSnafu { addr })?;
 
         if let Some(dur) = self.config().timeout {
             endpoint = endpoint.timeout(dur);
@@ -237,7 +237,7 @@ impl ChannelManager {
         if let Some(tls_config) = &self.inner.client_tls_config {
             endpoint = endpoint
                 .tls_config(tls_config.clone())
-                .context(CreateChannelSnafu)?;
+                .context(CreateChannelSnafu { addr })?;
         }
 
         endpoint = endpoint

src/common/grpc/src/error.rs

@@ -52,8 +52,9 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Failed to create gRPC channel"))]
+    #[snafu(display("Failed to create gRPC channel from '{addr}'"))]
     CreateChannel {
+        addr: String,
         #[snafu(source)]
         error: tonic::transport::Error,
         #[snafu(implicit)]

src/common/meta/Cargo.toml

@@ -17,7 +17,7 @@ workspace = true
 anymap2 = "0.13.0"
 api.workspace = true
 async-recursion = "1.0"
-async-stream = "0.3"
+async-stream.workspace = true
 async-trait.workspace = true
 backon = { workspace = true, optional = true }
 base64.workspace = true

src/common/meta/src/ddl.rs

@@ -50,7 +50,6 @@ pub mod drop_flow;
 pub mod drop_table;
 pub mod drop_view;
 pub mod flow_meta;
-mod physical_table_metadata;
 pub mod table_meta;
 #[cfg(any(test, feature = "testing"))]
 pub mod test_util;

src/common/meta/src/ddl/alter_logical_tables.rs

@@ -12,31 +12,32 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod check;
+mod executor;
-mod metadata;
-mod region_request;
-mod table_cache_keys;
 mod update_metadata;
+mod validator;
 
 use api::region::RegionResponse;
 use async_trait::async_trait;
 use common_catalog::format_full_table_name;
 use common_procedure::error::{FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu};
 use common_procedure::{Context, LockKey, Procedure, Status};
-use common_telemetry::{error, info, warn};
+use common_telemetry::{debug, error, info, warn};
-use futures_util::future;
+pub use executor::make_alter_region_request;
 use serde::{Deserialize, Serialize};
-use snafu::{ensure, ResultExt};
+use snafu::ResultExt;
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
 use strum::AsRefStr;
 use table::metadata::TableId;
 
-use crate::ddl::utils::{
+use crate::cache_invalidator::Context as CacheContext;
-    add_peer_context_if_needed, map_to_procedure_error, sync_follower_regions,
+use crate::ddl::alter_logical_tables::executor::AlterLogicalTablesExecutor;
+use crate::ddl::alter_logical_tables::validator::{
+    retain_unskipped, AlterLogicalTableValidator, ValidatorResult,
 };
+use crate::ddl::utils::{extract_column_metadatas, map_to_procedure_error, sync_follower_regions};
 use crate::ddl::DdlContext;
-use crate::error::{DecodeJsonSnafu, MetadataCorruptionSnafu, Result};
+use crate::error::Result;
 use crate::instruction::CacheIdent;
 use crate::key::table_info::TableInfoValue;
 use crate::key::table_route::PhysicalTableRouteValue;
@@ -44,13 +45,38 @@ use crate::key::DeserializedValueWithBytes;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
 use crate::metrics;
 use crate::rpc::ddl::AlterTableTask;
-use crate::rpc::router::{find_leaders, RegionRoute};
+use crate::rpc::router::RegionRoute;
 
 pub struct AlterLogicalTablesProcedure {
     pub context: DdlContext,
     pub data: AlterTablesData,
 }
 
+/// Builds the validator from the [`AlterTablesData`].
+fn build_validator_from_alter_table_data<'a>(
+    data: &'a AlterTablesData,
+) -> AlterLogicalTableValidator<'a> {
+    let phsycial_table_id = data.physical_table_id;
+    let alters = data
+        .tasks
+        .iter()
+        .map(|task| &task.alter_table)
+        .collect::<Vec<_>>();
+    AlterLogicalTableValidator::new(phsycial_table_id, alters)
+}
+
+/// Builds the executor from the [`AlterTablesData`].
+fn build_executor_from_alter_expr<'a>(data: &'a AlterTablesData) -> AlterLogicalTablesExecutor<'a> {
+    debug_assert_eq!(data.tasks.len(), data.table_info_values.len());
+    let alters = data
+        .tasks
+        .iter()
+        .zip(data.table_info_values.iter())
+        .map(|(task, table_info)| (table_info.table_info.ident.table_id, &task.alter_table))
+        .collect::<Vec<_>>();
+    AlterLogicalTablesExecutor::new(alters)
+}
+
 impl AlterLogicalTablesProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::AlterLogicalTables";
 
@@ -80,35 +106,44 @@ impl AlterLogicalTablesProcedure {
     }
 
     pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
-        // Checks all the tasks
+        let validator = build_validator_from_alter_table_data(&self.data);
-        self.check_input_tasks()?;
+        let ValidatorResult {
-        // Fills the table info values
+            num_skipped,
-        self.fill_table_info_values().await?;
+            skip_alter,
-        // Checks the physical table, must after [fill_table_info_values]
+            table_info_values,
-        self.check_physical_table().await?;
+            physical_table_info,
-        // Fills the physical table info
+            physical_table_route,
-        self.fill_physical_table_info().await?;
+        } = validator
-        // Filter the finished tasks
+            .validate(&self.context.table_metadata_manager)
-        let finished_tasks = self.check_finished_tasks()?;
+            .await?;
-        let already_finished_count = finished_tasks
+
-            .iter()
+        let num_tasks = self.data.tasks.len();
-            .map(|x| if *x { 1 } else { 0 })
+        if num_skipped == num_tasks {
-            .sum::<usize>();
-        let apply_tasks_count = self.data.tasks.len();
-        if already_finished_count == apply_tasks_count {
             info!("All the alter tasks are finished, will skip the procedure.");
+            let cache_ident_keys = AlterLogicalTablesExecutor::build_cache_ident_keys(
+                &physical_table_info,
+                &table_info_values
+                    .iter()
+                    .map(|v| v.get_inner_ref())
+                    .collect::<Vec<_>>(),
+            );
+            self.data.table_cache_keys_to_invalidate = cache_ident_keys;
             // Re-invalidate the table cache
             self.data.state = AlterTablesState::InvalidateTableCache;
             return Ok(Status::executing(true));
-        } else if already_finished_count > 0 {
+        } else if num_skipped > 0 {
             info!(
                 "There are {} alter tasks, {} of them were already finished.",
-                apply_tasks_count, already_finished_count
+                num_tasks, num_skipped
             );
         }
-        self.filter_task(&finished_tasks)?;
 
-        // Next state
+        // Updates the procedure state.
+        retain_unskipped(&mut self.data.tasks, &skip_alter);
+        self.data.physical_table_info = Some(physical_table_info);
+        self.data.physical_table_route = Some(physical_table_route);
+        self.data.table_info_values = table_info_values;
+        debug_assert_eq!(self.data.tasks.len(), self.data.table_info_values.len());
         self.data.state = AlterTablesState::SubmitAlterRegionRequests;
         Ok(Status::executing(true))
     }
@@ -116,57 +151,21 @@ impl AlterLogicalTablesProcedure {
     pub(crate) async fn on_submit_alter_region_requests(&mut self) -> Result<Status> {
         // Safety: we have checked the state in on_prepare
         let physical_table_route = &self.data.physical_table_route.as_ref().unwrap();
-        let leaders = find_leaders(&physical_table_route.region_routes);
+        let executor = build_executor_from_alter_expr(&self.data);
-        let mut alter_region_tasks = Vec::with_capacity(leaders.len());
+        let mut results = executor
+            .on_alter_regions(
+                &self.context.node_manager,
+                &physical_table_route.region_routes,
+            )
+            .await?;
 
-        for peer in leaders {
+        if let Some(column_metadatas) =
-            let requester = self.context.node_manager.datanode(&peer).await;
+            extract_column_metadatas(&mut results, ALTER_PHYSICAL_EXTENSION_KEY)?
-            let request = self.make_request(&peer, &physical_table_route.region_routes)?;
+        {
-
+            self.data.physical_columns = column_metadatas;
-            alter_region_tasks.push(async move {
-                requester
-                    .handle(request)
-                    .await
-                    .map_err(add_peer_context_if_needed(peer))
-            });
-        }
-
-        let mut results = future::join_all(alter_region_tasks)
-            .await
-            .into_iter()
-            .collect::<Result<Vec<_>>>()?;
-
-        // Collects responses from datanodes.
-        let phy_raw_schemas = results
-            .iter_mut()
-            .map(|res| res.extensions.remove(ALTER_PHYSICAL_EXTENSION_KEY))
-            .collect::<Vec<_>>();
-
-        if phy_raw_schemas.is_empty() {
-            self.submit_sync_region_requests(results, &physical_table_route.region_routes)
-                .await;
-            self.data.state = AlterTablesState::UpdateMetadata;
-            return Ok(Status::executing(true));
-        }
-
-        // Verify all the physical schemas are the same
-        // Safety: previous check ensures this vec is not empty
-        let first = phy_raw_schemas.first().unwrap();
-        ensure!(
-            phy_raw_schemas.iter().all(|x| x == first),
-            MetadataCorruptionSnafu {
-                err_msg: "The physical schemas from datanodes are not the same."
-            }
-        );
-
-        // Decodes the physical raw schemas
-        if let Some(phy_raw_schema) = first {
-            self.data.physical_columns =
-                ColumnMetadata::decode_list(phy_raw_schema).context(DecodeJsonSnafu)?;
         } else {
             warn!("altering logical table result doesn't contains extension key `{ALTER_PHYSICAL_EXTENSION_KEY}`,leaving the physical table's schema unchanged");
         }
-
         self.submit_sync_region_requests(results, &physical_table_route.region_routes)
             .await;
         self.data.state = AlterTablesState::UpdateMetadata;
@@ -182,7 +181,7 @@ impl AlterLogicalTablesProcedure {
         if let Err(err) = sync_follower_regions(
             &self.context,
             self.data.physical_table_id,
-            results,
+            &results,
             region_routes,
             table_info.meta.engine.as_str(),
         )
@@ -199,7 +198,18 @@ impl AlterLogicalTablesProcedure {
         self.update_physical_table_metadata().await?;
         self.update_logical_tables_metadata().await?;
 
-        self.data.build_cache_keys_to_invalidate();
+        let logical_table_info_values = self
+            .data
+            .table_info_values
+            .iter()
+            .map(|v| v.get_inner_ref())
+            .collect::<Vec<_>>();
+
+        let cache_ident_keys = AlterLogicalTablesExecutor::build_cache_ident_keys(
+            self.data.physical_table_info.as_ref().unwrap(),
+            &logical_table_info_values,
+        );
+        self.data.table_cache_keys_to_invalidate = cache_ident_keys;
         self.data.clear_metadata_fields();
 
         self.data.state = AlterTablesState::InvalidateTableCache;
@@ -209,9 +219,16 @@ impl AlterLogicalTablesProcedure {
     pub(crate) async fn on_invalidate_table_cache(&mut self) -> Result<Status> {
         let to_invalidate = &self.data.table_cache_keys_to_invalidate;
 
+        let ctx = CacheContext {
+            subject: Some(format!(
+                "Invalidate table cache by altering logical tables, physical_table_id: {}",
+                self.data.physical_table_id,
+            )),
+        };
+
         self.context
             .cache_invalidator
-            .invalidate(&Default::default(), to_invalidate)
+            .invalidate(&ctx, to_invalidate)
             .await?;
         Ok(Status::done())
     }
@@ -231,6 +248,10 @@ impl Procedure for AlterLogicalTablesProcedure {
         let _timer = metrics::METRIC_META_PROCEDURE_ALTER_TABLE
             .with_label_values(&[step])
             .start_timer();
+        debug!(
+            "Executing alter logical tables procedure, state: {:?}",
+            state
+        );
 
         match state {
             AlterTablesState::Prepare => self.on_prepare().await,

src/common/meta/src/ddl/alter_logical_tables/check.rs

@@ -1,136 +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::collections::HashSet;
-
-use api::v1::alter_table_expr::Kind;
-use snafu::{ensure, OptionExt};
-
-use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::error::{AlterLogicalTablesInvalidArgumentsSnafu, Result};
-use crate::key::table_info::TableInfoValue;
-use crate::key::table_route::TableRouteValue;
-use crate::rpc::ddl::AlterTableTask;
-
-impl AlterLogicalTablesProcedure {
-    pub(crate) fn check_input_tasks(&self) -> Result<()> {
-        self.check_schema()?;
-        self.check_alter_kind()?;
-        Ok(())
-    }
-
-    pub(crate) async fn check_physical_table(&self) -> Result<()> {
-        let table_route_manager = self.context.table_metadata_manager.table_route_manager();
-        let table_ids = self
-            .data
-            .table_info_values
-            .iter()
-            .map(|v| v.table_info.ident.table_id)
-            .collect::<Vec<_>>();
-        let table_routes = table_route_manager
-            .table_route_storage()
-            .batch_get(&table_ids)
-            .await?;
-        let physical_table_id = self.data.physical_table_id;
-        let is_same_physical_table = table_routes.iter().all(|r| {
-            if let Some(TableRouteValue::Logical(r)) = r {
-                r.physical_table_id() == physical_table_id
-            } else {
-                false
-            }
-        });
-
-        ensure!(
-            is_same_physical_table,
-            AlterLogicalTablesInvalidArgumentsSnafu {
-                err_msg: "All the tasks should have the same physical table id"
-            }
-        );
-
-        Ok(())
-    }
-
-    pub(crate) fn check_finished_tasks(&self) -> Result<Vec<bool>> {
-        let task = &self.data.tasks;
-        let table_info_values = &self.data.table_info_values;
-
-        Ok(task
-            .iter()
-            .zip(table_info_values.iter())
-            .map(|(task, table)| Self::check_finished_task(task, table))
-            .collect())
-    }
-
-    // Checks if the schemas of the tasks are the same
-    fn check_schema(&self) -> Result<()> {
-        let is_same_schema = self.data.tasks.windows(2).all(|pair| {
-            pair[0].alter_table.catalog_name == pair[1].alter_table.catalog_name
-                && pair[0].alter_table.schema_name == pair[1].alter_table.schema_name
-        });
-
-        ensure!(
-            is_same_schema,
-            AlterLogicalTablesInvalidArgumentsSnafu {
-                err_msg: "Schemas of the tasks are not the same"
-            }
-        );
-
-        Ok(())
-    }
-
-    fn check_alter_kind(&self) -> Result<()> {
-        for task in &self.data.tasks {
-            let kind = task.alter_table.kind.as_ref().context(
-                AlterLogicalTablesInvalidArgumentsSnafu {
-                    err_msg: "Alter kind is missing",
-                },
-            )?;
-            let Kind::AddColumns(_) = kind else {
-                return AlterLogicalTablesInvalidArgumentsSnafu {
-                    err_msg: "Only support add columns operation",
-                }
-                .fail();
-            };
-        }
-
-        Ok(())
-    }
-
-    fn check_finished_task(task: &AlterTableTask, table: &TableInfoValue) -> bool {
-        let columns = table
-            .table_info
-            .meta
-            .schema
-            .column_schemas
-            .iter()
-            .map(|c| &c.name)
-            .collect::<HashSet<_>>();
-
-        let Some(kind) = task.alter_table.kind.as_ref() else {
-            return true; // Never get here since we have checked it in `check_alter_kind`
-        };
-        let Kind::AddColumns(add_columns) = kind else {
-            return true; // Never get here since we have checked it in `check_alter_kind`
-        };
-
-        // We only check that all columns have been finished. That is to say,
-        // if one part is finished but another part is not, it will be considered
-        // unfinished.
-        add_columns
-            .add_columns
-            .iter()
-            .map(|add_column| add_column.column_def.as_ref().map(|c| &c.name))
-            .all(|column| column.map(|c| columns.contains(c)).unwrap_or(false))
-    }
-}

src/common/meta/src/ddl/alter_logical_tables/executor.rs

@@ -0,0 +1,216 @@
+// 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::collections::HashMap;
+
+use api::region::RegionResponse;
+use api::v1::alter_table_expr::Kind;
+use api::v1::region::{
+    alter_request, region_request, AddColumn, AddColumns, AlterRequest, AlterRequests,
+    RegionColumnDef, RegionRequest, RegionRequestHeader,
+};
+use api::v1::{self, AlterTableExpr};
+use common_telemetry::tracing_context::TracingContext;
+use common_telemetry::{debug, warn};
+use futures::future;
+use store_api::metadata::ColumnMetadata;
+use store_api::storage::{RegionId, RegionNumber, TableId};
+
+use crate::ddl::utils::{add_peer_context_if_needed, raw_table_info};
+use crate::error::Result;
+use crate::instruction::CacheIdent;
+use crate::key::table_info::TableInfoValue;
+use crate::key::{DeserializedValueWithBytes, RegionDistribution, TableMetadataManagerRef};
+use crate::node_manager::NodeManagerRef;
+use crate::rpc::router::{find_leaders, region_distribution, RegionRoute};
+
+/// [AlterLogicalTablesExecutor] performs:
+/// - Alters logical regions on the datanodes.
+/// - Updates table metadata for alter table operation.
+pub struct AlterLogicalTablesExecutor<'a> {
+    /// The alter table expressions.
+    ///
+    /// The first element is the logical table id, the second element is the alter table expression.
+    alters: Vec<(TableId, &'a AlterTableExpr)>,
+}
+
+impl<'a> AlterLogicalTablesExecutor<'a> {
+    pub fn new(alters: Vec<(TableId, &'a AlterTableExpr)>) -> Self {
+        Self { alters }
+    }
+
+    /// Alters logical regions on the datanodes.
+    pub(crate) async fn on_alter_regions(
+        &self,
+        node_manager: &NodeManagerRef,
+        region_routes: &[RegionRoute],
+    ) -> Result<Vec<RegionResponse>> {
+        let region_distribution = region_distribution(region_routes);
+        let leaders = find_leaders(region_routes)
+            .into_iter()
+            .map(|p| (p.id, p))
+            .collect::<HashMap<_, _>>();
+        let mut alter_region_tasks = Vec::with_capacity(leaders.len());
+        for (datanode_id, region_role_set) in region_distribution {
+            if region_role_set.leader_regions.is_empty() {
+                continue;
+            }
+            // Safety: must exists.
+            let peer = leaders.get(&datanode_id).unwrap();
+            let requester = node_manager.datanode(peer).await;
+            let requests = self.make_alter_region_request(&region_role_set.leader_regions);
+            let requester = requester.clone();
+            let peer = peer.clone();
+
+            debug!("Sending alter region requests to datanode {}", peer);
+            alter_region_tasks.push(async move {
+                requester
+                    .handle(make_request(requests))
+                    .await
+                    .map_err(add_peer_context_if_needed(peer))
+            });
+        }
+
+        future::join_all(alter_region_tasks)
+            .await
+            .into_iter()
+            .collect::<Result<Vec<_>>>()
+    }
+
+    fn make_alter_region_request(&self, region_numbers: &[RegionNumber]) -> AlterRequests {
+        let mut requests = Vec::with_capacity(region_numbers.len() * self.alters.len());
+        for (table_id, alter) in self.alters.iter() {
+            for region_number in region_numbers {
+                let region_id = RegionId::new(*table_id, *region_number);
+                let request = make_alter_region_request(region_id, alter);
+                requests.push(request);
+            }
+        }
+
+        AlterRequests { requests }
+    }
+
+    /// Updates table metadata for alter table operation.
+    ///
+    /// ## Panic:
+    /// - If the region distribution is not set when updating table metadata.
+    pub(crate) async fn on_alter_metadata(
+        physical_table_id: TableId,
+        table_metadata_manager: &TableMetadataManagerRef,
+        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
+        region_distribution: RegionDistribution,
+        physical_columns: &[ColumnMetadata],
+    ) -> Result<()> {
+        if physical_columns.is_empty() {
+            warn!("No physical columns found, leaving the physical table's schema unchanged when altering logical tables");
+            return Ok(());
+        }
+
+        let table_ref = current_table_info_value.table_ref();
+        let table_id = physical_table_id;
+
+        // Generates new table info
+        let old_raw_table_info = current_table_info_value.table_info.clone();
+        let new_raw_table_info =
+            raw_table_info::build_new_physical_table_info(old_raw_table_info, physical_columns);
+
+        debug!(
+            "Starting update table: {} metadata, table_id: {}, new table info: {:?}",
+            table_ref, table_id, new_raw_table_info
+        );
+
+        table_metadata_manager
+            .update_table_info(
+                current_table_info_value,
+                Some(region_distribution),
+                new_raw_table_info,
+            )
+            .await?;
+
+        Ok(())
+    }
+
+    /// Builds the cache ident keys for the alter logical tables.
+    ///
+    /// The cache ident keys are:
+    /// - The table id of the logical tables.
+    /// - The table name of the logical tables.
+    /// - The table id of the physical table.
+    pub(crate) fn build_cache_ident_keys(
+        physical_table_info: &TableInfoValue,
+        logical_table_info_values: &[&TableInfoValue],
+    ) -> Vec<CacheIdent> {
+        let mut cache_keys = Vec::with_capacity(logical_table_info_values.len() * 2 + 2);
+        cache_keys.extend(logical_table_info_values.iter().flat_map(|table| {
+            vec![
+                CacheIdent::TableId(table.table_info.ident.table_id),
+                CacheIdent::TableName(table.table_name()),
+            ]
+        }));
+        cache_keys.push(CacheIdent::TableId(
+            physical_table_info.table_info.ident.table_id,
+        ));
+        cache_keys.push(CacheIdent::TableName(physical_table_info.table_name()));
+
+        cache_keys
+    }
+}
+
+fn make_request(alter_requests: AlterRequests) -> RegionRequest {
+    RegionRequest {
+        header: Some(RegionRequestHeader {
+            tracing_context: TracingContext::from_current_span().to_w3c(),
+            ..Default::default()
+        }),
+        body: Some(region_request::Body::Alters(alter_requests)),
+    }
+}
+
+/// Makes an alter region request.
+pub fn make_alter_region_request(
+    region_id: RegionId,
+    alter_table_expr: &AlterTableExpr,
+) -> AlterRequest {
+    let region_id = region_id.as_u64();
+    let kind = match &alter_table_expr.kind {
+        Some(Kind::AddColumns(add_columns)) => Some(alter_request::Kind::AddColumns(
+            to_region_add_columns(add_columns),
+        )),
+        _ => unreachable!(), // Safety: we have checked the kind in check_input_tasks
+    };
+
+    AlterRequest {
+        region_id,
+        schema_version: 0,
+        kind,
+    }
+}
+
+fn to_region_add_columns(add_columns: &v1::AddColumns) -> AddColumns {
+    let add_columns = add_columns
+        .add_columns
+        .iter()
+        .map(|add_column| {
+            let region_column_def = RegionColumnDef {
+                column_def: add_column.column_def.clone(),
+                ..Default::default() // other fields are not used in alter logical table
+            };
+            AddColumn {
+                column_def: Some(region_column_def),
+                ..Default::default() // other fields are not used in alter logical table
+            }
+        })
+        .collect();
+    AddColumns { add_columns }
+}

src/common/meta/src/ddl/alter_logical_tables/metadata.rs

@@ -1,158 +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_catalog::format_full_table_name;
-use snafu::OptionExt;
-use table::metadata::TableId;
-
-use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::error::{
-    AlterLogicalTablesInvalidArgumentsSnafu, Result, TableInfoNotFoundSnafu, TableNotFoundSnafu,
-    TableRouteNotFoundSnafu,
-};
-use crate::key::table_info::TableInfoValue;
-use crate::key::table_name::TableNameKey;
-use crate::key::table_route::TableRouteValue;
-use crate::key::DeserializedValueWithBytes;
-use crate::rpc::ddl::AlterTableTask;
-
-impl AlterLogicalTablesProcedure {
-    pub(crate) fn filter_task(&mut self, finished_tasks: &[bool]) -> Result<()> {
-        debug_assert_eq!(finished_tasks.len(), self.data.tasks.len());
-        debug_assert_eq!(finished_tasks.len(), self.data.table_info_values.len());
-        self.data.tasks = self
-            .data
-            .tasks
-            .drain(..)
-            .zip(finished_tasks.iter())
-            .filter_map(|(task, finished)| if *finished { None } else { Some(task) })
-            .collect();
-        self.data.table_info_values = self
-            .data
-            .table_info_values
-            .drain(..)
-            .zip(finished_tasks.iter())
-            .filter_map(|(table_info_value, finished)| {
-                if *finished {
-                    None
-                } else {
-                    Some(table_info_value)
-                }
-            })
-            .collect();
-
-        Ok(())
-    }
-
-    pub(crate) async fn fill_physical_table_info(&mut self) -> Result<()> {
-        let (physical_table_info, physical_table_route) = self
-            .context
-            .table_metadata_manager
-            .get_full_table_info(self.data.physical_table_id)
-            .await?;
-
-        let physical_table_info = physical_table_info.with_context(|| TableInfoNotFoundSnafu {
-            table: format!("table id - {}", self.data.physical_table_id),
-        })?;
-        let physical_table_route = physical_table_route
-            .context(TableRouteNotFoundSnafu {
-                table_id: self.data.physical_table_id,
-            })?
-            .into_inner();
-
-        self.data.physical_table_info = Some(physical_table_info);
-        let TableRouteValue::Physical(physical_table_route) = physical_table_route else {
-            return AlterLogicalTablesInvalidArgumentsSnafu {
-                err_msg: format!(
-                    "expected a physical table but got a logical table: {:?}",
-                    self.data.physical_table_id
-                ),
-            }
-            .fail();
-        };
-        self.data.physical_table_route = Some(physical_table_route);
-
-        Ok(())
-    }
-
-    pub(crate) async fn fill_table_info_values(&mut self) -> Result<()> {
-        let table_ids = self.get_all_table_ids().await?;
-        let table_info_values = self.get_all_table_info_values(&table_ids).await?;
-        debug_assert_eq!(table_info_values.len(), self.data.tasks.len());
-        self.data.table_info_values = table_info_values;
-
-        Ok(())
-    }
-
-    async fn get_all_table_info_values(
-        &self,
-        table_ids: &[TableId],
-    ) -> Result<Vec<DeserializedValueWithBytes<TableInfoValue>>> {
-        let table_info_manager = self.context.table_metadata_manager.table_info_manager();
-        let mut table_info_map = table_info_manager.batch_get_raw(table_ids).await?;
-        let mut table_info_values = Vec::with_capacity(table_ids.len());
-        for (table_id, task) in table_ids.iter().zip(self.data.tasks.iter()) {
-            let table_info_value =
-                table_info_map
-                    .remove(table_id)
-                    .with_context(|| TableInfoNotFoundSnafu {
-                        table: extract_table_name(task),
-                    })?;
-            table_info_values.push(table_info_value);
-        }
-
-        Ok(table_info_values)
-    }
-
-    async fn get_all_table_ids(&self) -> Result<Vec<TableId>> {
-        let table_name_manager = self.context.table_metadata_manager.table_name_manager();
-        let table_name_keys = self
-            .data
-            .tasks
-            .iter()
-            .map(|task| extract_table_name_key(task))
-            .collect();
-
-        let table_name_values = table_name_manager.batch_get(table_name_keys).await?;
-        let mut table_ids = Vec::with_capacity(table_name_values.len());
-        for (value, task) in table_name_values.into_iter().zip(self.data.tasks.iter()) {
-            let table_id = value
-                .with_context(|| TableNotFoundSnafu {
-                    table_name: extract_table_name(task),
-                })?
-                .table_id();
-            table_ids.push(table_id);
-        }
-
-        Ok(table_ids)
-    }
-}
-
-#[inline]
-fn extract_table_name(task: &AlterTableTask) -> String {
-    format_full_table_name(
-        &task.alter_table.catalog_name,
-        &task.alter_table.schema_name,
-        &task.alter_table.table_name,
-    )
-}
-
-#[inline]
-fn extract_table_name_key(task: &AlterTableTask) -> TableNameKey {
-    TableNameKey::new(
-        &task.alter_table.catalog_name,
-        &task.alter_table.schema_name,
-        &task.alter_table.table_name,
-    )
-}

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

@@ -1,112 +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 api::v1;
-use api::v1::alter_table_expr::Kind;
-use api::v1::region::{
-    alter_request, region_request, AddColumn, AddColumns, AlterRequest, AlterRequests,
-    RegionColumnDef, RegionRequest, RegionRequestHeader,
-};
-use common_telemetry::tracing_context::TracingContext;
-use store_api::storage::RegionId;
-
-use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::error::Result;
-use crate::key::table_info::TableInfoValue;
-use crate::peer::Peer;
-use crate::rpc::ddl::AlterTableTask;
-use crate::rpc::router::{find_leader_regions, RegionRoute};
-
-impl AlterLogicalTablesProcedure {
-    pub(crate) fn make_request(
-        &self,
-        peer: &Peer,
-        region_routes: &[RegionRoute],
-    ) -> Result<RegionRequest> {
-        let alter_requests = self.make_alter_region_requests(peer, region_routes)?;
-        let request = RegionRequest {
-            header: Some(RegionRequestHeader {
-                tracing_context: TracingContext::from_current_span().to_w3c(),
-                ..Default::default()
-            }),
-            body: Some(region_request::Body::Alters(alter_requests)),
-        };
-
-        Ok(request)
-    }
-
-    fn make_alter_region_requests(
-        &self,
-        peer: &Peer,
-        region_routes: &[RegionRoute],
-    ) -> Result<AlterRequests> {
-        let tasks = &self.data.tasks;
-        let regions_on_this_peer = find_leader_regions(region_routes, peer);
-        let mut requests = Vec::with_capacity(tasks.len() * regions_on_this_peer.len());
-        for (task, table) in self
-            .data
-            .tasks
-            .iter()
-            .zip(self.data.table_info_values.iter())
-        {
-            for region_number in &regions_on_this_peer {
-                let region_id = RegionId::new(table.table_info.ident.table_id, *region_number);
-                let request = self.make_alter_region_request(region_id, task, table)?;
-                requests.push(request);
-            }
-        }
-
-        Ok(AlterRequests { requests })
-    }
-
-    fn make_alter_region_request(
-        &self,
-        region_id: RegionId,
-        task: &AlterTableTask,
-        table: &TableInfoValue,
-    ) -> Result<AlterRequest> {
-        let region_id = region_id.as_u64();
-        let schema_version = table.table_info.ident.version;
-        let kind = match &task.alter_table.kind {
-            Some(Kind::AddColumns(add_columns)) => Some(alter_request::Kind::AddColumns(
-                to_region_add_columns(add_columns),
-            )),
-            _ => unreachable!(), // Safety: we have checked the kind in check_input_tasks
-        };
-
-        Ok(AlterRequest {
-            region_id,
-            schema_version,
-            kind,
-        })
-    }
-}
-
-fn to_region_add_columns(add_columns: &v1::AddColumns) -> AddColumns {
-    let add_columns = add_columns
-        .add_columns
-        .iter()
-        .map(|add_column| {
-            let region_column_def = RegionColumnDef {
-                column_def: add_column.column_def.clone(),
-                ..Default::default() // other fields are not used in alter logical table
-            };
-            AddColumn {
-                column_def: Some(region_column_def),
-                ..Default::default() // other fields are not used in alter logical table
-            }
-        })
-        .collect();
-    AddColumns { add_columns }
-}

src/common/meta/src/ddl/alter_logical_tables/table_cache_keys.rs

@@ -1,50 +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 table::metadata::RawTableInfo;
-use table::table_name::TableName;
-
-use crate::ddl::alter_logical_tables::AlterTablesData;
-use crate::instruction::CacheIdent;
-
-impl AlterTablesData {
-    pub(crate) fn build_cache_keys_to_invalidate(&mut self) {
-        let mut cache_keys = self
-            .table_info_values
-            .iter()
-            .flat_map(|table| {
-                vec![
-                    CacheIdent::TableId(table.table_info.ident.table_id),
-                    CacheIdent::TableName(extract_table_name(&table.table_info)),
-                ]
-            })
-            .collect::<Vec<_>>();
-        cache_keys.push(CacheIdent::TableId(self.physical_table_id));
-        // Safety: physical_table_info already filled in previous steps
-        let physical_table_info = &self.physical_table_info.as_ref().unwrap().table_info;
-        cache_keys.push(CacheIdent::TableName(extract_table_name(
-            physical_table_info,
-        )));
-
-        self.table_cache_keys_to_invalidate = cache_keys;
-    }
-}
-
-fn extract_table_name(table_info: &RawTableInfo) -> TableName {
-    TableName::new(
-        &table_info.catalog_name,
-        &table_info.schema_name,
-        &table_info.name,
-    )
-}

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

@@ -13,41 +13,35 @@
 // limitations under the License.
 
 use common_grpc_expr::alter_expr_to_request;
-use common_telemetry::warn;
 use itertools::Itertools;
 use snafu::ResultExt;
 use table::metadata::{RawTableInfo, TableInfo};
 
+use crate::ddl::alter_logical_tables::executor::AlterLogicalTablesExecutor;
 use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
-use crate::ddl::physical_table_metadata;
 use crate::error;
 use crate::error::{ConvertAlterTableRequestSnafu, Result};
 use crate::key::table_info::TableInfoValue;
 use crate::key::DeserializedValueWithBytes;
 use crate::rpc::ddl::AlterTableTask;
+use crate::rpc::router::region_distribution;
 
 impl AlterLogicalTablesProcedure {
     pub(crate) async fn update_physical_table_metadata(&mut self) -> Result<()> {
-        if self.data.physical_columns.is_empty() {
-            warn!("No physical columns found, leaving the physical table's schema unchanged when altering logical tables");
-            return Ok(());
-        }
-
         // Safety: must exist.
         let physical_table_info = self.data.physical_table_info.as_ref().unwrap();
+        let physical_table_route = self.data.physical_table_route.as_ref().unwrap();
+        let region_distribution = region_distribution(&physical_table_route.region_routes);
 
-        // Generates new table info
+        // Updates physical table's metadata.
-        let old_raw_table_info = physical_table_info.table_info.clone();
+        AlterLogicalTablesExecutor::on_alter_metadata(
-        let new_raw_table_info = physical_table_metadata::build_new_physical_table_info(
+            self.data.physical_table_id,
-            old_raw_table_info,
+            &self.context.table_metadata_manager,
+            physical_table_info,
+            region_distribution,
             &self.data.physical_columns,
-        );
+        )
-
+        .await?;
-        // Updates physical table's metadata, and we don't need to touch per-region settings.
-        self.context
-            .table_metadata_manager
-            .update_table_info(physical_table_info, None, new_raw_table_info)
-            .await?;
 
         Ok(())
     }

src/common/meta/src/ddl/alter_logical_tables/validator.rs

@@ -0,0 +1,284 @@
+// 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::collections::HashSet;
+
+use api::v1::alter_table_expr::Kind;
+use api::v1::AlterTableExpr;
+use snafu::{ensure, OptionExt};
+use store_api::storage::TableId;
+use table::table_reference::TableReference;
+
+use crate::ddl::utils::table_id::get_all_table_ids_by_names;
+use crate::ddl::utils::table_info::get_all_table_info_values_by_table_ids;
+use crate::error::{
+    AlterLogicalTablesInvalidArgumentsSnafu, Result, TableInfoNotFoundSnafu,
+    TableRouteNotFoundSnafu,
+};
+use crate::key::table_info::TableInfoValue;
+use crate::key::table_route::{PhysicalTableRouteValue, TableRouteManager, TableRouteValue};
+use crate::key::{DeserializedValueWithBytes, TableMetadataManagerRef};
+
+/// [AlterLogicalTableValidator] validates the alter logical expressions.
+pub struct AlterLogicalTableValidator<'a> {
+    physical_table_id: TableId,
+    alters: Vec<&'a AlterTableExpr>,
+}
+
+impl<'a> AlterLogicalTableValidator<'a> {
+    pub fn new(physical_table_id: TableId, alters: Vec<&'a AlterTableExpr>) -> Self {
+        Self {
+            physical_table_id,
+            alters,
+        }
+    }
+
+    /// Validates all alter table expressions have the same schema and catalog.
+    fn validate_schema(&self) -> Result<()> {
+        let is_same_schema = self.alters.windows(2).all(|pair| {
+            pair[0].catalog_name == pair[1].catalog_name
+                && pair[0].schema_name == pair[1].schema_name
+        });
+
+        ensure!(
+            is_same_schema,
+            AlterLogicalTablesInvalidArgumentsSnafu {
+                err_msg: "Schemas of the alter table expressions are not the same"
+            }
+        );
+
+        Ok(())
+    }
+
+    /// Validates that all alter table expressions are of the supported kind.
+    /// Currently only supports `AddColumns` operations.
+    fn validate_alter_kind(&self) -> Result<()> {
+        for alter in &self.alters {
+            let kind = alter
+                .kind
+                .as_ref()
+                .context(AlterLogicalTablesInvalidArgumentsSnafu {
+                    err_msg: "Alter kind is missing",
+                })?;
+
+            let Kind::AddColumns(_) = kind else {
+                return AlterLogicalTablesInvalidArgumentsSnafu {
+                    err_msg: "Only support add columns operation",
+                }
+                .fail();
+            };
+        }
+
+        Ok(())
+    }
+
+    fn table_names(&self) -> Vec<TableReference> {
+        self.alters
+            .iter()
+            .map(|alter| {
+                TableReference::full(&alter.catalog_name, &alter.schema_name, &alter.table_name)
+            })
+            .collect()
+    }
+
+    /// Validates that the physical table info and route exist.
+    ///
+    /// This method performs the following validations:
+    /// 1. Retrieves the full table info and route for the given physical table id
+    /// 2. Ensures the table info and table route exists
+    /// 3. Verifies that the table route is actually a physical table route, not a logical one
+    ///
+    /// Returns a tuple containing the validated table info and physical table route.
+    async fn validate_physical_table(
+        &self,
+        table_metadata_manager: &TableMetadataManagerRef,
+    ) -> Result<(
+        DeserializedValueWithBytes<TableInfoValue>,
+        PhysicalTableRouteValue,
+    )> {
+        let (table_info, table_route) = table_metadata_manager
+            .get_full_table_info(self.physical_table_id)
+            .await?;
+
+        let table_info = table_info.with_context(|| TableInfoNotFoundSnafu {
+            table: format!("table id - {}", self.physical_table_id),
+        })?;
+
+        let physical_table_route = table_route
+            .context(TableRouteNotFoundSnafu {
+                table_id: self.physical_table_id,
+            })?
+            .into_inner();
+
+        let TableRouteValue::Physical(table_route) = physical_table_route else {
+            return AlterLogicalTablesInvalidArgumentsSnafu {
+                err_msg: format!(
+                    "expected a physical table but got a logical table: {:?}",
+                    self.physical_table_id
+                ),
+            }
+            .fail();
+        };
+
+        Ok((table_info, table_route))
+    }
+
+    /// Validates that all logical table routes have the same physical table id.
+    ///
+    /// This method performs the following validations:
+    /// 1. Retrieves table routes for all the given table ids.
+    /// 2. Ensures that all retrieved routes are logical table routes (not physical)
+    /// 3. Verifies that all logical table routes reference the same physical table id.
+    /// 4. Returns an error if any route is not logical or references a different physical table.
+    async fn validate_logical_table_routes(
+        &self,
+        table_route_manager: &TableRouteManager,
+        table_ids: &[TableId],
+    ) -> Result<()> {
+        let table_routes = table_route_manager
+            .table_route_storage()
+            .batch_get(table_ids)
+            .await?;
+
+        let physical_table_id = self.physical_table_id;
+
+        let is_same_physical_table = table_routes.iter().all(|r| {
+            if let Some(TableRouteValue::Logical(r)) = r {
+                r.physical_table_id() == physical_table_id
+            } else {
+                false
+            }
+        });
+
+        ensure!(
+            is_same_physical_table,
+            AlterLogicalTablesInvalidArgumentsSnafu {
+                err_msg: "All the tasks should have the same physical table id"
+            }
+        );
+
+        Ok(())
+    }
+
+    /// Validates the alter logical expressions.
+    ///
+    /// This method performs the following validations:
+    /// 1. Validates that all alter table expressions have the same schema and catalog.
+    /// 2. Validates that all alter table expressions are of the supported kind.
+    /// 3. Validates that the physical table info and route exist.
+    /// 4. Validates that all logical table routes have the same physical table id.
+    ///
+    /// Returns a [ValidatorResult] containing the validation results.
+    pub async fn validate(
+        &self,
+        table_metadata_manager: &TableMetadataManagerRef,
+    ) -> Result<ValidatorResult> {
+        self.validate_schema()?;
+        self.validate_alter_kind()?;
+        let (physical_table_info, physical_table_route) =
+            self.validate_physical_table(table_metadata_manager).await?;
+        let table_names = self.table_names();
+        let table_ids =
+            get_all_table_ids_by_names(table_metadata_manager.table_name_manager(), &table_names)
+                .await?;
+        let mut table_info_values = get_all_table_info_values_by_table_ids(
+            table_metadata_manager.table_info_manager(),
+            &table_ids,
+            &table_names,
+        )
+        .await?;
+        self.validate_logical_table_routes(
+            table_metadata_manager.table_route_manager(),
+            &table_ids,
+        )
+        .await?;
+        let skip_alter = self
+            .alters
+            .iter()
+            .zip(table_info_values.iter())
+            .map(|(task, table)| skip_alter_logical_region(task, table))
+            .collect::<Vec<_>>();
+        retain_unskipped(&mut table_info_values, &skip_alter);
+        let num_skipped = skip_alter.iter().filter(|&&x| x).count();
+
+        Ok(ValidatorResult {
+            num_skipped,
+            skip_alter,
+            table_info_values,
+            physical_table_info,
+            physical_table_route,
+        })
+    }
+}
+
+/// The result of the validator.
+pub(crate) struct ValidatorResult {
+    pub(crate) num_skipped: usize,
+    pub(crate) skip_alter: Vec<bool>,
+    pub(crate) table_info_values: Vec<DeserializedValueWithBytes<TableInfoValue>>,
+    pub(crate) physical_table_info: DeserializedValueWithBytes<TableInfoValue>,
+    pub(crate) physical_table_route: PhysicalTableRouteValue,
+}
+
+/// Retains the elements that are not skipped.
+pub(crate) fn retain_unskipped<T>(target: &mut Vec<T>, skipped: &[bool]) {
+    debug_assert_eq!(target.len(), skipped.len());
+    let mut iter = skipped.iter();
+    target.retain(|_| !iter.next().unwrap());
+}
+
+/// Returns true if does not required to alter the logical region.
+fn skip_alter_logical_region(alter: &AlterTableExpr, table: &TableInfoValue) -> bool {
+    let existing_columns = table
+        .table_info
+        .meta
+        .schema
+        .column_schemas
+        .iter()
+        .map(|c| &c.name)
+        .collect::<HashSet<_>>();
+
+    let Some(kind) = alter.kind.as_ref() else {
+        return true; // Never get here since we have checked it in `validate_alter_kind`
+    };
+    let Kind::AddColumns(add_columns) = kind else {
+        return true; // Never get here since we have checked it in `validate_alter_kind`
+    };
+
+    // We only check that all columns have been finished. That is to say,
+    // if one part is finished but another part is not, it will be considered
+    // unfinished.
+    add_columns
+        .add_columns
+        .iter()
+        .map(|add_column| add_column.column_def.as_ref().map(|c| &c.name))
+        .all(|column| {
+            column
+                .map(|c| existing_columns.contains(c))
+                .unwrap_or(false)
+        })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_retain_unskipped() {
+        let mut target = vec![1, 2, 3, 4, 5];
+        let skipped = vec![false, true, false, true, false];
+        retain_unskipped(&mut target, &skipped);
+        assert_eq!(target, vec![1, 3, 5]);
+    }
+}

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

@@ -12,10 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod check;
+mod executor;
 mod metadata;
 mod region_request;
-mod update_metadata;
 
 use std::vec;
 
@@ -29,30 +28,29 @@ use common_procedure::{
     Context as ProcedureContext, ContextProvider, Error as ProcedureError, LockKey, PoisonKey,
     PoisonKeys, Procedure, ProcedureId, Status, StringKey,
 };
-use common_telemetry::{debug, error, info};
+use common_telemetry::{error, info, warn};
-use futures::future::{self};
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, ResultExt};
-use store_api::storage::RegionId;
+use store_api::metadata::ColumnMetadata;
+use store_api::metric_engine_consts::TABLE_COLUMN_METADATA_EXTENSION_KEY;
 use strum::AsRefStr;
 use table::metadata::{RawTableInfo, TableId, TableInfo};
 use table::table_reference::TableReference;
 
-use crate::cache_invalidator::Context;
+use crate::ddl::alter_table::executor::AlterTableExecutor;
 use crate::ddl::utils::{
-    add_peer_context_if_needed, handle_multiple_results, map_to_procedure_error,
+    extract_column_metadatas, handle_multiple_results, map_to_procedure_error,
     sync_follower_regions, MultipleResults,
 };
 use crate::ddl::DdlContext;
 use crate::error::{AbortProcedureSnafu, NoLeaderSnafu, PutPoisonSnafu, Result, RetryLaterSnafu};
-use crate::instruction::CacheIdent;
 use crate::key::table_info::TableInfoValue;
 use crate::key::{DeserializedValueWithBytes, RegionDistribution};
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
 use crate::metrics;
 use crate::poison_key::table_poison_key;
 use crate::rpc::ddl::AlterTableTask;
-use crate::rpc::router::{find_leader_regions, find_leaders, region_distribution, RegionRoute};
+use crate::rpc::router::{find_leaders, region_distribution, RegionRoute};
 
 /// The alter table procedure
 pub struct AlterTableProcedure {
@@ -64,6 +62,24 @@ pub struct AlterTableProcedure {
     /// If we recover the procedure from json, then the table info value is not cached.
     /// But we already validated it in the prepare step.
     new_table_info: Option<TableInfo>,
+    /// The alter table executor.
+    executor: AlterTableExecutor,
+}
+
+/// Builds the executor from the [`AlterTableData`].
+///
+/// # Panics
+/// - If the alter kind is not set.
+fn build_executor_from_alter_expr(alter_data: &AlterTableData) -> AlterTableExecutor {
+    let table_name = alter_data.table_ref().into();
+    let table_id = alter_data.table_id;
+    let alter_kind = alter_data.task.alter_table.kind.as_ref().unwrap();
+    let new_table_name = if let Kind::RenameTable(RenameTable { new_table_name }) = alter_kind {
+        Some(new_table_name.to_string())
+    } else {
+        None
+    };
+    AlterTableExecutor::new(table_name, table_id, new_table_name)
 }
 
 impl AlterTableProcedure {
@@ -71,33 +87,42 @@ impl AlterTableProcedure {
 
     pub fn new(table_id: TableId, task: AlterTableTask, context: DdlContext) -> Result<Self> {
         task.validate()?;
+        let data = AlterTableData::new(task, table_id);
+        let executor = build_executor_from_alter_expr(&data);
         Ok(Self {
             context,
-            data: AlterTableData::new(task, table_id),
+            data,
             new_table_info: None,
+            executor,
         })
     }
 
     pub fn from_json(json: &str, context: DdlContext) -> ProcedureResult<Self> {
         let data: AlterTableData = serde_json::from_str(json).context(FromJsonSnafu)?;
+        let executor = build_executor_from_alter_expr(&data);
+
         Ok(AlterTableProcedure {
             context,
             data,
             new_table_info: None,
+            executor,
         })
     }
 
     // Checks whether the table exists.
     pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
-        self.check_alter().await?;
+        self.executor
+            .on_prepare(&self.context.table_metadata_manager)
+            .await?;
         self.fill_table_info().await?;
 
-        // Validates the request and builds the new table info.
+        // Safety: filled in `fill_table_info`.
-        // We need to build the new table info here because we should ensure the alteration
-        // is valid in `UpdateMeta` state as we already altered the region.
-        // Safety: `fill_table_info()` already set it.
         let table_info_value = self.data.table_info_value.as_ref().unwrap();
-        self.new_table_info = Some(self.build_new_table_info(&table_info_value.table_info)?);
+        let new_table_info = AlterTableExecutor::validate_alter_table_expr(
+            &table_info_value.table_info,
+            self.data.task.alter_table.clone(),
+        )?;
+        self.new_table_info = Some(new_table_info);
 
         // Safety: Checked in `AlterTableProcedure::new`.
         let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
@@ -140,9 +165,7 @@ impl AlterTableProcedure {
 
         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());
         let alter_kind = self.make_region_alter_kind()?;
 
         info!(
@@ -155,31 +178,14 @@ impl AlterTableProcedure {
         ensure!(!leaders.is_empty(), NoLeaderSnafu { table_id });
         // Puts the poison before submitting alter region requests to datanodes.
         self.put_poison(ctx_provider, procedure_id).await?;
-        for datanode in leaders {
+        let results = self
-            let requester = self.context.node_manager.datanode(&datanode).await;
+            .executor
-            let regions = find_leader_regions(&physical_table_route.region_routes, &datanode);
+            .on_alter_regions(
-
+                &self.context.node_manager,
-            for region in regions {
+                &physical_table_route.region_routes,
-                let region_id = RegionId::new(table_id, region);
+                alter_kind,
-                let request = self.make_alter_region_request(region_id, alter_kind.clone())?;
+            )
-                debug!("Submitting {request:?} to {datanode}");
+            .await;
-
-                let datanode = datanode.clone();
-                let requester = requester.clone();
-
-                alter_region_tasks.push(async move {
-                    requester
-                        .handle(request)
-                        .await
-                        .map_err(add_peer_context_if_needed(datanode))
-                });
-            }
-        }
-
-        let results = future::join_all(alter_region_tasks)
-            .await
-            .into_iter()
-            .collect::<Vec<_>>();
 
         match handle_multiple_results(results) {
             MultipleResults::PartialRetryable(error) => {
@@ -202,9 +208,9 @@ impl AlterTableProcedure {
                 })
             }
             MultipleResults::Ok(results) => {
-                self.submit_sync_region_requests(results, &physical_table_route.region_routes)
+                self.submit_sync_region_requests(&results, &physical_table_route.region_routes)
                     .await;
-                self.data.state = AlterTableState::UpdateMetadata;
+                self.handle_alter_region_response(results)?;
                 Ok(Status::executing_with_clean_poisons(true))
             }
             MultipleResults::AllNonRetryable(error) => {
@@ -220,9 +226,22 @@ impl AlterTableProcedure {
         }
     }
 
+    fn handle_alter_region_response(&mut self, mut results: Vec<RegionResponse>) -> Result<()> {
+        self.data.state = AlterTableState::UpdateMetadata;
+        if let Some(column_metadatas) =
+            extract_column_metadatas(&mut results, TABLE_COLUMN_METADATA_EXTENSION_KEY)?
+        {
+            self.data.column_metadatas = column_metadatas;
+        } else {
+            warn!("altering table result doesn't contains extension key `{TABLE_COLUMN_METADATA_EXTENSION_KEY}`,leaving the table's column metadata unchanged");
+        }
+
+        Ok(())
+    }
+
     async fn submit_sync_region_requests(
         &mut self,
-        results: Vec<RegionResponse>,
+        results: &[RegionResponse],
         region_routes: &[RegionRoute],
     ) {
         // Safety: filled in `prepare` step.
@@ -244,39 +263,34 @@ impl AlterTableProcedure {
     pub(crate) async fn on_update_metadata(&mut self) -> Result<Status> {
         let table_id = self.data.table_id();
         let table_ref = self.data.table_ref();
-        // Safety: checked before.
+        // Safety: filled in `fill_table_info`.
         let table_info_value = self.data.table_info_value.as_ref().unwrap();
+        // Safety: Checked in `AlterTableProcedure::new`.
+        let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
+
         // Gets the table info from the cache or builds it.
-        let new_info = match &self.new_table_info {
+        let  new_info = match &self.new_table_info {
             Some(cached) => cached.clone(),
-            None => self.build_new_table_info(&table_info_value.table_info)
+            None => AlterTableExecutor::validate_alter_table_expr(
+                &table_info_value.table_info,
+                self.data.task.alter_table.clone(),
+               )
                 .inspect_err(|e| {
                     // We already check the table info in the prepare step so this should not happen.
                     error!(e; "Unable to build info for table {} in update metadata step, table_id: {}", table_ref, table_id);
                 })?,
         };
 
-        debug!(
+        // Safety: region distribution is set in `submit_alter_region_requests`.
-            "Starting update table: {} metadata, new table info {:?}",
+        self.executor
-            table_ref.to_string(),
+            .on_alter_metadata(
-            new_info
+                &self.context.table_metadata_manager,
-        );
-
-        // Safety: Checked in `AlterTableProcedure::new`.
-        let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
-        if let Kind::RenameTable(RenameTable { new_table_name }) = alter_kind {
-            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,
+                self.data.region_distribution.as_ref(),
+                new_info.into(),
+                &self.data.column_metadatas,
             )
             .await?;
-        }
 
         info!("Updated table metadata for table {table_ref}, table_id: {table_id}, kind: {alter_kind:?}");
         self.data.state = AlterTableState::InvalidateTableCache;
@@ -285,18 +299,9 @@ impl AlterTableProcedure {
 
     /// Broadcasts the invalidating table cache instructions.
     async fn on_broadcast(&mut self) -> Result<Status> {
-        let cache_invalidator = &self.context.cache_invalidator;
+        self.executor
-
+            .invalidate_table_cache(&self.context.cache_invalidator)
-        cache_invalidator
-            .invalidate(
-                &Context::default(),
-                &[
-                    CacheIdent::TableId(self.data.table_id()),
-                    CacheIdent::TableName(self.data.table_ref().into()),
-                ],
-            )
             .await?;
-
         Ok(Status::done())
     }
 
@@ -318,6 +323,16 @@ impl AlterTableProcedure {
 
         lock_key
     }
+
+    #[cfg(test)]
+    pub(crate) fn data(&self) -> &AlterTableData {
+        &self.data
+    }
+
+    #[cfg(test)]
+    pub(crate) fn mut_data(&mut self) -> &mut AlterTableData {
+        &mut self.data
+    }
 }
 
 #[async_trait]
@@ -380,6 +395,8 @@ pub struct AlterTableData {
     state: AlterTableState,
     task: AlterTableTask,
     table_id: TableId,
+    #[serde(default)]
+    column_metadatas: Vec<ColumnMetadata>,
     /// Table info value before alteration.
     table_info_value: Option<DeserializedValueWithBytes<TableInfoValue>>,
     /// Region distribution for table in case we need to update region options.
@@ -392,6 +409,7 @@ impl AlterTableData {
             state: AlterTableState::Prepare,
             task,
             table_id,
+            column_metadatas: vec![],
             table_info_value: None,
             region_distribution: None,
         }
@@ -410,4 +428,14 @@ impl AlterTableData {
             .as_ref()
             .map(|value| &value.table_info)
     }
+
+    #[cfg(test)]
+    pub(crate) fn column_metadatas(&self) -> &[ColumnMetadata] {
+        &self.column_metadatas
+    }
+
+    #[cfg(test)]
+    pub(crate) fn set_column_metadatas(&mut self, column_metadatas: Vec<ColumnMetadata>) {
+        self.column_metadatas = column_metadatas;
+    }
 }