feat: support exact filter on time index column (#5671)

* feat: add predicate group

* feat: pass predicate group

* feat: memtable prune by time filters

* test: test PruneTimeIterator with time filters

* feat: push down returns exact for timestamp simple filters

---------

Co-authored-by: Ruihang Xia <waynestxia@gmail.com>

.github/actions/build-dev-builder-images/action.yml

@@ -41,7 +41,14 @@ runs:
         username: ${{ inputs.dockerhub-image-registry-username }}
         password: ${{ inputs.dockerhub-image-registry-token }}
 
-    - name: Build and push dev-builder-ubuntu image
+    - name: Set up qemu for multi-platform builds
+      uses: docker/setup-qemu-action@v3
+      with:
+        platforms: linux/amd64,linux/arm64
+        # The latest version will lead to segmentation fault.
+        image: tonistiigi/binfmt:qemu-v7.0.0-28
+
+    - name: Build and push dev-builder-ubuntu image # Build image for amd64 and arm64 platform.
       shell: bash
       if: ${{ inputs.build-dev-builder-ubuntu == 'true' }}
       run: |
@@ -52,7 +59,7 @@ runs:
           IMAGE_NAMESPACE=${{ inputs.dockerhub-image-namespace }} \
           DEV_BUILDER_IMAGE_TAG=${{ inputs.version }}
 
-    - name: Build and push dev-builder-centos image
+    - name: Build and push dev-builder-centos image # Only build image for amd64 platform.
       shell: bash
       if: ${{ inputs.build-dev-builder-centos == 'true' }}
       run: |
@@ -69,8 +76,7 @@ runs:
       run: |
         make dev-builder \
           BASE_IMAGE=android \
+          BUILDX_MULTI_PLATFORM_BUILD=amd64 \
           IMAGE_REGISTRY=${{ inputs.dockerhub-image-registry }} \
           IMAGE_NAMESPACE=${{ inputs.dockerhub-image-namespace }} \
-          DEV_BUILDER_IMAGE_TAG=${{ inputs.version }} && \
-
-        docker push ${{ inputs.dockerhub-image-registry }}/${{ inputs.dockerhub-image-namespace }}/dev-builder-android:${{ inputs.version }}
+          DEV_BUILDER_IMAGE_TAG=${{ inputs.version }}

.github/actions/start-runner/action.yml

@@ -56,7 +56,7 @@ runs:
 
     - name: Start EC2 runner
       if: startsWith(inputs.runner, 'ec2')
-      uses: machulav/ec2-github-runner@v2
+      uses: machulav/ec2-github-runner@v2.3.8
       id: start-linux-arm64-ec2-runner
       with:
         mode: start

.github/actions/stop-runner/action.yml

@@ -33,7 +33,7 @@ runs:
 
     - name: Stop EC2 runner
       if: ${{ inputs.label && inputs.ec2-instance-id }}
-      uses: machulav/ec2-github-runner@v2
+      uses: machulav/ec2-github-runner@v2.3.8
       with:
         mode: stop
         label: ${{ inputs.label }}

.github/workflows/apidoc.yml

@@ -14,7 +14,7 @@ name: Build API docs
 
 jobs:
   apidoc:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v4
       with:

.github/workflows/dev-build.yml

@@ -16,11 +16,11 @@ on:
         description: The runner uses to build linux-amd64 artifacts
         default: ec2-c6i.4xlarge-amd64
         options:
-          - ubuntu-20.04
-          - ubuntu-20.04-8-cores
-          - ubuntu-20.04-16-cores
-          - ubuntu-20.04-32-cores
-          - ubuntu-20.04-64-cores
+          - ubuntu-22.04
+          - ubuntu-22.04-8-cores
+          - ubuntu-22.04-16-cores
+          - ubuntu-22.04-32-cores
+          - ubuntu-22.04-64-cores
           - ec2-c6i.xlarge-amd64 # 4C8G
           - ec2-c6i.2xlarge-amd64 # 8C16G
           - ec2-c6i.4xlarge-amd64 # 16C32G
@@ -83,7 +83,7 @@ jobs:
   allocate-runners:
     name: Allocate runners
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     outputs:
       linux-amd64-runner: ${{ steps.start-linux-amd64-runner.outputs.label }}
       linux-arm64-runner: ${{ steps.start-linux-arm64-runner.outputs.label }}
@@ -218,7 +218,7 @@ jobs:
       build-linux-amd64-artifacts,
       build-linux-arm64-artifacts,
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     outputs:
       build-result: ${{ steps.set-build-result.outputs.build-result }}
     steps:
@@ -251,7 +251,7 @@ jobs:
       allocate-runners,
       release-images-to-dockerhub,
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     continue-on-error: true
     steps:
       - uses: actions/checkout@v4
@@ -283,7 +283,7 @@ jobs:
     name: Stop linux-amd64 runner
     # Only run this job when the runner is allocated.
     if: ${{ always() }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     needs: [
       allocate-runners,
       build-linux-amd64-artifacts,
@@ -309,7 +309,7 @@ jobs:
     name: Stop linux-arm64 runner
     # Only run this job when the runner is allocated.
     if: ${{ always() }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     needs: [
       allocate-runners,
       build-linux-arm64-artifacts,
@@ -337,7 +337,7 @@ jobs:
     needs: [
       release-images-to-dockerhub
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     permissions:
       issues: write
 

.github/workflows/develop.yml

@@ -23,7 +23,7 @@ concurrency:
 jobs:
   check-typos-and-docs:
     name: Check typos and docs
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
         with:
@@ -36,7 +36,7 @@ jobs:
           || (echo "'config/config.md' is not up-to-date, please run 'make config-docs'." && exit 1)
 
   license-header-check:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     name: Check License Header
     steps:
       - uses: actions/checkout@v4
@@ -49,7 +49,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ ubuntu-20.04 ]
+        os: [ ubuntu-latest ]
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
@@ -72,7 +72,7 @@ jobs:
 
   toml:
     name: Toml Check
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
@@ -89,7 +89,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ ubuntu-20.04 ]
+        os: [ ubuntu-latest ]
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
@@ -248,7 +248,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ ubuntu-20.04 ]
+        os: [ ubuntu-latest ]
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
@@ -568,7 +568,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ ubuntu-20.04 ]
+        os: [ ubuntu-latest ]
         mode:
           - name: "Basic"
             opts: ""
@@ -607,7 +607,7 @@ jobs:
 
   fmt:
     name: Rustfmt
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
@@ -624,7 +624,7 @@ jobs:
 
   clippy:
     name: Clippy
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
@@ -710,7 +710,7 @@ jobs:
 
   coverage:
     if: github.event_name == 'merge_group'
-    runs-on: ubuntu-20.04-8-cores
+    runs-on: ubuntu-22.04-8-cores
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
@@ -770,7 +770,7 @@ jobs:
   # compat:
   #   name: Compatibility Test
   #   needs: build
-  #   runs-on: ubuntu-20.04
+  #   runs-on: ubuntu-22.04
   #   timeout-minutes: 60
   #   steps:
   #     - uses: actions/checkout@v4

.github/workflows/docbot.yml

@@ -9,7 +9,7 @@ concurrency:
 
 jobs:
   docbot:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     permissions:
       pull-requests: write
       contents: read

.github/workflows/docs.yml

@@ -31,7 +31,7 @@ name: CI
 jobs:
   typos:
     name: Spell Check with Typos
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
         with:
@@ -39,7 +39,7 @@ jobs:
       - uses: crate-ci/typos@master
 
   license-header-check:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     name: Check License Header
     steps:
       - uses: actions/checkout@v4
@@ -49,29 +49,29 @@ jobs:
 
   check:
     name: Check
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - run: 'echo "No action required"'
 
   fmt:
     name: Rustfmt
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - run: 'echo "No action required"'
 
   clippy:
     name: Clippy
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - run: 'echo "No action required"'
 
   coverage:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - run: 'echo "No action required"'
 
   test:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - run: 'echo "No action required"'
 
@@ -80,7 +80,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ ubuntu-20.04 ]
+        os: [ ubuntu-latest ]
         mode:
           - name: "Basic"
           - name: "Remote WAL"

.github/workflows/nightly-build.yml

@@ -14,11 +14,11 @@ on:
         description: The runner uses to build linux-amd64 artifacts
         default: ec2-c6i.4xlarge-amd64
         options:
-          - ubuntu-20.04
-          - ubuntu-20.04-8-cores
-          - ubuntu-20.04-16-cores
-          - ubuntu-20.04-32-cores
-          - ubuntu-20.04-64-cores
+          - ubuntu-22.04
+          - ubuntu-22.04-8-cores
+          - ubuntu-22.04-16-cores
+          - ubuntu-22.04-32-cores
+          - ubuntu-22.04-64-cores
           - ec2-c6i.xlarge-amd64 # 4C8G
           - ec2-c6i.2xlarge-amd64 # 8C16G
           - ec2-c6i.4xlarge-amd64 # 16C32G
@@ -70,7 +70,7 @@ jobs:
   allocate-runners:
     name: Allocate runners
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     outputs:
       linux-amd64-runner: ${{ steps.start-linux-amd64-runner.outputs.label }}
       linux-arm64-runner: ${{ steps.start-linux-arm64-runner.outputs.label }}
@@ -182,7 +182,7 @@ jobs:
       build-linux-amd64-artifacts,
       build-linux-arm64-artifacts,
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     outputs:
       nightly-build-result: ${{ steps.set-nightly-build-result.outputs.nightly-build-result }}
     steps:
@@ -214,7 +214,7 @@ jobs:
       allocate-runners,
       release-images-to-dockerhub,
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     # When we push to ACR, it's easy to fail due to some unknown network issues.
     # However, we don't want to fail the whole workflow because of this.
     # The ACR have daily sync with DockerHub, so don't worry about the image not being updated.
@@ -249,7 +249,7 @@ jobs:
     name: Stop linux-amd64 runner
     # Only run this job when the runner is allocated.
     if: ${{ always() }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     needs: [
       allocate-runners,
       build-linux-amd64-artifacts,
@@ -275,7 +275,7 @@ jobs:
     name: Stop linux-arm64 runner
     # Only run this job when the runner is allocated.
     if: ${{ always() }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     needs: [
       allocate-runners,
       build-linux-arm64-artifacts,
@@ -303,7 +303,7 @@ jobs:
     needs: [
       release-images-to-dockerhub
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     permissions:
       issues: write
     env:

.github/workflows/nightly-ci.yml

@@ -13,7 +13,7 @@ jobs:
   sqlness-test:
     name: Run sqlness test
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-latest
     steps:
       - name: Checkout
         uses: actions/checkout@v4
@@ -133,7 +133,7 @@ jobs:
     name: Check status
     needs: [sqlness-test, sqlness-windows, test-on-windows]
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     outputs:
       check-result: ${{ steps.set-check-result.outputs.check-result }}
     steps:
@@ -146,7 +146,7 @@ jobs:
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' && always() }} # Not requiring successful dependent jobs, always run.
     name: Send notification to Greptime team
     needs: [check-status]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     env:
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL_DEVELOP_CHANNEL }}
     steps:

.github/workflows/release-dev-builder-images.yaml

@@ -29,7 +29,7 @@ jobs:
   release-dev-builder-images:
     name: Release dev builder images
     if: ${{ inputs.release_dev_builder_ubuntu_image || inputs.release_dev_builder_centos_image || inputs.release_dev_builder_android_image }} # Only manually trigger this job.
-    runs-on: ubuntu-22.04-16-cores
+    runs-on: ubuntu-latest
     outputs:
       version: ${{ steps.set-version.outputs.version }}
     steps:
@@ -63,7 +63,7 @@ jobs:
 
   release-dev-builder-images-ecr:
     name: Release dev builder images to AWS ECR
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-latest
     needs: [
       release-dev-builder-images
     ]
@@ -148,7 +148,7 @@ jobs:
 
   release-dev-builder-images-cn: # Note: Be careful issue: https://github.com/containers/skopeo/issues/1874 and we decide to use the latest stable skopeo container.
     name: Release dev builder images to CN region
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-latest
     needs: [
       release-dev-builder-images
     ]

.github/workflows/release.yml

@@ -18,11 +18,11 @@ on:
         description: The runner uses to build linux-amd64 artifacts
         default: ec2-c6i.4xlarge-amd64
         options:
-          - ubuntu-20.04
-          - ubuntu-20.04-8-cores
-          - ubuntu-20.04-16-cores
-          - ubuntu-20.04-32-cores
-          - ubuntu-20.04-64-cores
+          - ubuntu-22.04
+          - ubuntu-22.04-8-cores
+          - ubuntu-22.04-16-cores
+          - ubuntu-22.04-32-cores
+          - ubuntu-22.04-64-cores
           - ec2-c6i.xlarge-amd64 # 4C8G
           - ec2-c6i.2xlarge-amd64 # 8C16G
           - ec2-c6i.4xlarge-amd64 # 16C32G
@@ -97,7 +97,7 @@ jobs:
   allocate-runners:
     name: Allocate runners
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     outputs:
       linux-amd64-runner: ${{ steps.start-linux-amd64-runner.outputs.label }}
       linux-arm64-runner: ${{ steps.start-linux-arm64-runner.outputs.label }}
@@ -299,7 +299,7 @@ jobs:
       build-linux-amd64-artifacts,
       build-linux-arm64-artifacts,
     ]
-    runs-on: ubuntu-2004-16-cores
+    runs-on: ubuntu-latest
     outputs:
       build-image-result: ${{ steps.set-build-image-result.outputs.build-image-result }}
     steps:
@@ -335,7 +335,7 @@ jobs:
       build-windows-artifacts,
       release-images-to-dockerhub,
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     # When we push to ACR, it's easy to fail due to some unknown network issues.
     # However, we don't want to fail the whole workflow because of this.
     # The ACR have daily sync with DockerHub, so don't worry about the image not being updated.
@@ -377,7 +377,7 @@ jobs:
       build-windows-artifacts,
       release-images-to-dockerhub,
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
         with:
@@ -396,7 +396,7 @@ jobs:
     name: Stop linux-amd64 runner
     # Only run this job when the runner is allocated.
     if: ${{ always() }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     needs: [
       allocate-runners,
       build-linux-amd64-artifacts,
@@ -422,7 +422,7 @@ jobs:
     name: Stop linux-arm64 runner
     # Only run this job when the runner is allocated.
     if: ${{ always() }}
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     needs: [
       allocate-runners,
       build-linux-arm64-artifacts,
@@ -448,7 +448,7 @@ jobs:
     name: Bump doc version
     if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
     needs: [allocate-runners]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     # Permission reference: https://docs.github.com/en/actions/using-jobs/assigning-permissions-to-jobs
     permissions:
       issues: write # Allows the action to create issues for cyborg.
@@ -475,7 +475,7 @@ jobs:
       build-macos-artifacts,
       build-windows-artifacts,
     ]
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     # Permission reference: https://docs.github.com/en/actions/using-jobs/assigning-permissions-to-jobs
     permissions:
       issues: write # Allows the action to create issues for cyborg.

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

@@ -13,7 +13,7 @@ concurrency:
 
 jobs:
   check:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     timeout-minutes: 10
     steps:
       - uses: actions/checkout@v4

Cargo.lock

@@ -4701,7 +4701,7 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=072ce580502e015df1a6b03a185b60309a7c2a7a#072ce580502e015df1a6b03a185b60309a7c2a7a"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?rev=d92c9ac4e90ef4abdcf5c2eaf5a164e18ba09486#d92c9ac4e90ef4abdcf5c2eaf5a164e18ba09486"
 dependencies = [
  "prost 0.13.3",
  "serde",

Cargo.toml

@@ -129,7 +129,7 @@ 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 = "072ce580502e015df1a6b03a185b60309a7c2a7a" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "d92c9ac4e90ef4abdcf5c2eaf5a164e18ba09486" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"

Makefile

@@ -8,7 +8,7 @@ CARGO_BUILD_OPTS := --locked
 IMAGE_REGISTRY ?= docker.io
 IMAGE_NAMESPACE ?= greptime
 IMAGE_TAG ?= latest
-DEV_BUILDER_IMAGE_TAG ?= 2024-12-25-9d0fa5d5-20250124085746
+DEV_BUILDER_IMAGE_TAG ?= 2024-12-25-a71b93dd-20250305072908
 BUILDX_MULTI_PLATFORM_BUILD ?= false
 BUILDX_BUILDER_NAME ?= gtbuilder
 BASE_IMAGE ?= ubuntu
@@ -60,6 +60,8 @@ ifeq ($(BUILDX_MULTI_PLATFORM_BUILD), all)
 	BUILDX_MULTI_PLATFORM_BUILD_OPTS := --platform linux/amd64,linux/arm64 --push
 else ifeq ($(BUILDX_MULTI_PLATFORM_BUILD), amd64)
 	BUILDX_MULTI_PLATFORM_BUILD_OPTS := --platform linux/amd64 --push
+else ifeq ($(BUILDX_MULTI_PLATFORM_BUILD), arm64)
+	BUILDX_MULTI_PLATFORM_BUILD_OPTS := --platform linux/arm64 --push
 else
 	BUILDX_MULTI_PLATFORM_BUILD_OPTS := -o type=docker
 endif

docker/buildx/ubuntu/Dockerfile

@@ -1,4 +1,4 @@
-FROM ubuntu:20.04 as builder
+FROM ubuntu:22.04 as builder
 
 ARG CARGO_PROFILE
 ARG FEATURES

docker/ci/ubuntu/Dockerfile.fuzztests

@@ -1,4 +1,4 @@
-FROM ubuntu:22.04
+FROM ubuntu:latest
 
 # The binary name of GreptimeDB executable.
 # Defaults to "greptime", but sometimes in other projects it might be different.

docker/dev-builder/ubuntu/Dockerfile-20.04

@@ -41,7 +41,7 @@ RUN mv protoc3/include/* /usr/local/include/
 # and the repositories are pulled from trusted sources (still us, of course). Doing so does not violate the intention
 # of the Git's addition to the "safe.directory" at the first place (see the commit message here:
 # https://github.com/git/git/commit/8959555cee7ec045958f9b6dd62e541affb7e7d9).
-# There's also another solution to this, that we add the desired submodules to the safe directory, instead of using
+# There's also another solution to this, that we add the desired submodules to the safe directory, instead of using 
 # wildcard here. However, that requires the git's config files and the submodules all owned by the very same user.
 # It's troublesome to do this since the dev build runs in Docker, which is under user "root"; while outside the Docker,
 # it can be a different user that have prepared the submodules.

src/catalog/src/kvbackend/manager.rs

@@ -38,6 +38,7 @@ use partition::manager::{PartitionRuleManager, PartitionRuleManagerRef};
 use session::context::{Channel, QueryContext};
 use snafu::prelude::*;
 use table::dist_table::DistTable;
+use table::metadata::TableId;
 use table::table::numbers::{NumbersTable, NUMBERS_TABLE_NAME};
 use table::table_name::TableName;
 use table::TableRef;
@@ -286,6 +287,28 @@ impl CatalogManager for KvBackendCatalogManager {
         return Ok(None);
     }
 
+    async fn tables_by_ids(
+        &self,
+        catalog: &str,
+        schema: &str,
+        table_ids: &[TableId],
+    ) -> Result<Vec<TableRef>> {
+        let table_info_values = self
+            .table_metadata_manager
+            .table_info_manager()
+            .batch_get(table_ids)
+            .await
+            .context(TableMetadataManagerSnafu)?;
+
+        let tables = table_info_values
+            .into_values()
+            .filter(|t| t.table_info.catalog_name == catalog && t.table_info.schema_name == schema)
+            .map(build_table)
+            .collect::<Result<Vec<_>>>()?;
+
+        Ok(tables)
+    }
+
     fn tables<'a>(
         &'a self,
         catalog: &'a str,

src/catalog/src/lib.rs

@@ -87,6 +87,14 @@ pub trait CatalogManager: Send + Sync {
         query_ctx: Option<&QueryContext>,
     ) -> Result<Option<TableRef>>;
 
+    /// Returns the tables by table ids.
+    async fn tables_by_ids(
+        &self,
+        catalog: &str,
+        schema: &str,
+        table_ids: &[TableId],
+    ) -> Result<Vec<TableRef>>;
+
     /// Returns all tables with a stream by catalog and schema.
     fn tables<'a>(
         &'a self,

src/catalog/src/memory/manager.rs

@@ -14,7 +14,7 @@
 
 use std::any::Any;
 use std::collections::hash_map::Entry;
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::sync::{Arc, RwLock, Weak};
 
 use async_stream::{stream, try_stream};
@@ -28,6 +28,7 @@ use common_meta::kv_backend::memory::MemoryKvBackend;
 use futures_util::stream::BoxStream;
 use session::context::QueryContext;
 use snafu::OptionExt;
+use table::metadata::TableId;
 use table::TableRef;
 
 use crate::error::{CatalogNotFoundSnafu, Result, SchemaNotFoundSnafu, TableExistsSnafu};
@@ -143,6 +144,33 @@ impl CatalogManager for MemoryCatalogManager {
         Ok(result)
     }
 
+    async fn tables_by_ids(
+        &self,
+        catalog: &str,
+        schema: &str,
+        table_ids: &[TableId],
+    ) -> Result<Vec<TableRef>> {
+        let catalogs = self.catalogs.read().unwrap();
+
+        let schemas = catalogs.get(catalog).context(CatalogNotFoundSnafu {
+            catalog_name: catalog,
+        })?;
+
+        let tables = schemas
+            .get(schema)
+            .context(SchemaNotFoundSnafu { catalog, schema })?;
+
+        let filter_ids: HashSet<_> = table_ids.iter().collect();
+        // It is very inefficient, but we do not need to optimize it since it will not be called in `MemoryCatalogManager`.
+        let tables = tables
+            .values()
+            .filter(|t| filter_ids.contains(&t.table_info().table_id()))
+            .cloned()
+            .collect::<Vec<_>>();
+
+        Ok(tables)
+    }
+
     fn tables<'a>(
         &'a self,
         catalog: &'a str,

src/cmd/src/datanode.rs

@@ -287,7 +287,6 @@ impl StartCommand {
             .await
             .context(StartDatanodeSnafu)?;
 
-        let cluster_id = 0; // TODO(hl): read from config
         let member_id = opts
             .node_id
             .context(MissingConfigSnafu { msg: "'node_id'" })?;
@@ -296,13 +295,10 @@ impl StartCommand {
             msg: "'meta_client_options'",
         })?;
 
-        let meta_client = meta_client::create_meta_client(
-            cluster_id,
-            MetaClientType::Datanode { member_id },
-            meta_config,
-        )
-        .await
-        .context(MetaClientInitSnafu)?;
+        let meta_client =
+            meta_client::create_meta_client(MetaClientType::Datanode { member_id }, meta_config)
+                .await
+                .context(MetaClientInitSnafu)?;
 
         let meta_backend = Arc::new(MetaKvBackend {
             client: meta_client.clone(),

src/cmd/src/flownode.rs

@@ -241,9 +241,6 @@ impl StartCommand {
         let mut opts = opts.component;
         opts.grpc.detect_server_addr();
 
-        // TODO(discord9): make it not optionale after cluster id is required
-        let cluster_id = opts.cluster_id.unwrap_or(0);
-
         let member_id = opts
             .node_id
             .context(MissingConfigSnafu { msg: "'node_id'" })?;
@@ -252,13 +249,10 @@ impl StartCommand {
             msg: "'meta_client_options'",
         })?;
 
-        let meta_client = meta_client::create_meta_client(
-            cluster_id,
-            MetaClientType::Flownode { member_id },
-            meta_config,
-        )
-        .await
-        .context(MetaClientInitSnafu)?;
+        let meta_client =
+            meta_client::create_meta_client(MetaClientType::Flownode { member_id }, meta_config)
+                .await
+                .context(MetaClientInitSnafu)?;
 
         let cache_max_capacity = meta_config.metadata_cache_max_capacity;
         let cache_ttl = meta_config.metadata_cache_ttl;

src/cmd/src/frontend.rs

@@ -295,14 +295,10 @@ impl StartCommand {
         let cache_ttl = meta_client_options.metadata_cache_ttl;
         let cache_tti = meta_client_options.metadata_cache_tti;
 
-        let cluster_id = 0; // (TODO: jeremy): It is currently a reserved field and has not been enabled.
-        let meta_client = meta_client::create_meta_client(
-            cluster_id,
-            MetaClientType::Frontend,
-            meta_client_options,
-        )
-        .await
-        .context(MetaClientInitSnafu)?;
+        let meta_client =
+            meta_client::create_meta_client(MetaClientType::Frontend, meta_client_options)
+                .await
+                .context(MetaClientInitSnafu)?;
 
         // TODO(discord9): add helper function to ease the creation of cache registry&such
         let cached_meta_backend =

src/common/catalog/src/consts.rs

@@ -130,3 +130,10 @@ pub const SEMANTIC_TYPE_TIME_INDEX: &str = "TIMESTAMP";
 pub fn is_readonly_schema(schema: &str) -> bool {
     matches!(schema, INFORMATION_SCHEMA_NAME)
 }
+
+// ---- special table and fields ----
+pub const TRACE_ID_COLUMN: &str = "trace_id";
+pub const SPAN_ID_COLUMN: &str = "span_id";
+pub const SPAN_NAME_COLUMN: &str = "span_name";
+pub const PARENT_SPAN_ID_COLUMN: &str = "parent_span_id";
+// ---- End of special table and fields ----

src/common/function/src/aggr/hll.rs

@@ -12,6 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//! Two UDAFs are implemented for HyperLogLog:
+//!
+//! - `hll`: Accepts a string column and aggregates the values into a
+//!   HyperLogLog state.
+//! - `hll_merge`: Accepts a binary column of states generated by `hll`
+//!   and merges them into a single state.
+//!
+//! The states can be then used to estimate the cardinality of the
+//! values in the column by `hll_count` UDF.
+
 use std::sync::Arc;
 
 use common_query::prelude::*;

src/common/function/src/aggr/uddsketch_state.rs

@@ -12,6 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+//! Implementation of the `uddsketch_state` UDAF that generate the state of
+//! UDDSketch for a given set of values.
+//!
+//! The generated state can be used to compute approximate quantiles using
+//! `uddsketch_calc` UDF.
+
 use std::sync::Arc;
 
 use common_query::prelude::*;

src/common/function/src/scalars/aggregate.rs

@@ -12,24 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-mod argmax;
-mod argmin;
-mod diff;
-mod mean;
-mod polyval;
-mod scipy_stats_norm_cdf;
-mod scipy_stats_norm_pdf;
+//! # Deprecate Warning:
+//!
+//! This module is deprecated and will be removed in the future.
+//! All UDAF implementation here are not maintained and should
+//! not be used before they are refactored into the `src/aggr`
+//! version.
 
 use std::sync::Arc;
 
-pub use argmax::ArgmaxAccumulatorCreator;
-pub use argmin::ArgminAccumulatorCreator;
 use common_query::logical_plan::AggregateFunctionCreatorRef;
-pub use diff::DiffAccumulatorCreator;
-pub use mean::MeanAccumulatorCreator;
-pub use polyval::PolyvalAccumulatorCreator;
-pub use scipy_stats_norm_cdf::ScipyStatsNormCdfAccumulatorCreator;
-pub use scipy_stats_norm_pdf::ScipyStatsNormPdfAccumulatorCreator;
 
 use crate::function_registry::FunctionRegistry;
 use crate::scalars::vector::product::VectorProductCreator;
@@ -76,31 +68,22 @@ pub(crate) struct AggregateFunctions;
 
 impl AggregateFunctions {
     pub fn register(registry: &FunctionRegistry) {
-        macro_rules! register_aggr_func {
-            ($name :expr, $arg_count :expr, $creator :ty) => {
-                registry.register_aggregate_function(Arc::new(AggregateFunctionMeta::new(
-                    $name,
-                    $arg_count,
-                    Arc::new(|| Arc::new(<$creator>::default())),
-                )));
-            };
-        }
-
-        register_aggr_func!("diff", 1, DiffAccumulatorCreator);
-        register_aggr_func!("mean", 1, MeanAccumulatorCreator);
-        register_aggr_func!("polyval", 2, PolyvalAccumulatorCreator);
-        register_aggr_func!("argmax", 1, ArgmaxAccumulatorCreator);
-        register_aggr_func!("argmin", 1, ArgminAccumulatorCreator);
-        register_aggr_func!("scipystatsnormcdf", 2, ScipyStatsNormCdfAccumulatorCreator);
-        register_aggr_func!("scipystatsnormpdf", 2, ScipyStatsNormPdfAccumulatorCreator);
-        register_aggr_func!("vec_sum", 1, VectorSumCreator);
-        register_aggr_func!("vec_product", 1, VectorProductCreator);
+        registry.register_aggregate_function(Arc::new(AggregateFunctionMeta::new(
+            "vec_sum",
+            1,
+            Arc::new(|| Arc::new(VectorSumCreator::default())),
+        )));
+        registry.register_aggregate_function(Arc::new(AggregateFunctionMeta::new(
+            "vec_product",
+            1,
+            Arc::new(|| Arc::new(VectorProductCreator::default())),
+        )));
 
         #[cfg(feature = "geo")]
-        register_aggr_func!(
+        registry.register_aggregate_function(Arc::new(AggregateFunctionMeta::new(
             "json_encode_path",
             3,
-            super::geo::encoding::JsonPathEncodeFunctionCreator
-        );
+            Arc::new(|| Arc::new(super::geo::encoding::JsonPathEncodeFunctionCreator::default())),
+        )));
     }
 }

src/common/function/src/scalars/aggregate/argmax.rs

@@ -1,208 +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::cmp::Ordering;
-use std::sync::Arc;
-
-use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
-use common_query::error::{
-    BadAccumulatorImplSnafu, CreateAccumulatorSnafu, InvalidInputStateSnafu, Result,
-};
-use common_query::logical_plan::accumulator::AggrFuncTypeStore;
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
-use common_query::prelude::*;
-use datatypes::prelude::*;
-use datatypes::types::{LogicalPrimitiveType, WrapperType};
-use datatypes::vectors::{ConstantVector, Helper};
-use datatypes::with_match_primitive_type_id;
-use snafu::ensure;
-
-// https://numpy.org/doc/stable/reference/generated/numpy.argmax.html
-// return the index of the max value
-#[derive(Debug, Default)]
-pub struct Argmax<T> {
-    max: Option<T>,
-    n: u64,
-}
-
-impl<T> Argmax<T>
-where
-    T: PartialOrd + Copy,
-{
-    fn update(&mut self, value: T, index: u64) {
-        if let Some(Ordering::Less) = self.max.partial_cmp(&Some(value)) {
-            self.max = Some(value);
-            self.n = index;
-        }
-    }
-}
-
-impl<T> Accumulator for Argmax<T>
-where
-    T: WrapperType + PartialOrd,
-{
-    fn state(&self) -> Result<Vec<Value>> {
-        match self.max {
-            Some(max) => Ok(vec![max.into(), self.n.into()]),
-            _ => Ok(vec![Value::Null, self.n.into()]),
-        }
-    }
-
-    fn update_batch(&mut self, values: &[VectorRef]) -> Result<()> {
-        if values.is_empty() {
-            return Ok(());
-        }
-
-        let column = &values[0];
-        let column: &<T as Scalar>::VectorType = if column.is_const() {
-            let column: &ConstantVector = unsafe { Helper::static_cast(column) };
-            unsafe { Helper::static_cast(column.inner()) }
-        } else {
-            unsafe { Helper::static_cast(column) }
-        };
-        for (i, v) in column.iter_data().enumerate() {
-            if let Some(value) = v {
-                self.update(value, i as u64);
-            }
-        }
-        Ok(())
-    }
-
-    fn merge_batch(&mut self, states: &[VectorRef]) -> Result<()> {
-        if states.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(
-            states.len() == 2,
-            BadAccumulatorImplSnafu {
-                err_msg: "expect 2 states in `merge_batch`",
-            }
-        );
-
-        let max = &states[0];
-        let index = &states[1];
-        let max: &<T as Scalar>::VectorType = unsafe { Helper::static_cast(max) };
-        let index: &<u64 as Scalar>::VectorType = unsafe { Helper::static_cast(index) };
-        index
-            .iter_data()
-            .flatten()
-            .zip(max.iter_data().flatten())
-            .for_each(|(i, max)| self.update(max, i));
-        Ok(())
-    }
-
-    fn evaluate(&self) -> Result<Value> {
-        match self.max {
-            Some(_) => Ok(self.n.into()),
-            _ => Ok(Value::Null),
-        }
-    }
-}
-
-#[as_aggr_func_creator]
-#[derive(Debug, Default, AggrFuncTypeStore)]
-pub struct ArgmaxAccumulatorCreator {}
-
-impl AggregateFunctionCreator for ArgmaxAccumulatorCreator {
-    fn creator(&self) -> AccumulatorCreatorFunction {
-        let creator: AccumulatorCreatorFunction = Arc::new(move |types: &[ConcreteDataType]| {
-            let input_type = &types[0];
-            with_match_primitive_type_id!(
-                input_type.logical_type_id(),
-                |$S| {
-                    Ok(Box::new(Argmax::<<$S as LogicalPrimitiveType>::Wrapper>::default()))
-                },
-                {
-                    let err_msg = format!(
-                        "\"ARGMAX\" aggregate function not support data type {:?}",
-                        input_type.logical_type_id(),
-                    );
-                    CreateAccumulatorSnafu { err_msg }.fail()?
-                }
-            )
-        });
-        creator
-    }
-
-    fn output_type(&self) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint64_datatype())
-    }
-
-    fn state_types(&self) -> Result<Vec<ConcreteDataType>> {
-        let input_types = self.input_types()?;
-
-        ensure!(input_types.len() == 1, InvalidInputStateSnafu);
-
-        Ok(vec![
-            input_types.into_iter().next().unwrap(),
-            ConcreteDataType::uint64_datatype(),
-        ])
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use datatypes::vectors::Int32Vector;
-
-    use super::*;
-    #[test]
-    fn test_update_batch() {
-        // test update empty batch, expect not updating anything
-        let mut argmax = Argmax::<i32>::default();
-        argmax.update_batch(&[]).unwrap();
-        assert_eq!(Value::Null, argmax.evaluate().unwrap());
-
-        // test update one not-null value
-        let mut argmax = Argmax::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![Some(42)]))];
-        argmax.update_batch(&v).unwrap();
-        assert_eq!(Value::from(0_u64), argmax.evaluate().unwrap());
-
-        // test update one null value
-        let mut argmax = Argmax::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![Option::<i32>::None]))];
-        argmax.update_batch(&v).unwrap();
-        assert_eq!(Value::Null, argmax.evaluate().unwrap());
-
-        // test update no null-value batch
-        let mut argmax = Argmax::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![
-            Some(-1i32),
-            Some(1),
-            Some(3),
-        ]))];
-        argmax.update_batch(&v).unwrap();
-        assert_eq!(Value::from(2_u64), argmax.evaluate().unwrap());
-
-        // test update null-value batch
-        let mut argmax = Argmax::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![
-            Some(-2i32),
-            None,
-            Some(4),
-        ]))];
-        argmax.update_batch(&v).unwrap();
-        assert_eq!(Value::from(2_u64), argmax.evaluate().unwrap());
-
-        // test update with constant vector
-        let mut argmax = Argmax::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(ConstantVector::new(
-            Arc::new(Int32Vector::from_vec(vec![4])),
-            10,
-        ))];
-        argmax.update_batch(&v).unwrap();
-        assert_eq!(Value::from(0_u64), argmax.evaluate().unwrap());
-    }
-}

src/common/function/src/scalars/aggregate/argmin.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::cmp::Ordering;
-use std::sync::Arc;
-
-use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
-use common_query::error::{
-    BadAccumulatorImplSnafu, CreateAccumulatorSnafu, InvalidInputStateSnafu, Result,
-};
-use common_query::logical_plan::accumulator::AggrFuncTypeStore;
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
-use common_query::prelude::*;
-use datatypes::prelude::*;
-use datatypes::vectors::{ConstantVector, Helper};
-use datatypes::with_match_primitive_type_id;
-use snafu::ensure;
-
-// // https://numpy.org/doc/stable/reference/generated/numpy.argmin.html
-#[derive(Debug, Default)]
-pub struct Argmin<T> {
-    min: Option<T>,
-    n: u32,
-}
-
-impl<T> Argmin<T>
-where
-    T: Copy + PartialOrd,
-{
-    fn update(&mut self, value: T, index: u32) {
-        match self.min {
-            Some(min) => {
-                if let Some(Ordering::Greater) = min.partial_cmp(&value) {
-                    self.min = Some(value);
-                    self.n = index;
-                }
-            }
-            None => {
-                self.min = Some(value);
-                self.n = index;
-            }
-        }
-    }
-}
-
-impl<T> Accumulator for Argmin<T>
-where
-    T: WrapperType + PartialOrd,
-{
-    fn state(&self) -> Result<Vec<Value>> {
-        match self.min {
-            Some(min) => Ok(vec![min.into(), self.n.into()]),
-            _ => Ok(vec![Value::Null, self.n.into()]),
-        }
-    }
-
-    fn update_batch(&mut self, values: &[VectorRef]) -> Result<()> {
-        if values.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(values.len() == 1, InvalidInputStateSnafu);
-
-        let column = &values[0];
-        let column: &<T as Scalar>::VectorType = if column.is_const() {
-            let column: &ConstantVector = unsafe { Helper::static_cast(column) };
-            unsafe { Helper::static_cast(column.inner()) }
-        } else {
-            unsafe { Helper::static_cast(column) }
-        };
-        for (i, v) in column.iter_data().enumerate() {
-            if let Some(value) = v {
-                self.update(value, i as u32);
-            }
-        }
-        Ok(())
-    }
-
-    fn merge_batch(&mut self, states: &[VectorRef]) -> Result<()> {
-        if states.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(
-            states.len() == 2,
-            BadAccumulatorImplSnafu {
-                err_msg: "expect 2 states in `merge_batch`",
-            }
-        );
-
-        let min = &states[0];
-        let index = &states[1];
-        let min: &<T as Scalar>::VectorType = unsafe { Helper::static_cast(min) };
-        let index: &<u32 as Scalar>::VectorType = unsafe { Helper::static_cast(index) };
-        index
-            .iter_data()
-            .flatten()
-            .zip(min.iter_data().flatten())
-            .for_each(|(i, min)| self.update(min, i));
-        Ok(())
-    }
-
-    fn evaluate(&self) -> Result<Value> {
-        match self.min {
-            Some(_) => Ok(self.n.into()),
-            _ => Ok(Value::Null),
-        }
-    }
-}
-
-#[as_aggr_func_creator]
-#[derive(Debug, Default, AggrFuncTypeStore)]
-pub struct ArgminAccumulatorCreator {}
-
-impl AggregateFunctionCreator for ArgminAccumulatorCreator {
-    fn creator(&self) -> AccumulatorCreatorFunction {
-        let creator: AccumulatorCreatorFunction = Arc::new(move |types: &[ConcreteDataType]| {
-            let input_type = &types[0];
-            with_match_primitive_type_id!(
-                input_type.logical_type_id(),
-                |$S| {
-                    Ok(Box::new(Argmin::<<$S as LogicalPrimitiveType>::Wrapper>::default()))
-                },
-                {
-                    let err_msg = format!(
-                        "\"ARGMIN\" aggregate function not support data type {:?}",
-                        input_type.logical_type_id(),
-                    );
-                    CreateAccumulatorSnafu { err_msg }.fail()?
-                }
-            )
-        });
-        creator
-    }
-
-    fn output_type(&self) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint32_datatype())
-    }
-
-    fn state_types(&self) -> Result<Vec<ConcreteDataType>> {
-        let input_types = self.input_types()?;
-
-        ensure!(input_types.len() == 1, InvalidInputStateSnafu);
-
-        Ok(vec![
-            input_types.into_iter().next().unwrap(),
-            ConcreteDataType::uint32_datatype(),
-        ])
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use datatypes::vectors::Int32Vector;
-
-    use super::*;
-    #[test]
-    fn test_update_batch() {
-        // test update empty batch, expect not updating anything
-        let mut argmin = Argmin::<i32>::default();
-        argmin.update_batch(&[]).unwrap();
-        assert_eq!(Value::Null, argmin.evaluate().unwrap());
-
-        // test update one not-null value
-        let mut argmin = Argmin::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![Some(42)]))];
-        argmin.update_batch(&v).unwrap();
-        assert_eq!(Value::from(0_u32), argmin.evaluate().unwrap());
-
-        // test update one null value
-        let mut argmin = Argmin::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![Option::<i32>::None]))];
-        argmin.update_batch(&v).unwrap();
-        assert_eq!(Value::Null, argmin.evaluate().unwrap());
-
-        // test update no null-value batch
-        let mut argmin = Argmin::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![
-            Some(-1i32),
-            Some(1),
-            Some(3),
-        ]))];
-        argmin.update_batch(&v).unwrap();
-        assert_eq!(Value::from(0_u32), argmin.evaluate().unwrap());
-
-        // test update null-value batch
-        let mut argmin = Argmin::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![
-            Some(-2i32),
-            None,
-            Some(4),
-        ]))];
-        argmin.update_batch(&v).unwrap();
-        assert_eq!(Value::from(0_u32), argmin.evaluate().unwrap());
-
-        // test update with constant vector
-        let mut argmin = Argmin::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(ConstantVector::new(
-            Arc::new(Int32Vector::from_vec(vec![4])),
-            10,
-        ))];
-        argmin.update_batch(&v).unwrap();
-        assert_eq!(Value::from(0_u32), argmin.evaluate().unwrap());
-    }
-}

src/common/function/src/scalars/aggregate/diff.rs

@@ -1,252 +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::marker::PhantomData;
-use std::sync::Arc;
-
-use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
-use common_query::error::{
-    CreateAccumulatorSnafu, DowncastVectorSnafu, FromScalarValueSnafu, InvalidInputStateSnafu,
-    Result,
-};
-use common_query::logical_plan::accumulator::AggrFuncTypeStore;
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
-use common_query::prelude::*;
-use datatypes::prelude::*;
-use datatypes::value::ListValue;
-use datatypes::vectors::{ConstantVector, Helper, ListVector};
-use datatypes::with_match_primitive_type_id;
-use num_traits::AsPrimitive;
-use snafu::{ensure, OptionExt, ResultExt};
-
-// https://numpy.org/doc/stable/reference/generated/numpy.diff.html
-// I is the input type, O is the output type.
-#[derive(Debug, Default)]
-pub struct Diff<I, O> {
-    values: Vec<I>,
-    _phantom: PhantomData<O>,
-}
-
-impl<I, O> Diff<I, O> {
-    fn push(&mut self, value: I) {
-        self.values.push(value);
-    }
-}
-
-impl<I, O> Accumulator for Diff<I, O>
-where
-    I: WrapperType,
-    O: WrapperType,
-    I::Native: AsPrimitive<O::Native>,
-    O::Native: std::ops::Sub<Output = O::Native>,
-{
-    fn state(&self) -> Result<Vec<Value>> {
-        let nums = self
-            .values
-            .iter()
-            .map(|&n| n.into())
-            .collect::<Vec<Value>>();
-        Ok(vec![Value::List(ListValue::new(
-            nums,
-            I::LogicalType::build_data_type(),
-        ))])
-    }
-
-    fn update_batch(&mut self, values: &[VectorRef]) -> Result<()> {
-        if values.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(values.len() == 1, InvalidInputStateSnafu);
-
-        let column = &values[0];
-        let mut len = 1;
-        let column: &<I as Scalar>::VectorType = if column.is_const() {
-            len = column.len();
-            let column: &ConstantVector = unsafe { Helper::static_cast(column) };
-            unsafe { Helper::static_cast(column.inner()) }
-        } else {
-            unsafe { Helper::static_cast(column) }
-        };
-        (0..len).for_each(|_| {
-            for v in column.iter_data().flatten() {
-                self.push(v);
-            }
-        });
-        Ok(())
-    }
-
-    fn merge_batch(&mut self, states: &[VectorRef]) -> Result<()> {
-        if states.is_empty() {
-            return Ok(());
-        }
-
-        let states = &states[0];
-        let states = states
-            .as_any()
-            .downcast_ref::<ListVector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect ListVector, got vector type {}",
-                    states.vector_type_name()
-                ),
-            })?;
-        for state in states.values_iter() {
-            if let Some(state) = state.context(FromScalarValueSnafu)? {
-                self.update_batch(&[state])?;
-            }
-        }
-        Ok(())
-    }
-
-    fn evaluate(&self) -> Result<Value> {
-        if self.values.is_empty() || self.values.len() == 1 {
-            return Ok(Value::Null);
-        }
-        let diff = self
-            .values
-            .windows(2)
-            .map(|x| {
-                let native = x[1].into_native().as_() - x[0].into_native().as_();
-                O::from_native(native).into()
-            })
-            .collect::<Vec<Value>>();
-        let diff = Value::List(ListValue::new(diff, O::LogicalType::build_data_type()));
-        Ok(diff)
-    }
-}
-
-#[as_aggr_func_creator]
-#[derive(Debug, Default, AggrFuncTypeStore)]
-pub struct DiffAccumulatorCreator {}
-
-impl AggregateFunctionCreator for DiffAccumulatorCreator {
-    fn creator(&self) -> AccumulatorCreatorFunction {
-        let creator: AccumulatorCreatorFunction = Arc::new(move |types: &[ConcreteDataType]| {
-            let input_type = &types[0];
-            with_match_primitive_type_id!(
-                input_type.logical_type_id(),
-                |$S| {
-                    Ok(Box::new(Diff::<<$S as LogicalPrimitiveType>::Wrapper, <<$S as LogicalPrimitiveType>::LargestType as LogicalPrimitiveType>::Wrapper>::default()))
-                },
-                {
-                    let err_msg = format!(
-                        "\"DIFF\" aggregate function not support data type {:?}",
-                        input_type.logical_type_id(),
-                    );
-                    CreateAccumulatorSnafu { err_msg }.fail()?
-                }
-            )
-        });
-        creator
-    }
-
-    fn output_type(&self) -> Result<ConcreteDataType> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 1, InvalidInputStateSnafu);
-        with_match_primitive_type_id!(
-            input_types[0].logical_type_id(),
-            |$S| {
-                Ok(ConcreteDataType::list_datatype($S::default().into()))
-            },
-            {
-                unreachable!()
-            }
-        )
-    }
-
-    fn state_types(&self) -> Result<Vec<ConcreteDataType>> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 1, InvalidInputStateSnafu);
-        with_match_primitive_type_id!(
-            input_types[0].logical_type_id(),
-            |$S| {
-                Ok(vec![ConcreteDataType::list_datatype($S::default().into())])
-            },
-            {
-                unreachable!()
-            }
-        )
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use datatypes::vectors::Int32Vector;
-
-    use super::*;
-
-    #[test]
-    fn test_update_batch() {
-        // test update empty batch, expect not updating anything
-        let mut diff = Diff::<i32, i64>::default();
-        diff.update_batch(&[]).unwrap();
-        assert!(diff.values.is_empty());
-        assert_eq!(Value::Null, diff.evaluate().unwrap());
-
-        // test update one not-null value
-        let mut diff = Diff::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![Some(42)]))];
-        diff.update_batch(&v).unwrap();
-        assert_eq!(Value::Null, diff.evaluate().unwrap());
-
-        // test update one null value
-        let mut diff = Diff::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![Option::<i32>::None]))];
-        diff.update_batch(&v).unwrap();
-        assert_eq!(Value::Null, diff.evaluate().unwrap());
-
-        // test update no null-value batch
-        let mut diff = Diff::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![
-            Some(-1i32),
-            Some(1),
-            Some(2),
-        ]))];
-        let values = vec![Value::from(2_i64), Value::from(1_i64)];
-        diff.update_batch(&v).unwrap();
-        assert_eq!(
-            Value::List(ListValue::new(values, ConcreteDataType::int64_datatype())),
-            diff.evaluate().unwrap()
-        );
-
-        // test update null-value batch
-        let mut diff = Diff::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![
-            Some(-2i32),
-            None,
-            Some(3),
-            Some(4),
-        ]))];
-        let values = vec![Value::from(5_i64), Value::from(1_i64)];
-        diff.update_batch(&v).unwrap();
-        assert_eq!(
-            Value::List(ListValue::new(values, ConcreteDataType::int64_datatype())),
-            diff.evaluate().unwrap()
-        );
-
-        // test update with constant vector
-        let mut diff = Diff::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(ConstantVector::new(
-            Arc::new(Int32Vector::from_vec(vec![4])),
-            4,
-        ))];
-        let values = vec![Value::from(0_i64), Value::from(0_i64), Value::from(0_i64)];
-        diff.update_batch(&v).unwrap();
-        assert_eq!(
-            Value::List(ListValue::new(values, ConcreteDataType::int64_datatype())),
-            diff.evaluate().unwrap()
-        );
-    }
-}

src/common/function/src/scalars/aggregate/mean.rs

@@ -1,238 +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::marker::PhantomData;
-use std::sync::Arc;
-
-use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
-use common_query::error::{
-    BadAccumulatorImplSnafu, CreateAccumulatorSnafu, DowncastVectorSnafu, InvalidInputStateSnafu,
-    Result,
-};
-use common_query::logical_plan::accumulator::AggrFuncTypeStore;
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
-use common_query::prelude::*;
-use datatypes::prelude::*;
-use datatypes::types::WrapperType;
-use datatypes::vectors::{ConstantVector, Float64Vector, Helper, UInt64Vector};
-use datatypes::with_match_primitive_type_id;
-use num_traits::AsPrimitive;
-use snafu::{ensure, OptionExt};
-
-#[derive(Debug, Default)]
-pub struct Mean<T> {
-    sum: f64,
-    n: u64,
-    _phantom: PhantomData<T>,
-}
-
-impl<T> Mean<T>
-where
-    T: WrapperType,
-    T::Native: AsPrimitive<f64>,
-{
-    #[inline(always)]
-    fn push(&mut self, value: T) {
-        self.sum += value.into_native().as_();
-        self.n += 1;
-    }
-
-    #[inline(always)]
-    fn update(&mut self, sum: f64, n: u64) {
-        self.sum += sum;
-        self.n += n;
-    }
-}
-
-impl<T> Accumulator for Mean<T>
-where
-    T: WrapperType,
-    T::Native: AsPrimitive<f64>,
-{
-    fn state(&self) -> Result<Vec<Value>> {
-        Ok(vec![self.sum.into(), self.n.into()])
-    }
-
-    fn update_batch(&mut self, values: &[VectorRef]) -> Result<()> {
-        if values.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(values.len() == 1, InvalidInputStateSnafu);
-        let column = &values[0];
-        let mut len = 1;
-        let column: &<T as Scalar>::VectorType = if column.is_const() {
-            len = column.len();
-            let column: &ConstantVector = unsafe { Helper::static_cast(column) };
-            unsafe { Helper::static_cast(column.inner()) }
-        } else {
-            unsafe { Helper::static_cast(column) }
-        };
-        (0..len).for_each(|_| {
-            for v in column.iter_data().flatten() {
-                self.push(v);
-            }
-        });
-
-        Ok(())
-    }
-
-    fn merge_batch(&mut self, states: &[VectorRef]) -> Result<()> {
-        if states.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(
-            states.len() == 2,
-            BadAccumulatorImplSnafu {
-                err_msg: "expect 2 states in `merge_batch`",
-            }
-        );
-
-        let sum = &states[0];
-        let n = &states[1];
-
-        let sum = sum
-            .as_any()
-            .downcast_ref::<Float64Vector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect Float64Vector, got vector type {}",
-                    sum.vector_type_name()
-                ),
-            })?;
-
-        let n = n
-            .as_any()
-            .downcast_ref::<UInt64Vector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect UInt64Vector, got vector type {}",
-                    sum.vector_type_name()
-                ),
-            })?;
-
-        sum.iter_data().zip(n.iter_data()).for_each(|(sum, n)| {
-            if let (Some(sum), Some(n)) = (sum, n) {
-                self.update(sum, n);
-            }
-        });
-        Ok(())
-    }
-
-    fn evaluate(&self) -> Result<Value> {
-        if self.n == 0 {
-            return Ok(Value::Null);
-        }
-        let values = self.sum / self.n as f64;
-        Ok(values.into())
-    }
-}
-
-#[as_aggr_func_creator]
-#[derive(Debug, Default, AggrFuncTypeStore)]
-pub struct MeanAccumulatorCreator {}
-
-impl AggregateFunctionCreator for MeanAccumulatorCreator {
-    fn creator(&self) -> AccumulatorCreatorFunction {
-        let creator: AccumulatorCreatorFunction = Arc::new(move |types: &[ConcreteDataType]| {
-            let input_type = &types[0];
-            with_match_primitive_type_id!(
-                input_type.logical_type_id(),
-                |$S| {
-                    Ok(Box::new(Mean::<<$S as LogicalPrimitiveType>::Native>::default()))
-                },
-                {
-                    let err_msg = format!(
-                        "\"MEAN\" aggregate function not support data type {:?}",
-                        input_type.logical_type_id(),
-                    );
-                    CreateAccumulatorSnafu { err_msg }.fail()?
-                }
-            )
-        });
-        creator
-    }
-
-    fn output_type(&self) -> Result<ConcreteDataType> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 1, InvalidInputStateSnafu);
-        Ok(ConcreteDataType::float64_datatype())
-    }
-
-    fn state_types(&self) -> Result<Vec<ConcreteDataType>> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 1, InvalidInputStateSnafu);
-        Ok(vec![
-            ConcreteDataType::float64_datatype(),
-            ConcreteDataType::uint64_datatype(),
-        ])
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use datatypes::vectors::Int32Vector;
-
-    use super::*;
-    #[test]
-    fn test_update_batch() {
-        // test update empty batch, expect not updating anything
-        let mut mean = Mean::<i32>::default();
-        mean.update_batch(&[]).unwrap();
-        assert_eq!(Value::Null, mean.evaluate().unwrap());
-
-        // test update one not-null value
-        let mut mean = Mean::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![Some(42)]))];
-        mean.update_batch(&v).unwrap();
-        assert_eq!(Value::from(42.0_f64), mean.evaluate().unwrap());
-
-        // test update one null value
-        let mut mean = Mean::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![Option::<i32>::None]))];
-        mean.update_batch(&v).unwrap();
-        assert_eq!(Value::Null, mean.evaluate().unwrap());
-
-        // test update no null-value batch
-        let mut mean = Mean::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![
-            Some(-1i32),
-            Some(1),
-            Some(2),
-        ]))];
-        mean.update_batch(&v).unwrap();
-        assert_eq!(Value::from(0.6666666666666666), mean.evaluate().unwrap());
-
-        // test update null-value batch
-        let mut mean = Mean::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(Int32Vector::from(vec![
-            Some(-2i32),
-            None,
-            Some(3),
-            Some(4),
-        ]))];
-        mean.update_batch(&v).unwrap();
-        assert_eq!(Value::from(1.6666666666666667), mean.evaluate().unwrap());
-
-        // test update with constant vector
-        let mut mean = Mean::<i32>::default();
-        let v: Vec<VectorRef> = vec![Arc::new(ConstantVector::new(
-            Arc::new(Int32Vector::from_vec(vec![4])),
-            10,
-        ))];
-        mean.update_batch(&v).unwrap();
-        assert_eq!(Value::from(4.0), mean.evaluate().unwrap());
-    }
-}

src/common/function/src/scalars/aggregate/polyval.rs

@@ -1,329 +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::marker::PhantomData;
-use std::sync::Arc;
-
-use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
-use common_query::error::{
-    self, BadAccumulatorImplSnafu, CreateAccumulatorSnafu, DowncastVectorSnafu,
-    FromScalarValueSnafu, InvalidInputColSnafu, InvalidInputStateSnafu, Result,
-};
-use common_query::logical_plan::accumulator::AggrFuncTypeStore;
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
-use common_query::prelude::*;
-use datatypes::prelude::*;
-use datatypes::types::{LogicalPrimitiveType, WrapperType};
-use datatypes::value::ListValue;
-use datatypes::vectors::{ConstantVector, Helper, Int64Vector, ListVector};
-use datatypes::with_match_primitive_type_id;
-use num_traits::AsPrimitive;
-use snafu::{ensure, OptionExt, ResultExt};
-
-// https://numpy.org/doc/stable/reference/generated/numpy.polyval.html
-#[derive(Debug, Default)]
-pub struct Polyval<T, PolyT>
-where
-    T: WrapperType,
-    T::Native: AsPrimitive<PolyT::Native>,
-    PolyT: WrapperType,
-    PolyT::Native: std::ops::Mul<Output = PolyT::Native>,
-{
-    values: Vec<T>,
-    // DataFusion casts constant in into i64 type.
-    x: Option<i64>,
-    _phantom: PhantomData<PolyT>,
-}
-
-impl<T, PolyT> Polyval<T, PolyT>
-where
-    T: WrapperType,
-    T::Native: AsPrimitive<PolyT::Native>,
-    PolyT: WrapperType,
-    PolyT::Native: std::ops::Mul<Output = PolyT::Native>,
-{
-    fn push(&mut self, value: T) {
-        self.values.push(value);
-    }
-}
-
-impl<T, PolyT> Accumulator for Polyval<T, PolyT>
-where
-    T: WrapperType,
-    T::Native: AsPrimitive<PolyT::Native>,
-    PolyT: WrapperType + std::iter::Sum<<PolyT as WrapperType>::Native>,
-    PolyT::Native: std::ops::Mul<Output = PolyT::Native> + std::iter::Sum<PolyT::Native>,
-    i64: AsPrimitive<<PolyT as WrapperType>::Native>,
-{
-    fn state(&self) -> Result<Vec<Value>> {
-        let nums = self
-            .values
-            .iter()
-            .map(|&n| n.into())
-            .collect::<Vec<Value>>();
-        Ok(vec![
-            Value::List(ListValue::new(nums, T::LogicalType::build_data_type())),
-            self.x.into(),
-        ])
-    }
-
-    fn update_batch(&mut self, values: &[VectorRef]) -> Result<()> {
-        if values.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(values.len() == 2, InvalidInputStateSnafu);
-        ensure!(values[0].len() == values[1].len(), InvalidInputStateSnafu);
-        if values[0].len() == 0 {
-            return Ok(());
-        }
-        // This is a unary accumulator, so only one column is provided.
-        let column = &values[0];
-        let mut len = 1;
-        let column: &<T as Scalar>::VectorType = if column.is_const() {
-            len = column.len();
-            let column: &ConstantVector = unsafe { Helper::static_cast(column) };
-            unsafe { Helper::static_cast(column.inner()) }
-        } else {
-            unsafe { Helper::static_cast(column) }
-        };
-        (0..len).for_each(|_| {
-            for v in column.iter_data().flatten() {
-                self.push(v);
-            }
-        });
-
-        let x = &values[1];
-        let x = Helper::check_get_scalar::<i64>(x).context(error::InvalidInputTypeSnafu {
-            err_msg: "expecting \"POLYVAL\" function's second argument to be a positive integer",
-        })?;
-        // `get(0)` is safe because we have checked `values[1].len() == values[0].len() != 0`
-        let first = x.get(0);
-        ensure!(!first.is_null(), InvalidInputColSnafu);
-
-        for i in 1..x.len() {
-            ensure!(first == x.get(i), InvalidInputColSnafu);
-        }
-
-        let first = match first {
-            Value::Int64(v) => v,
-            // unreachable because we have checked `first` is not null and is i64 above
-            _ => unreachable!(),
-        };
-        if let Some(x) = self.x {
-            ensure!(x == first, InvalidInputColSnafu);
-        } else {
-            self.x = Some(first);
-        };
-        Ok(())
-    }
-
-    // DataFusion executes accumulators in partitions. In some execution stage, DataFusion will
-    // merge states from other accumulators (returned by `state()` method).
-    fn merge_batch(&mut self, states: &[VectorRef]) -> Result<()> {
-        if states.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(
-            states.len() == 2,
-            BadAccumulatorImplSnafu {
-                err_msg: "expect 2 states in `merge_batch`",
-            }
-        );
-
-        let x = &states[1];
-        let x = x
-            .as_any()
-            .downcast_ref::<Int64Vector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect Int64Vector, got vector type {}",
-                    x.vector_type_name()
-                ),
-            })?;
-        let x = x.get(0);
-        if x.is_null() {
-            return Ok(());
-        }
-        let x = match x {
-            Value::Int64(x) => x,
-            _ => unreachable!(),
-        };
-        self.x = Some(x);
-
-        let values = &states[0];
-        let values = values
-            .as_any()
-            .downcast_ref::<ListVector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect ListVector, got vector type {}",
-                    values.vector_type_name()
-                ),
-            })?;
-        for value in values.values_iter() {
-            if let Some(value) = value.context(FromScalarValueSnafu)? {
-                let column: &<T as Scalar>::VectorType = unsafe { Helper::static_cast(&value) };
-                for v in column.iter_data().flatten() {
-                    self.push(v);
-                }
-            }
-        }
-
-        Ok(())
-    }
-
-    // DataFusion expects this function to return the final value of this aggregator.
-    fn evaluate(&self) -> Result<Value> {
-        if self.values.is_empty() {
-            return Ok(Value::Null);
-        }
-        let x = if let Some(x) = self.x {
-            x
-        } else {
-            return Ok(Value::Null);
-        };
-        let len = self.values.len();
-        let polyval: PolyT = self
-            .values
-            .iter()
-            .enumerate()
-            .map(|(i, &value)| value.into_native().as_() * x.pow((len - 1 - i) as u32).as_())
-            .sum();
-        Ok(polyval.into())
-    }
-}
-
-#[as_aggr_func_creator]
-#[derive(Debug, Default, AggrFuncTypeStore)]
-pub struct PolyvalAccumulatorCreator {}
-
-impl AggregateFunctionCreator for PolyvalAccumulatorCreator {
-    fn creator(&self) -> AccumulatorCreatorFunction {
-        let creator: AccumulatorCreatorFunction = Arc::new(move |types: &[ConcreteDataType]| {
-            let input_type = &types[0];
-            with_match_primitive_type_id!(
-                input_type.logical_type_id(),
-                |$S| {
-                    Ok(Box::new(Polyval::<<$S as LogicalPrimitiveType>::Wrapper, <<$S as LogicalPrimitiveType>::LargestType as LogicalPrimitiveType>::Wrapper>::default()))
-                },
-                {
-                    let err_msg = format!(
-                        "\"POLYVAL\" aggregate function not support data type {:?}",
-                        input_type.logical_type_id(),
-                    );
-                    CreateAccumulatorSnafu { err_msg }.fail()?
-                }
-            )
-        });
-        creator
-    }
-
-    fn output_type(&self) -> Result<ConcreteDataType> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 2, InvalidInputStateSnafu);
-        let input_type = self.input_types()?[0].logical_type_id();
-        with_match_primitive_type_id!(
-            input_type,
-            |$S| {
-                Ok(<<$S as LogicalPrimitiveType>::LargestType as LogicalPrimitiveType>::build_data_type())
-            },
-            {
-                unreachable!()
-            }
-        )
-    }
-
-    fn state_types(&self) -> Result<Vec<ConcreteDataType>> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 2, InvalidInputStateSnafu);
-        Ok(vec![
-            ConcreteDataType::list_datatype(input_types.into_iter().next().unwrap()),
-            ConcreteDataType::int64_datatype(),
-        ])
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use datatypes::vectors::Int32Vector;
-
-    use super::*;
-    #[test]
-    fn test_update_batch() {
-        // test update empty batch, expect not updating anything
-        let mut polyval = Polyval::<i32, i64>::default();
-        polyval.update_batch(&[]).unwrap();
-        assert!(polyval.values.is_empty());
-        assert_eq!(Value::Null, polyval.evaluate().unwrap());
-
-        // test update one not-null value
-        let mut polyval = Polyval::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from(vec![Some(3)])),
-            Arc::new(Int64Vector::from(vec![Some(2_i64)])),
-        ];
-        polyval.update_batch(&v).unwrap();
-        assert_eq!(Value::Int64(3), polyval.evaluate().unwrap());
-
-        // test update one null value
-        let mut polyval = Polyval::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from(vec![Option::<i32>::None])),
-            Arc::new(Int64Vector::from(vec![Some(2_i64)])),
-        ];
-        polyval.update_batch(&v).unwrap();
-        assert_eq!(Value::Null, polyval.evaluate().unwrap());
-
-        // test update no null-value batch
-        let mut polyval = Polyval::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from(vec![Some(3), Some(0), Some(1)])),
-            Arc::new(Int64Vector::from(vec![
-                Some(2_i64),
-                Some(2_i64),
-                Some(2_i64),
-            ])),
-        ];
-        polyval.update_batch(&v).unwrap();
-        assert_eq!(Value::Int64(13), polyval.evaluate().unwrap());
-
-        // test update null-value batch
-        let mut polyval = Polyval::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from(vec![Some(3), Some(0), None, Some(1)])),
-            Arc::new(Int64Vector::from(vec![
-                Some(2_i64),
-                Some(2_i64),
-                Some(2_i64),
-                Some(2_i64),
-            ])),
-        ];
-        polyval.update_batch(&v).unwrap();
-        assert_eq!(Value::Int64(13), polyval.evaluate().unwrap());
-
-        // test update with constant vector
-        let mut polyval = Polyval::<i32, i64>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(ConstantVector::new(
-                Arc::new(Int32Vector::from_vec(vec![4])),
-                2,
-            )),
-            Arc::new(Int64Vector::from(vec![Some(5_i64), Some(5_i64)])),
-        ];
-        polyval.update_batch(&v).unwrap();
-        assert_eq!(Value::Int64(24), polyval.evaluate().unwrap());
-    }
-}

src/common/function/src/scalars/aggregate/scipy_stats_norm_cdf.rs

@@ -1,270 +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::sync::Arc;
-
-use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
-use common_query::error::{
-    self, BadAccumulatorImplSnafu, CreateAccumulatorSnafu, DowncastVectorSnafu,
-    FromScalarValueSnafu, GenerateFunctionSnafu, InvalidInputColSnafu, InvalidInputStateSnafu,
-    Result,
-};
-use common_query::logical_plan::accumulator::AggrFuncTypeStore;
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
-use common_query::prelude::*;
-use datatypes::prelude::*;
-use datatypes::value::{ListValue, OrderedFloat};
-use datatypes::vectors::{ConstantVector, Float64Vector, Helper, ListVector};
-use datatypes::with_match_primitive_type_id;
-use num_traits::AsPrimitive;
-use snafu::{ensure, OptionExt, ResultExt};
-use statrs::distribution::{ContinuousCDF, Normal};
-use statrs::statistics::Statistics;
-
-// https://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.norm.html
-
-#[derive(Debug, Default)]
-pub struct ScipyStatsNormCdf<T> {
-    values: Vec<T>,
-    x: Option<f64>,
-}
-
-impl<T> ScipyStatsNormCdf<T> {
-    fn push(&mut self, value: T) {
-        self.values.push(value);
-    }
-}
-
-impl<T> Accumulator for ScipyStatsNormCdf<T>
-where
-    T: WrapperType + std::iter::Sum<T>,
-    T::Native: AsPrimitive<f64>,
-{
-    fn state(&self) -> Result<Vec<Value>> {
-        let nums = self
-            .values
-            .iter()
-            .map(|&x| x.into())
-            .collect::<Vec<Value>>();
-        Ok(vec![
-            Value::List(ListValue::new(nums, T::LogicalType::build_data_type())),
-            self.x.into(),
-        ])
-    }
-
-    fn update_batch(&mut self, values: &[VectorRef]) -> Result<()> {
-        if values.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(values.len() == 2, InvalidInputStateSnafu);
-        ensure!(values[1].len() == values[0].len(), InvalidInputStateSnafu);
-
-        if values[0].len() == 0 {
-            return Ok(());
-        }
-
-        let column = &values[0];
-        let mut len = 1;
-        let column: &<T as Scalar>::VectorType = if column.is_const() {
-            len = column.len();
-            let column: &ConstantVector = unsafe { Helper::static_cast(column) };
-            unsafe { Helper::static_cast(column.inner()) }
-        } else {
-            unsafe { Helper::static_cast(column) }
-        };
-
-        let x = &values[1];
-        let x = Helper::check_get_scalar::<f64>(x).context(error::InvalidInputTypeSnafu {
-            err_msg: "expecting \"SCIPYSTATSNORMCDF\" function's second argument to be a positive integer",
-        })?;
-        let first = x.get(0);
-        ensure!(!first.is_null(), InvalidInputColSnafu);
-        let first = match first {
-            Value::Float64(OrderedFloat(v)) => v,
-            // unreachable because we have checked `first` is not null and is i64 above
-            _ => unreachable!(),
-        };
-        if let Some(x) = self.x {
-            ensure!(x == first, InvalidInputColSnafu);
-        } else {
-            self.x = Some(first);
-        };
-
-        (0..len).for_each(|_| {
-            for v in column.iter_data().flatten() {
-                self.push(v);
-            }
-        });
-        Ok(())
-    }
-
-    fn merge_batch(&mut self, states: &[VectorRef]) -> Result<()> {
-        if states.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(
-            states.len() == 2,
-            BadAccumulatorImplSnafu {
-                err_msg: "expect 2 states in `merge_batch`",
-            }
-        );
-
-        let x = &states[1];
-        let x = x
-            .as_any()
-            .downcast_ref::<Float64Vector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect Float64Vector, got vector type {}",
-                    x.vector_type_name()
-                ),
-            })?;
-        let x = x.get(0);
-        if x.is_null() {
-            return Ok(());
-        }
-        let x = match x {
-            Value::Float64(OrderedFloat(x)) => x,
-            _ => unreachable!(),
-        };
-        self.x = Some(x);
-
-        let values = &states[0];
-        let values = values
-            .as_any()
-            .downcast_ref::<ListVector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect ListVector, got vector type {}",
-                    values.vector_type_name()
-                ),
-            })?;
-        for value in values.values_iter() {
-            if let Some(value) = value.context(FromScalarValueSnafu)? {
-                let column: &<T as Scalar>::VectorType = unsafe { Helper::static_cast(&value) };
-                for v in column.iter_data().flatten() {
-                    self.push(v);
-                }
-            }
-        }
-        Ok(())
-    }
-
-    fn evaluate(&self) -> Result<Value> {
-        let mean = self.values.iter().map(|v| v.into_native().as_()).mean();
-        let std_dev = self.values.iter().map(|v| v.into_native().as_()).std_dev();
-        if mean.is_nan() || std_dev.is_nan() {
-            Ok(Value::Null)
-        } else {
-            let x = if let Some(x) = self.x {
-                x
-            } else {
-                return Ok(Value::Null);
-            };
-            let n = Normal::new(mean, std_dev).context(GenerateFunctionSnafu)?;
-            Ok(n.cdf(x).into())
-        }
-    }
-}
-
-#[as_aggr_func_creator]
-#[derive(Debug, Default, AggrFuncTypeStore)]
-pub struct ScipyStatsNormCdfAccumulatorCreator {}
-
-impl AggregateFunctionCreator for ScipyStatsNormCdfAccumulatorCreator {
-    fn creator(&self) -> AccumulatorCreatorFunction {
-        let creator: AccumulatorCreatorFunction = Arc::new(move |types: &[ConcreteDataType]| {
-            let input_type = &types[0];
-            with_match_primitive_type_id!(
-                input_type.logical_type_id(),
-                |$S| {
-                    Ok(Box::new(ScipyStatsNormCdf::<<$S as LogicalPrimitiveType>::Wrapper>::default()))
-                },
-                {
-                    let err_msg = format!(
-                        "\"SCIPYSTATSNORMCDF\" aggregate function not support data type {:?}",
-                        input_type.logical_type_id(),
-                    );
-                    CreateAccumulatorSnafu { err_msg }.fail()?
-                }
-            )
-        });
-        creator
-    }
-
-    fn output_type(&self) -> Result<ConcreteDataType> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 2, InvalidInputStateSnafu);
-        Ok(ConcreteDataType::float64_datatype())
-    }
-
-    fn state_types(&self) -> Result<Vec<ConcreteDataType>> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 2, InvalidInputStateSnafu);
-        Ok(vec![
-            ConcreteDataType::list_datatype(input_types[0].clone()),
-            ConcreteDataType::float64_datatype(),
-        ])
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use datatypes::vectors::{Float64Vector, Int32Vector};
-
-    use super::*;
-    #[test]
-    fn test_update_batch() {
-        // test update empty batch, expect not updating anything
-        let mut scipy_stats_norm_cdf = ScipyStatsNormCdf::<i32>::default();
-        scipy_stats_norm_cdf.update_batch(&[]).unwrap();
-        assert!(scipy_stats_norm_cdf.values.is_empty());
-        assert_eq!(Value::Null, scipy_stats_norm_cdf.evaluate().unwrap());
-
-        // test update no null-value batch
-        let mut scipy_stats_norm_cdf = ScipyStatsNormCdf::<i32>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from(vec![Some(-1i32), Some(1), Some(2)])),
-            Arc::new(Float64Vector::from(vec![
-                Some(2.0_f64),
-                Some(2.0_f64),
-                Some(2.0_f64),
-            ])),
-        ];
-        scipy_stats_norm_cdf.update_batch(&v).unwrap();
-        assert_eq!(
-            Value::from(0.8086334555398362),
-            scipy_stats_norm_cdf.evaluate().unwrap()
-        );
-
-        // test update null-value batch
-        let mut scipy_stats_norm_cdf = ScipyStatsNormCdf::<i32>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from(vec![Some(-2i32), None, Some(3), Some(4)])),
-            Arc::new(Float64Vector::from(vec![
-                Some(2.0_f64),
-                None,
-                Some(2.0_f64),
-                Some(2.0_f64),
-            ])),
-        ];
-        scipy_stats_norm_cdf.update_batch(&v).unwrap();
-        assert_eq!(
-            Value::from(0.5412943699039795),
-            scipy_stats_norm_cdf.evaluate().unwrap()
-        );
-    }
-}

src/common/function/src/scalars/aggregate/scipy_stats_norm_pdf.rs

@@ -1,271 +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::sync::Arc;
-
-use common_macro::{as_aggr_func_creator, AggrFuncTypeStore};
-use common_query::error::{
-    self, BadAccumulatorImplSnafu, CreateAccumulatorSnafu, DowncastVectorSnafu,
-    FromScalarValueSnafu, GenerateFunctionSnafu, InvalidInputColSnafu, InvalidInputStateSnafu,
-    Result,
-};
-use common_query::logical_plan::accumulator::AggrFuncTypeStore;
-use common_query::logical_plan::{Accumulator, AggregateFunctionCreator};
-use common_query::prelude::*;
-use datatypes::prelude::*;
-use datatypes::value::{ListValue, OrderedFloat};
-use datatypes::vectors::{ConstantVector, Float64Vector, Helper, ListVector};
-use datatypes::with_match_primitive_type_id;
-use num_traits::AsPrimitive;
-use snafu::{ensure, OptionExt, ResultExt};
-use statrs::distribution::{Continuous, Normal};
-use statrs::statistics::Statistics;
-
-// https://docs.scipy.org/doc/scipy/reference/generated/scipy.stats.norm.html
-
-#[derive(Debug, Default)]
-pub struct ScipyStatsNormPdf<T> {
-    values: Vec<T>,
-    x: Option<f64>,
-}
-
-impl<T> ScipyStatsNormPdf<T> {
-    fn push(&mut self, value: T) {
-        self.values.push(value);
-    }
-}
-
-impl<T> Accumulator for ScipyStatsNormPdf<T>
-where
-    T: WrapperType,
-    T::Native: AsPrimitive<f64> + std::iter::Sum<T>,
-{
-    fn state(&self) -> Result<Vec<Value>> {
-        let nums = self
-            .values
-            .iter()
-            .map(|&x| x.into())
-            .collect::<Vec<Value>>();
-        Ok(vec![
-            Value::List(ListValue::new(nums, T::LogicalType::build_data_type())),
-            self.x.into(),
-        ])
-    }
-
-    fn update_batch(&mut self, values: &[VectorRef]) -> Result<()> {
-        if values.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(values.len() == 2, InvalidInputStateSnafu);
-        ensure!(values[1].len() == values[0].len(), InvalidInputStateSnafu);
-
-        if values[0].len() == 0 {
-            return Ok(());
-        }
-
-        let column = &values[0];
-        let mut len = 1;
-        let column: &<T as Scalar>::VectorType = if column.is_const() {
-            len = column.len();
-            let column: &ConstantVector = unsafe { Helper::static_cast(column) };
-            unsafe { Helper::static_cast(column.inner()) }
-        } else {
-            unsafe { Helper::static_cast(column) }
-        };
-
-        let x = &values[1];
-        let x = Helper::check_get_scalar::<f64>(x).context(error::InvalidInputTypeSnafu {
-            err_msg: "expecting \"SCIPYSTATSNORMPDF\" function's second argument to be a positive integer",
-        })?;
-        let first = x.get(0);
-        ensure!(!first.is_null(), InvalidInputColSnafu);
-        let first = match first {
-            Value::Float64(OrderedFloat(v)) => v,
-            // unreachable because we have checked `first` is not null and is i64 above
-            _ => unreachable!(),
-        };
-        if let Some(x) = self.x {
-            ensure!(x == first, InvalidInputColSnafu);
-        } else {
-            self.x = Some(first);
-        };
-
-        (0..len).for_each(|_| {
-            for v in column.iter_data().flatten() {
-                self.push(v);
-            }
-        });
-        Ok(())
-    }
-
-    fn merge_batch(&mut self, states: &[VectorRef]) -> Result<()> {
-        if states.is_empty() {
-            return Ok(());
-        }
-
-        ensure!(
-            states.len() == 2,
-            BadAccumulatorImplSnafu {
-                err_msg: "expect 2 states in `merge_batch`",
-            }
-        );
-
-        let x = &states[1];
-        let x = x
-            .as_any()
-            .downcast_ref::<Float64Vector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect Float64Vector, got vector type {}",
-                    x.vector_type_name()
-                ),
-            })?;
-        let x = x.get(0);
-        if x.is_null() {
-            return Ok(());
-        }
-        let x = match x {
-            Value::Float64(OrderedFloat(x)) => x,
-            _ => unreachable!(),
-        };
-        self.x = Some(x);
-
-        let values = &states[0];
-        let values = values
-            .as_any()
-            .downcast_ref::<ListVector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!(
-                    "expect ListVector, got vector type {}",
-                    values.vector_type_name()
-                ),
-            })?;
-        for value in values.values_iter() {
-            if let Some(value) = value.context(FromScalarValueSnafu)? {
-                let column: &<T as Scalar>::VectorType = unsafe { Helper::static_cast(&value) };
-                for v in column.iter_data().flatten() {
-                    self.push(v);
-                }
-            }
-        }
-        Ok(())
-    }
-
-    fn evaluate(&self) -> Result<Value> {
-        let mean = self.values.iter().map(|v| v.into_native().as_()).mean();
-        let std_dev = self.values.iter().map(|v| v.into_native().as_()).std_dev();
-
-        if mean.is_nan() || std_dev.is_nan() {
-            Ok(Value::Null)
-        } else {
-            let x = if let Some(x) = self.x {
-                x
-            } else {
-                return Ok(Value::Null);
-            };
-            let n = Normal::new(mean, std_dev).context(GenerateFunctionSnafu)?;
-            Ok(n.pdf(x).into())
-        }
-    }
-}
-
-#[as_aggr_func_creator]
-#[derive(Debug, Default, AggrFuncTypeStore)]
-pub struct ScipyStatsNormPdfAccumulatorCreator {}
-
-impl AggregateFunctionCreator for ScipyStatsNormPdfAccumulatorCreator {
-    fn creator(&self) -> AccumulatorCreatorFunction {
-        let creator: AccumulatorCreatorFunction = Arc::new(move |types: &[ConcreteDataType]| {
-            let input_type = &types[0];
-            with_match_primitive_type_id!(
-                input_type.logical_type_id(),
-                |$S| {
-                    Ok(Box::new(ScipyStatsNormPdf::<<$S as LogicalPrimitiveType>::Wrapper>::default()))
-                },
-                {
-                    let err_msg = format!(
-                        "\"SCIPYSTATSNORMpdf\" aggregate function not support data type {:?}",
-                        input_type.logical_type_id(),
-                    );
-                    CreateAccumulatorSnafu { err_msg }.fail()?
-                }
-            )
-        });
-        creator
-    }
-
-    fn output_type(&self) -> Result<ConcreteDataType> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 2, InvalidInputStateSnafu);
-        Ok(ConcreteDataType::float64_datatype())
-    }
-
-    fn state_types(&self) -> Result<Vec<ConcreteDataType>> {
-        let input_types = self.input_types()?;
-        ensure!(input_types.len() == 2, InvalidInputStateSnafu);
-        Ok(vec![
-            ConcreteDataType::list_datatype(input_types[0].clone()),
-            ConcreteDataType::float64_datatype(),
-        ])
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use datatypes::vectors::{Float64Vector, Int32Vector};
-
-    use super::*;
-    #[test]
-    fn test_update_batch() {
-        // test update empty batch, expect not updating anything
-        let mut scipy_stats_norm_pdf = ScipyStatsNormPdf::<i32>::default();
-        scipy_stats_norm_pdf.update_batch(&[]).unwrap();
-        assert!(scipy_stats_norm_pdf.values.is_empty());
-        assert_eq!(Value::Null, scipy_stats_norm_pdf.evaluate().unwrap());
-
-        // test update no null-value batch
-        let mut scipy_stats_norm_pdf = ScipyStatsNormPdf::<i32>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from(vec![Some(-1i32), Some(1), Some(2)])),
-            Arc::new(Float64Vector::from(vec![
-                Some(2.0_f64),
-                Some(2.0_f64),
-                Some(2.0_f64),
-            ])),
-        ];
-        scipy_stats_norm_pdf.update_batch(&v).unwrap();
-        assert_eq!(
-            Value::from(0.17843340219081558),
-            scipy_stats_norm_pdf.evaluate().unwrap()
-        );
-
-        // test update null-value batch
-        let mut scipy_stats_norm_pdf = ScipyStatsNormPdf::<i32>::default();
-        let v: Vec<VectorRef> = vec![
-            Arc::new(Int32Vector::from(vec![Some(-2i32), None, Some(3), Some(4)])),
-            Arc::new(Float64Vector::from(vec![
-                Some(2.0_f64),
-                None,
-                Some(2.0_f64),
-                Some(2.0_f64),
-            ])),
-        ];
-        scipy_stats_norm_pdf.update_batch(&v).unwrap();
-        assert_eq!(
-            Value::from(0.12343972049858312),
-            scipy_stats_norm_pdf.evaluate().unwrap()
-        );
-    }
-}

src/common/meta/src/cluster.rs

@@ -28,7 +28,6 @@ use crate::error::{
     InvalidRoleSnafu, ParseNumSnafu, Result,
 };
 use crate::peer::Peer;
-use crate::ClusterId;
 
 const CLUSTER_NODE_INFO_PREFIX: &str = "__meta_cluster_node_info";
 
@@ -56,12 +55,9 @@ pub trait ClusterInfo {
     // TODO(jeremy): Other info, like region status, etc.
 }
 
-/// The key of [NodeInfo] in the storage. The format is `__meta_cluster_node_info-{cluster_id}-{role}-{node_id}`.
+/// The key of [NodeInfo] in the storage. The format is `__meta_cluster_node_info-0-{role}-{node_id}`.
 #[derive(Debug, Clone, Copy, Eq, Hash, PartialEq, Serialize, Deserialize)]
 pub struct NodeInfoKey {
-    /// The cluster id.
-    // todo(hl): remove cluster_id as it is not assigned anywhere.
-    pub cluster_id: ClusterId,
     /// The role of the node. It can be `[Role::Datanode]` or `[Role::Frontend]`.
     pub role: Role,
     /// The node id.
@@ -84,24 +80,15 @@ impl NodeInfoKey {
             _ => peer.id,
         };
 
-        Some(NodeInfoKey {
-            cluster_id: header.cluster_id,
-            role,
-            node_id,
-        })
+        Some(NodeInfoKey { role, node_id })
     }
 
-    pub fn key_prefix_with_cluster_id(cluster_id: u64) -> String {
-        format!("{}-{}-", CLUSTER_NODE_INFO_PREFIX, cluster_id)
+    pub fn key_prefix() -> String {
+        format!("{}-0-", CLUSTER_NODE_INFO_PREFIX)
     }
 
-    pub fn key_prefix_with_role(cluster_id: ClusterId, role: Role) -> String {
-        format!(
-            "{}-{}-{}-",
-            CLUSTER_NODE_INFO_PREFIX,
-            cluster_id,
-            i32::from(role)
-        )
+    pub fn key_prefix_with_role(role: Role) -> String {
+        format!("{}-0-{}-", CLUSTER_NODE_INFO_PREFIX, i32::from(role))
     }
 }
 
@@ -193,15 +180,10 @@ impl FromStr for NodeInfoKey {
         let caps = CLUSTER_NODE_INFO_PREFIX_PATTERN
             .captures(key)
             .context(InvalidNodeInfoKeySnafu { key })?;
-
         ensure!(caps.len() == 4, InvalidNodeInfoKeySnafu { key });
 
-        let cluster_id = caps[1].to_string();
         let role = caps[2].to_string();
         let node_id = caps[3].to_string();
-        let cluster_id: u64 = cluster_id.parse().context(ParseNumSnafu {
-            err_msg: format!("invalid cluster_id: {cluster_id}"),
-        })?;
         let role: i32 = role.parse().context(ParseNumSnafu {
             err_msg: format!("invalid role {role}"),
         })?;
@@ -210,11 +192,7 @@ impl FromStr for NodeInfoKey {
             err_msg: format!("invalid node_id: {node_id}"),
         })?;
 
-        Ok(Self {
-            cluster_id,
-            role,
-            node_id,
-        })
+        Ok(Self { role, node_id })
     }
 }
 
@@ -233,9 +211,8 @@ impl TryFrom<Vec<u8>> for NodeInfoKey {
 impl From<&NodeInfoKey> for Vec<u8> {
     fn from(key: &NodeInfoKey) -> Self {
         format!(
-            "{}-{}-{}-{}",
+            "{}-0-{}-{}",
             CLUSTER_NODE_INFO_PREFIX,
-            key.cluster_id,
             i32::from(key.role),
             key.node_id
         )
@@ -308,7 +285,6 @@ mod tests {
     #[test]
     fn test_node_info_key_round_trip() {
         let key = NodeInfoKey {
-            cluster_id: 1,
             role: Datanode,
             node_id: 2,
         };
@@ -316,7 +292,6 @@ mod tests {
         let key_bytes: Vec<u8> = (&key).into();
         let new_key: NodeInfoKey = key_bytes.try_into().unwrap();
 
-        assert_eq!(1, new_key.cluster_id);
         assert_eq!(Datanode, new_key.role);
         assert_eq!(2, new_key.node_id);
     }
@@ -362,11 +337,11 @@ mod tests {
 
     #[test]
     fn test_node_info_key_prefix() {
-        let prefix = NodeInfoKey::key_prefix_with_cluster_id(1);
-        assert_eq!(prefix, "__meta_cluster_node_info-1-");
+        let prefix = NodeInfoKey::key_prefix();
+        assert_eq!(prefix, "__meta_cluster_node_info-0-");
 
-        let prefix = NodeInfoKey::key_prefix_with_role(2, Frontend);
-        assert_eq!(prefix, "__meta_cluster_node_info-2-1-");
+        let prefix = NodeInfoKey::key_prefix_with_role(Frontend);
+        assert_eq!(prefix, "__meta_cluster_node_info-0-1-");
     }
 
     #[test]

src/common/meta/src/datanode.rs

@@ -25,8 +25,8 @@ use store_api::region_engine::{RegionRole, RegionStatistic};
 use store_api::storage::RegionId;
 use table::metadata::TableId;
 
+use crate::error;
 use crate::error::Result;
-use crate::{error, ClusterId};
 
 pub(crate) const DATANODE_LEASE_PREFIX: &str = "__meta_datanode_lease";
 const INACTIVE_REGION_PREFIX: &str = "__meta_inactive_region";
@@ -48,11 +48,10 @@ lazy_static! {
 
 /// The key of the datanode stat in the storage.
 ///
-/// The format is `__meta_datanode_stat-{cluster_id}-{node_id}`.
+/// The format is `__meta_datanode_stat-0-{node_id}`.
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 pub struct Stat {
     pub timestamp_millis: i64,
-    pub cluster_id: ClusterId,
     // The datanode Id.
     pub id: u64,
     // The datanode address.
@@ -102,10 +101,7 @@ impl Stat {
     }
 
     pub fn stat_key(&self) -> DatanodeStatKey {
-        DatanodeStatKey {
-            cluster_id: self.cluster_id,
-            node_id: self.id,
-        }
+        DatanodeStatKey { node_id: self.id }
     }
 
     /// Returns a tuple array containing [RegionId] and [RegionRole].
@@ -145,7 +141,7 @@ impl TryFrom<&HeartbeatRequest> for Stat {
         } = value;
 
         match (header, peer) {
-            (Some(header), Some(peer)) => {
+            (Some(_header), Some(peer)) => {
                 let region_stats = region_stats
                     .iter()
                     .map(RegionStat::from)
@@ -153,7 +149,6 @@ impl TryFrom<&HeartbeatRequest> for Stat {
 
                 Ok(Self {
                     timestamp_millis: time_util::current_time_millis(),
-                    cluster_id: header.cluster_id,
                     // datanode id
                     id: peer.id,
                     // datanode address
@@ -196,32 +191,24 @@ impl From<&api::v1::meta::RegionStat> for RegionStat {
 
 /// The key of the datanode stat in the memory store.
 ///
-/// The format is `__meta_datanode_stat-{cluster_id}-{node_id}`.
+/// The format is `__meta_datanode_stat-0-{node_id}`.
 #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
 pub struct DatanodeStatKey {
-    pub cluster_id: ClusterId,
     pub node_id: u64,
 }
 
 impl DatanodeStatKey {
     /// The key prefix.
     pub fn prefix_key() -> Vec<u8> {
-        format!("{DATANODE_STAT_PREFIX}-").into_bytes()
-    }
-
-    /// The key prefix with the cluster id.
-    pub fn key_prefix_with_cluster_id(cluster_id: ClusterId) -> String {
-        format!("{DATANODE_STAT_PREFIX}-{cluster_id}-")
+        // todo(hl): remove cluster id in prefix
+        format!("{DATANODE_STAT_PREFIX}-0-").into_bytes()
     }
 }
 
 impl From<DatanodeStatKey> for Vec<u8> {
     fn from(value: DatanodeStatKey) -> Self {
-        format!(
-            "{}-{}-{}",
-            DATANODE_STAT_PREFIX, value.cluster_id, value.node_id
-        )
-        .into_bytes()
+        // todo(hl): remove cluster id in prefix
+        format!("{}-0-{}", DATANODE_STAT_PREFIX, value.node_id).into_bytes()
     }
 }
 
@@ -234,20 +221,12 @@ impl FromStr for DatanodeStatKey {
             .context(error::InvalidStatKeySnafu { key })?;
 
         ensure!(caps.len() == 3, error::InvalidStatKeySnafu { key });
-
-        let cluster_id = caps[1].to_string();
         let node_id = caps[2].to_string();
-        let cluster_id: u64 = cluster_id.parse().context(error::ParseNumSnafu {
-            err_msg: format!("invalid cluster_id: {cluster_id}"),
-        })?;
         let node_id: u64 = node_id.parse().context(error::ParseNumSnafu {
             err_msg: format!("invalid node_id: {node_id}"),
         })?;
 
-        Ok(Self {
-            cluster_id,
-            node_id,
-        })
+        Ok(Self { node_id })
     }
 }
 
@@ -321,7 +300,6 @@ mod tests {
     #[test]
     fn test_stat_key() {
         let stat = Stat {
-            cluster_id: 3,
             id: 101,
             region_num: 10,
             ..Default::default()
@@ -329,14 +307,12 @@ mod tests {
 
         let stat_key = stat.stat_key();
 
-        assert_eq!(3, stat_key.cluster_id);
         assert_eq!(101, stat_key.node_id);
     }
 
     #[test]
     fn test_stat_val_round_trip() {
         let stat = Stat {
-            cluster_id: 0,
             id: 101,
             region_num: 100,
             ..Default::default()
@@ -351,7 +327,6 @@ mod tests {
         assert_eq!(1, stats.len());
 
         let stat = stats.first().unwrap();
-        assert_eq!(0, stat.cluster_id);
         assert_eq!(101, stat.id);
         assert_eq!(100, stat.region_num);
     }

src/common/meta/src/ddl.rs

@@ -30,7 +30,7 @@ use crate::node_manager::NodeManagerRef;
 use crate::region_keeper::MemoryRegionKeeperRef;
 use crate::rpc::ddl::{SubmitDdlTaskRequest, SubmitDdlTaskResponse};
 use crate::rpc::procedure::{MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse};
-use crate::{ClusterId, DatanodeId};
+use crate::DatanodeId;
 
 pub mod alter_database;
 pub mod alter_logical_tables;
@@ -57,7 +57,6 @@ pub mod utils;
 
 #[derive(Debug, Default)]
 pub struct ExecutorContext {
-    pub cluster_id: Option<u64>,
     pub tracing_context: Option<W3cTrace>,
 }
 
@@ -90,10 +89,6 @@ pub trait ProcedureExecutor: Send + Sync {
 
 pub type ProcedureExecutorRef = Arc<dyn ProcedureExecutor>;
 
-pub struct TableMetadataAllocatorContext {
-    pub cluster_id: ClusterId,
-}
-
 /// Metadata allocated to a table.
 #[derive(Default)]
 pub struct TableMetadata {
@@ -108,7 +103,7 @@ pub struct TableMetadata {
 
 pub type RegionFailureDetectorControllerRef = Arc<dyn RegionFailureDetectorController>;
 
-pub type DetectingRegion = (ClusterId, DatanodeId, RegionId);
+pub type DetectingRegion = (DatanodeId, RegionId);
 
 /// Used for actively registering Region failure detectors.
 ///

src/common/meta/src/ddl/alter_database.rs

@@ -30,7 +30,6 @@ use crate::key::DeserializedValueWithBytes;
 use crate::lock_key::{CatalogLock, SchemaLock};
 use crate::rpc::ddl::UnsetDatabaseOption::{self};
 use crate::rpc::ddl::{AlterDatabaseKind, AlterDatabaseTask, SetDatabaseOption};
-use crate::ClusterId;
 
 pub struct AlterDatabaseProcedure {
     pub context: DdlContext,
@@ -65,14 +64,10 @@ fn build_new_schema_value(
 impl AlterDatabaseProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::AlterDatabase";
 
-    pub fn new(
-        cluster_id: ClusterId,
-        task: AlterDatabaseTask,
-        context: DdlContext,
-    ) -> Result<Self> {
+    pub fn new(task: AlterDatabaseTask, context: DdlContext) -> Result<Self> {
         Ok(Self {
             context,
-            data: AlterDatabaseData::new(task, cluster_id)?,
+            data: AlterDatabaseData::new(task)?,
         })
     }
 
@@ -183,7 +178,6 @@ enum AlterDatabaseState {
 /// The data of alter database procedure.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct AlterDatabaseData {
-    cluster_id: ClusterId,
     state: AlterDatabaseState,
     kind: AlterDatabaseKind,
     catalog_name: String,
@@ -192,9 +186,8 @@ pub struct AlterDatabaseData {
 }
 
 impl AlterDatabaseData {
-    pub fn new(task: AlterDatabaseTask, cluster_id: ClusterId) -> Result<Self> {
+    pub fn new(task: AlterDatabaseTask) -> Result<Self> {
         Ok(Self {
-            cluster_id,
             state: AlterDatabaseState::Prepare,
             kind: AlterDatabaseKind::try_from(task.alter_expr.kind.unwrap())?,
             catalog_name: task.alter_expr.catalog_name,

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

@@ -37,9 +37,9 @@ use crate::key::table_info::TableInfoValue;
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::key::DeserializedValueWithBytes;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
+use crate::metrics;
 use crate::rpc::ddl::AlterTableTask;
 use crate::rpc::router::find_leaders;
-use crate::{metrics, ClusterId};
 
 pub struct AlterLogicalTablesProcedure {
     pub context: DdlContext,
@@ -50,7 +50,6 @@ impl AlterLogicalTablesProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::AlterLogicalTables";
 
     pub fn new(
-        cluster_id: ClusterId,
         tasks: Vec<AlterTableTask>,
         physical_table_id: TableId,
         context: DdlContext,
@@ -58,7 +57,6 @@ impl AlterLogicalTablesProcedure {
         Self {
             context,
             data: AlterTablesData {
-                cluster_id,
                 state: AlterTablesState::Prepare,
                 tasks,
                 table_info_values: vec![],
@@ -240,7 +238,6 @@ impl Procedure for AlterLogicalTablesProcedure {
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct AlterTablesData {
-    cluster_id: ClusterId,
     state: AlterTablesState,
     tasks: Vec<AlterTableTask>,
     /// Table info values before the alter operation.

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

@@ -45,9 +45,9 @@ 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::rpc::ddl::AlterTableTask;
 use crate::rpc::router::{find_leader_regions, find_leaders, region_distribution};
-use crate::{metrics, ClusterId};
 
 /// The alter table procedure
 pub struct AlterTableProcedure {
@@ -64,16 +64,11 @@ pub struct AlterTableProcedure {
 impl AlterTableProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::AlterTable";
 
-    pub fn new(
-        cluster_id: ClusterId,
-        table_id: TableId,
-        task: AlterTableTask,
-        context: DdlContext,
-    ) -> Result<Self> {
+    pub fn new(table_id: TableId, task: AlterTableTask, context: DdlContext) -> Result<Self> {
         task.validate()?;
         Ok(Self {
             context,
-            data: AlterTableData::new(task, table_id, cluster_id),
+            data: AlterTableData::new(task, table_id),
             new_table_info: None,
         })
     }
@@ -307,7 +302,6 @@ enum AlterTableState {
 // The serialized data of alter table.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct AlterTableData {
-    cluster_id: ClusterId,
     state: AlterTableState,
     task: AlterTableTask,
     table_id: TableId,
@@ -318,12 +312,11 @@ pub struct AlterTableData {
 }
 
 impl AlterTableData {
-    pub fn new(task: AlterTableTask, table_id: TableId, cluster_id: u64) -> Self {
+    pub fn new(task: AlterTableTask, table_id: TableId) -> Self {
         Self {
             state: AlterTableState::Prepare,
             task,
             table_id,
-            cluster_id,
             table_info_value: None,
             region_distribution: None,
         }

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

@@ -167,10 +167,9 @@ mod tests {
     use crate::test_util::{new_ddl_context, MockDatanodeManager};
 
     /// Prepares a region with schema `[ts: Timestamp, host: Tag, cpu: Field]`.
-    async fn prepare_ddl_context() -> (DdlContext, u64, TableId, RegionId, String) {
+    async fn prepare_ddl_context() -> (DdlContext, TableId, RegionId, String) {
         let datanode_manager = Arc::new(MockDatanodeManager::new(()));
         let ddl_context = new_ddl_context(datanode_manager);
-        let cluster_id = 1;
         let table_id = 1024;
         let region_id = RegionId::new(table_id, 1);
         let table_name = "foo";
@@ -225,19 +224,12 @@ mod tests {
             )
             .await
             .unwrap();
-        (
-            ddl_context,
-            cluster_id,
-            table_id,
-            region_id,
-            table_name.to_string(),
-        )
+        (ddl_context, table_id, region_id, table_name.to_string())
     }
 
     #[tokio::test]
     async fn test_make_alter_region_request() {
-        let (ddl_context, cluster_id, table_id, region_id, table_name) =
-            prepare_ddl_context().await;
+        let (ddl_context, table_id, region_id, table_name) = prepare_ddl_context().await;
 
         let task = AlterTableTask {
             alter_table: AlterTableExpr {
@@ -265,8 +257,7 @@ mod tests {
             },
         };
 
-        let mut procedure =
-            AlterTableProcedure::new(cluster_id, table_id, task, ddl_context).unwrap();
+        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)) = procedure
@@ -307,8 +298,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_make_alter_column_type_region_request() {
-        let (ddl_context, cluster_id, table_id, region_id, table_name) =
-            prepare_ddl_context().await;
+        let (ddl_context, table_id, region_id, table_name) = prepare_ddl_context().await;
 
         let task = AlterTableTask {
             alter_table: AlterTableExpr {
@@ -325,8 +315,7 @@ mod tests {
             },
         };
 
-        let mut procedure =
-            AlterTableProcedure::new(cluster_id, table_id, task, ddl_context).unwrap();
+        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)) = procedure

src/common/meta/src/ddl/create_flow.rs

@@ -46,9 +46,9 @@ use crate::key::flow::flow_route::FlowRouteValue;
 use crate::key::table_name::TableNameKey;
 use crate::key::{DeserializedValueWithBytes, FlowId, FlowPartitionId};
 use crate::lock_key::{CatalogLock, FlowNameLock, TableNameLock};
+use crate::metrics;
 use crate::peer::Peer;
 use crate::rpc::ddl::{CreateFlowTask, QueryContext};
-use crate::{metrics, ClusterId};
 
 /// The procedure of flow creation.
 pub struct CreateFlowProcedure {
@@ -60,16 +60,10 @@ impl CreateFlowProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::CreateFlow";
 
     /// Returns a new [CreateFlowProcedure].
-    pub fn new(
-        cluster_id: ClusterId,
-        task: CreateFlowTask,
-        query_context: QueryContext,
-        context: DdlContext,
-    ) -> Self {
+    pub fn new(task: CreateFlowTask, query_context: QueryContext, context: DdlContext) -> Self {
         Self {
             context,
             data: CreateFlowData {
-                cluster_id,
                 task,
                 flow_id: None,
                 peers: vec![],
@@ -363,7 +357,6 @@ impl fmt::Display for FlowType {
 /// The serializable data.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CreateFlowData {
-    pub(crate) cluster_id: ClusterId,
     pub(crate) state: CreateFlowState,
     pub(crate) task: CreateFlowTask,
     pub(crate) flow_id: Option<FlowId>,

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

@@ -23,11 +23,10 @@ impl CreateFlowProcedure {
     pub(crate) async fn allocate_flow_id(&mut self) -> Result<()> {
         //TODO(weny, ruihang): We doesn't support the partitions. It's always be 1, now.
         let partitions = 1;
-        let cluster_id = self.data.cluster_id;
         let (flow_id, peers) = self
             .context
             .flow_metadata_allocator
-            .create(cluster_id, partitions)
+            .create(partitions)
             .await?;
         self.data.flow_id = Some(flow_id);
         self.data.peers = peers;

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

@@ -36,9 +36,9 @@ use crate::ddl::DdlContext;
 use crate::error::{DecodeJsonSnafu, MetadataCorruptionSnafu, Result};
 use crate::key::table_route::TableRouteValue;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock, TableNameLock};
+use crate::metrics;
 use crate::rpc::ddl::CreateTableTask;
 use crate::rpc::router::{find_leaders, RegionRoute};
-use crate::{metrics, ClusterId};
 
 pub struct CreateLogicalTablesProcedure {
     pub context: DdlContext,
@@ -49,7 +49,6 @@ impl CreateLogicalTablesProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::CreateLogicalTables";
 
     pub fn new(
-        cluster_id: ClusterId,
         tasks: Vec<CreateTableTask>,
         physical_table_id: TableId,
         context: DdlContext,
@@ -57,7 +56,6 @@ impl CreateLogicalTablesProcedure {
         Self {
             context,
             data: CreateTablesData {
-                cluster_id,
                 state: CreateTablesState::Prepare,
                 tasks,
                 table_ids_already_exists: vec![],
@@ -245,7 +243,6 @@ impl Procedure for CreateLogicalTablesProcedure {
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CreateTablesData {
-    cluster_id: ClusterId,
     state: CreateTablesState,
     tasks: Vec<CreateTableTask>,
     table_ids_already_exists: Vec<Option<TableId>>,

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

@@ -37,17 +37,17 @@ use crate::ddl::utils::{
     add_peer_context_if_needed, convert_region_routes_to_detecting_regions, handle_retry_error,
     region_storage_path,
 };
-use crate::ddl::{DdlContext, TableMetadata, TableMetadataAllocatorContext};
+use crate::ddl::{DdlContext, TableMetadata};
 use crate::error::{self, Result};
 use crate::key::table_name::TableNameKey;
 use crate::key::table_route::{PhysicalTableRouteValue, TableRouteValue};
 use crate::lock_key::{CatalogLock, SchemaLock, TableNameLock};
+use crate::metrics;
 use crate::region_keeper::OperatingRegionGuard;
 use crate::rpc::ddl::CreateTableTask;
 use crate::rpc::router::{
     find_leader_regions, find_leaders, operating_leader_regions, RegionRoute,
 };
-use crate::{metrics, ClusterId};
 pub struct CreateTableProcedure {
     pub context: DdlContext,
     pub creator: TableCreator,
@@ -56,10 +56,10 @@ pub struct CreateTableProcedure {
 impl CreateTableProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::CreateTable";
 
-    pub fn new(cluster_id: ClusterId, task: CreateTableTask, context: DdlContext) -> Self {
+    pub fn new(task: CreateTableTask, context: DdlContext) -> Self {
         Self {
             context,
-            creator: TableCreator::new(cluster_id, task),
+            creator: TableCreator::new(task),
         }
     }
 
@@ -154,12 +154,7 @@ impl CreateTableProcedure {
         } = self
             .context
             .table_metadata_allocator
-            .create(
-                &TableMetadataAllocatorContext {
-                    cluster_id: self.creator.data.cluster_id,
-                },
-                &self.creator.data.task,
-            )
+            .create(&self.creator.data.task)
             .await?;
         self.creator
             .set_allocated_metadata(table_id, table_route, region_wal_options);
@@ -268,7 +263,6 @@ impl CreateTableProcedure {
     /// - Failed to create table metadata.
     async fn on_create_metadata(&mut self) -> Result<Status> {
         let table_id = self.table_id();
-        let cluster_id = self.creator.data.cluster_id;
         let manager = &self.context.table_metadata_manager;
 
         let raw_table_info = self.table_info().clone();
@@ -276,10 +270,8 @@ impl CreateTableProcedure {
         let region_wal_options = self.region_wal_options()?.clone();
         // Safety: the table_route must be allocated.
         let physical_table_route = self.table_route()?.clone();
-        let detecting_regions = convert_region_routes_to_detecting_regions(
-            cluster_id,
-            &physical_table_route.region_routes,
-        );
+        let detecting_regions =
+            convert_region_routes_to_detecting_regions(&physical_table_route.region_routes);
         let table_route = TableRouteValue::Physical(physical_table_route);
         manager
             .create_table_metadata(raw_table_info, table_route, region_wal_options)
@@ -351,11 +343,10 @@ pub struct TableCreator {
 }
 
 impl TableCreator {
-    pub fn new(cluster_id: ClusterId, task: CreateTableTask) -> Self {
+    pub fn new(task: CreateTableTask) -> Self {
         Self {
             data: CreateTableData {
                 state: CreateTableState::Prepare,
-                cluster_id,
                 task,
                 table_route: None,
                 region_wal_options: None,
@@ -421,7 +412,6 @@ pub struct CreateTableData {
     table_route: Option<PhysicalTableRouteValue>,
     /// None stands for not allocated yet.
     pub region_wal_options: Option<HashMap<RegionNumber, String>>,
-    pub cluster_id: ClusterId,
 }
 
 impl CreateTableData {

src/common/meta/src/ddl/create_view.rs

@@ -24,13 +24,13 @@ use table::table_reference::TableReference;
 
 use crate::cache_invalidator::Context;
 use crate::ddl::utils::handle_retry_error;
-use crate::ddl::{DdlContext, TableMetadata, TableMetadataAllocatorContext};
+use crate::ddl::{DdlContext, TableMetadata};
 use crate::error::{self, Result};
 use crate::instruction::CacheIdent;
 use crate::key::table_name::TableNameKey;
 use crate::lock_key::{CatalogLock, SchemaLock, TableNameLock};
+use crate::metrics;
 use crate::rpc::ddl::CreateViewTask;
-use crate::{metrics, ClusterId};
 
 // The procedure to execute `[CreateViewTask]`.
 pub struct CreateViewProcedure {
@@ -41,12 +41,11 @@ pub struct CreateViewProcedure {
 impl CreateViewProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::CreateView";
 
-    pub fn new(cluster_id: ClusterId, task: CreateViewTask, context: DdlContext) -> Self {
+    pub fn new(task: CreateViewTask, context: DdlContext) -> Self {
         Self {
             context,
             data: CreateViewData {
                 state: CreateViewState::Prepare,
-                cluster_id,
                 task,
                 need_update: false,
             },
@@ -144,12 +143,7 @@ impl CreateViewProcedure {
             let TableMetadata { table_id, .. } = self
                 .context
                 .table_metadata_allocator
-                .create_view(
-                    &TableMetadataAllocatorContext {
-                        cluster_id: self.data.cluster_id,
-                    },
-                    &None,
-                )
+                .create_view(&None)
                 .await?;
             self.data.set_allocated_metadata(table_id, false);
         }
@@ -285,7 +279,6 @@ pub enum CreateViewState {
 pub struct CreateViewData {
     pub state: CreateViewState,
     pub task: CreateViewTask,
-    pub cluster_id: ClusterId,
     /// Whether to update the view info.
     pub need_update: bool,
 }

src/common/meta/src/ddl/drop_database.rs

@@ -35,7 +35,6 @@ use crate::ddl::DdlContext;
 use crate::error::Result;
 use crate::key::table_name::TableNameValue;
 use crate::lock_key::{CatalogLock, SchemaLock};
-use crate::ClusterId;
 
 pub struct DropDatabaseProcedure {
     /// The context of procedure runtime.
@@ -54,7 +53,6 @@ pub(crate) enum DropTableTarget {
 
 /// Context of [DropDatabaseProcedure] execution.
 pub(crate) struct DropDatabaseContext {
-    cluster_id: ClusterId,
     catalog: String,
     schema: String,
     drop_if_exists: bool,
@@ -87,7 +85,6 @@ impl DropDatabaseProcedure {
         Self {
             runtime_context: context,
             context: DropDatabaseContext {
-                cluster_id: 0,
                 catalog,
                 schema,
                 drop_if_exists,
@@ -108,7 +105,6 @@ impl DropDatabaseProcedure {
         Ok(Self {
             runtime_context,
             context: DropDatabaseContext {
-                cluster_id: 0,
                 catalog,
                 schema,
                 drop_if_exists,

src/common/meta/src/ddl/drop_database/cursor.rs

@@ -217,11 +217,10 @@ mod tests {
     async fn test_next_without_logical_tables() {
         let node_manager = Arc::new(MockDatanodeManager::new(()));
         let ddl_context = new_ddl_context(node_manager);
-        create_physical_table(&ddl_context, 0, "phy").await;
+        create_physical_table(&ddl_context, "phy").await;
         // It always starts from Logical
         let mut state = DropDatabaseCursor::new(DropTableTarget::Logical);
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: DEFAULT_CATALOG_NAME.to_string(),
             schema: DEFAULT_SCHEMA_NAME.to_string(),
             drop_if_exists: false,
@@ -252,12 +251,11 @@ mod tests {
     async fn test_next_with_logical_tables() {
         let node_manager = Arc::new(MockDatanodeManager::new(()));
         let ddl_context = new_ddl_context(node_manager);
-        let physical_table_id = create_physical_table(&ddl_context, 0, "phy").await;
-        create_logical_table(ddl_context.clone(), 0, physical_table_id, "metric_0").await;
+        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
+        create_logical_table(ddl_context.clone(), physical_table_id, "metric_0").await;
         // It always starts from Logical
         let mut state = DropDatabaseCursor::new(DropTableTarget::Logical);
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: DEFAULT_CATALOG_NAME.to_string(),
             schema: DEFAULT_SCHEMA_NAME.to_string(),
             drop_if_exists: false,
@@ -286,7 +284,6 @@ mod tests {
         let ddl_context = new_ddl_context(node_manager);
         let mut state = DropDatabaseCursor::new(DropTableTarget::Physical);
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: DEFAULT_CATALOG_NAME.to_string(),
             schema: DEFAULT_SCHEMA_NAME.to_string(),
             drop_if_exists: false,

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

@@ -98,11 +98,10 @@ impl State for DropDatabaseExecutor {
     async fn next(
         &mut self,
         ddl_ctx: &DdlContext,
-        ctx: &mut DropDatabaseContext,
+        _ctx: &mut DropDatabaseContext,
     ) -> Result<(Box<dyn State>, Status)> {
         self.register_dropping_regions(ddl_ctx)?;
-        let executor =
-            DropTableExecutor::new(ctx.cluster_id, self.table_name.clone(), self.table_id, true);
+        let executor = DropTableExecutor::new(self.table_name.clone(), self.table_id, true);
         // Deletes metadata for table permanently.
         let table_route_value = TableRouteValue::new(
             self.table_id,
@@ -187,7 +186,7 @@ mod tests {
     async fn test_next_with_physical_table() {
         let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
         let ddl_context = new_ddl_context(node_manager);
-        let physical_table_id = create_physical_table(&ddl_context, 0, "phy").await;
+        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
         let (_, table_route) = ddl_context
             .table_metadata_manager
             .table_route_manager()
@@ -203,7 +202,6 @@ mod tests {
                 DropTableTarget::Physical,
             );
             let mut ctx = DropDatabaseContext {
-                cluster_id: 0,
                 catalog: DEFAULT_CATALOG_NAME.to_string(),
                 schema: DEFAULT_SCHEMA_NAME.to_string(),
                 drop_if_exists: false,
@@ -216,7 +214,6 @@ mod tests {
         }
         // Execute again
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: DEFAULT_CATALOG_NAME.to_string(),
             schema: DEFAULT_SCHEMA_NAME.to_string(),
             drop_if_exists: false,
@@ -239,8 +236,8 @@ mod tests {
     async fn test_next_logical_table() {
         let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
         let ddl_context = new_ddl_context(node_manager);
-        let physical_table_id = create_physical_table(&ddl_context, 0, "phy").await;
-        create_logical_table(ddl_context.clone(), 0, physical_table_id, "metric").await;
+        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
+        create_logical_table(ddl_context.clone(), physical_table_id, "metric").await;
         let logical_table_id = physical_table_id + 1;
         let (_, table_route) = ddl_context
             .table_metadata_manager
@@ -257,7 +254,6 @@ mod tests {
                 DropTableTarget::Logical,
             );
             let mut ctx = DropDatabaseContext {
-                cluster_id: 0,
                 catalog: DEFAULT_CATALOG_NAME.to_string(),
                 schema: DEFAULT_SCHEMA_NAME.to_string(),
                 drop_if_exists: false,
@@ -270,7 +266,6 @@ mod tests {
         }
         // Execute again
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: DEFAULT_CATALOG_NAME.to_string(),
             schema: DEFAULT_SCHEMA_NAME.to_string(),
             drop_if_exists: false,
@@ -345,7 +340,7 @@ mod tests {
     async fn test_next_retryable_err() {
         let node_manager = Arc::new(MockDatanodeManager::new(RetryErrorDatanodeHandler));
         let ddl_context = new_ddl_context(node_manager);
-        let physical_table_id = create_physical_table(&ddl_context, 0, "phy").await;
+        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
         let (_, table_route) = ddl_context
             .table_metadata_manager
             .table_route_manager()
@@ -360,7 +355,6 @@ mod tests {
             DropTableTarget::Physical,
         );
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: DEFAULT_CATALOG_NAME.to_string(),
             schema: DEFAULT_SCHEMA_NAME.to_string(),
             drop_if_exists: false,
@@ -374,7 +368,7 @@ mod tests {
     async fn test_on_recovery() {
         let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
         let ddl_context = new_ddl_context(node_manager);
-        let physical_table_id = create_physical_table(&ddl_context, 0, "phy").await;
+        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
         let (_, table_route) = ddl_context
             .table_metadata_manager
             .table_route_manager()
@@ -390,7 +384,6 @@ mod tests {
                 DropTableTarget::Physical,
             );
             let mut ctx = DropDatabaseContext {
-                cluster_id: 0,
                 catalog: DEFAULT_CATALOG_NAME.to_string(),
                 schema: DEFAULT_SCHEMA_NAME.to_string(),
                 drop_if_exists: false,

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

@@ -118,7 +118,6 @@ mod tests {
             .unwrap();
         let mut state = DropDatabaseRemoveMetadata;
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: "foo".to_string(),
             schema: "bar".to_string(),
             drop_if_exists: true,
@@ -145,7 +144,6 @@ mod tests {
         // Schema not exists
         let mut state = DropDatabaseRemoveMetadata;
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: "foo".to_string(),
             schema: "bar".to_string(),
             drop_if_exists: true,

src/common/meta/src/ddl/drop_database/start.rs

@@ -89,7 +89,6 @@ mod tests {
         let ddl_context = new_ddl_context(node_manager);
         let mut step = DropDatabaseStart;
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: "foo".to_string(),
             schema: "bar".to_string(),
             drop_if_exists: false,
@@ -105,7 +104,6 @@ mod tests {
         let ddl_context = new_ddl_context(node_manager);
         let mut state = DropDatabaseStart;
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: "foo".to_string(),
             schema: "bar".to_string(),
             drop_if_exists: true,
@@ -128,7 +126,6 @@ mod tests {
             .unwrap();
         let mut state = DropDatabaseStart;
         let mut ctx = DropDatabaseContext {
-            cluster_id: 0,
             catalog: "foo".to_string(),
             schema: "bar".to_string(),
             drop_if_exists: false,

src/common/meta/src/ddl/drop_flow.rs

@@ -37,8 +37,8 @@ use crate::instruction::{CacheIdent, DropFlow};
 use crate::key::flow::flow_info::FlowInfoValue;
 use crate::key::flow::flow_route::FlowRouteValue;
 use crate::lock_key::{CatalogLock, FlowLock};
+use crate::metrics;
 use crate::rpc::ddl::DropFlowTask;
-use crate::{metrics, ClusterId};
 
 /// The procedure for dropping a flow.
 pub struct DropFlowProcedure {
@@ -51,12 +51,11 @@ pub struct DropFlowProcedure {
 impl DropFlowProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::DropFlow";
 
-    pub fn new(cluster_id: ClusterId, task: DropFlowTask, context: DdlContext) -> Self {
+    pub fn new(task: DropFlowTask, context: DdlContext) -> Self {
         Self {
             context,
             data: DropFlowData {
                 state: DropFlowState::Prepare,
-                cluster_id,
                 task,
                 flow_info_value: None,
                 flow_route_values: vec![],
@@ -218,7 +217,6 @@ impl Procedure for DropFlowProcedure {
 #[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct DropFlowData {
     state: DropFlowState,
-    cluster_id: ClusterId,
     task: DropFlowTask,
     pub(crate) flow_info_value: Option<FlowInfoValue>,
     pub(crate) flow_route_values: Vec<FlowRouteValue>,

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

@@ -40,10 +40,10 @@ use crate::ddl::DdlContext;
 use crate::error::{self, Result};
 use crate::key::table_route::TableRouteValue;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
+use crate::metrics;
 use crate::region_keeper::OperatingRegionGuard;
 use crate::rpc::ddl::DropTableTask;
 use crate::rpc::router::{operating_leader_regions, RegionRoute};
-use crate::{metrics, ClusterId};
 
 pub struct DropTableProcedure {
     /// The context of procedure runtime.
@@ -59,8 +59,8 @@ pub struct DropTableProcedure {
 impl DropTableProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::DropTable";
 
-    pub fn new(cluster_id: ClusterId, task: DropTableTask, context: DdlContext) -> Self {
-        let data = DropTableData::new(cluster_id, task);
+    pub fn new(task: DropTableTask, context: DdlContext) -> Self {
+        let data = DropTableData::new(task);
         let executor = data.build_executor();
         Self {
             context,
@@ -268,7 +268,6 @@ impl Procedure for DropTableProcedure {
 #[derive(Debug, Serialize, Deserialize)]
 pub struct DropTableData {
     pub state: DropTableState,
-    pub cluster_id: ClusterId,
     pub task: DropTableTask,
     pub physical_region_routes: Vec<RegionRoute>,
     pub physical_table_id: Option<TableId>,
@@ -279,10 +278,9 @@ pub struct DropTableData {
 }
 
 impl DropTableData {
-    pub fn new(cluster_id: ClusterId, task: DropTableTask) -> Self {
+    pub fn new(task: DropTableTask) -> Self {
         Self {
             state: DropTableState::Prepare,
-            cluster_id,
             task,
             physical_region_routes: vec![],
             physical_table_id: None,
@@ -301,7 +299,6 @@ impl DropTableData {
 
     fn build_executor(&self) -> DropTableExecutor {
         DropTableExecutor::new(
-            self.cluster_id,
             self.task.table_name(),
             self.task.table_id,
             self.task.drop_if_exists,

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

@@ -36,7 +36,6 @@ use crate::instruction::CacheIdent;
 use crate::key::table_name::TableNameKey;
 use crate::key::table_route::TableRouteValue;
 use crate::rpc::router::{find_leader_regions, find_leaders, RegionRoute};
-use crate::ClusterId;
 
 /// [Control] indicated to the caller whether to go to the next step.
 #[derive(Debug)]
@@ -54,14 +53,8 @@ impl<T> Control<T> {
 
 impl DropTableExecutor {
     /// Returns the [DropTableExecutor].
-    pub fn new(
-        cluster_id: ClusterId,
-        table: TableName,
-        table_id: TableId,
-        drop_if_exists: bool,
-    ) -> Self {
+    pub fn new(table: TableName, table_id: TableId, drop_if_exists: bool) -> Self {
         Self {
-            cluster_id,
             table,
             table_id,
             drop_if_exists,
@@ -74,7 +67,6 @@ impl DropTableExecutor {
 /// - Invalidates the cache on the Frontend nodes.
 /// - Drops the regions on the Datanode nodes.
 pub struct DropTableExecutor {
-    cluster_id: ClusterId,
     table: TableName,
     table_id: TableId,
     drop_if_exists: bool,
@@ -164,7 +156,7 @@ impl DropTableExecutor {
         let detecting_regions = if table_route_value.is_physical() {
             // Safety: checked.
             let regions = table_route_value.region_routes().unwrap();
-            convert_region_routes_to_detecting_regions(self.cluster_id, regions)
+            convert_region_routes_to_detecting_regions(regions)
         } else {
             vec![]
         };
@@ -321,7 +313,6 @@ mod tests {
         let node_manager = Arc::new(MockDatanodeManager::new(()));
         let ctx = new_ddl_context(node_manager);
         let executor = DropTableExecutor::new(
-            0,
             TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_table"),
             1024,
             true,
@@ -331,7 +322,6 @@ mod tests {
 
         // Drops a non-exists table
         let executor = DropTableExecutor::new(
-            0,
             TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_table"),
             1024,
             false,
@@ -341,7 +331,6 @@ mod tests {
 
         // Drops a exists table
         let executor = DropTableExecutor::new(
-            0,
             TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_table"),
             1024,
             false,

src/common/meta/src/ddl/drop_view.rs

@@ -31,8 +31,8 @@ use crate::error::{self, Result};
 use crate::instruction::CacheIdent;
 use crate::key::table_name::TableNameKey;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
+use crate::metrics;
 use crate::rpc::ddl::DropViewTask;
-use crate::{metrics, ClusterId};
 
 /// The procedure for dropping a view.
 pub struct DropViewProcedure {
@@ -45,12 +45,11 @@ pub struct DropViewProcedure {
 impl DropViewProcedure {
     pub const TYPE_NAME: &'static str = "metasrv-procedure::DropView";
 
-    pub fn new(cluster_id: ClusterId, task: DropViewTask, context: DdlContext) -> Self {
+    pub fn new(task: DropViewTask, context: DdlContext) -> Self {
         Self {
             context,
             data: DropViewData {
                 state: DropViewState::Prepare,
-                cluster_id,
                 task,
             },
         }
@@ -216,7 +215,6 @@ impl Procedure for DropViewProcedure {
 #[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct DropViewData {
     state: DropViewState,
-    cluster_id: ClusterId,
     task: DropViewTask,
 }
 

src/common/meta/src/ddl/flow_meta.rs

@@ -20,7 +20,6 @@ use crate::error::Result;
 use crate::key::FlowId;
 use crate::peer::Peer;
 use crate::sequence::SequenceRef;
-use crate::ClusterId;
 
 /// The reference of [FlowMetadataAllocator].
 pub type FlowMetadataAllocatorRef = Arc<FlowMetadataAllocator>;
@@ -60,16 +59,9 @@ impl FlowMetadataAllocator {
     }
 
     /// Allocates the [FlowId] and [Peer]s.
-    pub async fn create(
-        &self,
-        cluster_id: ClusterId,
-        partitions: usize,
-    ) -> Result<(FlowId, Vec<Peer>)> {
+    pub async fn create(&self, partitions: usize) -> Result<(FlowId, Vec<Peer>)> {
         let flow_id = self.allocate_flow_id().await?;
-        let peers = self
-            .partition_peer_allocator
-            .alloc(cluster_id, partitions)
-            .await?;
+        let peers = self.partition_peer_allocator.alloc(partitions).await?;
 
         Ok((flow_id, peers))
     }
@@ -79,7 +71,7 @@ impl FlowMetadataAllocator {
 #[async_trait]
 pub trait PartitionPeerAllocator: Send + Sync {
     /// Allocates [Peer] nodes for storing partitions.
-    async fn alloc(&self, cluster_id: ClusterId, partitions: usize) -> Result<Vec<Peer>>;
+    async fn alloc(&self, partitions: usize) -> Result<Vec<Peer>>;
 }
 
 /// [PartitionPeerAllocatorRef] allocates [Peer]s for partitions.
@@ -89,7 +81,7 @@ struct NoopPartitionPeerAllocator;
 
 #[async_trait]
 impl PartitionPeerAllocator for NoopPartitionPeerAllocator {
-    async fn alloc(&self, _cluster_id: ClusterId, partitions: usize) -> Result<Vec<Peer>> {
+    async fn alloc(&self, partitions: usize) -> Result<Vec<Peer>> {
         Ok(vec![Peer::default(); partitions])
     }
 }

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

@@ -20,7 +20,7 @@ use common_telemetry::{debug, info};
 use snafu::ensure;
 use store_api::storage::{RegionId, RegionNumber, TableId};
 
-use crate::ddl::{TableMetadata, TableMetadataAllocatorContext};
+use crate::ddl::TableMetadata;
 use crate::error::{self, Result, UnsupportedSnafu};
 use crate::key::table_route::PhysicalTableRouteValue;
 use crate::peer::Peer;
@@ -109,7 +109,6 @@ impl TableMetadataAllocator {
 
     async fn create_table_route(
         &self,
-        ctx: &TableMetadataAllocatorContext,
         table_id: TableId,
         task: &CreateTableTask,
     ) -> Result<PhysicalTableRouteValue> {
@@ -121,7 +120,7 @@ impl TableMetadataAllocator {
             }
         );
 
-        let peers = self.peer_allocator.alloc(ctx, regions).await?;
+        let peers = self.peer_allocator.alloc(regions).await?;
         let region_routes = task
             .partitions
             .iter()
@@ -147,11 +146,7 @@ impl TableMetadataAllocator {
     }
 
     /// Create VIEW metadata
-    pub async fn create_view(
-        &self,
-        _ctx: &TableMetadataAllocatorContext,
-        table_id: &Option<api::v1::TableId>,
-    ) -> Result<TableMetadata> {
+    pub async fn create_view(&self, table_id: &Option<api::v1::TableId>) -> Result<TableMetadata> {
         let table_id = self.allocate_table_id(table_id).await?;
 
         Ok(TableMetadata {
@@ -160,13 +155,9 @@ impl TableMetadataAllocator {
         })
     }
 
-    pub async fn create(
-        &self,
-        ctx: &TableMetadataAllocatorContext,
-        task: &CreateTableTask,
-    ) -> Result<TableMetadata> {
+    pub async fn create(&self, task: &CreateTableTask) -> Result<TableMetadata> {
         let table_id = self.allocate_table_id(&task.create_table.table_id).await?;
-        let table_route = self.create_table_route(ctx, table_id, task).await?;
+        let table_route = self.create_table_route(table_id, task).await?;
         let region_wal_options = self.create_wal_options(&table_route)?;
 
         debug!(
@@ -188,19 +179,14 @@ pub type PeerAllocatorRef = Arc<dyn PeerAllocator>;
 #[async_trait]
 pub trait PeerAllocator: Send + Sync {
     /// Allocates `regions` size [`Peer`]s.
-    async fn alloc(&self, ctx: &TableMetadataAllocatorContext, regions: usize)
-        -> Result<Vec<Peer>>;
+    async fn alloc(&self, regions: usize) -> Result<Vec<Peer>>;
 }
 
 struct NoopPeerAllocator;
 
 #[async_trait]
 impl PeerAllocator for NoopPeerAllocator {
-    async fn alloc(
-        &self,
-        _ctx: &TableMetadataAllocatorContext,
-        regions: usize,
-    ) -> Result<Vec<Peer>> {
+    async fn alloc(&self, regions: usize) -> Result<Vec<Peer>> {
         Ok(vec![Peer::default(); regions])
     }
 }

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

@@ -31,10 +31,9 @@ use crate::ddl::test_util::columns::TestColumnDefBuilder;
 use crate::ddl::test_util::create_table::{
     build_raw_table_info_from_expr, TestCreateTableExprBuilder,
 };
-use crate::ddl::{DdlContext, TableMetadata, TableMetadataAllocatorContext};
+use crate::ddl::{DdlContext, TableMetadata};
 use crate::key::table_route::TableRouteValue;
 use crate::rpc::ddl::CreateTableTask;
-use crate::ClusterId;
 
 pub async fn create_physical_table_metadata(
     ddl_context: &DdlContext,
@@ -48,11 +47,7 @@ pub async fn create_physical_table_metadata(
         .unwrap();
 }
 
-pub async fn create_physical_table(
-    ddl_context: &DdlContext,
-    cluster_id: ClusterId,
-    name: &str,
-) -> TableId {
+pub async fn create_physical_table(ddl_context: &DdlContext, name: &str) -> TableId {
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task(name);
     let TableMetadata {
@@ -61,10 +56,7 @@ pub async fn create_physical_table(
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -80,15 +72,13 @@ pub async fn create_physical_table(
 
 pub async fn create_logical_table(
     ddl_context: DdlContext,
-    cluster_id: ClusterId,
     physical_table_id: TableId,
     table_name: &str,
 ) -> TableId {
     use std::assert_matches::assert_matches;
 
     let tasks = vec![test_create_logical_table_task(table_name)];
-    let mut procedure =
-        CreateLogicalTablesProcedure::new(cluster_id, tasks, physical_table_id, ddl_context);
+    let mut procedure = CreateLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Executing { persist: true });
     let status = procedure.on_create_metadata().await.unwrap();

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

@@ -86,7 +86,6 @@ fn make_alter_logical_table_rename_task(
 async fn test_on_prepare_check_schema() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let tasks = vec![
         make_alter_logical_table_add_column_task(
             Some("schema1"),
@@ -100,8 +99,7 @@ async fn test_on_prepare_check_schema() {
         ),
     ];
     let physical_table_id = 1024u32;
-    let mut procedure =
-        AlterLogicalTablesProcedure::new(cluster_id, tasks, physical_table_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, AlterLogicalTablesInvalidArguments { .. });
 }
@@ -110,50 +108,46 @@ async fn test_on_prepare_check_schema() {
 async fn test_on_prepare_check_alter_kind() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let tasks = vec![make_alter_logical_table_rename_task(
         "schema1",
         "table1",
         "new_table1",
     )];
     let physical_table_id = 1024u32;
-    let mut procedure =
-        AlterLogicalTablesProcedure::new(cluster_id, tasks, physical_table_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, AlterLogicalTablesInvalidArguments { .. });
 }
 
 #[tokio::test]
 async fn test_on_prepare_different_physical_table() {
-    let cluster_id = 1;
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
 
-    let phy1_id = create_physical_table(&ddl_context, cluster_id, "phy1").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy1_id, "table1").await;
-    let phy2_id = create_physical_table(&ddl_context, cluster_id, "phy2").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy2_id, "table2").await;
+    let phy1_id = create_physical_table(&ddl_context, "phy1").await;
+    create_logical_table(ddl_context.clone(), phy1_id, "table1").await;
+    let phy2_id = create_physical_table(&ddl_context, "phy2").await;
+    create_logical_table(ddl_context.clone(), phy2_id, "table2").await;
 
     let tasks = vec![
         make_alter_logical_table_add_column_task(None, "table1", vec!["column1".to_string()]),
         make_alter_logical_table_add_column_task(None, "table2", vec!["column2".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(cluster_id, tasks, phy1_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy1_id, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, AlterLogicalTablesInvalidArguments { .. });
 }
 
 #[tokio::test]
 async fn test_on_prepare_logical_table_not_exists() {
-    let cluster_id = 1;
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
 
     // Creates physical table
-    let phy_id = create_physical_table(&ddl_context, cluster_id, "phy").await;
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates 3 logical tables
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table1").await;
+    create_logical_table(ddl_context.clone(), phy_id, "table1").await;
 
     let tasks = vec![
         make_alter_logical_table_add_column_task(None, "table1", vec!["column1".to_string()]),
@@ -161,23 +155,22 @@ async fn test_on_prepare_logical_table_not_exists() {
         make_alter_logical_table_add_column_task(None, "table2", vec!["column2".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(cluster_id, tasks, phy_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, TableNotFound { .. });
 }
 
 #[tokio::test]
 async fn test_on_prepare() {
-    let cluster_id = 1;
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
 
     // Creates physical table
-    let phy_id = create_physical_table(&ddl_context, cluster_id, "phy").await;
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates 3 logical tables
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table1").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table2").await;
-    create_logical_table(ddl_context.clone(), cluster_id, 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;
 
     let tasks = vec![
         make_alter_logical_table_add_column_task(None, "table1", vec!["column1".to_string()]),
@@ -185,25 +178,24 @@ async fn test_on_prepare() {
         make_alter_logical_table_add_column_task(None, "table3", vec!["column3".to_string()]),
     ];
 
-    let mut procedure = AlterLogicalTablesProcedure::new(cluster_id, tasks, phy_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
     let result = procedure.on_prepare().await;
     assert_matches!(result, Ok(Status::Executing { persist: true }));
 }
 
 #[tokio::test]
 async fn test_on_update_metadata() {
-    let cluster_id = 1;
     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, cluster_id, "phy").await;
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates 3 logical tables
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table1").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table2").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table3").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table4").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table5").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;
 
     let tasks = vec![
         make_alter_logical_table_add_column_task(None, "table1", vec!["new_col".to_string()]),
@@ -211,7 +203,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(cluster_id, tasks, phy_id, ddl_context);
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context);
     let mut status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Executing { persist: true });
 
@@ -229,23 +221,21 @@ async fn test_on_update_metadata() {
 
 #[tokio::test]
 async fn test_on_part_duplicate_alter_request() {
-    let cluster_id = 1;
     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, cluster_id, "phy").await;
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates 3 logical tables
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table1").await;
-    create_logical_table(ddl_context.clone(), cluster_id, 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()]),
         make_alter_logical_table_add_column_task(None, "table2", vec!["col_0".to_string()]),
     ];
 
-    let mut procedure =
-        AlterLogicalTablesProcedure::new(cluster_id, tasks, phy_id, ddl_context.clone());
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context.clone());
     let mut status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Executing { persist: true });
 
@@ -278,8 +268,7 @@ async fn test_on_part_duplicate_alter_request() {
         ),
     ];
 
-    let mut procedure =
-        AlterLogicalTablesProcedure::new(cluster_id, tasks, phy_id, ddl_context.clone());
+    let mut procedure = AlterLogicalTablesProcedure::new(tasks, phy_id, ddl_context.clone());
     let mut status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Executing { persist: true });
 

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

@@ -59,7 +59,6 @@ fn test_rename_alter_table_task(table_name: &str, new_table_name: &str) -> Alter
 async fn test_on_prepare_table_exists_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_create_table_task("foo", 1024);
     // Puts a value to table name key.
     ddl_context
@@ -73,7 +72,7 @@ async fn test_on_prepare_table_exists_err() {
         .unwrap();
 
     let task = test_rename_alter_table_task("non-exists", "foo");
-    let mut procedure = AlterTableProcedure::new(cluster_id, 1024, task, ddl_context).unwrap();
+    let mut procedure = AlterTableProcedure::new(1024, task, ddl_context).unwrap();
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err.status_code(), StatusCode::TableAlreadyExists);
 }
@@ -82,9 +81,8 @@ async fn test_on_prepare_table_exists_err() {
 async fn test_on_prepare_table_not_exists_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_rename_alter_table_task("non-exists", "foo");
-    let mut procedure = AlterTableProcedure::new(cluster_id, 1024, task, ddl_context).unwrap();
+    let mut procedure = AlterTableProcedure::new(1024, task, ddl_context).unwrap();
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err.status_code(), StatusCode::TableNotFound);
 }
@@ -95,7 +93,6 @@ async fn test_on_submit_alter_request() {
     let datanode_handler = DatanodeWatcher(tx);
     let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let table_id = 1024;
     let table_name = "foo";
     let task = test_create_table_task(table_name, table_id);
@@ -144,8 +141,7 @@ async fn test_on_submit_alter_request() {
             })),
         },
     };
-    let mut procedure =
-        AlterTableProcedure::new(cluster_id, table_id, alter_table_task, ddl_context).unwrap();
+    let mut procedure = AlterTableProcedure::new(table_id, alter_table_task, ddl_context).unwrap();
     procedure.on_prepare().await.unwrap();
     procedure.submit_alter_region_requests().await.unwrap();
 
@@ -181,7 +177,6 @@ async fn test_on_submit_alter_request_with_outdated_request() {
         RequestOutdatedErrorDatanodeHandler,
     ));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let table_id = 1024;
     let table_name = "foo";
     let task = test_create_table_task(table_name, table_id);
@@ -230,8 +225,7 @@ async fn test_on_submit_alter_request_with_outdated_request() {
             })),
         },
     };
-    let mut procedure =
-        AlterTableProcedure::new(cluster_id, table_id, alter_table_task, ddl_context).unwrap();
+    let mut procedure = AlterTableProcedure::new(table_id, alter_table_task, ddl_context).unwrap();
     procedure.on_prepare().await.unwrap();
     procedure.submit_alter_region_requests().await.unwrap();
 }
@@ -240,7 +234,6 @@ async fn test_on_submit_alter_request_with_outdated_request() {
 async fn test_on_update_metadata_rename() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let table_name = "foo";
     let new_table_name = "bar";
     let table_id = 1024;
@@ -257,8 +250,7 @@ async fn test_on_update_metadata_rename() {
         .unwrap();
 
     let task = test_rename_alter_table_task(table_name, new_table_name);
-    let mut procedure =
-        AlterTableProcedure::new(cluster_id, table_id, task, ddl_context.clone()).unwrap();
+    let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context.clone()).unwrap();
     procedure.on_prepare().await.unwrap();
     procedure.on_update_metadata().await.unwrap();
 
@@ -291,7 +283,6 @@ async fn test_on_update_metadata_rename() {
 async fn test_on_update_metadata_add_columns() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let table_name = "foo";
     let table_id = 1024;
     let task = test_create_table_task(table_name, table_id);
@@ -335,8 +326,7 @@ async fn test_on_update_metadata_add_columns() {
             })),
         },
     };
-    let mut procedure =
-        AlterTableProcedure::new(cluster_id, table_id, task, ddl_context.clone()).unwrap();
+    let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context.clone()).unwrap();
     procedure.on_prepare().await.unwrap();
     procedure.submit_alter_region_requests().await.unwrap();
     procedure.on_update_metadata().await.unwrap();
@@ -361,7 +351,6 @@ async fn test_on_update_metadata_add_columns() {
 async fn test_on_update_table_options() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let table_name = "foo";
     let table_id = 1024;
     let task = test_create_table_task(table_name, table_id);
@@ -398,8 +387,7 @@ async fn test_on_update_table_options() {
             })),
         },
     };
-    let mut procedure =
-        AlterTableProcedure::new(cluster_id, table_id, task, ddl_context.clone()).unwrap();
+    let mut procedure = AlterTableProcedure::new(table_id, task, ddl_context.clone()).unwrap();
     procedure.on_prepare().await.unwrap();
     procedure.submit_alter_region_requests().await.unwrap();
     procedure.on_update_metadata().await.unwrap();

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

@@ -25,11 +25,11 @@ use crate::ddl::create_flow::CreateFlowProcedure;
 use crate::ddl::test_util::create_table::test_create_table_task;
 use crate::ddl::test_util::flownode_handler::NaiveFlownodeHandler;
 use crate::ddl::DdlContext;
+use crate::error;
 use crate::key::table_route::TableRouteValue;
 use crate::key::FlowId;
 use crate::rpc::ddl::CreateFlowTask;
 use crate::test_util::{new_ddl_context, MockFlownodeManager};
-use crate::{error, ClusterId};
 
 pub(crate) fn test_create_flow_task(
     name: &str,
@@ -53,7 +53,6 @@ pub(crate) fn test_create_flow_task(
 
 #[tokio::test]
 async fn test_create_flow_source_table_not_found() {
-    let cluster_id = 1;
     let source_table_names = vec![TableName::new(
         DEFAULT_CATALOG_NAME,
         DEFAULT_SCHEMA_NAME,
@@ -65,14 +64,13 @@ async fn test_create_flow_source_table_not_found() {
     let node_manager = Arc::new(MockFlownodeManager::new(NaiveFlownodeHandler));
     let ddl_context = new_ddl_context(node_manager);
     let query_ctx = QueryContext::arc().into();
-    let mut procedure = CreateFlowProcedure::new(cluster_id, task, query_ctx, ddl_context);
+    let mut procedure = CreateFlowProcedure::new(task, query_ctx, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, error::Error::TableNotFound { .. });
 }
 
 pub(crate) async fn create_test_flow(
     ddl_context: &DdlContext,
-    cluster_id: ClusterId,
     flow_name: &str,
     source_table_names: Vec<TableName>,
     sink_table_name: TableName,
@@ -84,8 +82,7 @@ pub(crate) async fn create_test_flow(
         false,
     );
     let query_ctx = QueryContext::arc().into();
-    let mut procedure =
-        CreateFlowProcedure::new(cluster_id, task.clone(), query_ctx, ddl_context.clone());
+    let mut procedure = CreateFlowProcedure::new(task.clone(), query_ctx, ddl_context.clone());
     let output = execute_procedure_until_done(&mut procedure).await.unwrap();
     let flow_id = output.downcast_ref::<FlowId>().unwrap();
 
@@ -94,7 +91,6 @@ pub(crate) async fn create_test_flow(
 
 #[tokio::test]
 async fn test_create_flow() {
-    let cluster_id = 1;
     let table_id = 1024;
     let source_table_names = vec![TableName::new(
         DEFAULT_CATALOG_NAME,
@@ -118,7 +114,6 @@ async fn test_create_flow() {
         .unwrap();
     let flow_id = create_test_flow(
         &ddl_context,
-        cluster_id,
         "my_flow",
         source_table_names.clone(),
         sink_table_name.clone(),
@@ -134,8 +129,7 @@ async fn test_create_flow() {
         true,
     );
     let query_ctx = QueryContext::arc().into();
-    let mut procedure =
-        CreateFlowProcedure::new(cluster_id, task.clone(), query_ctx, ddl_context.clone());
+    let mut procedure = CreateFlowProcedure::new(task.clone(), query_ctx, ddl_context.clone());
     let output = execute_procedure_until_done(&mut procedure).await.unwrap();
     let flow_id = output.downcast_ref::<FlowId>().unwrap();
     assert_eq!(*flow_id, 1024);
@@ -143,7 +137,7 @@ async fn test_create_flow() {
     // Creates again
     let task = test_create_flow_task("my_flow", source_table_names, sink_table_name, false);
     let query_ctx = QueryContext::arc().into();
-    let mut procedure = CreateFlowProcedure::new(cluster_id, task.clone(), query_ctx, ddl_context);
+    let mut procedure = CreateFlowProcedure::new(task.clone(), query_ctx, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, error::Error::FlowAlreadyExists { .. });
 }

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

@@ -26,7 +26,7 @@ use crate::ddl::test_util::datanode_handler::NaiveDatanodeHandler;
 use crate::ddl::test_util::{
     create_physical_table_metadata, test_create_logical_table_task, test_create_physical_table_task,
 };
-use crate::ddl::{TableMetadata, TableMetadataAllocatorContext};
+use crate::ddl::TableMetadata;
 use crate::error::Error;
 use crate::key::table_route::TableRouteValue;
 use crate::test_util::{new_ddl_context, MockDatanodeManager};
@@ -35,11 +35,9 @@ use crate::test_util::{new_ddl_context, MockDatanodeManager};
 async fn test_on_prepare_physical_table_not_found() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let tasks = vec![test_create_logical_table_task("foo")];
     let physical_table_id = 1024u32;
-    let mut procedure =
-        CreateLogicalTablesProcedure::new(cluster_id, tasks, physical_table_id, ddl_context);
+    let mut procedure = CreateLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, Error::TableRouteNotFound { .. });
 }
@@ -48,7 +46,6 @@ async fn test_on_prepare_physical_table_not_found() {
 async fn test_on_prepare() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
     let TableMetadata {
@@ -57,10 +54,7 @@ async fn test_on_prepare() {
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -73,8 +67,7 @@ async fn test_on_prepare() {
     // The create logical table procedure.
     let tasks = vec![test_create_logical_table_task("foo")];
     let physical_table_id = table_id;
-    let mut procedure =
-        CreateLogicalTablesProcedure::new(cluster_id, tasks, physical_table_id, ddl_context);
+    let mut procedure = CreateLogicalTablesProcedure::new(tasks, physical_table_id, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Executing { persist: true });
 }
@@ -83,7 +76,6 @@ async fn test_on_prepare() {
 async fn test_on_prepare_logical_table_exists_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
     let TableMetadata {
@@ -92,10 +84,7 @@ async fn test_on_prepare_logical_table_exists_err() {
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -119,7 +108,7 @@ async fn test_on_prepare_logical_table_exists_err() {
     // The create logical table procedure.
     let physical_table_id = table_id;
     let mut procedure =
-        CreateLogicalTablesProcedure::new(cluster_id, vec![task], physical_table_id, ddl_context);
+        CreateLogicalTablesProcedure::new(vec![task], physical_table_id, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, Error::TableAlreadyExists { .. });
     assert_eq!(err.status_code(), StatusCode::TableAlreadyExists);
@@ -129,7 +118,6 @@ async fn test_on_prepare_logical_table_exists_err() {
 async fn test_on_prepare_with_create_if_table_exists() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
     let TableMetadata {
@@ -138,10 +126,7 @@ async fn test_on_prepare_with_create_if_table_exists() {
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -167,7 +152,7 @@ async fn test_on_prepare_with_create_if_table_exists() {
     // Sets `create_if_not_exists`
     task.create_table.create_if_not_exists = true;
     let mut procedure =
-        CreateLogicalTablesProcedure::new(cluster_id, vec![task], physical_table_id, ddl_context);
+        CreateLogicalTablesProcedure::new(vec![task], physical_table_id, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
     let output = status.downcast_output_ref::<Vec<u32>>().unwrap();
     assert_eq!(*output, vec![8192]);
@@ -177,7 +162,6 @@ async fn test_on_prepare_with_create_if_table_exists() {
 async fn test_on_prepare_part_logical_tables_exist() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
     let TableMetadata {
@@ -186,10 +170,7 @@ async fn test_on_prepare_part_logical_tables_exist() {
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -216,7 +197,6 @@ async fn test_on_prepare_part_logical_tables_exist() {
     task.create_table.create_if_not_exists = true;
     let non_exist_task = test_create_logical_table_task("non_exists");
     let mut procedure = CreateLogicalTablesProcedure::new(
-        cluster_id,
         vec![task, non_exist_task],
         physical_table_id,
         ddl_context,
@@ -229,7 +209,6 @@ async fn test_on_prepare_part_logical_tables_exist() {
 async fn test_on_create_metadata() {
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
     let TableMetadata {
@@ -238,10 +217,7 @@ async fn test_on_create_metadata() {
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -257,7 +233,6 @@ async fn test_on_create_metadata() {
     let task = test_create_logical_table_task("foo");
     let yet_another_task = test_create_logical_table_task("bar");
     let mut procedure = CreateLogicalTablesProcedure::new(
-        cluster_id,
         vec![task, yet_another_task],
         physical_table_id,
         ddl_context,
@@ -279,7 +254,6 @@ async fn test_on_create_metadata() {
 async fn test_on_create_metadata_part_logical_tables_exist() {
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
     let TableMetadata {
@@ -288,10 +262,7 @@ async fn test_on_create_metadata_part_logical_tables_exist() {
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -318,7 +289,6 @@ async fn test_on_create_metadata_part_logical_tables_exist() {
     task.create_table.create_if_not_exists = true;
     let non_exist_task = test_create_logical_table_task("non_exists");
     let mut procedure = CreateLogicalTablesProcedure::new(
-        cluster_id,
         vec![task, non_exist_task],
         physical_table_id,
         ddl_context,
@@ -340,7 +310,6 @@ async fn test_on_create_metadata_part_logical_tables_exist() {
 async fn test_on_create_metadata_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
     let TableMetadata {
@@ -349,10 +318,7 @@ async fn test_on_create_metadata_err() {
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -368,7 +334,6 @@ async fn test_on_create_metadata_err() {
     let task = test_create_logical_table_task("foo");
     let yet_another_task = test_create_logical_table_task("bar");
     let mut procedure = CreateLogicalTablesProcedure::new(
-        cluster_id,
         vec![task.clone(), yet_another_task],
         physical_table_id,
         ddl_context.clone(),

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

@@ -87,7 +87,6 @@ pub(crate) fn test_create_table_task(name: &str) -> CreateTableTask {
 async fn test_on_prepare_table_exists_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_create_table_task("foo");
     assert!(!task.create_table.create_if_not_exists);
     // Puts a value to table name key.
@@ -100,7 +99,7 @@ async fn test_on_prepare_table_exists_err() {
         )
         .await
         .unwrap();
-    let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateTableProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, Error::TableAlreadyExists { .. });
     assert_eq!(err.status_code(), StatusCode::TableAlreadyExists);
@@ -110,7 +109,6 @@ async fn test_on_prepare_table_exists_err() {
 async fn test_on_prepare_with_create_if_table_exists() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let mut task = test_create_table_task("foo");
     task.create_table.create_if_not_exists = true;
     task.table_info.ident.table_id = 1024;
@@ -124,7 +122,7 @@ async fn test_on_prepare_with_create_if_table_exists() {
         )
         .await
         .unwrap();
-    let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateTableProcedure::new(task, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Done { output: Some(..) });
     let table_id = *status.downcast_output_ref::<u32>().unwrap();
@@ -135,10 +133,9 @@ async fn test_on_prepare_with_create_if_table_exists() {
 async fn test_on_prepare_without_create_if_table_exists() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let mut task = test_create_table_task("foo");
     task.create_table.create_if_not_exists = true;
-    let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateTableProcedure::new(task, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Executing { persist: true });
     assert_eq!(procedure.table_id(), 1024);
@@ -148,11 +145,10 @@ async fn test_on_prepare_without_create_if_table_exists() {
 async fn test_on_prepare_with_no_partition_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let mut task = test_create_table_task("foo");
     task.partitions = vec![];
     task.create_table.create_if_not_exists = true;
-    let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateTableProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, Error::Unexpected { .. });
     assert!(err
@@ -165,10 +161,9 @@ async fn test_on_datanode_create_regions_should_retry() {
     common_telemetry::init_default_ut_logging();
     let node_manager = Arc::new(MockDatanodeManager::new(RetryErrorDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_create_table_task("foo");
     assert!(!task.create_table.create_if_not_exists);
-    let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateTableProcedure::new(task, ddl_context);
     procedure.on_prepare().await.unwrap();
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
@@ -183,10 +178,9 @@ async fn test_on_datanode_create_regions_should_not_retry() {
     common_telemetry::init_default_ut_logging();
     let node_manager = Arc::new(MockDatanodeManager::new(UnexpectedErrorDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_create_table_task("foo");
     assert!(!task.create_table.create_if_not_exists);
-    let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateTableProcedure::new(task, ddl_context);
     procedure.on_prepare().await.unwrap();
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
@@ -201,10 +195,9 @@ async fn test_on_create_metadata_error() {
     common_telemetry::init_default_ut_logging();
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_create_table_task("foo");
     assert!(!task.create_table.create_if_not_exists);
-    let mut procedure = CreateTableProcedure::new(cluster_id, task.clone(), ddl_context.clone());
+    let mut procedure = CreateTableProcedure::new(task.clone(), ddl_context.clone());
     procedure.on_prepare().await.unwrap();
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
@@ -233,10 +226,9 @@ async fn test_on_create_metadata() {
     common_telemetry::init_default_ut_logging();
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_create_table_task("foo");
     assert!(!task.create_table.create_if_not_exists);
-    let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateTableProcedure::new(task, ddl_context);
     procedure.on_prepare().await.unwrap();
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
@@ -251,14 +243,12 @@ async fn test_on_create_metadata() {
 
 #[tokio::test]
 async fn test_memory_region_keeper_guard_dropped_on_procedure_done() {
-    let cluster_id = 1;
-
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let kv_backend = Arc::new(MemoryKvBackend::new());
     let ddl_context = new_ddl_context_with_kv_backend(node_manager, kv_backend);
 
     let task = test_create_table_task("foo");
-    let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = CreateTableProcedure::new(task, ddl_context.clone());
 
     execute_procedure_until(&mut procedure, |p| {
         p.creator.data.state == CreateTableState::CreateMetadata

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

@@ -97,7 +97,6 @@ pub(crate) fn test_create_view_task(name: &str) -> CreateViewTask {
 async fn test_on_prepare_view_exists_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_create_view_task("foo");
     assert!(!task.create_view.create_if_not_exists);
     // Puts a value to table name key.
@@ -113,7 +112,7 @@ async fn test_on_prepare_view_exists_err() {
         )
         .await
         .unwrap();
-    let mut procedure = CreateViewProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateViewProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, Error::ViewAlreadyExists { .. });
     assert_eq!(err.status_code(), StatusCode::TableAlreadyExists);
@@ -123,7 +122,6 @@ async fn test_on_prepare_view_exists_err() {
 async fn test_on_prepare_with_create_if_view_exists() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let mut task = test_create_view_task("foo");
     task.create_view.create_if_not_exists = true;
     task.view_info.ident.table_id = 1024;
@@ -140,7 +138,7 @@ async fn test_on_prepare_with_create_if_view_exists() {
         )
         .await
         .unwrap();
-    let mut procedure = CreateViewProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateViewProcedure::new(task, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Done { output: Some(..) });
     let table_id = *status.downcast_output_ref::<u32>().unwrap();
@@ -151,10 +149,9 @@ async fn test_on_prepare_with_create_if_view_exists() {
 async fn test_on_prepare_without_create_if_table_exists() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let mut task = test_create_view_task("foo");
     task.create_view.create_if_not_exists = true;
-    let mut procedure = CreateViewProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateViewProcedure::new(task, ddl_context);
     let status = procedure.on_prepare().await.unwrap();
     assert_matches!(status, Status::Executing { persist: true });
     assert_eq!(procedure.view_id(), 1024);
@@ -165,10 +162,9 @@ async fn test_on_create_metadata() {
     common_telemetry::init_default_ut_logging();
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let task = test_create_view_task("foo");
     assert!(!task.create_view.create_if_not_exists);
-    let mut procedure = CreateViewProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = CreateViewProcedure::new(task, ddl_context);
     procedure.on_prepare().await.unwrap();
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
@@ -185,10 +181,9 @@ async fn test_replace_view_metadata() {
     common_telemetry::init_default_ut_logging();
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager.clone());
-    let cluster_id = 1;
     let task = test_create_view_task("foo");
     assert!(!task.create_view.create_if_not_exists);
-    let mut procedure = CreateViewProcedure::new(cluster_id, task.clone(), ddl_context.clone());
+    let mut procedure = CreateViewProcedure::new(task.clone(), ddl_context.clone());
     procedure.on_prepare().await.unwrap();
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
@@ -213,7 +208,7 @@ async fn test_replace_view_metadata() {
     let mut task = test_create_view_task("foo");
     // The view already exists, prepare should fail
     {
-        let mut procedure = CreateViewProcedure::new(cluster_id, task.clone(), ddl_context.clone());
+        let mut procedure = CreateViewProcedure::new(task.clone(), ddl_context.clone());
         let err = procedure.on_prepare().await.unwrap_err();
         assert_matches!(err, Error::ViewAlreadyExists { .. });
         assert_eq!(err.status_code(), StatusCode::TableAlreadyExists);
@@ -224,7 +219,7 @@ async fn test_replace_view_metadata() {
     task.create_view.logical_plan = vec![4, 5, 6];
     task.create_view.definition = "new_definition".to_string();
 
-    let mut procedure = CreateViewProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = CreateViewProcedure::new(task, ddl_context.clone());
     procedure.on_prepare().await.unwrap();
     let ctx = ProcedureContext {
         procedure_id: ProcedureId::random(),
@@ -254,12 +249,11 @@ async fn test_replace_table() {
     common_telemetry::init_default_ut_logging();
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager.clone());
-    let cluster_id = 1;
 
     {
         // Create a `foo` table.
         let task = test_create_table_task("foo");
-        let mut procedure = CreateTableProcedure::new(cluster_id, task, ddl_context.clone());
+        let mut procedure = CreateTableProcedure::new(task, ddl_context.clone());
         procedure.on_prepare().await.unwrap();
         let ctx = ProcedureContext {
             procedure_id: ProcedureId::random(),
@@ -272,7 +266,7 @@ async fn test_replace_table() {
     // Try to replace a view named `foo` too.
     let mut task = test_create_view_task("foo");
     task.create_view.or_replace = true;
-    let mut procedure = CreateViewProcedure::new(cluster_id, task.clone(), ddl_context.clone());
+    let mut procedure = CreateViewProcedure::new(task.clone(), ddl_context.clone());
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, Error::TableAlreadyExists { .. });
     assert_eq!(err.status_code(), StatusCode::TableAlreadyExists);

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

@@ -31,7 +31,6 @@ use crate::test_util::{new_ddl_context, MockDatanodeManager};
 #[tokio::test]
 async fn test_drop_database_with_logical_tables() {
     common_telemetry::init_default_ut_logging();
-    let cluster_id = 1;
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
     ddl_context
@@ -45,11 +44,11 @@ async fn test_drop_database_with_logical_tables() {
         .await
         .unwrap();
     // Creates physical table
-    let phy_id = create_physical_table(&ddl_context, cluster_id, "phy").await;
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates 3 logical tables
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table1").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table2").await;
-    create_logical_table(ddl_context.clone(), cluster_id, 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;
 
     let mut procedure = DropDatabaseProcedure::new(
         DEFAULT_CATALOG_NAME.to_string(),
@@ -80,7 +79,6 @@ async fn test_drop_database_with_logical_tables() {
 #[tokio::test]
 async fn test_drop_database_retryable_error() {
     common_telemetry::init_default_ut_logging();
-    let cluster_id = 1;
     let node_manager = Arc::new(MockDatanodeManager::new(RetryErrorDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
     ddl_context
@@ -94,11 +92,11 @@ async fn test_drop_database_retryable_error() {
         .await
         .unwrap();
     // Creates physical table
-    let phy_id = create_physical_table(&ddl_context, cluster_id, "phy").await;
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates 3 logical tables
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table1").await;
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table2").await;
-    create_logical_table(ddl_context.clone(), cluster_id, 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;
 
     let mut procedure = DropDatabaseProcedure::new(
         DEFAULT_CATALOG_NAME.to_string(),
@@ -128,7 +126,6 @@ async fn test_drop_database_retryable_error() {
 #[tokio::test]
 async fn test_drop_database_recover() {
     common_telemetry::init_default_ut_logging();
-    let cluster_id = 1;
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let ddl_context = new_ddl_context(node_manager);
     ddl_context
@@ -142,9 +139,9 @@ async fn test_drop_database_recover() {
         .await
         .unwrap();
     // Creates a physical table
-    let phy_id = create_physical_table(&ddl_context, cluster_id, "phy").await;
+    let phy_id = create_physical_table(&ddl_context, "phy").await;
     // Creates a logical tables
-    create_logical_table(ddl_context.clone(), cluster_id, phy_id, "table1").await;
+    create_logical_table(ddl_context.clone(), phy_id, "table1").await;
     let mut procedure = DropDatabaseProcedure::new(
         DEFAULT_CATALOG_NAME.to_string(),
         DEFAULT_SCHEMA_NAME.to_string(),

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

@@ -40,12 +40,11 @@ fn test_drop_flow_task(flow_name: &str, flow_id: u32, drop_if_exists: bool) -> D
 
 #[tokio::test]
 async fn test_drop_flow_not_found() {
-    let cluster_id = 1;
     let flow_id = 1024;
     let node_manager = Arc::new(MockFlownodeManager::new(NaiveFlownodeHandler));
     let ddl_context = new_ddl_context(node_manager);
     let task = test_drop_flow_task("my_flow", flow_id, false);
-    let mut procedure = DropFlowProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropFlowProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, error::Error::FlowNotFound { .. });
 }
@@ -53,7 +52,6 @@ async fn test_drop_flow_not_found() {
 #[tokio::test]
 async fn test_drop_flow() {
     // create a flow
-    let cluster_id = 1;
     let table_id = 1024;
     let source_table_names = vec![TableName::new(
         DEFAULT_CATALOG_NAME,
@@ -75,27 +73,21 @@ async fn test_drop_flow() {
         )
         .await
         .unwrap();
-    let flow_id = create_test_flow(
-        &ddl_context,
-        cluster_id,
-        "my_flow",
-        source_table_names,
-        sink_table_name,
-    )
-    .await;
+    let flow_id =
+        create_test_flow(&ddl_context, "my_flow", source_table_names, sink_table_name).await;
     // Drops the flows
     let task = test_drop_flow_task("my_flow", flow_id, false);
-    let mut procedure = DropFlowProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = DropFlowProcedure::new(task, ddl_context.clone());
     execute_procedure_until_done(&mut procedure).await;
 
     // Drops if not exists
     let task = test_drop_flow_task("my_flow", flow_id, true);
-    let mut procedure = DropFlowProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = DropFlowProcedure::new(task, ddl_context.clone());
     execute_procedure_until_done(&mut procedure).await;
 
     // Drops again
     let task = test_drop_flow_task("my_flow", flow_id, false);
-    let mut procedure = DropFlowProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropFlowProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_matches!(err, error::Error::FlowNotFound { .. });
 }

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

@@ -35,7 +35,7 @@ use crate::ddl::test_util::{
     create_logical_table, create_physical_table, create_physical_table_metadata,
     test_create_logical_table_task, test_create_physical_table_task,
 };
-use crate::ddl::{TableMetadata, TableMetadataAllocatorContext};
+use crate::ddl::TableMetadata;
 use crate::key::table_route::TableRouteValue;
 use crate::kv_backend::memory::MemoryKvBackend;
 use crate::peer::Peer;
@@ -47,7 +47,6 @@ use crate::test_util::{new_ddl_context, new_ddl_context_with_kv_backend, MockDat
 async fn test_on_prepare_table_not_exists_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let table_name = "foo";
     let table_id = 1024;
     let task = test_create_table_task(table_name, table_id);
@@ -63,7 +62,7 @@ async fn test_on_prepare_table_not_exists_err() {
         .unwrap();
 
     let task = new_drop_table_task("bar", table_id, false);
-    let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropTableProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_eq!(err.status_code(), StatusCode::TableNotFound);
 }
@@ -72,7 +71,6 @@ async fn test_on_prepare_table_not_exists_err() {
 async fn test_on_prepare_table() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let table_name = "foo";
     let table_id = 1024;
     let task = test_create_table_task(table_name, table_id);
@@ -89,13 +87,13 @@ async fn test_on_prepare_table() {
 
     let task = new_drop_table_task("bar", table_id, true);
     // Drop if exists
-    let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = DropTableProcedure::new(task, ddl_context.clone());
     procedure.on_prepare().await.unwrap();
     assert!(!procedure.rollback_supported());
 
     let task = new_drop_table_task(table_name, table_id, false);
     // Drop table
-    let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropTableProcedure::new(task, ddl_context);
     procedure.on_prepare().await.unwrap();
 }
 
@@ -105,7 +103,6 @@ async fn test_on_datanode_drop_regions() {
     let datanode_handler = DatanodeWatcher(tx);
     let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let table_id = 1024;
     let table_name = "foo";
     let task = test_create_table_task(table_name, table_id);
@@ -144,7 +141,7 @@ async fn test_on_datanode_drop_regions() {
 
     let task = new_drop_table_task(table_name, table_id, false);
     // Drop table
-    let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropTableProcedure::new(task, ddl_context);
     procedure.on_prepare().await.unwrap();
     procedure.on_datanode_drop_regions().await.unwrap();
 
@@ -179,7 +176,6 @@ async fn test_on_rollback() {
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let kv_backend = Arc::new(MemoryKvBackend::new());
     let ddl_context = new_ddl_context_with_kv_backend(node_manager, kv_backend.clone());
-    let cluster_id = 1;
     // Prepares physical table metadata.
     let mut create_physical_table_task = test_create_physical_table_task("phy_table");
     let TableMetadata {
@@ -188,10 +184,7 @@ async fn test_on_rollback() {
         ..
     } = ddl_context
         .table_metadata_allocator
-        .create(
-            &TableMetadataAllocatorContext { cluster_id },
-            &create_physical_table_task,
-        )
+        .create(&create_physical_table_task)
         .await
         .unwrap();
     create_physical_table_task.set_table_id(table_id);
@@ -205,12 +198,8 @@ async fn test_on_rollback() {
     let physical_table_id = table_id;
     // Creates the logical table metadata.
     let task = test_create_logical_table_task("foo");
-    let mut procedure = CreateLogicalTablesProcedure::new(
-        cluster_id,
-        vec![task],
-        physical_table_id,
-        ddl_context.clone(),
-    );
+    let mut procedure =
+        CreateLogicalTablesProcedure::new(vec![task], physical_table_id, ddl_context.clone());
     procedure.on_prepare().await.unwrap();
     let ctx = new_test_procedure_context();
     procedure.execute(&ctx).await.unwrap();
@@ -223,7 +212,7 @@ async fn test_on_rollback() {
     // Drops the physical table
     {
         let task = new_drop_table_task("phy_table", physical_table_id, false);
-        let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context.clone());
+        let mut procedure = DropTableProcedure::new(task, ddl_context.clone());
         procedure.on_prepare().await.unwrap();
         assert!(procedure.rollback_supported());
         procedure.on_delete_metadata().await.unwrap();
@@ -238,7 +227,7 @@ async fn test_on_rollback() {
 
     // Drops the logical table
     let task = new_drop_table_task("foo", table_ids[0], false);
-    let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = DropTableProcedure::new(task, ddl_context.clone());
     procedure.on_prepare().await.unwrap();
     assert!(!procedure.rollback_supported());
 }
@@ -255,18 +244,15 @@ fn new_drop_table_task(table_name: &str, table_id: TableId, drop_if_exists: bool
 
 #[tokio::test]
 async fn test_memory_region_keeper_guard_dropped_on_procedure_done() {
-    let cluster_id = 1;
-
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let kv_backend = Arc::new(MemoryKvBackend::new());
     let ddl_context = new_ddl_context_with_kv_backend(node_manager, kv_backend);
 
-    let physical_table_id = create_physical_table(&ddl_context, cluster_id, "t").await;
-    let logical_table_id =
-        create_logical_table(ddl_context.clone(), cluster_id, physical_table_id, "s").await;
+    let physical_table_id = create_physical_table(&ddl_context, "t").await;
+    let logical_table_id = create_logical_table(ddl_context.clone(), physical_table_id, "s").await;
 
     let inner_test = |task: DropTableTask| async {
-        let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context.clone());
+        let mut procedure = DropTableProcedure::new(task, ddl_context.clone());
         execute_procedure_until(&mut procedure, |p| {
             p.data.state == DropTableState::InvalidateTableCache
         })
@@ -304,14 +290,13 @@ async fn test_from_json() {
         (DropTableState::DatanodeDropRegions, 1, 1),
         (DropTableState::DeleteTombstone, 1, 0),
     ] {
-        let cluster_id = 1;
         let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
         let kv_backend = Arc::new(MemoryKvBackend::new());
         let ddl_context = new_ddl_context_with_kv_backend(node_manager, kv_backend);
 
-        let physical_table_id = create_physical_table(&ddl_context, cluster_id, "t").await;
+        let physical_table_id = create_physical_table(&ddl_context, "t").await;
         let task = new_drop_table_task("t", physical_table_id, false);
-        let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context.clone());
+        let mut procedure = DropTableProcedure::new(task, ddl_context.clone());
         execute_procedure_until(&mut procedure, |p| p.data.state == state).await;
         let data = procedure.dump().unwrap();
         assert_eq!(
@@ -334,14 +319,13 @@ async fn test_from_json() {
 
     let num_operating_regions = 0;
     let num_operating_regions_after_recovery = 0;
-    let cluster_id = 1;
     let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
     let kv_backend = Arc::new(MemoryKvBackend::new());
     let ddl_context = new_ddl_context_with_kv_backend(node_manager, kv_backend);
 
-    let physical_table_id = create_physical_table(&ddl_context, cluster_id, "t").await;
+    let physical_table_id = create_physical_table(&ddl_context, "t").await;
     let task = new_drop_table_task("t", physical_table_id, false);
-    let mut procedure = DropTableProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = DropTableProcedure::new(task, ddl_context.clone());
     execute_procedure_until_done(&mut procedure).await;
     let data = procedure.dump().unwrap();
     assert_eq!(

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

@@ -41,7 +41,6 @@ fn new_drop_view_task(view: &str, view_id: TableId, drop_if_exists: bool) -> Dro
 async fn test_on_prepare_view_not_exists_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let view_id = 1024;
     let mut task = test_create_view_task("foo");
     task.view_info.ident.table_id = view_id;
@@ -60,7 +59,7 @@ async fn test_on_prepare_view_not_exists_err() {
         .unwrap();
 
     let task = new_drop_view_task("bar", view_id, false);
-    let mut procedure = DropViewProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropViewProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_eq!(err.status_code(), StatusCode::TableNotFound);
 }
@@ -69,7 +68,6 @@ async fn test_on_prepare_view_not_exists_err() {
 async fn test_on_prepare_not_view_err() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let view_id = 1024;
     let view_name = "foo";
     let task = test_create_table_task(view_name, view_id);
@@ -85,7 +83,7 @@ async fn test_on_prepare_not_view_err() {
         .unwrap();
 
     let task = new_drop_view_task(view_name, view_id, false);
-    let mut procedure = DropViewProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropViewProcedure::new(task, ddl_context);
     // It's not a view, expect error
     let err = procedure.on_prepare().await.unwrap_err();
     assert_eq!(err.status_code(), StatusCode::InvalidArguments);
@@ -95,7 +93,6 @@ async fn test_on_prepare_not_view_err() {
 async fn test_on_prepare_success() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let view_id = 1024;
     let view_name = "foo";
     let mut task = test_create_view_task("foo");
@@ -116,12 +113,12 @@ async fn test_on_prepare_success() {
 
     let task = new_drop_view_task("bar", view_id, true);
     // Drop if exists
-    let mut procedure = DropViewProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = DropViewProcedure::new(task, ddl_context.clone());
     procedure.on_prepare().await.unwrap();
 
     let task = new_drop_view_task(view_name, view_id, false);
     // Prepare success
-    let mut procedure = DropViewProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropViewProcedure::new(task, ddl_context);
     procedure.on_prepare().await.unwrap();
     assert_eq!(DropViewState::DeleteMetadata, procedure.state());
 }
@@ -130,7 +127,6 @@ async fn test_on_prepare_success() {
 async fn test_drop_view_success() {
     let node_manager = Arc::new(MockDatanodeManager::new(()));
     let ddl_context = new_ddl_context(node_manager);
-    let cluster_id = 1;
     let view_id = 1024;
     let view_name = "foo";
     let mut task = test_create_view_task("foo");
@@ -159,7 +155,7 @@ async fn test_drop_view_success() {
 
     let task = new_drop_view_task(view_name, view_id, false);
     // Prepare success
-    let mut procedure = DropViewProcedure::new(cluster_id, task, ddl_context.clone());
+    let mut procedure = DropViewProcedure::new(task, ddl_context.clone());
     execute_procedure_until_done(&mut procedure).await;
     assert_eq!(DropViewState::InvalidateViewCache, procedure.state());
 
@@ -174,7 +170,7 @@ async fn test_drop_view_success() {
 
     // Drop again
     let task = new_drop_view_task(view_name, view_id, false);
-    let mut procedure = DropViewProcedure::new(cluster_id, task, ddl_context);
+    let mut procedure = DropViewProcedure::new(task, ddl_context);
     let err = procedure.on_prepare().await.unwrap_err();
     assert_eq!(err.status_code(), StatusCode::TableNotFound);
 }

src/common/meta/src/ddl/truncate_table.rs

@@ -39,9 +39,9 @@ use crate::key::table_info::TableInfoValue;
 use crate::key::table_name::TableNameKey;
 use crate::key::DeserializedValueWithBytes;
 use crate::lock_key::{CatalogLock, SchemaLock, TableLock};
+use crate::metrics;
 use crate::rpc::ddl::TruncateTableTask;
 use crate::rpc::router::{find_leader_regions, find_leaders, RegionRoute};
-use crate::{metrics, ClusterId};
 
 pub struct TruncateTableProcedure {
     context: DdlContext,
@@ -91,7 +91,6 @@ impl TruncateTableProcedure {
     pub(crate) const TYPE_NAME: &'static str = "metasrv-procedure::TruncateTable";
 
     pub(crate) fn new(
-        cluster_id: ClusterId,
         task: TruncateTableTask,
         table_info_value: DeserializedValueWithBytes<TableInfoValue>,
         region_routes: Vec<RegionRoute>,
@@ -99,7 +98,7 @@ impl TruncateTableProcedure {
     ) -> Self {
         Self {
             context,
-            data: TruncateTableData::new(cluster_id, task, table_info_value, region_routes),
+            data: TruncateTableData::new(task, table_info_value, region_routes),
         }
     }
 
@@ -189,7 +188,6 @@ impl TruncateTableProcedure {
 #[derive(Debug, Serialize, Deserialize)]
 pub struct TruncateTableData {
     state: TruncateTableState,
-    cluster_id: ClusterId,
     task: TruncateTableTask,
     table_info_value: DeserializedValueWithBytes<TableInfoValue>,
     region_routes: Vec<RegionRoute>,
@@ -197,14 +195,12 @@ pub struct TruncateTableData {
 
 impl TruncateTableData {
     pub fn new(
-        cluster_id: ClusterId,
         task: TruncateTableTask,
         table_info_value: DeserializedValueWithBytes<TableInfoValue>,
         region_routes: Vec<RegionRoute>,
     ) -> Self {
         Self {
             state: TruncateTableState::Prepare,
-            cluster_id,
             task,
             table_info_value,
             region_routes,

src/common/meta/src/ddl/utils.rs

@@ -34,7 +34,6 @@ use crate::key::TableMetadataManagerRef;
 use crate::peer::Peer;
 use crate::rpc::ddl::CreateTableTask;
 use crate::rpc::router::RegionRoute;
-use crate::ClusterId;
 
 /// Adds [Peer] context if the error is unretryable.
 pub fn add_peer_context_if_needed(datanode: Peer) -> impl FnOnce(Error) -> Error {
@@ -144,7 +143,6 @@ pub async fn get_physical_table_id(
 
 /// Converts a list of [`RegionRoute`] to a list of [`DetectingRegion`].
 pub fn convert_region_routes_to_detecting_regions(
-    cluster_id: ClusterId,
     region_routes: &[RegionRoute],
 ) -> Vec<DetectingRegion> {
     region_routes
@@ -153,7 +151,7 @@ pub fn convert_region_routes_to_detecting_regions(
             route
                 .leader_peer
                 .as_ref()
-                .map(|peer| (cluster_id, peer.id, route.region.id))
+                .map(|peer| (peer.id, route.region.id))
         })
         .collect::<Vec<_>>()
 }

src/common/meta/src/ddl_manager.rs

@@ -60,7 +60,6 @@ use crate::rpc::ddl::{
 use crate::rpc::procedure;
 use crate::rpc::procedure::{MigrateRegionRequest, MigrateRegionResponse, ProcedureStateResponse};
 use crate::rpc::router::RegionRoute;
-use crate::ClusterId;
 
 pub type DdlManagerRef = Arc<DdlManager>;
 
@@ -154,13 +153,12 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_alter_table_task(
         &self,
-        cluster_id: ClusterId,
         table_id: TableId,
         alter_table_task: AlterTableTask,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
 
-        let procedure = AlterTableProcedure::new(cluster_id, table_id, alter_table_task, context)?;
+        let procedure = AlterTableProcedure::new(table_id, alter_table_task, context)?;
 
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
@@ -171,12 +169,11 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_create_table_task(
         &self,
-        cluster_id: ClusterId,
         create_table_task: CreateTableTask,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
 
-        let procedure = CreateTableProcedure::new(cluster_id, create_table_task, context);
+        let procedure = CreateTableProcedure::new(create_table_task, context);
 
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
@@ -187,12 +184,11 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_create_view_task(
         &self,
-        cluster_id: ClusterId,
         create_view_task: CreateViewTask,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
 
-        let procedure = CreateViewProcedure::new(cluster_id, create_view_task, context);
+        let procedure = CreateViewProcedure::new(create_view_task, context);
 
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
@@ -203,18 +199,13 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_create_logical_table_tasks(
         &self,
-        cluster_id: ClusterId,
         create_table_tasks: Vec<CreateTableTask>,
         physical_table_id: TableId,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
 
-        let procedure = CreateLogicalTablesProcedure::new(
-            cluster_id,
-            create_table_tasks,
-            physical_table_id,
-            context,
-        );
+        let procedure =
+            CreateLogicalTablesProcedure::new(create_table_tasks, physical_table_id, context);
 
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
@@ -225,18 +216,13 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_alter_logical_table_tasks(
         &self,
-        cluster_id: ClusterId,
         alter_table_tasks: Vec<AlterTableTask>,
         physical_table_id: TableId,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
 
-        let procedure = AlterLogicalTablesProcedure::new(
-            cluster_id,
-            alter_table_tasks,
-            physical_table_id,
-            context,
-        );
+        let procedure =
+            AlterLogicalTablesProcedure::new(alter_table_tasks, physical_table_id, context);
 
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
@@ -247,12 +233,11 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_drop_table_task(
         &self,
-        cluster_id: ClusterId,
         drop_table_task: DropTableTask,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
 
-        let procedure = DropTableProcedure::new(cluster_id, drop_table_task, context);
+        let procedure = DropTableProcedure::new(drop_table_task, context);
 
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
@@ -263,7 +248,6 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_create_database(
         &self,
-        _cluster_id: ClusterId,
         CreateDatabaseTask {
             catalog,
             schema,
@@ -283,7 +267,6 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_drop_database(
         &self,
-        _cluster_id: ClusterId,
         DropDatabaseTask {
             catalog,
             schema,
@@ -299,11 +282,10 @@ impl DdlManager {
 
     pub async fn submit_alter_database(
         &self,
-        cluster_id: ClusterId,
         alter_database_task: AlterDatabaseTask,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
-        let procedure = AlterDatabaseProcedure::new(cluster_id, alter_database_task, context)?;
+        let procedure = AlterDatabaseProcedure::new(alter_database_task, context)?;
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
         self.submit_procedure(procedure_with_id).await
@@ -313,12 +295,11 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_create_flow_task(
         &self,
-        cluster_id: ClusterId,
         create_flow: CreateFlowTask,
         query_context: QueryContext,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
-        let procedure = CreateFlowProcedure::new(cluster_id, create_flow, query_context, context);
+        let procedure = CreateFlowProcedure::new(create_flow, query_context, context);
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
         self.submit_procedure(procedure_with_id).await
@@ -328,11 +309,10 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_drop_flow_task(
         &self,
-        cluster_id: ClusterId,
         drop_flow: DropFlowTask,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
-        let procedure = DropFlowProcedure::new(cluster_id, drop_flow, context);
+        let procedure = DropFlowProcedure::new(drop_flow, context);
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
         self.submit_procedure(procedure_with_id).await
@@ -342,11 +322,10 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_drop_view_task(
         &self,
-        cluster_id: ClusterId,
         drop_view: DropViewTask,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
-        let procedure = DropViewProcedure::new(cluster_id, drop_view, context);
+        let procedure = DropViewProcedure::new(drop_view, context);
         let procedure_with_id = ProcedureWithId::with_random_id(Box::new(procedure));
 
         self.submit_procedure(procedure_with_id).await
@@ -356,14 +335,12 @@ impl DdlManager {
     #[tracing::instrument(skip_all)]
     pub async fn submit_truncate_table_task(
         &self,
-        cluster_id: ClusterId,
         truncate_table_task: TruncateTableTask,
         table_info_value: DeserializedValueWithBytes<TableInfoValue>,
         region_routes: Vec<RegionRoute>,
     ) -> Result<(ProcedureId, Option<Output>)> {
         let context = self.create_context();
         let procedure = TruncateTableProcedure::new(
-            cluster_id,
             truncate_table_task,
             table_info_value,
             region_routes,
@@ -397,7 +374,6 @@ impl DdlManager {
 
 async fn handle_truncate_table_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     truncate_table_task: TruncateTableTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let table_id = truncate_table_task.table_id;
@@ -416,12 +392,7 @@ async fn handle_truncate_table_task(
     let table_route = table_route_value.into_inner().region_routes()?.clone();
 
     let (id, _) = ddl_manager
-        .submit_truncate_table_task(
-            cluster_id,
-            truncate_table_task,
-            table_info_value,
-            table_route,
-        )
+        .submit_truncate_table_task(truncate_table_task, table_info_value, table_route)
         .await?;
 
     info!("Table: {table_id} is truncated via procedure_id {id:?}");
@@ -434,7 +405,6 @@ async fn handle_truncate_table_task(
 
 async fn handle_alter_table_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     alter_table_task: AlterTableTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let table_ref = alter_table_task.table_ref();
@@ -468,7 +438,7 @@ async fn handle_alter_table_task(
     );
 
     let (id, _) = ddl_manager
-        .submit_alter_table_task(cluster_id, table_id, alter_table_task)
+        .submit_alter_table_task(table_id, alter_table_task)
         .await?;
 
     info!("Table: {table_id} is altered via procedure_id {id:?}");
@@ -481,13 +451,10 @@ async fn handle_alter_table_task(
 
 async fn handle_drop_table_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     drop_table_task: DropTableTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let table_id = drop_table_task.table_id;
-    let (id, _) = ddl_manager
-        .submit_drop_table_task(cluster_id, drop_table_task)
-        .await?;
+    let (id, _) = ddl_manager.submit_drop_table_task(drop_table_task).await?;
 
     info!("Table: {table_id} is dropped via procedure_id {id:?}");
 
@@ -499,11 +466,10 @@ async fn handle_drop_table_task(
 
 async fn handle_create_table_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     create_table_task: CreateTableTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let (id, output) = ddl_manager
-        .submit_create_table_task(cluster_id, create_table_task)
+        .submit_create_table_task(create_table_task)
         .await?;
 
     let procedure_id = id.to_string();
@@ -525,7 +491,6 @@ async fn handle_create_table_task(
 
 async fn handle_create_logical_table_tasks(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     create_table_tasks: Vec<CreateTableTask>,
 ) -> Result<SubmitDdlTaskResponse> {
     ensure!(
@@ -542,7 +507,7 @@ async fn handle_create_logical_table_tasks(
     let num_logical_tables = create_table_tasks.len();
 
     let (id, output) = ddl_manager
-        .submit_create_logical_table_tasks(cluster_id, create_table_tasks, physical_table_id)
+        .submit_create_logical_table_tasks(create_table_tasks, physical_table_id)
         .await?;
 
     info!("{num_logical_tables} logical tables on physical table: {physical_table_id:?} is created via procedure_id {id:?}");
@@ -568,11 +533,10 @@ async fn handle_create_logical_table_tasks(
 
 async fn handle_create_database_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     create_database_task: CreateDatabaseTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let (id, _) = ddl_manager
-        .submit_create_database(cluster_id, create_database_task.clone())
+        .submit_create_database(create_database_task.clone())
         .await?;
 
     let procedure_id = id.to_string();
@@ -589,11 +553,10 @@ async fn handle_create_database_task(
 
 async fn handle_drop_database_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     drop_database_task: DropDatabaseTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let (id, _) = ddl_manager
-        .submit_drop_database(cluster_id, drop_database_task.clone())
+        .submit_drop_database(drop_database_task.clone())
         .await?;
 
     let procedure_id = id.to_string();
@@ -610,11 +573,10 @@ async fn handle_drop_database_task(
 
 async fn handle_alter_database_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     alter_database_task: AlterDatabaseTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let (id, _) = ddl_manager
-        .submit_alter_database(cluster_id, alter_database_task.clone())
+        .submit_alter_database(alter_database_task.clone())
         .await?;
 
     let procedure_id = id.to_string();
@@ -632,11 +594,10 @@ async fn handle_alter_database_task(
 
 async fn handle_drop_flow_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     drop_flow_task: DropFlowTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let (id, _) = ddl_manager
-        .submit_drop_flow_task(cluster_id, drop_flow_task.clone())
+        .submit_drop_flow_task(drop_flow_task.clone())
         .await?;
 
     let procedure_id = id.to_string();
@@ -653,11 +614,10 @@ async fn handle_drop_flow_task(
 
 async fn handle_drop_view_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     drop_view_task: DropViewTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let (id, _) = ddl_manager
-        .submit_drop_view_task(cluster_id, drop_view_task.clone())
+        .submit_drop_view_task(drop_view_task.clone())
         .await?;
 
     let procedure_id = id.to_string();
@@ -675,12 +635,11 @@ async fn handle_drop_view_task(
 
 async fn handle_create_flow_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     create_flow_task: CreateFlowTask,
     query_context: QueryContext,
 ) -> Result<SubmitDdlTaskResponse> {
     let (id, output) = ddl_manager
-        .submit_create_flow_task(cluster_id, create_flow_task.clone(), query_context)
+        .submit_create_flow_task(create_flow_task.clone(), query_context)
         .await?;
 
     let procedure_id = id.to_string();
@@ -712,7 +671,6 @@ async fn handle_create_flow_task(
 
 async fn handle_alter_logical_table_tasks(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     alter_table_tasks: Vec<AlterTableTask>,
 ) -> Result<SubmitDdlTaskResponse> {
     ensure!(
@@ -733,7 +691,7 @@ async fn handle_alter_logical_table_tasks(
     let num_logical_tables = alter_table_tasks.len();
 
     let (id, _) = ddl_manager
-        .submit_alter_logical_table_tasks(cluster_id, alter_table_tasks, physical_table_id)
+        .submit_alter_logical_table_tasks(alter_table_tasks, physical_table_id)
         .await?;
 
     info!("{num_logical_tables} logical tables on physical table: {physical_table_id:?} is altered via procedure_id {id:?}");
@@ -749,11 +707,10 @@ async fn handle_alter_logical_table_tasks(
 /// Handle the `[CreateViewTask]` and returns the DDL response when success.
 async fn handle_create_view_task(
     ddl_manager: &DdlManager,
-    cluster_id: ClusterId,
     create_view_task: CreateViewTask,
 ) -> Result<SubmitDdlTaskResponse> {
     let (id, output) = ddl_manager
-        .submit_create_view_task(cluster_id, create_view_task)
+        .submit_create_view_task(create_view_task)
         .await?;
 
     let procedure_id = id.to_string();
@@ -788,55 +745,43 @@ impl ProcedureExecutor for DdlManager {
             .unwrap_or(TracingContext::from_current_span())
             .attach(tracing::info_span!("DdlManager::submit_ddl_task"));
         async move {
-            let cluster_id = ctx.cluster_id.unwrap_or_default();
             debug!("Submitting Ddl task: {:?}", request.task);
             match request.task {
                 CreateTable(create_table_task) => {
-                    handle_create_table_task(self, cluster_id, create_table_task).await
-                }
-                DropTable(drop_table_task) => {
-                    handle_drop_table_task(self, cluster_id, drop_table_task).await
+                    handle_create_table_task(self, create_table_task).await
                 }
+                DropTable(drop_table_task) => handle_drop_table_task(self, drop_table_task).await,
                 AlterTable(alter_table_task) => {
-                    handle_alter_table_task(self, cluster_id, alter_table_task).await
+                    handle_alter_table_task(self, alter_table_task).await
                 }
                 TruncateTable(truncate_table_task) => {
-                    handle_truncate_table_task(self, cluster_id, truncate_table_task).await
+                    handle_truncate_table_task(self, truncate_table_task).await
                 }
                 CreateLogicalTables(create_table_tasks) => {
-                    handle_create_logical_table_tasks(self, cluster_id, create_table_tasks).await
+                    handle_create_logical_table_tasks(self, create_table_tasks).await
                 }
                 AlterLogicalTables(alter_table_tasks) => {
-                    handle_alter_logical_table_tasks(self, cluster_id, alter_table_tasks).await
+                    handle_alter_logical_table_tasks(self, alter_table_tasks).await
                 }
                 DropLogicalTables(_) => todo!(),
                 CreateDatabase(create_database_task) => {
-                    handle_create_database_task(self, cluster_id, create_database_task).await
+                    handle_create_database_task(self, create_database_task).await
                 }
                 DropDatabase(drop_database_task) => {
-                    handle_drop_database_task(self, cluster_id, drop_database_task).await
+                    handle_drop_database_task(self, drop_database_task).await
                 }
                 AlterDatabase(alter_database_task) => {
-                    handle_alter_database_task(self, cluster_id, alter_database_task).await
+                    handle_alter_database_task(self, alter_database_task).await
                 }
                 CreateFlow(create_flow_task) => {
-                    handle_create_flow_task(
-                        self,
-                        cluster_id,
-                        create_flow_task,
-                        request.query_context.into(),
-                    )
-                    .await
-                }
-                DropFlow(drop_flow_task) => {
-                    handle_drop_flow_task(self, cluster_id, drop_flow_task).await
+                    handle_create_flow_task(self, create_flow_task, request.query_context.into())
+                        .await
                 }
+                DropFlow(drop_flow_task) => handle_drop_flow_task(self, drop_flow_task).await,
                 CreateView(create_view_task) => {
-                    handle_create_view_task(self, cluster_id, create_view_task).await
-                }
-                DropView(drop_view_task) => {
-                    handle_drop_view_task(self, cluster_id, drop_view_task).await
+                    handle_create_view_task(self, create_view_task).await
                 }
+                DropView(drop_view_task) => handle_drop_view_task(self, drop_view_task).await,
             }
         }
         .trace(span)

src/common/meta/src/instruction.rs

@@ -26,11 +26,10 @@ use crate::flow_name::FlowName;
 use crate::key::schema_name::SchemaName;
 use crate::key::FlowId;
 use crate::peer::Peer;
-use crate::{ClusterId, DatanodeId, FlownodeId};
+use crate::{DatanodeId, FlownodeId};
 
 #[derive(Eq, Hash, PartialEq, Clone, Debug, Serialize, Deserialize)]
 pub struct RegionIdent {
-    pub cluster_id: ClusterId,
     pub datanode_id: DatanodeId,
     pub table_id: TableId,
     pub region_number: RegionNumber,
@@ -47,8 +46,8 @@ impl Display for RegionIdent {
     fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "RegionIdent(datanode_id='{}.{}', table_id={}, region_number={}, engine = {})",
-            self.cluster_id, self.datanode_id, self.table_id, self.region_number, self.engine
+            "RegionIdent(datanode_id='{}', table_id={}, region_number={}, engine = {})",
+            self.datanode_id, self.table_id, self.region_number, self.engine
         )
     }
 }
@@ -262,7 +261,6 @@ mod tests {
     fn test_serialize_instruction() {
         let open_region = Instruction::OpenRegion(OpenRegion::new(
             RegionIdent {
-                cluster_id: 1,
                 datanode_id: 2,
                 table_id: 1024,
                 region_number: 1,
@@ -277,12 +275,11 @@ mod tests {
         let serialized = serde_json::to_string(&open_region).unwrap();
 
         assert_eq!(
-            r#"{"OpenRegion":{"region_ident":{"cluster_id":1,"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"},"region_storage_path":"test/foo","region_options":{},"region_wal_options":{},"skip_wal_replay":false}}"#,
+            r#"{"OpenRegion":{"region_ident":{"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"},"region_storage_path":"test/foo","region_options":{},"region_wal_options":{},"skip_wal_replay":false}}"#,
             serialized
         );
 
         let close_region = Instruction::CloseRegion(RegionIdent {
-            cluster_id: 1,
             datanode_id: 2,
             table_id: 1024,
             region_number: 1,
@@ -292,7 +289,7 @@ mod tests {
         let serialized = serde_json::to_string(&close_region).unwrap();
 
         assert_eq!(
-            r#"{"CloseRegion":{"cluster_id":1,"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"}}"#,
+            r#"{"CloseRegion":{"datanode_id":2,"table_id":1024,"region_number":1,"engine":"mito2"}}"#,
             serialized
         );
     }
@@ -307,7 +304,6 @@ mod tests {
     #[test]
     fn test_compatible_serialize_open_region() {
         let region_ident = RegionIdent {
-            cluster_id: 1,
             datanode_id: 2,
             table_id: 1024,
             region_number: 1,

src/common/meta/src/lib.rs

@@ -47,8 +47,6 @@ pub mod test_util;
 pub mod util;
 pub mod wal_options_allocator;
 
-// The id of the cluster.
-pub type ClusterId = u64;
 // The id of the datanode.
 pub type DatanodeId = u64;
 // The id of the flownode.

src/common/meta/src/node_expiry_listener.rs

@@ -99,7 +99,7 @@ impl NodeExpiryListener {
         in_memory: &ResettableKvBackendRef,
         max_idle_time: Duration,
     ) -> error::Result<impl Iterator<Item = NodeInfoKey>> {
-        let prefix = NodeInfoKey::key_prefix_with_cluster_id(0);
+        let prefix = NodeInfoKey::key_prefix();
         let req = RangeRequest::new().with_prefix(prefix);
         let current_time_millis = common_time::util::current_time_millis();
         let resp = in_memory.range(req).await?;

src/common/meta/src/peer.rs

@@ -19,7 +19,7 @@ use api::v1::meta::Peer as PbPeer;
 use serde::{Deserialize, Serialize};
 
 use crate::error::Error;
-use crate::{ClusterId, DatanodeId, FlownodeId};
+use crate::{DatanodeId, FlownodeId};
 
 #[derive(Debug, Default, Clone, Hash, Eq, PartialEq, Deserialize, Serialize)]
 pub struct Peer {
@@ -72,8 +72,8 @@ impl Display for Peer {
 /// can query peer given a node id
 #[async_trait::async_trait]
 pub trait PeerLookupService {
-    async fn datanode(&self, cluster_id: ClusterId, id: DatanodeId) -> Result<Option<Peer>, Error>;
-    async fn flownode(&self, cluster_id: ClusterId, id: FlownodeId) -> Result<Option<Peer>, Error>;
+    async fn datanode(&self, id: DatanodeId) -> Result<Option<Peer>, Error>;
+    async fn flownode(&self, id: FlownodeId) -> Result<Option<Peer>, Error>;
 }
 
 pub type PeerLookupServiceRef = Arc<dyn PeerLookupService + Send + Sync>;

src/common/meta/src/rpc.rs

@@ -31,11 +31,6 @@ impl ResponseHeader {
         self.0.protocol_version
     }
 
-    #[inline]
-    pub fn cluster_id(&self) -> u64 {
-        self.0.cluster_id
-    }
-
     #[inline]
     pub fn error_code(&self) -> i32 {
         match self.0.error.as_ref() {
@@ -143,7 +138,6 @@ mod tests {
     fn test_response_header_trans() {
         let pb_header = PbResponseHeader {
             protocol_version: 101,
-            cluster_id: 1,
             error: Some(Error {
                 code: 100,
                 err_msg: "test".to_string(),
@@ -152,7 +146,6 @@ mod tests {
 
         let header = ResponseHeader(pb_header);
         assert_eq!(101, header.protocol_version());
-        assert_eq!(1, header.cluster_id());
         assert_eq!(100, header.error_code());
         assert_eq!("test".to_string(), header.error_msg());
     }

src/common/meta/src/test_util.rs

@@ -37,7 +37,7 @@ use crate::peer::{Peer, PeerLookupService};
 use crate::region_keeper::MemoryRegionKeeper;
 use crate::sequence::SequenceBuilder;
 use crate::wal_options_allocator::WalOptionsAllocator;
-use crate::{ClusterId, DatanodeId, FlownodeId};
+use crate::{DatanodeId, FlownodeId};
 
 #[async_trait::async_trait]
 pub trait MockDatanodeHandler: Sync + Send + Clone {
@@ -189,11 +189,11 @@ pub struct NoopPeerLookupService;
 
 #[async_trait::async_trait]
 impl PeerLookupService for NoopPeerLookupService {
-    async fn datanode(&self, _cluster_id: ClusterId, id: DatanodeId) -> Result<Option<Peer>> {
+    async fn datanode(&self, id: DatanodeId) -> Result<Option<Peer>> {
         Ok(Some(Peer::empty(id)))
     }
 
-    async fn flownode(&self, _cluster_id: ClusterId, id: FlownodeId) -> Result<Option<Peer>> {
+    async fn flownode(&self, id: FlownodeId) -> Result<Option<Peer>> {
         Ok(Some(Peer::empty(id)))
     }
 }

src/common/query/src/error.rs

@@ -24,7 +24,6 @@ use datatypes::arrow::datatypes::DataType as ArrowDatatype;
 use datatypes::error::Error as DataTypeError;
 use datatypes::prelude::ConcreteDataType;
 use snafu::{Location, Snafu};
-use statrs::StatsError;
 
 #[derive(Snafu)]
 #[snafu(visibility(pub))]
@@ -38,14 +37,6 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Failed to generate function"))]
-    GenerateFunction {
-        #[snafu(source)]
-        error: StatsError,
-        #[snafu(implicit)]
-        location: Location,
-    },
-
     #[snafu(display("Failed to cast scalar value into vector"))]
     FromScalarValue {
         #[snafu(implicit)]
@@ -97,12 +88,6 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("unexpected: not constant column"))]
-    InvalidInputCol {
-        #[snafu(implicit)]
-        location: Location,
-    },
-
     #[snafu(display("General DataFusion error"))]
     GeneralDataFusion {
         #[snafu(source)]
@@ -248,8 +233,6 @@ impl ErrorExt for Error {
             Error::CreateAccumulator { .. }
             | Error::DowncastVector { .. }
             | Error::InvalidInputState { .. }
-            | Error::InvalidInputCol { .. }
-            | Error::GenerateFunction { .. }
             | Error::BadAccumulatorImpl { .. }
             | Error::ToScalarValue { .. }
             | Error::GetScalarVector { .. }

src/datanode/src/heartbeat/handler.rs

@@ -235,7 +235,6 @@ mod tests {
         Instruction::CloseRegion(RegionIdent {
             table_id: region_id.table_id(),
             region_number: region_id.region_number(),
-            cluster_id: 1,
             datanode_id: 2,
             engine: MITO_ENGINE_NAME.to_string(),
         })
@@ -246,7 +245,6 @@ mod tests {
             RegionIdent {
                 table_id: region_id.table_id(),
                 region_number: region_id.region_number(),
-                cluster_id: 1,
                 datanode_id: 2,
                 engine: MITO_ENGINE_NAME.to_string(),
             },

src/datatypes/src/schema/column_schema.rs

@@ -597,7 +597,7 @@ impl fmt::Display for FulltextAnalyzer {
 }
 
 /// Skipping options for a column.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default, Visit, VisitMut)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Visit, VisitMut)]
 #[serde(rename_all = "kebab-case")]
 pub struct SkippingIndexOptions {
     /// The granularity of the skip index.
@@ -607,6 +607,15 @@ pub struct SkippingIndexOptions {
     pub index_type: SkippingIndexType,
 }
 
+impl Default for SkippingIndexOptions {
+    fn default() -> Self {
+        Self {
+            granularity: DEFAULT_GRANULARITY,
+            index_type: SkippingIndexType::default(),
+        }
+    }
+}
+
 impl fmt::Display for SkippingIndexOptions {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(f, "granularity={}", self.granularity)?;

src/flow/src/adapter.rs

@@ -103,7 +103,6 @@ impl Default for FlowConfig {
 #[serde(default)]
 pub struct FlownodeOptions {
     pub mode: Mode,
-    pub cluster_id: Option<u64>,
     pub node_id: Option<u64>,
     pub flow: FlowConfig,
     pub grpc: GrpcOptions,
@@ -118,7 +117,6 @@ impl Default for FlownodeOptions {
     fn default() -> Self {
         Self {
             mode: servers::Mode::Standalone,
-            cluster_id: None,
             node_id: None,
             flow: FlowConfig::default(),
             grpc: GrpcOptions::default().with_bind_addr("127.0.0.1:3004"),

src/frontend/src/instance/jaeger.rs

@@ -35,12 +35,11 @@ use servers::error::{
     CatalogSnafu, CollectRecordbatchSnafu, DataFusionSnafu, Result as ServerResult,
     TableNotFoundSnafu,
 };
-use servers::http::jaeger::QueryTraceParams;
-use servers::otlp::trace::v0::{
-    DURATION_NANO_COLUMN, SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN, SPAN_ID_COLUMN,
-    SPAN_KIND_COLUMN, SPAN_KIND_PREFIX, SPAN_NAME_COLUMN, TIMESTAMP_COLUMN, TRACE_ID_COLUMN,
+use servers::http::jaeger::{QueryTraceParams, FIND_TRACES_COLS};
+use servers::otlp::trace::{
+    DURATION_NANO_COLUMN, SERVICE_NAME_COLUMN, SPAN_ATTRIBUTES_COLUMN, SPAN_KIND_COLUMN,
+    SPAN_KIND_PREFIX, SPAN_NAME_COLUMN, TIMESTAMP_COLUMN, TRACE_ID_COLUMN, TRACE_TABLE_NAME,
 };
-use servers::otlp::trace::TRACE_TABLE_NAME;
 use servers::query_handler::JaegerQueryHandler;
 use session::context::QueryContextRef;
 use snafu::{OptionExt, ResultExt};
@@ -102,16 +101,9 @@ impl JaegerQueryHandler for Instance {
     }
 
     async fn get_trace(&self, ctx: QueryContextRef, trace_id: &str) -> ServerResult<Output> {
-        // It's equivalent to `SELECT trace_id, timestamp, duration_nano, service_name, span_name, span_id, span_attributes FROM {db}.{trace_table} WHERE trace_id = '{trace_id}'`.
-        let selects = vec![
-            col(TRACE_ID_COLUMN),
-            col(TIMESTAMP_COLUMN),
-            col(DURATION_NANO_COLUMN),
-            col(SERVICE_NAME_COLUMN),
-            col(SPAN_NAME_COLUMN),
-            col(SPAN_ID_COLUMN),
-            col(SPAN_ATTRIBUTES_COLUMN),
-        ];
+        // It's equivalent to `SELECT trace_id, timestamp, duration_nano, service_name, span_name, span_id, span_attributes, resource_attributes, parent_span_id
+        // FROM {db}.{trace_table} WHERE trace_id = '{trace_id}'`.
+        let selects: Vec<Expr> = FIND_TRACES_COLS.clone();
 
         let filters = vec![col(TRACE_ID_COLUMN).eq(lit(trace_id))];
 
@@ -133,15 +125,7 @@ impl JaegerQueryHandler for Instance {
         ctx: QueryContextRef,
         query_params: QueryTraceParams,
     ) -> ServerResult<Output> {
-        let selects = vec![
-            col(TRACE_ID_COLUMN),
-            col(TIMESTAMP_COLUMN),
-            col(DURATION_NANO_COLUMN),
-            col(SERVICE_NAME_COLUMN),
-            col(SPAN_NAME_COLUMN),
-            col(SPAN_ID_COLUMN),
-            col(SPAN_ATTRIBUTES_COLUMN),
-        ];
+        let selects: Vec<Expr> = FIND_TRACES_COLS.clone();
 
         let mut filters = vec![];
 

src/frontend/src/instance/log_handler.rs

@@ -127,4 +127,26 @@ impl Instance {
             .map_err(BoxedError::new)
             .context(ExecuteGrpcRequestSnafu)
     }
+
+    pub async fn handle_trace_inserts(
+        &self,
+        rows: RowInsertRequests,
+        ctx: QueryContextRef,
+    ) -> ServerResult<Output> {
+        let _guard = if let Some(limiter) = &self.limiter {
+            let result = limiter.limit_row_inserts(&rows);
+            if result.is_none() {
+                return InFlightWriteBytesExceededSnafu.fail();
+            }
+            result
+        } else {
+            None
+        };
+
+        self.inserter
+            .handle_trace_inserts(rows, ctx, self.statement_executor.as_ref())
+            .await
+            .map_err(BoxedError::new)
+            .context(ExecuteGrpcRequestSnafu)
+    }
 }

src/frontend/src/instance/otlp.rs

@@ -101,17 +101,7 @@ impl OpenTelemetryProtocolHandler for Instance {
 
         OTLP_TRACES_ROWS.inc_by(rows as u64);
 
-        let _guard = if let Some(limiter) = &self.limiter {
-            let result = limiter.limit_row_inserts(&requests);
-            if result.is_none() {
-                return InFlightWriteBytesExceededSnafu.fail();
-            }
-            result
-        } else {
-            None
-        };
-
-        self.handle_log_inserts(requests, ctx)
+        self.handle_trace_inserts(requests, ctx)
             .await
             .map_err(BoxedError::new)
             .context(error::ExecuteGrpcQuerySnafu)

src/meta-client/examples/meta_client.rs

@@ -31,13 +31,13 @@ fn main() {
 
 #[tokio::main]
 async fn run() {
-    let id = (1000u64, 2000u64);
+    let id = 2000u64;
     let config = ChannelConfig::new()
         .timeout(Duration::from_secs(3))
         .connect_timeout(Duration::from_secs(5))
         .tcp_nodelay(true);
     let channel_manager = ChannelManager::with_config(config);
-    let mut meta_client = MetaClientBuilder::datanode_default_options(id.0, id.1)
+    let mut meta_client = MetaClientBuilder::datanode_default_options(id)
         .channel_manager(channel_manager)
         .build();
     meta_client.start(&["127.0.0.1:3002"]).await.unwrap();

src/meta-client/src/client.rs

@@ -47,7 +47,6 @@ use common_meta::rpc::store::{
     DeleteRangeResponse, PutRequest, PutResponse, RangeRequest, RangeResponse,
 };
 use common_meta::rpc::KeyValue;
-use common_meta::ClusterId;
 use common_telemetry::info;
 use futures::TryStreamExt;
 use heartbeat::Client as HeartbeatClient;
@@ -61,7 +60,7 @@ use crate::error::{
     Result,
 };
 
-pub type Id = (u64, u64);
+pub type Id = u64;
 
 const DEFAULT_ASK_LEADER_MAX_RETRY: usize = 3;
 const DEFAULT_SUBMIT_DDL_MAX_RETRY: usize = 3;
@@ -81,18 +80,18 @@ pub struct MetaClientBuilder {
 }
 
 impl MetaClientBuilder {
-    pub fn new(cluster_id: ClusterId, member_id: u64, role: Role) -> Self {
+    pub fn new(member_id: u64, role: Role) -> Self {
         Self {
-            id: (cluster_id, member_id),
+            id: member_id,
             role,
             ..Default::default()
         }
     }
 
     /// Returns the role of Frontend's default options.
-    pub fn frontend_default_options(cluster_id: ClusterId) -> Self {
+    pub fn frontend_default_options() -> Self {
         // Frontend does not need a member id.
-        Self::new(cluster_id, 0, Role::Frontend)
+        Self::new(0, Role::Frontend)
             .enable_store()
             .enable_heartbeat()
             .enable_procedure()
@@ -100,15 +99,15 @@ impl MetaClientBuilder {
     }
 
     /// Returns the role of Datanode's default options.
-    pub fn datanode_default_options(cluster_id: ClusterId, member_id: u64) -> Self {
-        Self::new(cluster_id, member_id, Role::Datanode)
+    pub fn datanode_default_options(member_id: u64) -> Self {
+        Self::new(member_id, Role::Datanode)
             .enable_store()
             .enable_heartbeat()
     }
 
     /// Returns the role of Flownode's default options.
-    pub fn flownode_default_options(cluster_id: ClusterId, member_id: u64) -> Self {
-        Self::new(cluster_id, member_id, Role::Flownode)
+    pub fn flownode_default_options(member_id: u64) -> Self {
+        Self::new(member_id, Role::Flownode)
             .enable_store()
             .enable_heartbeat()
             .enable_procedure()
@@ -273,15 +272,9 @@ impl ClusterInfo for MetaClient {
         let cluster_client = self.cluster_client()?;
 
         let (get_metasrv_nodes, nodes_key_prefix) = match role {
-            None => (
-                true,
-                Some(NodeInfoKey::key_prefix_with_cluster_id(self.id.0)),
-            ),
+            None => (true, Some(NodeInfoKey::key_prefix())),
             Some(ClusterRole::Metasrv) => (true, None),
-            Some(role) => (
-                false,
-                Some(NodeInfoKey::key_prefix_with_role(self.id.0, role)),
-            ),
+            Some(role) => (false, Some(NodeInfoKey::key_prefix_with_role(role))),
         };
 
         let mut nodes = if get_metasrv_nodes {
@@ -324,7 +317,7 @@ impl ClusterInfo for MetaClient {
 
     async fn list_region_stats(&self) -> Result<Vec<RegionStat>> {
         let cluster_kv_backend = Arc::new(self.cluster_client()?);
-        let range_prefix = DatanodeStatKey::key_prefix_with_cluster_id(self.id.0);
+        let range_prefix = DatanodeStatKey::prefix_key();
         let req = RangeRequest::new().with_prefix(range_prefix);
         let stream =
             PaginationStream::new(cluster_kv_backend, req, 256, decode_stats).into_stream();
@@ -555,6 +548,8 @@ impl MetaClient {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
     use api::v1::meta::{HeartbeatRequest, Peer};
     use common_meta::kv_backend::{KvBackendRef, ResettableKvBackendRef};
     use rand::Rng;
@@ -624,31 +619,31 @@ mod tests {
     async fn test_meta_client_builder() {
         let urls = &["127.0.0.1:3001", "127.0.0.1:3002"];
 
-        let mut meta_client = MetaClientBuilder::new(0, 0, Role::Datanode)
+        let mut meta_client = MetaClientBuilder::new(0, Role::Datanode)
             .enable_heartbeat()
             .build();
         let _ = meta_client.heartbeat_client().unwrap();
         assert!(meta_client.store_client().is_err());
         meta_client.start(urls).await.unwrap();
 
-        let mut meta_client = MetaClientBuilder::new(0, 0, Role::Datanode).build();
+        let mut meta_client = MetaClientBuilder::new(0, Role::Datanode).build();
         assert!(meta_client.heartbeat_client().is_err());
         assert!(meta_client.store_client().is_err());
         meta_client.start(urls).await.unwrap();
 
-        let mut meta_client = MetaClientBuilder::new(0, 0, Role::Datanode)
+        let mut meta_client = MetaClientBuilder::new(0, Role::Datanode)
             .enable_store()
             .build();
         assert!(meta_client.heartbeat_client().is_err());
         let _ = meta_client.store_client().unwrap();
         meta_client.start(urls).await.unwrap();
 
-        let mut meta_client = MetaClientBuilder::new(1, 2, Role::Datanode)
+        let mut meta_client = MetaClientBuilder::new(2, Role::Datanode)
             .enable_heartbeat()
             .enable_store()
             .build();
-        assert_eq!(1, meta_client.id().0);
-        assert_eq!(2, meta_client.id().1);
+        assert_eq!(2, meta_client.id());
+        assert_eq!(2, meta_client.id());
         let _ = meta_client.heartbeat_client().unwrap();
         let _ = meta_client.store_client().unwrap();
         meta_client.start(urls).await.unwrap();
@@ -657,7 +652,7 @@ mod tests {
     #[tokio::test]
     async fn test_not_start_heartbeat_client() {
         let urls = &["127.0.0.1:3001", "127.0.0.1:3002"];
-        let mut meta_client = MetaClientBuilder::new(0, 0, Role::Datanode)
+        let mut meta_client = MetaClientBuilder::new(0, Role::Datanode)
             .enable_store()
             .build();
         meta_client.start(urls).await.unwrap();
@@ -668,7 +663,7 @@ mod tests {
     #[tokio::test]
     async fn test_not_start_store_client() {
         let urls = &["127.0.0.1:3001", "127.0.0.1:3002"];
-        let mut meta_client = MetaClientBuilder::new(0, 0, Role::Datanode)
+        let mut meta_client = MetaClientBuilder::new(0, Role::Datanode)
             .enable_heartbeat()
             .build();
 
@@ -688,6 +683,9 @@ mod tests {
         let tc = new_client("test_heartbeat").await;
         let (sender, mut receiver) = tc.client.heartbeat().await.unwrap();
         // send heartbeats
+
+        let request_sent = Arc::new(AtomicUsize::new(0));
+        let request_sent_clone = request_sent.clone();
         let _handle = tokio::spawn(async move {
             for _ in 0..5 {
                 let req = HeartbeatRequest {
@@ -698,14 +696,24 @@ mod tests {
                     ..Default::default()
                 };
                 sender.send(req).await.unwrap();
+                request_sent_clone.fetch_add(1, Ordering::Relaxed);
             }
         });
 
-        let _handle = tokio::spawn(async move {
-            while let Some(res) = receiver.message().await.unwrap() {
-                assert_eq!(1000, res.header.unwrap().cluster_id);
+        let heartbeat_count = Arc::new(AtomicUsize::new(0));
+        let heartbeat_count_clone = heartbeat_count.clone();
+        let handle = tokio::spawn(async move {
+            while let Some(_resp) = receiver.message().await.unwrap() {
+                heartbeat_count_clone.fetch_add(1, Ordering::Relaxed);
             }
         });
+
+        handle.await.unwrap();
+        //+1 for the initial response
+        assert_eq!(
+            request_sent.load(Ordering::Relaxed) + 1,
+            heartbeat_count.load(Ordering::Relaxed)
+        );
     }
 
     #[tokio::test]

src/meta-client/src/client/heartbeat.rs

@@ -272,7 +272,7 @@ mod test {
 
     #[tokio::test]
     async fn test_already_start() {
-        let mut client = Client::new((0, 0), Role::Datanode, ChannelManager::default(), 3);
+        let mut client = Client::new(0, Role::Datanode, ChannelManager::default(), 3);
         client
             .start(&["127.0.0.1:1000", "127.0.0.1:1001"])
             .await
@@ -288,7 +288,7 @@ mod test {
     #[tokio::test]
     async fn test_heartbeat_stream() {
         let (sender, mut receiver) = mpsc::channel::<HeartbeatRequest>(100);
-        let sender = HeartbeatSender::new((8, 8), Role::Datanode, sender);
+        let sender = HeartbeatSender::new(8, Role::Datanode, sender);
         let _handle = tokio::spawn(async move {
             for _ in 0..10 {
                 sender.send(HeartbeatRequest::default()).await.unwrap();
@@ -296,7 +296,6 @@ mod test {
         });
         while let Some(req) = receiver.recv().await {
             let header = req.header.unwrap();
-            assert_eq!(8, header.cluster_id);
             assert_eq!(8, header.member_id);
         }
     }

src/meta-client/src/client/store.rs

@@ -255,7 +255,7 @@ mod test {
 
     #[tokio::test]
     async fn test_already_start() {
-        let mut client = Client::new((0, 0), Role::Frontend, ChannelManager::default());
+        let mut client = Client::new(0, Role::Frontend, ChannelManager::default());
         client
             .start(&["127.0.0.1:1000", "127.0.0.1:1001"])
             .await
@@ -270,7 +270,7 @@ mod test {
 
     #[tokio::test]
     async fn test_start_with_duplicate_peers() {
-        let mut client = Client::new((0, 0), Role::Frontend, ChannelManager::default());
+        let mut client = Client::new(0, Role::Frontend, ChannelManager::default());
         client
             .start(&["127.0.0.1:1000", "127.0.0.1:1000", "127.0.0.1:1000"])
             .await

src/meta-client/src/lib.rs

@@ -71,23 +71,22 @@ pub enum MetaClientType {
 pub type MetaClientRef = Arc<client::MetaClient>;
 
 pub async fn create_meta_client(
-    cluster_id: u64,
     client_type: MetaClientType,
     meta_client_options: &MetaClientOptions,
 ) -> error::Result<MetaClientRef> {
     info!(
-        "Creating {:?} instance from cluster {} with Metasrv addrs {:?}",
-        client_type, cluster_id, meta_client_options.metasrv_addrs
+        "Creating {:?} instance with Metasrv addrs {:?}",
+        client_type, meta_client_options.metasrv_addrs
     );
 
     let mut builder = match client_type {
         MetaClientType::Datanode { member_id } => {
-            MetaClientBuilder::datanode_default_options(cluster_id, member_id)
+            MetaClientBuilder::datanode_default_options(member_id)
         }
         MetaClientType::Flownode { member_id } => {
-            MetaClientBuilder::flownode_default_options(cluster_id, member_id)
+            MetaClientBuilder::flownode_default_options(member_id)
         }
-        MetaClientType::Frontend => MetaClientBuilder::frontend_default_options(cluster_id),
+        MetaClientType::Frontend => MetaClientBuilder::frontend_default_options(),
     };
 
     let base_config = ChannelConfig::new()

src/meta-client/src/mocks.rs

@@ -60,8 +60,8 @@ pub async fn mock_client_with_etcdstore(addr: &str) -> (MetaClient, MockMetaCont
 }
 
 pub async fn mock_client_by(server_addr: String, channel_manager: ChannelManager) -> MetaClient {
-    let id = (1000u64, 2000u64);
-    let mut meta_client = MetaClientBuilder::datanode_default_options(id.0, id.1)
+    let id = 2000u64;
+    let mut meta_client = MetaClientBuilder::datanode_default_options(id)
         .enable_access_cluster_info()
         .channel_manager(channel_manager)
         .build();

src/meta-srv/src/cluster.rs

@@ -375,13 +375,9 @@ mod tests {
 
     #[test]
     fn test_to_stat_kv_map() {
-        let stat_key = DatanodeStatKey {
-            cluster_id: 0,
-            node_id: 100,
-        };
+        let stat_key = DatanodeStatKey { node_id: 100 };
 
         let stat = Stat {
-            cluster_id: 0,
             id: 100,
             addr: "127.0.0.1:3001".to_string(),
             ..Default::default()
@@ -400,7 +396,6 @@ mod tests {
         let stat_val = kv_map.get(&stat_key).unwrap();
         let stat = stat_val.stats.first().unwrap();
 
-        assert_eq!(0, stat.cluster_id);
         assert_eq!(100, stat.id);
         assert_eq!("127.0.0.1:3001", stat.addr);
     }

src/meta-srv/src/flow_meta_alloc.rs

@@ -15,7 +15,6 @@
 use common_error::ext::BoxedError;
 use common_meta::ddl::flow_meta::PartitionPeerAllocator;
 use common_meta::peer::Peer;
-use common_meta::ClusterId;
 use snafu::ResultExt;
 
 use crate::metasrv::{SelectorContext, SelectorRef};
@@ -34,14 +33,9 @@ impl FlowPeerAllocator {
 
 #[async_trait::async_trait]
 impl PartitionPeerAllocator for FlowPeerAllocator {
-    async fn alloc(
-        &self,
-        cluster_id: ClusterId,
-        partitions: usize,
-    ) -> common_meta::error::Result<Vec<Peer>> {
+    async fn alloc(&self, partitions: usize) -> common_meta::error::Result<Vec<Peer>> {
         self.selector
             .select(
-                cluster_id,
                 &self.ctx,
                 SelectorOptions {
                     min_required_items: partitions,

src/meta-srv/src/handler.rs

@@ -20,8 +20,8 @@ use std::time::{Duration, Instant};
 
 use api::v1::meta::mailbox_message::Payload;
 use api::v1::meta::{
-    HeartbeatRequest, HeartbeatResponse, MailboxMessage, RegionLease, RequestHeader,
-    ResponseHeader, Role, PROTOCOL_VERSION,
+    HeartbeatRequest, HeartbeatResponse, MailboxMessage, RegionLease, ResponseHeader, Role,
+    PROTOCOL_VERSION,
 };
 use check_leader_handler::CheckLeaderHandler;
 use collect_cluster_info_handler::{
@@ -153,13 +153,9 @@ pub struct Pusher {
 }
 
 impl Pusher {
-    pub fn new(
-        sender: Sender<std::result::Result<HeartbeatResponse, tonic::Status>>,
-        req_header: &RequestHeader,
-    ) -> Self {
+    pub fn new(sender: Sender<std::result::Result<HeartbeatResponse, tonic::Status>>) -> Self {
         let res_header = ResponseHeader {
             protocol_version: PROTOCOL_VERSION,
-            cluster_id: req_header.cluster_id,
             ..Default::default()
         };
 
@@ -772,7 +768,7 @@ mod tests {
     use std::sync::Arc;
     use std::time::Duration;
 
-    use api::v1::meta::{MailboxMessage, RequestHeader, Role, PROTOCOL_VERSION};
+    use api::v1::meta::{MailboxMessage, Role};
     use common_meta::kv_backend::memory::MemoryKvBackend;
     use common_meta::sequence::SequenceBuilder;
     use tokio::sync::mpsc;
@@ -814,12 +810,8 @@ mod tests {
     async fn push_msg_via_mailbox() -> (MailboxRef, MailboxReceiver) {
         let datanode_id = 12;
         let (pusher_tx, mut pusher_rx) = mpsc::channel(16);
-        let res_header = RequestHeader {
-            protocol_version: PROTOCOL_VERSION,
-            ..Default::default()
-        };
         let pusher_id = PusherId::new(Role::Datanode, datanode_id);
-        let pusher: Pusher = Pusher::new(pusher_tx, &res_header);
+        let pusher: Pusher = Pusher::new(pusher_tx);
         let handler_group = HeartbeatHandlerGroup::default();
         handler_group.register_pusher(pusher_id, pusher).await;
 

src/meta-srv/src/handler/collect_stats_handler.rs

@@ -262,15 +262,11 @@ mod tests {
         let handler = CollectStatsHandler::default();
         handle_request_many_times(ctx.clone(), &handler, 1).await;
 
-        let key = DatanodeStatKey {
-            cluster_id: 3,
-            node_id: 101,
-        };
+        let key = DatanodeStatKey { node_id: 101 };
         let key: Vec<u8> = key.into();
         let res = ctx.in_memory.get(&key).await.unwrap();
         let kv = res.unwrap();
         let key: DatanodeStatKey = kv.key.clone().try_into().unwrap();
-        assert_eq!(3, key.cluster_id);
         assert_eq!(101, key.node_id);
         let val: DatanodeStatValue = kv.value.try_into().unwrap();
         // first new stat must be set in kv store immediately
@@ -295,7 +291,6 @@ mod tests {
         for i in 1..=loop_times {
             let mut acc = HeartbeatAccumulator {
                 stat: Some(Stat {
-                    cluster_id: 3,
                     id: 101,
                     region_num: i as _,
                     ..Default::default()

src/meta-srv/src/handler/failure_handler.rs

@@ -101,7 +101,6 @@ mod tests {
             }
         }
         acc.stat = Some(Stat {
-            cluster_id: 1,
             id: 42,
             region_stats: vec![new_region_stat(1), new_region_stat(2), new_region_stat(3)],
             timestamp_millis: 1000,

src/meta-srv/src/handler/keep_lease_handler.rs

@@ -38,17 +38,14 @@ impl HeartbeatHandler for DatanodeKeepLeaseHandler {
         _acc: &mut HeartbeatAccumulator,
     ) -> Result<HandleControl> {
         let HeartbeatRequest { header, peer, .. } = req;
-        let Some(header) = &header else {
+        let Some(_header) = &header else {
             return Ok(HandleControl::Continue);
         };
         let Some(peer) = &peer else {
             return Ok(HandleControl::Continue);
         };
 
-        let key = DatanodeLeaseKey {
-            cluster_id: header.cluster_id,
-            node_id: peer.id,
-        };
+        let key = DatanodeLeaseKey { node_id: peer.id };
         let value = LeaseValue {
             timestamp_millis: time_util::current_time_millis(),
             node_addr: peer.addr.clone(),
@@ -80,17 +77,14 @@ impl HeartbeatHandler for FlownodeKeepLeaseHandler {
         _acc: &mut HeartbeatAccumulator,
     ) -> Result<HandleControl> {
         let HeartbeatRequest { header, peer, .. } = req;
-        let Some(header) = &header else {
+        let Some(_header) = &header else {
             return Ok(HandleControl::Continue);
         };
         let Some(peer) = &peer else {
             return Ok(HandleControl::Continue);
         };
 
-        let key = FlownodeLeaseKey {
-            cluster_id: header.cluster_id,
-            node_id: peer.id,
-        };
+        let key = FlownodeLeaseKey { node_id: peer.id };
         let value = LeaseValue {
             timestamp_millis: time_util::current_time_millis(),
             node_addr: peer.addr.clone(),

src/meta-srv/src/handler/region_lease_handler.rs

@@ -64,7 +64,6 @@ impl HeartbeatHandler for RegionLeaseHandler {
         };
 
         let regions = stat.regions();
-        let cluster_id = stat.cluster_id;
         let datanode_id = stat.id;
 
         let RenewRegionLeasesResponse {
@@ -72,7 +71,7 @@ impl HeartbeatHandler for RegionLeaseHandler {
             renewed,
         } = self
             .region_lease_keeper
-            .renew_region_leases(cluster_id, datanode_id, &regions)
+            .renew_region_leases(datanode_id, &regions)
             .await?;
 
         let renewed = renewed
@@ -153,7 +152,6 @@ mod test {
         let peer = Peer::empty(datanode_id);
         let follower_peer = Peer::empty(datanode_id + 1);
         let table_info = new_test_table_info(table_id, vec![region_number]).into();
-        let cluster_id = 1;
 
         let region_routes = vec![RegionRoute {
             region: Region::new_test(region_id),
@@ -181,7 +179,6 @@ mod test {
         let acc = &mut HeartbeatAccumulator::default();
 
         acc.stat = Some(Stat {
-            cluster_id,
             id: peer.id,
             region_stats: vec![
                 new_empty_region_stat(region_id, RegionRole::Follower),
@@ -215,7 +212,6 @@ mod test {
         let acc = &mut HeartbeatAccumulator::default();
 
         acc.stat = Some(Stat {
-            cluster_id,
             id: follower_peer.id,
             region_stats: vec![
                 new_empty_region_stat(region_id, RegionRole::Follower),
@@ -249,7 +245,6 @@ mod test {
         let acc = &mut HeartbeatAccumulator::default();
 
         acc.stat = Some(Stat {
-            cluster_id,
             id: follower_peer.id,
             region_stats: vec![
                 new_empty_region_stat(region_id, RegionRole::Follower),
@@ -292,7 +287,6 @@ mod test {
         let peer = Peer::empty(datanode_id);
         let follower_peer = Peer::empty(datanode_id + 1);
         let table_info = new_test_table_info(table_id, vec![region_number]).into();
-        let cluster_id = 1;
 
         let region_routes = vec![
             RegionRoute {
@@ -333,7 +327,6 @@ mod test {
         let acc = &mut HeartbeatAccumulator::default();
 
         acc.stat = Some(Stat {
-            cluster_id,
             id: peer.id,
             region_stats: vec![
                 new_empty_region_stat(region_id, RegionRole::Leader),