poc/create-alter-for-metrics:

### Commit Message

 Enhance Prometheus Bulk Write Handling

 - **`server.rs`**: Introduced `start_background_task` in `PromBulkState` to handle asynchronous batch processing and SST file writing. Added a new `tx` field to manage task communication.
 - **`access_layer.rs`**: Added `file_id` method to `ParquetWriter` for file identification.
 - **`batch_builder.rs`**: Modified `MetricsBatchBuilder` to utilize session catalog and schema, and updated batch processing logic to handle column metadata.
 - **`prom_store.rs`**: Updated `remote_write` to use `decode_remote_write_request_to_batch` for batch processing and send data to the background task.
 - **`prom_row_builder.rs`**: Made `TableBuilder` and `TablesBuilder` fields public for external access.
 - **`proto.rs`**: Exposed `table_data` in `PromWriteRequest` for batch processing.

Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>

.cargo/config.toml

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

.github/actions/setup-etcd-cluster/action.yml

@@ -12,7 +12,7 @@ runs:
   steps:
   - name: Install Etcd cluster
     shell: bash
-    run: |
+    run: | 
       helm upgrade \
         --install etcd oci://registry-1.docker.io/bitnamicharts/etcd \
         --set replicaCount=${{ inputs.etcd-replicas }} \
@@ -24,9 +24,4 @@ runs:
         --set auth.rbac.token.enabled=false \
         --set persistence.size=2Gi \
         --create-namespace \
-        --set global.security.allowInsecureImages=true \
-        --set image.registry=docker.io \
-        --set image.repository=greptime/etcd \
-        --set image.tag=3.6.1-debian-12-r3 \
-        --version 12.0.8 \
         -n ${{ inputs.namespace }}

.github/actions/setup-kafka-cluster/action.yml

@@ -12,7 +12,7 @@ runs:
   steps:
   - name: Install Kafka cluster
     shell: bash
-    run: |
+    run: | 
       helm upgrade \
         --install kafka oci://registry-1.docker.io/bitnamicharts/kafka \
         --set controller.replicaCount=${{ inputs.controller-replicas }} \
@@ -23,8 +23,4 @@ runs:
         --set listeners.controller.protocol=PLAINTEXT \
         --set listeners.client.protocol=PLAINTEXT \
         --create-namespace \
-        --set image.registry=docker.io \
-        --set image.repository=greptime/kafka \
-        --set image.tag=3.9.0-debian-12-r1 \
-        --version 31.0.0 \
         -n ${{ inputs.namespace }}

.github/actions/setup-postgres-cluster/action.yml

@@ -6,7 +6,9 @@ inputs:
     description: "Number of PostgreSQL replicas"
   namespace:
     default: "postgres-namespace"
-    description: "The PostgreSQL namespace"
+  postgres-version:
+    default: "14.2"
+    description: "PostgreSQL version"
   storage-size:
     default: "1Gi"
     description: "Storage size for PostgreSQL"
@@ -20,11 +22,7 @@ runs:
       helm upgrade \
         --install postgresql oci://registry-1.docker.io/bitnamicharts/postgresql \
         --set replicaCount=${{ inputs.postgres-replicas }} \
-        --set global.security.allowInsecureImages=true \
-        --set image.registry=docker.io \
-        --set image.repository=greptime/postgresql \
-        --set image.tag=17.5.0-debian-12-r3 \
-        --version 16.7.4 \
+        --set image.tag=${{ inputs.postgres-version }} \
         --set persistence.size=${{ inputs.storage-size }} \
         --set postgresql.username=greptimedb \
         --set postgresql.password=admin \

.github/scripts/check-version.sh

@@ -1,42 +0,0 @@
-#!/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/scripts/pull-test-deps-images.sh

@@ -1,34 +0,0 @@
-#!/bin/bash
-
-# This script is used to pull the test dependency images that are stored in public ECR one by one to avoid rate limiting.
-
-set -e
-
-MAX_RETRIES=3
-
-IMAGES=(
-  "greptime/zookeeper:3.7"
-  "greptime/kafka:3.9.0-debian-12-r1"
-  "greptime/etcd:3.6.1-debian-12-r3"
-  "greptime/minio:2024"
-  "greptime/mysql:5.7"
-)
-
-for image in "${IMAGES[@]}"; do
-  for ((attempt=1; attempt<=MAX_RETRIES; attempt++)); do
-    if docker pull "$image"; then
-      # Successfully pulled the image.
-      break
-    else
-      # Use some simple exponential backoff to avoid rate limiting.
-      if [ $attempt -lt $MAX_RETRIES ]; then
-        sleep_seconds=$((attempt * 5))
-        echo "Attempt $attempt failed for $image, waiting $sleep_seconds seconds"
-        sleep $sleep_seconds  # 5s, 10s delays
-      else
-        echo "Failed to pull $image after $MAX_RETRIES attempts"
-        exit 1
-      fi
-    fi
-  done
-done

.github/workflows/develop.yml

@@ -719,10 +719,6 @@ jobs:
           save-if: ${{ github.ref == 'refs/heads/main' }}
       - name: Install latest nextest release
         uses: taiki-e/install-action@nextest
-
-      - name: Pull test dependencies images
-        run: ./.github/scripts/pull-test-deps-images.sh
-
       - name: Setup external services
         working-directory: tests-integration/fixtures
         run: docker compose up -d --wait

.github/workflows/release.yml

@@ -110,8 +110,6 @@ 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
@@ -137,11 +135,6 @@ 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
@@ -321,7 +314,7 @@ jobs:
           image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
           image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
           version: ${{ needs.allocate-runners.outputs.version }}
-          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' }}
+          push-latest-tag: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
 
       - name: Set build image result
         id: set-build-image-result
@@ -368,7 +361,7 @@ jobs:
           dev-mode: false
           upload-to-s3: true
           update-version-info: true
-          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' }}
+          push-latest-tag: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
 
   publish-github-release:
     name: Create GitHub release and upload artifacts

Cargo.toml

@@ -71,7 +71,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.15.5"
+version = "0.15.0"
 edition = "2021"
 license = "Apache-2.0"
 
@@ -121,6 +121,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,12 +131,11 @@ 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 = "f3103a8c9b8ce162457d0a3e3ca00d53d1a8bd06" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "464226cf8a4a22696503536a123d0b9e318582f4" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -221,8 +221,6 @@ tokio-util = { version = "0.7", features = ["io-util", "compat"] }
 toml = "0.8.8"
 tonic = { version = "0.12", features = ["tls", "gzip", "zstd"] }
 tower = "0.5"
-tower-http = "0.6"
-tracing = "0.1"
 tracing-appender = "0.2"
 tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "fmt"] }
 typetag = "0.2"

Makefile

@@ -8,7 +8,7 @@ CARGO_BUILD_OPTS := --locked
 IMAGE_REGISTRY ?= docker.io
 IMAGE_NAMESPACE ?= greptime
 IMAGE_TAG ?= latest
-DEV_BUILDER_IMAGE_TAG ?= 2025-05-19-f55023f3-20250829091211
+DEV_BUILDER_IMAGE_TAG ?= 2025-05-19-b2377d4b-20250520045554
 BUILDX_MULTI_PLATFORM_BUILD ?= false
 BUILDX_BUILDER_NAME ?= gtbuilder
 BASE_IMAGE ?= ubuntu

config/config.md

@@ -147,7 +147,6 @@
 | `region_engine.mito.write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
 | `region_engine.mito.parallel_scan_channel_size` | Integer | `32` | Capacity of the channel to send data from parallel scan tasks to the main task. |
-| `region_engine.mito.max_concurrent_scan_files` | Integer | `128` | Maximum number of SST files to scan concurrently. |
 | `region_engine.mito.allow_stale_entries` | Bool | `false` | Whether to allow stale WAL entries read during replay. |
 | `region_engine.mito.min_compaction_interval` | String | `0m` | Minimum time interval between two compactions.<br/>To align with the old behavior, the default value is 0 (no restrictions). |
 | `region_engine.mito.index` | -- | -- | The options for index in Mito engine. |
@@ -186,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. |
@@ -289,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. |
@@ -324,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` | Delay before initializing region failure detectors.<br/>This delay helps prevent premature initialization of region failure detectors in cases where<br/>cluster maintenance mode is enabled right after metasrv starts, especially when the cluster<br/>is not deployed via the recommended GreptimeDB Operator. Without this delay, early detector registration<br/>may trigger unnecessary region failovers during datanode startup. |
 | `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. |
@@ -371,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. |
@@ -497,7 +500,6 @@
 | `region_engine.mito.write_cache_ttl` | String | Unset | TTL for write cache. |
 | `region_engine.mito.sst_write_buffer_size` | String | `8MB` | Buffer size for SST writing. |
 | `region_engine.mito.parallel_scan_channel_size` | Integer | `32` | Capacity of the channel to send data from parallel scan tasks to the main task. |
-| `region_engine.mito.max_concurrent_scan_files` | Integer | `128` | Maximum number of SST files to scan concurrently. |
 | `region_engine.mito.allow_stale_entries` | Bool | `false` | Whether to allow stale WAL entries read during replay. |
 | `region_engine.mito.min_compaction_interval` | String | `0m` | Minimum time interval between two compactions.<br/>To align with the old behavior, the default value is 0 (no restrictions). |
 | `region_engine.mito.index` | -- | -- | The options for index in Mito engine. |
@@ -536,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. |
@@ -586,10 +589,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` | -- |
 | `tracing` | -- | -- | The tracing options. Only effect when compiled with `tokio-console` feature. |

config/datanode.example.toml

@@ -474,9 +474,6 @@ sst_write_buffer_size = "8MB"
 ## Capacity of the channel to send data from parallel scan tasks to the main task.
 parallel_scan_channel_size = 32
 
-## Maximum number of SST files to scan concurrently.
-max_concurrent_scan_files = 128
-
 ## Whether to allow stale WAL entries read during replay.
 allow_stale_entries = false
 
@@ -632,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
@@ -643,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

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,13 @@ use_memory_store = false
 ## - Using shared storage (e.g., s3).
 enable_region_failover = false
 
+## Delay before initializing region failure detectors.
+## This delay helps prevent premature initialization of region failure detectors in cases where
+## cluster maintenance mode is enabled right after metasrv starts, especially when the cluster
+## is not deployed via the recommended GreptimeDB Operator. Without this delay, early detector registration
+## may trigger unnecessary region failovers during datanode startup.
+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 +227,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 +238,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

@@ -565,9 +565,6 @@ sst_write_buffer_size = "8MB"
 ## Capacity of the channel to send data from parallel scan tasks to the main task.
 parallel_scan_channel_size = 32
 
-## Maximum number of SST files to scan concurrently.
-max_concurrent_scan_files = 128
-
 ## Whether to allow stale WAL entries read during replay.
 allow_stale_entries = false
 
@@ -723,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
@@ -734,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

licenserc.toml

@@ -31,6 +31,7 @@ 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/api/src/v1/column_def.rs

@@ -24,7 +24,7 @@ use greptime_proto::v1::{
 };
 use snafu::ResultExt;
 
-use crate::error::{self, ConvertColumnDefaultConstraintSnafu, Result};
+use crate::error::{self, Result};
 use crate::helper::ColumnDataTypeWrapper;
 use crate::v1::{ColumnDef, ColumnOptions, SemanticType};
 
@@ -77,48 +77,6 @@ pub fn try_as_column_schema(column_def: &ColumnDef) -> Result<ColumnSchema> {
         })
 }
 
-/// Tries to construct a `ColumnDef` from the given `ColumnSchema`.
-///
-/// TODO(weny): Add tests for this function.
-pub fn try_as_column_def(column_schema: &ColumnSchema, is_primary_key: bool) -> Result<ColumnDef> {
-    let column_datatype =
-        ColumnDataTypeWrapper::try_from(column_schema.data_type.clone()).map(|w| w.to_parts())?;
-
-    let semantic_type = if column_schema.is_time_index() {
-        SemanticType::Timestamp
-    } else if is_primary_key {
-        SemanticType::Tag
-    } else {
-        SemanticType::Field
-    } as i32;
-    let comment = column_schema
-        .metadata()
-        .get(COMMENT_KEY)
-        .cloned()
-        .unwrap_or_default();
-
-    let default_constraint = match column_schema.default_constraint() {
-        None => vec![],
-        Some(v) => v
-            .clone()
-            .try_into()
-            .context(ConvertColumnDefaultConstraintSnafu {
-                column: &column_schema.name,
-            })?,
-    };
-    let options = options_from_column_schema(column_schema);
-    Ok(ColumnDef {
-        name: column_schema.name.clone(),
-        data_type: column_datatype.0 as i32,
-        is_nullable: column_schema.is_nullable(),
-        default_constraint,
-        semantic_type,
-        comment,
-        datatype_extension: column_datatype.1,
-        options,
-    })
-}
-
 /// Constructs a `ColumnOptions` from the given `ColumnSchema`.
 pub fn options_from_column_schema(column_schema: &ColumnSchema) -> Option<ColumnOptions> {
     let mut options = ColumnOptions::default();
@@ -268,20 +226,18 @@ mod tests {
         assert!(options.is_none());
 
         let mut schema = ColumnSchema::new("test", ConcreteDataType::string_datatype(), true)
-            .with_fulltext_options(FulltextOptions::new_unchecked(
-                true,
-                FulltextAnalyzer::English,
-                false,
-                FulltextBackend::Bloom,
-                10240,
-                0.01,
-            ))
+            .with_fulltext_options(FulltextOptions {
+                enable: true,
+                analyzer: FulltextAnalyzer::English,
+                case_sensitive: false,
+                backend: FulltextBackend::Bloom,
+            })
             .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\",\"granularity\":10240,\"false-positive-rate-in-10000\":100}"
+            "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\"}"
         );
         assert_eq!(
             options.options.get(INVERTED_INDEX_GRPC_KEY).unwrap(),
@@ -291,18 +247,16 @@ mod tests {
 
     #[test]
     fn test_options_with_fulltext() {
-        let fulltext = FulltextOptions::new_unchecked(
-            true,
-            FulltextAnalyzer::English,
-            false,
-            FulltextBackend::Bloom,
-            10240,
-            0.01,
-        );
+        let fulltext = FulltextOptions {
+            enable: true,
+            analyzer: FulltextAnalyzer::English,
+            case_sensitive: false,
+            backend: FulltextBackend::Bloom,
+        };
         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\",\"granularity\":10240,\"false-positive-rate-in-10000\":100}"
+            "{\"enable\":true,\"analyzer\":\"English\",\"case-sensitive\":false,\"backend\":\"bloom\"}"
         );
     }
 

src/catalog/Cargo.toml

@@ -43,8 +43,6 @@ moka = { workspace = true, features = ["future", "sync"] }
 partition.workspace = true
 paste.workspace = true
 prometheus.workspace = true
-promql-parser.workspace = true
-rand.workspace = true
 rustc-hash.workspace = true
 serde_json.workspace = true
 session.workspace = true

src/catalog/src/information_extension.rs

@@ -16,8 +16,8 @@ use api::v1::meta::ProcedureStatus;
 use common_error::ext::BoxedError;
 use common_meta::cluster::{ClusterInfo, NodeInfo};
 use common_meta::datanode::RegionStat;
+use common_meta::ddl::{ExecutorContext, ProcedureExecutor};
 use common_meta::key::flow::flow_state::FlowStat;
-use common_meta::procedure_executor::{ExecutorContext, ProcedureExecutor};
 use common_meta::rpc::procedure;
 use common_procedure::{ProcedureInfo, ProcedureState};
 use meta_client::MetaClientRef;

src/catalog/src/kvbackend/manager.rs

@@ -28,7 +28,7 @@ use common_meta::cache::{
 use common_meta::key::catalog_name::CatalogNameKey;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::schema_name::SchemaNameKey;
-use common_meta::key::table_info::{TableInfoManager, TableInfoValue};
+use common_meta::key::table_info::TableInfoValue;
 use common_meta::key::table_name::TableNameKey;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
@@ -39,7 +39,6 @@ use moka::sync::Cache;
 use partition::manager::{PartitionRuleManager, PartitionRuleManagerRef};
 use session::context::{Channel, QueryContext};
 use snafu::prelude::*;
-use store_api::metric_engine_consts::METRIC_ENGINE_NAME;
 use table::dist_table::DistTable;
 use table::metadata::TableId;
 use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
@@ -143,61 +142,6 @@ 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]
@@ -324,7 +268,10 @@ impl CatalogManager for KvBackendCatalogManager {
         let table_cache: TableCacheRef = self.cache_registry.get().context(CacheNotFoundSnafu {
             name: "table_cache",
         })?;
-
+        let table_route_cache: TableRouteCacheRef =
+            self.cache_registry.get().context(CacheNotFoundSnafu {
+                name: "table_route_cache",
+            })?;
         let table = table_cache
             .get_by_ref(&TableName {
                 catalog_name: catalog_name.to_string(),
@@ -334,18 +281,55 @@ impl CatalogManager for KvBackendCatalogManager {
             .await
             .context(GetTableCacheSnafu)?;
 
-        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);
+        // Override logical table's partition key indices with physical table's.
+        if let Some(table) = &table
+            && let Some(table_route_value) = table_route_cache
+                .get(table.table_info().table_id())
+                .await
+                .context(TableMetadataManagerSnafu)?
+            && let TableRoute::Logical(logical_route) = &*table_route_value
+            && let Some(physical_table_info_value) = self
+                .table_metadata_manager
+                .table_info_manager()
+                .get(logical_route.physical_table_id())
+                .await
+                .context(TableMetadataManagerSnafu)?
+        {
+            let mut new_table_info = (*table.table_info()).clone();
+            // Gather all column names from the logical table
+            let logical_column_names: std::collections::HashSet<_> = new_table_info
+                .meta
+                .schema
+                .column_schemas()
+                .iter()
+                .map(|col| &col.name)
+                .collect();
+
+            // Only preserve partition key indices where the corresponding columns exist in logical table
+            new_table_info.meta.partition_key_indices = physical_table_info_value
+                .table_info
+                .meta
+                .partition_key_indices
+                .iter()
+                .filter(|&&index| {
+                    if let Some(physical_column) = physical_table_info_value
+                        .table_info
+                        .meta
+                        .schema
+                        .column_schemas
+                        .get(index)
+                    {
+                        logical_column_names.contains(&physical_column.name)
+                    } else {
+                        false
+                    }
+                })
+                .cloned()
+                .collect();
+
+            let new_table = DistTable::table(Arc::new(new_table_info));
+
+            return Ok(Some(new_table));
         }
 
         if channel == Channel::Postgres {
@@ -358,7 +342,7 @@ impl CatalogManager for KvBackendCatalogManager {
             }
         }
 
-        Ok(None)
+        Ok(table)
     }
 
     async fn tables_by_ids(
@@ -410,20 +394,8 @@ 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)
@@ -450,7 +422,6 @@ 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;
@@ -468,16 +439,6 @@ 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;
                         }

src/catalog/src/process_manager.rs

@@ -14,24 +14,17 @@
 
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
-use std::fmt::{Debug, Display, Formatter};
+use std::fmt::{Debug, Formatter};
 use std::sync::atomic::{AtomicU32, Ordering};
 use std::sync::{Arc, RwLock};
-use std::time::{Duration, Instant, UNIX_EPOCH};
 
 use api::v1::frontend::{KillProcessRequest, ListProcessRequest, ProcessInfo};
 use common_base::cancellation::CancellationHandle;
 use common_frontend::selector::{FrontendSelector, MetaClientSelector};
-use common_frontend::slow_query_event::SlowQueryEvent;
-use common_telemetry::{debug, error, info, warn};
+use common_telemetry::{debug, info};
 use common_time::util::current_time_millis;
 use meta_client::MetaClientRef;
-use promql_parser::parser::EvalStmt;
-use rand::random;
-use session::context::QueryContextRef;
 use snafu::{ensure, OptionExt, ResultExt};
-use sql::statements::statement::Statement;
-use tokio::sync::mpsc::Sender;
 
 use crate::error;
 use crate::metrics::{PROCESS_KILL_COUNT, PROCESS_LIST_COUNT};
@@ -51,23 +44,6 @@ pub struct ProcessManager {
     frontend_selector: Option<MetaClientSelector>,
 }
 
-/// Represents a parsed query statement, functionally equivalent to [query::parser::QueryStatement].
-/// This enum is defined here to avoid cyclic dependencies with the query parser module.
-#[derive(Debug, Clone)]
-pub enum QueryStatement {
-    Sql(Statement),
-    Promql(EvalStmt),
-}
-
-impl Display for QueryStatement {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        match self {
-            QueryStatement::Sql(stmt) => write!(f, "{}", stmt),
-            QueryStatement::Promql(eval_stmt) => write!(f, "{}", eval_stmt),
-        }
-    }
-}
-
 impl ProcessManager {
     /// Create a [ProcessManager] instance with server address and kv client.
     pub fn new(server_addr: String, meta_client: Option<MetaClientRef>) -> Self {
@@ -91,7 +67,6 @@ impl ProcessManager {
         query: String,
         client: String,
         query_id: Option<ProcessId>,
-        _slow_query_timer: Option<SlowQueryTimer>,
     ) -> Ticket {
         let id = query_id.unwrap_or_else(|| self.next_id.fetch_add(1, Ordering::Relaxed));
         let process = ProcessInfo {
@@ -118,7 +93,6 @@ impl ProcessManager {
             manager: self.clone(),
             id,
             cancellation_handle,
-            _slow_query_timer,
         }
     }
 
@@ -167,20 +141,14 @@ impl ProcessManager {
                 .await
                 .context(error::InvokeFrontendSnafu)?;
             for mut f in frontends {
-                let result = f
-                    .list_process(ListProcessRequest {
+                processes.extend(
+                    f.list_process(ListProcessRequest {
                         catalog: catalog.unwrap_or_default().to_string(),
                     })
                     .await
-                    .context(error::InvokeFrontendSnafu);
-                match result {
-                    Ok(resp) => {
-                        processes.extend(resp.processes);
-                    }
-                    Err(e) => {
-                        warn!(e; "Skipping failing node: {:?}", f)
-                    }
-                }
+                    .context(error::InvokeFrontendSnafu)?
+                    .processes,
+                );
             }
         }
         processes.extend(self.local_processes(catalog)?);
@@ -249,7 +217,6 @@ pub struct Ticket {
     pub(crate) manager: ProcessManagerRef,
     pub(crate) id: ProcessId,
     pub cancellation_handle: Arc<CancellationHandle>,
-    _slow_query_timer: Option<SlowQueryTimer>,
 }
 
 impl Drop for Ticket {
@@ -290,107 +257,6 @@ impl Debug for CancellableProcess {
     }
 }
 
-/// SlowQueryTimer is used to log slow query when it's dropped.
-/// In drop(), it will check if the query is slow and send the slow query event to the handler.
-pub struct SlowQueryTimer {
-    start: Instant,
-    stmt: QueryStatement,
-    query_ctx: QueryContextRef,
-    threshold: Option<Duration>,
-    sample_ratio: Option<f64>,
-    tx: Sender<SlowQueryEvent>,
-}
-
-impl SlowQueryTimer {
-    pub fn new(
-        stmt: QueryStatement,
-        query_ctx: QueryContextRef,
-        threshold: Option<Duration>,
-        sample_ratio: Option<f64>,
-        tx: Sender<SlowQueryEvent>,
-    ) -> Self {
-        Self {
-            start: Instant::now(),
-            stmt,
-            query_ctx,
-            threshold,
-            sample_ratio,
-            tx,
-        }
-    }
-}
-
-impl SlowQueryTimer {
-    fn send_slow_query_event(&self, elapsed: Duration, threshold: Duration) {
-        let mut slow_query_event = SlowQueryEvent {
-            cost: elapsed.as_millis() as u64,
-            threshold: threshold.as_millis() as u64,
-            query: "".to_string(),
-            query_ctx: self.query_ctx.clone(),
-
-            // The following fields are only used for PromQL queries.
-            is_promql: false,
-            promql_range: None,
-            promql_step: None,
-            promql_start: None,
-            promql_end: None,
-        };
-
-        match &self.stmt {
-            QueryStatement::Promql(stmt) => {
-                slow_query_event.is_promql = true;
-                slow_query_event.query = stmt.expr.to_string();
-                slow_query_event.promql_step = Some(stmt.interval.as_millis() as u64);
-
-                let start = stmt
-                    .start
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap_or_default()
-                    .as_millis() as i64;
-
-                let end = stmt
-                    .end
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap_or_default()
-                    .as_millis() as i64;
-
-                slow_query_event.promql_range = Some((end - start) as u64);
-                slow_query_event.promql_start = Some(start);
-                slow_query_event.promql_end = Some(end);
-            }
-            QueryStatement::Sql(stmt) => {
-                slow_query_event.query = stmt.to_string();
-            }
-        }
-
-        // Send SlowQueryEvent to the handler.
-        if let Err(e) = self.tx.try_send(slow_query_event) {
-            error!(e; "Failed to send slow query event");
-        }
-    }
-}
-
-impl Drop for SlowQueryTimer {
-    fn drop(&mut self) {
-        if let Some(threshold) = self.threshold {
-            // Calculate the elaspsed duration since the timer is created.
-            let elapsed = self.start.elapsed();
-            if elapsed > threshold {
-                if let Some(ratio) = self.sample_ratio {
-                    // Only capture a portion of slow queries based on sample_ratio.
-                    // Generate a random number in [0, 1) and compare it with sample_ratio.
-                    if ratio >= 1.0 || random::<f64>() <= ratio {
-                        self.send_slow_query_event(elapsed, threshold);
-                    }
-                } else {
-                    // Captures all slow queries if sample_ratio is not set.
-                    self.send_slow_query_event(elapsed, threshold);
-                }
-            }
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -406,7 +272,6 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "".to_string(),
             None,
-            None,
         );
 
         let running_processes = process_manager.local_processes(None).unwrap();
@@ -430,7 +295,6 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "client1".to_string(),
             Some(custom_id),
-            None,
         );
 
         assert_eq!(ticket.id, custom_id);
@@ -451,7 +315,6 @@ mod tests {
             "SELECT * FROM table1".to_string(),
             "client1".to_string(),
             None,
-            None,
         );
 
         let ticket2 = process_manager.clone().register_query(
@@ -460,7 +323,6 @@ mod tests {
             "SELECT * FROM table2".to_string(),
             "client2".to_string(),
             None,
-            None,
         );
 
         let running_processes = process_manager.local_processes(Some("public")).unwrap();
@@ -482,7 +344,6 @@ mod tests {
             "SELECT * FROM table1".to_string(),
             "client1".to_string(),
             None,
-            None,
         );
 
         let _ticket2 = process_manager.clone().register_query(
@@ -491,7 +352,6 @@ mod tests {
             "SELECT * FROM table2".to_string(),
             "client2".to_string(),
             None,
-            None,
         );
 
         // Test listing processes for specific catalog
@@ -518,7 +378,6 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "client1".to_string(),
             None,
-            None,
         );
         assert_eq!(process_manager.local_processes(None).unwrap().len(), 1);
         process_manager.deregister_query("public".to_string(), ticket.id);
@@ -535,7 +394,6 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "client1".to_string(),
             None,
-            None,
         );
 
         assert!(!ticket.cancellation_handle.is_cancelled());
@@ -553,7 +411,6 @@ mod tests {
             "SELECT * FROM table".to_string(),
             "client1".to_string(),
             None,
-            None,
         );
         assert!(!ticket.cancellation_handle.is_cancelled());
         let killed = process_manager
@@ -599,7 +456,6 @@ mod tests {
             "SELECT COUNT(*) FROM users WHERE age > 18".to_string(),
             "test_client".to_string(),
             Some(42),
-            None,
         );
 
         let processes = process_manager.local_processes(None).unwrap();
@@ -626,7 +482,6 @@ mod tests {
                 "SELECT * FROM table".to_string(),
                 "client1".to_string(),
                 None,
-                None,
             );
 
             // Process should be registered

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;

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";
-const CATALOG: &str = "catalog";
-const SCHEMAS: &str = "schemas";
-const QUERY: &str = "query";
-const CLIENT: &str = "client";
-const FRONTEND: &str = "frontend";
-const START_TIMESTAMP: &str = "start_timestamp";
-const ELAPSED_TIME: &str = "elapsed_time";
+pub const ID: &str = "id";
+pub const CATALOG: &str = "catalog";
+pub const SCHEMAS: &str = "schemas";
+pub const QUERY: &str = "query";
+pub const CLIENT: &str = "client";
+pub const FRONTEND: &str = "frontend";
+pub const START_TIMESTAMP: &str = "start_timestamp";
+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/region_statistics.rs

@@ -40,7 +40,6 @@ const REGION_ID: &str = "region_id";
 const TABLE_ID: &str = "table_id";
 const REGION_NUMBER: &str = "region_number";
 const REGION_ROWS: &str = "region_rows";
-const WRITTEN_BYTES: &str = "written_bytes_since_open";
 const DISK_SIZE: &str = "disk_size";
 const MEMTABLE_SIZE: &str = "memtable_size";
 const MANIFEST_SIZE: &str = "manifest_size";
@@ -57,7 +56,6 @@ const INIT_CAPACITY: usize = 42;
 /// - `table_id`: The table id.
 /// - `region_number`: The region number.
 /// - `region_rows`: The number of rows in region.
-/// - `written_bytes_since_open`: The total bytes written of the region since region opened.
 /// - `memtable_size`: The memtable size in bytes.
 /// - `disk_size`: The approximate disk size in bytes.
 /// - `manifest_size`: The manifest size in bytes.
@@ -85,7 +83,6 @@ impl InformationSchemaRegionStatistics {
             ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
             ColumnSchema::new(REGION_NUMBER, ConcreteDataType::uint32_datatype(), false),
             ColumnSchema::new(REGION_ROWS, ConcreteDataType::uint64_datatype(), true),
-            ColumnSchema::new(WRITTEN_BYTES, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(DISK_SIZE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(MEMTABLE_SIZE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(MANIFEST_SIZE, ConcreteDataType::uint64_datatype(), true),
@@ -148,7 +145,6 @@ struct InformationSchemaRegionStatisticsBuilder {
     table_ids: UInt32VectorBuilder,
     region_numbers: UInt32VectorBuilder,
     region_rows: UInt64VectorBuilder,
-    written_bytes: UInt64VectorBuilder,
     disk_sizes: UInt64VectorBuilder,
     memtable_sizes: UInt64VectorBuilder,
     manifest_sizes: UInt64VectorBuilder,
@@ -167,7 +163,6 @@ impl InformationSchemaRegionStatisticsBuilder {
             table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
             region_numbers: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
             region_rows: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
-            written_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             disk_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             memtable_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             manifest_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
@@ -198,7 +193,6 @@ impl InformationSchemaRegionStatisticsBuilder {
             (TABLE_ID, &Value::from(region_stat.id.table_id())),
             (REGION_NUMBER, &Value::from(region_stat.id.region_number())),
             (REGION_ROWS, &Value::from(region_stat.num_rows)),
-            (WRITTEN_BYTES, &Value::from(region_stat.written_bytes)),
             (DISK_SIZE, &Value::from(region_stat.approximate_bytes)),
             (MEMTABLE_SIZE, &Value::from(region_stat.memtable_size)),
             (MANIFEST_SIZE, &Value::from(region_stat.manifest_size)),
@@ -217,7 +211,6 @@ impl InformationSchemaRegionStatisticsBuilder {
         self.region_numbers
             .push(Some(region_stat.id.region_number()));
         self.region_rows.push(Some(region_stat.num_rows));
-        self.written_bytes.push(Some(region_stat.written_bytes));
         self.disk_sizes.push(Some(region_stat.approximate_bytes));
         self.memtable_sizes.push(Some(region_stat.memtable_size));
         self.manifest_sizes.push(Some(region_stat.manifest_size));
@@ -233,7 +226,6 @@ impl InformationSchemaRegionStatisticsBuilder {
             Arc::new(self.table_ids.finish()),
             Arc::new(self.region_numbers.finish()),
             Arc::new(self.region_rows.finish()),
-            Arc::new(self.written_bytes.finish()),
             Arc::new(self.disk_sizes.finish()),
             Arc::new(self.memtable_sizes.finish()),
             Arc::new(self.manifest_sizes.finish()),

src/cli/Cargo.toml

@@ -43,6 +43,7 @@ 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

src/cli/src/bench.rs

@@ -160,7 +160,6 @@ 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/metadata/control/get.rs

@@ -29,7 +29,7 @@ use futures::TryStreamExt;
 
 use crate::error::InvalidArgumentsSnafu;
 use crate::metadata::common::StoreConfig;
-use crate::metadata::control::utils::{decode_key_value, get_table_id_by_name, json_formatter};
+use crate::metadata::control::utils::{decode_key_value, get_table_id_by_name, json_fromatter};
 use crate::Tool;
 
 /// Getting metadata from metadata store.
@@ -206,7 +206,7 @@ impl Tool for GetTableTool {
             println!(
                 "{}\n{}",
                 TableInfoKey::new(table_id),
-                json_formatter(self.pretty, &*table_info)
+                json_fromatter(self.pretty, &*table_info)
             );
         } else {
             println!("Table info not found");
@@ -221,7 +221,7 @@ impl Tool for GetTableTool {
             println!(
                 "{}\n{}",
                 TableRouteKey::new(table_id),
-                json_formatter(self.pretty, &table_route)
+                json_fromatter(self.pretty, &table_route)
             );
         } else {
             println!("Table route not found");

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

@@ -27,7 +27,7 @@ pub fn decode_key_value(kv: KeyValue) -> CommonMetaResult<(String, String)> {
 }
 
 /// Formats a value as a JSON string.
-pub fn json_formatter<T>(pretty: bool, value: &T) -> String
+pub fn json_fromatter<T>(pretty: bool, value: &T) -> String
 where
     T: Serialize,
 {

src/cli/src/metadata/repair.rs

@@ -241,6 +241,7 @@ impl RepairTool {
             let alter_table_request = alter_table::make_alter_region_request_for_peer(
                 logical_table_id,
                 &alter_table_expr,
+                full_table_metadata.table_info.ident.version,
                 peer,
                 physical_region_routes,
             )?;

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

@@ -66,6 +66,7 @@ pub fn generate_alter_table_expr_for_all_columns(
 pub fn make_alter_region_request_for_peer(
     logical_table_id: TableId,
     alter_table_expr: &AlterTableExpr,
+    schema_version: u64,
     peer: &Peer,
     region_routes: &[RegionRoute],
 ) -> Result<RegionRequest> {
@@ -73,7 +74,7 @@ pub fn make_alter_region_request_for_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);
+        let request = make_alter_region_request(region_id, alter_table_expr, schema_version);
         requests.push(request);
     }
 

src/client/src/database.rs

@@ -211,18 +211,12 @@ impl Database {
                         retries += 1;
                         warn!("Retrying {} times with error = {:?}", retries, err);
                         continue;
-                    } else {
-                        error!(
-                            err; "Failed to send request to grpc handle, retries = {}, not retryable error, aborting",
-                            retries
-                        );
-                        return Err(err.into());
                     }
                 }
                 (Err(err), false) => {
                     error!(
-                        err; "Failed to send request to grpc handle after {} retries",
-                        retries,
+                        "Failed to send request to grpc handle after {} retries, error = {:?}",
+                        retries, err
                     );
                     return Err(err.into());
                 }

src/client/src/region.rs

@@ -163,70 +163,19 @@ impl RegionRequester {
             let _span = tracing_context.attach(common_telemetry::tracing::info_span!(
                 "poll_flight_data_stream"
             ));
-
-            let mut buffered_message: Option<FlightMessage> = None;
-            let mut stream_ended = false;
-
-            while !stream_ended {
-                // get the next message from the buffered message or read from the flight message stream
-                let flight_message_item = if let Some(msg) = buffered_message.take() {
-                    Some(Ok(msg))
-                } else {
-                    flight_message_stream.next().await
-                };
-
-                let flight_message = match flight_message_item {
-                    Some(Ok(message)) => message,
-                    Some(Err(e)) => {
-                        yield Err(BoxedError::new(e)).context(ExternalSnafu);
-                        break;
-                    }
-                    None => break,
-                };
+            while let Some(flight_message) = flight_message_stream.next().await {
+                let flight_message = flight_message
+                    .map_err(BoxedError::new)
+                    .context(ExternalSnafu)?;
 
                 match flight_message {
                     FlightMessage::RecordBatch(record_batch) => {
-                        let result_to_yield = RecordBatch::try_from_df_record_batch(
+                        yield RecordBatch::try_from_df_record_batch(
                             schema_cloned.clone(),
                             record_batch,
-                        );
-
-                        // get the next message from the stream. normally it should be a metrics message.
-                        if let Some(next_flight_message_result) = flight_message_stream.next().await
-                        {
-                            match next_flight_message_result {
-                                Ok(FlightMessage::Metrics(s)) => {
-                                    let m = serde_json::from_str(&s).ok().map(Arc::new);
-                                    metrics_ref.swap(m);
-                                }
-                                Ok(FlightMessage::RecordBatch(rb)) => {
-                                    // for some reason it's not a metrics message, so we need to buffer this record batch
-                                    // and yield it in the next iteration.
-                                    buffered_message = Some(FlightMessage::RecordBatch(rb));
-                                }
-                                Ok(_) => {
-                                    yield IllegalFlightMessagesSnafu {
-                                        reason: "A RecordBatch message can only be succeeded by a Metrics message or another RecordBatch message"
-                                    }
-                                    .fail()
-                                    .map_err(BoxedError::new)
-                                    .context(ExternalSnafu);
-                                    break;
-                                }
-                                Err(e) => {
-                                    yield Err(BoxedError::new(e)).context(ExternalSnafu);
-                                    break;
-                                }
-                            }
-                        } else {
-                            // the stream has ended
-                            stream_ended = true;
-                        }
-
-                        yield result_to_yield;
+                        )
                     }
                     FlightMessage::Metrics(s) => {
-                        // just a branch in case of some metrics message comes after other things.
                         let m = serde_json::from_str(&s).ok().map(Arc::new);
                         metrics_ref.swap(m);
                         break;

src/cmd/Cargo.toml

@@ -52,6 +52,7 @@ 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
@@ -66,6 +67,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::{verbose_version, version};
+use common_version::version;
 use servers::install_ring_crypto_provider;
 
 #[derive(Parser)]
-#[command(name = "greptime", author, version, long_version = verbose_version(), about)]
+#[command(name = "greptime", author, version, long_version = version(), about)]
 #[command(propagate_version = true)]
 pub(crate) struct Command {
     #[clap(subcommand)]
@@ -143,8 +143,10 @@ async fn start(cli: Command) -> Result<()> {
 }
 
 fn setup_human_panic() {
-    human_panic::setup_panic!(human_panic::Metadata::new("GreptimeDB", version())
-        .homepage("https://github.com/GreptimeTeam/greptimedb/discussions"));
+    human_panic::setup_panic!(
+        human_panic::Metadata::new("GreptimeDB", env!("CARGO_PKG_VERSION"))
+            .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, verbose_version};
+use common_version::{short_version, version};
 use datanode::datanode::DatanodeBuilder;
 use datanode::service::DatanodeServiceBuilder;
 use meta_client::MetaClientType;
@@ -67,7 +67,7 @@ impl InstanceBuilder {
             None,
         );
 
-        log_versions(verbose_version(), short_version(), APP_NAME);
+        log_versions(version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         plugins::setup_datanode_plugins(plugins, &opts.plugins, dn_opts)

src/cmd/src/flownode.rs

@@ -32,7 +32,7 @@ use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::TableMetadataManager;
 use common_telemetry::info;
 use common_telemetry::logging::{TracingOptions, DEFAULT_LOGGING_DIR};
-use common_version::{short_version, verbose_version};
+use common_version::{short_version, version};
 use flow::{
     get_flow_auth_options, FlownodeBuilder, FlownodeInstance, FlownodeServiceBuilder,
     FrontendClient, FrontendInvoker,
@@ -279,7 +279,7 @@ impl StartCommand {
             None,
         );
 
-        log_versions(verbose_version(), short_version(), APP_NAME);
+        log_versions(version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Flownode start command: {:#?}", self);

src/cmd/src/frontend.rs

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

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(&[common_version::version(), short_version, app])
+        .with_label_values(&[env!("CARGO_PKG_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, verbose_version};
+use common_version::{short_version, version};
 use meta_srv::bootstrap::MetasrvInstance;
 use meta_srv::metasrv::BackendImpl;
 use snafu::ResultExt;
@@ -54,10 +54,6 @@ impl Instance {
     pub fn get_inner(&self) -> &MetasrvInstance {
         &self.instance
     }
-
-    pub fn mut_inner(&mut self) -> &mut MetasrvInstance {
-        &mut self.instance
-    }
 }
 
 #[async_trait]
@@ -324,7 +320,7 @@ impl StartCommand {
             None,
         );
 
-        log_versions(verbose_version(), short_version(), APP_NAME);
+        log_versions(version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Metasrv start command: {:#?}", self);
@@ -345,7 +341,7 @@ impl StartCommand {
             .context(error::BuildMetaServerSnafu)?;
         let metasrv = builder.build().await.context(error::BuildMetaServerSnafu)?;
 
-        let instance = MetasrvInstance::new(metasrv)
+        let instance = MetasrvInstance::new(opts, plugins, metasrv)
             .await
             .context(error::BuildMetaServerSnafu)?;
 

src/cmd/src/standalone.rs

@@ -34,14 +34,13 @@ use common_meta::cluster::{NodeInfo, NodeStatus};
 use common_meta::datanode::RegionStat;
 use common_meta::ddl::flow_meta::FlowMetadataAllocator;
 use common_meta::ddl::table_meta::TableMetadataAllocator;
-use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl};
+use common_meta::ddl::{DdlContext, NoopRegionFailureDetectorControl, ProcedureExecutorRef};
 use common_meta::ddl_manager::DdlManager;
 use common_meta::key::flow::flow_state::FlowStat;
 use common_meta::key::flow::FlowMetadataManager;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::KvBackendRef;
 use common_meta::peer::Peer;
-use common_meta::procedure_executor::LocalProcedureExecutor;
 use common_meta::region_keeper::MemoryRegionKeeper;
 use common_meta::region_registry::LeaderRegionRegistry;
 use common_meta::sequence::SequenceBuilder;
@@ -52,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, verbose_version};
+use common_version::{short_version, version};
 use common_wal::config::DatanodeWalConfig;
 use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, StorageConfig};
 use datanode::datanode::{Datanode, DatanodeBuilder};
@@ -258,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]
@@ -467,7 +485,7 @@ impl StartCommand {
             opts.component.slow_query.as_ref(),
         );
 
-        log_versions(verbose_version(), short_version(), APP_NAME);
+        log_versions(version(), short_version(), APP_NAME);
         create_resource_limit_metrics(APP_NAME);
 
         info!("Standalone start command: {:#?}", self);
@@ -547,13 +565,14 @@ impl StartCommand {
         // actually make a connection
         let (frontend_client, frontend_instance_handler) =
             FrontendClient::from_empty_grpc_handler();
+        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()
@@ -610,8 +629,9 @@ impl StartCommand {
             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.clone(), true)
+        let ddl_manager = DdlManager::try_new(ddl_context, procedure_manager_c, true)
             .context(error::InitDdlManagerSnafu)?;
         #[cfg(feature = "enterprise")]
         let ddl_manager = {
@@ -619,11 +639,7 @@ impl StartCommand {
                 plugins.get();
             ddl_manager.with_trigger_ddl_manager(trigger_ddl_manager)
         };
-
-        let procedure_executor = Arc::new(LocalProcedureExecutor::new(
-            Arc::new(ddl_manager),
-            procedure_manager.clone(),
-        ));
+        let ddl_task_executor: ProcedureExecutorRef = Arc::new(ddl_manager);
 
         let fe_instance = FrontendBuilder::new(
             fe_opts.clone(),
@@ -631,7 +647,7 @@ impl StartCommand {
             layered_cache_registry.clone(),
             catalog_manager.clone(),
             node_manager.clone(),
-            procedure_executor.clone(),
+            ddl_task_executor.clone(),
             process_manager,
         )
         .with_plugin(plugins.clone())
@@ -656,7 +672,7 @@ impl StartCommand {
             catalog_manager.clone(),
             kv_backend.clone(),
             layered_cache_registry.clone(),
-            procedure_executor,
+            ddl_task_executor.clone(),
             node_manager,
         )
         .await
@@ -666,7 +682,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)?;
 
@@ -677,12 +693,22 @@ 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,
         })
     }
@@ -792,7 +818,6 @@ impl InformationExtension for StandaloneInformationExtension {
                     region_manifest: region_stat.manifest.into(),
                     data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
                     metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
-                    written_bytes: region_stat.written_bytes,
                 }
             })
             .collect::<Vec<_>>();
@@ -823,7 +848,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;
@@ -942,15 +967,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,7 +18,7 @@ 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};
@@ -81,7 +81,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 +124,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 +172,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()
             },
@@ -229,7 +229,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/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/datasource/src/lib.rs

@@ -21,6 +21,7 @@ pub mod error;
 pub mod file_format;
 pub mod lister;
 pub mod object_store;
+pub mod parquet_writer;
 pub mod share_buffer;
 #[cfg(test)]
 pub mod test_util;

src/common/datasource/src/parquet_writer.rs

@@ -0,0 +1,52 @@
+// 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 bytes::Bytes;
+use futures::future::BoxFuture;
+use object_store::Writer;
+use parquet::arrow::async_writer::AsyncFileWriter;
+use parquet::errors::ParquetError;
+
+/// Bridges opendal [Writer] with parquet [AsyncFileWriter].
+pub struct AsyncWriter {
+    inner: Writer,
+}
+
+impl AsyncWriter {
+    /// Create a [`AsyncWriter`] by given [`Writer`].
+    pub fn new(writer: Writer) -> Self {
+        Self { inner: writer }
+    }
+}
+
+impl AsyncFileWriter for AsyncWriter {
+    fn write(&mut self, bs: Bytes) -> BoxFuture<'_, parquet::errors::Result<()>> {
+        Box::pin(async move {
+            self.inner
+                .write(bs)
+                .await
+                .map_err(|err| ParquetError::External(Box::new(err)))
+        })
+    }
+
+    fn complete(&mut self) -> BoxFuture<'_, parquet::errors::Result<()>> {
+        Box::pin(async move {
+            self.inner
+                .close()
+                .await
+                .map(|_| ())
+                .map_err(|err| ParquetError::External(Box::new(err)))
+        })
+    }
+}

src/common/frontend/Cargo.toml

@@ -12,7 +12,6 @@ common-macro.workspace = true
 common-meta.workspace = true
 greptime-proto.workspace = true
 meta-client.workspace = true
-session.workspace = true
 snafu.workspace = true
 tonic.workspace = true
 

src/common/frontend/src/lib.rs

@@ -19,7 +19,6 @@ use snafu::OptionExt;
 
 pub mod error;
 pub mod selector;
-pub mod slow_query_event;
 
 #[derive(Debug, Clone, Eq, PartialEq)]
 pub struct DisplayProcessId {

src/common/frontend/src/selector.rs

@@ -12,7 +12,6 @@
 // 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};
@@ -31,7 +30,7 @@ use crate::error::{MetaSnafu, Result};
 pub type FrontendClientPtr = Box<dyn FrontendClient>;
 
 #[async_trait::async_trait]
-pub trait FrontendClient: Send + Debug {
+pub trait FrontendClient: Send {
     async fn list_process(&mut self, req: ListProcessRequest) -> Result<ListProcessResponse>;
 
     async fn kill_process(&mut self, req: KillProcessRequest) -> Result<KillProcessResponse>;

src/common/frontend/src/slow_query_event.rs

@@ -1,28 +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 session::context::QueryContextRef;
-
-#[derive(Debug)]
-pub struct SlowQueryEvent {
-    pub cost: u64,
-    pub threshold: u64,
-    pub query: String,
-    pub is_promql: bool,
-    pub query_ctx: QueryContextRef,
-    pub promql_range: Option<u64>,
-    pub promql_step: Option<u64>,
-    pub promql_start: Option<i64>,
-    pub promql_end: Option<i64>,
-}

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/admin.rs

@@ -16,9 +16,6 @@ mod add_region_follower;
 mod flush_compact_region;
 mod flush_compact_table;
 mod migrate_region;
-mod reconcile_catalog;
-mod reconcile_database;
-mod reconcile_table;
 mod remove_region_follower;
 
 use std::sync::Arc;
@@ -27,9 +24,6 @@ use add_region_follower::AddRegionFollowerFunction;
 use flush_compact_region::{CompactRegionFunction, FlushRegionFunction};
 use flush_compact_table::{CompactTableFunction, FlushTableFunction};
 use migrate_region::MigrateRegionFunction;
-use reconcile_catalog::ReconcileCatalogFunction;
-use reconcile_database::ReconcileDatabaseFunction;
-use reconcile_table::ReconcileTableFunction;
 use remove_region_follower::RemoveRegionFollowerFunction;
 
 use crate::flush_flow::FlushFlowFunction;
@@ -49,8 +43,5 @@ impl AdminFunction {
         registry.register_async(Arc::new(FlushTableFunction));
         registry.register_async(Arc::new(CompactTableFunction));
         registry.register_async(Arc::new(FlushFlowFunction));
-        registry.register_async(Arc::new(ReconcileCatalogFunction));
-        registry.register_async(Arc::new(ReconcileDatabaseFunction));
-        registry.register_async(Arc::new(ReconcileTableFunction));
     }
 }

src/common/function/src/admin/reconcile_catalog.rs

@@ -1,179 +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::meta::reconcile_request::Target;
-use api::v1::meta::{ReconcileCatalog, ReconcileRequest};
-use common_macro::admin_fn;
-use common_query::error::{
-    InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, Result,
-    UnsupportedInputDataTypeSnafu,
-};
-use common_query::prelude::{Signature, TypeSignature, Volatility};
-use common_telemetry::info;
-use datatypes::prelude::*;
-use session::context::QueryContextRef;
-
-use crate::handlers::ProcedureServiceHandlerRef;
-use crate::helper::{
-    cast_u32, default_parallelism, default_resolve_strategy, get_string_from_params,
-    parse_resolve_strategy,
-};
-
-const FN_NAME: &str = "reconcile_catalog";
-
-/// A function to reconcile a catalog.
-/// Returns the procedure id if success.
-///
-/// - `reconcile_catalog(resolve_strategy)`.
-/// - `reconcile_catalog(resolve_strategy, parallelism)`.
-///
-/// - `reconcile_catalog()`.
-#[admin_fn(
-    name = ReconcileCatalogFunction,
-    display_name = reconcile_catalog,
-    sig_fn = signature,
-    ret = string
-)]
-pub(crate) async fn reconcile_catalog(
-    procedure_service_handler: &ProcedureServiceHandlerRef,
-    query_ctx: &QueryContextRef,
-    params: &[ValueRef<'_>],
-) -> Result<Value> {
-    let (resolve_strategy, parallelism) = match params.len() {
-        0 => (default_resolve_strategy(), default_parallelism()),
-        1 => (
-            parse_resolve_strategy(get_string_from_params(params, 0, FN_NAME)?)?,
-            default_parallelism(),
-        ),
-        2 => {
-            let Some(parallelism) = cast_u32(&params[1])? else {
-                return UnsupportedInputDataTypeSnafu {
-                    function: FN_NAME,
-                    datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-                }
-                .fail();
-            };
-            (
-                parse_resolve_strategy(get_string_from_params(params, 0, FN_NAME)?)?,
-                parallelism,
-            )
-        }
-        size => {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "The length of the args is not correct, expect 0, 1 or 2, have: {}",
-                    size
-                ),
-            }
-            .fail();
-        }
-    };
-    info!(
-        "Reconciling catalog with resolve_strategy: {:?}, parallelism: {}",
-        resolve_strategy, parallelism
-    );
-    let pid = procedure_service_handler
-        .reconcile(ReconcileRequest {
-            target: Some(Target::ReconcileCatalog(ReconcileCatalog {
-                catalog_name: query_ctx.current_catalog().to_string(),
-                parallelism,
-                resolve_strategy: resolve_strategy as i32,
-            })),
-            ..Default::default()
-        })
-        .await?;
-    match pid {
-        Some(pid) => Ok(Value::from(pid)),
-        None => Ok(Value::Null),
-    }
-}
-
-fn signature() -> Signature {
-    let nums = ConcreteDataType::numerics();
-    let mut signs = Vec::with_capacity(2 + nums.len());
-    signs.extend([
-        // reconcile_catalog()
-        TypeSignature::NullAry,
-        // reconcile_catalog(resolve_strategy)
-        TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-    ]);
-    for sign in nums {
-        // reconcile_catalog(resolve_strategy, parallelism)
-        signs.push(TypeSignature::Exact(vec![
-            ConcreteDataType::string_datatype(),
-            sign,
-        ]));
-    }
-    Signature::one_of(signs, Volatility::Immutable)
-}
-
-#[cfg(test)]
-mod tests {
-    use std::assert_matches::assert_matches;
-    use std::sync::Arc;
-
-    use common_query::error::Error;
-    use datatypes::vectors::{StringVector, UInt64Vector, VectorRef};
-
-    use crate::admin::reconcile_catalog::ReconcileCatalogFunction;
-    use crate::function::{AsyncFunction, FunctionContext};
-
-    #[tokio::test]
-    async fn test_reconcile_catalog() {
-        common_telemetry::init_default_ut_logging();
-
-        // reconcile_catalog()
-        let f = ReconcileCatalogFunction;
-        let args = vec![];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_catalog(resolve_strategy)
-        let f = ReconcileCatalogFunction;
-        let args = vec![Arc::new(StringVector::from(vec!["UseMetasrv"])) as _];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_catalog(resolve_strategy, parallelism)
-        let f = ReconcileCatalogFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt64Vector::from_slice([10])) as _,
-        ];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // unsupported input data type
-        let f = ReconcileCatalogFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(StringVector::from(vec!["test"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::UnsupportedInputDataType { .. });
-
-        // invalid function args
-        let f = ReconcileCatalogFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt64Vector::from_slice([10])) as _,
-            Arc::new(StringVector::from(vec!["10"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::InvalidFuncArgs { .. });
-    }
-}

src/common/function/src/admin/reconcile_database.rs

@@ -1,198 +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::meta::reconcile_request::Target;
-use api::v1::meta::{ReconcileDatabase, ReconcileRequest};
-use common_macro::admin_fn;
-use common_query::error::{
-    InvalidFuncArgsSnafu, MissingProcedureServiceHandlerSnafu, Result,
-    UnsupportedInputDataTypeSnafu,
-};
-use common_query::prelude::{Signature, TypeSignature, Volatility};
-use common_telemetry::info;
-use datatypes::prelude::*;
-use session::context::QueryContextRef;
-
-use crate::handlers::ProcedureServiceHandlerRef;
-use crate::helper::{
-    cast_u32, default_parallelism, default_resolve_strategy, get_string_from_params,
-    parse_resolve_strategy,
-};
-
-const FN_NAME: &str = "reconcile_database";
-
-/// A function to reconcile a database.
-/// Returns the procedure id if success.
-///
-/// - `reconcile_database(database_name)`.
-/// - `reconcile_database(database_name, resolve_strategy)`.
-/// - `reconcile_database(database_name, resolve_strategy, parallelism)`.
-///
-/// The parameters:
-/// - `database_name`:  the database name
-#[admin_fn(
-    name = ReconcileDatabaseFunction,
-    display_name = reconcile_database,
-    sig_fn = signature,
-    ret = string
-)]
-pub(crate) async fn reconcile_database(
-    procedure_service_handler: &ProcedureServiceHandlerRef,
-    query_ctx: &QueryContextRef,
-    params: &[ValueRef<'_>],
-) -> Result<Value> {
-    let (database_name, resolve_strategy, parallelism) = match params.len() {
-        1 => (
-            get_string_from_params(params, 0, FN_NAME)?,
-            default_resolve_strategy(),
-            default_parallelism(),
-        ),
-        2 => (
-            get_string_from_params(params, 0, FN_NAME)?,
-            parse_resolve_strategy(get_string_from_params(params, 1, FN_NAME)?)?,
-            default_parallelism(),
-        ),
-        3 => {
-            let Some(parallelism) = cast_u32(&params[2])? else {
-                return UnsupportedInputDataTypeSnafu {
-                    function: FN_NAME,
-                    datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-                }
-                .fail();
-            };
-            (
-                get_string_from_params(params, 0, FN_NAME)?,
-                parse_resolve_strategy(get_string_from_params(params, 1, FN_NAME)?)?,
-                parallelism,
-            )
-        }
-        size => {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "The length of the args is not correct, expect 1, 2 or 3, have: {}",
-                    size
-                ),
-            }
-            .fail();
-        }
-    };
-    info!(
-        "Reconciling database: {}, resolve_strategy: {:?}, parallelism: {}",
-        database_name, resolve_strategy, parallelism
-    );
-    let pid = procedure_service_handler
-        .reconcile(ReconcileRequest {
-            target: Some(Target::ReconcileDatabase(ReconcileDatabase {
-                catalog_name: query_ctx.current_catalog().to_string(),
-                database_name: database_name.to_string(),
-                parallelism,
-                resolve_strategy: resolve_strategy as i32,
-            })),
-            ..Default::default()
-        })
-        .await?;
-    match pid {
-        Some(pid) => Ok(Value::from(pid)),
-        None => Ok(Value::Null),
-    }
-}
-
-fn signature() -> Signature {
-    let nums = ConcreteDataType::numerics();
-    let mut signs = Vec::with_capacity(2 + nums.len());
-    signs.extend([
-        // reconcile_database(datanode_name)
-        TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-        // reconcile_database(database_name, resolve_strategy)
-        TypeSignature::Exact(vec![
-            ConcreteDataType::string_datatype(),
-            ConcreteDataType::string_datatype(),
-        ]),
-    ]);
-    for sign in nums {
-        // reconcile_database(database_name, resolve_strategy, parallelism)
-        signs.push(TypeSignature::Exact(vec![
-            ConcreteDataType::string_datatype(),
-            ConcreteDataType::string_datatype(),
-            sign,
-        ]));
-    }
-    Signature::one_of(signs, Volatility::Immutable)
-}
-
-#[cfg(test)]
-mod tests {
-    use std::assert_matches::assert_matches;
-    use std::sync::Arc;
-
-    use common_query::error::Error;
-    use datatypes::vectors::{StringVector, UInt32Vector, VectorRef};
-
-    use crate::admin::reconcile_database::ReconcileDatabaseFunction;
-    use crate::function::{AsyncFunction, FunctionContext};
-
-    #[tokio::test]
-    async fn test_reconcile_catalog() {
-        common_telemetry::init_default_ut_logging();
-
-        // reconcile_database(database_name)
-        let f = ReconcileDatabaseFunction;
-        let args = vec![Arc::new(StringVector::from(vec!["test"])) as _];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_database(database_name, resolve_strategy)
-        let f = ReconcileDatabaseFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["test"])) as _,
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-        ];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_database(database_name, resolve_strategy, parallelism)
-        let f = ReconcileDatabaseFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["test"])) as _,
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt32Vector::from_slice([10])) as _,
-        ];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // invalid function args
-        let f = ReconcileDatabaseFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt32Vector::from_slice([10])) as _,
-            Arc::new(StringVector::from(vec!["v1"])) as _,
-            Arc::new(StringVector::from(vec!["v2"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::InvalidFuncArgs { .. });
-
-        // unsupported input data type
-        let f = ReconcileDatabaseFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["UseLatest"])) as _,
-            Arc::new(UInt32Vector::from_slice([10])) as _,
-            Arc::new(StringVector::from(vec!["v1"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::UnsupportedInputDataType { .. });
-    }
-}

src/common/function/src/admin/reconcile_table.rs

@@ -1,149 +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::meta::reconcile_request::Target;
-use api::v1::meta::{ReconcileRequest, ReconcileTable, ResolveStrategy};
-use common_catalog::format_full_table_name;
-use common_error::ext::BoxedError;
-use common_macro::admin_fn;
-use common_query::error::{
-    MissingProcedureServiceHandlerSnafu, Result, TableMutationSnafu, UnsupportedInputDataTypeSnafu,
-};
-use common_query::prelude::{Signature, TypeSignature, Volatility};
-use common_telemetry::info;
-use datatypes::prelude::*;
-use session::context::QueryContextRef;
-use session::table_name::table_name_to_full_name;
-use snafu::ResultExt;
-
-use crate::handlers::ProcedureServiceHandlerRef;
-use crate::helper::parse_resolve_strategy;
-
-const FN_NAME: &str = "reconcile_table";
-
-/// A function to reconcile a table.
-/// Returns the procedure id if success.
-///
-/// - `reconcile_table(table_name)`.
-/// - `reconcile_table(table_name, resolve_strategy)`.
-///
-/// The parameters:
-/// - `table_name`:  the table name
-#[admin_fn(
-    name = ReconcileTableFunction,
-    display_name = reconcile_table,
-    sig_fn = signature,
-    ret = string
-)]
-pub(crate) async fn reconcile_table(
-    procedure_service_handler: &ProcedureServiceHandlerRef,
-    query_ctx: &QueryContextRef,
-    params: &[ValueRef<'_>],
-) -> Result<Value> {
-    let (table_name, resolve_strategy) = match params {
-        [ValueRef::String(table_name)] => (table_name, ResolveStrategy::UseLatest),
-        [ValueRef::String(table_name), ValueRef::String(resolve_strategy)] => {
-            (table_name, parse_resolve_strategy(resolve_strategy)?)
-        }
-        _ => {
-            return UnsupportedInputDataTypeSnafu {
-                function: FN_NAME,
-                datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-            }
-            .fail()
-        }
-    };
-    let (catalog_name, schema_name, table_name) = table_name_to_full_name(table_name, query_ctx)
-        .map_err(BoxedError::new)
-        .context(TableMutationSnafu)?;
-    info!(
-        "Reconciling table: {} with resolve_strategy: {:?}",
-        format_full_table_name(&catalog_name, &schema_name, &table_name),
-        resolve_strategy
-    );
-    let pid = procedure_service_handler
-        .reconcile(ReconcileRequest {
-            target: Some(Target::ReconcileTable(ReconcileTable {
-                catalog_name,
-                schema_name,
-                table_name,
-                resolve_strategy: resolve_strategy as i32,
-            })),
-            ..Default::default()
-        })
-        .await?;
-    match pid {
-        Some(pid) => Ok(Value::from(pid)),
-        None => Ok(Value::Null),
-    }
-}
-
-fn signature() -> Signature {
-    Signature::one_of(
-        vec![
-            // reconcile_table(table_name)
-            TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-            // reconcile_table(table_name, resolve_strategy)
-            TypeSignature::Exact(vec![
-                ConcreteDataType::string_datatype(),
-                ConcreteDataType::string_datatype(),
-            ]),
-        ],
-        Volatility::Immutable,
-    )
-}
-
-#[cfg(test)]
-mod tests {
-    use std::assert_matches::assert_matches;
-    use std::sync::Arc;
-
-    use common_query::error::Error;
-    use datatypes::vectors::{StringVector, VectorRef};
-
-    use crate::admin::reconcile_table::ReconcileTableFunction;
-    use crate::function::{AsyncFunction, FunctionContext};
-
-    #[tokio::test]
-    async fn test_reconcile_table() {
-        common_telemetry::init_default_ut_logging();
-
-        // reconcile_table(table_name)
-        let f = ReconcileTableFunction;
-        let args = vec![Arc::new(StringVector::from(vec!["test"])) as _];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // reconcile_table(table_name, resolve_strategy)
-        let f = ReconcileTableFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["test"])) as _,
-            Arc::new(StringVector::from(vec!["UseMetasrv"])) as _,
-        ];
-        let result = f.eval(FunctionContext::mock(), &args).await.unwrap();
-        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_pid"]));
-        assert_eq!(expect, result);
-
-        // unsupported input data type
-        let f = ReconcileTableFunction;
-        let args = vec![
-            Arc::new(StringVector::from(vec!["test"])) as _,
-            Arc::new(StringVector::from(vec!["UseMetasrv"])) as _,
-            Arc::new(StringVector::from(vec!["10"])) as _,
-        ];
-        let err = f.eval(FunctionContext::mock(), &args).await.unwrap_err();
-        assert_matches!(err, Error::UnsupportedInputDataType { .. });
-    }
-}

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/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/handlers.rs

@@ -14,7 +14,6 @@
 
 use std::sync::Arc;
 
-use api::v1::meta::ReconcileRequest;
 use async_trait::async_trait;
 use catalog::CatalogManagerRef;
 use common_base::AffectedRows;
@@ -66,9 +65,6 @@ pub trait ProcedureServiceHandler: Send + Sync {
     /// Migrate a region from source peer to target peer, returns the procedure id if success.
     async fn migrate_region(&self, request: MigrateRegionRequest) -> Result<Option<String>>;
 
-    /// Reconcile a table, database or catalog, returns the procedure id if success.
-    async fn reconcile(&self, request: ReconcileRequest) -> Result<Option<String>>;
-
     /// Query the procedure' state by its id
     async fn query_procedure_state(&self, pid: &str) -> Result<ProcedureStateResponse>;
 

src/common/function/src/helper.rs

@@ -12,15 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use api::v1::meta::ResolveStrategy;
-use common_query::error::{
-    InvalidFuncArgsSnafu, InvalidInputTypeSnafu, Result, UnsupportedInputDataTypeSnafu,
-};
+use common_query::error::{InvalidInputTypeSnafu, Result};
 use common_query::prelude::{Signature, TypeSignature, Volatility};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::types::cast::cast;
 use datatypes::value::ValueRef;
-use snafu::{OptionExt, ResultExt};
+use snafu::ResultExt;
 
 /// Create a function signature with oneof signatures of interleaving two arguments.
 pub fn one_of_sigs2(args1: Vec<ConcreteDataType>, args2: Vec<ConcreteDataType>) -> Signature {
@@ -46,64 +43,3 @@ pub fn cast_u64(value: &ValueRef) -> Result<Option<u64>> {
         })
         .map(|v| v.as_u64())
 }
-
-/// Cast a [`ValueRef`] to u32, returns `None` if fails
-pub fn cast_u32(value: &ValueRef) -> Result<Option<u32>> {
-    cast((*value).into(), &ConcreteDataType::uint32_datatype())
-        .context(InvalidInputTypeSnafu {
-            err_msg: format!(
-                "Failed to cast input into uint32, actual type: {:#?}",
-                value.data_type(),
-            ),
-        })
-        .map(|v| v.as_u64().map(|v| v as u32))
-}
-
-/// Parse a resolve strategy from a string.
-pub fn parse_resolve_strategy(strategy: &str) -> Result<ResolveStrategy> {
-    ResolveStrategy::from_str_name(strategy).context(InvalidFuncArgsSnafu {
-        err_msg: format!("Invalid resolve strategy: {}", strategy),
-    })
-}
-
-/// Default parallelism for reconcile operations.
-pub fn default_parallelism() -> u32 {
-    64
-}
-
-/// Default resolve strategy for reconcile operations.
-pub fn default_resolve_strategy() -> ResolveStrategy {
-    ResolveStrategy::UseLatest
-}
-
-/// Get the string value from the params.
-///
-/// # Errors
-/// Returns an error if the input type is not a string.
-pub fn get_string_from_params<'a>(
-    params: &'a [ValueRef<'a>],
-    index: usize,
-    fn_name: &'a str,
-) -> Result<&'a str> {
-    let ValueRef::String(s) = &params[index] else {
-        return UnsupportedInputDataTypeSnafu {
-            function: fn_name,
-            datatypes: params.iter().map(|v| v.data_type()).collect::<Vec<_>>(),
-        }
-        .fail();
-    };
-    Ok(s)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_parse_resolve_strategy() {
-        assert_eq!(
-            parse_resolve_strategy("UseLatest").unwrap(),
-            ResolveStrategy::UseLatest
-        );
-    }
-}

src/common/function/src/lib.rs

@@ -14,7 +14,6 @@
 
 #![feature(let_chains)]
 #![feature(try_blocks)]
-#![feature(assert_matches)]
 
 mod admin;
 mod flush_flow;

src/common/function/src/state.rs

@@ -32,7 +32,7 @@ impl FunctionState {
     pub fn mock() -> Self {
         use std::sync::Arc;
 
-        use api::v1::meta::{ProcedureStatus, ReconcileRequest};
+        use api::v1::meta::ProcedureStatus;
         use async_trait::async_trait;
         use catalog::CatalogManagerRef;
         use common_base::AffectedRows;
@@ -63,10 +63,6 @@ impl FunctionState {
                 Ok(Some("test_pid".to_string()))
             }
 
-            async fn reconcile(&self, _request: ReconcileRequest) -> Result<Option<String>> {
-                Ok(Some("test_pid".to_string()))
-            }
-
             async fn query_procedure_state(&self, _pid: &str) -> Result<ProcedureStateResponse> {
                 Ok(ProcedureStateResponse {
                     status: ProcedureStatus::Done.into(),

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 {}",
-            common_version::version()
+            env!("CARGO_PKG_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()),
-                    common_version::version()
+                    env!("CARGO_PKG_VERSION")
                 )
             }
             Channel::Postgres => {
-                format!("16.3-greptimedb-{}", common_version::version())
+                format!("16.3-greptimedb-{}", env!("CARGO_PKG_VERSION"))
             }
-            _ => common_version::version().to_string(),
+            _ => env!("CARGO_PKG_VERSION").to_string(),
         };
         let result = StringVector::from(vec![version]);
         Ok(Arc::new(result))

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

@@ -29,12 +29,12 @@ use snafu::{ensure, OptionExt, ResultExt};
 use store_api::region_request::{SetRegionOption, UnsetRegionOption};
 use table::metadata::TableId;
 use table::requests::{
-    AddColumnRequest, AlterKind, AlterTableRequest, ModifyColumnTypeRequest, SetIndexOption,
-    UnsetIndexOption,
+    AddColumnRequest, AlterKind, AlterTableRequest, ModifyColumnTypeRequest, SetIndexOptions,
+    UnsetIndexOptions,
 };
 
 use crate::error::{
-    InvalidColumnDefSnafu, InvalidIndexOptionSnafu, InvalidSetFulltextOptionRequestSnafu,
+    InvalidColumnDefSnafu, InvalidSetFulltextOptionRequestSnafu,
     InvalidSetSkippingIndexOptionRequestSnafu, InvalidSetTableOptionRequestSnafu,
     InvalidUnsetTableOptionRequestSnafu, MissingAlterIndexOptionSnafu, MissingFieldSnafu,
     MissingTimestampColumnSnafu, Result, UnknownLocationTypeSnafu,
@@ -43,59 +43,6 @@ use crate::error::{
 const LOCATION_TYPE_FIRST: i32 = LocationType::First as i32;
 const LOCATION_TYPE_AFTER: i32 = LocationType::After as i32;
 
-fn set_index_option_from_proto(set_index: api::v1::SetIndex) -> Result<SetIndexOption> {
-    let options = set_index.options.context(MissingAlterIndexOptionSnafu)?;
-    Ok(match options {
-        api::v1::set_index::Options::Fulltext(f) => SetIndexOption::Fulltext {
-            column_name: f.column_name.clone(),
-            options: FulltextOptions::new(
-                f.enable,
-                as_fulltext_option_analyzer(
-                    Analyzer::try_from(f.analyzer).context(InvalidSetFulltextOptionRequestSnafu)?,
-                ),
-                f.case_sensitive,
-                as_fulltext_option_backend(
-                    PbFulltextBackend::try_from(f.backend)
-                        .context(InvalidSetFulltextOptionRequestSnafu)?,
-                ),
-                f.granularity as u32,
-                f.false_positive_rate,
-            )
-            .context(InvalidIndexOptionSnafu)?,
-        },
-        api::v1::set_index::Options::Inverted(i) => SetIndexOption::Inverted {
-            column_name: i.column_name,
-        },
-        api::v1::set_index::Options::Skipping(s) => SetIndexOption::Skipping {
-            column_name: s.column_name,
-            options: SkippingIndexOptions::new(
-                s.granularity as u32,
-                s.false_positive_rate,
-                as_skipping_index_type(
-                    PbSkippingIndexType::try_from(s.skipping_index_type)
-                        .context(InvalidSetSkippingIndexOptionRequestSnafu)?,
-                ),
-            )
-            .context(InvalidIndexOptionSnafu)?,
-        },
-    })
-}
-
-fn unset_index_option_from_proto(unset_index: api::v1::UnsetIndex) -> Result<UnsetIndexOption> {
-    let options = unset_index.options.context(MissingAlterIndexOptionSnafu)?;
-    Ok(match options {
-        api::v1::unset_index::Options::Fulltext(f) => UnsetIndexOption::Fulltext {
-            column_name: f.column_name,
-        },
-        api::v1::unset_index::Options::Inverted(i) => UnsetIndexOption::Inverted {
-            column_name: i.column_name,
-        },
-        api::v1::unset_index::Options::Skipping(s) => UnsetIndexOption::Skipping {
-            column_name: s.column_name,
-        },
-    })
-}
-
 /// Convert an [`AlterTableExpr`] to an [`AlterTableRequest`]
 pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<AlterTableRequest> {
     let catalog_name = expr.catalog_name;
@@ -174,34 +121,65 @@ pub fn alter_expr_to_request(table_id: TableId, expr: AlterTableExpr) -> Result<
                     .context(InvalidUnsetTableOptionRequestSnafu)?,
             }
         }
-        Kind::SetIndex(o) => {
-            let option = set_index_option_from_proto(o)?;
-            AlterKind::SetIndexes {
-                options: vec![option],
-            }
-        }
-        Kind::UnsetIndex(o) => {
-            let option = unset_index_option_from_proto(o)?;
-            AlterKind::UnsetIndexes {
-                options: vec![option],
-            }
-        }
-        Kind::SetIndexes(o) => {
-            let options = o
-                .set_indexes
-                .into_iter()
-                .map(set_index_option_from_proto)
-                .collect::<Result<Vec<_>>>()?;
-            AlterKind::SetIndexes { options }
-        }
-        Kind::UnsetIndexes(o) => {
-            let options = o
-                .unset_indexes
-                .into_iter()
-                .map(unset_index_option_from_proto)
-                .collect::<Result<Vec<_>>>()?;
-            AlterKind::UnsetIndexes { options }
-        }
+        Kind::SetIndex(o) => match o.options {
+            Some(opt) => match opt {
+                api::v1::set_index::Options::Fulltext(f) => AlterKind::SetIndex {
+                    options: SetIndexOptions::Fulltext {
+                        column_name: f.column_name.clone(),
+                        options: FulltextOptions {
+                            enable: f.enable,
+                            analyzer: as_fulltext_option_analyzer(
+                                Analyzer::try_from(f.analyzer)
+                                    .context(InvalidSetFulltextOptionRequestSnafu)?,
+                            ),
+                            case_sensitive: f.case_sensitive,
+                            backend: as_fulltext_option_backend(
+                                PbFulltextBackend::try_from(f.backend)
+                                    .context(InvalidSetFulltextOptionRequestSnafu)?,
+                            ),
+                        },
+                    },
+                },
+                api::v1::set_index::Options::Inverted(i) => AlterKind::SetIndex {
+                    options: SetIndexOptions::Inverted {
+                        column_name: i.column_name,
+                    },
+                },
+                api::v1::set_index::Options::Skipping(s) => AlterKind::SetIndex {
+                    options: SetIndexOptions::Skipping {
+                        column_name: s.column_name,
+                        options: SkippingIndexOptions {
+                            granularity: s.granularity as u32,
+                            index_type: as_skipping_index_type(
+                                PbSkippingIndexType::try_from(s.skipping_index_type)
+                                    .context(InvalidSetSkippingIndexOptionRequestSnafu)?,
+                            ),
+                        },
+                    },
+                },
+            },
+            None => return MissingAlterIndexOptionSnafu.fail(),
+        },
+        Kind::UnsetIndex(o) => match o.options {
+            Some(opt) => match opt {
+                api::v1::unset_index::Options::Fulltext(f) => AlterKind::UnsetIndex {
+                    options: UnsetIndexOptions::Fulltext {
+                        column_name: f.column_name,
+                    },
+                },
+                api::v1::unset_index::Options::Inverted(i) => AlterKind::UnsetIndex {
+                    options: UnsetIndexOptions::Inverted {
+                        column_name: i.column_name,
+                    },
+                },
+                api::v1::unset_index::Options::Skipping(s) => AlterKind::UnsetIndex {
+                    options: UnsetIndexOptions::Skipping {
+                        column_name: s.column_name,
+                    },
+                },
+            },
+            None => return MissingAlterIndexOptionSnafu.fail(),
+        },
         Kind::DropDefaults(o) => {
             let names = o
                 .drop_defaults

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

@@ -153,14 +153,6 @@ 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>;
@@ -188,8 +180,7 @@ impl ErrorExt for Error {
             | Error::InvalidUnsetTableOptionRequest { .. }
             | Error::InvalidSetFulltextOptionRequest { .. }
             | Error::InvalidSetSkippingIndexOptionRequest { .. }
-            | Error::MissingAlterIndexOption { .. }
-            | Error::InvalidIndexOption { .. } => StatusCode::InvalidArguments,
+            | Error::MissingAlterIndexOption { .. } => StatusCode::InvalidArguments,
         }
     }
 

src/common/meta/Cargo.toml

@@ -32,7 +32,6 @@ common-procedure.workspace = true
 common-procedure-test.workspace = true
 common-query.workspace = true
 common-recordbatch.workspace = true
-common-runtime.workspace = true
 common-telemetry.workspace = true
 common-time.workspace = true
 common-wal.workspace = true

src/common/meta/src/cache/flow/table_flownode.rs

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

src/common/meta/src/cache_invalidator.rs

@@ -22,7 +22,6 @@ use crate::key::flow::flow_name::FlowNameKey;
 use crate::key::flow::flow_route::FlowRouteKey;
 use crate::key::flow::flownode_flow::FlownodeFlowKey;
 use crate::key::flow::table_flow::TableFlowKey;
-use crate::key::node_address::NodeAddressKey;
 use crate::key::schema_name::SchemaNameKey;
 use crate::key::table_info::TableInfoKey;
 use crate::key::table_name::TableNameKey;
@@ -54,10 +53,6 @@ pub struct Context {
 #[async_trait::async_trait]
 pub trait CacheInvalidator: Send + Sync {
     async fn invalidate(&self, ctx: &Context, caches: &[CacheIdent]) -> Result<()>;
-
-    fn name(&self) -> &'static str {
-        std::any::type_name::<Self>()
-    }
 }
 
 pub type CacheInvalidatorRef = Arc<dyn CacheInvalidator>;
@@ -142,13 +137,6 @@ where
                     let key = FlowInfoKey::new(*flow_id);
                     self.invalidate_key(&key.to_bytes()).await;
                 }
-                CacheIdent::FlowNodeAddressChange(node_id) => {
-                    // other caches doesn't need to be invalidated
-                    // since this is only for flownode address change not id change
-                    common_telemetry::info!("Invalidate flow node cache for node_id: {}", node_id);
-                    let key = NodeAddressKey::with_flownode(*node_id);
-                    self.invalidate_key(&key.to_bytes()).await;
-                }
             }
         }
         Ok(())

src/common/meta/src/datanode.rs

@@ -97,8 +97,6 @@ pub struct RegionStat {
     pub index_size: u64,
     /// The manifest infoof the region.
     pub region_manifest: RegionManifestInfo,
-    /// The total bytes written of the region since region opened.
-    pub written_bytes: u64,
     /// The latest entry id of topic used by data.
     /// **Only used by remote WAL prune.**
     pub data_topic_latest_entry_id: u64,
@@ -279,7 +277,6 @@ impl From<&api::v1::meta::RegionStat> for RegionStat {
             sst_size: region_stat.sst_size,
             index_size: region_stat.index_size,
             region_manifest: region_stat.manifest.into(),
-            written_bytes: region_stat.written_bytes,
             data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
             metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
         }

src/common/meta/src/ddl.rs

@@ -15,17 +15,25 @@
 use std::collections::HashMap;
 use std::sync::Arc;
 
+use api::v1::meta::ProcedureDetailResponse;
+use common_telemetry::tracing_context::W3cTrace;
 use store_api::storage::{RegionId, RegionNumber, TableId};
 
 use crate::cache_invalidator::CacheInvalidatorRef;
 use crate::ddl::flow_meta::FlowMetadataAllocatorRef;
 use crate::ddl::table_meta::TableMetadataAllocatorRef;
+use crate::error::{Result, UnsupportedSnafu};
 use crate::key::flow::FlowMetadataManagerRef;
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::key::TableMetadataManagerRef;
 use crate::node_manager::NodeManagerRef;
 use crate::region_keeper::MemoryRegionKeeperRef;
 use crate::region_registry::LeaderRegionRegistryRef;
+use crate::rpc::ddl::{SubmitDdlTaskRequest, SubmitDdlTaskResponse};
+use crate::rpc::procedure::{
+    AddRegionFollowerRequest, MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse,
+    RemoveRegionFollowerRequest,
+};
 use crate::DatanodeId;
 
 pub mod alter_database;
@@ -36,13 +44,13 @@ pub mod create_flow;
 pub mod create_logical_tables;
 pub mod create_table;
 mod create_table_template;
-pub(crate) use create_table_template::{build_template_from_raw_table_info, CreateRequestBuilder};
 pub mod create_view;
 pub mod drop_database;
 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;
@@ -51,6 +59,64 @@ pub(crate) mod tests;
 pub mod truncate_table;
 pub mod utils;
 
+#[derive(Debug, Default)]
+pub struct ExecutorContext {
+    pub tracing_context: Option<W3cTrace>,
+}
+
+/// The procedure executor that accepts ddl, region migration task etc.
+#[async_trait::async_trait]
+pub trait ProcedureExecutor: Send + Sync {
+    /// Submit a ddl task
+    async fn submit_ddl_task(
+        &self,
+        ctx: &ExecutorContext,
+        request: SubmitDdlTaskRequest,
+    ) -> Result<SubmitDdlTaskResponse>;
+
+    /// Add a region follower
+    async fn add_region_follower(
+        &self,
+        _ctx: &ExecutorContext,
+        _request: AddRegionFollowerRequest,
+    ) -> Result<()> {
+        UnsupportedSnafu {
+            operation: "add_region_follower",
+        }
+        .fail()
+    }
+
+    /// Remove a region follower
+    async fn remove_region_follower(
+        &self,
+        _ctx: &ExecutorContext,
+        _request: RemoveRegionFollowerRequest,
+    ) -> Result<()> {
+        UnsupportedSnafu {
+            operation: "remove_region_follower",
+        }
+        .fail()
+    }
+
+    /// Submit a region migration task
+    async fn migrate_region(
+        &self,
+        ctx: &ExecutorContext,
+        request: MigrateRegionRequest,
+    ) -> Result<MigrateRegionResponse>;
+
+    /// Query the procedure state by its id
+    async fn query_procedure_state(
+        &self,
+        ctx: &ExecutorContext,
+        pid: &str,
+    ) -> Result<ProcedureStateResponse>;
+
+    async fn list_procedures(&self, ctx: &ExecutorContext) -> Result<ProcedureDetailResponse>;
+}
+
+pub type ProcedureExecutorRef = Arc<dyn ProcedureExecutor>;
+
 /// Metadata allocated to a table.
 #[derive(Default)]
 pub struct TableMetadata {

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

@@ -12,32 +12,32 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod executor;
+mod check;
+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::{debug, error, info, warn};
-pub use executor::make_alter_region_request;
+use common_telemetry::{error, info, warn};
+use futures_util::future;
+pub use region_request::make_alter_region_request;
 use serde::{Deserialize, Serialize};
-use snafu::ResultExt;
+use snafu::{ensure, 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::cache_invalidator::Context as CacheContext;
-use crate::ddl::alter_logical_tables::executor::AlterLogicalTablesExecutor;
-use crate::ddl::alter_logical_tables::validator::{
-    retain_unskipped, AlterLogicalTableValidator, ValidatorResult,
+use crate::ddl::utils::{
+    add_peer_context_if_needed, map_to_procedure_error, sync_follower_regions,
 };
-use crate::ddl::utils::{extract_column_metadatas, map_to_procedure_error, sync_follower_regions};
 use crate::ddl::DdlContext;
-use crate::error::Result;
+use crate::error::{DecodeJsonSnafu, MetadataCorruptionSnafu, Result};
 use crate::instruction::CacheIdent;
 use crate::key::table_info::TableInfoValue;
 use crate::key::table_route::PhysicalTableRouteValue;
@@ -45,38 +45,13 @@ use crate::key::DeserializedValueWithBytes;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
 use crate::metrics;
 use crate::rpc::ddl::AlterTableTask;
-use crate::rpc::router::RegionRoute;
+use crate::rpc::router::{find_leaders, 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";
 
@@ -106,44 +81,35 @@ impl AlterLogicalTablesProcedure {
     }
 
     pub(crate) async fn on_prepare(&mut self) -> Result<Status> {
-        let validator = build_validator_from_alter_table_data(&self.data);
-        let ValidatorResult {
-            num_skipped,
-            skip_alter,
-            table_info_values,
-            physical_table_info,
-            physical_table_route,
-        } = validator
-            .validate(&self.context.table_metadata_manager)
-            .await?;
-
-        let num_tasks = self.data.tasks.len();
-        if num_skipped == num_tasks {
+        // Checks all the tasks
+        self.check_input_tasks()?;
+        // Fills the table info values
+        self.fill_table_info_values().await?;
+        // Checks the physical table, must after [fill_table_info_values]
+        self.check_physical_table().await?;
+        // Fills the physical table info
+        self.fill_physical_table_info().await?;
+        // Filter the finished tasks
+        let finished_tasks = self.check_finished_tasks()?;
+        let already_finished_count = finished_tasks
+            .iter()
+            .map(|x| if *x { 1 } else { 0 })
+            .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 num_skipped > 0 {
+        } else if already_finished_count > 0 {
             info!(
                 "There are {} alter tasks, {} of them were already finished.",
-                num_tasks, num_skipped
+                apply_tasks_count, already_finished_count
             );
         }
+        self.filter_task(&finished_tasks)?;
 
-        // 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());
+        // Next state
         self.data.state = AlterTablesState::SubmitAlterRegionRequests;
         Ok(Status::executing(true))
     }
@@ -151,21 +117,57 @@ 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 executor = build_executor_from_alter_expr(&self.data);
-        let mut results = executor
-            .on_alter_regions(
-                &self.context.node_manager,
-                &physical_table_route.region_routes,
-            )
-            .await?;
+        let leaders = find_leaders(&physical_table_route.region_routes);
+        let mut alter_region_tasks = Vec::with_capacity(leaders.len());
 
-        if let Some(column_metadatas) =
-            extract_column_metadatas(&mut results, ALTER_PHYSICAL_EXTENSION_KEY)?
-        {
-            self.data.physical_columns = column_metadatas;
+        for peer in leaders {
+            let requester = self.context.node_manager.datanode(&peer).await;
+            let request = self.make_request(&peer, &physical_table_route.region_routes)?;
+
+            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;
@@ -181,7 +183,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(),
         )
@@ -198,18 +200,7 @@ impl AlterLogicalTablesProcedure {
         self.update_physical_table_metadata().await?;
         self.update_logical_tables_metadata().await?;
 
-        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.build_cache_keys_to_invalidate();
         self.data.clear_metadata_fields();
 
         self.data.state = AlterTablesState::InvalidateTableCache;
@@ -219,16 +210,9 @@ 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(&ctx, to_invalidate)
+            .invalidate(&Default::default(), to_invalidate)
             .await?;
         Ok(Status::done())
     }
@@ -248,10 +232,6 @@ 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

@@ -0,0 +1,136 @@
+// 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

@@ -1,216 +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::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

@@ -0,0 +1,158 @@
+// 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

@@ -0,0 +1,113 @@
+// 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::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 store_api::storage::RegionId;
+
+use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
+use crate::error::Result;
+use crate::peer::Peer;
+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 = make_alter_region_request(
+                    region_id,
+                    &task.alter_table,
+                    table.table_info.ident.version,
+                );
+                requests.push(request);
+            }
+        }
+
+        Ok(AlterRequests { requests })
+    }
+}
+
+/// Makes an alter region request.
+pub fn make_alter_region_request(
+    region_id: RegionId,
+    alter_table_expr: &AlterTableExpr,
+    schema_version: u64,
+) -> 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,
+        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

@@ -0,0 +1,50 @@
+// 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,43 +13,66 @@
 // 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::utils::table_info::batch_update_table_info_values;
+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);
 
-        // Updates physical table's metadata.
-        AlterLogicalTablesExecutor::on_alter_metadata(
-            self.data.physical_table_id,
-            &self.context.table_metadata_manager,
-            physical_table_info,
-            region_distribution,
+        // Generates new table info
+        let old_raw_table_info = physical_table_info.table_info.clone();
+        let new_raw_table_info = physical_table_metadata::build_new_physical_table_info(
+            old_raw_table_info,
             &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(())
     }
 
     pub(crate) async fn update_logical_tables_metadata(&mut self) -> Result<()> {
         let table_info_values = self.build_update_metadata()?;
-        batch_update_table_info_values(&self.context.table_metadata_manager, table_info_values)
-            .await
+        let manager = &self.context.table_metadata_manager;
+        let chunk_size = manager.batch_update_table_info_value_chunk_size();
+        if table_info_values.len() > chunk_size {
+            let chunks = table_info_values
+                .into_iter()
+                .chunks(chunk_size)
+                .into_iter()
+                .map(|check| check.collect::<Vec<_>>())
+                .collect::<Vec<_>>();
+            for chunk in chunks {
+                manager.batch_update_table_info_values(chunk).await?;
+            }
+        } else {
+            manager
+                .batch_update_table_info_values(table_info_values)
+                .await?;
+        }
+
+        Ok(())
     }
 
     pub(crate) fn build_update_metadata(

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

@@ -1,279 +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 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::{
-    all_logical_table_routes_have_same_physical_id, 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 all_logical_table_routes_have_same_physical_id =
-            all_logical_table_routes_have_same_physical_id(
-                table_route_manager,
-                table_ids,
-                self.physical_table_id,
-            )
-            .await?;
-
-        ensure!(
-            all_logical_table_routes_have_same_physical_id,
-            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,9 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod executor;
+mod check;
 mod metadata;
 mod region_request;
+mod update_metadata;
 
 use std::vec;
 
@@ -28,29 +29,30 @@ use common_procedure::{
     Context as ProcedureContext, ContextProvider, Error as ProcedureError, LockKey, PoisonKey,
     PoisonKeys, Procedure, ProcedureId, Status, StringKey,
 };
-use common_telemetry::{error, info, warn};
+use common_telemetry::{debug, error, info};
+use futures::future::{self};
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, ResultExt};
-use store_api::metadata::ColumnMetadata;
-use store_api::metric_engine_consts::TABLE_COLUMN_METADATA_EXTENSION_KEY;
+use store_api::storage::RegionId;
 use strum::AsRefStr;
 use table::metadata::{RawTableInfo, TableId, TableInfo};
 use table::table_reference::TableReference;
 
-use crate::ddl::alter_table::executor::AlterTableExecutor;
+use crate::cache_invalidator::Context;
 use crate::ddl::utils::{
-    extract_column_metadatas, handle_multiple_results, map_to_procedure_error,
+    add_peer_context_if_needed, 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_leaders, region_distribution, RegionRoute};
+use crate::rpc::router::{find_leader_regions, find_leaders, region_distribution, RegionRoute};
 
 /// The alter table procedure
 pub struct AlterTableProcedure {
@@ -62,24 +64,6 @@ 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 {
@@ -87,42 +71,33 @@ 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,
+            data: AlterTableData::new(task, table_id),
             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.executor
-            .on_prepare(&self.context.table_metadata_manager)
-            .await?;
+        self.check_alter().await?;
         self.fill_table_info().await?;
 
-        // Safety: filled in `fill_table_info`.
+        // Validates the request and builds the new 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();
-        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);
+        self.new_table_info = Some(self.build_new_table_info(&table_info_value.table_info)?);
 
         // Safety: Checked in `AlterTableProcedure::new`.
         let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
@@ -165,7 +140,9 @@ 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!(
@@ -178,14 +155,31 @@ 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?;
-        let results = self
-            .executor
-            .on_alter_regions(
-                &self.context.node_manager,
-                &physical_table_route.region_routes,
-                alter_kind,
-            )
-            .await;
+        for datanode in leaders {
+            let requester = self.context.node_manager.datanode(&datanode).await;
+            let regions = find_leader_regions(&physical_table_route.region_routes, &datanode);
+
+            for region in regions {
+                let region_id = RegionId::new(table_id, region);
+                let request = self.make_alter_region_request(region_id, alter_kind.clone())?;
+                debug!("Submitting {request:?} to {datanode}");
+
+                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) => {
@@ -208,9 +202,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.handle_alter_region_response(results)?;
+                self.data.state = AlterTableState::UpdateMetadata;
                 Ok(Status::executing_with_clean_poisons(true))
             }
             MultipleResults::AllNonRetryable(error) => {
@@ -226,21 +220,9 @@ impl AlterTableProcedure {
         }
     }
 
-    fn handle_alter_region_response(&mut self, mut results: Vec<RegionResponse>) -> Result<()> {
-        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");
-        }
-        self.data.state = AlterTableState::UpdateMetadata;
-        Ok(())
-    }
-
     async fn submit_sync_region_requests(
         &mut self,
-        results: &[RegionResponse],
+        results: Vec<RegionResponse>,
         region_routes: &[RegionRoute],
     ) {
         // Safety: filled in `prepare` step.
@@ -262,34 +244,39 @@ 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: filled in `fill_table_info`.
+        // Safety: checked before.
         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 => AlterTableExecutor::validate_alter_table_expr(
-                &table_info_value.table_info,
-                self.data.task.alter_table.clone(),
-               )
+            None => self.build_new_table_info(&table_info_value.table_info)
                 .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);
                 })?,
         };
 
-        // Safety: region distribution is set in `submit_alter_region_requests`.
-        self.executor
-            .on_alter_metadata(
-                &self.context.table_metadata_manager,
-                table_info_value,
-                self.data.region_distribution.as_ref(),
+        debug!(
+            "Starting update table: {} metadata, new table info {:?}",
+            table_ref.to_string(),
+            new_info
+        );
+
+        // 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(),
-                &self.data.column_metadatas,
+                region_distribution,
+                table_info_value,
             )
             .await?;
+        }
 
         info!("Updated table metadata for table {table_ref}, table_id: {table_id}, kind: {alter_kind:?}");
         self.data.state = AlterTableState::InvalidateTableCache;
@@ -298,9 +285,18 @@ impl AlterTableProcedure {
 
     /// Broadcasts the invalidating table cache instructions.
     async fn on_broadcast(&mut self) -> Result<Status> {
-        self.executor
-            .invalidate_table_cache(&self.context.cache_invalidator)
+        let cache_invalidator = &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())
     }
 
@@ -322,16 +318,6 @@ 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]
@@ -394,8 +380,6 @@ 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.
@@ -408,7 +392,6 @@ impl AlterTableData {
             state: AlterTableState::Prepare,
             task,
             table_id,
-            column_metadatas: vec![],
             table_info_value: None,
             region_distribution: None,
         }
@@ -427,14 +410,4 @@ 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;
-    }
 }

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

@@ -0,0 +1,62 @@
+// 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::alter_table_expr::Kind;
+use api::v1::RenameTable;
+use common_catalog::format_full_table_name;
+use snafu::ensure;
+
+use crate::ddl::alter_table::AlterTableProcedure;
+use crate::error::{self, Result};
+use crate::key::table_name::TableNameKey;
+
+impl AlterTableProcedure {
+    /// Checks:
+    /// - The new table name doesn't exist (rename).
+    /// - Table exists.
+    pub(crate) async fn check_alter(&self) -> Result<()> {
+        let alter_expr = &self.data.task.alter_table;
+        let catalog = &alter_expr.catalog_name;
+        let schema = &alter_expr.schema_name;
+        let table_name = &alter_expr.table_name;
+        // Safety: Checked in `AlterTableProcedure::new`.
+        let alter_kind = self.data.task.alter_table.kind.as_ref().unwrap();
+
+        let manager = &self.context.table_metadata_manager;
+        if let Kind::RenameTable(RenameTable { new_table_name }) = alter_kind {
+            let new_table_name_key = TableNameKey::new(catalog, schema, new_table_name);
+            let exists = manager
+                .table_name_manager()
+                .exists(new_table_name_key)
+                .await?;
+            ensure!(
+                !exists,
+                error::TableAlreadyExistsSnafu {
+                    table_name: format_full_table_name(catalog, schema, new_table_name),
+                }
+            )
+        }
+
+        let table_name_key = TableNameKey::new(catalog, schema, table_name);
+        let exists = manager.table_name_manager().exists(table_name_key).await?;
+        ensure!(
+            exists,
+            error::TableNotFoundSnafu {
+                table_name: format_full_table_name(catalog, schema, &alter_expr.table_name),
+            }
+        );
+
+        Ok(())
+    }
+}

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

@@ -1,308 +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::HashMap;
-
-use api::region::RegionResponse;
-use api::v1::region::region_request::Body;
-use api::v1::region::{alter_request, AlterRequest, RegionRequest, RegionRequestHeader};
-use api::v1::AlterTableExpr;
-use common_catalog::format_full_table_name;
-use common_grpc_expr::alter_expr_to_request;
-use common_telemetry::debug;
-use common_telemetry::tracing_context::TracingContext;
-use futures::future;
-use snafu::{ensure, ResultExt};
-use store_api::metadata::ColumnMetadata;
-use store_api::storage::{RegionId, TableId};
-use table::metadata::{RawTableInfo, TableInfo};
-use table::requests::AlterKind;
-use table::table_name::TableName;
-
-use crate::cache_invalidator::{CacheInvalidatorRef, Context};
-use crate::ddl::utils::{add_peer_context_if_needed, raw_table_info};
-use crate::error::{self, Result, UnexpectedSnafu};
-use crate::instruction::CacheIdent;
-use crate::key::table_info::TableInfoValue;
-use crate::key::table_name::TableNameKey;
-use crate::key::{DeserializedValueWithBytes, RegionDistribution, TableMetadataManagerRef};
-use crate::node_manager::NodeManagerRef;
-use crate::rpc::router::{find_leaders, region_distribution, RegionRoute};
-
-/// [AlterTableExecutor] performs:
-/// - Alters the metadata of the table.
-/// - Alters regions on the datanode nodes.
-pub struct AlterTableExecutor {
-    table: TableName,
-    table_id: TableId,
-    /// The new table name if the alter kind is rename table.
-    new_table_name: Option<String>,
-}
-
-impl AlterTableExecutor {
-    /// Creates a new [`AlterTableExecutor`].
-    pub fn new(table: TableName, table_id: TableId, new_table_name: Option<String>) -> Self {
-        Self {
-            table,
-            table_id,
-            new_table_name,
-        }
-    }
-
-    /// Prepares to alter the table.
-    ///
-    /// ## Checks:
-    /// - The new table name doesn't exist (rename).
-    /// - Table exists.
-    pub(crate) async fn on_prepare(
-        &self,
-        table_metadata_manager: &TableMetadataManagerRef,
-    ) -> Result<()> {
-        let catalog = &self.table.catalog_name;
-        let schema = &self.table.schema_name;
-        let table_name = &self.table.table_name;
-
-        let manager = table_metadata_manager;
-        if let Some(new_table_name) = &self.new_table_name {
-            let new_table_name_key = TableNameKey::new(catalog, schema, new_table_name);
-            let exists = manager
-                .table_name_manager()
-                .exists(new_table_name_key)
-                .await?;
-            ensure!(
-                !exists,
-                error::TableAlreadyExistsSnafu {
-                    table_name: format_full_table_name(catalog, schema, new_table_name),
-                }
-            )
-        }
-
-        let table_name_key = TableNameKey::new(catalog, schema, table_name);
-        let exists = manager.table_name_manager().exists(table_name_key).await?;
-        ensure!(
-            exists,
-            error::TableNotFoundSnafu {
-                table_name: format_full_table_name(catalog, schema, table_name),
-            }
-        );
-
-        Ok(())
-    }
-
-    /// Validates the alter table expression and builds the new table info.
-    ///
-    /// This validation is performed early to ensure the alteration is valid before
-    /// proceeding to the `on_alter_metadata` state, where regions have already been altered.
-    /// Building the new table info here allows us to catch any issues with the
-    /// alteration before committing metadata changes.
-    pub(crate) fn validate_alter_table_expr(
-        table_info: &RawTableInfo,
-        alter_table_expr: AlterTableExpr,
-    ) -> Result<TableInfo> {
-        build_new_table_info(table_info, alter_table_expr)
-    }
-
-    /// Updates table metadata for alter table operation.
-    pub(crate) async fn on_alter_metadata(
-        &self,
-        table_metadata_manager: &TableMetadataManagerRef,
-        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
-        region_distribution: Option<&RegionDistribution>,
-        mut raw_table_info: RawTableInfo,
-        column_metadatas: &[ColumnMetadata],
-    ) -> Result<()> {
-        let table_ref = self.table.table_ref();
-        let table_id = self.table_id;
-
-        if let Some(new_table_name) = &self.new_table_name {
-            debug!(
-                "Starting update table: {} metadata, table_id: {}, new table info: {:?}, new table name: {}",
-                table_ref, table_id, raw_table_info, new_table_name
-            );
-
-            table_metadata_manager
-                .rename_table(current_table_info_value, new_table_name.to_string())
-                .await?;
-        } else {
-            debug!(
-                "Starting update table: {} metadata, table_id: {}, new table info: {:?}",
-                table_ref, table_id, raw_table_info
-            );
-
-            ensure!(
-                region_distribution.is_some(),
-                UnexpectedSnafu {
-                    err_msg: "region distribution is not set when updating table metadata",
-                }
-            );
-
-            if !column_metadatas.is_empty() {
-                raw_table_info::update_table_info_column_ids(&mut raw_table_info, column_metadatas);
-            }
-            table_metadata_manager
-                .update_table_info(
-                    current_table_info_value,
-                    region_distribution.cloned(),
-                    raw_table_info,
-                )
-                .await?;
-        }
-
-        Ok(())
-    }
-
-    /// Alters regions on the datanode nodes.
-    pub(crate) async fn on_alter_regions(
-        &self,
-        node_manager: &NodeManagerRef,
-        region_routes: &[RegionRoute],
-        kind: Option<alter_request::Kind>,
-    ) -> Vec<Result<RegionResponse>> {
-        let region_distribution = region_distribution(region_routes);
-        let leaders = find_leaders(region_routes)
-            .into_iter()
-            .map(|p| (p.id, p))
-            .collect::<HashMap<_, _>>();
-        let total_num_region = region_distribution
-            .values()
-            .map(|r| r.leader_regions.len())
-            .sum::<usize>();
-        let mut alter_region_tasks = Vec::with_capacity(total_num_region);
-        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;
-
-            for region_id in region_role_set.leader_regions {
-                let region_id = RegionId::new(self.table_id, region_id);
-                let request = make_alter_region_request(region_id, kind.clone());
-
-                let requester = requester.clone();
-                let peer = peer.clone();
-
-                alter_region_tasks.push(async move {
-                    requester
-                        .handle(request)
-                        .await
-                        .map_err(add_peer_context_if_needed(peer))
-                });
-            }
-        }
-
-        future::join_all(alter_region_tasks)
-            .await
-            .into_iter()
-            .collect::<Vec<_>>()
-    }
-
-    /// Invalidates cache for the table.
-    pub(crate) async fn invalidate_table_cache(
-        &self,
-        cache_invalidator: &CacheInvalidatorRef,
-    ) -> Result<()> {
-        let ctx = Context {
-            subject: Some(format!(
-                "Invalidate table cache by altering table {}, table_id: {}",
-                self.table.table_ref(),
-                self.table_id,
-            )),
-        };
-
-        cache_invalidator
-            .invalidate(
-                &ctx,
-                &[
-                    CacheIdent::TableName(self.table.clone()),
-                    CacheIdent::TableId(self.table_id),
-                ],
-            )
-            .await?;
-
-        Ok(())
-    }
-}
-
-/// Makes alter region request.
-pub(crate) fn make_alter_region_request(
-    region_id: RegionId,
-    kind: Option<alter_request::Kind>,
-) -> RegionRequest {
-    RegionRequest {
-        header: Some(RegionRequestHeader {
-            tracing_context: TracingContext::from_current_span().to_w3c(),
-            ..Default::default()
-        }),
-        body: Some(Body::Alter(AlterRequest {
-            region_id: region_id.as_u64(),
-            kind,
-            ..Default::default()
-        })),
-    }
-}
-
-/// Builds new table info after alteration.
-///
-/// This function creates a new table info by applying the alter table expression
-/// to the existing table info. For add column operations, it increments the
-/// `next_column_id` by the number of columns being added, which may result in gaps
-/// in the column id sequence.
-fn build_new_table_info(
-    table_info: &RawTableInfo,
-    alter_table_expr: AlterTableExpr,
-) -> Result<TableInfo> {
-    let table_info =
-        TableInfo::try_from(table_info.clone()).context(error::ConvertRawTableInfoSnafu)?;
-    let schema_name = &table_info.schema_name;
-    let catalog_name = &table_info.catalog_name;
-    let table_name = &table_info.name;
-    let table_id = table_info.ident.table_id;
-    let request = alter_expr_to_request(table_id, alter_table_expr)
-        .context(error::ConvertAlterTableRequestSnafu)?;
-
-    let new_meta = table_info
-        .meta
-        .builder_with_alter_kind(table_name, &request.alter_kind)
-        .context(error::TableSnafu)?
-        .build()
-        .with_context(|_| error::BuildTableMetaSnafu {
-            table_name: format_full_table_name(catalog_name, schema_name, table_name),
-        })?;
-
-    let mut new_info = table_info.clone();
-    new_info.meta = new_meta;
-    new_info.ident.version = table_info.ident.version + 1;
-    match request.alter_kind {
-        AlterKind::AddColumns { columns } => {
-            // Bumps the column id for the new columns.
-            // It may bump more than the actual number of columns added if there are
-            // existing columns, but it's fine.
-            new_info.meta.next_column_id += columns.len() as u32;
-        }
-        AlterKind::RenameTable { new_table_name } => {
-            new_info.name = new_table_name.to_string();
-        }
-        AlterKind::DropColumns { .. }
-        | AlterKind::ModifyColumnTypes { .. }
-        | AlterKind::SetTableOptions { .. }
-        | AlterKind::UnsetTableOptions { .. }
-        | AlterKind::SetIndexes { .. }
-        | AlterKind::UnsetIndexes { .. }
-        | AlterKind::DropDefaults { .. } => {}
-    }
-
-    Ok(new_info)
-}

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

@@ -15,16 +15,43 @@
 use std::collections::HashSet;
 
 use api::v1::alter_table_expr::Kind;
+use api::v1::region::region_request::Body;
 use api::v1::region::{
-    alter_request, AddColumn, AddColumns, DropColumn, DropColumns, RegionColumnDef,
+    alter_request, AddColumn, AddColumns, AlterRequest, DropColumn, DropColumns, RegionColumnDef,
+    RegionRequest, RegionRequestHeader,
 };
+use common_telemetry::tracing_context::TracingContext;
 use snafu::OptionExt;
+use store_api::storage::RegionId;
 use table::metadata::RawTableInfo;
 
 use crate::ddl::alter_table::AlterTableProcedure;
 use crate::error::{InvalidProtoMsgSnafu, Result};
 
 impl AlterTableProcedure {
+    /// Makes alter region request from existing an alter kind.
+    /// Region alter request always add columns if not exist.
+    pub(crate) fn make_alter_region_request(
+        &self,
+        region_id: RegionId,
+        kind: Option<alter_request::Kind>,
+    ) -> Result<RegionRequest> {
+        // Safety: checked
+        let table_info = self.data.table_info().unwrap();
+
+        Ok(RegionRequest {
+            header: Some(RegionRequestHeader {
+                tracing_context: TracingContext::from_current_span().to_w3c(),
+                ..Default::default()
+            }),
+            body: Some(Body::Alter(AlterRequest {
+                region_id: region_id.as_u64(),
+                schema_version: table_info.ident.version,
+                kind,
+            })),
+        })
+    }
+
     /// Makes alter kind proto that all regions can reuse.
     /// Region alter request always add columns if not exist.
     pub(crate) fn make_region_alter_kind(&self) -> Result<Option<alter_request::Kind>> {
@@ -108,8 +135,6 @@ fn create_proto_alter_kind(
         Kind::UnsetTableOptions(v) => Ok(Some(alter_request::Kind::UnsetTableOptions(v.clone()))),
         Kind::SetIndex(v) => Ok(Some(alter_request::Kind::SetIndex(v.clone()))),
         Kind::UnsetIndex(v) => Ok(Some(alter_request::Kind::UnsetIndex(v.clone()))),
-        Kind::SetIndexes(v) => Ok(Some(alter_request::Kind::SetIndexes(v.clone()))),
-        Kind::UnsetIndexes(v) => Ok(Some(alter_request::Kind::UnsetIndexes(v.clone()))),
         Kind::DropDefaults(v) => Ok(Some(alter_request::Kind::DropDefaults(v.clone()))),
     }
 }
@@ -130,7 +155,6 @@ mod tests {
     use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
     use store_api::storage::{RegionId, TableId};
 
-    use crate::ddl::alter_table::executor::make_alter_region_request;
     use crate::ddl::alter_table::AlterTableProcedure;
     use crate::ddl::test_util::columns::TestColumnDefBuilder;
     use crate::ddl::test_util::create_table::{
@@ -237,13 +261,15 @@ mod tests {
         let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context).unwrap();
         procedure.on_prepare().await.unwrap();
         let alter_kind = procedure.make_region_alter_kind().unwrap();
-        let Some(Body::Alter(alter_region_request)) =
-            make_alter_region_request(region_id, alter_kind).body
+        let Some(Body::Alter(alter_region_request)) = procedure
+            .make_alter_region_request(region_id, alter_kind)
+            .unwrap()
+            .body
         else {
             unreachable!()
         };
         assert_eq!(alter_region_request.region_id, region_id.as_u64());
-        assert_eq!(alter_region_request.schema_version, 0);
+        assert_eq!(alter_region_request.schema_version, 1);
         assert_eq!(
             alter_region_request.kind,
             Some(region::alter_request::Kind::AddColumns(
@@ -293,13 +319,15 @@ mod tests {
         let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context).unwrap();
         procedure.on_prepare().await.unwrap();
         let alter_kind = procedure.make_region_alter_kind().unwrap();
-        let Some(Body::Alter(alter_region_request)) =
-            make_alter_region_request(region_id, alter_kind).body
+        let Some(Body::Alter(alter_region_request)) = procedure
+            .make_alter_region_request(region_id, alter_kind)
+            .unwrap()
+            .body
         else {
             unreachable!()
         };
         assert_eq!(alter_region_request.region_id, region_id.as_u64());
-        assert_eq!(alter_region_request.schema_version, 0);
+        assert_eq!(alter_region_request.schema_version, 1);
         assert_eq!(
             alter_region_request.kind,
             Some(region::alter_request::Kind::ModifyColumnTypes(

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

@@ -0,0 +1,103 @@
+// 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_grpc_expr::alter_expr_to_request;
+use snafu::ResultExt;
+use table::metadata::{RawTableInfo, TableInfo};
+use table::requests::AlterKind;
+
+use crate::ddl::alter_table::AlterTableProcedure;
+use crate::error::{self, Result};
+use crate::key::table_info::TableInfoValue;
+use crate::key::{DeserializedValueWithBytes, RegionDistribution};
+
+impl AlterTableProcedure {
+    /// Builds new table info after alteration.
+    /// It bumps the column id of the table by the number of the add column requests.
+    /// So there may be holes in the column id sequence.
+    pub(crate) fn build_new_table_info(&self, table_info: &RawTableInfo) -> Result<TableInfo> {
+        let table_info =
+            TableInfo::try_from(table_info.clone()).context(error::ConvertRawTableInfoSnafu)?;
+        let table_ref = self.data.table_ref();
+        let alter_expr = self.data.task.alter_table.clone();
+        let request = alter_expr_to_request(self.data.table_id(), alter_expr)
+            .context(error::ConvertAlterTableRequestSnafu)?;
+
+        let new_meta = table_info
+            .meta
+            .builder_with_alter_kind(table_ref.table, &request.alter_kind)
+            .context(error::TableSnafu)?
+            .build()
+            .with_context(|_| error::BuildTableMetaSnafu {
+                table_name: table_ref.table,
+            })?;
+
+        let mut new_info = table_info.clone();
+        new_info.meta = new_meta;
+        new_info.ident.version = table_info.ident.version + 1;
+        match request.alter_kind {
+            AlterKind::AddColumns { columns } => {
+                // Bumps the column id for the new columns.
+                // It may bump more than the actual number of columns added if there are
+                // existing columns, but it's fine.
+                new_info.meta.next_column_id += columns.len() as u32;
+            }
+            AlterKind::RenameTable { new_table_name } => {
+                new_info.name = new_table_name.to_string();
+            }
+            AlterKind::DropColumns { .. }
+            | AlterKind::ModifyColumnTypes { .. }
+            | AlterKind::SetTableOptions { .. }
+            | AlterKind::UnsetTableOptions { .. }
+            | AlterKind::SetIndex { .. }
+            | AlterKind::UnsetIndex { .. }
+            | AlterKind::DropDefaults { .. } => {}
+        }
+
+        Ok(new_info)
+    }
+
+    /// Updates table metadata for rename table operation.
+    pub(crate) async fn on_update_metadata_for_rename(
+        &self,
+        new_table_name: String,
+        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
+    ) -> Result<()> {
+        let table_metadata_manager = &self.context.table_metadata_manager;
+        table_metadata_manager
+            .rename_table(current_table_info_value, new_table_name)
+            .await?;
+
+        Ok(())
+    }
+
+    /// Updates table metadata for alter table operation.
+    pub(crate) async fn on_update_metadata_for_alter(
+        &self,
+        new_table_info: RawTableInfo,
+        region_distribution: RegionDistribution,
+        current_table_info_value: &DeserializedValueWithBytes<TableInfoValue>,
+    ) -> Result<()> {
+        let table_metadata_manager = &self.context.table_metadata_manager;
+        table_metadata_manager
+            .update_table_info(
+                current_table_info_value,
+                Some(region_distribution),
+                new_table_info,
+            )
+            .await?;
+
+        Ok(())
+    }
+}

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

@@ -27,7 +27,7 @@ use common_telemetry::{debug, error, warn};
 use futures::future;
 pub use region_request::create_region_request_builder;
 use serde::{Deserialize, Serialize};
-use snafu::ResultExt;
+use snafu::{ensure, ResultExt};
 use store_api::metadata::ColumnMetadata;
 use store_api::metric_engine_consts::ALTER_PHYSICAL_EXTENSION_KEY;
 use store_api::storage::{RegionId, RegionNumber};
@@ -35,11 +35,10 @@ use strum::AsRefStr;
 use table::metadata::{RawTableInfo, TableId};
 
 use crate::ddl::utils::{
-    add_peer_context_if_needed, extract_column_metadatas, map_to_procedure_error,
-    sync_follower_regions,
+    add_peer_context_if_needed, map_to_procedure_error, sync_follower_regions,
 };
 use crate::ddl::DdlContext;
-use crate::error::Result;
+use crate::error::{DecodeJsonSnafu, MetadataCorruptionSnafu, Result};
 use crate::key::table_route::TableRouteValue;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
 use crate::metrics;
@@ -167,23 +166,47 @@ impl CreateLogicalTablesProcedure {
             .into_iter()
             .collect::<Result<Vec<_>>>()?;
 
-        if let Some(column_metadatas) =
-            extract_column_metadatas(&mut results, ALTER_PHYSICAL_EXTENSION_KEY)?
-        {
-            self.data.physical_columns = column_metadatas;
+        // Collects response 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, region_routes)
+                .await;
+            self.data.state = CreateTablesState::CreateMetadata;
+            return Ok(Status::executing(false));
+        }
+
+        // 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_schemas) = first {
+            self.data.physical_columns =
+                ColumnMetadata::decode_list(phy_raw_schemas).context(DecodeJsonSnafu)?;
         } else {
             warn!("creating 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, region_routes)
+        self.submit_sync_region_requests(results, region_routes)
             .await;
         self.data.state = CreateTablesState::CreateMetadata;
+
         Ok(Status::executing(true))
     }
 
     async fn submit_sync_region_requests(
         &self,
-        results: &[RegionResponse],
+        results: Vec<RegionResponse>,
         region_routes: &[RegionRoute],
     ) {
         if let Err(err) = sync_follower_regions(

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

@@ -22,7 +22,7 @@ use table::table_name::TableName;
 
 use crate::cache_invalidator::Context;
 use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
-use crate::ddl::utils::raw_table_info;
+use crate::ddl::physical_table_metadata;
 use crate::error::{Result, TableInfoNotFoundSnafu};
 use crate::instruction::CacheIdent;
 
@@ -47,7 +47,7 @@ impl CreateLogicalTablesProcedure {
         // Generates new table info
         let raw_table_info = physical_table_info.deref().table_info.clone();
 
-        let new_table_info = raw_table_info::build_new_physical_table_info(
+        let new_table_info = physical_table_metadata::build_new_physical_table_info(
             raw_table_info,
             &self.data.physical_columns,
         );

src/common/meta/src/ddl/create_table.rs

@@ -21,24 +21,21 @@ use common_error::ext::BoxedError;
 use common_procedure::error::{
     ExternalSnafu, FromJsonSnafu, Result as ProcedureResult, ToJsonSnafu,
 };
-use common_procedure::{Context as ProcedureContext, LockKey, Procedure, ProcedureId, Status};
+use common_procedure::{Context as ProcedureContext, LockKey, Procedure, Status};
+use common_telemetry::info;
 use common_telemetry::tracing_context::TracingContext;
-use common_telemetry::{info, warn};
 use futures::future::join_all;
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, OptionExt, ResultExt};
-use store_api::metadata::ColumnMetadata;
-use store_api::metric_engine_consts::TABLE_COLUMN_METADATA_EXTENSION_KEY;
 use store_api::storage::{RegionId, RegionNumber};
 use strum::AsRefStr;
 use table::metadata::{RawTableInfo, TableId};
 use table::table_reference::TableReference;
 
 use crate::ddl::create_table_template::{build_template, CreateRequestBuilder};
-use crate::ddl::utils::raw_table_info::update_table_info_column_ids;
 use crate::ddl::utils::{
-    add_peer_context_if_needed, convert_region_routes_to_detecting_regions,
-    extract_column_metadatas, map_to_procedure_error, region_storage_path,
+    add_peer_context_if_needed, convert_region_routes_to_detecting_regions, map_to_procedure_error,
+    region_storage_path,
 };
 use crate::ddl::{DdlContext, TableMetadata};
 use crate::error::{self, Result};
@@ -246,20 +243,14 @@ impl CreateTableProcedure {
             }
         }
 
-        let mut results = join_all(create_region_tasks)
+        join_all(create_region_tasks)
             .await
             .into_iter()
             .collect::<Result<Vec<_>>>()?;
 
-        if let Some(column_metadatas) =
-            extract_column_metadatas(&mut results, TABLE_COLUMN_METADATA_EXTENSION_KEY)?
-        {
-            self.creator.data.column_metadatas = column_metadatas;
-        } else {
-            warn!("creating table result doesn't contains extension key `{TABLE_COLUMN_METADATA_EXTENSION_KEY}`,leaving the table's column metadata unchanged");
-        }
-
         self.creator.data.state = CreateTableState::CreateMetadata;
+
+        // TODO(weny): Add more tests.
         Ok(Status::executing(true))
     }
 
@@ -267,15 +258,11 @@ impl CreateTableProcedure {
     ///
     /// Abort(not-retry):
     /// - Failed to create table metadata.
-    async fn on_create_metadata(&mut self, pid: ProcedureId) -> Result<Status> {
+    async fn on_create_metadata(&mut self) -> Result<Status> {
         let table_id = self.table_id();
-        let table_ref = self.creator.data.table_ref();
         let manager = &self.context.table_metadata_manager;
 
-        let mut raw_table_info = self.table_info().clone();
-        if !self.creator.data.column_metadatas.is_empty() {
-            update_table_info_column_ids(&mut raw_table_info, &self.creator.data.column_metadatas);
-        }
+        let raw_table_info = self.table_info().clone();
         // Safety: the region_wal_options must be allocated.
         let region_wal_options = self.region_wal_options()?.clone();
         // Safety: the table_route must be allocated.
@@ -289,10 +276,7 @@ impl CreateTableProcedure {
         self.context
             .register_failure_detectors(detecting_regions)
             .await;
-        info!(
-            "Successfully created table: {}, table_id: {}, procedure_id: {}",
-            table_ref, table_id, pid
-        );
+        info!("Created table metadata for table {table_id}");
 
         self.creator.opening_regions.clear();
         Ok(Status::done_with_output(table_id))
@@ -320,7 +304,7 @@ impl Procedure for CreateTableProcedure {
         Ok(())
     }
 
-    async fn execute(&mut self, ctx: &ProcedureContext) -> ProcedureResult<Status> {
+    async fn execute(&mut self, _ctx: &ProcedureContext) -> ProcedureResult<Status> {
         let state = &self.creator.data.state;
 
         let _timer = metrics::METRIC_META_PROCEDURE_CREATE_TABLE
@@ -330,7 +314,7 @@ impl Procedure for CreateTableProcedure {
         match state {
             CreateTableState::Prepare => self.on_prepare().await,
             CreateTableState::DatanodeCreateRegions => self.on_datanode_create_regions().await,
-            CreateTableState::CreateMetadata => self.on_create_metadata(ctx.procedure_id).await,
+            CreateTableState::CreateMetadata => self.on_create_metadata().await,
         }
         .map_err(map_to_procedure_error)
     }
@@ -362,7 +346,6 @@ impl TableCreator {
         Self {
             data: CreateTableData {
                 state: CreateTableState::Prepare,
-                column_metadatas: vec![],
                 task,
                 table_route: None,
                 region_wal_options: None,
@@ -424,8 +407,6 @@ pub enum CreateTableState {
 pub struct CreateTableData {
     pub state: CreateTableState,
     pub task: CreateTableTask,
-    #[serde(default)]
-    pub column_metadatas: Vec<ColumnMetadata>,
     /// None stands for not allocated yet.
     table_route: Option<PhysicalTableRouteValue>,
     /// None stands for not allocated yet.

src/common/meta/src/ddl/create_table_template.rs

@@ -14,57 +14,17 @@
 
 use std::collections::HashMap;
 
-use api::v1::column_def::try_as_column_def;
 use api::v1::region::{CreateRequest, RegionColumnDef};
 use api::v1::{ColumnDef, CreateTableExpr, SemanticType};
-use snafu::{OptionExt, ResultExt};
-use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME};
+use snafu::OptionExt;
+use store_api::metric_engine_consts::LOGICAL_TABLE_METADATA_KEY;
 use store_api::storage::{RegionId, RegionNumber};
-use table::metadata::{RawTableInfo, TableId};
+use table::metadata::TableId;
 
-use crate::error::{self, Result};
+use crate::error;
+use crate::error::Result;
 use crate::wal_options_allocator::prepare_wal_options;
 
-/// Builds a [CreateRequest] from a [RawTableInfo].
-///
-/// Note: **This method is only used for creating logical tables.**
-pub(crate) fn build_template_from_raw_table_info(
-    raw_table_info: &RawTableInfo,
-) -> Result<CreateRequest> {
-    let primary_key_indices = &raw_table_info.meta.primary_key_indices;
-    let column_defs = raw_table_info
-        .meta
-        .schema
-        .column_schemas
-        .iter()
-        .enumerate()
-        .map(|(i, c)| {
-            let is_primary_key = primary_key_indices.contains(&i);
-            let column_def = try_as_column_def(c, is_primary_key)
-                .context(error::ConvertColumnDefSnafu { column: &c.name })?;
-
-            Ok(RegionColumnDef {
-                column_def: Some(column_def),
-                // The column id will be overridden by the metric engine.
-                // So we just use the index as the column id.
-                column_id: i as u32,
-            })
-        })
-        .collect::<Result<Vec<_>>>()?;
-
-    let options = HashMap::from(&raw_table_info.meta.options);
-    let template = CreateRequest {
-        region_id: 0,
-        engine: METRIC_ENGINE_NAME.to_string(),
-        column_defs,
-        primary_key: primary_key_indices.iter().map(|i| *i as u32).collect(),
-        path: String::new(),
-        options,
-    };
-
-    Ok(template)
-}
-
 pub(crate) fn build_template(create_table_expr: &CreateTableExpr) -> Result<CreateRequest> {
     let column_defs = create_table_expr
         .column_defs

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

@@ -185,15 +185,11 @@ impl DropTableExecutor {
             .await
     }
 
-    /// Invalidates caches for the table.
+    /// Invalidates frontend caches
     pub async fn invalidate_table_cache(&self, ctx: &DdlContext) -> Result<()> {
         let cache_invalidator = &ctx.cache_invalidator;
         let ctx = Context {
-            subject: Some(format!(
-                "Invalidate table cache by dropping table {}, table_id: {}",
-                self.table.table_ref(),
-                self.table_id,
-            )),
+            subject: Some("Invalidate table cache by dropping table".to_string()),
         };
 
         cache_invalidator

src/common/meta/src/ddl/physical_table_metadata.rs

@@ -0,0 +1,60 @@
+// 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::SemanticType;
+use store_api::metadata::ColumnMetadata;
+use table::metadata::RawTableInfo;
+
+/// Generate the new physical table info.
+pub(crate) fn build_new_physical_table_info(
+    mut raw_table_info: RawTableInfo,
+    physical_columns: &[ColumnMetadata],
+) -> RawTableInfo {
+    let existing_columns = raw_table_info
+        .meta
+        .schema
+        .column_schemas
+        .iter()
+        .map(|col| col.name.clone())
+        .collect::<HashSet<_>>();
+    let primary_key_indices = &mut raw_table_info.meta.primary_key_indices;
+    let value_indices = &mut raw_table_info.meta.value_indices;
+    value_indices.clear();
+    let time_index = &mut raw_table_info.meta.schema.timestamp_index;
+    let columns = &mut raw_table_info.meta.schema.column_schemas;
+    columns.clear();
+
+    for (idx, col) in physical_columns.iter().enumerate() {
+        match col.semantic_type {
+            SemanticType::Tag => {
+                // push new primary key to the end.
+                if !existing_columns.contains(&col.column_schema.name) {
+                    primary_key_indices.push(idx);
+                }
+            }
+            SemanticType::Field => value_indices.push(idx),
+            SemanticType::Timestamp => *time_index = Some(idx),
+        }
+
+        columns.push(col.column_schema.clone());
+    }
+
+    if let Some(time_index) = *time_index {
+        raw_table_info.meta.schema.column_schemas[time_index].set_time_index();
+    }
+
+    raw_table_info
+}

src/common/meta/src/ddl/table_meta.rs

@@ -122,7 +122,6 @@ impl TableMetadataAllocator {
         );
 
         let peers = self.peer_allocator.alloc(regions).await?;
-        debug!("Allocated peers {:?} for table {}", peers, table_id);
         let region_routes = task
             .partitions
             .iter()
@@ -175,10 +174,6 @@ impl TableMetadataAllocator {
             region_wal_options,
         })
     }
-
-    pub fn table_id_sequence(&self) -> SequenceRef {
-        self.table_id_sequence.clone()
-    }
 }
 
 pub type PeerAllocatorRef = Arc<dyn PeerAllocator>;

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

@@ -17,7 +17,6 @@ pub mod columns;
 pub mod create_table;
 pub mod datanode_handler;
 pub mod flownode_handler;
-pub mod region_metadata;
 
 use std::assert_matches::assert_matches;
 use std::collections::HashMap;
@@ -25,14 +24,7 @@ use std::collections::HashMap;
 use api::v1::meta::Partition;
 use api::v1::{ColumnDataType, SemanticType};
 use common_procedure::Status;
-use datatypes::prelude::ConcreteDataType;
-use datatypes::schema::ColumnSchema;
-use store_api::metadata::ColumnMetadata;
-use store_api::metric_engine_consts::{
-    DATA_SCHEMA_TABLE_ID_COLUMN_NAME, DATA_SCHEMA_TSID_COLUMN_NAME, LOGICAL_TABLE_METADATA_KEY,
-    METRIC_ENGINE_NAME,
-};
-use store_api::storage::consts::ReservedColumnId;
+use store_api::metric_engine_consts::{LOGICAL_TABLE_METADATA_KEY, METRIC_ENGINE_NAME};
 use table::metadata::{RawTableInfo, TableId};
 
 use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
@@ -154,7 +146,6 @@ pub fn test_create_logical_table_task(name: &str) -> CreateTableTask {
     }
 }
 
-/// Creates a physical table task with a single region.
 pub fn test_create_physical_table_task(name: &str) -> CreateTableTask {
     let create_table = TestCreateTableExprBuilder::default()
         .column_defs([
@@ -191,95 +182,3 @@ pub fn test_create_physical_table_task(name: &str) -> CreateTableTask {
         table_info,
     }
 }
-
-/// Creates a column metadata list with tag fields.
-pub fn test_column_metadatas(tag_fields: &[&str]) -> Vec<ColumnMetadata> {
-    let mut output = Vec::with_capacity(tag_fields.len() + 4);
-    output.extend([
-        ColumnMetadata {
-            column_schema: ColumnSchema::new(
-                "ts",
-                ConcreteDataType::timestamp_millisecond_datatype(),
-                false,
-            ),
-            semantic_type: SemanticType::Timestamp,
-            column_id: 0,
-        },
-        ColumnMetadata {
-            column_schema: ColumnSchema::new("value", ConcreteDataType::float64_datatype(), false),
-            semantic_type: SemanticType::Field,
-            column_id: 1,
-        },
-        ColumnMetadata {
-            column_schema: ColumnSchema::new(
-                DATA_SCHEMA_TABLE_ID_COLUMN_NAME,
-                ConcreteDataType::timestamp_millisecond_datatype(),
-                false,
-            ),
-            semantic_type: SemanticType::Tag,
-            column_id: ReservedColumnId::table_id(),
-        },
-        ColumnMetadata {
-            column_schema: ColumnSchema::new(
-                DATA_SCHEMA_TSID_COLUMN_NAME,
-                ConcreteDataType::float64_datatype(),
-                false,
-            ),
-            semantic_type: SemanticType::Tag,
-            column_id: ReservedColumnId::tsid(),
-        },
-    ]);
-
-    for (i, name) in tag_fields.iter().enumerate() {
-        output.push(ColumnMetadata {
-            column_schema: ColumnSchema::new(
-                name.to_string(),
-                ConcreteDataType::string_datatype(),
-                true,
-            ),
-            semantic_type: SemanticType::Tag,
-            column_id: (i + 2) as u32,
-        });
-    }
-
-    output
-}
-
-/// Asserts the column names.
-pub fn assert_column_name(table_info: &RawTableInfo, expected_column_names: &[&str]) {
-    assert_eq!(
-        table_info
-            .meta
-            .schema
-            .column_schemas
-            .iter()
-            .map(|c| c.name.to_string())
-            .collect::<Vec<_>>(),
-        expected_column_names
-    );
-}
-
-/// Asserts the column metadatas
-pub fn assert_column_name_and_id(column_metadatas: &[ColumnMetadata], expected: &[(&str, u32)]) {
-    assert_eq!(expected.len(), column_metadatas.len());
-    for (name, id) in expected {
-        let column_metadata = column_metadatas
-            .iter()
-            .find(|c| c.column_id == *id)
-            .unwrap();
-        assert_eq!(column_metadata.column_schema.name, *name);
-    }
-}
-
-/// Gets the raw table info.
-pub async fn get_raw_table_info(ddl_context: &DdlContext, table_id: TableId) -> RawTableInfo {
-    ddl_context
-        .table_metadata_manager
-        .table_info_manager()
-        .get(table_id)
-        .await
-        .unwrap()
-        .unwrap()
-        .into_inner()
-        .table_info
-}

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

@@ -132,7 +132,6 @@ pub fn build_raw_table_info_from_expr(expr: &CreateTableExpr) -> RawTableInfo {
             options: TableOptions::try_from_iter(&expr.table_options).unwrap(),
             created_on: DateTime::default(),
             partition_key_indices: vec![],
-            column_ids: vec![],
         },
         table_type: TableType::Base,
     }

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

@@ -12,11 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::collections::HashMap;
-use std::sync::Arc;
-
 use api::region::RegionResponse;
-use api::v1::region::region_request::Body;
 use api::v1::region::RegionRequest;
 use common_error::ext::{BoxedError, ErrorExt, StackError};
 use common_error::status_code::StatusCode;
@@ -24,8 +20,6 @@ use common_query::request::QueryRequest;
 use common_recordbatch::SendableRecordBatchStream;
 use common_telemetry::debug;
 use snafu::{ResultExt, Snafu};
-use store_api::metadata::RegionMetadata;
-use store_api::storage::RegionId;
 use tokio::sync::mpsc;
 
 use crate::error::{self, Error, Result};
@@ -51,13 +45,10 @@ impl MockDatanodeHandler for () {
     }
 }
 
-type RegionRequestHandler =
-    Arc<dyn Fn(Peer, RegionRequest) -> Result<RegionResponse> + Send + Sync>;
-
 #[derive(Clone)]
 pub struct DatanodeWatcher {
     sender: mpsc::Sender<(Peer, RegionRequest)>,
-    handler: Option<RegionRequestHandler>,
+    handler: Option<fn(Peer, RegionRequest) -> Result<RegionResponse>>,
 }
 
 impl DatanodeWatcher {
@@ -70,9 +61,9 @@ impl DatanodeWatcher {
 
     pub fn with_handler(
         mut self,
-        user_handler: impl Fn(Peer, RegionRequest) -> Result<RegionResponse> + Send + Sync + 'static,
+        user_handler: fn(Peer, RegionRequest) -> Result<RegionResponse>,
     ) -> Self {
-        self.handler = Some(Arc::new(user_handler));
+        self.handler = Some(user_handler);
         self
     }
 }
@@ -85,7 +76,7 @@ impl MockDatanodeHandler for DatanodeWatcher {
             .send((peer.clone(), request.clone()))
             .await
             .unwrap();
-        if let Some(handler) = self.handler.as_ref() {
+        if let Some(handler) = self.handler {
             handler(peer.clone(), request)
         } else {
             Ok(RegionResponse::new(0))
@@ -282,47 +273,3 @@ impl MockDatanodeHandler for AllFailureDatanodeHandler {
         unreachable!()
     }
 }
-
-#[derive(Clone)]
-pub struct ListMetadataDatanodeHandler {
-    pub region_metadatas: HashMap<RegionId, Option<RegionMetadata>>,
-}
-
-impl ListMetadataDatanodeHandler {
-    pub fn new(region_metadatas: HashMap<RegionId, Option<RegionMetadata>>) -> Self {
-        Self { region_metadatas }
-    }
-}
-
-#[async_trait::async_trait]
-impl MockDatanodeHandler for ListMetadataDatanodeHandler {
-    async fn handle(&self, _peer: &Peer, request: RegionRequest) -> Result<RegionResponse> {
-        let Some(Body::ListMetadata(req)) = request.body else {
-            unreachable!()
-        };
-        let mut response = RegionResponse::new(0);
-
-        let mut output = Vec::with_capacity(req.region_ids.len());
-        for region_id in req.region_ids {
-            match self.region_metadatas.get(&RegionId::from_u64(region_id)) {
-                Some(metadata) => {
-                    output.push(metadata.clone());
-                }
-                None => {
-                    output.push(None);
-                }
-            }
-        }
-
-        response.metadata = serde_json::to_vec(&output).unwrap();
-        Ok(response)
-    }
-
-    async fn handle_query(
-        &self,
-        _peer: &Peer,
-        _request: QueryRequest,
-    ) -> Result<SendableRecordBatchStream> {
-        unreachable!()
-    }
-}

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

@@ -1,34 +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::SemanticType;
-use store_api::metadata::{ColumnMetadata, RegionMetadata, RegionMetadataBuilder};
-use store_api::storage::RegionId;
-
-/// Builds a region metadata with the given column metadatas.
-pub fn build_region_metadata(
-    region_id: RegionId,
-    column_metadatas: &[ColumnMetadata],
-) -> RegionMetadata {
-    let mut builder = RegionMetadataBuilder::new(region_id);
-    let mut primary_key = vec![];
-    for column_metadata in column_metadatas {
-        builder.push_column_metadata(column_metadata.clone());
-        if column_metadata.semantic_type == SemanticType::Tag {
-            primary_key.push(column_metadata.column_id);
-        }
-    }
-    builder.primary_key(primary_key);
-    builder.build().unwrap()
-}

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

@@ -23,20 +23,17 @@ use api::v1::{ColumnDataType, SemanticType};
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
 use common_procedure::{Procedure, ProcedureId, Status};
 use common_procedure_test::MockContextProvider;
-use store_api::metadata::ColumnMetadata;
-use store_api::metric_engine_consts::{ALTER_PHYSICAL_EXTENSION_KEY, MANIFEST_INFO_EXTENSION_KEY};
+use store_api::metric_engine_consts::MANIFEST_INFO_EXTENSION_KEY;
 use store_api::region_engine::RegionManifestInfo;
-use store_api::storage::consts::ReservedColumnId;
 use store_api::storage::RegionId;
 use tokio::sync::mpsc;
 
 use crate::ddl::alter_logical_tables::AlterLogicalTablesProcedure;
 use crate::ddl::test_util::alter_table::TestAlterTableExprBuilder;
 use crate::ddl::test_util::columns::TestColumnDefBuilder;
-use crate::ddl::test_util::datanode_handler::DatanodeWatcher;
+use crate::ddl::test_util::datanode_handler::{DatanodeWatcher, NaiveDatanodeHandler};
 use crate::ddl::test_util::{
-    assert_column_name, create_logical_table, create_physical_table,
-    create_physical_table_metadata, get_raw_table_info, test_column_metadatas,
+    create_logical_table, create_physical_table, create_physical_table_metadata,
     test_create_physical_table_task,
 };
 use crate::error::Error::{AlterLogicalTablesInvalidArguments, TableNotFound};
@@ -99,52 +96,6 @@ fn make_alter_logical_table_rename_task(
     }
 }
 
-fn make_alters_request_handler(
-    column_metadatas: Vec<ColumnMetadata>,
-) -> impl Fn(Peer, RegionRequest) -> Result<RegionResponse> {
-    move |_peer: Peer, request: RegionRequest| {
-        if let region_request::Body::Alters(_) = request.body.unwrap() {
-            let mut response = RegionResponse::new(0);
-            // Default region id for physical table.
-            let region_id = RegionId::new(1000, 1);
-            response.extensions.insert(
-                MANIFEST_INFO_EXTENSION_KEY.to_string(),
-                RegionManifestInfo::encode_list(&[(
-                    region_id,
-                    RegionManifestInfo::metric(1, 0, 2, 0),
-                )])
-                .unwrap(),
-            );
-            response.extensions.insert(
-                ALTER_PHYSICAL_EXTENSION_KEY.to_string(),
-                ColumnMetadata::encode_list(&column_metadatas).unwrap(),
-            );
-            return Ok(response);
-        }
-        Ok(RegionResponse::new(0))
-    }
-}
-
-fn assert_alters_request(
-    peer: Peer,
-    request: RegionRequest,
-    expected_peer_id: u64,
-    expected_region_ids: &[RegionId],
-) {
-    assert_eq!(peer.id, expected_peer_id,);
-    let Some(region_request::Body::Alters(req)) = request.body else {
-        unreachable!();
-    };
-    for (i, region_id) in expected_region_ids.iter().enumerate() {
-        assert_eq!(
-            req.requests[i].region_id,
-            *region_id,
-            "actual region id: {}",
-            RegionId::from_u64(req.requests[i].region_id)
-        );
-    }
-}
-
 #[tokio::test]
 async fn test_on_prepare_check_schema() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
@@ -254,20 +205,15 @@ async fn test_on_prepare() {
 
 #[tokio::test]
 async fn test_on_update_metadata() {
-    common_telemetry::init_default_ut_logging();
-    let (tx, mut rx) = mpsc::channel(8);
-    let test_column_metadatas = test_column_metadatas(&["new_col", "mew_col"]);
-    let datanode_handler =
-        DatanodeWatcher::new(tx).with_handler(make_alters_request_handler(test_column_metadatas));
-    let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler));
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
 
     // Creates physical table
     let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates 3 logical tables
-    let logical_table1_id = create_logical_table(ddl_context.clone(), phy_id, "table1").await;
-    let logical_table2_id = create_logical_table(ddl_context.clone(), phy_id, "table2").await;
-    let logical_table3_id = create_logical_table(ddl_context.clone(), phy_id, "table3").await;
+    create_logical_table(ddl_context.clone(), phy_id, "table1").await;
+    create_logical_table(ddl_context.clone(), phy_id, "table2").await;
+    create_logical_table(ddl_context.clone(), phy_id, "table3").await;
     create_logical_table(ddl_context.clone(), phy_id, "table4").await;
     create_logical_table(ddl_context.clone(), phy_id, "table5").await;
 
@@ -277,7 +223,7 @@ async fn test_on_update_metadata() {
         make_alter_logical_table_add_column_task(None, "table3", vec!["new_col".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context.clone());
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
     let mut status = procedure.on_prepare().await.unwrap();
     assert_matches!(
         status,
@@ -309,52 +255,18 @@ async fn test_on_update_metadata() {
             clean_poisons: false
         }
     );
-    let (peer, request) = rx.try_recv().unwrap();
-    rx.try_recv().unwrap_err();
-    assert_alters_request(
-        peer,
-        request,
-        0,
-        &[
-            RegionId::new(logical_table1_id, 0),
-            RegionId::new(logical_table2_id, 0),
-            RegionId::new(logical_table3_id, 0),
-        ],
-    );
-
-    let table_info = get_raw_table_info(&ddl_context, phy_id).await;
-    assert_column_name(
-        &table_info,
-        &["ts", "value", "__table_id", "__tsid", "new_col", "mew_col"],
-    );
-    assert_eq!(
-        table_info.meta.column_ids,
-        vec![
-            0,
-            1,
-            ReservedColumnId::table_id(),
-            ReservedColumnId::tsid(),
-            2,
-            3
-        ]
-    );
 }
 
 #[tokio::test]
 async fn test_on_part_duplicate_alter_request() {
-    common_telemetry::init_default_ut_logging();
-    let (tx, mut rx) = mpsc::channel(8);
-    let column_metadatas = test_column_metadatas(&["col_0"]);
-    let handler =
-        DatanodeWatcher::new(tx).with_handler(make_alters_request_handler(column_metadatas));
-    let node_manager = Arc::new(MockDatanodeManager::new(handler));
-    let mut ddl_context = new_ddl_context(node_manager);
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
+    let ddl_context = new_ddl_context(node_manager);
 
     // Creates physical table
     let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates 3 logical tables
-    let logical_table1_id = create_logical_table(ddl_context.clone(), phy_id, "table1").await;
-    let logical_table2_id = create_logical_table(ddl_context.clone(), phy_id, "table2").await;
+    create_logical_table(ddl_context.clone(), phy_id, "table1").await;
+    create_logical_table(ddl_context.clone(), phy_id, "table2").await;
 
     let tasks = vec![
         make_alter_logical_table_add_column_task(None, "table1", vec!["col_0".to_string()]),
@@ -393,40 +305,6 @@ async fn test_on_part_duplicate_alter_request() {
             clean_poisons: false
         }
     );
-    let (peer, request) = rx.try_recv().unwrap();
-    rx.try_recv().unwrap_err();
-    assert_alters_request(
-        peer,
-        request,
-        0,
-        &[
-            RegionId::new(logical_table1_id, 0),
-            RegionId::new(logical_table2_id, 0),
-        ],
-    );
-
-    let table_info = get_raw_table_info(&ddl_context, phy_id).await;
-    assert_column_name(
-        &table_info,
-        &["ts", "value", "__table_id", "__tsid", "col_0"],
-    );
-    assert_eq!(
-        table_info.meta.column_ids,
-        vec![
-            0,
-            1,
-            ReservedColumnId::table_id(),
-            ReservedColumnId::tsid(),
-            2
-        ]
-    );
-
-    let (tx, mut rx) = mpsc::channel(8);
-    let column_metadatas = test_column_metadatas(&["col_0", "new_col_1", "new_col_2"]);
-    let handler =
-        DatanodeWatcher::new(tx).with_handler(make_alters_request_handler(column_metadatas));
-    let node_manager = Arc::new(MockDatanodeManager::new(handler));
-    ddl_context.node_manager = node_manager;
 
     // re-alter
     let tasks = vec![
@@ -479,44 +357,6 @@ async fn test_on_part_duplicate_alter_request() {
         }
     );
 
-    let (peer, request) = rx.try_recv().unwrap();
-    rx.try_recv().unwrap_err();
-    assert_alters_request(
-        peer,
-        request,
-        0,
-        &[
-            RegionId::new(logical_table1_id, 0),
-            RegionId::new(logical_table2_id, 0),
-        ],
-    );
-
-    let table_info = get_raw_table_info(&ddl_context, phy_id).await;
-    assert_column_name(
-        &table_info,
-        &[
-            "ts",
-            "value",
-            "__table_id",
-            "__tsid",
-            "col_0",
-            "new_col_1",
-            "new_col_2",
-        ],
-    );
-    assert_eq!(
-        table_info.meta.column_ids,
-        vec![
-            0,
-            1,
-            ReservedColumnId::table_id(),
-            ReservedColumnId::tsid(),
-            2,
-            3,
-            4,
-        ]
-    );
-
     let table_name_keys = vec![
         TableNameKey::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "table1"),
         TableNameKey::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "table2"),
@@ -582,13 +422,27 @@ async fn test_on_part_duplicate_alter_request() {
     );
 }
 
+fn alters_request_handler(_peer: Peer, request: RegionRequest) -> Result<RegionResponse> {
+    if let region_request::Body::Alters(_) = request.body.unwrap() {
+        let mut response = RegionResponse::new(0);
+        // Default region id for physical table.
+        let region_id = RegionId::new(1000, 1);
+        response.extensions.insert(
+            MANIFEST_INFO_EXTENSION_KEY.to_string(),
+            RegionManifestInfo::encode_list(&[(region_id, RegionManifestInfo::metric(1, 0, 2, 0))])
+                .unwrap(),
+        );
+        return Ok(response);
+    }
+
+    Ok(RegionResponse::new(0))
+}
+
 #[tokio::test]
 async fn test_on_submit_alter_region_request() {
     common_telemetry::init_default_ut_logging();
     let (tx, mut rx) = mpsc::channel(8);
-    let column_metadatas = test_column_metadatas(&["new_col", "mew_col"]);
-    let handler =
-        DatanodeWatcher::new(tx).with_handler(make_alters_request_handler(column_metadatas));
+    let handler = DatanodeWatcher::new(tx).with_handler(alters_request_handler);
     let node_manager = Arc::new(MockDatanodeManager::new(handler));
     let ddl_context = new_ddl_context(node_manager);
 

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

@@ -30,12 +30,7 @@ use common_error::status_code::StatusCode;
 use common_procedure::store::poison_store::PoisonStore;
 use common_procedure::{ProcedureId, Status};
 use common_procedure_test::MockContextProvider;
-use datatypes::prelude::ConcreteDataType;
-use datatypes::schema::ColumnSchema;
-use store_api::metadata::ColumnMetadata;
-use store_api::metric_engine_consts::{
-    MANIFEST_INFO_EXTENSION_KEY, TABLE_COLUMN_METADATA_EXTENSION_KEY,
-};
+use store_api::metric_engine_consts::MANIFEST_INFO_EXTENSION_KEY;
 use store_api::region_engine::RegionManifestInfo;
 use store_api::storage::RegionId;
 use table::requests::TTL_KEY;
@@ -48,7 +43,6 @@ use crate::ddl::test_util::datanode_handler::{
     AllFailureDatanodeHandler, DatanodeWatcher, PartialSuccessDatanodeHandler,
     RequestOutdatedErrorDatanodeHandler,
 };
-use crate::ddl::test_util::{assert_column_name, assert_column_name_and_id};
 use crate::error::{Error, Result};
 use crate::key::datanode_table::DatanodeTableKey;
 use crate::key::table_name::TableNameKey;
@@ -185,30 +179,6 @@ fn alter_request_handler(_peer: Peer, request: RegionRequest) -> Result<RegionRe
             RegionManifestInfo::encode_list(&[(region_id, RegionManifestInfo::mito(1, 1))])
                 .unwrap(),
         );
-        response.extensions.insert(
-            TABLE_COLUMN_METADATA_EXTENSION_KEY.to_string(),
-            ColumnMetadata::encode_list(&[
-                ColumnMetadata {
-                    column_schema: ColumnSchema::new(
-                        "ts",
-                        ConcreteDataType::timestamp_millisecond_datatype(),
-                        false,
-                    ),
-                    semantic_type: SemanticType::Timestamp,
-                    column_id: 0,
-                },
-                ColumnMetadata {
-                    column_schema: ColumnSchema::new(
-                        "host",
-                        ConcreteDataType::float64_datatype(),
-                        false,
-                    ),
-                    semantic_type: SemanticType::Tag,
-                    column_id: 1,
-                },
-            ])
-            .unwrap(),
-        );
         return Ok(response);
     }
 
@@ -217,7 +187,6 @@ fn alter_request_handler(_peer: Peer, request: RegionRequest) -> Result<RegionRe
 
 #[tokio::test]
 async fn test_on_submit_alter_request() {
-    common_telemetry::init_default_ut_logging();
     let (tx, mut rx) = mpsc::channel(8);
     let datanode_handler = DatanodeWatcher::new(tx).with_handler(alter_request_handler);
     let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler));
@@ -265,8 +234,6 @@ async fn test_on_submit_alter_request() {
     assert_sync_request(peer, request, 4, RegionId::new(table_id, 2), 1);
     let (peer, request) = results.remove(0);
     assert_sync_request(peer, request, 5, RegionId::new(table_id, 1), 1);
-    let column_metadatas = procedure.data().column_metadatas();
-    assert_column_name_and_id(column_metadatas, &[("ts", 0), ("host", 1)]);
 }
 
 #[tokio::test]
@@ -411,7 +378,6 @@ async fn test_on_update_metadata_rename() {
 
 #[tokio::test]
 async fn test_on_update_metadata_add_columns() {
-    common_telemetry::init_default_ut_logging();
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
     let table_name = "foo";
@@ -465,34 +431,6 @@ async fn test_on_update_metadata_add_columns() {
         .submit_alter_region_requests(procedure_id, provider.as_ref())
         .await
         .unwrap();
-    // Returned column metadatas is empty.
-    assert!(procedure.data().column_metadatas().is_empty());
-    procedure.mut_data().set_column_metadatas(vec![
-        ColumnMetadata {
-            column_schema: ColumnSchema::new(
-                "ts",
-                ConcreteDataType::timestamp_millisecond_datatype(),
-                false,
-            ),
-            semantic_type: SemanticType::Timestamp,
-            column_id: 0,
-        },
-        ColumnMetadata {
-            column_schema: ColumnSchema::new("host", ConcreteDataType::float64_datatype(), false),
-            semantic_type: SemanticType::Tag,
-            column_id: 1,
-        },
-        ColumnMetadata {
-            column_schema: ColumnSchema::new("cpu", ConcreteDataType::float64_datatype(), false),
-            semantic_type: SemanticType::Tag,
-            column_id: 2,
-        },
-        ColumnMetadata {
-            column_schema: ColumnSchema::new("my_tag3", ConcreteDataType::string_datatype(), true),
-            semantic_type: SemanticType::Tag,
-            column_id: 3,
-        },
-    ]);
     procedure.on_update_metadata().await.unwrap();
 
     let table_info = ddl_context
@@ -509,8 +447,6 @@ async fn test_on_update_metadata_add_columns() {
         table_info.meta.schema.column_schemas.len() as u32,
         table_info.meta.next_column_id
     );
-    assert_column_name(&table_info, &["ts", "host", "cpu", "my_tag3"]);
-    assert_eq!(table_info.meta.column_ids, vec![0, 1, 2, 3]);
 }
 
 #[tokio::test]

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

@@ -23,18 +23,15 @@ use common_error::ext::ErrorExt;
 use common_error::status_code::StatusCode;
 use common_procedure::{Context as ProcedureContext, Procedure, ProcedureId, Status};
 use common_procedure_test::MockContextProvider;
-use store_api::metadata::ColumnMetadata;
-use store_api::metric_engine_consts::{ALTER_PHYSICAL_EXTENSION_KEY, MANIFEST_INFO_EXTENSION_KEY};
+use store_api::metric_engine_consts::MANIFEST_INFO_EXTENSION_KEY;
 use store_api::region_engine::RegionManifestInfo;
-use store_api::storage::consts::ReservedColumnId;
 use store_api::storage::RegionId;
 use tokio::sync::mpsc;
 
 use crate::ddl::create_logical_tables::CreateLogicalTablesProcedure;
 use crate::ddl::test_util::datanode_handler::{DatanodeWatcher, NaiveDatanodeHandler};
 use crate::ddl::test_util::{
-    assert_column_name, create_physical_table_metadata, get_raw_table_info, test_column_metadatas,
-    test_create_logical_table_task, test_create_physical_table_task,
+    create_physical_table_metadata, test_create_logical_table_task, test_create_physical_table_task,
 };
 use crate::ddl::TableMetadata;
 use crate::error::{Error, Result};
@@ -42,54 +39,6 @@ use crate::key::table_route::{PhysicalTableRouteValue, TableRouteValue};
 use crate::rpc::router::{Region, RegionRoute};
 use crate::test_util::{new_ddl_context, MockDatanodeManager};
 
-fn make_creates_request_handler(
-    column_metadatas: Vec<ColumnMetadata>,
-) -> impl Fn(Peer, RegionRequest) -> Result<RegionResponse> {
-    move |_peer, request| {
-        let _ = _peer;
-        if let region_request::Body::Creates(_) = request.body.unwrap() {
-            let mut response = RegionResponse::new(0);
-            // Default region id for physical table.
-            let region_id = RegionId::new(1024, 1);
-            response.extensions.insert(
-                MANIFEST_INFO_EXTENSION_KEY.to_string(),
-                RegionManifestInfo::encode_list(&[(
-                    region_id,
-                    RegionManifestInfo::metric(1, 0, 2, 0),
-                )])
-                .unwrap(),
-            );
-            response.extensions.insert(
-                ALTER_PHYSICAL_EXTENSION_KEY.to_string(),
-                ColumnMetadata::encode_list(&column_metadatas).unwrap(),
-            );
-            return Ok(response);
-        }
-
-        Ok(RegionResponse::new(0))
-    }
-}
-
-fn assert_creates_request(
-    peer: Peer,
-    request: RegionRequest,
-    expected_peer_id: u64,
-    expected_region_ids: &[RegionId],
-) {
-    assert_eq!(peer.id, expected_peer_id,);
-    let Some(region_request::Body::Creates(req)) = request.body else {
-        unreachable!();
-    };
-    for (i, region_id) in expected_region_ids.iter().enumerate() {
-        assert_eq!(
-            req.requests[i].region_id,
-            *region_id,
-            "actual region id: {}",
-            RegionId::from_u64(req.requests[i].region_id)
-        );
-    }
-}
-
 #[tokio::test]
 async fn test_on_prepare_physical_table_not_found() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
@@ -278,12 +227,7 @@ async fn test_on_prepare_part_logical_tables_exist() {
 
 #[tokio::test]
 async fn test_on_create_metadata() {
-    common_telemetry::init_default_ut_logging();
-    let (tx, mut rx) = mpsc::channel(8);
-    let column_metadatas = test_column_metadatas(&["host", "cpu"]);
-    let datanode_handler =
-        DatanodeWatcher::new(tx).with_handler(make_creates_request_handler(column_metadatas));
-    let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler));
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
@@ -311,7 +255,7 @@ async fn test_on_create_metadata() {
     let mut procedure = CreateLogicalTablesProcedure::new(
         vec![task, yet_another_task],
         physical_table_id,
-        ddl_context.clone(),
+        ddl_context,
     );
     let status = procedure.on_prepare().await.unwrap();
     assert_matches!(
@@ -330,42 +274,11 @@ async fn test_on_create_metadata() {
     let status = procedure.execute(&ctx).await.unwrap();
     let table_ids = status.downcast_output_ref::<Vec<u32>>().unwrap();
     assert_eq!(*table_ids, vec![1025, 1026]);
-
-    let (peer, request) = rx.try_recv().unwrap();
-    rx.try_recv().unwrap_err();
-    assert_creates_request(
-        peer,
-        request,
-        0,
-        &[RegionId::new(1025, 0), RegionId::new(1026, 0)],
-    );
-
-    let table_info = get_raw_table_info(&ddl_context, table_id).await;
-    assert_column_name(
-        &table_info,
-        &["ts", "value", "__table_id", "__tsid", "host", "cpu"],
-    );
-    assert_eq!(
-        table_info.meta.column_ids,
-        vec![
-            0,
-            1,
-            ReservedColumnId::table_id(),
-            ReservedColumnId::tsid(),
-            2,
-            3
-        ]
-    );
 }
 
 #[tokio::test]
 async fn test_on_create_metadata_part_logical_tables_exist() {
-    common_telemetry::init_default_ut_logging();
-    let (tx, mut rx) = mpsc::channel(8);
-    let column_metadatas = test_column_metadatas(&["host", "cpu"]);
-    let datanode_handler =
-        DatanodeWatcher::new(tx).with_handler(make_creates_request_handler(column_metadatas));
-    let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler));
+    let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
@@ -404,7 +317,7 @@ async fn test_on_create_metadata_part_logical_tables_exist() {
     let mut procedure = CreateLogicalTablesProcedure::new(
         vec![task, non_exist_task],
         physical_table_id,
-        ddl_context.clone(),
+        ddl_context,
     );
     let status = procedure.on_prepare().await.unwrap();
     assert_matches!(
@@ -423,27 +336,6 @@ async fn test_on_create_metadata_part_logical_tables_exist() {
     let status = procedure.execute(&ctx).await.unwrap();
     let table_ids = status.downcast_output_ref::<Vec<u32>>().unwrap();
     assert_eq!(*table_ids, vec![8192, 1025]);
-
-    let (peer, request) = rx.try_recv().unwrap();
-    rx.try_recv().unwrap_err();
-    assert_creates_request(peer, request, 0, &[RegionId::new(1025, 0)]);
-
-    let table_info = get_raw_table_info(&ddl_context, table_id).await;
-    assert_column_name(
-        &table_info,
-        &["ts", "value", "__table_id", "__tsid", "host", "cpu"],
-    );
-    assert_eq!(
-        table_info.meta.column_ids,
-        vec![
-            0,
-            1,
-            ReservedColumnId::table_id(),
-            ReservedColumnId::tsid(),
-            2,
-            3
-        ]
-    );
 }
 
 #[tokio::test]
@@ -507,13 +399,27 @@ async fn test_on_create_metadata_err() {
     assert!(!error.is_retry_later());
 }
 
+fn creates_request_handler(_peer: Peer, request: RegionRequest) -> Result<RegionResponse> {
+    if let region_request::Body::Creates(_) = request.body.unwrap() {
+        let mut response = RegionResponse::new(0);
+        // Default region id for physical table.
+        let region_id = RegionId::new(1024, 1);
+        response.extensions.insert(
+            MANIFEST_INFO_EXTENSION_KEY.to_string(),
+            RegionManifestInfo::encode_list(&[(region_id, RegionManifestInfo::metric(1, 0, 2, 0))])
+                .unwrap(),
+        );
+        return Ok(response);
+    }
+
+    Ok(RegionResponse::new(0))
+}
+
 #[tokio::test]
 async fn test_on_submit_create_request() {
     common_telemetry::init_default_ut_logging();
     let (tx, mut rx) = mpsc::channel(8);
-    let column_metadatas = test_column_metadatas(&["host", "cpu"]);
-    let handler =
-        DatanodeWatcher::new(tx).with_handler(make_creates_request_handler(column_metadatas));
+    let handler = DatanodeWatcher::new(tx).with_handler(creates_request_handler);
     let node_manager = Arc::new(MockDatanodeManager::new(handler));
     let ddl_context = new_ddl_context(node_manager);
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");