feat: async decode

Signed-off-by: yihong0618 <zouzou0208@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/build-linux-artifacts/action.yml

@@ -52,7 +52,7 @@ runs:
       uses: ./.github/actions/build-greptime-binary
       with:
         base-image: ubuntu
-        features: servers/dashboard,pg_kvbackend
+        features: servers/dashboard,pg_kvbackend,mysql_kvbackend
         cargo-profile: ${{ inputs.cargo-profile }}
         artifacts-dir: greptime-linux-${{ inputs.arch }}-${{ inputs.version }}
         version: ${{ inputs.version }}
@@ -70,7 +70,7 @@ runs:
       if: ${{ inputs.arch == 'amd64' && inputs.dev-mode == 'false' }} # Builds greptime for centos if the host machine is amd64.
       with:
         base-image: centos
-        features: servers/dashboard,pg_kvbackend
+        features: servers/dashboard,pg_kvbackend,mysql_kvbackend
         cargo-profile: ${{ inputs.cargo-profile }}
         artifacts-dir: greptime-linux-${{ inputs.arch }}-centos-${{ inputs.version }}
         version: ${{ 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:
@@ -238,6 +238,13 @@ jobs:
           version: ${{ needs.allocate-runners.outputs.version }}
           push-latest-tag: false # Don't push the latest tag to registry.
           dev-mode: true # Only build the standard images.
+          
+      - name: Echo Docker image tag to step summary
+        run: |
+          echo "## Docker Image Tag" >> $GITHUB_STEP_SUMMARY
+          echo "Image Tag: \`${{ needs.allocate-runners.outputs.version }}\`" >> $GITHUB_STEP_SUMMARY
+          echo "Full Image Name: \`docker.io/${{ vars.IMAGE_NAMESPACE }}/${{ vars.DEV_BUILD_IMAGE_NAME }}:${{ needs.allocate-runners.outputs.version }}\`" >> $GITHUB_STEP_SUMMARY
+          echo "Pull Command: \`docker pull docker.io/${{ vars.IMAGE_NAMESPACE }}/${{ vars.DEV_BUILD_IMAGE_NAME }}:${{ needs.allocate-runners.outputs.version }}\`" >> $GITHUB_STEP_SUMMARY
 
       - name: Set build result
         id: set-build-result
@@ -251,7 +258,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 +290,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 +316,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 +344,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
@@ -111,7 +111,7 @@ jobs:
       - name: Build greptime binaries
         shell: bash
         # `cargo gc` will invoke `cargo build` with specified args
-        run: cargo gc -- --bin greptime --bin sqlness-runner --features pg_kvbackend
+        run: cargo gc -- --bin greptime --bin sqlness-runner --features "pg_kvbackend,mysql_kvbackend"
       - name: Pack greptime binaries
         shell: bash
         run: |
@@ -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
@@ -270,7 +270,7 @@ jobs:
       - name: Build greptime bianry
         shell: bash
         # `cargo gc` will invoke `cargo build` with specified args
-        run: cargo gc --profile ci -- --bin greptime --features pg_kvbackend
+        run: cargo gc --profile ci -- --bin greptime --features "pg_kvbackend,mysql_kvbackend"
       - name: Pack greptime binary
         shell: bash
         run: |
@@ -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
@@ -687,7 +687,7 @@ jobs:
         working-directory: tests-integration/fixtures
         run: docker compose up -d --wait
       - name: Run nextest cases
-        run: cargo nextest run --workspace -F dashboard -F pg_kvbackend
+        run: cargo nextest run --workspace -F dashboard -F pg_kvbackend -F mysql_kvbackend
         env:
           CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold"
           RUST_BACKTRACE: 1
@@ -704,13 +704,14 @@ jobs:
           GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000
           GT_ETCD_ENDPOINTS: http://127.0.0.1:2379
           GT_POSTGRES_ENDPOINTS: postgres://greptimedb:admin@127.0.0.1:5432/postgres
+          GT_MYSQL_ENDPOINTS: mysql://greptimedb:admin@127.0.0.1:3306/mysql
           GT_KAFKA_ENDPOINTS: 127.0.0.1:9092
           GT_KAFKA_SASL_ENDPOINTS: 127.0.0.1:9093
           UNITTEST_LOG_DIR: "__unittest_logs"
 
   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
@@ -739,7 +740,7 @@ jobs:
         working-directory: tests-integration/fixtures
         run: docker compose up -d --wait
       - name: Run nextest cases
-        run: cargo llvm-cov nextest --workspace --lcov --output-path lcov.info -F dashboard -F pg_kvbackend
+        run: cargo llvm-cov nextest --workspace --lcov --output-path lcov.info -F dashboard -F pg_kvbackend -F mysql_kvbackend
         env:
           CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold"
           RUST_BACKTRACE: 1
@@ -755,6 +756,7 @@ jobs:
           GT_MINIO_ENDPOINT_URL: http://127.0.0.1:9000
           GT_ETCD_ENDPOINTS: http://127.0.0.1:2379
           GT_POSTGRES_ENDPOINTS: postgres://greptimedb:admin@127.0.0.1:5432/postgres
+          GT_MYSQL_ENDPOINTS: mysql://greptimedb:admin@127.0.0.1:3306/mysql
           GT_KAFKA_ENDPOINTS: 127.0.0.1:9092
           GT_KAFKA_SASL_ENDPOINTS: 127.0.0.1:9093
           UNITTEST_LOG_DIR: "__unittest_logs"
@@ -770,7 +772,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/grafana.yml

@@ -0,0 +1,52 @@
+name: Check Grafana Panels
+
+on:
+  pull_request:
+    branches:
+      - main
+    paths:
+      - 'grafana/**'  # Trigger only when files under the grafana/ directory change
+
+jobs:
+  check-panels:
+    runs-on: ubuntu-latest
+
+    steps:
+      # Check out the repository
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      # Install jq (required for the script)
+      - name: Install jq
+        run: sudo apt-get install -y jq
+
+      # Make the check.sh script executable
+      - name: Make check.sh executable
+        run: chmod +x grafana/check.sh
+
+      # Run the check.sh script
+      - name: Run check.sh
+        run: ./grafana/check.sh
+
+      # Only run summary.sh for pull_request events (not for merge queues or final pushes)
+      - name: Check if this is a pull request
+        id: check-pr
+        run: |
+          if [[ "${{ github.event_name }}" == "pull_request" ]]; then
+            echo "is_pull_request=true" >> $GITHUB_OUTPUT
+          else
+            echo "is_pull_request=false" >> $GITHUB_OUTPUT
+          fi
+
+      # Make the summary.sh script executable
+      - name: Make summary.sh executable
+        if: steps.check-pr.outputs.is_pull_request == 'true'
+        run: chmod +x grafana/summary.sh
+
+      # Run the summary.sh script and add its output to the GitHub Job Summary
+      - name: Run summary.sh and add to Job Summary
+        if: steps.check-pr.outputs.is_pull_request == 'true'
+        run: |
+          SUMMARY=$(./grafana/summary.sh)
+          echo "### Summary of Grafana Panels" >> $GITHUB_STEP_SUMMARY
+          echo "$SUMMARY" >> $GITHUB_STEP_SUMMARY

.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

@@ -1693,6 +1693,7 @@ dependencies = [
  "humantime",
  "meta-client",
  "nu-ansi-term",
+ "opendal",
  "query",
  "rand",
  "reqwest",
@@ -2015,6 +2016,7 @@ dependencies = [
  "arc-swap",
  "async-trait",
  "bincode",
+ "chrono",
  "common-base",
  "common-catalog",
  "common-error",
@@ -2196,6 +2198,7 @@ dependencies = [
  "serde_with",
  "session",
  "snafu 0.8.5",
+ "sqlx",
  "store-api",
  "strum 0.25.0",
  "table",
@@ -4116,11 +4119,12 @@ dependencies = [
 
 [[package]]
 name = "flate2"
-version = "1.0.34"
+version = "1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a1b589b4dc103969ad3cf85c950899926ec64300a1a46d76c03a6072957036f0"
+checksum = "11faaf5a5236997af9848be0bef4db95824b1d534ebc64d0f0c6cf3e67bd38dc"
 dependencies = [
  "crc32fast",
+ "libz-rs-sys",
  "libz-sys",
  "miniz_oxide",
 ]
@@ -4167,6 +4171,7 @@ dependencies = [
  "bytes",
  "cache",
  "catalog",
+ "chrono",
  "client",
  "common-base",
  "common-catalog",
@@ -4701,7 +4706,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=c5419bbd20cb42e568ec325a4d71a3c94cc327e1#c5419bbd20cb42e568ec325a4d71a3c94cc327e1"
 dependencies = [
  "prost 0.13.3",
  "serde",
@@ -5566,6 +5571,7 @@ dependencies = [
  "rand",
  "regex",
  "regex-automata 0.4.8",
+ "roaring",
  "serde",
  "serde_json",
  "snafu 0.8.5",
@@ -5897,15 +5903,15 @@ dependencies = [
 
 [[package]]
 name = "jsonpath-rust"
-version = "0.7.3"
+version = "0.7.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "69a61b87f6a55cc6c28fed5739dd36b9642321ce63e4a5e4a4715d69106f4a10"
+checksum = "0c00ae348f9f8fd2d09f82a98ca381c60df9e0820d8d79fce43e649b4dc3128b"
 dependencies = [
  "pest",
  "pest_derive",
  "regex",
  "serde_json",
- "thiserror 1.0.64",
+ "thiserror 2.0.12",
 ]
 
 [[package]]
@@ -6273,6 +6279,15 @@ dependencies = [
  "vcpkg",
 ]
 
+[[package]]
+name = "libz-rs-sys"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "902bc563b5d65ad9bba616b490842ef0651066a1a1dc3ce1087113ffcb873c8d"
+dependencies = [
+ "zlib-rs",
+]
+
 [[package]]
 name = "libz-sys"
 version = "1.1.20"
@@ -6716,6 +6731,7 @@ dependencies = [
  "session",
  "snafu 0.8.5",
  "store-api",
+ "strum 0.25.0",
  "table",
  "tokio",
  "tokio-postgres",
@@ -6816,9 +6832,9 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
 
 [[package]]
 name = "miniz_oxide"
-version = "0.8.0"
+version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2d80299ef12ff69b16a84bb182e3b9df68b5a91574d3d4fa6e41b65deec4df1"
+checksum = "8e3e04debbb59698c15bacbb6d93584a8c0ca9cc3213cb423d31f760d8843ce5"
 dependencies = [
  "adler2",
 ]
@@ -8270,7 +8286,7 @@ dependencies = [
  "rand",
  "ring",
  "rust_decimal",
- "thiserror 2.0.6",
+ "thiserror 2.0.12",
  "tokio",
  "tokio-rustls 0.26.0",
  "tokio-util",
@@ -8382,7 +8398,7 @@ dependencies = [
  "greptime-proto",
  "itertools 0.10.5",
  "jsonb",
- "jsonpath-rust 0.7.3",
+ "jsonpath-rust 0.7.5",
  "lazy_static",
  "moka",
  "once_cell",
@@ -8760,6 +8776,7 @@ dependencies = [
  "common-recordbatch",
  "common-telemetry",
  "datafusion",
+ "datafusion-common",
  "datafusion-expr",
  "datatypes",
  "futures",
@@ -8773,8 +8790,9 @@ dependencies = [
 
 [[package]]
 name = "promql-parser"
-version = "0.4.3"
-source = "git+https://github.com/GreptimeTeam/promql-parser.git?rev=27abb8e16003a50c720f00d6c85f41f5fa2a2a8e#27abb8e16003a50c720f00d6c85f41f5fa2a2a8e"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7c6b1429bdd199d53bd58b745075c1652efedbe2746e5d4f0d56d3184dda48ec"
 dependencies = [
  "cfgrammar",
  "chrono",
@@ -9632,6 +9650,16 @@ dependencies = [
  "syn 1.0.109",
 ]
 
+[[package]]
+name = "roaring"
+version = "0.10.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "41589aba99537475bf697f2118357cad1c31590c5a1b9f6d9fc4ad6d07503661"
+dependencies = [
+ "bytemuck",
+ "byteorder",
+]
+
 [[package]]
 name = "robust"
 version = "1.1.0"
@@ -11051,7 +11079,7 @@ dependencies = [
  "serde_json",
  "sha2",
  "smallvec",
- "thiserror 2.0.6",
+ "thiserror 2.0.12",
  "tokio",
  "tokio-stream",
  "tracing",
@@ -11136,7 +11164,7 @@ dependencies = [
  "smallvec",
  "sqlx-core",
  "stringprep",
- "thiserror 2.0.6",
+ "thiserror 2.0.12",
  "tracing",
  "whoami",
 ]
@@ -11174,7 +11202,7 @@ dependencies = [
  "smallvec",
  "sqlx-core",
  "stringprep",
- "thiserror 2.0.6",
+ "thiserror 2.0.12",
  "tracing",
  "whoami",
 ]
@@ -11955,11 +11983,11 @@ dependencies = [
 
 [[package]]
 name = "thiserror"
-version = "2.0.6"
+version = "2.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fec2a1820ebd077e2b90c4df007bebf344cd394098a13c563957d0afc83ea47"
+checksum = "567b8a2dae586314f7be2a752ec7474332959c6460e02bde30d702a66d488708"
 dependencies = [
- "thiserror-impl 2.0.6",
+ "thiserror-impl 2.0.12",
 ]
 
 [[package]]
@@ -11975,9 +12003,9 @@ dependencies = [
 
 [[package]]
 name = "thiserror-impl"
-version = "2.0.6"
+version = "2.0.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d65750cab40f4ff1929fb1ba509e9914eb756131cef4210da8d5d700d26f6312"
+checksum = "7f7cf42b4507d8ea322120659672cf1b9dbb93f8f2d4ecfd6e51350ff5b17a1d"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -13936,6 +13964,12 @@ dependencies = [
  "syn 2.0.96",
 ]
 
+[[package]]
+name = "zlib-rs"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8b20717f0917c908dc63de2e44e97f1e6b126ca58d0e391cee86d504eb8fbd05"
+
 [[package]]
 name = "zstd"
 version = "0.11.2+zstd.1.5.2"

Cargo.toml

@@ -126,10 +126,11 @@ deadpool-postgres = "0.12"
 derive_builder = "0.12"
 dotenv = "0.15"
 etcd-client = "0.14"
+flate2 = { version = "1.1.0", default-features = false, features = ["zlib-rs"] }
 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 = "c5419bbd20cb42e568ec325a4d71a3c94cc327e1" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -160,9 +161,7 @@ parquet = { version = "53.0.0", default-features = false, features = ["arrow", "
 paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
-promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", features = [
-    "ser",
-], rev = "27abb8e16003a50c720f00d6c85f41f5fa2a2a8e" }
+promql-parser = { version = "0.5", features = ["ser"] }
 prost = "0.13"
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.8"
@@ -190,6 +189,10 @@ shadow-rs = "0.38"
 similar-asserts = "1.6.0"
 smallvec = { version = "1", features = ["serde"] }
 snafu = "0.8"
+sqlx = { version = "0.8", features = [
+    "runtime-tokio-rustls",
+    "mysql",
+] }
 sysinfo = "0.30"
 # on branch v0.52.x
 sqlparser = { git = "https://github.com/GreptimeTeam/sqlparser-rs.git", rev = "71dd86058d2af97b9925093d40c4e03360403170", features = [

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

config/datanode.example.toml

@@ -231,6 +231,7 @@ overwrite_entry_start_id = false
 # secret_access_key = "123456"
 # endpoint = "https://s3.amazonaws.com"
 # region = "us-west-2"
+# enable_virtual_host_style = false
 
 # Example of using Oss as the storage.
 # [storage]

config/standalone.example.toml

@@ -318,6 +318,7 @@ retry_delay = "500ms"
 # secret_access_key = "123456"
 # endpoint = "https://s3.amazonaws.com"
 # region = "us-west-2"
+# enable_virtual_host_style = false
 
 # Example of using Oss as the storage.
 # [storage]

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.

grafana/check.sh

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+
+BASEDIR=$(dirname "$0")
+
+# Use jq to check for panels with empty or missing descriptions
+invalid_panels=$(cat $BASEDIR/greptimedb-cluster.json | jq -r '
+  .panels[]
+  | select((.type == "stats" or .type == "timeseries") and (.description == "" or .description == null))
+')
+
+# Check if any invalid panels were found
+if [[ -n "$invalid_panels" ]]; then
+  echo "Error: The following panels have empty or missing descriptions:"
+  echo "$invalid_panels"
+  exit 1
+else
+  echo "All panels with type 'stats' or 'timeseries' have valid descriptions."
+  exit 0
+fi

grafana/summary.sh

@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+
+BASEDIR=$(dirname "$0")
+echo '| Title | Description | Expressions |
+|---|---|---|'
+
+cat $BASEDIR/greptimedb-cluster.json | jq -r '
+  .panels |
+  map(select(.type == "stat" or .type == "timeseries")) |
+  .[] | "| \(.title) | \(.description | gsub("\n"; "<br>")) | \(.targets | map(.expr // .rawSql | "`\(.|gsub("\n"; "<br>"))`")  | join("<br>")) |"
+'

src/api/src/helper.rs

@@ -19,9 +19,7 @@ use common_decimal::decimal128::{DECIMAL128_DEFAULT_SCALE, DECIMAL128_MAX_PRECIS
 use common_decimal::Decimal128;
 use common_time::time::Time;
 use common_time::timestamp::TimeUnit;
-use common_time::{
-    Date, DateTime, IntervalDayTime, IntervalMonthDayNano, IntervalYearMonth, Timestamp,
-};
+use common_time::{Date, IntervalDayTime, IntervalMonthDayNano, IntervalYearMonth, Timestamp};
 use datatypes::prelude::{ConcreteDataType, ValueRef};
 use datatypes::scalars::ScalarVector;
 use datatypes::types::{
@@ -29,8 +27,8 @@ use datatypes::types::{
 };
 use datatypes::value::{OrderedF32, OrderedF64, Value};
 use datatypes::vectors::{
-    BinaryVector, BooleanVector, DateTimeVector, DateVector, Decimal128Vector, Float32Vector,
-    Float64Vector, Int32Vector, Int64Vector, IntervalDayTimeVector, IntervalMonthDayNanoVector,
+    BinaryVector, BooleanVector, DateVector, Decimal128Vector, Float32Vector, Float64Vector,
+    Int32Vector, Int64Vector, IntervalDayTimeVector, IntervalMonthDayNanoVector,
     IntervalYearMonthVector, PrimitiveVector, StringVector, TimeMicrosecondVector,
     TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector, TimestampMicrosecondVector,
     TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, UInt32Vector,
@@ -118,7 +116,7 @@ impl From<ColumnDataTypeWrapper> for ConcreteDataType {
             ColumnDataType::Json => ConcreteDataType::json_datatype(),
             ColumnDataType::String => ConcreteDataType::string_datatype(),
             ColumnDataType::Date => ConcreteDataType::date_datatype(),
-            ColumnDataType::Datetime => ConcreteDataType::datetime_datatype(),
+            ColumnDataType::Datetime => ConcreteDataType::timestamp_microsecond_datatype(),
             ColumnDataType::TimestampSecond => ConcreteDataType::timestamp_second_datatype(),
             ColumnDataType::TimestampMillisecond => {
                 ConcreteDataType::timestamp_millisecond_datatype()
@@ -271,7 +269,6 @@ impl TryFrom<ConcreteDataType> for ColumnDataTypeWrapper {
             ConcreteDataType::Binary(_) => ColumnDataType::Binary,
             ConcreteDataType::String(_) => ColumnDataType::String,
             ConcreteDataType::Date(_) => ColumnDataType::Date,
-            ConcreteDataType::DateTime(_) => ColumnDataType::Datetime,
             ConcreteDataType::Timestamp(t) => match t {
                 TimestampType::Second(_) => ColumnDataType::TimestampSecond,
                 TimestampType::Millisecond(_) => ColumnDataType::TimestampMillisecond,
@@ -476,7 +473,6 @@ pub fn push_vals(column: &mut Column, origin_count: usize, vector: VectorRef) {
         Value::String(val) => values.string_values.push(val.as_utf8().to_string()),
         Value::Binary(val) => values.binary_values.push(val.to_vec()),
         Value::Date(val) => values.date_values.push(val.val()),
-        Value::DateTime(val) => values.datetime_values.push(val.val()),
         Value::Timestamp(val) => match val.unit() {
             TimeUnit::Second => values.timestamp_second_values.push(val.value()),
             TimeUnit::Millisecond => values.timestamp_millisecond_values.push(val.value()),
@@ -577,12 +573,11 @@ pub fn pb_value_to_value_ref<'a>(
         ValueData::BinaryValue(bytes) => ValueRef::Binary(bytes.as_slice()),
         ValueData::StringValue(string) => ValueRef::String(string.as_str()),
         ValueData::DateValue(d) => ValueRef::Date(Date::from(*d)),
-        ValueData::DatetimeValue(d) => ValueRef::DateTime(DateTime::new(*d)),
         ValueData::TimestampSecondValue(t) => ValueRef::Timestamp(Timestamp::new_second(*t)),
         ValueData::TimestampMillisecondValue(t) => {
             ValueRef::Timestamp(Timestamp::new_millisecond(*t))
         }
-        ValueData::TimestampMicrosecondValue(t) => {
+        ValueData::DatetimeValue(t) | ValueData::TimestampMicrosecondValue(t) => {
             ValueRef::Timestamp(Timestamp::new_microsecond(*t))
         }
         ValueData::TimestampNanosecondValue(t) => {
@@ -651,7 +646,6 @@ pub fn pb_values_to_vector_ref(data_type: &ConcreteDataType, values: Values) ->
         ConcreteDataType::Binary(_) => Arc::new(BinaryVector::from(values.binary_values)),
         ConcreteDataType::String(_) => Arc::new(StringVector::from_vec(values.string_values)),
         ConcreteDataType::Date(_) => Arc::new(DateVector::from_vec(values.date_values)),
-        ConcreteDataType::DateTime(_) => Arc::new(DateTimeVector::from_vec(values.datetime_values)),
         ConcreteDataType::Timestamp(unit) => match unit {
             TimestampType::Second(_) => Arc::new(TimestampSecondVector::from_vec(
                 values.timestamp_second_values,
@@ -787,11 +781,6 @@ pub fn pb_values_to_values(data_type: &ConcreteDataType, values: Values) -> Vec<
             .into_iter()
             .map(|val| val.into())
             .collect(),
-        ConcreteDataType::DateTime(_) => values
-            .datetime_values
-            .into_iter()
-            .map(|v| Value::DateTime(v.into()))
-            .collect(),
         ConcreteDataType::Date(_) => values
             .date_values
             .into_iter()
@@ -947,9 +936,6 @@ pub fn to_proto_value(value: Value) -> Option<v1::Value> {
         Value::Date(v) => v1::Value {
             value_data: Some(ValueData::DateValue(v.val())),
         },
-        Value::DateTime(v) => v1::Value {
-            value_data: Some(ValueData::DatetimeValue(v.val())),
-        },
         Value::Timestamp(v) => match v.unit() {
             TimeUnit::Second => v1::Value {
                 value_data: Some(ValueData::TimestampSecondValue(v.value())),
@@ -1066,7 +1052,6 @@ pub fn value_to_grpc_value(value: Value) -> GrpcValue {
             Value::String(v) => Some(ValueData::StringValue(v.as_utf8().to_string())),
             Value::Binary(v) => Some(ValueData::BinaryValue(v.to_vec())),
             Value::Date(v) => Some(ValueData::DateValue(v.val())),
-            Value::DateTime(v) => Some(ValueData::DatetimeValue(v.val())),
             Value::Timestamp(v) => Some(match v.unit() {
                 TimeUnit::Second => ValueData::TimestampSecondValue(v.value()),
                 TimeUnit::Millisecond => ValueData::TimestampMillisecondValue(v.value()),
@@ -1248,7 +1233,7 @@ mod tests {
             ColumnDataTypeWrapper::date_datatype().into()
         );
         assert_eq!(
-            ConcreteDataType::datetime_datatype(),
+            ConcreteDataType::timestamp_microsecond_datatype(),
             ColumnDataTypeWrapper::datetime_datatype().into()
         );
         assert_eq!(
@@ -1339,10 +1324,6 @@ mod tests {
             ColumnDataTypeWrapper::date_datatype(),
             ConcreteDataType::date_datatype().try_into().unwrap()
         );
-        assert_eq!(
-            ColumnDataTypeWrapper::datetime_datatype(),
-            ConcreteDataType::datetime_datatype().try_into().unwrap()
-        );
         assert_eq!(
             ColumnDataTypeWrapper::timestamp_millisecond_datatype(),
             ConcreteDataType::timestamp_millisecond_datatype()
@@ -1830,17 +1811,6 @@ mod tests {
         ]
     );
 
-    test_convert_values!(
-        datetime,
-        vec![1.into(), 2.into(), 3.into()],
-        datetime,
-        vec![
-            Value::DateTime(1.into()),
-            Value::DateTime(2.into()),
-            Value::DateTime(3.into())
-        ]
-    );
-
     #[test]
     fn test_vectors_to_rows_for_different_types() {
         let boolean_vec = BooleanVector::from_vec(vec![true, false, true]);

src/api/src/v1/column_def.rs

@@ -132,6 +132,15 @@ pub fn options_from_skipping(skipping: &SkippingIndexOptions) -> Result<Option<C
     Ok((!options.options.is_empty()).then_some(options))
 }
 
+/// Tries to construct a `ColumnOptions` for inverted index.
+pub fn options_from_inverted() -> ColumnOptions {
+    let mut options = ColumnOptions::default();
+    options
+        .options
+        .insert(INVERTED_INDEX_GRPC_KEY.to_string(), "true".to_string());
+    options
+}
+
 /// Tries to construct a `FulltextAnalyzer` from the given analyzer.
 pub fn as_fulltext_option(analyzer: Analyzer) -> FulltextAnalyzer {
     match analyzer {

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/catalog/src/system_schema.rs

@@ -77,7 +77,7 @@ trait SystemSchemaProviderInner {
     fn system_table(&self, name: &str) -> Option<SystemTableRef>;
 
     fn table_info(catalog_name: String, table: &SystemTableRef) -> TableInfoRef {
-        let table_meta = TableMetaBuilder::default()
+        let table_meta = TableMetaBuilder::empty()
             .schema(table.schema())
             .primary_key_indices(vec![])
             .next_column_id(0)

src/catalog/src/system_schema/information_schema/columns.rs

@@ -365,10 +365,6 @@ impl InformationSchemaColumnsBuilder {
             self.numeric_scales.push(None);
 
             match &column_schema.data_type {
-                ConcreteDataType::DateTime(datetime_type) => {
-                    self.datetime_precisions
-                        .push(Some(datetime_type.precision() as i64));
-                }
                 ConcreteDataType::Timestamp(ts_type) => {
                     self.datetime_precisions
                         .push(Some(ts_type.precision() as i64));

src/catalog/src/system_schema/information_schema/flows.rs

@@ -28,16 +28,19 @@ use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
 use datatypes::prelude::ConcreteDataType as CDT;
 use datatypes::scalars::ScalarVectorBuilder;
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
+use datatypes::timestamp::TimestampMillisecond;
 use datatypes::value::Value;
 use datatypes::vectors::{
-    Int64VectorBuilder, StringVectorBuilder, UInt32VectorBuilder, UInt64VectorBuilder, VectorRef,
+    Int64VectorBuilder, StringVectorBuilder, TimestampMillisecondVectorBuilder,
+    UInt32VectorBuilder, UInt64VectorBuilder, VectorRef,
 };
 use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::{ScanRequest, TableId};
 
 use crate::error::{
-    CreateRecordBatchSnafu, FlowInfoNotFoundSnafu, InternalSnafu, JsonSnafu, ListFlowsSnafu, Result,
+    CreateRecordBatchSnafu, FlowInfoNotFoundSnafu, InternalSnafu, JsonSnafu, ListFlowsSnafu,
+    Result, UpgradeWeakCatalogManagerRefSnafu,
 };
 use crate::information_schema::{Predicates, FLOWS};
 use crate::system_schema::information_schema::InformationTable;
@@ -59,6 +62,10 @@ pub const SOURCE_TABLE_IDS: &str = "source_table_ids";
 pub const SINK_TABLE_NAME: &str = "sink_table_name";
 pub const FLOWNODE_IDS: &str = "flownode_ids";
 pub const OPTIONS: &str = "options";
+pub const CREATED_TIME: &str = "created_time";
+pub const UPDATED_TIME: &str = "updated_time";
+pub const LAST_EXECUTION_TIME: &str = "last_execution_time";
+pub const SOURCE_TABLE_NAMES: &str = "source_table_names";
 
 /// The `information_schema.flows` to provides information about flows in databases.
 #[derive(Debug)]
@@ -99,6 +106,14 @@ impl InformationSchemaFlows {
                 (SINK_TABLE_NAME, CDT::string_datatype(), false),
                 (FLOWNODE_IDS, CDT::string_datatype(), true),
                 (OPTIONS, CDT::string_datatype(), true),
+                (CREATED_TIME, CDT::timestamp_millisecond_datatype(), false),
+                (UPDATED_TIME, CDT::timestamp_millisecond_datatype(), false),
+                (
+                    LAST_EXECUTION_TIME,
+                    CDT::timestamp_millisecond_datatype(),
+                    true,
+                ),
+                (SOURCE_TABLE_NAMES, CDT::string_datatype(), true),
             ]
             .into_iter()
             .map(|(name, ty, nullable)| ColumnSchema::new(name, ty, nullable))
@@ -170,6 +185,10 @@ struct InformationSchemaFlowsBuilder {
     sink_table_names: StringVectorBuilder,
     flownode_id_groups: StringVectorBuilder,
     option_groups: StringVectorBuilder,
+    created_time: TimestampMillisecondVectorBuilder,
+    updated_time: TimestampMillisecondVectorBuilder,
+    last_execution_time: TimestampMillisecondVectorBuilder,
+    source_table_names: StringVectorBuilder,
 }
 
 impl InformationSchemaFlowsBuilder {
@@ -196,6 +215,10 @@ impl InformationSchemaFlowsBuilder {
             sink_table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             flownode_id_groups: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             option_groups: StringVectorBuilder::with_capacity(INIT_CAPACITY),
+            created_time: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            updated_time: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            last_execution_time: TimestampMillisecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            source_table_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
         }
     }
 
@@ -235,13 +258,14 @@ impl InformationSchemaFlowsBuilder {
                     catalog_name: catalog_name.to_string(),
                     flow_name: flow_name.to_string(),
                 })?;
-            self.add_flow(&predicates, flow_id.flow_id(), flow_info, &flow_stat)?;
+            self.add_flow(&predicates, flow_id.flow_id(), flow_info, &flow_stat)
+                .await?;
         }
 
         self.finish()
     }
 
-    fn add_flow(
+    async fn add_flow(
         &mut self,
         predicates: &Predicates,
         flow_id: FlowId,
@@ -290,6 +314,36 @@ impl InformationSchemaFlowsBuilder {
                     input: format!("{:?}", flow_info.options()),
                 },
             )?));
+        self.created_time
+            .push(Some(flow_info.created_time().timestamp_millis().into()));
+        self.updated_time
+            .push(Some(flow_info.updated_time().timestamp_millis().into()));
+        self.last_execution_time
+            .push(flow_stat.as_ref().and_then(|state| {
+                state
+                    .last_exec_time_map
+                    .get(&flow_id)
+                    .map(|v| TimestampMillisecond::new(*v))
+            }));
+
+        let mut source_table_names = vec![];
+        let catalog_name = self.catalog_name.clone();
+        let catalog_manager = self
+            .catalog_manager
+            .upgrade()
+            .context(UpgradeWeakCatalogManagerRefSnafu)?;
+        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
+            source_table_names.extend(
+                catalog_manager
+                    .tables_by_ids(&catalog_name, &schema_name, flow_info.source_table_ids())
+                    .await?
+                    .into_iter()
+                    .map(|table| table.table_info().full_table_name()),
+            );
+        }
+
+        let source_table_names = source_table_names.join(",");
+        self.source_table_names.push(Some(&source_table_names));
 
         Ok(())
     }
@@ -307,6 +361,10 @@ impl InformationSchemaFlowsBuilder {
             Arc::new(self.sink_table_names.finish()),
             Arc::new(self.flownode_id_groups.finish()),
             Arc::new(self.option_groups.finish()),
+            Arc::new(self.created_time.finish()),
+            Arc::new(self.updated_time.finish()),
+            Arc::new(self.last_execution_time.finish()),
+            Arc::new(self.source_table_names.finish()),
         ];
         RecordBatch::new(self.schema.clone(), columns).context(CreateRecordBatchSnafu)
     }

src/catalog/src/system_schema/information_schema/information_memory_table.rs

@@ -20,7 +20,7 @@ use datatypes::vectors::{Int64Vector, StringVector, VectorRef};
 
 use super::table_names::*;
 use crate::system_schema::utils::tables::{
-    bigint_column, datetime_column, string_column, string_columns,
+    bigint_column, string_column, string_columns, timestamp_micro_column,
 };
 
 const NO_VALUE: &str = "NO";
@@ -163,17 +163,17 @@ pub(super) fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>
                 string_column("EVENT_BODY"),
                 string_column("EVENT_DEFINITION"),
                 string_column("EVENT_TYPE"),
-                datetime_column("EXECUTE_AT"),
+                timestamp_micro_column("EXECUTE_AT"),
                 bigint_column("INTERVAL_VALUE"),
                 string_column("INTERVAL_FIELD"),
                 string_column("SQL_MODE"),
-                datetime_column("STARTS"),
-                datetime_column("ENDS"),
+                timestamp_micro_column("STARTS"),
+                timestamp_micro_column("ENDS"),
                 string_column("STATUS"),
                 string_column("ON_COMPLETION"),
-                datetime_column("CREATED"),
-                datetime_column("LAST_ALTERED"),
-                datetime_column("LAST_EXECUTED"),
+                timestamp_micro_column("CREATED"),
+                timestamp_micro_column("LAST_ALTERED"),
+                timestamp_micro_column("LAST_EXECUTED"),
                 string_column("EVENT_COMMENT"),
                 bigint_column("ORIGINATOR"),
                 string_column("CHARACTER_SET_CLIENT"),
@@ -204,10 +204,10 @@ pub(super) fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>
                 bigint_column("INITIAL_SIZE"),
                 bigint_column("MAXIMUM_SIZE"),
                 bigint_column("AUTOEXTEND_SIZE"),
-                datetime_column("CREATION_TIME"),
-                datetime_column("LAST_UPDATE_TIME"),
-                datetime_column("LAST_ACCESS_TIME"),
-                datetime_column("RECOVER_TIME"),
+                timestamp_micro_column("CREATION_TIME"),
+                timestamp_micro_column("LAST_UPDATE_TIME"),
+                timestamp_micro_column("LAST_ACCESS_TIME"),
+                timestamp_micro_column("RECOVER_TIME"),
                 bigint_column("TRANSACTION_COUNTER"),
                 string_column("VERSION"),
                 string_column("ROW_FORMAT"),
@@ -217,9 +217,9 @@ pub(super) fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>
                 bigint_column("MAX_DATA_LENGTH"),
                 bigint_column("INDEX_LENGTH"),
                 bigint_column("DATA_FREE"),
-                datetime_column("CREATE_TIME"),
-                datetime_column("UPDATE_TIME"),
-                datetime_column("CHECK_TIME"),
+                timestamp_micro_column("CREATE_TIME"),
+                timestamp_micro_column("UPDATE_TIME"),
+                timestamp_micro_column("CHECK_TIME"),
                 string_column("CHECKSUM"),
                 string_column("STATUS"),
                 string_column("EXTRA"),
@@ -330,8 +330,8 @@ pub(super) fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>
                 string_column("SQL_DATA_ACCESS"),
                 string_column("SQL_PATH"),
                 string_column("SECURITY_TYPE"),
-                datetime_column("CREATED"),
-                datetime_column("LAST_ALTERED"),
+                timestamp_micro_column("CREATED"),
+                timestamp_micro_column("LAST_ALTERED"),
                 string_column("SQL_MODE"),
                 string_column("ROUTINE_COMMENT"),
                 string_column("DEFINER"),
@@ -383,7 +383,7 @@ pub(super) fn get_schema_columns(table_name: &str) -> (SchemaRef, Vec<VectorRef>
                 string_column("ACTION_REFERENCE_NEW_TABLE"),
                 string_column("ACTION_REFERENCE_OLD_ROW"),
                 string_column("ACTION_REFERENCE_NEW_ROW"),
-                datetime_column("CREATED"),
+                timestamp_micro_column("CREATED"),
                 string_column("SQL_MODE"),
                 string_column("DEFINER"),
                 string_column("CHARACTER_SET_CLIENT"),

src/catalog/src/system_schema/information_schema/partitions.rs

@@ -20,17 +20,18 @@ use common_catalog::consts::INFORMATION_SCHEMA_PARTITIONS_TABLE_ID;
 use common_error::ext::BoxedError;
 use common_recordbatch::adapter::RecordBatchStreamAdapter;
 use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
-use common_time::datetime::DateTime;
 use datafusion::execution::TaskContext;
 use datafusion::physical_plan::stream::RecordBatchStreamAdapter as DfRecordBatchStreamAdapter;
 use datafusion::physical_plan::streaming::PartitionStream as DfPartitionStream;
 use datafusion::physical_plan::SendableRecordBatchStream as DfSendableRecordBatchStream;
 use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
+use datatypes::timestamp::TimestampMicrosecond;
 use datatypes::value::Value;
 use datatypes::vectors::{
-    ConstantVector, DateTimeVector, DateTimeVectorBuilder, Int64Vector, Int64VectorBuilder,
-    MutableVector, StringVector, StringVectorBuilder, UInt64VectorBuilder,
+    ConstantVector, Int64Vector, Int64VectorBuilder, MutableVector, StringVector,
+    StringVectorBuilder, TimestampMicrosecondVector, TimestampMicrosecondVectorBuilder,
+    UInt64VectorBuilder,
 };
 use futures::{StreamExt, TryStreamExt};
 use partition::manager::PartitionInfo;
@@ -127,9 +128,21 @@ impl InformationSchemaPartitions {
             ColumnSchema::new("max_data_length", ConcreteDataType::int64_datatype(), true),
             ColumnSchema::new("index_length", ConcreteDataType::int64_datatype(), true),
             ColumnSchema::new("data_free", ConcreteDataType::int64_datatype(), true),
-            ColumnSchema::new("create_time", ConcreteDataType::datetime_datatype(), true),
-            ColumnSchema::new("update_time", ConcreteDataType::datetime_datatype(), true),
-            ColumnSchema::new("check_time", ConcreteDataType::datetime_datatype(), true),
+            ColumnSchema::new(
+                "create_time",
+                ConcreteDataType::timestamp_microsecond_datatype(),
+                true,
+            ),
+            ColumnSchema::new(
+                "update_time",
+                ConcreteDataType::timestamp_microsecond_datatype(),
+                true,
+            ),
+            ColumnSchema::new(
+                "check_time",
+                ConcreteDataType::timestamp_microsecond_datatype(),
+                true,
+            ),
             ColumnSchema::new("checksum", ConcreteDataType::int64_datatype(), true),
             ColumnSchema::new(
                 "partition_comment",
@@ -200,7 +213,7 @@ struct InformationSchemaPartitionsBuilder {
     partition_names: StringVectorBuilder,
     partition_ordinal_positions: Int64VectorBuilder,
     partition_expressions: StringVectorBuilder,
-    create_times: DateTimeVectorBuilder,
+    create_times: TimestampMicrosecondVectorBuilder,
     partition_ids: UInt64VectorBuilder,
 }
 
@@ -220,7 +233,7 @@ impl InformationSchemaPartitionsBuilder {
             partition_names: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             partition_ordinal_positions: Int64VectorBuilder::with_capacity(INIT_CAPACITY),
             partition_expressions: StringVectorBuilder::with_capacity(INIT_CAPACITY),
-            create_times: DateTimeVectorBuilder::with_capacity(INIT_CAPACITY),
+            create_times: TimestampMicrosecondVectorBuilder::with_capacity(INIT_CAPACITY),
             partition_ids: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
         }
     }
@@ -324,7 +337,7 @@ impl InformationSchemaPartitionsBuilder {
             };
 
             self.partition_expressions.push(expressions.as_deref());
-            self.create_times.push(Some(DateTime::from(
+            self.create_times.push(Some(TimestampMicrosecond::from(
                 table_info.meta.created_on.timestamp_millis(),
             )));
             self.partition_ids.push(Some(partition.id.as_u64()));
@@ -342,8 +355,8 @@ impl InformationSchemaPartitionsBuilder {
             Arc::new(Int64Vector::from(vec![None])),
             rows_num,
         ));
-        let null_datetime_vector = Arc::new(ConstantVector::new(
-            Arc::new(DateTimeVector::from(vec![None])),
+        let null_timestampmicrosecond_vector = Arc::new(ConstantVector::new(
+            Arc::new(TimestampMicrosecondVector::from(vec![None])),
             rows_num,
         ));
         let partition_methods = Arc::new(ConstantVector::new(
@@ -373,8 +386,8 @@ impl InformationSchemaPartitionsBuilder {
             null_i64_vector.clone(),
             Arc::new(self.create_times.finish()),
             // TODO(dennis): supports update_time
-            null_datetime_vector.clone(),
-            null_datetime_vector,
+            null_timestampmicrosecond_vector.clone(),
+            null_timestampmicrosecond_vector,
             null_i64_vector,
             null_string_vector.clone(),
             null_string_vector.clone(),

src/catalog/src/system_schema/information_schema/tables.rs

@@ -30,7 +30,8 @@ use datatypes::prelude::{ConcreteDataType, ScalarVectorBuilder, VectorRef};
 use datatypes::schema::{ColumnSchema, Schema, SchemaRef};
 use datatypes::value::Value;
 use datatypes::vectors::{
-    DateTimeVectorBuilder, StringVectorBuilder, UInt32VectorBuilder, UInt64VectorBuilder,
+    StringVectorBuilder, TimestampMicrosecondVectorBuilder, UInt32VectorBuilder,
+    UInt64VectorBuilder,
 };
 use futures::TryStreamExt;
 use snafu::{OptionExt, ResultExt};
@@ -105,9 +106,21 @@ impl InformationSchemaTables {
             ColumnSchema::new(TABLE_ROWS, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(DATA_FREE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(AUTO_INCREMENT, ConcreteDataType::uint64_datatype(), true),
-            ColumnSchema::new(CREATE_TIME, ConcreteDataType::datetime_datatype(), true),
-            ColumnSchema::new(UPDATE_TIME, ConcreteDataType::datetime_datatype(), true),
-            ColumnSchema::new(CHECK_TIME, ConcreteDataType::datetime_datatype(), true),
+            ColumnSchema::new(
+                CREATE_TIME,
+                ConcreteDataType::timestamp_microsecond_datatype(),
+                true,
+            ),
+            ColumnSchema::new(
+                UPDATE_TIME,
+                ConcreteDataType::timestamp_microsecond_datatype(),
+                true,
+            ),
+            ColumnSchema::new(
+                CHECK_TIME,
+                ConcreteDataType::timestamp_microsecond_datatype(),
+                true,
+            ),
             ColumnSchema::new(TABLE_COLLATION, ConcreteDataType::string_datatype(), true),
             ColumnSchema::new(CHECKSUM, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(CREATE_OPTIONS, ConcreteDataType::string_datatype(), true),
@@ -182,9 +195,9 @@ struct InformationSchemaTablesBuilder {
     max_index_length: UInt64VectorBuilder,
     data_free: UInt64VectorBuilder,
     auto_increment: UInt64VectorBuilder,
-    create_time: DateTimeVectorBuilder,
-    update_time: DateTimeVectorBuilder,
-    check_time: DateTimeVectorBuilder,
+    create_time: TimestampMicrosecondVectorBuilder,
+    update_time: TimestampMicrosecondVectorBuilder,
+    check_time: TimestampMicrosecondVectorBuilder,
     table_collation: StringVectorBuilder,
     checksum: UInt64VectorBuilder,
     create_options: StringVectorBuilder,
@@ -219,9 +232,9 @@ impl InformationSchemaTablesBuilder {
             max_index_length: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             data_free: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             auto_increment: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
-            create_time: DateTimeVectorBuilder::with_capacity(INIT_CAPACITY),
-            update_time: DateTimeVectorBuilder::with_capacity(INIT_CAPACITY),
-            check_time: DateTimeVectorBuilder::with_capacity(INIT_CAPACITY),
+            create_time: TimestampMicrosecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            update_time: TimestampMicrosecondVectorBuilder::with_capacity(INIT_CAPACITY),
+            check_time: TimestampMicrosecondVectorBuilder::with_capacity(INIT_CAPACITY),
             table_collation: StringVectorBuilder::with_capacity(INIT_CAPACITY),
             checksum: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             create_options: StringVectorBuilder::with_capacity(INIT_CAPACITY),

src/catalog/src/system_schema/utils/tables.rs

@@ -51,10 +51,10 @@ pub fn bigint_column(name: &str) -> ColumnSchema {
     )
 }
 
-pub fn datetime_column(name: &str) -> ColumnSchema {
+pub fn timestamp_micro_column(name: &str) -> ColumnSchema {
     ColumnSchema::new(
         str::to_lowercase(name),
-        ConcreteDataType::datetime_datatype(),
+        ConcreteDataType::timestamp_microsecond_datatype(),
         false,
     )
 }

src/cli/Cargo.toml

@@ -6,6 +6,7 @@ license.workspace = true
 
 [features]
 pg_kvbackend = ["common-meta/pg_kvbackend"]
+mysql_kvbackend = ["common-meta/mysql_kvbackend"]
 
 [lints]
 workspace = true
@@ -43,6 +44,10 @@ futures.workspace = true
 humantime.workspace = true
 meta-client.workspace = true
 nu-ansi-term = "0.46"
+opendal = { version = "0.51.1", features = [
+    "services-fs",
+    "services-s3",
+] }
 query.workspace = true
 rand.workspace = true
 reqwest.workspace = true

src/cli/src/bench.rs

@@ -23,6 +23,8 @@ use common_error::ext::BoxedError;
 use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::etcd::EtcdStore;
 use common_meta::kv_backend::memory::MemoryKvBackend;
+#[cfg(feature = "mysql_kvbackend")]
+use common_meta::kv_backend::rds::MySqlStore;
 #[cfg(feature = "pg_kvbackend")]
 use common_meta::kv_backend::rds::PgStore;
 use common_meta::peer::Peer;
@@ -63,6 +65,9 @@ pub struct BenchTableMetadataCommand {
     #[cfg(feature = "pg_kvbackend")]
     #[clap(long)]
     postgres_addr: Option<String>,
+    #[cfg(feature = "mysql_kvbackend")]
+    #[clap(long)]
+    mysql_addr: Option<String>,
     #[clap(long)]
     count: u32,
 }
@@ -86,6 +91,16 @@ impl BenchTableMetadataCommand {
             kv_backend
         };
 
+        #[cfg(feature = "mysql_kvbackend")]
+        let kv_backend = if let Some(mysql_addr) = &self.mysql_addr {
+            info!("Using mysql as kv backend");
+            MySqlStore::with_url(mysql_addr, "greptime_metakv", 128)
+                .await
+                .unwrap()
+        } else {
+            kv_backend
+        };
+
         let table_metadata_manager = Arc::new(TableMetadataManager::new(kv_backend));
 
         let tool = BenchTableMetadata {

src/cli/src/error.rs

@@ -276,6 +276,24 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+
+    #[snafu(display("OpenDAL operator failed"))]
+    OpenDal {
+        #[snafu(implicit)]
+        location: Location,
+        #[snafu(source)]
+        error: opendal::Error,
+    },
+    #[snafu(display("S3 config need be set"))]
+    S3ConfigNotSet {
+        #[snafu(implicit)]
+        location: Location,
+    },
+    #[snafu(display("Output directory not set"))]
+    OutputDirNotSet {
+        #[snafu(implicit)]
+        location: Location,
+    },
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -319,6 +337,9 @@ impl ErrorExt for Error {
             | Error::BuildClient { .. } => StatusCode::Unexpected,
 
             Error::Other { source, .. } => source.status_code(),
+            Error::OpenDal { .. } => StatusCode::Internal,
+            Error::S3ConfigNotSet { .. } => StatusCode::InvalidArguments,
+            Error::OutputDirNotSet { .. } => StatusCode::InvalidArguments,
 
             Error::BuildRuntime { source, .. } => source.status_code(),
 

src/cli/src/export.rs

@@ -21,15 +21,18 @@ use async_trait::async_trait;
 use clap::{Parser, ValueEnum};
 use common_error::ext::BoxedError;
 use common_telemetry::{debug, error, info};
+use opendal::layers::LoggingLayer;
+use opendal::{services, Operator};
 use serde_json::Value;
 use snafu::{OptionExt, ResultExt};
-use tokio::fs::File;
-use tokio::io::{AsyncWriteExt, BufWriter};
 use tokio::sync::Semaphore;
 use tokio::time::Instant;
 
 use crate::database::{parse_proxy_opts, DatabaseClient};
-use crate::error::{EmptyResultSnafu, Error, FileIoSnafu, Result, SchemaNotFoundSnafu};
+use crate::error::{
+    EmptyResultSnafu, Error, OpenDalSnafu, OutputDirNotSetSnafu, Result, S3ConfigNotSetSnafu,
+    SchemaNotFoundSnafu,
+};
 use crate::{database, Tool};
 
 type TableReference = (String, String, String);
@@ -52,8 +55,9 @@ pub struct ExportCommand {
     addr: String,
 
     /// Directory to put the exported data. E.g.: /tmp/greptimedb-export
+    /// for local export.
     #[clap(long)]
-    output_dir: String,
+    output_dir: Option<String>,
 
     /// The name of the catalog to export.
     #[clap(long, default_value = "greptime-*")]
@@ -101,10 +105,51 @@ pub struct ExportCommand {
     /// Disable proxy server, if set, will not use any proxy.
     #[clap(long)]
     no_proxy: bool,
+
+    /// if export data to s3
+    #[clap(long)]
+    s3: bool,
+
+    /// The s3 bucket name
+    /// if s3 is set, this is required
+    #[clap(long)]
+    s3_bucket: Option<String>,
+
+    /// The s3 endpoint
+    /// if s3 is set, this is required
+    #[clap(long)]
+    s3_endpoint: Option<String>,
+
+    /// The s3 access key
+    /// if s3 is set, this is required
+    #[clap(long)]
+    s3_access_key: Option<String>,
+
+    /// The s3 secret key
+    /// if s3 is set, this is required
+    #[clap(long)]
+    s3_secret_key: Option<String>,
+
+    /// The s3 region
+    /// if s3 is set, this is required
+    #[clap(long)]
+    s3_region: Option<String>,
 }
 
 impl ExportCommand {
     pub async fn build(&self) -> std::result::Result<Box<dyn Tool>, BoxedError> {
+        if self.s3
+            && (self.s3_bucket.is_none()
+                || self.s3_endpoint.is_none()
+                || self.s3_access_key.is_none()
+                || self.s3_secret_key.is_none()
+                || self.s3_region.is_none())
+        {
+            return Err(BoxedError::new(S3ConfigNotSetSnafu {}.build()));
+        }
+        if !self.s3 && self.output_dir.is_none() {
+            return Err(BoxedError::new(OutputDirNotSetSnafu {}.build()));
+        }
         let (catalog, schema) =
             database::split_database(&self.database).map_err(BoxedError::new)?;
         let proxy = parse_proxy_opts(self.proxy.clone(), self.no_proxy)?;
@@ -126,24 +171,43 @@ impl ExportCommand {
             target: self.target.clone(),
             start_time: self.start_time.clone(),
             end_time: self.end_time.clone(),
+            s3: self.s3,
+            s3_bucket: self.s3_bucket.clone(),
+            s3_endpoint: self.s3_endpoint.clone(),
+            s3_access_key: self.s3_access_key.clone(),
+            s3_secret_key: self.s3_secret_key.clone(),
+            s3_region: self.s3_region.clone(),
         }))
     }
 }
 
+#[derive(Clone)]
 pub struct Export {
     catalog: String,
     schema: Option<String>,
     database_client: DatabaseClient,
-    output_dir: String,
+    output_dir: Option<String>,
     parallelism: usize,
     target: ExportTarget,
     start_time: Option<String>,
     end_time: Option<String>,
+    s3: bool,
+    s3_bucket: Option<String>,
+    s3_endpoint: Option<String>,
+    s3_access_key: Option<String>,
+    s3_secret_key: Option<String>,
+    s3_region: Option<String>,
 }
 
 impl Export {
     fn catalog_path(&self) -> PathBuf {
-        PathBuf::from(&self.output_dir).join(&self.catalog)
+        if self.s3 {
+            PathBuf::from(&self.catalog)
+        } else if let Some(dir) = &self.output_dir {
+            PathBuf::from(dir).join(&self.catalog)
+        } else {
+            unreachable!("catalog_path: output_dir must be set when not using s3")
+        }
     }
 
     async fn get_db_names(&self) -> Result<Vec<String>> {
@@ -300,19 +364,23 @@ impl Export {
         let timer = Instant::now();
         let db_names = self.get_db_names().await?;
         let db_count = db_names.len();
+        let operator = self.build_operator().await?;
+
         for schema in db_names {
-            let db_dir = self.catalog_path().join(format!("{schema}/"));
-            tokio::fs::create_dir_all(&db_dir)
-                .await
-                .context(FileIoSnafu)?;
-            let file = db_dir.join("create_database.sql");
-            let mut file = File::create(file).await.context(FileIoSnafu)?;
             let create_database = self
                 .show_create("DATABASE", &self.catalog, &schema, None)
                 .await?;
-            file.write_all(create_database.as_bytes())
-                .await
-                .context(FileIoSnafu)?;
+
+            let file_path = self.get_file_path(&schema, "create_database.sql");
+            self.write_to_storage(&operator, &file_path, create_database.into_bytes())
+                .await?;
+
+            info!(
+                "Exported {}.{} database creation SQL to {}",
+                self.catalog,
+                schema,
+                self.format_output_path(&file_path)
+            );
         }
 
         let elapsed = timer.elapsed();
@@ -326,149 +394,267 @@ impl Export {
         let semaphore = Arc::new(Semaphore::new(self.parallelism));
         let db_names = self.get_db_names().await?;
         let db_count = db_names.len();
+        let operator = Arc::new(self.build_operator().await?);
         let mut tasks = Vec::with_capacity(db_names.len());
+
         for schema in db_names {
             let semaphore_moved = semaphore.clone();
+            let export_self = self.clone();
+            let operator = operator.clone();
             tasks.push(async move {
                 let _permit = semaphore_moved.acquire().await.unwrap();
-                let (metric_physical_tables, remaining_tables, views) =
-                    self.get_table_list(&self.catalog, &schema).await?;
-                let table_count =
-                    metric_physical_tables.len() + remaining_tables.len() + views.len();
-                let db_dir = self.catalog_path().join(format!("{schema}/"));
-                tokio::fs::create_dir_all(&db_dir)
-                    .await
-                    .context(FileIoSnafu)?;
-                let file = db_dir.join("create_tables.sql");
-                let mut file = File::create(file).await.context(FileIoSnafu)?;
-                for (c, s, t) in metric_physical_tables.into_iter().chain(remaining_tables) {
-                    let create_table = self.show_create("TABLE", &c, &s, Some(&t)).await?;
-                    file.write_all(create_table.as_bytes())
-                        .await
-                        .context(FileIoSnafu)?;
-                }
-                for (c, s, v) in views {
-                    let create_view = self.show_create("VIEW", &c, &s, Some(&v)).await?;
-                    file.write_all(create_view.as_bytes())
-                        .await
-                        .context(FileIoSnafu)?;
+                let (metric_physical_tables, remaining_tables, views) = export_self
+                    .get_table_list(&export_self.catalog, &schema)
+                    .await?;
+
+                // Create directory if needed for file system storage
+                if !export_self.s3 {
+                    let db_dir = format!("{}/{}/", export_self.catalog, schema);
+                    operator.create_dir(&db_dir).await.context(OpenDalSnafu)?;
                 }
 
+                let file_path = export_self.get_file_path(&schema, "create_tables.sql");
+                let mut content = Vec::new();
+
+                // Add table creation SQL
+                for (c, s, t) in metric_physical_tables.iter().chain(&remaining_tables) {
+                    let create_table = export_self.show_create("TABLE", c, s, Some(t)).await?;
+                    content.extend_from_slice(create_table.as_bytes());
+                }
+
+                // Add view creation SQL
+                for (c, s, v) in &views {
+                    let create_view = export_self.show_create("VIEW", c, s, Some(v)).await?;
+                    content.extend_from_slice(create_view.as_bytes());
+                }
+
+                // Write to storage
+                export_self
+                    .write_to_storage(&operator, &file_path, content)
+                    .await?;
+
                 info!(
-                    "Finished exporting {}.{schema} with {table_count} table schemas to path: {}",
-                    self.catalog,
-                    db_dir.to_string_lossy()
+                    "Finished exporting {}.{schema} with {} table schemas to path: {}",
+                    export_self.catalog,
+                    metric_physical_tables.len() + remaining_tables.len() + views.len(),
+                    export_self.format_output_path(&file_path)
                 );
 
                 Ok::<(), Error>(())
             });
         }
 
-        let success = futures::future::join_all(tasks)
-            .await
-            .into_iter()
-            .filter(|r| match r {
-                Ok(_) => true,
-                Err(e) => {
-                    error!(e; "export schema job failed");
-                    false
-                }
-            })
-            .count();
-
+        let success = self.execute_tasks(tasks).await;
         let elapsed = timer.elapsed();
         info!("Success {success}/{db_count} jobs, cost: {elapsed:?}");
 
         Ok(())
     }
 
+    async fn build_operator(&self) -> Result<Operator> {
+        if self.s3 {
+            self.build_s3_operator().await
+        } else {
+            self.build_fs_operator().await
+        }
+    }
+
+    async fn build_s3_operator(&self) -> Result<Operator> {
+        let mut builder = services::S3::default().root("").bucket(
+            self.s3_bucket
+                .as_ref()
+                .expect("s3_bucket must be provided when s3 is enabled"),
+        );
+
+        if let Some(endpoint) = self.s3_endpoint.as_ref() {
+            builder = builder.endpoint(endpoint);
+        }
+
+        if let Some(region) = self.s3_region.as_ref() {
+            builder = builder.region(region);
+        }
+
+        if let Some(key_id) = self.s3_access_key.as_ref() {
+            builder = builder.access_key_id(key_id);
+        }
+
+        if let Some(secret_key) = self.s3_secret_key.as_ref() {
+            builder = builder.secret_access_key(secret_key);
+        }
+
+        let op = Operator::new(builder)
+            .context(OpenDalSnafu)?
+            .layer(LoggingLayer::default())
+            .finish();
+        Ok(op)
+    }
+
+    async fn build_fs_operator(&self) -> Result<Operator> {
+        let root = self
+            .output_dir
+            .as_ref()
+            .context(OutputDirNotSetSnafu)?
+            .clone();
+        let op = Operator::new(services::Fs::default().root(&root))
+            .context(OpenDalSnafu)?
+            .layer(LoggingLayer::default())
+            .finish();
+        Ok(op)
+    }
+
     async fn export_database_data(&self) -> Result<()> {
         let timer = Instant::now();
         let semaphore = Arc::new(Semaphore::new(self.parallelism));
         let db_names = self.get_db_names().await?;
         let db_count = db_names.len();
         let mut tasks = Vec::with_capacity(db_count);
+        let operator = Arc::new(self.build_operator().await?);
+        let with_options = build_with_options(&self.start_time, &self.end_time);
+
         for schema in db_names {
             let semaphore_moved = semaphore.clone();
+            let export_self = self.clone();
+            let with_options_clone = with_options.clone();
+            let operator = operator.clone();
+
             tasks.push(async move {
                 let _permit = semaphore_moved.acquire().await.unwrap();
-                let db_dir = self.catalog_path().join(format!("{schema}/"));
-                tokio::fs::create_dir_all(&db_dir)
-                    .await
-                    .context(FileIoSnafu)?;
 
-                let with_options = match (&self.start_time, &self.end_time) {
-                    (Some(start_time), Some(end_time)) => {
-                        format!(
-                            "WITH (FORMAT='parquet', start_time='{}', end_time='{}')",
-                            start_time, end_time
-                        )
-                    }
-                    (Some(start_time), None) => {
-                        format!("WITH (FORMAT='parquet', start_time='{}')", start_time)
-                    }
-                    (None, Some(end_time)) => {
-                        format!("WITH (FORMAT='parquet', end_time='{}')", end_time)
-                    }
-                    (None, None) => "WITH (FORMAT='parquet')".to_string(),
-                };
+                // Create directory if not using S3
+                if !export_self.s3 {
+                    let db_dir = format!("{}/{}/", export_self.catalog, schema);
+                    operator.create_dir(&db_dir).await.context(OpenDalSnafu)?;
+                }
 
+                let (path, connection_part) = export_self.get_storage_params(&schema);
+
+                // Execute COPY DATABASE TO command
                 let sql = format!(
-                    r#"COPY DATABASE "{}"."{}" TO '{}' {};"#,
-                    self.catalog,
-                    schema,
-                    db_dir.to_str().unwrap(),
-                    with_options
+                    r#"COPY DATABASE "{}"."{}" TO '{}' WITH ({}){};"#,
+                    export_self.catalog, schema, path, with_options_clone, connection_part
+                );
+                info!("Executing sql: {sql}");
+                export_self.database_client.sql_in_public(&sql).await?;
+                info!(
+                    "Finished exporting {}.{} data to {}",
+                    export_self.catalog, schema, path
                 );
 
-                info!("Executing sql: {sql}");
+                // Create copy_from.sql file
+                let copy_database_from_sql = format!(
+                    r#"COPY DATABASE "{}"."{}" FROM '{}' WITH ({}){};"#,
+                    export_self.catalog, schema, path, with_options_clone, connection_part
+                );
 
-                self.database_client.sql_in_public(&sql).await?;
+                let copy_from_path = export_self.get_file_path(&schema, "copy_from.sql");
+                export_self
+                    .write_to_storage(
+                        &operator,
+                        &copy_from_path,
+                        copy_database_from_sql.into_bytes(),
+                    )
+                    .await?;
 
                 info!(
-                    "Finished exporting {}.{schema} data into path: {}",
-                    self.catalog,
-                    db_dir.to_string_lossy()
-                );
-
-                // The export copy from sql
-                let copy_from_file = db_dir.join("copy_from.sql");
-                let mut writer =
-                    BufWriter::new(File::create(copy_from_file).await.context(FileIoSnafu)?);
-                let copy_database_from_sql = format!(
-                    r#"COPY DATABASE "{}"."{}" FROM '{}' WITH (FORMAT='parquet');"#,
-                    self.catalog,
+                    "Finished exporting {}.{} copy_from.sql to {}",
+                    export_self.catalog,
                     schema,
-                    db_dir.to_str().unwrap()
+                    export_self.format_output_path(&copy_from_path)
                 );
-                writer
-                    .write(copy_database_from_sql.as_bytes())
-                    .await
-                    .context(FileIoSnafu)?;
-                writer.flush().await.context(FileIoSnafu)?;
-
-                info!("Finished exporting {}.{schema} copy_from.sql", self.catalog);
 
                 Ok::<(), Error>(())
-            })
+            });
         }
 
-        let success = futures::future::join_all(tasks)
+        let success = self.execute_tasks(tasks).await;
+        let elapsed = timer.elapsed();
+        info!("Success {success}/{db_count} jobs, costs: {elapsed:?}");
+
+        Ok(())
+    }
+
+    fn get_file_path(&self, schema: &str, file_name: &str) -> String {
+        format!("{}/{}/{}", self.catalog, schema, file_name)
+    }
+
+    fn format_output_path(&self, file_path: &str) -> String {
+        if self.s3 {
+            format!(
+                "s3://{}/{}",
+                self.s3_bucket.as_ref().unwrap_or(&String::new()),
+                file_path
+            )
+        } else {
+            format!(
+                "{}/{}",
+                self.output_dir.as_ref().unwrap_or(&String::new()),
+                file_path
+            )
+        }
+    }
+
+    async fn write_to_storage(
+        &self,
+        op: &Operator,
+        file_path: &str,
+        content: Vec<u8>,
+    ) -> Result<()> {
+        op.write(file_path, content).await.context(OpenDalSnafu)
+    }
+
+    fn get_storage_params(&self, schema: &str) -> (String, String) {
+        if self.s3 {
+            let s3_path = format!(
+                "s3://{}/{}/{}/",
+                // Safety: s3_bucket is required when s3 is enabled
+                self.s3_bucket.as_ref().unwrap(),
+                self.catalog,
+                schema
+            );
+
+            // endpoint is optional
+            let endpoint_option = if let Some(endpoint) = self.s3_endpoint.as_ref() {
+                format!(", ENDPOINT='{}'", endpoint)
+            } else {
+                String::new()
+            };
+
+            // Safety: All s3 options are required
+            let connection_options = format!(
+                "ACCESS_KEY_ID='{}', SECRET_ACCESS_KEY='{}', REGION='{}'{}",
+                self.s3_access_key.as_ref().unwrap(),
+                self.s3_secret_key.as_ref().unwrap(),
+                self.s3_region.as_ref().unwrap(),
+                endpoint_option
+            );
+
+            (s3_path, format!(" CONNECTION ({})", connection_options))
+        } else {
+            (
+                self.catalog_path()
+                    .join(format!("{schema}/"))
+                    .to_string_lossy()
+                    .to_string(),
+                String::new(),
+            )
+        }
+    }
+
+    async fn execute_tasks(
+        &self,
+        tasks: Vec<impl std::future::Future<Output = Result<()>>>,
+    ) -> usize {
+        futures::future::join_all(tasks)
             .await
             .into_iter()
             .filter(|r| match r {
                 Ok(_) => true,
                 Err(e) => {
-                    error!(e; "export database job failed");
+                    error!(e; "export job failed");
                     false
                 }
             })
-            .count();
-        let elapsed = timer.elapsed();
-
-        info!("Success {success}/{db_count} jobs, costs: {elapsed:?}");
-
-        Ok(())
+            .count()
     }
 }
 
@@ -493,3 +679,15 @@ impl Tool for Export {
         }
     }
 }
+
+/// Builds the WITH options string for SQL commands, assuming consistent syntax across S3 and local exports.
+fn build_with_options(start_time: &Option<String>, end_time: &Option<String>) -> String {
+    let mut options = vec!["format = 'parquet'".to_string()];
+    if let Some(start) = start_time {
+        options.push(format!("start_time = '{}'", start));
+    }
+    if let Some(end) = end_time {
+        options.push(format!("end_time = '{}'", end));
+    }
+    options.join(", ")
+}

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/Cargo.toml

@@ -17,6 +17,7 @@ api.workspace = true
 arc-swap = "1.0"
 async-trait.workspace = true
 bincode = "1.3"
+chrono.workspace = true
 common-base.workspace = true
 common-catalog.workspace = true
 common-error.workspace = true

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/function/src/scalars/date/date_format.rs

@@ -43,7 +43,6 @@ impl Function for DateFormatFunction {
         helper::one_of_sigs2(
             vec![
                 ConcreteDataType::date_datatype(),
-                ConcreteDataType::datetime_datatype(),
                 ConcreteDataType::timestamp_second_datatype(),
                 ConcreteDataType::timestamp_millisecond_datatype(),
                 ConcreteDataType::timestamp_microsecond_datatype(),
@@ -105,22 +104,6 @@ impl Function for DateFormatFunction {
                     results.push(result.as_deref());
                 }
             }
-            ConcreteDataType::DateTime(_) => {
-                for i in 0..size {
-                    let datetime = left.get(i).as_datetime();
-                    let format = formats.get(i).as_string();
-
-                    let result = match (datetime, format) {
-                        (Some(datetime), Some(fmt)) => datetime
-                            .as_formatted_string(&fmt, Some(&func_ctx.query_ctx.timezone()))
-                            .map_err(BoxedError::new)
-                            .context(error::ExecuteSnafu)?,
-                        _ => None,
-                    };
-
-                    results.push(result.as_deref());
-                }
-            }
             _ => {
                 return UnsupportedInputDataTypeSnafu {
                     function: NAME,
@@ -147,7 +130,7 @@ mod tests {
     use common_query::prelude::{TypeSignature, Volatility};
     use datatypes::prelude::{ConcreteDataType, ScalarVector};
     use datatypes::value::Value;
-    use datatypes::vectors::{DateTimeVector, DateVector, StringVector, TimestampSecondVector};
+    use datatypes::vectors::{DateVector, StringVector, TimestampSecondVector};
 
     use super::{DateFormatFunction, *};
 
@@ -169,16 +152,11 @@ mod tests {
             ConcreteDataType::string_datatype(),
             f.return_type(&[ConcreteDataType::date_datatype()]).unwrap()
         );
-        assert_eq!(
-            ConcreteDataType::string_datatype(),
-            f.return_type(&[ConcreteDataType::datetime_datatype()])
-                .unwrap()
-        );
         assert!(matches!(f.signature(),
                          Signature {
                              type_signature: TypeSignature::OneOf(sigs),
                              volatility: Volatility::Immutable
-                         } if  sigs.len() == 6));
+                         } if  sigs.len() == 5));
     }
 
     #[test]
@@ -262,45 +240,4 @@ mod tests {
             }
         }
     }
-
-    #[test]
-    fn test_datetime_date_format() {
-        let f = DateFormatFunction;
-
-        let dates = vec![Some(123), None, Some(42), None];
-        let formats = vec![
-            "%Y-%m-%d %T.%3f",
-            "%Y-%m-%d %T.%3f",
-            "%Y-%m-%d %T.%3f",
-            "%Y-%m-%d %T.%3f",
-        ];
-        let results = [
-            Some("1970-01-01 00:00:00.123"),
-            None,
-            Some("1970-01-01 00:00:00.042"),
-            None,
-        ];
-
-        let date_vector = DateTimeVector::from(dates.clone());
-        let interval_vector = StringVector::from_vec(formats);
-        let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
-
-        assert_eq!(4, vector.len());
-        for (i, _t) in dates.iter().enumerate() {
-            let v = vector.get(i);
-            let result = results.get(i).unwrap();
-
-            if result.is_none() {
-                assert_eq!(Value::Null, v);
-                continue;
-            }
-            match v {
-                Value::String(s) => {
-                    assert_eq!(s.as_utf8(), result.unwrap());
-                }
-                _ => unreachable!(),
-            }
-        }
-    }
 }

src/common/function/src/scalars/date/date_sub.rs

@@ -118,11 +118,6 @@ mod tests {
             ConcreteDataType::date_datatype(),
             f.return_type(&[ConcreteDataType::date_datatype()]).unwrap()
         );
-        assert_eq!(
-            ConcreteDataType::datetime_datatype(),
-            f.return_type(&[ConcreteDataType::datetime_datatype()])
-                .unwrap()
-        );
         assert!(
             matches!(f.signature(),
                          Signature {

src/common/function/src/scalars/timestamp/greatest.rs

@@ -23,7 +23,7 @@ use datatypes::arrow::array::AsArray;
 use datatypes::arrow::compute::cast;
 use datatypes::arrow::compute::kernels::zip;
 use datatypes::arrow::datatypes::{
-    DataType as ArrowDataType, Date32Type, Date64Type, TimestampMicrosecondType,
+    DataType as ArrowDataType, Date32Type, TimeUnit, TimestampMicrosecondType,
     TimestampMillisecondType, TimestampNanosecondType, TimestampSecondType,
 };
 use datatypes::prelude::ConcreteDataType;
@@ -69,9 +69,8 @@ impl Function for GreatestFunction {
         );
 
         match &input_types[0] {
-            ConcreteDataType::String(_) => Ok(ConcreteDataType::datetime_datatype()),
+            ConcreteDataType::String(_) => Ok(ConcreteDataType::timestamp_millisecond_datatype()),
             ConcreteDataType::Date(_) => Ok(ConcreteDataType::date_datatype()),
-            ConcreteDataType::DateTime(_) => Ok(ConcreteDataType::datetime_datatype()),
             ConcreteDataType::Timestamp(ts_type) => Ok(ConcreteDataType::Timestamp(*ts_type)),
             _ => UnsupportedInputDataTypeSnafu {
                 function: NAME,
@@ -87,7 +86,6 @@ impl Function for GreatestFunction {
             vec![
                 ConcreteDataType::string_datatype(),
                 ConcreteDataType::date_datatype(),
-                ConcreteDataType::datetime_datatype(),
                 ConcreteDataType::timestamp_nanosecond_datatype(),
                 ConcreteDataType::timestamp_microsecond_datatype(),
                 ConcreteDataType::timestamp_millisecond_datatype(),
@@ -109,20 +107,24 @@ impl Function for GreatestFunction {
         );
         match columns[0].data_type() {
             ConcreteDataType::String(_) => {
-                // Treats string as `DateTime` type.
-                let column1 = cast(&columns[0].to_arrow_array(), &ArrowDataType::Date64)
-                    .context(ArrowComputeSnafu)?;
-                let column1 = column1.as_primitive::<Date64Type>();
-                let column2 = cast(&columns[1].to_arrow_array(), &ArrowDataType::Date64)
-                    .context(ArrowComputeSnafu)?;
-                let column2 = column2.as_primitive::<Date64Type>();
+                let column1 = cast(
+                    &columns[0].to_arrow_array(),
+                    &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
+                )
+                .context(ArrowComputeSnafu)?;
+                let column1 = column1.as_primitive::<TimestampMillisecondType>();
+                let column2 = cast(
+                    &columns[1].to_arrow_array(),
+                    &ArrowDataType::Timestamp(TimeUnit::Millisecond, None),
+                )
+                .context(ArrowComputeSnafu)?;
+                let column2 = column2.as_primitive::<TimestampMillisecondType>();
                 let boolean_array = gt(&column1, &column2).context(ArrowComputeSnafu)?;
                 let result =
                     zip::zip(&boolean_array, &column1, &column2).context(ArrowComputeSnafu)?;
                 Ok(Helper::try_into_vector(&result).context(error::FromArrowArraySnafu)?)
             }
             ConcreteDataType::Date(_) => gt_time_types!(Date32Type, columns),
-            ConcreteDataType::DateTime(_) => gt_time_types!(Date64Type, columns),
             ConcreteDataType::Timestamp(ts_type) => match ts_type {
                 TimestampType::Second(_) => gt_time_types!(TimestampSecondType, columns),
                 TimestampType::Millisecond(_) => {
@@ -155,15 +157,15 @@ mod tests {
     use std::sync::Arc;
 
     use common_time::timestamp::TimeUnit;
-    use common_time::{Date, DateTime, Timestamp};
+    use common_time::{Date, Timestamp};
     use datatypes::types::{
-        DateTimeType, DateType, TimestampMicrosecondType, TimestampMillisecondType,
-        TimestampNanosecondType, TimestampSecondType,
+        DateType, TimestampMicrosecondType, TimestampMillisecondType, TimestampNanosecondType,
+        TimestampSecondType,
     };
     use datatypes::value::Value;
     use datatypes::vectors::{
-        DateTimeVector, DateVector, StringVector, TimestampMicrosecondVector,
-        TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, Vector,
+        DateVector, StringVector, TimestampMicrosecondVector, TimestampMillisecondVector,
+        TimestampNanosecondVector, TimestampSecondVector, Vector,
     };
     use paste::paste;
 
@@ -178,7 +180,7 @@ mod tests {
                     ConcreteDataType::string_datatype()
                 ])
                 .unwrap(),
-            ConcreteDataType::DateTime(DateTimeType)
+            ConcreteDataType::timestamp_millisecond_datatype()
         );
         let columns = vec![
             Arc::new(StringVector::from(vec![
@@ -194,15 +196,18 @@ mod tests {
         let result = function
             .eval(&FunctionContext::default(), &columns)
             .unwrap();
-        let result = result.as_any().downcast_ref::<DateTimeVector>().unwrap();
+        let result = result
+            .as_any()
+            .downcast_ref::<TimestampMillisecondVector>()
+            .unwrap();
         assert_eq!(result.len(), 2);
         assert_eq!(
             result.get(0),
-            Value::DateTime(DateTime::from_str("2001-02-01 00:00:00", None).unwrap())
+            Value::Timestamp(Timestamp::from_str("2001-02-01 00:00:00", None).unwrap())
         );
         assert_eq!(
             result.get(1),
-            Value::DateTime(DateTime::from_str("2012-12-23 00:00:00", None).unwrap())
+            Value::Timestamp(Timestamp::from_str("2012-12-23 00:00:00", None).unwrap())
         );
     }
 
@@ -245,30 +250,33 @@ mod tests {
         assert_eq!(
             function
                 .return_type(&[
-                    ConcreteDataType::datetime_datatype(),
-                    ConcreteDataType::datetime_datatype()
+                    ConcreteDataType::timestamp_millisecond_datatype(),
+                    ConcreteDataType::timestamp_millisecond_datatype()
                 ])
                 .unwrap(),
-            ConcreteDataType::DateTime(DateTimeType)
+            ConcreteDataType::timestamp_millisecond_datatype()
         );
 
         let columns = vec![
-            Arc::new(DateTimeVector::from_slice(vec![-1, 2])) as _,
-            Arc::new(DateTimeVector::from_slice(vec![0, 1])) as _,
+            Arc::new(TimestampMillisecondVector::from_slice(vec![-1, 2])) as _,
+            Arc::new(TimestampMillisecondVector::from_slice(vec![0, 1])) as _,
         ];
 
         let result = function
             .eval(&FunctionContext::default(), &columns)
             .unwrap();
-        let result = result.as_any().downcast_ref::<DateTimeVector>().unwrap();
+        let result = result
+            .as_any()
+            .downcast_ref::<TimestampMillisecondVector>()
+            .unwrap();
         assert_eq!(result.len(), 2);
         assert_eq!(
             result.get(0),
-            Value::DateTime(DateTime::from_str("1970-01-01 00:00:00", None).unwrap())
+            Value::Timestamp(Timestamp::from_str("1970-01-01 00:00:00", None).unwrap())
         );
         assert_eq!(
             result.get(1),
-            Value::DateTime(DateTime::from_str("1970-01-01 00:00:00.002", None).unwrap())
+            Value::Timestamp(Timestamp::from_str("1970-01-01 00:00:00.002", None).unwrap())
         );
     }
 

src/common/function/src/scalars/timestamp/to_unixtime.rs

@@ -17,7 +17,7 @@ use std::sync::Arc;
 
 use common_query::error::{InvalidFuncArgsSnafu, Result, UnsupportedInputDataTypeSnafu};
 use common_query::prelude::{Signature, Volatility};
-use common_time::{Date, DateTime, Timestamp};
+use common_time::{Date, Timestamp};
 use datatypes::prelude::ConcreteDataType;
 use datatypes::vectors::{Int64Vector, VectorRef};
 use snafu::ensure;
@@ -32,10 +32,6 @@ const NAME: &str = "to_unixtime";
 
 fn convert_to_seconds(arg: &str, func_ctx: &FunctionContext) -> Option<i64> {
     let timezone = &func_ctx.query_ctx.timezone();
-    if let Ok(dt) = DateTime::from_str(arg, Some(timezone)) {
-        return Some(dt.val() / 1000);
-    }
-
     if let Ok(ts) = Timestamp::from_str(arg, Some(timezone)) {
         return Some(ts.split().0);
     }
@@ -59,12 +55,6 @@ fn convert_dates_to_seconds(vector: &VectorRef) -> Vec<Option<i64>> {
         .collect::<Vec<Option<i64>>>()
 }
 
-fn convert_datetimes_to_seconds(vector: &VectorRef) -> Vec<Option<i64>> {
-    (0..vector.len())
-        .map(|i| vector.get(i).as_datetime().map(|dt| dt.val() / 1000))
-        .collect::<Vec<Option<i64>>>()
-}
-
 impl Function for ToUnixtimeFunction {
     fn name(&self) -> &str {
         NAME
@@ -82,7 +72,6 @@ impl Function for ToUnixtimeFunction {
                 ConcreteDataType::int32_datatype(),
                 ConcreteDataType::int64_datatype(),
                 ConcreteDataType::date_datatype(),
-                ConcreteDataType::datetime_datatype(),
                 ConcreteDataType::timestamp_second_datatype(),
                 ConcreteDataType::timestamp_millisecond_datatype(),
                 ConcreteDataType::timestamp_microsecond_datatype(),
@@ -119,10 +108,6 @@ impl Function for ToUnixtimeFunction {
                 let seconds = convert_dates_to_seconds(vector);
                 Ok(Arc::new(Int64Vector::from(seconds)))
             }
-            ConcreteDataType::DateTime(_) => {
-                let seconds = convert_datetimes_to_seconds(vector);
-                Ok(Arc::new(Int64Vector::from(seconds)))
-            }
             ConcreteDataType::Timestamp(_) => {
                 let seconds = convert_timestamps_to_seconds(vector);
                 Ok(Arc::new(Int64Vector::from(seconds)))
@@ -148,7 +133,7 @@ mod tests {
     use datatypes::prelude::ConcreteDataType;
     use datatypes::value::Value;
     use datatypes::vectors::{
-        DateTimeVector, DateVector, StringVector, TimestampMillisecondVector, TimestampSecondVector,
+        DateVector, StringVector, TimestampMillisecondVector, TimestampSecondVector,
     };
 
     use super::{ToUnixtimeFunction, *};
@@ -171,7 +156,6 @@ mod tests {
                              ConcreteDataType::int32_datatype(),
                              ConcreteDataType::int64_datatype(),
                              ConcreteDataType::date_datatype(),
-                             ConcreteDataType::datetime_datatype(),
                              ConcreteDataType::timestamp_second_datatype(),
                              ConcreteDataType::timestamp_millisecond_datatype(),
                              ConcreteDataType::timestamp_microsecond_datatype(),
@@ -253,31 +237,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn test_datetime_to_unixtime() {
-        let f = ToUnixtimeFunction;
-
-        let times = vec![Some(123000), None, Some(42000), None];
-        let results = [Some(123), None, Some(42), None];
-        let date_vector = DateTimeVector::from(times.clone());
-        let args: Vec<VectorRef> = vec![Arc::new(date_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
-        assert_eq!(4, vector.len());
-        for (i, _t) in times.iter().enumerate() {
-            let v = vector.get(i);
-            if i == 1 || i == 3 {
-                assert_eq!(Value::Null, v);
-                continue;
-            }
-            match v {
-                Value::Int64(ts) => {
-                    assert_eq!(ts, (*results.get(i).unwrap()).unwrap());
-                }
-                _ => unreachable!(),
-            }
-        }
-    }
-
     #[test]
     fn test_timestamp_to_unixtime() {
         let f = ToUnixtimeFunction;

src/common/grpc/src/select.rs

@@ -17,8 +17,8 @@ use api::v1::column::Values;
 use common_base::BitVec;
 use datatypes::types::{IntervalType, TimeType, TimestampType, WrapperType};
 use datatypes::vectors::{
-    BinaryVector, BooleanVector, DateTimeVector, DateVector, Decimal128Vector, Float32Vector,
-    Float64Vector, Int16Vector, Int32Vector, Int64Vector, Int8Vector, IntervalDayTimeVector,
+    BinaryVector, BooleanVector, DateVector, Decimal128Vector, Float32Vector, Float64Vector,
+    Int16Vector, Int32Vector, Int64Vector, Int8Vector, IntervalDayTimeVector,
     IntervalMonthDayNanoVector, IntervalYearMonthVector, StringVector, TimeMicrosecondVector,
     TimeMillisecondVector, TimeNanosecondVector, TimeSecondVector, TimestampMicrosecondVector,
     TimestampMillisecondVector, TimestampNanosecondVector, TimestampSecondVector, UInt16Vector,
@@ -141,12 +141,6 @@ pub fn values(arrays: &[VectorRef]) -> Result<Values> {
         (ConcreteDataType::Date(_), DateVector, date_values, |x| {
             x.val()
         }),
-        (
-            ConcreteDataType::DateTime(_),
-            DateTimeVector,
-            datetime_values,
-            |x| { x.val() }
-        ),
         (
             ConcreteDataType::Timestamp(TimestampType::Second(_)),
             TimestampSecondVector,

src/common/macro/src/lib.rs

@@ -18,11 +18,13 @@ mod print_caller;
 mod range_fn;
 mod stack_trace_debug;
 mod utils;
+
 use aggr_func::{impl_aggr_func_type_store, impl_as_aggr_func_creator};
 use print_caller::process_print_caller;
 use proc_macro::TokenStream;
+use quote::quote;
 use range_fn::process_range_fn;
-use syn::{parse_macro_input, DeriveInput};
+use syn::{parse_macro_input, Data, DeriveInput, Fields};
 
 use crate::admin_fn::process_admin_fn;
 
@@ -136,3 +138,51 @@ pub fn print_caller(args: TokenStream, input: TokenStream) -> TokenStream {
 pub fn stack_trace_debug(args: TokenStream, input: TokenStream) -> TokenStream {
     stack_trace_debug::stack_trace_style_impl(args.into(), input.into()).into()
 }
+
+/// Generates implementation for `From<&TableMeta> for TableMetaBuilder`
+#[proc_macro_derive(ToMetaBuilder)]
+pub fn derive_meta_builder(input: TokenStream) -> TokenStream {
+    let input = parse_macro_input!(input as DeriveInput);
+
+    let Data::Struct(data_struct) = input.data else {
+        panic!("ToMetaBuilder can only be derived for structs");
+    };
+
+    let Fields::Named(fields) = data_struct.fields else {
+        panic!("ToMetaBuilder can only be derived for structs with named fields");
+    };
+
+    // Check that this is being applied to TableMeta struct
+    if input.ident != "TableMeta" {
+        panic!("ToMetaBuilder can only be derived for TableMeta struct");
+    }
+
+    let field_init = fields.named.iter().map(|field| {
+        let field_name = field.ident.as_ref().unwrap();
+        quote! {
+            #field_name: Default::default(),
+        }
+    });
+
+    let field_assignments = fields.named.iter().map(|field| {
+        let field_name = field.ident.as_ref().unwrap();
+        quote! {
+            builder.#field_name(meta.#field_name.clone());
+        }
+    });
+
+    let gen = quote! {
+        impl From<&TableMeta> for TableMetaBuilder {
+            fn from(meta: &TableMeta) -> Self {
+                let mut builder = Self {
+                    #(#field_init)*
+                };
+
+                #(#field_assignments)*
+                builder
+            }
+        }
+    };
+
+    gen.into()
+}

src/common/meta/Cargo.toml

@@ -7,6 +7,7 @@ license.workspace = true
 [features]
 testing = []
 pg_kvbackend = ["dep:tokio-postgres", "dep:backon", "dep:deadpool-postgres", "dep:deadpool"]
+mysql_kvbackend = ["dep:sqlx", "dep:backon"]
 
 [lints]
 workspace = true
@@ -57,9 +58,10 @@ serde_json.workspace = true
 serde_with.workspace = true
 session.workspace = true
 snafu.workspace = true
+sqlx = { workspace = true, optional = true }
 store-api.workspace = true
 strum.workspace = true
-table.workspace = true
+table = { workspace = true, features = ["testing"] }
 tokio.workspace = true
 tokio-postgres = { workspace = true, optional = true }
 tonic.workspace = true

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

@@ -192,6 +192,8 @@ mod tests {
                     expire_after: Some(300),
                     comment: "comment".to_string(),
                     options: Default::default(),
+                    created_time: chrono::Utc::now(),
+                    updated_time: chrono::Utc::now(),
                 },
                 (1..=3)
                     .map(|i| {

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>,
@@ -432,7 +425,14 @@ impl From<&CreateFlowData> for (FlowInfoValue, Vec<(FlowPartitionId, FlowRouteVa
         let flow_type = value.flow_type.unwrap_or_default().to_string();
         options.insert("flow_type".to_string(), flow_type);
 
-        let flow_info = FlowInfoValue {
+        let mut create_time = chrono::Utc::now();
+        if let Some(prev_flow_value) = value.prev_flow_info_value.as_ref()
+            && value.task.or_replace
+        {
+            create_time = prev_flow_value.get_inner_ref().created_time;
+        }
+
+        let flow_info: FlowInfoValue = FlowInfoValue {
             source_table_ids: value.source_table_ids.clone(),
             sink_table_name,
             flownode_ids,
@@ -442,6 +442,8 @@ impl From<&CreateFlowData> for (FlowInfoValue, Vec<(FlowPartitionId, FlowRouteVa
             expire_after,
             comment,
             options,
+            created_time: create_time,
+            updated_time: chrono::Utc::now(),
         };
 
         (flow_info, flow_routes)

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,