fix: show tables too slow cherry pick (#7244 )

v0.12.1 fix: `show tables` is too slow under large tables Signed-off-by: luofucong <luofc@foxmail.com>
ci: update github runner on old branch (#7249 )
2025-12-25 23:49:58 +00:00 · 2025-11-18 12:29:40 +00:00 · 2025-11-18 19:52:03 +08:00 · 2025-06-04 11:36:42 +08:00
190 changed files with 1328 additions and 6413 deletions
--- a/.github/CODEOWNERS
+++ b/.github/CODEOWNERS
@@ -4,7 +4,7 @@

 * @GreptimeTeam/db-approver

-## [Module] Database Engine
+## [Module] Databse Engine
 /src/index @zhongzc
 /src/mito2 @evenyag @v0y4g3r @waynexia
 /src/query @evenyag
--- a/.github/actions/build-dev-builder-images/action.yml
+++ b/.github/actions/build-dev-builder-images/action.yml
@@ -41,14 +41,7 @@ runs:
        username: ${{ inputs.dockerhub-image-registry-username }}
        password: ${{ inputs.dockerhub-image-registry-token }}

-    - 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.
+    - name: Build and push dev-builder-ubuntu image
      shell: bash
      if: ${{ inputs.build-dev-builder-ubuntu == 'true' }}
      run: |
@@ -59,7 +52,7 @@ runs:
          IMAGE_NAMESPACE=${{ inputs.dockerhub-image-namespace }} \
          DEV_BUILDER_IMAGE_TAG=${{ inputs.version }}

-    - name: Build and push dev-builder-centos image # Only build image for amd64 platform.
+    - name: Build and push dev-builder-centos image
      shell: bash
      if: ${{ inputs.build-dev-builder-centos == 'true' }}
      run: |
@@ -76,7 +69,8 @@ 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 }}
+          DEV_BUILDER_IMAGE_TAG=${{ inputs.version }} && \
+
+        docker push ${{ inputs.dockerhub-image-registry }}/${{ inputs.dockerhub-image-namespace }}/dev-builder-android:${{ inputs.version }}
--- a/.github/actions/build-linux-artifacts/action.yml
+++ b/.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
+        features: servers/dashboard,pg_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
+        features: servers/dashboard,pg_kvbackend
        cargo-profile: ${{ inputs.cargo-profile }}
        artifacts-dir: greptime-linux-${{ inputs.arch }}-centos-${{ inputs.version }}
        version: ${{ inputs.version }}
--- a/.github/actions/build-windows-artifacts/action.yml
+++ b/.github/actions/build-windows-artifacts/action.yml
@@ -47,6 +47,7 @@ runs:
      shell: pwsh
      run: make test sqlness-test
      env:
+        RUSTUP_WINDOWS_PATH_ADD_BIN: 1 # Workaround for https://github.com/nextest-rs/nextest/issues/1493
        RUST_BACKTRACE: 1
        SQLNESS_OPTS: "--preserve-state"

--- a/.github/actions/release-cn-artifacts/action.yaml
+++ b/.github/actions/release-cn-artifacts/action.yaml
@@ -64,11 +64,11 @@ inputs:
  upload-max-retry-times:
    description: Max retry times for uploading artifacts to S3
    required: false
-    default: "30"
+    default: "20"
  upload-retry-timeout:
    description: Timeout for uploading artifacts to S3
    required: false
-    default: "120" # minutes
+    default: "30" # minutes
 runs:
  using: composite
  steps:
--- a/.github/actions/setup-etcd-cluster/action.yml
+++ b/.github/actions/setup-etcd-cluster/action.yml
@@ -24,4 +24,9 @@ runs:
        --set auth.rbac.token.enabled=false \
        --set persistence.size=2Gi \
        --create-namespace \
+        --set global.security.allowInsecureImages=true \
+        --set image.registry=docker.io \
+        --set image.repository=greptime/etcd \
+        --set image.tag=3.6.1-debian-12-r3 \
+        --version 12.0.8 \
        -n ${{ inputs.namespace }}
--- a/.github/actions/setup-greptimedb-cluster/action.yml
+++ b/.github/actions/setup-greptimedb-cluster/action.yml
@@ -8,15 +8,15 @@ inputs:
    default: 2
    description: "Number of Datanode replicas"
  meta-replicas:
-    default: 2
+    default: 1
    description: "Number of Metasrv replicas"
-  image-registry:
+  image-registry: 
    default: "docker.io"
    description: "Image registry"
-  image-repository:
+  image-repository: 
    default: "greptime/greptimedb"
    description: "Image repository"
-  image-tag:
+  image-tag: 
    default: "latest"
    description: 'Image tag'
  etcd-endpoints:
@@ -32,12 +32,12 @@ runs:
  steps:
  - name: Install GreptimeDB operator
    uses: nick-fields/retry@v3
-    with:
+    with: 
      timeout_minutes: 3
      max_attempts: 3
      shell: bash
      command: |
-        helm repo add greptime https://greptimeteam.github.io/helm-charts/
+        helm repo add greptime https://greptimeteam.github.io/helm-charts/ 
        helm repo update
        helm upgrade \
          --install \
@@ -48,7 +48,7 @@ runs:
          --wait-for-jobs
  - name: Install GreptimeDB cluster
    shell: bash
-    run: |
+    run: | 
      helm upgrade \
        --install my-greptimedb \
        --set meta.backendStorage.etcd.endpoints=${{ inputs.etcd-endpoints }} \
@@ -59,7 +59,7 @@ runs:
        --set base.podTemplate.main.resources.requests.cpu=50m \
        --set base.podTemplate.main.resources.requests.memory=256Mi \
        --set base.podTemplate.main.resources.limits.cpu=2000m \
-        --set base.podTemplate.main.resources.limits.memory=3Gi \
+        --set base.podTemplate.main.resources.limits.memory=2Gi \
        --set frontend.replicas=${{ inputs.frontend-replicas }} \
        --set datanode.replicas=${{ inputs.datanode-replicas }} \
        --set meta.replicas=${{ inputs.meta-replicas }} \
@@ -72,7 +72,7 @@ runs:
  - name: Wait for GreptimeDB
    shell: bash
    run: |
-      while true; do
+      while true; do 
        PHASE=$(kubectl -n my-greptimedb get gtc my-greptimedb -o jsonpath='{.status.clusterPhase}')
        if [ "$PHASE" == "Running" ]; then
          echo "Cluster is ready"
@@ -86,10 +86,10 @@ runs:
  - name: Print GreptimeDB info
    if: always()
    shell: bash
-    run: |
+    run: | 
      kubectl get all --show-labels -n my-greptimedb
  - name: Describe Nodes
    if: always()
    shell: bash
-    run: |
+    run: | 
      kubectl describe nodes
--- a/.github/actions/setup-greptimedb-cluster/with-remote-wal.yaml
+++ b/.github/actions/setup-greptimedb-cluster/with-remote-wal.yaml
@@ -2,14 +2,13 @@ meta:
  configData: |-
    [runtime]
    global_rt_size = 4
-
+    
    [wal]
    provider = "kafka"
    broker_endpoints = ["kafka.kafka-cluster.svc.cluster.local:9092"]
    num_topics = 3
-    auto_prune_interval = "30s"
-    trigger_flush_threshold = 100

+        
    [datanode]
    [datanode.client]
    timeout = "120s"
@@ -22,7 +21,7 @@ datanode:
    [wal]
    provider = "kafka"
    broker_endpoints = ["kafka.kafka-cluster.svc.cluster.local:9092"]
-    overwrite_entry_start_id = true
+    linger = "2ms"
 frontend:
  configData: |-
    [runtime]
--- a/.github/actions/setup-kafka-cluster/action.yml
+++ b/.github/actions/setup-kafka-cluster/action.yml
@@ -23,4 +23,8 @@ runs:
        --set listeners.controller.protocol=PLAINTEXT \
        --set listeners.client.protocol=PLAINTEXT \
        --create-namespace \
+        --set image.registry=docker.io \
+        --set image.repository=greptime/kafka \
+        --set image.tag=3.9.0-debian-12-r1 \
+        --version 31.0.0 \
        -n ${{ inputs.namespace }}
--- a/.github/actions/start-runner/action.yml
+++ b/.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.3.8
+      uses: machulav/ec2-github-runner@v2
      id: start-linux-arm64-ec2-runner
      with:
        mode: start
--- a/.github/actions/stop-runner/action.yml
+++ b/.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.3.8
+      uses: machulav/ec2-github-runner@v2
      with:
        mode: stop
        label: ${{ inputs.label }}
--- a/.github/labeler.yaml
+++ b/.github/labeler.yaml
@@ -1,15 +0,0 @@
-ci:
-  - changed-files:
-      - any-glob-to-any-file: .github/**
-
-docker:
-  - changed-files:
-      - any-glob-to-any-file: docker/**
-
-documentation:
-  - changed-files:
-      - any-glob-to-any-file: docs/**
-
-dashboard:
-  - changed-files:
-      - any-glob-to-any-file: grafana/**
--- a/.github/scripts/check-version.sh
+++ b/.github/scripts/check-version.sh
@@ -1,42 +0,0 @@
-#!/bin/bash
-
-# Get current version
-CURRENT_VERSION=$1
-if [ -z "$CURRENT_VERSION" ]; then
-  echo "Error: Failed to get current version"
-  exit 1
-fi
-
-# Get the latest version from GitHub Releases
-API_RESPONSE=$(curl -s "https://api.github.com/repos/GreptimeTeam/greptimedb/releases/latest")
-
-if [ -z "$API_RESPONSE" ] || [ "$(echo "$API_RESPONSE" | jq -r '.message')" = "Not Found" ]; then
-  echo "Error: Failed to fetch latest version from GitHub"
-  exit 1
-fi
-
-# Get the latest version
-LATEST_VERSION=$(echo "$API_RESPONSE" | jq -r '.tag_name')
-
-if [ -z "$LATEST_VERSION" ] || [ "$LATEST_VERSION" = "null" ]; then
-  echo "Error: No valid version found in GitHub releases"
-  exit 1
-fi
-
-# Cleaned up version number format (removed possible 'v' prefix and -nightly suffix)
-CLEAN_CURRENT=$(echo "$CURRENT_VERSION" | sed 's/^v//' | sed 's/-nightly-.*//')
-CLEAN_LATEST=$(echo "$LATEST_VERSION" | sed 's/^v//' | sed 's/-nightly-.*//')
-
-echo "Current version: $CLEAN_CURRENT"
-echo "Latest release version: $CLEAN_LATEST"
-
-# Use sort -V to compare versions
-HIGHER_VERSION=$(printf "%s\n%s" "$CLEAN_CURRENT" "$CLEAN_LATEST" | sort -V | tail -n1)
-
-if [ "$HIGHER_VERSION" = "$CLEAN_CURRENT" ]; then
-  echo "Current version ($CLEAN_CURRENT) is NEWER than or EQUAL to latest ($CLEAN_LATEST)"
-  echo "should-push-latest-tag=true" >> $GITHUB_OUTPUT
-else
-  echo "Current version ($CLEAN_CURRENT) is OLDER than latest ($CLEAN_LATEST)"
-  echo "should-push-latest-tag=false" >> $GITHUB_OUTPUT
-fi
--- a/.github/scripts/create-version.sh
+++ b/.github/scripts/create-version.sh
@@ -8,25 +8,24 @@ set -e
 # - If it's a nightly build, the version is 'nightly-YYYYMMDD-$(git rev-parse --short HEAD)', like 'nightly-20230712-e5b243c'.
 # create_version ${GIHUB_EVENT_NAME} ${NEXT_RELEASE_VERSION} ${NIGHTLY_RELEASE_PREFIX}
 function create_version() {
-  # Read from environment variables.
+  # Read from envrionment variables.
  if [ -z "$GITHUB_EVENT_NAME" ]; then
-      echo "GITHUB_EVENT_NAME is empty" >&2
+      echo "GITHUB_EVENT_NAME is empty"
      exit 1
  fi

  if [ -z "$NEXT_RELEASE_VERSION" ]; then
-      echo "NEXT_RELEASE_VERSION is empty, use version from Cargo.toml" >&2
-      # NOTE: Need a `v` prefix for the version string.
-      export NEXT_RELEASE_VERSION=v$(grep '^version = ' Cargo.toml | cut -d '"' -f 2 | head -n 1)
+      echo "NEXT_RELEASE_VERSION is empty"
+      exit 1
  fi

  if [ -z "$NIGHTLY_RELEASE_PREFIX" ]; then
-      echo "NIGHTLY_RELEASE_PREFIX is empty" >&2
+      echo "NIGHTLY_RELEASE_PREFIX is empty"
      exit 1
  fi

  # Reuse $NEXT_RELEASE_VERSION to identify whether it's a nightly build.
-  # It will be like 'nightly-20230808-7d0d8dc6'.
+  # It will be like 'nigtly-20230808-7d0d8dc6'.
  if [ "$NEXT_RELEASE_VERSION" = nightly ]; then
    echo "$NIGHTLY_RELEASE_PREFIX-$(date "+%Y%m%d")-$(git rev-parse --short HEAD)"
    exit 0
@@ -36,7 +35,7 @@ function create_version() {
  # It will be like 'dev-2023080819-f0e7216c'.
  if [ "$NEXT_RELEASE_VERSION" = dev ]; then
    if [ -z "$COMMIT_SHA" ]; then
-      echo "COMMIT_SHA is empty in dev build" >&2
+      echo "COMMIT_SHA is empty in dev build"
      exit 1
    fi
    echo "dev-$(date "+%Y%m%d-%s")-$(echo "$COMMIT_SHA" | cut -c1-8)"
@@ -46,7 +45,7 @@ function create_version() {
  # Note: Only output 'version=xxx' to stdout when everything is ok, so that it can be used in GitHub Actions Outputs.
  if [ "$GITHUB_EVENT_NAME" = push ]; then
    if [ -z "$GITHUB_REF_NAME" ]; then
-      echo "GITHUB_REF_NAME is empty in push event" >&2
+      echo "GITHUB_REF_NAME is empty in push event"
      exit 1
    fi
    echo "$GITHUB_REF_NAME"
@@ -55,15 +54,15 @@ function create_version() {
  elif [ "$GITHUB_EVENT_NAME" = schedule ]; then
    echo "$NEXT_RELEASE_VERSION-$NIGHTLY_RELEASE_PREFIX-$(date "+%Y%m%d")"
  else
-    echo "Unsupported GITHUB_EVENT_NAME: $GITHUB_EVENT_NAME" >&2
+    echo "Unsupported GITHUB_EVENT_NAME: $GITHUB_EVENT_NAME"
    exit 1
  fi
 }

 # You can run as following examples:
-#  GITHUB_EVENT_NAME=push NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nightly GITHUB_REF_NAME=v0.3.0 ./create-version.sh
-#  GITHUB_EVENT_NAME=workflow_dispatch NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nightly ./create-version.sh
-#  GITHUB_EVENT_NAME=schedule NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nightly ./create-version.sh
-#  GITHUB_EVENT_NAME=schedule NEXT_RELEASE_VERSION=nightly NIGHTLY_RELEASE_PREFIX=nightly ./create-version.sh
-#  GITHUB_EVENT_NAME=workflow_dispatch COMMIT_SHA=f0e7216c4bb6acce9b29a21ec2d683be2e3f984a NEXT_RELEASE_VERSION=dev NIGHTLY_RELEASE_PREFIX=nightly ./create-version.sh
+#  GITHUB_EVENT_NAME=push NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nigtly GITHUB_REF_NAME=v0.3.0 ./create-version.sh
+#  GITHUB_EVENT_NAME=workflow_dispatch NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nigtly ./create-version.sh
+#  GITHUB_EVENT_NAME=schedule NEXT_RELEASE_VERSION=v0.4.0 NIGHTLY_RELEASE_PREFIX=nigtly ./create-version.sh
+#  GITHUB_EVENT_NAME=schedule NEXT_RELEASE_VERSION=nightly NIGHTLY_RELEASE_PREFIX=nigtly ./create-version.sh
+#  GITHUB_EVENT_NAME=workflow_dispatch COMMIT_SHA=f0e7216c4bb6acce9b29a21ec2d683be2e3f984a NEXT_RELEASE_VERSION=dev NIGHTLY_RELEASE_PREFIX=nigtly ./create-version.sh
 create_version
--- a/.github/scripts/deploy-greptimedb.sh
+++ b/.github/scripts/deploy-greptimedb.sh
@@ -10,7 +10,7 @@ GREPTIMEDB_IMAGE_TAG=${GREPTIMEDB_IMAGE_TAG:-latest}
 ETCD_CHART="oci://registry-1.docker.io/bitnamicharts/etcd"
 GREPTIME_CHART="https://greptimeteam.github.io/helm-charts/"

-# Create a cluster with 1 control-plane node and 5 workers.
+# Ceate a cluster with 1 control-plane node and 5 workers.
 function create_kind_cluster() {
  cat <<EOF | kind create cluster --name "${CLUSTER}" --image kindest/node:"$KUBERNETES_VERSION" --config=-
 kind: Cluster
@@ -68,7 +68,7 @@ function deploy_greptimedb_cluster() {

  helm install "$cluster_name" greptime/greptimedb-cluster \
    --set image.tag="$GREPTIMEDB_IMAGE_TAG" \
-    --set meta.backendStorage.etcd.endpoints="etcd.$install_namespace:2379" \
+    --set meta.etcdEndpoints="etcd.$install_namespace:2379" \
    -n "$install_namespace"

  # Wait for greptimedb cluster to be ready.
@@ -103,7 +103,7 @@ function deploy_greptimedb_cluster_with_s3_storage() {

  helm install "$cluster_name" greptime/greptimedb-cluster -n "$install_namespace" \
    --set image.tag="$GREPTIMEDB_IMAGE_TAG" \
-    --set meta.backendStorage.etcd.endpoints="etcd.$install_namespace:2379" \
+    --set meta.etcdEndpoints="etcd.$install_namespace:2379" \
    --set storage.s3.bucket="$AWS_CI_TEST_BUCKET" \
    --set storage.s3.region="$AWS_REGION" \
    --set storage.s3.root="$DATA_ROOT" \
--- a/.github/scripts/pull-test-deps-images.sh
+++ b/.github/scripts/pull-test-deps-images.sh
@@ -0,0 +1,34 @@
+#!/bin/bash
+
+# This script is used to pull the test dependency images that are stored in public ECR one by one to avoid rate limiting.
+
+set -e
+
+MAX_RETRIES=3
+
+IMAGES=(
+  "greptime/zookeeper:3.7"
+  "greptime/kafka:3.9.0-debian-12-r1"
+  "greptime/etcd:3.6.1-debian-12-r3"
+  "greptime/minio:2024"
+  "greptime/mysql:5.7"
+)
+
+for image in "${IMAGES[@]}"; do
+  for ((attempt=1; attempt<=MAX_RETRIES; attempt++)); do
+    if docker pull "$image"; then
+      # Successfully pulled the image.
+      break
+    else
+      # Use some simple exponential backoff to avoid rate limiting.
+      if [ $attempt -lt $MAX_RETRIES ]; then
+        sleep_seconds=$((attempt * 5))
+        echo "Attempt $attempt failed for $image, waiting $sleep_seconds seconds"
+        sleep $sleep_seconds  # 5s, 10s delays
+      else
+        echo "Failed to pull $image after $MAX_RETRIES attempts"
+        exit 1
+      fi
+    fi
+  done
+done
--- a/.github/scripts/update-dev-builder-version.sh
+++ b/.github/scripts/update-dev-builder-version.sh
@@ -1,37 +0,0 @@
-#!/bin/bash
-
-DEV_BUILDER_IMAGE_TAG=$1
-
-update_dev_builder_version() {
-  if [ -z "$DEV_BUILDER_IMAGE_TAG" ]; then
-    echo "Error: Should specify the dev-builder image tag"
-    exit 1
-  fi
-
-  # Configure Git configs.
-  git config --global user.email greptimedb-ci@greptime.com
-  git config --global user.name greptimedb-ci
-
-  # Checkout a new branch.
-  BRANCH_NAME="ci/update-dev-builder-$(date +%Y%m%d%H%M%S)"
-  git checkout -b $BRANCH_NAME
-
-  # Update the dev-builder image tag in the Makefile.
-  sed -i "s/DEV_BUILDER_IMAGE_TAG ?=.*/DEV_BUILDER_IMAGE_TAG ?= ${DEV_BUILDER_IMAGE_TAG}/g" Makefile
-
-  # Commit the changes.
-  git add Makefile
-  git commit -m "ci: update dev-builder image tag"
-  git push origin $BRANCH_NAME
-
-  # Create a Pull Request.
-  gh pr create \
-      --title "ci: update dev-builder image tag" \
-      --body "This PR updates the dev-builder image tag" \
-      --base main \
-      --head $BRANCH_NAME \
-      --reviewer zyy17 \
-      --reviewer daviderli614
-}
-
-update_dev_builder_version
--- a/.github/scripts/update-helm-charts-version.sh
+++ b/.github/scripts/update-helm-charts-version.sh
@@ -1,46 +0,0 @@
-#!/bin/bash
-
-set -e
-
-VERSION=${VERSION}
-GITHUB_TOKEN=${GITHUB_TOKEN}
-
-update_helm_charts_version() {
-  # Configure Git configs.
-  git config --global user.email update-helm-charts-version@greptime.com
-  git config --global user.name update-helm-charts-version
-
-  # Clone helm-charts repository.
-  git clone "https://x-access-token:${GITHUB_TOKEN}@github.com/GreptimeTeam/helm-charts.git"
-  cd helm-charts
-
-  # Set default remote for gh CLI
-  gh repo set-default GreptimeTeam/helm-charts
-
-  # Checkout a new branch.
-  BRANCH_NAME="chore/greptimedb-${VERSION}"
-  git checkout -b $BRANCH_NAME
-
-  # Update version.
-  make update-version CHART=greptimedb-cluster VERSION=${VERSION}
-  make update-version CHART=greptimedb-standalone VERSION=${VERSION}
-
-  # Update docs.
-  make docs
-
-  # Commit the changes.
-  git add .
-  git commit -s -m "chore: Update GreptimeDB version to ${VERSION}"
-  git push origin $BRANCH_NAME
-
-  # Create a Pull Request.
-  gh pr create \
-    --title "chore: Update GreptimeDB version to ${VERSION}" \
-    --body "This PR updates the GreptimeDB version." \
-    --base main \
-    --head $BRANCH_NAME \
-    --reviewer zyy17 \
-    --reviewer daviderli614
-}
-
-update_helm_charts_version
--- a/.github/scripts/update-homebrew-greptme-version.sh
+++ b/.github/scripts/update-homebrew-greptme-version.sh
@@ -1,42 +0,0 @@
-#!/bin/bash
-
-set -e
-
-VERSION=${VERSION}
-GITHUB_TOKEN=${GITHUB_TOKEN}
-
-update_homebrew_greptime_version() {
-  # Configure Git configs.
-  git config --global user.email update-greptime-version@greptime.com
-  git config --global user.name update-greptime-version
-
-  # Clone helm-charts repository.
-  git clone "https://x-access-token:${GITHUB_TOKEN}@github.com/GreptimeTeam/homebrew-greptime.git"
-  cd homebrew-greptime
-
-  # Set default remote for gh CLI
-  gh repo set-default GreptimeTeam/homebrew-greptime
-
-  # Checkout a new branch.
-  BRANCH_NAME="chore/greptimedb-${VERSION}"
-  git checkout -b $BRANCH_NAME
-
-  # Update version.
-  make update-greptime-version VERSION=${VERSION}
-
-  # Commit the changes.
-  git add .
-  git commit -s -m "chore: Update GreptimeDB version to ${VERSION}"
-  git push origin $BRANCH_NAME
-
-  # Create a Pull Request.
-  gh pr create \
-    --title "chore: Update GreptimeDB version to ${VERSION}" \
-    --body "This PR updates the GreptimeDB version." \
-    --base main \
-    --head $BRANCH_NAME \
-    --reviewer zyy17 \
-    --reviewer daviderli614
-}
-
-update_homebrew_greptime_version
--- a/.github/scripts/upload-artifacts-to-s3.sh
+++ b/.github/scripts/upload-artifacts-to-s3.sh
@@ -41,7 +41,7 @@ function upload_artifacts() {
 # Updates the latest version information in AWS S3 if UPDATE_VERSION_INFO is true.
 function update_version_info() {
  if [ "$UPDATE_VERSION_INFO" == "true" ]; then
-    # If it's the official release(like v1.0.0, v1.0.1, v1.0.2, etc.), update latest-version.txt.
+    # If it's the officail release(like v1.0.0, v1.0.1, v1.0.2, etc.), update latest-version.txt.
    if [[ "$VERSION" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
      echo "Updating latest-version.txt"
      echo "$VERSION" > latest-version.txt
--- a/.github/workflows/apidoc.yml
+++ b/.github/workflows/apidoc.yml
@@ -14,7 +14,7 @@ name: Build API docs

 jobs:
  apidoc:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
    - uses: actions/checkout@v4
      with:
--- a/.github/workflows/dev-build.yml
+++ b/.github/workflows/dev-build.yml
@@ -55,11 +55,6 @@ on:
        description: Build and push images to DockerHub and ACR
        required: false
        default: true
-      upload_artifacts_to_s3:
-        type: boolean
-        description: Whether upload artifacts to s3
-        required: false
-        default: false
      cargo_profile:
        type: choice
        description: The cargo profile to use in building GreptimeDB.
@@ -88,7 +83,7 @@ jobs:
  allocate-runners:
    name: Allocate runners
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    outputs:
      linux-amd64-runner: ${{ steps.start-linux-amd64-runner.outputs.label }}
      linux-arm64-runner: ${{ steps.start-linux-arm64-runner.outputs.label }}
@@ -223,7 +218,7 @@ jobs:
      build-linux-amd64-artifacts,
      build-linux-arm64-artifacts,
    ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    outputs:
      build-result: ${{ steps.set-build-result.outputs.build-result }}
    steps:
@@ -244,13 +239,6 @@ jobs:
          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
        run: |
@@ -263,7 +251,7 @@ jobs:
      allocate-runners,
      release-images-to-dockerhub,
    ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    continue-on-error: true
    steps:
      - uses: actions/checkout@v4
@@ -286,7 +274,7 @@ jobs:
          aws-cn-access-key-id: ${{ secrets.AWS_CN_ACCESS_KEY_ID }}
          aws-cn-secret-access-key: ${{ secrets.AWS_CN_SECRET_ACCESS_KEY }}
          aws-cn-region: ${{ vars.AWS_RELEASE_BUCKET_REGION }}
-          upload-to-s3: ${{ inputs.upload_artifacts_to_s3 }}
+          upload-to-s3: false
          dev-mode: true                     # Only build the standard images(exclude centos images).
          push-latest-tag: false             # Don't push the latest tag to registry.
          update-version-info: false         # Don't update the version info in S3.
@@ -295,7 +283,7 @@ jobs:
    name: Stop linux-amd64 runner
    # Only run this job when the runner is allocated.
    if: ${{ always() }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    needs: [
      allocate-runners,
      build-linux-amd64-artifacts,
@@ -321,7 +309,7 @@ jobs:
    name: Stop linux-arm64 runner
    # Only run this job when the runner is allocated.
    if: ${{ always() }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    needs: [
      allocate-runners,
      build-linux-arm64-artifacts,
@@ -349,7 +337,7 @@ jobs:
    needs: [
      release-images-to-dockerhub
    ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    permissions:
      issues: write

--- a/.github/workflows/develop.yml
+++ b/.github/workflows/develop.yml
@@ -21,15 +21,13 @@ concurrency:
  cancel-in-progress: true

 jobs:
-  check-typos-and-docs:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    name: Check typos and docs
-    runs-on: ubuntu-latest
+  check-docs:
+    name: Check docs
+    runs-on: ubuntu-22.04
    steps:
      - uses: actions/checkout@v4
        with:
          persist-credentials: false
-      - uses: crate-ci/typos@master
      - name: Check the config docs
        run: |
          make config-docs && \
@@ -37,8 +35,7 @@ jobs:
          || (echo "'config/config.md' is not up-to-date, please run 'make config-docs'." && exit 1)

  license-header-check:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    name: Check License Header
    steps:
      - uses: actions/checkout@v4
@@ -47,12 +44,11 @@ jobs:
      - uses: korandoru/hawkeye@v5

  check:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Check
    runs-on: ${{ matrix.os }}
    strategy:
      matrix:
-        os: [ ubuntu-latest ]
+        os: [ ubuntu-22.04 ]
    timeout-minutes: 60
    steps:
      - uses: actions/checkout@v4
@@ -74,9 +70,8 @@ jobs:
        run: cargo check --locked --workspace --all-targets

  toml:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Toml Check
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    timeout-minutes: 60
    steps:
      - uses: actions/checkout@v4
@@ -89,12 +84,11 @@ jobs:
        run: taplo format --check

  build:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Build GreptimeDB binaries
    runs-on: ${{ matrix.os }}
    strategy:
      matrix:
-        os: [ ubuntu-latest ]
+        os: [ ubuntu-22.04 ]
    timeout-minutes: 60
    steps:
      - uses: actions/checkout@v4
@@ -132,7 +126,6 @@ jobs:
          version: current

  fuzztest:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Fuzz Test
    needs: build
    runs-on: ubuntu-latest
@@ -189,13 +182,11 @@ jobs:
          max-total-time: 120

  unstable-fuzztest:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Unstable Fuzz Test
    needs: build-greptime-ci
    runs-on: ubuntu-latest
    timeout-minutes: 60
    strategy:
-      fail-fast: false
      matrix:
        target: [ "unstable_fuzz_create_table_standalone" ]
    steps:
@@ -223,12 +214,12 @@ jobs:
        run: |
          sudo apt update && sudo apt install -y libfuzzer-14-dev
          cargo install cargo-fuzz cargo-gc-bin --force
-      - name: Download pre-built binary
+      - name: Download pre-built binariy
        uses: actions/download-artifact@v4
        with:
          name: bin
          path: .
-      - name: Unzip binary
+      - name: Unzip bianry
        run: |
          tar -xvf ./bin.tar.gz
          rm ./bin.tar.gz
@@ -250,19 +241,13 @@ jobs:
          name: unstable-fuzz-logs
          path: /tmp/unstable-greptime/
          retention-days: 3
-      - name: Describe pods
-        if: failure()
-        shell: bash
-        run: |
-          kubectl describe pod -n my-greptimedb

  build-greptime-ci:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Build GreptimeDB binary (profile-CI)
    runs-on: ${{ matrix.os }}
    strategy:
      matrix:
-        os: [ ubuntu-latest ]
+        os: [ ubuntu-22.04 ]
    timeout-minutes: 60
    steps:
      - uses: actions/checkout@v4
@@ -281,7 +266,7 @@ jobs:
      - name: Install cargo-gc-bin
        shell: bash
        run: cargo install cargo-gc-bin --force
-      - name: Build greptime binary
+      - 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
@@ -299,13 +284,11 @@ jobs:
          version: current

  distributed-fuzztest:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Fuzz Test (Distributed, ${{ matrix.mode.name }}, ${{ matrix.target }})
    runs-on: ubuntu-latest
    needs:  build-greptime-ci
    timeout-minutes: 60
    strategy:
-      fail-fast: false
      matrix:
        target: [ "fuzz_create_table", "fuzz_alter_table", "fuzz_create_database", "fuzz_create_logical_table", "fuzz_alter_logical_table", "fuzz_insert", "fuzz_insert_logical_table" ]
        mode:
@@ -335,9 +318,9 @@ jobs:
        name: Setup Minio
        uses: ./.github/actions/setup-minio
      - if: matrix.mode.kafka
-        name: Setup Kafka cluster
+        name: Setup Kafka cluser
        uses: ./.github/actions/setup-kafka-cluster
-      - name: Setup Etcd cluster
+      - name: Setup Etcd cluser
        uses: ./.github/actions/setup-etcd-cluster
      # Prepares for fuzz tests
      - uses: arduino/setup-protoc@v3
@@ -410,11 +393,6 @@ jobs:
        shell: bash
        run: |
          kubectl describe nodes
-      - name: Describe pod
-        if: failure()
-        shell: bash
-        run: |
-          kubectl describe pod -n my-greptimedb
      - name: Export kind logs
        if: failure()
        shell: bash
@@ -437,13 +415,11 @@ jobs:
          docker system prune -f

  distributed-fuzztest-with-chaos:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Fuzz Test with Chaos (Distributed, ${{ matrix.mode.name }}, ${{ matrix.target }})
    runs-on: ubuntu-latest
    needs:  build-greptime-ci
    timeout-minutes: 60
    strategy:
-      fail-fast: false
      matrix:
        target: ["fuzz_migrate_mito_regions", "fuzz_migrate_metric_regions", "fuzz_failover_mito_regions", "fuzz_failover_metric_regions"]
        mode:
@@ -488,9 +464,9 @@ jobs:
        name: Setup Minio
        uses: ./.github/actions/setup-minio
      - if: matrix.mode.kafka
-        name: Setup Kafka cluster
+        name: Setup Kafka cluser
        uses: ./.github/actions/setup-kafka-cluster
-      - name: Setup Etcd cluster
+      - name: Setup Etcd cluser
        uses: ./.github/actions/setup-etcd-cluster
      # Prepares for fuzz tests
      - uses: arduino/setup-protoc@v3
@@ -564,11 +540,6 @@ jobs:
        shell: bash
        run: |
          kubectl describe nodes
-      - name: Describe pods
-        if: failure()
-        shell: bash
-        run: |
-          kubectl describe pod -n my-greptimedb
      - name: Export kind logs
        if: failure()
        shell: bash
@@ -591,14 +562,12 @@ jobs:
          docker system prune -f

  sqlness:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Sqlness Test (${{ matrix.mode.name }})
    needs: build
    runs-on: ${{ matrix.os }}
    strategy:
-      fail-fast: false
      matrix:
-        os: [ ubuntu-latest ]
+        os: [ ubuntu-22.04 ]
        mode:
          - name: "Basic"
            opts: ""
@@ -606,7 +575,7 @@ jobs:
          - name: "Remote WAL"
            opts: "-w kafka -k 127.0.0.1:9092"
            kafka: true
-          - name: "PostgreSQL KvBackend"
+          - name: "Pg Kvbackend"
            opts: "--setup-pg"
            kafka: false
    timeout-minutes: 60
@@ -617,7 +586,8 @@ jobs:
      - if: matrix.mode.kafka
        name: Setup kafka server
        working-directory: tests-integration/fixtures
-        run: docker compose up -d --wait kafka
+        run:  ../../.github/scripts/pull-test-deps-images.sh && docker compose up -d --wait kafka
+
      - name: Download pre-built binaries
        uses: actions/download-artifact@v4
        with:
@@ -636,9 +606,8 @@ jobs:
          retention-days: 3

  fmt:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Rustfmt
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    timeout-minutes: 60
    steps:
      - uses: actions/checkout@v4
@@ -654,9 +623,8 @@ jobs:
        run: make fmt-check

  clippy:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Clippy
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    timeout-minutes: 60
    steps:
      - uses: actions/checkout@v4
@@ -680,7 +648,6 @@ jobs:
        run: make clippy

  conflict-check:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
    name: Check for conflict
    runs-on: ubuntu-latest
    steps:
@@ -691,7 +658,7 @@ jobs:
        uses: olivernybroe/action-conflict-finder@v4.0

  test:
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' && github.event_name != 'merge_group' }}
+    if: github.event_name != 'merge_group'
    runs-on: ubuntu-22.04-arm
    timeout-minutes: 60
    needs:  [conflict-check, clippy, fmt]
@@ -706,7 +673,7 @@ jobs:
      - name: Install toolchain
        uses: actions-rust-lang/setup-rust-toolchain@v1
        with:
-          cache: false
+            cache: false
      - name: Rust Cache
        uses: Swatinem/rust-cache@v2
        with:
@@ -718,7 +685,8 @@ jobs:
        uses: taiki-e/install-action@nextest
      - name: Setup external services
        working-directory: tests-integration/fixtures
-        run: docker compose up -d --wait
+        run: ../../.github/scripts/pull-test-deps-images.sh && docker compose up -d --wait
+
      - name: Run nextest cases
        run: cargo nextest run --workspace -F dashboard -F pg_kvbackend
        env:
@@ -737,13 +705,12 @@ 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.repository == 'GreptimeTeam/greptimedb' && github.event_name == 'merge_group' }}
+    if: github.event_name == 'merge_group'
    runs-on: ubuntu-22.04-8-cores
    timeout-minutes: 60
    steps:
@@ -771,7 +738,8 @@ jobs:
        uses: taiki-e/install-action@cargo-llvm-cov
      - name: Setup external services
        working-directory: tests-integration/fixtures
-        run: docker compose up -d --wait
+        run: ../../.github/scripts/pull-test-deps-images.sh && 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
        env:
@@ -789,7 +757,6 @@ 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"
@@ -803,7 +770,6 @@ jobs:
          verbose: true

  # compat:
-  #   if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
  #   name: Compatibility Test
  #   needs: build
  #   runs-on: ubuntu-22.04
--- a/.github/workflows/docbot.yml
+++ b/.github/workflows/docbot.yml
@@ -3,13 +3,9 @@ on:
  pull_request_target:
    types: [opened, edited]

-concurrency:
-  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
-  cancel-in-progress: true
-
 jobs:
  docbot:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    permissions:
      pull-requests: write
      contents: read
--- a/.github/workflows/docs.yml
+++ b/.github/workflows/docs.yml
@@ -31,7 +31,7 @@ name: CI
 jobs:
  typos:
    name: Spell Check with Typos
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
      - uses: actions/checkout@v4
        with:
@@ -39,7 +39,7 @@ jobs:
      - uses: crate-ci/typos@master

  license-header-check:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    name: Check License Header
    steps:
      - uses: actions/checkout@v4
@@ -49,29 +49,29 @@ jobs:

  check:
    name: Check
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
      - run: 'echo "No action required"'

  fmt:
    name: Rustfmt
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
      - run: 'echo "No action required"'

  clippy:
    name: Clippy
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
      - run: 'echo "No action required"'

  coverage:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
      - run: 'echo "No action required"'

  test:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
      - run: 'echo "No action required"'

@@ -80,7 +80,7 @@ jobs:
    runs-on: ${{ matrix.os }}
    strategy:
      matrix:
-        os: [ ubuntu-latest ]
+        os: [ ubuntu-22.04 ]
        mode:
          - name: "Basic"
          - name: "Remote WAL"
--- a/.github/workflows/nightly-build.yml
+++ b/.github/workflows/nightly-build.yml
@@ -70,7 +70,7 @@ jobs:
  allocate-runners:
    name: Allocate runners
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    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-latest
+    runs-on: ubuntu-22.04
    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-latest
+    runs-on: ubuntu-22.04
    # 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-latest
+    runs-on: ubuntu-22.04
    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-latest
+    runs-on: ubuntu-22.04
    needs: [
      allocate-runners,
      build-linux-arm64-artifacts,
@@ -303,7 +303,7 @@ jobs:
    needs: [
      release-images-to-dockerhub
    ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    permissions:
      issues: write
    env:
--- a/.github/workflows/nightly-ci.yml
+++ b/.github/workflows/nightly-ci.yml
@@ -13,7 +13,7 @@ jobs:
  sqlness-test:
    name: Run sqlness test
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
      - name: Checkout
        uses: actions/checkout@v4
@@ -107,6 +107,7 @@ jobs:
          CARGO_BUILD_RUSTFLAGS: "-C linker=lld-link"
          RUST_BACKTRACE: 1
          CARGO_INCREMENTAL: 0
+          RUSTUP_WINDOWS_PATH_ADD_BIN: 1 # Workaround for https://github.com/nextest-rs/nextest/issues/1493
          GT_S3_BUCKET: ${{ vars.AWS_CI_TEST_BUCKET }}
          GT_S3_ACCESS_KEY_ID: ${{ secrets.AWS_CI_TEST_ACCESS_KEY_ID }}
          GT_S3_ACCESS_KEY: ${{ secrets.AWS_CI_TEST_SECRET_ACCESS_KEY }}
@@ -117,22 +118,22 @@ jobs:
    name: Run clean build on Linux
    runs-on: ubuntu-latest
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    timeout-minutes: 45
+    timeout-minutes: 60
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0
          persist-credentials: false
-      - uses: cachix/install-nix-action@v31
-      - run: nix develop --command cargo check --bin greptime
-        env:
-          CARGO_BUILD_RUSTFLAGS: "-C link-arg=-fuse-ld=mold"
+      - uses: cachix/install-nix-action@v27
+        with:
+          nix_path: nixpkgs=channel:nixos-24.11
+      - run: nix develop --command cargo build

  check-status:
    name: Check status
    needs: [sqlness-test, sqlness-windows, test-on-windows]
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    outputs:
      check-result: ${{ steps.set-check-result.outputs.check-result }}
    steps:
@@ -145,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-latest
+    runs-on: ubuntu-22.04
    env:
      SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL_DEVELOP_CHANNEL }}
    steps:
--- a/.github/workflows/pr-labeling.yaml
+++ b/.github/workflows/pr-labeling.yaml
@@ -1,42 +0,0 @@
-name: 'PR Labeling'
-
-on:
-  pull_request_target:
-    types:
-      - opened
-      - synchronize
-      - reopened
-
-permissions:
-  contents: read
-  pull-requests: write
-  issues: write
-
-jobs:
-  labeler:
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout sources
-        uses: actions/checkout@v4
-
-      - uses: actions/labeler@v5
-        with:
-          configuration-path: ".github/labeler.yaml"
-          repo-token: "${{ secrets.GITHUB_TOKEN }}"
-
-  size-label:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: pascalgn/size-label-action@v0.5.5
-        env:
-          GITHUB_TOKEN: "${{ secrets.GITHUB_TOKEN }}"
-        with:
-          sizes: >
-            {
-              "0": "XS",
-              "100": "S",
-              "300": "M",
-              "1000": "L",
-              "1500": "XL",
-              "2000": "XXL"
-            }
--- a/.github/workflows/release-dev-builder-images.yaml
+++ b/.github/workflows/release-dev-builder-images.yaml
@@ -24,20 +24,12 @@ on:
        description: Release dev-builder-android image
        required: false
        default: false
-      update_dev_builder_image_tag:
-        type: boolean
-        description: Update the DEV_BUILDER_IMAGE_TAG in Makefile and create a PR
-        required: false
-        default: false

 jobs:
  release-dev-builder-images:
    name: Release dev builder images
-    # The jobs are triggered by the following events:
-    # 1. Manually triggered workflow_dispatch event
-    # 2. Push event when the PR that modifies the `rust-toolchain.toml` or `docker/dev-builder/**` is merged to main
-    if: ${{ github.event_name == 'push' || inputs.release_dev_builder_ubuntu_image || inputs.release_dev_builder_centos_image || inputs.release_dev_builder_android_image }}
-    runs-on: ubuntu-latest
+    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
    outputs:
      version: ${{ steps.set-version.outputs.version }}
    steps:
@@ -65,13 +57,13 @@ jobs:
          version: ${{ env.VERSION }}
          dockerhub-image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
          dockerhub-image-registry-token: ${{ secrets.DOCKERHUB_TOKEN }}
-          build-dev-builder-ubuntu: ${{ inputs.release_dev_builder_ubuntu_image || github.event_name == 'push' }}
-          build-dev-builder-centos: ${{ inputs.release_dev_builder_centos_image || github.event_name == 'push' }}
-          build-dev-builder-android: ${{ inputs.release_dev_builder_android_image || github.event_name == 'push' }}
+          build-dev-builder-ubuntu: ${{ inputs.release_dev_builder_ubuntu_image }}
+          build-dev-builder-centos: ${{ inputs.release_dev_builder_centos_image }}
+          build-dev-builder-android: ${{ inputs.release_dev_builder_android_image }}

  release-dev-builder-images-ecr:
    name: Release dev builder images to AWS ECR
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    needs: [
      release-dev-builder-images
    ]
@@ -93,7 +85,7 @@ jobs:

      - name: Push dev-builder-ubuntu image
        shell: bash
-        if: ${{ inputs.release_dev_builder_ubuntu_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_ubuntu_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -114,7 +106,7 @@ jobs:

      - name: Push dev-builder-centos image
        shell: bash
-        if: ${{ inputs.release_dev_builder_centos_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_centos_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -135,7 +127,7 @@ jobs:

      - name: Push dev-builder-android image
        shell: bash
-        if: ${{ inputs.release_dev_builder_android_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_android_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -156,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-latest
+    runs-on: ubuntu-22.04
    needs: [
      release-dev-builder-images
    ]
@@ -170,7 +162,7 @@ jobs:

      - name: Push dev-builder-ubuntu image
        shell: bash
-        if: ${{ inputs.release_dev_builder_ubuntu_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_ubuntu_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -184,7 +176,7 @@ jobs:

      - name: Push dev-builder-centos image
        shell: bash
-        if: ${{ inputs.release_dev_builder_centos_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_centos_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -198,7 +190,7 @@ jobs:

      - name: Push dev-builder-android image
        shell: bash
-        if: ${{ inputs.release_dev_builder_android_image || github.event_name == 'push' }}
+        if: ${{ inputs.release_dev_builder_android_image }}
        env:
          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
@@ -209,24 +201,3 @@ jobs:
            quay.io/skopeo/stable:latest \
            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-android:$IMAGE_VERSION \
            docker://$ACR_IMAGE_REGISTRY/$IMAGE_NAMESPACE/dev-builder-android:$IMAGE_VERSION
-  
-  update-dev-builder-image-tag:
-    name: Update dev-builder image tag
-    runs-on: ubuntu-latest
-    permissions:
-      contents: write
-      pull-requests: write
-    if: ${{ github.event_name == 'push' || inputs.update_dev_builder_image_tag }}
-    needs: [
-      release-dev-builder-images
-    ]
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-
-      - name: Update dev-builder image tag
-        shell: bash
-        env:
-          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-        run: |
-          ./.github/scripts/update-dev-builder-version.sh ${{ needs.release-dev-builder-images.outputs.version }}
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -88,14 +88,16 @@ env:
  # Controls whether to run tests, include unit-test, integration-test and sqlness.
  DISABLE_RUN_TESTS: ${{ inputs.skip_test || vars.DEFAULT_SKIP_TEST }}

-  # The scheduled version is '${{ env.NEXT_RELEASE_VERSION }}-nightly-YYYYMMDD', like v0.2.0-nightly-20230313;
+  # The scheduled version is '${{ env.NEXT_RELEASE_VERSION }}-nightly-YYYYMMDD', like v0.2.0-nigthly-20230313;
  NIGHTLY_RELEASE_PREFIX: nightly
+  # Note: The NEXT_RELEASE_VERSION should be modified manually by every formal release.
+  NEXT_RELEASE_VERSION: v0.12.0

 jobs:
  allocate-runners:
    name: Allocate runners
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    outputs:
      linux-amd64-runner: ${{ steps.start-linux-amd64-runner.outputs.label }}
      linux-arm64-runner: ${{ steps.start-linux-arm64-runner.outputs.label }}
@@ -110,8 +112,6 @@ jobs:

      # The 'version' use as the global tag name of the release workflow.
      version: ${{ steps.create-version.outputs.version }}
-
-      should-push-latest-tag: ${{ steps.check-version.outputs.should-push-latest-tag }}
    steps:
      - name: Checkout
        uses: actions/checkout@v4
@@ -126,7 +126,7 @@ jobs:

      # The create-version will create a global variable named 'version' in the global workflows.
      # - If it's a tag push release, the version is the tag name(${{ github.ref_name }});
-      # - If it's a scheduled release, the version is '${{ env.NEXT_RELEASE_VERSION }}-nightly-$buildTime', like v0.2.0-nightly-20230313;
+      # - If it's a scheduled release, the version is '${{ env.NEXT_RELEASE_VERSION }}-nightly-$buildTime', like v0.2.0-nigthly-20230313;
      # - If it's a manual release, the version is '${{ env.NEXT_RELEASE_VERSION }}-<short-git-sha>-YYYYMMDDSS', like v0.2.0-e5b243c-2023071245;
      - name: Create version
        id: create-version
@@ -135,13 +135,9 @@ jobs:
        env:
          GITHUB_EVENT_NAME: ${{ github.event_name }}
          GITHUB_REF_NAME: ${{ github.ref_name }}
+          NEXT_RELEASE_VERSION: ${{ env.NEXT_RELEASE_VERSION }}
          NIGHTLY_RELEASE_PREFIX: ${{ env.NIGHTLY_RELEASE_PREFIX }}

-      - name: Check version
-        id: check-version
-        run: |
-          ./.github/scripts/check-version.sh "${{ steps.create-version.outputs.version }}"
-
      - name: Allocate linux-amd64 runner
        if: ${{ inputs.build_linux_amd64_artifacts || github.event_name == 'push' || github.event_name == 'schedule' }}
        uses: ./.github/actions/start-runner
@@ -303,7 +299,7 @@ jobs:
      build-linux-amd64-artifacts,
      build-linux-arm64-artifacts,
    ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-2004-16-cores
    outputs:
      build-image-result: ${{ steps.set-build-image-result.outputs.build-image-result }}
    steps:
@@ -321,7 +317,7 @@ jobs:
          image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
          image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
          version: ${{ needs.allocate-runners.outputs.version }}
-          push-latest-tag: ${{ needs.allocate-runners.outputs.should-push-latest-tag == 'true' && github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
+          push-latest-tag: true

      - name: Set build image result
        id: set-build-image-result
@@ -339,7 +335,7 @@ jobs:
      build-windows-artifacts,
      release-images-to-dockerhub,
    ]
-    runs-on: ubuntu-latest-16-cores
+    runs-on: ubuntu-22.04
    # 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.
@@ -368,7 +364,7 @@ jobs:
          dev-mode: false
          upload-to-s3: true
          update-version-info: true
-          push-latest-tag: ${{ needs.allocate-runners.outputs.should-push-latest-tag == 'true' && github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
+          push-latest-tag: true

  publish-github-release:
    name: Create GitHub release and upload artifacts
@@ -381,7 +377,7 @@ jobs:
      build-windows-artifacts,
      release-images-to-dockerhub,
    ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    steps:
      - uses: actions/checkout@v4
        with:
@@ -395,12 +391,12 @@ jobs:

  ### Stop runners ###
  # It's very necessary to split the job of releasing runners into 'stop-linux-amd64-runner' and 'stop-linux-arm64-runner'.
-  # Because we can terminate the specified EC2 instance immediately after the job is finished without unnecessary waiting.
+  # Because we can terminate the specified EC2 instance immediately after the job is finished without uncessary waiting.
  stop-linux-amd64-runner: # It's always run as the last job in the workflow to make sure that the runner is released.
    name: Stop linux-amd64 runner
    # Only run this job when the runner is allocated.
    if: ${{ always() }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    needs: [
      allocate-runners,
      build-linux-amd64-artifacts,
@@ -426,7 +422,7 @@ jobs:
    name: Stop linux-arm64 runner
    # Only run this job when the runner is allocated.
    if: ${{ always() }}
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    needs: [
      allocate-runners,
      build-linux-arm64-artifacts,
@@ -448,11 +444,11 @@ jobs:
          aws-region: ${{ vars.EC2_RUNNER_REGION }}
          github-token: ${{ secrets.GH_PERSONAL_ACCESS_TOKEN }}

-  bump-downstream-repo-versions:
-    name: Bump downstream repo versions
+  bump-doc-version:
+    name: Bump doc version
    if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
-    needs: [allocate-runners, publish-github-release]
-    runs-on: ubuntu-latest
+    needs: [allocate-runners]
+    runs-on: ubuntu-22.04
    # 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.
@@ -463,58 +459,13 @@ jobs:
          fetch-depth: 0
          persist-credentials: false
      - uses: ./.github/actions/setup-cyborg
-      - name: Bump downstream repo versions
+      - name: Bump doc version
        working-directory: cyborg
-        run: pnpm tsx bin/bump-versions.ts
+        run: pnpm tsx bin/bump-doc-version.ts
        env:
-          TARGET_REPOS: website,docs,demo
          VERSION: ${{ needs.allocate-runners.outputs.version }}
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-          WEBSITE_REPO_TOKEN: ${{ secrets.WEBSITE_REPO_TOKEN }}
          DOCS_REPO_TOKEN: ${{ secrets.DOCS_REPO_TOKEN }}
-          DEMO_REPO_TOKEN: ${{ secrets.DEMO_REPO_TOKEN }}
-
-  bump-helm-charts-version:
-    name: Bump helm charts version
-    if: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
-    needs: [allocate-runners, publish-github-release]
-    runs-on: ubuntu-latest
-    permissions:
-      contents: write
-      pull-requests: write
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-
-      - name: Bump helm charts version
-        env:
-          GITHUB_TOKEN: ${{ secrets.HELM_CHARTS_REPO_TOKEN }}
-          VERSION: ${{ needs.allocate-runners.outputs.version }}
-        run: |
-          ./.github/scripts/update-helm-charts-version.sh
-
-  bump-homebrew-greptime-version:
-    name: Bump homebrew greptime version
-    if: ${{ github.ref_type == 'tag' && !contains(github.ref_name, 'nightly') && github.event_name != 'schedule' }}
-    needs: [allocate-runners, publish-github-release]
-    runs-on: ubuntu-latest
-    permissions:
-      contents: write
-      pull-requests: write
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-
-      - name: Bump homebrew greptime version
-        env:
-          GITHUB_TOKEN: ${{ secrets.HOMEBREW_GREPTIME_REPO_TOKEN }}
-          VERSION: ${{ needs.allocate-runners.outputs.version }}
-        run: |
-          ./.github/scripts/update-homebrew-greptme-version.sh

  notification:
    if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name == 'push' || github.event_name == 'schedule') && always() }}
@@ -524,7 +475,7 @@ jobs:
      build-macos-artifacts,
      build-windows-artifacts,
    ]
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    # 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.
--- a/.github/workflows/semantic-pull-request.yml
+++ b/.github/workflows/semantic-pull-request.yml
@@ -7,21 +7,14 @@ on:
      - reopened
      - edited

-concurrency:
-  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
-  cancel-in-progress: true
-
-permissions:
-  issues: write
-  contents: write
-  pull-requests: write
-
 jobs:
  check:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-22.04
    timeout-minutes: 10
    steps:
      - uses: actions/checkout@v4
+        with:
+          persist-credentials: false
      - uses: ./.github/actions/setup-cyborg
      - name: Check Pull Request
        working-directory: cyborg
--- a/.gitignore
+++ b/.gitignore
@@ -54,10 +54,3 @@ tests-fuzz/corpus/
 # Nix
 .direnv
 .envrc
-
-## default data home
-greptimedb_data
-
-# github
-!/.github
-
--- a/AUTHOR.md
+++ b/AUTHOR.md
@@ -3,28 +3,30 @@
 ## Individual Committers (in alphabetical order)

 * [CookiePieWw](https://github.com/CookiePieWw)
+* [KKould](https://github.com/KKould)
+* [NiwakaDev](https://github.com/NiwakaDev)
 * [etolbakov](https://github.com/etolbakov)
 * [irenjj](https://github.com/irenjj)
-* [KKould](https://github.com/KKould)
-* [Lanqing Yang](https://github.com/lyang24)
-* [NiwakaDev](https://github.com/NiwakaDev)
 * [tisonkun](https://github.com/tisonkun)
-
+* [Lanqing Yang](https://github.com/lyang24)

 ## Team Members (in alphabetical order)

+* [Breeze-P](https://github.com/Breeze-P)
+* [GrepTime](https://github.com/GrepTime)
+* [MichaelScofield](https://github.com/MichaelScofield)
+* [Wenjie0329](https://github.com/Wenjie0329)
+* [WenyXu](https://github.com/WenyXu)
+* [ZonaHex](https://github.com/ZonaHex)
 * [apdong2022](https://github.com/apdong2022)
 * [beryl678](https://github.com/beryl678)
-* [Breeze-P](https://github.com/Breeze-P)
 * [daviderli614](https://github.com/daviderli614)
 * [discord9](https://github.com/discord9)
 * [evenyag](https://github.com/evenyag)
 * [fengjiachun](https://github.com/fengjiachun)
 * [fengys1996](https://github.com/fengys1996)
-* [GrepTime](https://github.com/GrepTime)
 * [holalengyu](https://github.com/holalengyu)
 * [killme2008](https://github.com/killme2008)
-* [MichaelScofield](https://github.com/MichaelScofield)
 * [nicecui](https://github.com/nicecui)
 * [paomian](https://github.com/paomian)
 * [shuiyisong](https://github.com/shuiyisong)
@@ -32,14 +34,11 @@
 * [sunng87](https://github.com/sunng87)
 * [v0y4g3r](https://github.com/v0y4g3r)
 * [waynexia](https://github.com/waynexia)
-* [Wenjie0329](https://github.com/Wenjie0329)
-* [WenyXu](https://github.com/WenyXu)
 * [xtang](https://github.com/xtang)
 * [zhaoyingnan01](https://github.com/zhaoyingnan01)
 * [zhongzc](https://github.com/zhongzc)
-* [ZonaHex](https://github.com/ZonaHex)
 * [zyy17](https://github.com/zyy17)

 ## All Contributors

-To see the full list of contributors, please visit our [Contributors page](https://github.com/GreptimeTeam/greptimedb/graphs/contributors)
+[![All Contributors](https://contrib.rocks/image?repo=GreptimeTeam/greptimedb)](https://github.com/GreptimeTeam/greptimedb/graphs/contributors)
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -185,7 +185,7 @@ checksum = "d301b3b94cb4b2f23d7917810addbbaff90738e0ca2be692bd027e70d7e0330c"

 [[package]]
 name = "api"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "common-base",
 "common-decimal",
@@ -432,7 +432,7 @@ dependencies = [
 "arrow-schema",
 "chrono",
 "half",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "lexical-core",
 "num",
 "serde",
@@ -710,7 +710,7 @@ dependencies = [

 [[package]]
 name = "auth"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "async-trait",
@@ -1324,7 +1324,7 @@ dependencies = [

 [[package]]
 name = "cache"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "catalog",
 "common-error",
@@ -1348,7 +1348,7 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"

 [[package]]
 name = "catalog"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "arrow",
@@ -1475,7 +1475,7 @@ version = "0.13.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6026d8cd82ada8bbcfe337805dd1eb6afdc9e80fa4d57e977b3a36315e0c5525"
 dependencies = [
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "lazy_static",
 "num-traits",
 "regex",
@@ -1661,7 +1661,7 @@ checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97"

 [[package]]
 name = "cli"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "auth",
@@ -1703,7 +1703,7 @@ dependencies = [
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.12.2",
+ "substrait 0.12.1",
 "table",
 "tempfile",
 "tokio",
@@ -1712,7 +1712,7 @@ dependencies = [

 [[package]]
 name = "client"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "arc-swap",
@@ -1739,7 +1739,7 @@ dependencies = [
 "rand",
 "serde_json",
 "snafu 0.8.5",
- "substrait 0.12.2",
+ "substrait 0.12.1",
 "substrait 0.37.3",
 "tokio",
 "tokio-stream",
@@ -1780,7 +1780,7 @@ dependencies = [

 [[package]]
 name = "cmd"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "auth",
@@ -1791,7 +1791,6 @@ dependencies = [
 "clap 4.5.19",
 "cli",
 "client",
- "colored",
 "common-base",
 "common-catalog",
 "common-config",
@@ -1826,10 +1825,7 @@ dependencies = [
 "mito2",
 "moka",
 "nu-ansi-term",
- "object-store",
- "parquet",
 "plugins",
- "pprof",
 "prometheus",
 "prost 0.13.3",
 "query",
@@ -1845,7 +1841,7 @@ dependencies = [
 "similar-asserts",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.12.2",
+ "substrait 0.12.1",
 "table",
 "temp-env",
 "tempfile",
@@ -1862,16 +1858,6 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d3fd119d74b830634cea2a0f58bbd0d54540518a14397557951e79340abc28c0"

-[[package]]
-name = "colored"
-version = "2.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "117725a109d387c937a1533ce01b450cbde6b88abceea8473c4d7a85853cda3c"
-dependencies = [
- "lazy_static",
- "windows-sys 0.59.0",
-]
-
 [[package]]
 name = "combine"
 version = "4.6.7"
@@ -1901,7 +1887,7 @@ checksum = "55b672471b4e9f9e95499ea597ff64941a309b2cdbffcc46f2cc5e2d971fd335"

 [[package]]
 name = "common-base"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "anymap2",
 "async-trait",
@@ -1923,11 +1909,11 @@ dependencies = [

 [[package]]
 name = "common-catalog"
-version = "0.12.2"
+version = "0.12.1"

 [[package]]
 name = "common-config"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "common-base",
 "common-error",
@@ -1952,7 +1938,7 @@ dependencies = [

 [[package]]
 name = "common-datasource"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "arrow",
 "arrow-schema",
@@ -1988,7 +1974,7 @@ dependencies = [

 [[package]]
 name = "common-decimal"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "bigdecimal 0.4.5",
 "common-error",
@@ -2001,7 +1987,7 @@ dependencies = [

 [[package]]
 name = "common-error"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "http 1.1.0",
 "snafu 0.8.5",
@@ -2011,7 +1997,7 @@ dependencies = [

 [[package]]
 name = "common-frontend"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "common-error",
@@ -2021,14 +2007,12 @@ dependencies = [

 [[package]]
 name = "common-function"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
- "ahash 0.8.11",
 "api",
 "approx 0.5.1",
 "arc-swap",
 "async-trait",
- "bincode",
 "common-base",
 "common-catalog",
 "common-error",
@@ -2046,7 +2030,6 @@ dependencies = [
 "geo-types",
 "geohash",
 "h3o",
- "hyperloglogplus",
 "jsonb",
 "nalgebra 0.33.2",
 "num",
@@ -2063,13 +2046,12 @@ dependencies = [
 "store-api",
 "table",
 "tokio",
- "uddsketch",
 "wkt",
 ]

 [[package]]
 name = "common-greptimedb-telemetry"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "common-runtime",
@@ -2086,7 +2068,7 @@ dependencies = [

 [[package]]
 name = "common-grpc"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "arrow-flight",
@@ -2114,7 +2096,7 @@ dependencies = [

 [[package]]
 name = "common-grpc-expr"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "common-base",
@@ -2133,7 +2115,7 @@ dependencies = [

 [[package]]
 name = "common-macro"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "arc-swap",
 "common-query",
@@ -2147,7 +2129,7 @@ dependencies = [

 [[package]]
 name = "common-mem-prof"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2160,7 +2142,7 @@ dependencies = [

 [[package]]
 name = "common-meta"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "anymap2",
 "api",
@@ -2220,7 +2202,7 @@ dependencies = [

 [[package]]
 name = "common-options"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "common-grpc",
 "humantime-serde",
@@ -2229,11 +2211,11 @@ dependencies = [

 [[package]]
 name = "common-plugins"
-version = "0.12.2"
+version = "0.12.1"

 [[package]]
 name = "common-pprof"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2245,7 +2227,7 @@ dependencies = [

 [[package]]
 name = "common-procedure"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -2272,7 +2254,7 @@ dependencies = [

 [[package]]
 name = "common-procedure-test"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "common-procedure",
@@ -2280,7 +2262,7 @@ dependencies = [

 [[package]]
 name = "common-query"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "async-trait",
@@ -2306,7 +2288,7 @@ dependencies = [

 [[package]]
 name = "common-recordbatch"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "arc-swap",
 "common-error",
@@ -2325,7 +2307,7 @@ dependencies = [

 [[package]]
 name = "common-runtime"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -2355,7 +2337,7 @@ dependencies = [

 [[package]]
 name = "common-telemetry"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "atty",
 "backtrace",
@@ -2383,7 +2365,7 @@ dependencies = [

 [[package]]
 name = "common-test-util"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "client",
 "common-query",
@@ -2395,7 +2377,7 @@ dependencies = [

 [[package]]
 name = "common-time"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "arrow",
 "chrono",
@@ -2413,7 +2395,7 @@ dependencies = [

 [[package]]
 name = "common-version"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "build-data",
 "const_format",
@@ -2423,7 +2405,7 @@ dependencies = [

 [[package]]
 name = "common-wal"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "common-base",
 "common-error",
@@ -2990,7 +2972,7 @@ dependencies = [
 "chrono",
 "half",
 "hashbrown 0.14.5",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "libc",
 "object_store",
 "parquet",
@@ -3050,7 +3032,7 @@ dependencies = [
 "datafusion-functions-aggregate-common",
 "datafusion-functions-window-common",
 "datafusion-physical-expr-common",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "paste",
 "recursive",
 "serde_json",
@@ -3172,7 +3154,7 @@ dependencies = [
 "datafusion-physical-expr-common",
 "datafusion-physical-plan",
 "half",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "log",
 "parking_lot 0.12.3",
 "paste",
@@ -3223,7 +3205,7 @@ dependencies = [
 "datafusion-common",
 "datafusion-expr",
 "datafusion-physical-expr",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "itertools 0.13.0",
 "log",
 "recursive",
@@ -3248,7 +3230,7 @@ dependencies = [
 "datafusion-physical-expr-common",
 "half",
 "hashbrown 0.14.5",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "itertools 0.13.0",
 "log",
 "paste",
@@ -3307,7 +3289,7 @@ dependencies = [
 "futures",
 "half",
 "hashbrown 0.14.5",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "itertools 0.13.0",
 "log",
 "once_cell",
@@ -3327,7 +3309,7 @@ dependencies = [
 "arrow-schema",
 "datafusion-common",
 "datafusion-expr",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "log",
 "recursive",
 "regex",
@@ -3354,7 +3336,7 @@ dependencies = [

 [[package]]
 name = "datanode"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "arrow-flight",
@@ -3406,7 +3388,7 @@ dependencies = [
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.12.2",
+ "substrait 0.12.1",
 "table",
 "tokio",
 "toml 0.8.19",
@@ -3415,7 +3397,7 @@ dependencies = [

 [[package]]
 name = "datatypes"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4059,7 +4041,7 @@ dependencies = [

 [[package]]
 name = "file-engine"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "async-trait",
@@ -4169,7 +4151,7 @@ checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8"

 [[package]]
 name = "flow"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "arrow",
@@ -4230,7 +4212,7 @@ dependencies = [
 "snafu 0.8.5",
 "store-api",
 "strum 0.25.0",
- "substrait 0.12.2",
+ "substrait 0.12.1",
 "table",
 "tokio",
 "tonic 0.12.3",
@@ -4285,7 +4267,7 @@ checksum = "6c2141d6d6c8512188a7891b4b01590a45f6dac67afb4f255c4124dbb86d4eaa"

 [[package]]
 name = "frontend"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "arc-swap",
@@ -4713,7 +4695,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=a25adc8a01340231121646d8f0a29d0e92f45461#a25adc8a01340231121646d8f0a29d0e92f45461"
 dependencies = [
 "prost 0.13.3",
 "serde",
@@ -4736,7 +4718,7 @@ dependencies = [
 "futures-sink",
 "futures-util",
 "http 0.2.12",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "slab",
 "tokio",
 "tokio-util",
@@ -4755,7 +4737,7 @@ dependencies = [
 "futures-core",
 "futures-sink",
 "http 1.1.0",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "slab",
 "tokio",
 "tokio-util",
@@ -5305,15 +5287,6 @@ dependencies = [
 "tracing",
 ]

-[[package]]
-name = "hyperloglogplus"
-version = "0.4.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "621debdf94dcac33e50475fdd76d34d5ea9c0362a834b9db08c3024696c1fbe3"
-dependencies = [
- "serde",
-]
-
 [[package]]
 name = "i_float"
 version = "1.3.1"
@@ -5553,7 +5526,7 @@ dependencies = [

 [[package]]
 name = "index"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "asynchronous-codec",
@@ -5602,9 +5575,9 @@ dependencies = [

 [[package]]
 name = "indexmap"
-version = "2.7.1"
+version = "2.6.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8c9c992b02b5b4c94ea26e32fe5bccb7aa7d9f390ab5c1221ff895bc7ea8b652"
+checksum = "707907fe3c25f5424cce2cb7e1cbcafee6bdbe735ca90ef77c29e84591e5b9da"
 dependencies = [
 "equivalent",
 "hashbrown 0.15.2",
@@ -5618,7 +5591,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "232929e1d75fe899576a3d5c7416ad0d88dbfbb3c3d6aa00873a7408a50ddb88"
 dependencies = [
 "ahash 0.8.11",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "is-terminal",
 "itoa",
 "log",
@@ -5965,7 +5938,7 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ee7893dab2e44ae5f9d0173f26ff4aa327c10b01b06a72b52dd9405b628640d"
 dependencies = [
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 ]

 [[package]]
@@ -6345,7 +6318,7 @@ checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"

 [[package]]
 name = "log-query"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "chrono",
 "common-error",
@@ -6357,7 +6330,7 @@ dependencies = [

 [[package]]
 name = "log-store"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -6448,7 +6421,7 @@ dependencies = [
 "cactus",
 "cfgrammar",
 "filetime",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "lazy_static",
 "lrtable",
 "num-traits",
@@ -6650,7 +6623,7 @@ dependencies = [

 [[package]]
 name = "meta-client"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "async-trait",
@@ -6677,7 +6650,7 @@ dependencies = [

 [[package]]
 name = "meta-srv"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "async-trait",
@@ -6763,7 +6736,7 @@ dependencies = [

 [[package]]
 name = "metric-engine"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "aquamarine",
@@ -6861,7 +6834,7 @@ dependencies = [

 [[package]]
 name = "mito2"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "aquamarine",
@@ -7558,7 +7531,7 @@ dependencies = [

 [[package]]
 name = "object-store"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "anyhow",
 "bytes",
@@ -7689,7 +7662,7 @@ checksum = "1e32339a5dc40459130b3bd269e9892439f55b33e772d2a9d402a789baaf4e8a"
 dependencies = [
 "futures-core",
 "futures-sink",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "js-sys",
 "once_cell",
 "pin-project-lite",
@@ -7807,7 +7780,7 @@ dependencies = [

 [[package]]
 name = "operator"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -7855,7 +7828,7 @@ dependencies = [
 "sql",
 "sqlparser 0.52.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=71dd86058d2af97b9925093d40c4e03360403170)",
 "store-api",
- "substrait 0.12.2",
+ "substrait 0.12.1",
 "table",
 "tokio",
 "tokio-util",
@@ -8092,7 +8065,7 @@ dependencies = [

 [[package]]
 name = "partition"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "async-trait",
@@ -8261,7 +8234,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b4c5cc86750666a3ed20bdaf5ca2a0344f9c67674cae0515bec2da16fbaa47db"
 dependencies = [
 "fixedbitset",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 ]

 [[package]]
@@ -8360,7 +8333,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"

 [[package]]
 name = "pipeline"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -8500,7 +8473,7 @@ dependencies = [

 [[package]]
 name = "plugins"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "auth",
 "clap 4.5.19",
@@ -8762,7 +8735,7 @@ dependencies = [

 [[package]]
 name = "promql"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "ahash 0.8.11",
 "async-trait",
@@ -9007,7 +8980,7 @@ dependencies = [

 [[package]]
 name = "puffin"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-compression 0.4.13",
 "async-trait",
@@ -9048,7 +9021,7 @@ dependencies = [

 [[package]]
 name = "query"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -9113,7 +9086,7 @@ dependencies = [
 "sqlparser 0.52.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=71dd86058d2af97b9925093d40c4e03360403170)",
 "statrs",
 "store-api",
- "substrait 0.12.2",
+ "substrait 0.12.1",
 "table",
 "tokio",
 "tokio-stream",
@@ -10352,7 +10325,7 @@ version = "1.0.137"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "930cfb6e6abf99298aaad7d29abbef7a9999a9a8806a40088f55f0dcec03146b"
 dependencies = [
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "itoa",
 "memchr",
 "ryu",
@@ -10423,7 +10396,7 @@ dependencies = [
 "chrono",
 "hex",
 "indexmap 1.9.3",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "serde",
 "serde_derive",
 "serde_json",
@@ -10449,7 +10422,7 @@ version = "0.9.34+deprecated"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
 dependencies = [
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "itoa",
 "ryu",
 "serde",
@@ -10458,7 +10431,7 @@ dependencies = [

 [[package]]
 name = "servers"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -10510,7 +10483,6 @@ dependencies = [
 "humantime",
 "humantime-serde",
 "hyper 1.4.1",
- "indexmap 2.7.1",
 "influxdb_line_protocol",
 "itertools 0.10.5",
 "json5",
@@ -10575,7 +10547,7 @@ dependencies = [

 [[package]]
 name = "session"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "arc-swap",
@@ -10884,7 +10856,7 @@ dependencies = [

 [[package]]
 name = "sql"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "chrono",
@@ -10921,12 +10893,12 @@ dependencies = [
 [[package]]
 name = "sqlness"
 version = "0.6.1"
-source = "git+https://github.com/CeresDB/sqlness.git?rev=bb91f31ff58993e07ea89845791235138283a24c#bb91f31ff58993e07ea89845791235138283a24c"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "308a7338f2211813d6e9da117e9b9b7aee5d072872d11a934002fd2bd4ab5276"
 dependencies = [
 "async-trait",
 "derive_builder 0.11.2",
 "duration-str",
- "futures",
 "minijinja",
 "prettydiff",
 "regex",
@@ -10938,7 +10910,7 @@ dependencies = [

 [[package]]
 name = "sqlness-runner"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -10952,7 +10924,6 @@ dependencies = [
 "hex",
 "local-ip-address",
 "mysql",
- "num_cpus",
 "reqwest",
 "serde",
 "serde_json",
@@ -11052,7 +11023,7 @@ dependencies = [
 "futures-util",
 "hashbrown 0.15.2",
 "hashlink",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "log",
 "memchr",
 "once_cell",
@@ -11255,7 +11226,7 @@ dependencies = [

 [[package]]
 name = "store-api"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "aquamarine",
@@ -11385,7 +11356,7 @@ dependencies = [

 [[package]]
 name = "substrait"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "async-trait",
 "bytes",
@@ -11566,7 +11537,7 @@ dependencies = [

 [[package]]
 name = "table"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "async-trait",
@@ -11817,7 +11788,7 @@ checksum = "3369f5ac52d5eb6ab48c6b4ffdc8efbcad6b89c765749064ba298f2c68a16a76"

 [[package]]
 name = "tests-fuzz"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "arbitrary",
 "async-trait",
@@ -11861,7 +11832,7 @@ dependencies = [

 [[package]]
 name = "tests-integration"
-version = "0.12.2"
+version = "0.12.1"
 dependencies = [
 "api",
 "arrow-flight",
@@ -11927,7 +11898,7 @@ dependencies = [
 "sql",
 "sqlx",
 "store-api",
- "substrait 0.12.2",
+ "substrait 0.12.1",
 "table",
 "tempfile",
 "time",
@@ -12348,7 +12319,7 @@ version = "0.19.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1b5bb770da30e5cbfde35a2d7b9b8a2c4b8ef89548a7a6aeab5c9a576e3e7421"
 dependencies = [
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "toml_datetime",
 "winnow 0.5.40",
 ]
@@ -12359,7 +12330,7 @@ version = "0.22.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ae48d6208a266e853d946088ed816055e556cc6028c5e8e2b84d9fa5dd7c7f5"
 dependencies = [
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "serde",
 "serde_spanned",
 "toml_datetime",
@@ -12497,7 +12468,7 @@ dependencies = [
 "futures-core",
 "futures-util",
 "hdrhistogram",
- "indexmap 2.7.1",
+ "indexmap 2.6.0",
 "pin-project-lite",
 "slab",
 "sync_wrapper 1.0.1",
@@ -12985,14 +12956,6 @@ version = "0.1.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2896d95c02a80c6d6a5d6e953d479f5ddf2dfdb6a244441010e373ac0fb88971"

-[[package]]
-name = "uddsketch"
-version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/timescaledb-toolkit.git?rev=84828fe8fb494a6a61412a3da96517fc80f7bb20#84828fe8fb494a6a61412a3da96517fc80f7bb20"
-dependencies = [
- "serde",
-]
-
 [[package]]
 name = "unescaper"
 version = "0.1.5"
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -67,7 +67,7 @@ members = [
 resolver = "2"

 [workspace.package]
-version = "0.12.2"
+version = "0.12.1"
 edition = "2021"
 license = "Apache-2.0"

@@ -129,7 +129,7 @@ etcd-client = "0.14"
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "072ce580502e015df1a6b03a185b60309a7c2a7a" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "a25adc8a01340231121646d8f0a29d0e92f45461" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
--- a/config/config.md
+++ b/config/config.md
@@ -319,7 +319,6 @@
 | `selector` | String | `round_robin` | Datanode selector type.<br/>- `round_robin` (default value)<br/>- `lease_based`<br/>- `load_based`<br/>For details, please see "https://docs.greptime.com/developer-guide/metasrv/selector". |
 | `use_memory_store` | Bool | `false` | Store data in memory. |
 | `enable_region_failover` | Bool | `false` | Whether to enable region failover.<br/>This feature is only available on GreptimeDB running on cluster mode and<br/>- Using Remote WAL<br/>- Using shared storage (e.g., s3). |
-| `node_max_idle_time` | String | `24hours` | Max allowed idle time before removing node info from metasrv memory. |
 | `enable_telemetry` | Bool | `true` | Whether to enable greptimedb telemetry. Enabled by default. |
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
--- a/config/datanode.example.toml
+++ b/config/datanode.example.toml
@@ -231,7 +231,6 @@ 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]
--- a/config/metasrv.example.toml
+++ b/config/metasrv.example.toml
@@ -50,9 +50,6 @@ use_memory_store = false
 ## - Using shared storage (e.g., s3).
 enable_region_failover = false

-## Max allowed idle time before removing node info from metasrv memory.
-node_max_idle_time = "24hours"
-
 ## Whether to enable greptimedb telemetry. Enabled by default.
 #+ enable_telemetry = true

--- a/config/standalone.example.toml
+++ b/config/standalone.example.toml
@@ -318,7 +318,6 @@ 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]
--- a/cyborg/bin/bump-versions.ts
+++ b/cyborg/bin/bump-versions.ts
@@ -1,156 +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.
- */
-
-import * as core from "@actions/core";
-import {obtainClient} from "@/common";
-
-interface RepoConfig {
-  tokenEnv: string;
-  repo: string;
-  workflowLogic: (version: string) => [string, string] | null;
-}
-
-const REPO_CONFIGS: Record<string, RepoConfig> = {
-  website: {
-    tokenEnv: "WEBSITE_REPO_TOKEN",
-    repo: "website",
-    workflowLogic: (version: string) => {
-      // Skip nightly versions for website
-      if (version.includes('nightly')) {
-        console.log('Nightly version detected for website, skipping workflow trigger.');
-        return null;
-      }
-      return ['bump-patch-version.yml', version];
-    }
-  },
-  demo: {
-    tokenEnv: "DEMO_REPO_TOKEN",
-    repo: "demo-scene",
-    workflowLogic: (version: string) => {
-      // Skip nightly versions for demo
-      if (version.includes('nightly')) {
-        console.log('Nightly version detected for demo, skipping workflow trigger.');
-        return null;
-      }
-      return ['bump-patch-version.yml', version];
-    }
-  },
-  docs: {
-    tokenEnv: "DOCS_REPO_TOKEN",
-    repo: "docs",
-    workflowLogic: (version: string) => {
-      // Check if it's a nightly version
-      if (version.includes('nightly')) {
-        return ['bump-nightly-version.yml', version];
-      }
-
-      const parts = version.split('.');
-      if (parts.length !== 3) {
-        throw new Error('Invalid version format');
-      }
-
-      // If patch version (last number) is 0, it's a major version
-      // Return only major.minor version
-      if (parts[2] === '0') {
-        return ['bump-version.yml', `${parts[0]}.${parts[1]}`];
-      }
-
-      // Otherwise it's a patch version, use full version
-      return ['bump-patch-version.yml', version];
-    }
-  }
-};
-
-async function triggerWorkflow(repoConfig: RepoConfig, workflowId: string, version: string) {
-  const client = obtainClient(repoConfig.tokenEnv);
-  try {
-    await client.rest.actions.createWorkflowDispatch({
-      owner: "GreptimeTeam",
-      repo: repoConfig.repo,
-      workflow_id: workflowId,
-      ref: "main",
-      inputs: {
-        version,
-      },
-    });
-    console.log(`Successfully triggered ${workflowId} workflow for ${repoConfig.repo} with version ${version}`);
-  } catch (error) {
-    core.setFailed(`Failed to trigger workflow for ${repoConfig.repo}: ${error.message}`);
-    throw error;
-  }
-}
-
-async function processRepo(repoName: string, version: string) {
-  const repoConfig = REPO_CONFIGS[repoName];
-  if (!repoConfig) {
-    throw new Error(`Unknown repository: ${repoName}`);
-  }
-
-  try {
-    const workflowResult = repoConfig.workflowLogic(version);
-    if (workflowResult === null) {
-      // Skip this repo (e.g., nightly version for website)
-      return;
-    }
-
-    const [workflowId, apiVersion] = workflowResult;
-    await triggerWorkflow(repoConfig, workflowId, apiVersion);
-  } catch (error) {
-    core.setFailed(`Error processing ${repoName} with version ${version}: ${error.message}`);
-    throw error;
-  }
-}
-
-async function main() {
-  const version = process.env.VERSION;
-  if (!version) {
-    core.setFailed("VERSION environment variable is required");
-    process.exit(1);
-  }
-
-  // Remove 'v' prefix if exists
-  const cleanVersion = version.startsWith('v') ? version.slice(1) : version;
-
-  // Get target repositories from environment variable
-  // Default to both if not specified
-  const targetRepos = process.env.TARGET_REPOS?.split(',').map(repo => repo.trim()) || ['website', 'docs'];
-
-  console.log(`Processing version ${cleanVersion} for repositories: ${targetRepos.join(', ')}`);
-
-  const errors: string[] = [];
-
-  // Process each repository
-  for (const repo of targetRepos) {
-    try {
-      await processRepo(repo, cleanVersion);
-    } catch (error) {
-      errors.push(`${repo}: ${error.message}`);
-    }
-  }
-
-  if (errors.length > 0) {
-    core.setFailed(`Failed to process some repositories: ${errors.join('; ')}`);
-    process.exit(1);
-  }
-
-  console.log('All repositories processed successfully');
-}
-
-// Execute main function
-main().catch((error) => {
-  core.setFailed(`Unexpected error: ${error.message}`);
-  process.exit(1);
-});
--- a/cyborg/bin/follow-up-docs-issue.ts
+++ b/cyborg/bin/follow-up-docs-issue.ts
@@ -55,25 +55,12 @@ async function main() {
        await client.rest.issues.addLabels({
            owner, repo, issue_number: number, labels: [labelDocsRequired],
        })
-
-        // Get available assignees for the docs repo
-        const assigneesResponse = await docsClient.rest.issues.listAssignees({
-            owner: 'GreptimeTeam',
-            repo: 'docs',
-        })
-        const validAssignees = assigneesResponse.data.map(assignee => assignee.login)
-        core.info(`Available assignees: ${validAssignees.join(', ')}`)
-
-        // Check if the actor is a valid assignee, otherwise fallback to fengjiachun
-        const assignee = validAssignees.includes(actor) ? actor : 'fengjiachun'
-        core.info(`Assigning issue to: ${assignee}`)
-
        await docsClient.rest.issues.create({
            owner: 'GreptimeTeam',
            repo: 'docs',
            title: `Update docs for ${title}`,
            body: `A document change request is generated from ${html_url}`,
-            assignee: assignee,
+            assignee: actor,
        }).then((res) => {
            core.info(`Created issue ${res.data}`)
        })
--- a/scripts/install.sh
+++ b/scripts/install.sh
@@ -53,54 +53,6 @@ get_arch_type() {
  esac
 }

-# Verify SHA256 checksum
-verify_sha256() {
-  file="$1"
-  expected_sha256="$2"
-
-  if command -v sha256sum >/dev/null 2>&1; then
-    actual_sha256=$(sha256sum "$file" | cut -d' ' -f1)
-  elif command -v shasum >/dev/null 2>&1; then
-    actual_sha256=$(shasum -a 256 "$file" | cut -d' ' -f1)
-  else
-    echo "Warning: No SHA256 verification tool found (sha256sum or shasum). Skipping checksum verification."
-    return 0
-  fi
-
-  if [ "$actual_sha256" = "$expected_sha256" ]; then
-    echo "SHA256 checksum verified successfully."
-    return 0
-  else
-    echo "Error: SHA256 checksum verification failed!"
-    echo "Expected: $expected_sha256"
-    echo "Actual: $actual_sha256"
-    return 1
-  fi
-}
-
-# Prompt for user confirmation (compatible with different shells)
-prompt_confirmation() {
-  message="$1"
-  printf "%s (y/N): " "$message"
-
-  # Try to read user input, fallback if read fails
-  answer=""
-  if read answer </dev/tty 2>/dev/null; then
-    case "$answer" in
-      [Yy]|[Yy][Ee][Ss])
-        return 0
-        ;;
-      *)
-        return 1
-        ;;
-    esac
-  else
-    echo ""
-    echo "Cannot read user input. Defaulting to No."
-    return 1
-  fi
-}
-
 download_artifact() {
  if [ -n "${OS_TYPE}" ] && [ -n "${ARCH_TYPE}" ]; then
    # Use the latest stable released version.
@@ -119,104 +71,17 @@ download_artifact() {
    fi

    echo "Downloading ${BIN}, OS: ${OS_TYPE}, Arch: ${ARCH_TYPE}, Version: ${VERSION}"
-    PKG_NAME="${BIN}-${OS_TYPE}-${ARCH_TYPE}-${VERSION}"
-    PACKAGE_NAME="${PKG_NAME}.tar.gz"
-    SHA256_FILE="${PKG_NAME}.sha256sum"
+    PACKAGE_NAME="${BIN}-${OS_TYPE}-${ARCH_TYPE}-${VERSION}.tar.gz"

    if [ -n "${PACKAGE_NAME}" ]; then
-      # Check if files already exist and prompt for override
-      if [ -f "${PACKAGE_NAME}" ]; then
-        echo "File ${PACKAGE_NAME} already exists."
-        if prompt_confirmation "Do you want to override it?"; then
-          echo "Overriding existing file..."
-          rm -f "${PACKAGE_NAME}"
-        else
-          echo "Skipping download. Using existing file."
-        fi
-      fi
-
-      if [ -f "${BIN}" ]; then
-        echo "Binary ${BIN} already exists."
-        if prompt_confirmation "Do you want to override it?"; then
-          echo "Will override existing binary..."
-          rm -f "${BIN}"
-        else
-          echo "Installation cancelled."
-          exit 0
-        fi
-      fi
-
-      # Download package if not exists
-      if [ ! -f "${PACKAGE_NAME}" ]; then
-        echo "Downloading ${PACKAGE_NAME}..."
-        # Use curl instead of wget for better compatibility
-        if command -v curl >/dev/null 2>&1; then
-          if ! curl -L -o "${PACKAGE_NAME}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"; then
-            echo "Error: Failed to download ${PACKAGE_NAME}"
-            exit 1
-          fi
-        elif command -v wget >/dev/null 2>&1; then
-          if ! wget -O "${PACKAGE_NAME}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"; then
-            echo "Error: Failed to download ${PACKAGE_NAME}"
-            exit 1
-          fi
-        else
-          echo "Error: Neither curl nor wget is available for downloading."
-          exit 1
-        fi
-      fi
-
-      # Download and verify SHA256 checksum
-      echo "Downloading SHA256 checksum..."
-      sha256_download_success=0
-      if command -v curl >/dev/null 2>&1; then
-        if curl -L -s -o "${SHA256_FILE}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${SHA256_FILE}" 2>/dev/null; then
-          sha256_download_success=1
-        fi
-      elif command -v wget >/dev/null 2>&1; then
-        if wget -q -O "${SHA256_FILE}" "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${SHA256_FILE}" 2>/dev/null; then
-          sha256_download_success=1
-        fi
-      fi
-
-      if [ $sha256_download_success -eq 1 ] && [ -f "${SHA256_FILE}" ]; then
-        expected_sha256=$(cat "${SHA256_FILE}" | cut -d' ' -f1)
-        if [ -n "$expected_sha256" ]; then
-          if ! verify_sha256 "${PACKAGE_NAME}" "${expected_sha256}"; then
-            echo "SHA256 verification failed. Removing downloaded file."
-            rm -f "${PACKAGE_NAME}" "${SHA256_FILE}"
-            exit 1
-          fi
-        else
-          echo "Warning: Could not parse SHA256 checksum from file."
-        fi
-        rm -f "${SHA256_FILE}"
-      else
-        echo "Warning: Could not download SHA256 checksum file. Skipping verification."
-      fi
+      wget "https://github.com/${GITHUB_ORG}/${GITHUB_REPO}/releases/download/${VERSION}/${PACKAGE_NAME}"

      # Extract the binary and clean the rest.
-      echo "Extracting ${PACKAGE_NAME}..."
-      if ! tar xf "${PACKAGE_NAME}"; then
-        echo "Error: Failed to extract ${PACKAGE_NAME}"
-        exit 1
-      fi
-
-      # Find the binary in the extracted directory
-      extracted_dir="${PACKAGE_NAME%.tar.gz}"
-      if [ -f "${extracted_dir}/${BIN}" ]; then
-        mv "${extracted_dir}/${BIN}" "${PWD}/"
-        rm -f "${PACKAGE_NAME}"
-        rm -rf "${extracted_dir}"
-        chmod +x "${BIN}"
-        echo "Installation completed successfully!"
-        echo "Run './${BIN} --help' to get started"
-      else
-        echo "Error: Binary ${BIN} not found in extracted archive"
-        rm -f "${PACKAGE_NAME}"
-        rm -rf "${extracted_dir}"
-        exit 1
-      fi
+      tar xvf "${PACKAGE_NAME}" && \
+      mv "${PACKAGE_NAME%.tar.gz}/${BIN}" "${PWD}" && \
+      rm -r "${PACKAGE_NAME}" && \
+      rm -r "${PACKAGE_NAME%.tar.gz}" && \
+      echo "Run './${BIN} --help' to get started"
    fi
  fi
 }
--- a/src/catalog/src/system_schema/information_schema/tables.rs
+++ b/src/catalog/src/system_schema/information_schema/tables.rs
@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::collections::HashSet;
 use std::sync::{Arc, Weak};

 use arrow_schema::SchemaRef as ArrowSchemaRef;
@@ -244,14 +243,17 @@ impl InformationSchemaTablesBuilder {
        // TODO(dennis): `region_stats` API is not stable in distributed cluster because of network issue etc.
        // But we don't want the statements such as `show tables` fail,
        // so using `unwrap_or_else` here instead of `?` operator.
-        let region_stats = information_extension
-            .region_stats()
-            .await
-            .map_err(|e| {
-                error!(e; "Failed to call region_stats");
-                e
-            })
-            .unwrap_or_else(|_| vec![]);
+        let region_stats = {
+            let mut x = information_extension
+                .region_stats()
+                .await
+                .unwrap_or_else(|e| {
+                    error!(e; "Failed to find region stats in information_schema, fallback to all empty");
+                    vec![]
+                });
+            x.sort_unstable_by_key(|x| x.id);
+            x
+        };

        for schema_name in catalog_manager.schema_names(&catalog_name, None).await? {
            let mut stream = catalog_manager.tables(&catalog_name, &schema_name, None);
@@ -262,16 +264,16 @@ impl InformationSchemaTablesBuilder {
                // TODO(dennis): make it working for metric engine
                let table_region_stats =
                    if table_info.meta.engine == MITO_ENGINE || table_info.is_physical_table() {
-                        let region_ids = table_info
+                        table_info
                            .meta
                            .region_numbers
                            .iter()
                            .map(|n| RegionId::new(table_info.ident.table_id, *n))
-                            .collect::<HashSet<_>>();
-
-                        region_stats
-                            .iter()
-                            .filter(|stat| region_ids.contains(&stat.id))
+                            .flat_map(|region_id| {
+                                region_stats
+                                    .binary_search_by_key(&region_id, |x| x.id)
+                                    .map(|i| &region_stats[i])
+                            })
                            .collect::<Vec<_>>()
                    } else {
                        vec![]
--- a/src/cmd/Cargo.toml
+++ b/src/cmd/Cargo.toml
@@ -9,10 +9,6 @@ default-run = "greptime"
 name = "greptime"
 path = "src/bin/greptime.rs"

-[[bin]]
-name = "objbench"
-path = "src/bin/objbench.rs"
-
 [features]
 default = ["servers/pprof", "servers/mem-prof"]
 tokio-console = ["common-telemetry/tokio-console"]
@@ -24,7 +20,6 @@ workspace = true
 async-trait.workspace = true
 auth.workspace = true
 base64.workspace = true
-colored = "2.0"
 cache.workspace = true
 catalog.workspace = true
 chrono.workspace = true
@@ -60,9 +55,6 @@ futures.workspace = true
 human-panic = "2.0"
 humantime.workspace = true
 lazy_static.workspace = true
-object-store.workspace = true
-parquet = "53"
-pprof = "0.14"
 meta-client.workspace = true
 meta-srv.workspace = true
 metric-engine.workspace = true
--- a/src/cmd/src/bin/greptime.rs
+++ b/src/cmd/src/bin/greptime.rs
@@ -21,8 +21,6 @@ use cmd::{cli, datanode, flownode, frontend, metasrv, standalone, App};
 use common_version::version;
 use servers::install_ring_crypto_provider;

-pub mod objbench;
-
 #[derive(Parser)]
 #[command(name = "greptime", author, version, long_version = version(), about)]
 #[command(propagate_version = true)]
--- a/src/cmd/src/bin/objbench.rs
+++ b/src/cmd/src/bin/objbench.rs
@@ -1,602 +0,0 @@
-// Copyright 2025 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::path::{Path, PathBuf};
-use std::time::Instant;
-
-use clap::Parser;
-use cmd::error::{self, Result};
-use colored::Colorize;
-use datanode::config::ObjectStoreConfig;
-use mito2::config::{FulltextIndexConfig, MitoConfig, Mode};
-use mito2::read::Source;
-use mito2::sst::file::{FileHandle, FileId, FileMeta};
-use mito2::sst::file_purger::{FilePurger, FilePurgerRef, PurgeRequest};
-use mito2::sst::parquet::{WriteOptions, PARQUET_METADATA_KEY};
-use mito2::{build_access_layer, Metrics, OperationType, SstWriteRequest};
-use object_store::ObjectStore;
-use serde::{Deserialize, Serialize};
-use store_api::metadata::{RegionMetadata, RegionMetadataRef};
-
-#[tokio::main]
-pub async fn main() {
-    // common_telemetry::init_default_ut_logging();
-    let cmd = Command::parse();
-    if let Err(e) = cmd.run().await {
-        eprintln!("{}: {}", "Error".red().bold(), e);
-        std::process::exit(1);
-    }
-}
-
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
-#[serde(default)]
-pub struct StorageConfigWrapper {
-    storage: StorageConfig,
-}
-
-/// Storage engine config
-#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
-#[serde(default)]
-pub struct StorageConfig {
-    /// The working directory of database
-    pub data_home: String,
-    #[serde(flatten)]
-    pub store: ObjectStoreConfig,
-}
-
-#[derive(Debug, Parser)]
-pub struct Command {
-    /// Path to the object-store config file (TOML). Must deserialize into datanode::config::ObjectStoreConfig.
-    #[clap(long, value_name = "FILE")]
-    pub config: PathBuf,
-
-    /// Source SST file path in object-store (e.g. "region_dir/<uuid>.parquet").
-    #[clap(long, value_name = "PATH")]
-    pub source: String,
-
-    /// Target SST file path in object-store; its parent directory is used as destination region dir.
-    #[clap(long, value_name = "PATH")]
-    pub target: String,
-
-    /// Verbose output
-    #[clap(short, long, default_value_t = false)]
-    pub verbose: bool,
-
-    /// Output file path for pprof flamegraph (enables profiling)
-    #[clap(long, value_name = "FILE")]
-    pub pprof_file: Option<PathBuf>,
-}
-
-impl Command {
-    pub async fn run(&self) -> Result<()> {
-        if self.verbose {
-            common_telemetry::init_default_ut_logging();
-        }
-
-        println!("{}", "Starting objbench...".cyan().bold());
-
-        // Build object store from config
-        let cfg_str = std::fs::read_to_string(&self.config).map_err(|e| {
-            error::IllegalConfigSnafu {
-                msg: format!("failed to read config {}: {e}", self.config.display()),
-            }
-            .build()
-        })?;
-        let store_cfg: StorageConfigWrapper = toml::from_str(&cfg_str).map_err(|e| {
-            error::IllegalConfigSnafu {
-                msg: format!("failed to parse config {}: {e}", self.config.display()),
-            }
-            .build()
-        })?;
-
-        let object_store = build_object_store(&store_cfg.storage).await?;
-        println!("{} Object store initialized", "✓".green());
-
-        // Prepare source identifiers
-        let (src_region_dir, src_file_id) = split_sst_path(&self.source)?;
-        println!("{} Source path parsed: {}", "✓".green(), self.source);
-
-        // Load parquet metadata to extract RegionMetadata and file stats
-        println!("{}", "Loading parquet metadata...".yellow());
-        let file_size = object_store
-            .stat(&self.source)
-            .await
-            .map_err(|e| {
-                error::IllegalConfigSnafu {
-                    msg: format!("stat failed: {e}"),
-                }
-                .build()
-            })?
-            .content_length();
-        let parquet_meta = load_parquet_metadata(object_store.clone(), &self.source, file_size)
-            .await
-            .map_err(|e| {
-                error::IllegalConfigSnafu {
-                    msg: format!("read parquet metadata failed: {e}"),
-                }
-                .build()
-            })?;
-
-        let region_meta = extract_region_metadata(&self.source, &parquet_meta)?;
-        let num_rows = parquet_meta.file_metadata().num_rows() as u64;
-        let num_row_groups = parquet_meta.num_row_groups() as u64;
-
-        println!(
-            "{} Metadata loaded - rows: {}, size: {} bytes",
-            "✓".green(),
-            num_rows,
-            file_size
-        );
-
-        // Build a FileHandle for the source file
-        let file_meta = FileMeta {
-            region_id: region_meta.region_id,
-            file_id: src_file_id,
-            time_range: Default::default(),
-            level: 0,
-            file_size,
-            available_indexes: Default::default(),
-            index_file_size: 0,
-            num_rows,
-            num_row_groups,
-            sequence: None,
-        };
-        let src_handle = FileHandle::new(file_meta, new_noop_file_purger());
-
-        // Build the reader for a single file via ParquetReaderBuilder
-        println!("{}", "Building reader...".yellow());
-        let (_src_access_layer, _cache_manager) =
-            build_access_layer_simple(src_region_dir.clone(), object_store.clone()).await?;
-        let reader_build_start = Instant::now();
-        let reader = mito2::sst::parquet::reader::ParquetReaderBuilder::new(
-            src_region_dir.clone(),
-            src_handle.clone(),
-            object_store.clone(),
-        )
-        .expected_metadata(Some(region_meta.clone()))
-        .build()
-        .await
-        .map_err(|e| {
-            error::IllegalConfigSnafu {
-                msg: format!("build reader failed: {e}"),
-            }
-            .build()
-        })?;
-
-        let reader_build_elapsed = reader_build_start.elapsed();
-        let total_rows = reader.parquet_metadata().file_metadata().num_rows();
-        println!("{} Reader built in {:?}", "✓".green(), reader_build_elapsed);
-
-        // Prepare target access layer for writing
-        println!("{}", "Preparing target access layer...".yellow());
-        let (tgt_access_layer, tgt_cache_manager) =
-            build_access_layer_simple(self.target.clone(), object_store.clone()).await?;
-
-        // Build write request
-        let fulltext_index_config = FulltextIndexConfig {
-            create_on_compaction: Mode::Disable,
-            ..Default::default()
-        };
-        let write_opts = WriteOptions::default();
-        let write_req = SstWriteRequest {
-            op_type: OperationType::Compact,
-            metadata: region_meta,
-            source: Source::Reader(Box::new(reader)),
-            cache_manager: tgt_cache_manager,
-            storage: None,
-            max_sequence: None,
-            index_options: Default::default(),
-            inverted_index_config: MitoConfig::default().inverted_index,
-            fulltext_index_config,
-            bloom_filter_index_config: MitoConfig::default().bloom_filter_index,
-        };
-
-        // Write SST
-        println!("{}", "Writing SST...".yellow());
-        let mut metrics = Metrics::default();
-
-        // Start profiling if pprof_file is specified
-        #[cfg(unix)]
-        let profiler_guard = if self.pprof_file.is_some() {
-            println!("{} Starting profiling...", "⚡".yellow());
-            Some(
-                pprof::ProfilerGuardBuilder::default()
-                    .frequency(99)
-                    .blocklist(&["libc", "libgcc", "pthread", "vdso"])
-                    .build()
-                    .map_err(|e| {
-                        error::IllegalConfigSnafu {
-                            msg: format!("Failed to start profiler: {e}"),
-                        }
-                        .build()
-                    })?,
-            )
-        } else {
-            None
-        };
-
-        #[cfg(not(unix))]
-        if self.pprof_file.is_some() {
-            eprintln!(
-                "{}: Profiling is not supported on this platform",
-                "Warning".yellow()
-            );
-        }
-
-        let write_start = Instant::now();
-        let infos = tgt_access_layer
-            .write_sst(write_req, &write_opts, &mut metrics)
-            .await
-            .map_err(|e| {
-                error::IllegalConfigSnafu {
-                    msg: format!("write_sst failed: {e}"),
-                }
-                .build()
-            })?;
-
-        let write_elapsed = write_start.elapsed();
-
-        // Stop profiling and generate flamegraph if enabled
-        #[cfg(unix)]
-        if let (Some(guard), Some(pprof_file)) = (profiler_guard, &self.pprof_file) {
-            println!("{} Generating flamegraph...", "🔥".yellow());
-            match guard.report().build() {
-                Ok(report) => {
-                    let mut flamegraph_data = Vec::new();
-                    if let Err(e) = report.flamegraph(&mut flamegraph_data) {
-                        eprintln!(
-                            "{}: Failed to generate flamegraph: {}",
-                            "Warning".yellow(),
-                            e
-                        );
-                    } else if let Err(e) = std::fs::write(pprof_file, flamegraph_data) {
-                        eprintln!(
-                            "{}: Failed to write flamegraph to {}: {}",
-                            "Warning".yellow(),
-                            pprof_file.display(),
-                            e
-                        );
-                    } else {
-                        println!(
-                            "{} Flamegraph saved to {}",
-                            "✓".green(),
-                            pprof_file.display().to_string().cyan()
-                        );
-                    }
-                }
-                Err(e) => {
-                    eprintln!(
-                        "{}: Failed to generate pprof report: {}",
-                        "Warning".yellow(),
-                        e
-                    );
-                }
-            }
-        }
-        assert_eq!(infos.len(), 1);
-        let dst_file_id = infos[0].file_id;
-        let dst_file_path = format!("{}{}", self.target, dst_file_id.as_parquet(),);
-
-        // Report results with ANSI colors
-        println!("\n{} {}", "Write complete!".green().bold(), "✓".green());
-        println!("  {}: {}", "Destination file".bold(), dst_file_path.cyan());
-        println!("  {}: {}", "Rows".bold(), total_rows.to_string().cyan());
-        println!(
-            "  {}: {}",
-            "File size".bold(),
-            format!("{} bytes", file_size).cyan()
-        );
-        println!(
-            "  {}: {:?}",
-            "Reader build time".bold(),
-            reader_build_elapsed
-        );
-        println!("  {}: {:?}", "Total time".bold(), write_elapsed);
-
-        // Print metrics in a formatted way
-        println!(
-            "  {}: {:?}, sum: {:?}",
-            "Metrics".bold(),
-            metrics,
-            metrics.sum()
-        );
-
-        // Print infos
-        println!("  {}: {:?}", "Index".bold(), infos[0].index_metadata);
-
-        // Cleanup
-        println!("\n{}", "Cleaning up...".yellow());
-        object_store.delete(&dst_file_path).await.map_err(|e| {
-            error::IllegalConfigSnafu {
-                msg: format!("Failed to delete dest file {}: {}", dst_file_path, e),
-            }
-            .build()
-        })?;
-        println!("{} Temporary file deleted", "✓".green());
-
-        println!("\n{}", "Benchmark completed successfully!".green().bold());
-        Ok(())
-    }
-}
-
-fn split_sst_path(path: &str) -> Result<(String, FileId)> {
-    let p = Path::new(path);
-    let file_name = p.file_name().and_then(|s| s.to_str()).ok_or_else(|| {
-        error::IllegalConfigSnafu {
-            msg: "invalid source path".to_string(),
-        }
-        .build()
-    })?;
-    let uuid_str = file_name.strip_suffix(".parquet").ok_or_else(|| {
-        error::IllegalConfigSnafu {
-            msg: "expect .parquet file".to_string(),
-        }
-        .build()
-    })?;
-    let file_id = FileId::parse_str(uuid_str).map_err(|e| {
-        error::IllegalConfigSnafu {
-            msg: format!("invalid file id: {e}"),
-        }
-        .build()
-    })?;
-    let parent = p
-        .parent()
-        .and_then(|s| s.to_str())
-        .unwrap_or("")
-        .to_string();
-    Ok((parent, file_id))
-}
-
-fn extract_region_metadata(
-    file_path: &str,
-    meta: &parquet::file::metadata::ParquetMetaData,
-) -> Result<RegionMetadataRef> {
-    use parquet::format::KeyValue;
-    let kvs: Option<&Vec<KeyValue>> = meta.file_metadata().key_value_metadata();
-    let Some(kvs) = kvs else {
-        return Err(error::IllegalConfigSnafu {
-            msg: format!("{file_path}: missing parquet key_value metadata"),
-        }
-        .build());
-    };
-    let json = kvs
-        .iter()
-        .find(|kv| kv.key == PARQUET_METADATA_KEY)
-        .and_then(|kv| kv.value.as_ref())
-        .ok_or_else(|| {
-            error::IllegalConfigSnafu {
-                msg: format!("{file_path}: key {PARQUET_METADATA_KEY} not found or empty"),
-            }
-            .build()
-        })?;
-    let region: RegionMetadata = RegionMetadata::from_json(json).map_err(|e| {
-        error::IllegalConfigSnafu {
-            msg: format!("invalid region metadata json: {e}"),
-        }
-        .build()
-    })?;
-    Ok(std::sync::Arc::new(region))
-}
-
-async fn build_object_store(sc: &StorageConfig) -> Result<ObjectStore> {
-    use datanode::config::ObjectStoreConfig::*;
-    let oss = &sc.store;
-    match oss {
-        File(_) => {
-            use object_store::services::Fs;
-            let builder = Fs::default().root(&sc.data_home);
-            Ok(ObjectStore::new(builder)
-                .map_err(|e| {
-                    error::IllegalConfigSnafu {
-                        msg: format!("init fs backend failed: {e}"),
-                    }
-                    .build()
-                })?
-                .finish())
-        }
-        S3(s3) => {
-            use common_base::secrets::ExposeSecret;
-            use object_store::services::S3;
-            use object_store::util;
-            let root = util::normalize_dir(&s3.root);
-            let mut builder = S3::default()
-                .root(&root)
-                .bucket(&s3.bucket)
-                .access_key_id(s3.access_key_id.expose_secret())
-                .secret_access_key(s3.secret_access_key.expose_secret());
-            if let Some(ep) = &s3.endpoint {
-                builder = builder.endpoint(ep);
-            }
-            if let Some(region) = &s3.region {
-                builder = builder.region(region);
-            }
-            if s3.enable_virtual_host_style {
-                builder = builder.enable_virtual_host_style();
-            }
-            Ok(ObjectStore::new(builder)
-                .map_err(|e| {
-                    error::IllegalConfigSnafu {
-                        msg: format!("init s3 backend failed: {e}"),
-                    }
-                    .build()
-                })?
-                .finish())
-        }
-        Oss(oss) => {
-            use common_base::secrets::ExposeSecret;
-            use object_store::services::Oss;
-            use object_store::util;
-            let root = util::normalize_dir(&oss.root);
-            let builder = Oss::default()
-                .root(&root)
-                .bucket(&oss.bucket)
-                .endpoint(&oss.endpoint)
-                .access_key_id(oss.access_key_id.expose_secret())
-                .access_key_secret(oss.access_key_secret.expose_secret());
-            Ok(ObjectStore::new(builder)
-                .map_err(|e| {
-                    error::IllegalConfigSnafu {
-                        msg: format!("init oss backend failed: {e}"),
-                    }
-                    .build()
-                })?
-                .finish())
-        }
-        Azblob(az) => {
-            use common_base::secrets::ExposeSecret;
-            use object_store::services::Azblob;
-            use object_store::util;
-            let root = util::normalize_dir(&az.root);
-            let mut builder = Azblob::default()
-                .root(&root)
-                .container(&az.container)
-                .endpoint(&az.endpoint)
-                .account_name(az.account_name.expose_secret())
-                .account_key(az.account_key.expose_secret());
-            if let Some(token) = &az.sas_token {
-                builder = builder.sas_token(token);
-            }
-            Ok(ObjectStore::new(builder)
-                .map_err(|e| {
-                    error::IllegalConfigSnafu {
-                        msg: format!("init azblob backend failed: {e}"),
-                    }
-                    .build()
-                })?
-                .finish())
-        }
-        Gcs(gcs) => {
-            use common_base::secrets::ExposeSecret;
-            use object_store::services::Gcs;
-            use object_store::util;
-            let root = util::normalize_dir(&gcs.root);
-            let builder = Gcs::default()
-                .root(&root)
-                .bucket(&gcs.bucket)
-                .scope(&gcs.scope)
-                .credential_path(gcs.credential_path.expose_secret())
-                .credential(gcs.credential.expose_secret())
-                .endpoint(&gcs.endpoint);
-            Ok(ObjectStore::new(builder)
-                .map_err(|e| {
-                    error::IllegalConfigSnafu {
-                        msg: format!("init gcs backend failed: {e}"),
-                    }
-                    .build()
-                })?
-                .finish())
-        }
-    }
-}
-
-async fn build_access_layer_simple(
-    region_dir: String,
-    object_store: ObjectStore,
-) -> Result<(
-    std::sync::Arc<mito2::AccessLayer>,
-    std::sync::Arc<mito2::CacheManager>,
-)> {
-    // Minimal index aux path setup
-    let mut mito_cfg = MitoConfig::default();
-    // Use a temporary directory as aux path
-    let data_home = std::env::temp_dir().join("greptime_objbench");
-    let _ = std::fs::create_dir_all(&data_home);
-    let _ = mito_cfg.index.sanitize(
-        data_home.to_str().unwrap_or("/tmp"),
-        &mito_cfg.inverted_index,
-    );
-    let access_layer = build_access_layer(&region_dir, object_store, &mito_cfg)
-        .await
-        .map_err(|e| {
-            error::IllegalConfigSnafu {
-                msg: format!("build_access_layer failed: {e}"),
-            }
-            .build()
-        })?;
-    Ok((
-        access_layer,
-        std::sync::Arc::new(mito2::CacheManager::default()),
-    ))
-}
-
-fn new_noop_file_purger() -> FilePurgerRef {
-    #[derive(Debug)]
-    struct Noop;
-    impl FilePurger for Noop {
-        fn send_request(&self, _request: PurgeRequest) {}
-    }
-    std::sync::Arc::new(Noop)
-}
-
-async fn load_parquet_metadata(
-    object_store: ObjectStore,
-    path: &str,
-    file_size: u64,
-) -> std::result::Result<
-    parquet::file::metadata::ParquetMetaData,
-    Box<dyn std::error::Error + Send + Sync>,
-> {
-    use parquet::file::metadata::ParquetMetaDataReader;
-    use parquet::file::FOOTER_SIZE;
-    let actual_size = if file_size == 0 {
-        object_store.stat(path).await?.content_length()
-    } else {
-        file_size
-    };
-    if actual_size < FOOTER_SIZE as u64 {
-        return Err("file too small".into());
-    }
-    let prefetch: u64 = 64 * 1024;
-    let start = actual_size.saturating_sub(prefetch);
-    let buffer = object_store
-        .read_with(path)
-        .range(start..actual_size)
-        .await?
-        .to_vec();
-    let buffer_len = buffer.len();
-    let mut footer = [0; 8];
-    footer.copy_from_slice(&buffer[buffer_len - FOOTER_SIZE..]);
-    let metadata_len = ParquetMetaDataReader::decode_footer(&footer)? as u64;
-    if actual_size - (FOOTER_SIZE as u64) < metadata_len {
-        return Err("invalid footer/metadata length".into());
-    }
-    if (metadata_len as usize) <= buffer_len - FOOTER_SIZE {
-        let metadata_start = buffer_len - metadata_len as usize - FOOTER_SIZE;
-        let meta = ParquetMetaDataReader::decode_metadata(
-            &buffer[metadata_start..buffer_len - FOOTER_SIZE],
-        )?;
-        Ok(meta)
-    } else {
-        let metadata_start = actual_size - metadata_len - FOOTER_SIZE as u64;
-        let data = object_store
-            .read_with(path)
-            .range(metadata_start..(actual_size - FOOTER_SIZE as u64))
-            .await?
-            .to_vec();
-        let meta = ParquetMetaDataReader::decode_metadata(&data)?;
-        Ok(meta)
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::StorageConfigWrapper;
-
-    #[test]
-    fn test_decode() {
-        let cfg = std::fs::read_to_string("/home/lei/datanode-bulk.toml").unwrap();
-        let storage: StorageConfigWrapper = toml::from_str(&cfg).unwrap();
-        println!("{:?}", storage);
-    }
-}
--- a/src/common/function/Cargo.toml
+++ b/src/common/function/Cargo.toml
@@ -12,11 +12,9 @@ default = ["geo"]
 geo = ["geohash", "h3o", "s2", "wkt", "geo-types", "dep:geo"]

 [dependencies]
-ahash = "0.8"
 api.workspace = true
 arc-swap = "1.0"
 async-trait.workspace = true
-bincode = "1.3"
 common-base.workspace = true
 common-catalog.workspace = true
 common-error.workspace = true
@@ -34,7 +32,6 @@ geo = { version = "0.29", optional = true }
 geo-types = { version = "0.7", optional = true }
 geohash = { version = "0.13", optional = true }
 h3o = { version = "0.6", optional = true }
-hyperloglogplus = "0.4"
 jsonb.workspace = true
 nalgebra.workspace = true
 num = "0.4"
@@ -50,7 +47,6 @@ sql.workspace = true
 statrs = "0.16"
 store-api.workspace = true
 table.workspace = true
-uddsketch = { git = "https://github.com/GreptimeTeam/timescaledb-toolkit.git", rev = "84828fe8fb494a6a61412a3da96517fc80f7bb20" }
 wkt = { version = "0.11", optional = true }

 [dev-dependencies]
--- a/src/common/function/src/aggr.rs
+++ b/src/common/function/src/aggr.rs
@@ -1,20 +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.
-
-mod hll;
-mod uddsketch_state;
-
-pub(crate) use hll::HllStateType;
-pub use hll::{HllState, HLL_MERGE_NAME, HLL_NAME};
-pub use uddsketch_state::{UddSketchState, UDDSKETCH_STATE_NAME};
--- a/src/common/function/src/aggr/hll.rs
+++ b/src/common/function/src/aggr/hll.rs
@@ -1,319 +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_query::prelude::*;
-use common_telemetry::trace;
-use datafusion::arrow::array::ArrayRef;
-use datafusion::common::cast::{as_binary_array, as_string_array};
-use datafusion::common::not_impl_err;
-use datafusion::error::{DataFusionError, Result as DfResult};
-use datafusion::logical_expr::function::AccumulatorArgs;
-use datafusion::logical_expr::{Accumulator as DfAccumulator, AggregateUDF};
-use datafusion::prelude::create_udaf;
-use datatypes::arrow::datatypes::DataType;
-use hyperloglogplus::{HyperLogLog, HyperLogLogPlus};
-
-use crate::utils::FixedRandomState;
-
-pub const HLL_NAME: &str = "hll";
-pub const HLL_MERGE_NAME: &str = "hll_merge";
-
-const DEFAULT_PRECISION: u8 = 14;
-
-pub(crate) type HllStateType = HyperLogLogPlus<String, FixedRandomState>;
-
-pub struct HllState {
-    hll: HllStateType,
-}
-
-impl std::fmt::Debug for HllState {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "HllState<Opaque>")
-    }
-}
-
-impl Default for HllState {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl HllState {
-    pub fn new() -> Self {
-        Self {
-            // Safety: the DEFAULT_PRECISION is fixed and valid
-            hll: HllStateType::new(DEFAULT_PRECISION, FixedRandomState::new()).unwrap(),
-        }
-    }
-
-    /// Create a UDF for the `hll` function.
-    ///
-    /// `hll` accepts a string column and aggregates the
-    /// values into a HyperLogLog state.
-    pub fn state_udf_impl() -> AggregateUDF {
-        create_udaf(
-            HLL_NAME,
-            vec![DataType::Utf8],
-            Arc::new(DataType::Binary),
-            Volatility::Immutable,
-            Arc::new(Self::create_accumulator),
-            Arc::new(vec![DataType::Binary]),
-        )
-    }
-
-    /// Create a UDF for the `hll_merge` function.
-    ///
-    /// `hll_merge` accepts a binary column of states generated by `hll`
-    /// and merges them into a single state.
-    pub fn merge_udf_impl() -> AggregateUDF {
-        create_udaf(
-            HLL_MERGE_NAME,
-            vec![DataType::Binary],
-            Arc::new(DataType::Binary),
-            Volatility::Immutable,
-            Arc::new(Self::create_merge_accumulator),
-            Arc::new(vec![DataType::Binary]),
-        )
-    }
-
-    fn update(&mut self, value: &str) {
-        self.hll.insert(value);
-    }
-
-    fn merge(&mut self, raw: &[u8]) {
-        if let Ok(serialized) = bincode::deserialize::<HllStateType>(raw) {
-            if let Ok(()) = self.hll.merge(&serialized) {
-                return;
-            }
-        }
-        trace!("Warning: Failed to merge HyperLogLog from {:?}", raw);
-    }
-
-    fn create_accumulator(acc_args: AccumulatorArgs) -> DfResult<Box<dyn DfAccumulator>> {
-        let data_type = acc_args.exprs[0].data_type(acc_args.schema)?;
-
-        match data_type {
-            DataType::Utf8 => Ok(Box::new(HllState::new())),
-            other => not_impl_err!("{HLL_NAME} does not support data type: {other}"),
-        }
-    }
-
-    fn create_merge_accumulator(acc_args: AccumulatorArgs) -> DfResult<Box<dyn DfAccumulator>> {
-        let data_type = acc_args.exprs[0].data_type(acc_args.schema)?;
-
-        match data_type {
-            DataType::Binary => Ok(Box::new(HllState::new())),
-            other => not_impl_err!("{HLL_MERGE_NAME} does not support data type: {other}"),
-        }
-    }
-}
-
-impl DfAccumulator for HllState {
-    fn update_batch(&mut self, values: &[ArrayRef]) -> DfResult<()> {
-        let array = &values[0];
-
-        match array.data_type() {
-            DataType::Utf8 => {
-                let string_array = as_string_array(array)?;
-                for value in string_array.iter().flatten() {
-                    self.update(value);
-                }
-            }
-            DataType::Binary => {
-                let binary_array = as_binary_array(array)?;
-                for v in binary_array.iter().flatten() {
-                    self.merge(v);
-                }
-            }
-            _ => {
-                return not_impl_err!(
-                    "HLL functions do not support data type: {}",
-                    array.data_type()
-                )
-            }
-        }
-
-        Ok(())
-    }
-
-    fn evaluate(&mut self) -> DfResult<ScalarValue> {
-        Ok(ScalarValue::Binary(Some(
-            bincode::serialize(&self.hll).map_err(|e| {
-                DataFusionError::Internal(format!("Failed to serialize HyperLogLog: {}", e))
-            })?,
-        )))
-    }
-
-    fn size(&self) -> usize {
-        std::mem::size_of_val(&self.hll)
-    }
-
-    fn state(&mut self) -> DfResult<Vec<ScalarValue>> {
-        Ok(vec![ScalarValue::Binary(Some(
-            bincode::serialize(&self.hll).map_err(|e| {
-                DataFusionError::Internal(format!("Failed to serialize HyperLogLog: {}", e))
-            })?,
-        ))])
-    }
-
-    fn merge_batch(&mut self, states: &[ArrayRef]) -> DfResult<()> {
-        let array = &states[0];
-        let binary_array = as_binary_array(array)?;
-        for v in binary_array.iter().flatten() {
-            self.merge(v);
-        }
-
-        Ok(())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use datafusion::arrow::array::{BinaryArray, StringArray};
-
-    use super::*;
-
-    #[test]
-    fn test_hll_basic() {
-        let mut state = HllState::new();
-        state.update("1");
-        state.update("2");
-        state.update("3");
-
-        let result = state.evaluate().unwrap();
-        if let ScalarValue::Binary(Some(bytes)) = result {
-            let mut hll: HllStateType = bincode::deserialize(&bytes).unwrap();
-            assert_eq!(hll.count().trunc() as u32, 3);
-        } else {
-            panic!("Expected binary scalar value");
-        }
-    }
-
-    #[test]
-    fn test_hll_roundtrip() {
-        let mut state = HllState::new();
-        state.update("1");
-        state.update("2");
-
-        // Serialize
-        let serialized = state.evaluate().unwrap();
-
-        // Create new state and merge the serialized data
-        let mut new_state = HllState::new();
-        if let ScalarValue::Binary(Some(bytes)) = &serialized {
-            new_state.merge(bytes);
-
-            // Verify the merged state matches original
-            let result = new_state.evaluate().unwrap();
-            if let ScalarValue::Binary(Some(new_bytes)) = result {
-                let mut original: HllStateType = bincode::deserialize(bytes).unwrap();
-                let mut merged: HllStateType = bincode::deserialize(&new_bytes).unwrap();
-                assert_eq!(original.count(), merged.count());
-            } else {
-                panic!("Expected binary scalar value");
-            }
-        } else {
-            panic!("Expected binary scalar value");
-        }
-    }
-
-    #[test]
-    fn test_hll_batch_update() {
-        let mut state = HllState::new();
-
-        // Test string values
-        let str_values = vec!["a", "b", "c", "d", "e", "f", "g", "h", "i"];
-        let str_array = Arc::new(StringArray::from(str_values)) as ArrayRef;
-        state.update_batch(&[str_array]).unwrap();
-
-        let result = state.evaluate().unwrap();
-        if let ScalarValue::Binary(Some(bytes)) = result {
-            let mut hll: HllStateType = bincode::deserialize(&bytes).unwrap();
-            assert_eq!(hll.count().trunc() as u32, 9);
-        } else {
-            panic!("Expected binary scalar value");
-        }
-    }
-
-    #[test]
-    fn test_hll_merge_batch() {
-        let mut state1 = HllState::new();
-        state1.update("1");
-        let state1_binary = state1.evaluate().unwrap();
-
-        let mut state2 = HllState::new();
-        state2.update("2");
-        let state2_binary = state2.evaluate().unwrap();
-
-        let mut merged_state = HllState::new();
-        if let (ScalarValue::Binary(Some(bytes1)), ScalarValue::Binary(Some(bytes2))) =
-            (&state1_binary, &state2_binary)
-        {
-            let binary_array = Arc::new(BinaryArray::from(vec![
-                bytes1.as_slice(),
-                bytes2.as_slice(),
-            ])) as ArrayRef;
-            merged_state.merge_batch(&[binary_array]).unwrap();
-
-            let result = merged_state.evaluate().unwrap();
-            if let ScalarValue::Binary(Some(bytes)) = result {
-                let mut hll: HllStateType = bincode::deserialize(&bytes).unwrap();
-                assert_eq!(hll.count().trunc() as u32, 2);
-            } else {
-                panic!("Expected binary scalar value");
-            }
-        } else {
-            panic!("Expected binary scalar values");
-        }
-    }
-
-    #[test]
-    fn test_hll_merge_function() {
-        // Create two HLL states with different values
-        let mut state1 = HllState::new();
-        state1.update("1");
-        state1.update("2");
-        let state1_binary = state1.evaluate().unwrap();
-
-        let mut state2 = HllState::new();
-        state2.update("2");
-        state2.update("3");
-        let state2_binary = state2.evaluate().unwrap();
-
-        // Create a merge state and merge both states
-        let mut merge_state = HllState::new();
-        if let (ScalarValue::Binary(Some(bytes1)), ScalarValue::Binary(Some(bytes2))) =
-            (&state1_binary, &state2_binary)
-        {
-            let binary_array = Arc::new(BinaryArray::from(vec![
-                bytes1.as_slice(),
-                bytes2.as_slice(),
-            ])) as ArrayRef;
-            merge_state.update_batch(&[binary_array]).unwrap();
-
-            let result = merge_state.evaluate().unwrap();
-            if let ScalarValue::Binary(Some(bytes)) = result {
-                let mut hll: HllStateType = bincode::deserialize(&bytes).unwrap();
-                // Should have 3 unique values: "1", "2", "3"
-                assert_eq!(hll.count().trunc() as u32, 3);
-            } else {
-                panic!("Expected binary scalar value");
-            }
-        } else {
-            panic!("Expected binary scalar values");
-        }
-    }
-}
--- a/src/common/function/src/aggr/uddsketch_state.rs
+++ b/src/common/function/src/aggr/uddsketch_state.rs
@@ -1,307 +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_query::prelude::*;
-use common_telemetry::trace;
-use datafusion::common::cast::{as_binary_array, as_primitive_array};
-use datafusion::common::not_impl_err;
-use datafusion::error::{DataFusionError, Result as DfResult};
-use datafusion::logical_expr::function::AccumulatorArgs;
-use datafusion::logical_expr::{Accumulator as DfAccumulator, AggregateUDF};
-use datafusion::physical_plan::expressions::Literal;
-use datafusion::prelude::create_udaf;
-use datatypes::arrow::array::ArrayRef;
-use datatypes::arrow::datatypes::{DataType, Float64Type};
-use uddsketch::{SketchHashKey, UDDSketch};
-
-pub const UDDSKETCH_STATE_NAME: &str = "uddsketch_state";
-
-#[derive(Debug)]
-pub struct UddSketchState {
-    uddsketch: UDDSketch,
-}
-
-impl UddSketchState {
-    pub fn new(bucket_size: u64, error_rate: f64) -> Self {
-        Self {
-            uddsketch: UDDSketch::new(bucket_size, error_rate),
-        }
-    }
-
-    pub fn udf_impl() -> AggregateUDF {
-        create_udaf(
-            UDDSKETCH_STATE_NAME,
-            vec![DataType::Int64, DataType::Float64, DataType::Float64],
-            Arc::new(DataType::Binary),
-            Volatility::Immutable,
-            Arc::new(|args| {
-                let (bucket_size, error_rate) = downcast_accumulator_args(args)?;
-                Ok(Box::new(UddSketchState::new(bucket_size, error_rate)))
-            }),
-            Arc::new(vec![DataType::Binary]),
-        )
-    }
-
-    fn update(&mut self, value: f64) {
-        self.uddsketch.add_value(value);
-    }
-
-    fn merge(&mut self, raw: &[u8]) {
-        if let Ok(uddsketch) = bincode::deserialize::<UDDSketch>(raw) {
-            if uddsketch.count() != 0 {
-                self.uddsketch.merge_sketch(&uddsketch);
-            }
-        } else {
-            trace!("Warning: Failed to deserialize UDDSketch from {:?}", raw);
-        }
-    }
-}
-
-fn downcast_accumulator_args(args: AccumulatorArgs) -> DfResult<(u64, f64)> {
-    let bucket_size = match args.exprs[0]
-        .as_any()
-        .downcast_ref::<Literal>()
-        .map(|lit| lit.value())
-    {
-        Some(ScalarValue::Int64(Some(value))) => *value as u64,
-        _ => {
-            return not_impl_err!(
-                "{} not supported for bucket size: {}",
-                UDDSKETCH_STATE_NAME,
-                &args.exprs[0]
-            )
-        }
-    };
-
-    let error_rate = match args.exprs[1]
-        .as_any()
-        .downcast_ref::<Literal>()
-        .map(|lit| lit.value())
-    {
-        Some(ScalarValue::Float64(Some(value))) => *value,
-        _ => {
-            return not_impl_err!(
-                "{} not supported for error rate: {}",
-                UDDSKETCH_STATE_NAME,
-                &args.exprs[1]
-            )
-        }
-    };
-
-    Ok((bucket_size, error_rate))
-}
-
-impl DfAccumulator for UddSketchState {
-    fn update_batch(&mut self, values: &[ArrayRef]) -> DfResult<()> {
-        let array = &values[2]; // the third column is data value
-        let f64_array = as_primitive_array::<Float64Type>(array)?;
-        for v in f64_array.iter().flatten() {
-            self.update(v);
-        }
-
-        Ok(())
-    }
-
-    fn evaluate(&mut self) -> DfResult<ScalarValue> {
-        Ok(ScalarValue::Binary(Some(
-            bincode::serialize(&self.uddsketch).map_err(|e| {
-                DataFusionError::Internal(format!("Failed to serialize UDDSketch: {}", e))
-            })?,
-        )))
-    }
-
-    fn size(&self) -> usize {
-        // Base size of UDDSketch struct fields
-        let mut total_size = std::mem::size_of::<f64>() * 3 + // alpha, gamma, values_sum
-                            std::mem::size_of::<u32>() +      // compactions
-                            std::mem::size_of::<u64>() * 2; // max_buckets, num_values
-
-        // Size of buckets (SketchHashMap)
-        // Each bucket entry contains:
-        // - SketchHashKey (enum with i64/Zero/Invalid variants)
-        // - SketchHashEntry (count: u64, next: SketchHashKey)
-        let bucket_entry_size = std::mem::size_of::<SketchHashKey>() + // key
-                               std::mem::size_of::<u64>() +            // count
-                               std::mem::size_of::<SketchHashKey>(); // next
-
-        total_size += self.uddsketch.current_buckets_count() * bucket_entry_size;
-
-        total_size
-    }
-
-    fn state(&mut self) -> DfResult<Vec<ScalarValue>> {
-        Ok(vec![ScalarValue::Binary(Some(
-            bincode::serialize(&self.uddsketch).map_err(|e| {
-                DataFusionError::Internal(format!("Failed to serialize UDDSketch: {}", e))
-            })?,
-        ))])
-    }
-
-    fn merge_batch(&mut self, states: &[ArrayRef]) -> DfResult<()> {
-        let array = &states[0];
-        let binary_array = as_binary_array(array)?;
-        for v in binary_array.iter().flatten() {
-            self.merge(v);
-        }
-
-        Ok(())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use datafusion::arrow::array::{BinaryArray, Float64Array};
-
-    use super::*;
-
-    #[test]
-    fn test_uddsketch_state_basic() {
-        let mut state = UddSketchState::new(10, 0.01);
-        state.update(1.0);
-        state.update(2.0);
-        state.update(3.0);
-
-        let result = state.evaluate().unwrap();
-        if let ScalarValue::Binary(Some(bytes)) = result {
-            let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
-            assert_eq!(deserialized.count(), 3);
-        } else {
-            panic!("Expected binary scalar value");
-        }
-    }
-
-    #[test]
-    fn test_uddsketch_state_roundtrip() {
-        let mut state = UddSketchState::new(10, 0.01);
-        state.update(1.0);
-        state.update(2.0);
-
-        // Serialize
-        let serialized = state.evaluate().unwrap();
-
-        // Create new state and merge the serialized data
-        let mut new_state = UddSketchState::new(10, 0.01);
-        if let ScalarValue::Binary(Some(bytes)) = &serialized {
-            new_state.merge(bytes);
-
-            // Verify the merged state matches original by comparing deserialized values
-            let original_sketch: UDDSketch = bincode::deserialize(bytes).unwrap();
-            let new_result = new_state.evaluate().unwrap();
-            if let ScalarValue::Binary(Some(new_bytes)) = new_result {
-                let new_sketch: UDDSketch = bincode::deserialize(&new_bytes).unwrap();
-                assert_eq!(original_sketch.count(), new_sketch.count());
-                assert_eq!(original_sketch.sum(), new_sketch.sum());
-                assert_eq!(original_sketch.mean(), new_sketch.mean());
-                assert_eq!(original_sketch.max_error(), new_sketch.max_error());
-                // Compare a few quantiles to ensure statistical equivalence
-                for q in [0.1, 0.5, 0.9].iter() {
-                    assert!(
-                        (original_sketch.estimate_quantile(*q) - new_sketch.estimate_quantile(*q))
-                            .abs()
-                            < 1e-10,
-                        "Quantile {} mismatch: original={}, new={}",
-                        q,
-                        original_sketch.estimate_quantile(*q),
-                        new_sketch.estimate_quantile(*q)
-                    );
-                }
-            } else {
-                panic!("Expected binary scalar value");
-            }
-        } else {
-            panic!("Expected binary scalar value");
-        }
-    }
-
-    #[test]
-    fn test_uddsketch_state_batch_update() {
-        let mut state = UddSketchState::new(10, 0.01);
-        let values = vec![1.0f64, 2.0, 3.0];
-        let array = Arc::new(Float64Array::from(values)) as ArrayRef;
-
-        state
-            .update_batch(&[array.clone(), array.clone(), array])
-            .unwrap();
-
-        let result = state.evaluate().unwrap();
-        if let ScalarValue::Binary(Some(bytes)) = result {
-            let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
-            assert_eq!(deserialized.count(), 3);
-        } else {
-            panic!("Expected binary scalar value");
-        }
-    }
-
-    #[test]
-    fn test_uddsketch_state_merge_batch() {
-        let mut state1 = UddSketchState::new(10, 0.01);
-        state1.update(1.0);
-        let state1_binary = state1.evaluate().unwrap();
-
-        let mut state2 = UddSketchState::new(10, 0.01);
-        state2.update(2.0);
-        let state2_binary = state2.evaluate().unwrap();
-
-        let mut merged_state = UddSketchState::new(10, 0.01);
-        if let (ScalarValue::Binary(Some(bytes1)), ScalarValue::Binary(Some(bytes2))) =
-            (&state1_binary, &state2_binary)
-        {
-            let binary_array = Arc::new(BinaryArray::from(vec![
-                bytes1.as_slice(),
-                bytes2.as_slice(),
-            ])) as ArrayRef;
-            merged_state.merge_batch(&[binary_array]).unwrap();
-
-            let result = merged_state.evaluate().unwrap();
-            if let ScalarValue::Binary(Some(bytes)) = result {
-                let deserialized: UDDSketch = bincode::deserialize(&bytes).unwrap();
-                assert_eq!(deserialized.count(), 2);
-            } else {
-                panic!("Expected binary scalar value");
-            }
-        } else {
-            panic!("Expected binary scalar values");
-        }
-    }
-
-    #[test]
-    fn test_uddsketch_state_size() {
-        let mut state = UddSketchState::new(10, 0.01);
-        let initial_size = state.size();
-
-        // Add some values to create buckets
-        state.update(1.0);
-        state.update(2.0);
-        state.update(3.0);
-
-        let size_with_values = state.size();
-        assert!(
-            size_with_values > initial_size,
-            "Size should increase after adding values: initial={}, with_values={}",
-            initial_size,
-            size_with_values
-        );
-
-        // Verify size increases with more buckets
-        state.update(10.0); // This should create a new bucket
-        assert!(
-            state.size() > size_with_values,
-            "Size should increase after adding new bucket: prev={}, new={}",
-            size_with_values,
-            state.size()
-        );
-    }
-}
--- a/src/common/function/src/function_registry.rs
+++ b/src/common/function/src/function_registry.rs
@@ -22,12 +22,10 @@ use crate::function::{AsyncFunctionRef, FunctionRef};
 use crate::scalars::aggregate::{AggregateFunctionMetaRef, AggregateFunctions};
 use crate::scalars::date::DateFunction;
 use crate::scalars::expression::ExpressionFunction;
-use crate::scalars::hll_count::HllCalcFunction;
 use crate::scalars::json::JsonFunction;
 use crate::scalars::matches::MatchesFunction;
 use crate::scalars::math::MathFunction;
 use crate::scalars::timestamp::TimestampFunction;
-use crate::scalars::uddsketch_calc::UddSketchCalcFunction;
 use crate::scalars::vector::VectorFunction;
 use crate::system::SystemFunction;
 use crate::table::TableFunction;
@@ -107,8 +105,6 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
    TimestampFunction::register(&function_registry);
    DateFunction::register(&function_registry);
    ExpressionFunction::register(&function_registry);
-    UddSketchCalcFunction::register(&function_registry);
-    HllCalcFunction::register(&function_registry);

    // Aggregate functions
    AggregateFunctions::register(&function_registry);
--- a/src/common/function/src/lib.rs
+++ b/src/common/function/src/lib.rs
@@ -21,7 +21,6 @@ pub mod scalars;
 mod system;
 mod table;

-pub mod aggr;
 pub mod function;
 pub mod function_registry;
 pub mod handlers;
--- a/src/common/function/src/scalars.rs
+++ b/src/common/function/src/scalars.rs
@@ -22,9 +22,7 @@ pub mod matches;
 pub mod math;
 pub mod vector;

-pub(crate) mod hll_count;
 #[cfg(test)]
 pub(crate) mod test;
 pub(crate) mod timestamp;
-pub(crate) mod uddsketch_calc;
 pub mod udf;
--- a/src/common/function/src/scalars/hll_count.rs
+++ b/src/common/function/src/scalars/hll_count.rs
@@ -1,175 +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.
-
-//! Implementation of the scalar function `hll_count`.
-
-use std::fmt;
-use std::fmt::Display;
-use std::sync::Arc;
-
-use common_query::error::{DowncastVectorSnafu, InvalidFuncArgsSnafu, Result};
-use common_query::prelude::{Signature, Volatility};
-use datatypes::data_type::ConcreteDataType;
-use datatypes::prelude::Vector;
-use datatypes::scalars::{ScalarVector, ScalarVectorBuilder};
-use datatypes::vectors::{BinaryVector, MutableVector, UInt64VectorBuilder, VectorRef};
-use hyperloglogplus::HyperLogLog;
-use snafu::OptionExt;
-
-use crate::aggr::HllStateType;
-use crate::function::{Function, FunctionContext};
-use crate::function_registry::FunctionRegistry;
-
-const NAME: &str = "hll_count";
-
-/// HllCalcFunction implements the scalar function `hll_count`.
-///
-/// It accepts one argument:
-/// 1. The serialized HyperLogLogPlus state, as produced by the aggregator (binary).
-///
-/// For each row, it deserializes the sketch and returns the estimated cardinality.
-#[derive(Debug, Default)]
-pub struct HllCalcFunction;
-
-impl HllCalcFunction {
-    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(HllCalcFunction));
-    }
-}
-
-impl Display for HllCalcFunction {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "{}", NAME.to_ascii_uppercase())
-    }
-}
-
-impl Function for HllCalcFunction {
-    fn name(&self) -> &str {
-        NAME
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        // Only argument: HyperLogLogPlus state (binary)
-        Signature::exact(
-            vec![ConcreteDataType::binary_datatype()],
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        if columns.len() != 1 {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!("hll_count expects 1 argument, got {}", columns.len()),
-            }
-            .fail();
-        }
-
-        let hll_vec = columns[0]
-            .as_any()
-            .downcast_ref::<BinaryVector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!("expect BinaryVector, got {}", columns[0].vector_type_name()),
-            })?;
-        let len = hll_vec.len();
-        let mut builder = UInt64VectorBuilder::with_capacity(len);
-
-        for i in 0..len {
-            let hll_opt = hll_vec.get_data(i);
-
-            if hll_opt.is_none() {
-                builder.push_null();
-                continue;
-            }
-
-            let hll_bytes = hll_opt.unwrap();
-
-            // Deserialize the HyperLogLogPlus from its bincode representation
-            let mut hll: HllStateType = match bincode::deserialize(hll_bytes) {
-                Ok(h) => h,
-                Err(e) => {
-                    common_telemetry::trace!("Failed to deserialize HyperLogLogPlus: {}", e);
-                    builder.push_null();
-                    continue;
-                }
-            };
-
-            builder.push(Some(hll.count().round() as u64));
-        }
-
-        Ok(builder.to_vector())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use datatypes::vectors::BinaryVector;
-
-    use super::*;
-    use crate::utils::FixedRandomState;
-
-    #[test]
-    fn test_hll_count_function() {
-        let function = HllCalcFunction;
-        assert_eq!("hll_count", function.name());
-        assert_eq!(
-            ConcreteDataType::uint64_datatype(),
-            function
-                .return_type(&[ConcreteDataType::uint64_datatype()])
-                .unwrap()
-        );
-
-        // Create a test HLL
-        let mut hll = HllStateType::new(14, FixedRandomState::new()).unwrap();
-        for i in 1..=10 {
-            hll.insert(&i.to_string());
-        }
-
-        let serialized_bytes = bincode::serialize(&hll).unwrap();
-        let args: Vec<VectorRef> = vec![Arc::new(BinaryVector::from(vec![Some(serialized_bytes)]))];
-
-        let result = function.eval(FunctionContext::default(), &args).unwrap();
-        assert_eq!(result.len(), 1);
-
-        // Test cardinality estimate
-        if let datatypes::value::Value::UInt64(v) = result.get(0) {
-            assert_eq!(v, 10);
-        } else {
-            panic!("Expected uint64 value");
-        }
-    }
-
-    #[test]
-    fn test_hll_count_function_errors() {
-        let function = HllCalcFunction;
-
-        // Test with invalid number of arguments
-        let args: Vec<VectorRef> = vec![];
-        let result = function.eval(FunctionContext::default(), &args);
-        assert!(result.is_err());
-        assert!(result
-            .unwrap_err()
-            .to_string()
-            .contains("hll_count expects 1 argument"));
-
-        // Test with invalid binary data
-        let args: Vec<VectorRef> = vec![Arc::new(BinaryVector::from(vec![Some(vec![1, 2, 3])]))]; // Invalid binary data
-        let result = function.eval(FunctionContext::default(), &args).unwrap();
-        assert_eq!(result.len(), 1);
-        assert!(matches!(result.get(0), datatypes::value::Value::Null));
-    }
-}
--- a/src/common/function/src/scalars/uddsketch_calc.rs
+++ b/src/common/function/src/scalars/uddsketch_calc.rs
@@ -1,211 +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.
-
-//! Implementation of the scalar function `uddsketch_calc`.
-
-use std::fmt;
-use std::fmt::Display;
-use std::sync::Arc;
-
-use common_query::error::{DowncastVectorSnafu, InvalidFuncArgsSnafu, Result};
-use common_query::prelude::{Signature, Volatility};
-use datatypes::data_type::ConcreteDataType;
-use datatypes::prelude::Vector;
-use datatypes::scalars::{ScalarVector, ScalarVectorBuilder};
-use datatypes::vectors::{BinaryVector, Float64VectorBuilder, MutableVector, VectorRef};
-use snafu::OptionExt;
-use uddsketch::UDDSketch;
-
-use crate::function::{Function, FunctionContext};
-use crate::function_registry::FunctionRegistry;
-
-const NAME: &str = "uddsketch_calc";
-
-/// UddSketchCalcFunction implements the scalar function `uddsketch_calc`.
-///
-/// It accepts two arguments:
-/// 1. A percentile (as f64) for which to compute the estimated quantile (e.g. 0.95 for p95).
-/// 2. The serialized UDDSketch state, as produced by the aggregator (binary).
-///
-/// For each row, it deserializes the sketch and returns the computed quantile value.
-#[derive(Debug, Default)]
-pub struct UddSketchCalcFunction;
-
-impl UddSketchCalcFunction {
-    pub fn register(registry: &FunctionRegistry) {
-        registry.register(Arc::new(UddSketchCalcFunction));
-    }
-}
-
-impl Display for UddSketchCalcFunction {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        write!(f, "{}", NAME.to_ascii_uppercase())
-    }
-}
-
-impl Function for UddSketchCalcFunction {
-    fn name(&self) -> &str {
-        NAME
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::float64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        // First argument: percentile (float64)
-        // Second argument: UDDSketch state (binary)
-        Signature::exact(
-            vec![
-                ConcreteDataType::float64_datatype(),
-                ConcreteDataType::binary_datatype(),
-            ],
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        if columns.len() != 2 {
-            return InvalidFuncArgsSnafu {
-                err_msg: format!("uddsketch_calc expects 2 arguments, got {}", columns.len()),
-            }
-            .fail();
-        }
-
-        let perc_vec = &columns[0];
-        let sketch_vec = columns[1]
-            .as_any()
-            .downcast_ref::<BinaryVector>()
-            .with_context(|| DowncastVectorSnafu {
-                err_msg: format!("expect BinaryVector, got {}", columns[1].vector_type_name()),
-            })?;
-        let len = sketch_vec.len();
-        let mut builder = Float64VectorBuilder::with_capacity(len);
-
-        for i in 0..len {
-            let perc_opt = perc_vec.get(i).as_f64_lossy();
-            let sketch_opt = sketch_vec.get_data(i);
-
-            if sketch_opt.is_none() || perc_opt.is_none() {
-                builder.push_null();
-                continue;
-            }
-
-            let sketch_bytes = sketch_opt.unwrap();
-            let perc = perc_opt.unwrap();
-
-            // Deserialize the UDDSketch from its bincode representation
-            let sketch: UDDSketch = match bincode::deserialize(sketch_bytes) {
-                Ok(s) => s,
-                Err(e) => {
-                    common_telemetry::trace!("Failed to deserialize UDDSketch: {}", e);
-                    builder.push_null();
-                    continue;
-                }
-            };
-
-            // Compute the estimated quantile from the sketch
-            let result = sketch.estimate_quantile(perc);
-            builder.push(Some(result));
-        }
-
-        Ok(builder.to_vector())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use datatypes::vectors::{BinaryVector, Float64Vector};
-
-    use super::*;
-
-    #[test]
-    fn test_uddsketch_calc_function() {
-        let function = UddSketchCalcFunction;
-        assert_eq!("uddsketch_calc", function.name());
-        assert_eq!(
-            ConcreteDataType::float64_datatype(),
-            function
-                .return_type(&[ConcreteDataType::float64_datatype()])
-                .unwrap()
-        );
-
-        // Create a test sketch
-        let mut sketch = UDDSketch::new(128, 0.01);
-        sketch.add_value(10.0);
-        sketch.add_value(20.0);
-        sketch.add_value(30.0);
-        sketch.add_value(40.0);
-        sketch.add_value(50.0);
-        sketch.add_value(60.0);
-        sketch.add_value(70.0);
-        sketch.add_value(80.0);
-        sketch.add_value(90.0);
-        sketch.add_value(100.0);
-
-        // Get expected values directly from the sketch
-        let expected_p50 = sketch.estimate_quantile(0.5);
-        let expected_p90 = sketch.estimate_quantile(0.9);
-        let expected_p95 = sketch.estimate_quantile(0.95);
-
-        let serialized = bincode::serialize(&sketch).unwrap();
-        let percentiles = vec![0.5, 0.9, 0.95];
-
-        let args: Vec<VectorRef> = vec![
-            Arc::new(Float64Vector::from_vec(percentiles.clone())),
-            Arc::new(BinaryVector::from(vec![Some(serialized.clone()); 3])),
-        ];
-
-        let result = function.eval(FunctionContext::default(), &args).unwrap();
-        assert_eq!(result.len(), 3);
-
-        // Test median (p50)
-        assert!(
-            matches!(result.get(0), datatypes::value::Value::Float64(v) if (v - expected_p50).abs() < 1e-10)
-        );
-        // Test p90
-        assert!(
-            matches!(result.get(1), datatypes::value::Value::Float64(v) if (v - expected_p90).abs() < 1e-10)
-        );
-        // Test p95
-        assert!(
-            matches!(result.get(2), datatypes::value::Value::Float64(v) if (v - expected_p95).abs() < 1e-10)
-        );
-    }
-
-    #[test]
-    fn test_uddsketch_calc_function_errors() {
-        let function = UddSketchCalcFunction;
-
-        // Test with invalid number of arguments
-        let args: Vec<VectorRef> = vec![Arc::new(Float64Vector::from_vec(vec![0.95]))];
-        let result = function.eval(FunctionContext::default(), &args);
-        assert!(result.is_err());
-        assert!(result
-            .unwrap_err()
-            .to_string()
-            .contains("uddsketch_calc expects 2 arguments"));
-
-        // Test with invalid binary data
-        let args: Vec<VectorRef> = vec![
-            Arc::new(Float64Vector::from_vec(vec![0.95])),
-            Arc::new(BinaryVector::from(vec![Some(vec![1, 2, 3])])), // Invalid binary data
-        ];
-        let result = function.eval(FunctionContext::default(), &args).unwrap();
-        assert_eq!(result.len(), 1);
-        assert!(matches!(result.get(0), datatypes::value::Value::Null));
-    }
-}
--- a/src/common/function/src/utils.rs
+++ b/src/common/function/src/utils.rs
@@ -12,11 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use std::hash::BuildHasher;
-
-use ahash::RandomState;
-use serde::{Deserialize, Serialize};
-
 /// Escapes special characters in the provided pattern string for `LIKE`.
 ///
 /// Specifically, it prefixes the backslash (`\`), percent (`%`), and underscore (`_`)
@@ -37,71 +32,6 @@ pub fn escape_like_pattern(pattern: &str) -> String {
        })
        .collect::<String>()
 }
-
-/// A random state with fixed seeds.
-///
-/// This is used to ensure that the hash values are consistent across
-/// different processes, and easy to serialize and deserialize.
-#[derive(Debug)]
-pub struct FixedRandomState {
-    state: RandomState,
-}
-
-impl FixedRandomState {
-    // some random seeds
-    const RANDOM_SEED_0: u64 = 0x517cc1b727220a95;
-    const RANDOM_SEED_1: u64 = 0x428a2f98d728ae22;
-    const RANDOM_SEED_2: u64 = 0x7137449123ef65cd;
-    const RANDOM_SEED_3: u64 = 0xb5c0fbcfec4d3b2f;
-
-    pub fn new() -> Self {
-        Self {
-            state: ahash::RandomState::with_seeds(
-                Self::RANDOM_SEED_0,
-                Self::RANDOM_SEED_1,
-                Self::RANDOM_SEED_2,
-                Self::RANDOM_SEED_3,
-            ),
-        }
-    }
-}
-
-impl Default for FixedRandomState {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl BuildHasher for FixedRandomState {
-    type Hasher = ahash::AHasher;
-
-    fn build_hasher(&self) -> Self::Hasher {
-        self.state.build_hasher()
-    }
-
-    fn hash_one<T: std::hash::Hash>(&self, x: T) -> u64 {
-        self.state.hash_one(x)
-    }
-}
-
-impl Serialize for FixedRandomState {
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: serde::Serializer,
-    {
-        serializer.serialize_unit()
-    }
-}
-
-impl<'de> Deserialize<'de> for FixedRandomState {
-    fn deserialize<D>(_deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        Ok(Self::new())
-    }
-}
-
 #[cfg(test)]
 mod tests {
    use super::*;
--- a/src/common/meta/src/cache/flow/table_flownode.rs
+++ b/src/common/meta/src/cache/flow/table_flownode.rs
@@ -16,6 +16,7 @@ use std::collections::HashMap;
 use std::sync::Arc;

 use futures::future::BoxFuture;
+use futures::TryStreamExt;
 use moka::future::Cache;
 use moka::ops::compute::Op;
 use table::metadata::TableId;
@@ -53,13 +54,9 @@ fn init_factory(table_flow_manager: TableFlowManagerRef) -> Initializer<TableId,
        Box::pin(async move {
            table_flow_manager
                .flows(table_id)
+                .map_ok(|(key, value)| (key.flownode_id(), value.peer))
+                .try_collect::<HashMap<_, _>>()
                .await
-                .map(|flows| {
-                    flows
-                        .into_iter()
-                        .map(|(key, value)| (key.flownode_id(), value.peer))
-                        .collect::<HashMap<_, _>>()
-                })
                // We must cache the `HashSet` even if it's empty,
                // to avoid future requests to the remote storage next time;
                // If the value is added to the remote storage,
--- a/src/common/meta/src/cluster.rs
+++ b/src/common/meta/src/cluster.rs
@@ -57,10 +57,12 @@ pub trait ClusterInfo {
 }

 /// The key of [NodeInfo] in the storage. The format is `__meta_cluster_node_info-{cluster_id}-{role}-{node_id}`.
+///
+/// This key cannot be used to describe the `Metasrv` because the `Metasrv` does not have
+/// a `cluster_id`, it serves multiple clusters.
 #[derive(Debug, Clone, 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,
@@ -230,8 +232,8 @@ impl TryFrom<Vec<u8>> for NodeInfoKey {
    }
 }

-impl From<&NodeInfoKey> for Vec<u8> {
-    fn from(key: &NodeInfoKey) -> Self {
+impl From<NodeInfoKey> for Vec<u8> {
+    fn from(key: NodeInfoKey) -> Self {
        format!(
            "{}-{}-{}-{}",
            CLUSTER_NODE_INFO_PREFIX,
@@ -313,7 +315,7 @@ mod tests {
            node_id: 2,
        };

-        let key_bytes: Vec<u8> = (&key).into();
+        let key_bytes: Vec<u8> = key.into();
        let new_key: NodeInfoKey = key_bytes.try_into().unwrap();

        assert_eq!(1, new_key.cluster_id);
--- a/src/common/meta/src/ddl/create_flow.rs
+++ b/src/common/meta/src/ddl/create_flow.rs
@@ -15,7 +15,6 @@
 mod metadata;

 use std::collections::BTreeMap;
-use std::fmt;

 use api::v1::flow::flow_request::Body as PbFlowRequest;
 use api::v1::flow::{CreateRequest, FlowRequest, FlowRequestHeader};
@@ -29,6 +28,7 @@ use common_procedure::{
 use common_telemetry::info;
 use common_telemetry::tracing_context::TracingContext;
 use futures::future::join_all;
+use futures::TryStreamExt;
 use itertools::Itertools;
 use serde::{Deserialize, Serialize};
 use snafu::{ensure, ResultExt};
@@ -77,7 +77,6 @@ impl CreateFlowProcedure {
                query_context,
                state: CreateFlowState::Prepare,
                prev_flow_info_value: None,
-                flow_type: None,
            },
        }
    }
@@ -105,7 +104,7 @@ impl CreateFlowProcedure {
        if create_if_not_exists && or_replace {
            // this is forbidden because not clear what does that mean exactly
            return error::UnsupportedSnafu {
-                operation: "Create flow with both `IF NOT EXISTS` and `OR REPLACE`",
+                operation: "Create flow with both `IF NOT EXISTS` and `OR REPLACE`".to_string(),
            }
            .fail();
        }
@@ -130,10 +129,9 @@ impl CreateFlowProcedure {
                .flow_metadata_manager
                .flow_route_manager()
                .routes(flow_id)
-                .await?
-                .into_iter()
-                .map(|(_, value)| value.peer)
-                .collect::<Vec<_>>();
+                .map_ok(|(_, value)| value.peer)
+                .try_collect::<Vec<_>>()
+                .await?;
            self.data.flow_id = Some(flow_id);
            self.data.peers = peers;
            info!("Replacing flow, flow_id: {}", flow_id);
@@ -177,8 +175,6 @@ impl CreateFlowProcedure {
            self.allocate_flow_id().await?;
        }
        self.data.state = CreateFlowState::CreateFlows;
-        // determine flow type
-        self.data.flow_type = Some(determine_flow_type(&self.data.task));

        Ok(Status::executing(true))
    }
@@ -313,11 +309,6 @@ impl Procedure for CreateFlowProcedure {
    }
 }

-pub fn determine_flow_type(_flow_task: &CreateFlowTask) -> FlowType {
-    // TODO(discord9): determine flow type
-    FlowType::RecordingRule
-}
-
 /// The state of [CreateFlowProcedure].
 #[derive(Debug, Clone, Serialize, Deserialize, AsRefStr, PartialEq)]
 pub enum CreateFlowState {
@@ -331,35 +322,6 @@ pub enum CreateFlowState {
    CreateMetadata,
 }

-/// The type of flow.
-#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
-pub enum FlowType {
-    /// The flow is a recording rule task.
-    RecordingRule,
-    /// The flow is a streaming task.
-    Streaming,
-}
-
-impl FlowType {
-    pub const RECORDING_RULE: &str = "recording_rule";
-    pub const STREAMING: &str = "streaming";
-}
-
-impl Default for FlowType {
-    fn default() -> Self {
-        Self::RecordingRule
-    }
-}
-
-impl fmt::Display for FlowType {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        match self {
-            FlowType::RecordingRule => write!(f, "{}", FlowType::RECORDING_RULE),
-            FlowType::Streaming => write!(f, "{}", FlowType::STREAMING),
-        }
-    }
-}
-
 /// The serializable data.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct CreateFlowData {
@@ -373,7 +335,6 @@ pub struct CreateFlowData {
    /// For verify if prev value is consistent when need to update flow metadata.
    /// only set when `or_replace` is true.
    pub(crate) prev_flow_info_value: Option<DeserializedValueWithBytes<FlowInfoValue>>,
-    pub(crate) flow_type: Option<FlowType>,
 }

 impl From<&CreateFlowData> for CreateRequest {
@@ -381,7 +342,7 @@ impl From<&CreateFlowData> for CreateRequest {
        let flow_id = value.flow_id.unwrap();
        let source_table_ids = &value.source_table_ids;

-        let mut req = CreateRequest {
+        CreateRequest {
            flow_id: Some(api::v1::FlowId { id: flow_id }),
            source_table_ids: source_table_ids
                .iter()
@@ -395,11 +356,7 @@ impl From<&CreateFlowData> for CreateRequest {
            comment: value.task.comment.clone(),
            sql: value.task.sql.clone(),
            flow_options: value.task.flow_options.clone(),
-        };
-
-        let flow_type = value.flow_type.unwrap_or_default().to_string();
-        req.flow_options.insert("flow_type".to_string(), flow_type);
-        req
+        }
    }
 }

@@ -412,7 +369,7 @@ impl From<&CreateFlowData> for (FlowInfoValue, Vec<(FlowPartitionId, FlowRouteVa
            expire_after,
            comment,
            sql,
-            flow_options: mut options,
+            flow_options: options,
            ..
        } = value.task.clone();

@@ -429,21 +386,19 @@ impl From<&CreateFlowData> for (FlowInfoValue, Vec<(FlowPartitionId, FlowRouteVa
            .map(|(idx, peer)| (idx as u32, FlowRouteValue { peer: peer.clone() }))
            .collect::<Vec<_>>();

-        let flow_type = value.flow_type.unwrap_or_default().to_string();
-        options.insert("flow_type".to_string(), flow_type);
-
-        let flow_info = FlowInfoValue {
-            source_table_ids: value.source_table_ids.clone(),
-            sink_table_name,
-            flownode_ids,
-            catalog_name,
-            flow_name,
-            raw_sql: sql,
-            expire_after,
-            comment,
-            options,
-        };
-
-        (flow_info, flow_routes)
+        (
+            FlowInfoValue {
+                source_table_ids: value.source_table_ids.clone(),
+                sink_table_name,
+                flownode_ids,
+                catalog_name,
+                flow_name,
+                raw_sql: sql,
+                expire_after,
+                comment,
+                options,
+            },
+            flow_routes,
+        )
    }
 }
--- a/src/common/meta/src/ddl/drop_database/executor.rs
+++ b/src/common/meta/src/ddl/drop_database/executor.rs
@@ -128,7 +128,7 @@ impl State for DropDatabaseExecutor {
            .await?;
        executor.invalidate_table_cache(ddl_ctx).await?;
        executor
-            .on_drop_regions(ddl_ctx, &self.physical_region_routes, true)
+            .on_drop_regions(ddl_ctx, &self.physical_region_routes)
            .await?;
        info!("Table: {}({}) is dropped", self.table_name, self.table_id);

--- a/src/common/meta/src/ddl/drop_flow/metadata.rs
+++ b/src/common/meta/src/ddl/drop_flow/metadata.rs
@@ -13,6 +13,7 @@
 // limitations under the License.

 use common_catalog::format_full_flow_name;
+use futures::TryStreamExt;
 use snafu::{ensure, OptionExt};

 use crate::ddl::drop_flow::DropFlowProcedure;
@@ -38,10 +39,9 @@ impl DropFlowProcedure {
            .flow_metadata_manager
            .flow_route_manager()
            .routes(self.data.task.flow_id)
-            .await?
-            .into_iter()
-            .map(|(_, value)| value)
-            .collect::<Vec<_>>();
+            .map_ok(|(_, value)| value)
+            .try_collect::<Vec<_>>()
+            .await?;
        ensure!(
            !flow_route_values.is_empty(),
            error::FlowRouteNotFoundSnafu {
--- a/src/common/meta/src/ddl/drop_table.rs
+++ b/src/common/meta/src/ddl/drop_table.rs
@@ -156,7 +156,7 @@ impl DropTableProcedure {

    pub async fn on_datanode_drop_regions(&mut self) -> Result<Status> {
        self.executor
-            .on_drop_regions(&self.context, &self.data.physical_region_routes, false)
+            .on_drop_regions(&self.context, &self.data.physical_region_routes)
            .await?;
        self.data.state = DropTableState::DeleteTombstone;
        Ok(Status::executing(true))
--- a/src/common/meta/src/ddl/drop_table/executor.rs
+++ b/src/common/meta/src/ddl/drop_table/executor.rs
@@ -214,7 +214,6 @@ impl DropTableExecutor {
        &self,
        ctx: &DdlContext,
        region_routes: &[RegionRoute],
-        fast_path: bool,
    ) -> Result<()> {
        let leaders = find_leaders(region_routes);
        let mut drop_region_tasks = Vec::with_capacity(leaders.len());
@@ -237,7 +236,6 @@ impl DropTableExecutor {
                    }),
                    body: Some(region_request::Body::Drop(PbDropRegionRequest {
                        region_id: region_id.as_u64(),
-                        fast_path,
                    })),
                };
                let datanode = datanode.clone();
--- a/src/common/meta/src/key/flow.rs
+++ b/src/common/meta/src/key/flow.rs
@@ -16,9 +16,9 @@ pub mod flow_info;
 pub(crate) mod flow_name;
 pub(crate) mod flow_route;
 pub mod flow_state;
-mod flownode_addr_helper;
 pub(crate) mod flownode_flow;
 pub(crate) mod table_flow;
+
 use std::ops::Deref;
 use std::sync::Arc;

@@ -506,6 +506,7 @@ mod tests {
        let routes = flow_metadata_manager
            .flow_route_manager()
            .routes(flow_id)
+            .try_collect::<Vec<_>>()
            .await
            .unwrap();
        assert_eq!(
@@ -537,6 +538,7 @@ mod tests {
            let nodes = flow_metadata_manager
                .table_flow_manager()
                .flows(table_id)
+                .try_collect::<Vec<_>>()
                .await
                .unwrap();
            assert_eq!(
@@ -725,6 +727,7 @@ mod tests {
        let routes = flow_metadata_manager
            .flow_route_manager()
            .routes(flow_id)
+            .try_collect::<Vec<_>>()
            .await
            .unwrap();
        assert_eq!(
@@ -756,6 +759,7 @@ mod tests {
            let nodes = flow_metadata_manager
                .table_flow_manager()
                .flows(table_id)
+                .try_collect::<Vec<_>>()
                .await
                .unwrap();
            assert_eq!(
--- a/src/common/meta/src/key/flow/flow_route.rs
+++ b/src/common/meta/src/key/flow/flow_route.rs
@@ -12,15 +12,14 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use futures::TryStreamExt;
+use futures::stream::BoxStream;
 use lazy_static::lazy_static;
 use regex::Regex;
 use serde::{Deserialize, Serialize};
 use snafu::OptionExt;

 use crate::error::{self, Result};
-use crate::key::flow::{flownode_addr_helper, FlowScoped};
-use crate::key::node_address::NodeAddressKey;
+use crate::key::flow::FlowScoped;
 use crate::key::{BytesAdapter, FlowId, FlowPartitionId, MetadataKey, MetadataValue};
 use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
@@ -168,7 +167,10 @@ impl FlowRouteManager {
    }

    /// Retrieves all [FlowRouteValue]s of the specified `flow_id`.
-    pub async fn routes(&self, flow_id: FlowId) -> Result<Vec<(FlowRouteKey, FlowRouteValue)>> {
+    pub fn routes(
+        &self,
+        flow_id: FlowId,
+    ) -> BoxStream<'static, Result<(FlowRouteKey, FlowRouteValue)>> {
        let start_key = FlowRouteKey::range_start_key(flow_id);
        let req = RangeRequest::new().with_prefix(start_key);
        let stream = PaginationStream::new(
@@ -179,9 +181,7 @@ impl FlowRouteManager {
        )
        .into_stream();

-        let mut res = stream.try_collect::<Vec<_>>().await?;
-        self.remap_flow_route_addresses(&mut res).await?;
-        Ok(res)
+        Box::pin(stream)
    }

    /// Builds a create flow routes transaction.
@@ -203,28 +203,6 @@ impl FlowRouteManager {

        Ok(Txn::new().and_then(txns))
    }
-
-    async fn remap_flow_route_addresses(
-        &self,
-        flow_routes: &mut [(FlowRouteKey, FlowRouteValue)],
-    ) -> Result<()> {
-        let keys = flow_routes
-            .iter()
-            .map(|(_, value)| NodeAddressKey::with_flownode(value.peer.id))
-            .collect();
-        let flow_node_addrs =
-            flownode_addr_helper::get_flownode_addresses(&self.kv_backend, keys).await?;
-        for (_, flow_route_value) in flow_routes.iter_mut() {
-            let flownode_id = flow_route_value.peer.id;
-            // If an id lacks a corresponding address in the `flow_node_addrs`,
-            // it means the old address in `table_flow_value` is still valid,
-            // which is expected.
-            if let Some(node_addr) = flow_node_addrs.get(&flownode_id) {
-                flow_route_value.peer.addr = node_addr.peer.addr.clone();
-            }
-        }
-        Ok(())
-    }
 }

 #[cfg(test)]
--- a/src/common/meta/src/key/flow/flownode_addr_helper.rs
+++ b/src/common/meta/src/key/flow/flownode_addr_helper.rs
@@ -1,47 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use std::collections::HashMap;
-
-use crate::error::Result;
-use crate::key::node_address::{NodeAddressKey, NodeAddressValue};
-use crate::key::{MetadataKey, MetadataValue};
-use crate::kv_backend::KvBackendRef;
-use crate::rpc::store::BatchGetRequest;
-
-/// Get the addresses of the flownodes.
-/// The result is a map: node_id -> NodeAddressValue
-pub(crate) async fn get_flownode_addresses(
-    kv_backend: &KvBackendRef,
-    keys: Vec<NodeAddressKey>,
-) -> Result<HashMap<u64, NodeAddressValue>> {
-    if keys.is_empty() {
-        return Ok(HashMap::default());
-    }
-
-    let req = BatchGetRequest {
-        keys: keys.into_iter().map(|k| k.to_bytes()).collect(),
-    };
-    kv_backend
-        .batch_get(req)
-        .await?
-        .kvs
-        .into_iter()
-        .map(|kv| {
-            let key = NodeAddressKey::from_bytes(&kv.key)?;
-            let value = NodeAddressValue::try_from_raw_value(&kv.value)?;
-            Ok((key.node_id, value))
-        })
-        .collect()
-}
--- a/src/common/meta/src/key/flow/table_flow.rs
+++ b/src/common/meta/src/key/flow/table_flow.rs
@@ -14,7 +14,7 @@

 use std::sync::Arc;

-use futures::TryStreamExt;
+use futures::stream::BoxStream;
 use lazy_static::lazy_static;
 use regex::Regex;
 use serde::{Deserialize, Serialize};
@@ -22,8 +22,7 @@ use snafu::OptionExt;
 use table::metadata::TableId;

 use crate::error::{self, Result};
-use crate::key::flow::{flownode_addr_helper, FlowScoped};
-use crate::key::node_address::NodeAddressKey;
+use crate::key::flow::FlowScoped;
 use crate::key::{BytesAdapter, FlowId, FlowPartitionId, MetadataKey, MetadataValue};
 use crate::kv_backend::txn::{Txn, TxnOp};
 use crate::kv_backend::KvBackendRef;
@@ -197,7 +196,10 @@ impl TableFlowManager {
    /// Retrieves all [TableFlowKey]s of the specified `table_id`.
    ///
    /// TODO(discord9): add cache for it since range request does not support cache.
-    pub async fn flows(&self, table_id: TableId) -> Result<Vec<(TableFlowKey, TableFlowValue)>> {
+    pub fn flows(
+        &self,
+        table_id: TableId,
+    ) -> BoxStream<'static, Result<(TableFlowKey, TableFlowValue)>> {
        let start_key = TableFlowKey::range_start_key(table_id);
        let req = RangeRequest::new().with_prefix(start_key);
        let stream = PaginationStream::new(
@@ -208,9 +210,7 @@ impl TableFlowManager {
        )
        .into_stream();

-        let mut res = stream.try_collect::<Vec<_>>().await?;
-        self.remap_table_flow_addresses(&mut res).await?;
-        Ok(res)
+        Box::pin(stream)
    }

    /// Builds a create table flow transaction.
@@ -238,28 +238,6 @@ impl TableFlowManager {

        Ok(Txn::new().and_then(txns))
    }
-
-    async fn remap_table_flow_addresses(
-        &self,
-        table_flows: &mut [(TableFlowKey, TableFlowValue)],
-    ) -> Result<()> {
-        let keys = table_flows
-            .iter()
-            .map(|(_, value)| NodeAddressKey::with_flownode(value.peer.id))
-            .collect::<Vec<_>>();
-        let flownode_addrs =
-            flownode_addr_helper::get_flownode_addresses(&self.kv_backend, keys).await?;
-        for (_, table_flow_value) in table_flows.iter_mut() {
-            let flownode_id = table_flow_value.peer.id;
-            // If an id lacks a corresponding address in the `flow_node_addrs`,
-            // it means the old address in `table_flow_value` is still valid,
-            // which is expected.
-            if let Some(flownode_addr) = flownode_addrs.get(&flownode_id) {
-                table_flow_value.peer.addr = flownode_addr.peer.addr.clone();
-            }
-        }
-        Ok(())
-    }
 }

 #[cfg(test)]
--- a/src/common/meta/src/key/node_address.rs
+++ b/src/common/meta/src/key/node_address.rs
@@ -39,10 +39,6 @@ impl NodeAddressKey {
    pub fn with_datanode(node_id: u64) -> Self {
        Self::new(Role::Datanode, node_id)
    }
-
-    pub fn with_flownode(node_id: u64) -> Self {
-        Self::new(Role::Flownode, node_id)
-    }
 }

 #[derive(Debug, PartialEq, Serialize, Deserialize, Clone)]
--- a/src/common/meta/src/lib.rs
+++ b/src/common/meta/src/lib.rs
@@ -34,7 +34,6 @@ pub mod kv_backend;
 pub mod leadership_notifier;
 pub mod lock_key;
 pub mod metrics;
-pub mod node_expiry_listener;
 pub mod node_manager;
 pub mod peer;
 pub mod range_stream;
--- a/src/common/meta/src/node_expiry_listener.rs
+++ b/src/common/meta/src/node_expiry_listener.rs
@@ -1,152 +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::Mutex;
-use std::time::Duration;
-
-use common_telemetry::{debug, error, info, warn};
-use tokio::task::JoinHandle;
-use tokio::time::{interval, MissedTickBehavior};
-
-use crate::cluster::{NodeInfo, NodeInfoKey};
-use crate::error;
-use crate::kv_backend::ResettableKvBackendRef;
-use crate::leadership_notifier::LeadershipChangeListener;
-use crate::rpc::store::RangeRequest;
-use crate::rpc::KeyValue;
-
-/// [NodeExpiryListener] periodically checks all node info in memory and removes
-/// expired node info to prevent memory leak.
-pub struct NodeExpiryListener {
-    handle: Mutex<Option<JoinHandle<()>>>,
-    max_idle_time: Duration,
-    in_memory: ResettableKvBackendRef,
-}
-
-impl Drop for NodeExpiryListener {
-    fn drop(&mut self) {
-        self.stop();
-    }
-}
-
-impl NodeExpiryListener {
-    pub fn new(max_idle_time: Duration, in_memory: ResettableKvBackendRef) -> Self {
-        Self {
-            handle: Mutex::new(None),
-            max_idle_time,
-            in_memory,
-        }
-    }
-
-    async fn start(&self) {
-        let mut handle = self.handle.lock().unwrap();
-        if handle.is_none() {
-            let in_memory = self.in_memory.clone();
-
-            let max_idle_time = self.max_idle_time;
-            let ticker_loop = tokio::spawn(async move {
-                // Run clean task every minute.
-                let mut interval = interval(Duration::from_secs(60));
-                interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
-                loop {
-                    interval.tick().await;
-                    if let Err(e) = Self::clean_expired_nodes(&in_memory, max_idle_time).await {
-                        error!(e; "Failed to clean expired node");
-                    }
-                }
-            });
-            *handle = Some(ticker_loop);
-        }
-    }
-
-    fn stop(&self) {
-        if let Some(handle) = self.handle.lock().unwrap().take() {
-            handle.abort();
-            info!("Node expiry listener stopped")
-        }
-    }
-
-    /// Cleans expired nodes from memory.
-    async fn clean_expired_nodes(
-        in_memory: &ResettableKvBackendRef,
-        max_idle_time: Duration,
-    ) -> error::Result<()> {
-        let node_keys = Self::list_expired_nodes(in_memory, max_idle_time).await?;
-        for key in node_keys {
-            let key_bytes: Vec<u8> = (&key).into();
-            if let Err(e) = in_memory.delete(&key_bytes, false).await {
-                warn!(e; "Failed to delete expired node: {:?}", key_bytes);
-            } else {
-                debug!("Deleted expired node key: {:?}", key);
-            }
-        }
-        Ok(())
-    }
-
-    /// Lists expired nodes that have been inactive more than `max_idle_time`.
-    async fn list_expired_nodes(
-        in_memory: &ResettableKvBackendRef,
-        max_idle_time: Duration,
-    ) -> error::Result<impl Iterator<Item = NodeInfoKey>> {
-        let prefix = NodeInfoKey::key_prefix_with_cluster_id(0);
-        let req = RangeRequest::new().with_prefix(prefix);
-        let current_time_millis = common_time::util::current_time_millis();
-        let resp = in_memory.range(req).await?;
-        Ok(resp
-            .kvs
-            .into_iter()
-            .filter_map(move |KeyValue { key, value }| {
-                let Ok(info) = NodeInfo::try_from(value).inspect_err(|e| {
-                    warn!(e; "Unrecognized node info value");
-                }) else {
-                    return None;
-                };
-                if (current_time_millis - info.last_activity_ts) > max_idle_time.as_millis() as i64
-                {
-                    NodeInfoKey::try_from(key)
-                        .inspect_err(|e| {
-                            warn!(e; "Unrecognized node info key: {:?}", info.peer);
-                        })
-                        .ok()
-                        .inspect(|node_key| {
-                            debug!("Found expired node: {:?}", node_key);
-                        })
-                } else {
-                    None
-                }
-            }))
-    }
-}
-
-#[async_trait::async_trait]
-impl LeadershipChangeListener for NodeExpiryListener {
-    fn name(&self) -> &str {
-        "NodeExpiryListener"
-    }
-
-    async fn on_leader_start(&self) -> error::Result<()> {
-        self.start().await;
-        info!(
-            "On leader start, node expiry listener started with max idle time: {:?}",
-            self.max_idle_time
-        );
-        Ok(())
-    }
-
-    async fn on_leader_stop(&self) -> error::Result<()> {
-        self.stop();
-        info!("On leader stop, node expiry listener stopped");
-        Ok(())
-    }
-}
--- a/src/datanode/src/config.rs
+++ b/src/datanode/src/config.rs
@@ -171,10 +171,6 @@ pub struct S3Config {
    pub secret_access_key: SecretString,
    pub endpoint: Option<String>,
    pub region: Option<String>,
-    /// Enable virtual host style so that opendal will send API requests in virtual host style instead of path style.
-    /// By default, opendal will send API to https://s3.us-east-1.amazonaws.com/bucket_name
-    /// Enabled, opendal will send API to https://bucket_name.s3.us-east-1.amazonaws.com
-    pub enable_virtual_host_style: bool,
    #[serde(flatten)]
    pub cache: ObjectStorageCacheConfig,
    pub http_client: HttpClientConfig,
@@ -189,7 +185,6 @@ impl PartialEq for S3Config {
            && self.secret_access_key.expose_secret() == other.secret_access_key.expose_secret()
            && self.endpoint == other.endpoint
            && self.region == other.region
-            && self.enable_virtual_host_style == other.enable_virtual_host_style
            && self.cache == other.cache
            && self.http_client == other.http_client
    }
@@ -294,7 +289,6 @@ impl Default for S3Config {
            root: String::default(),
            access_key_id: SecretString::from(String::default()),
            secret_access_key: SecretString::from(String::default()),
-            enable_virtual_host_style: false,
            endpoint: Option::default(),
            region: Option::default(),
            cache: ObjectStorageCacheConfig::default(),
--- a/src/datanode/src/region_server.rs
+++ b/src/datanode/src/region_server.rs
@@ -1218,10 +1218,7 @@ mod tests {
        );

        let response = mock_region_server
-            .handle_request(
-                region_id,
-                RegionRequest::Drop(RegionDropRequest { fast_path: false }),
-            )
+            .handle_request(region_id, RegionRequest::Drop(RegionDropRequest {}))
            .await
            .unwrap();
        assert_eq!(response.affected_rows, 0);
@@ -1313,10 +1310,7 @@ mod tests {
            .insert(region_id, RegionEngineWithStatus::Ready(engine.clone()));

        mock_region_server
-            .handle_request(
-                region_id,
-                RegionRequest::Drop(RegionDropRequest { fast_path: false }),
-            )
+            .handle_request(region_id, RegionRequest::Drop(RegionDropRequest {}))
            .await
            .unwrap_err();

--- a/src/datanode/src/store/s3.rs
+++ b/src/datanode/src/store/s3.rs
@@ -41,13 +41,10 @@ pub(crate) async fn new_s3_object_store(s3_config: &S3Config) -> Result<ObjectSt

    if s3_config.endpoint.is_some() {
        builder = builder.endpoint(s3_config.endpoint.as_ref().unwrap());
-    }
+    };
    if s3_config.region.is_some() {
        builder = builder.region(s3_config.region.as_ref().unwrap());
-    }
-    if s3_config.enable_virtual_host_style {
-        builder = builder.enable_virtual_host_style();
-    }
+    };

    Ok(ObjectStore::new(builder)
        .context(error::InitBackendSnafu)?
--- a/src/datatypes/src/lib.rs
+++ b/src/datatypes/src/lib.rs
@@ -32,5 +32,5 @@ pub mod types;
 pub mod value;
 pub mod vectors;

-pub use arrow::{self, compute};
+pub use arrow;
 pub use error::{Error, Result};
--- a/src/flow/src/adapter/flownode_impl.rs
+++ b/src/flow/src/adapter/flownode_impl.rs
@@ -133,7 +133,7 @@ impl Flownode for FlowWorkerManager {
                    .map_err(to_meta_err(snafu::location!()))?;

                debug!(
-                    "Done to flush flow_id={:?} with {} input rows flushed, {} rows sent and {} output rows flushed",
+                    "Done to flush flow_id={:?} with {} input rows flushed, {} rows sended and {} output rows flushed",
                    flow_id, flushed_input_rows, rows_send, row
                );
                Ok(FlowResponse {
--- a/src/flow/src/heartbeat.rs
+++ b/src/flow/src/heartbeat.rs
@@ -60,12 +60,12 @@ async fn query_flow_state(
 #[derive(Clone)]
 pub struct HeartbeatTask {
    node_id: u64,
-    node_epoch: u64,
    peer_addr: String,
    meta_client: Arc<MetaClient>,
    report_interval: Duration,
    retry_interval: Duration,
    resp_handler_executor: HeartbeatResponseHandlerExecutorRef,
+    start_time_ms: u64,
    running: Arc<AtomicBool>,
    query_stat_size: Option<SizeReportSender>,
 }
@@ -83,12 +83,12 @@ impl HeartbeatTask {
    ) -> Self {
        Self {
            node_id: opts.node_id.unwrap_or(0),
-            node_epoch: common_time::util::current_time_millis() as u64,
            peer_addr: addrs::resolve_addr(&opts.grpc.bind_addr, Some(&opts.grpc.server_addr)),
            meta_client,
            report_interval: heartbeat_opts.interval,
            retry_interval: heartbeat_opts.retry_interval,
            resp_handler_executor,
+            start_time_ms: common_time::util::current_time_millis() as u64,
            running: Arc::new(AtomicBool::new(false)),
            query_stat_size: None,
        }
@@ -103,11 +103,6 @@ impl HeartbeatTask {
            warn!("Heartbeat task started multiple times");
            return Ok(());
        }
-
-        self.create_streams().await
-    }
-
-    async fn create_streams(&self) -> Result<(), Error> {
        info!("Start to establish the heartbeat connection to metasrv.");
        let (req_sender, resp_stream) = self
            .meta_client
@@ -186,7 +181,7 @@ impl HeartbeatTask {
        mut outgoing_rx: mpsc::Receiver<OutgoingMessage>,
    ) {
        let report_interval = self.report_interval;
-        let node_epoch = self.node_epoch;
+        let start_time_ms = self.start_time_ms;
        let self_peer = Some(Peer {
            id: self.node_id,
            addr: self.peer_addr.clone(),
@@ -203,8 +198,7 @@ impl HeartbeatTask {

            let heartbeat_request = HeartbeatRequest {
                peer: self_peer,
-                node_epoch,
-                info: Self::build_node_info(node_epoch),
+                info: Self::build_node_info(start_time_ms),
                ..Default::default()
            };

@@ -214,7 +208,6 @@ impl HeartbeatTask {
                        if let Some(message) = message {
                            Self::new_heartbeat_request(&heartbeat_request, Some(message), &latest_report)
                        } else {
-                            warn!("Sender has been dropped, exiting the heartbeat loop");
                            // Receives None that means Sender was dropped, we need to break the current loop
                            break
                        }
@@ -237,8 +230,6 @@ impl HeartbeatTask {
                // set the timeout to half of the report interval so that it wouldn't delay heartbeat if something went horribly wrong
                latest_report = query_flow_state(&query_stat_size, report_interval / 2).await;
            }
-
-            info!("flownode heartbeat task stopped.");
        });
    }

@@ -256,11 +247,7 @@ impl HeartbeatTask {
                            error!(e; "Error while handling heartbeat response");
                        }
                    }
-                    Ok(None) => {
-                        warn!("Heartbeat response stream closed");
-                        capture_self.start_with_retry(retry_interval).await;
-                        break;
-                    }
+                    Ok(None) => break,
                    Err(e) => {
                        error!(e; "Occur error while reading heartbeat response");
                        capture_self.start_with_retry(retry_interval).await;
@@ -286,7 +273,7 @@ impl HeartbeatTask {

            info!("Try to re-establish the heartbeat connection to metasrv.");

-            if self.create_streams().await.is_ok() {
+            if self.start().await.is_ok() {
                break;
            }
        }
--- a/src/frontend/src/heartbeat.rs
+++ b/src/frontend/src/heartbeat.rs
@@ -23,7 +23,7 @@ use common_meta::heartbeat::handler::{
 };
 use common_meta::heartbeat::mailbox::{HeartbeatMailbox, MailboxRef, OutgoingMessage};
 use common_meta::heartbeat::utils::outgoing_message_to_mailbox_message;
-use common_telemetry::{debug, error, info, warn};
+use common_telemetry::{debug, error, info};
 use meta_client::client::{HeartbeatSender, HeartbeatStream, MetaClient};
 use servers::addrs;
 use servers::heartbeat_options::HeartbeatOptions;
@@ -42,8 +42,8 @@ use crate::metrics::{HEARTBEAT_RECV_COUNT, HEARTBEAT_SENT_COUNT};
 pub struct HeartbeatTask {
    peer_addr: String,
    meta_client: Arc<MetaClient>,
-    report_interval: Duration,
-    retry_interval: Duration,
+    report_interval: u64,
+    retry_interval: u64,
    resp_handler_executor: HeartbeatResponseHandlerExecutorRef,
    start_time_ms: u64,
 }
@@ -58,8 +58,8 @@ impl HeartbeatTask {
        HeartbeatTask {
            peer_addr: addrs::resolve_addr(&opts.grpc.bind_addr, Some(&opts.grpc.server_addr)),
            meta_client,
-            report_interval: heartbeat_opts.interval,
-            retry_interval: heartbeat_opts.retry_interval,
+            report_interval: heartbeat_opts.interval.as_millis() as u64,
+            retry_interval: heartbeat_opts.retry_interval.as_millis() as u64,
            resp_handler_executor,
            start_time_ms: common_time::util::current_time_millis() as u64,
        }
@@ -103,15 +103,13 @@ impl HeartbeatTask {
                            HEARTBEAT_RECV_COUNT.with_label_values(&["success"]).inc();
                        }
                    }
-                    Ok(None) => {
-                        warn!("Heartbeat response stream closed");
-                        capture_self.start_with_retry(retry_interval).await;
-                        break;
-                    }
+                    Ok(None) => break,
                    Err(e) => {
                        HEARTBEAT_RECV_COUNT.with_label_values(&["error"]).inc();
                        error!(e; "Occur error while reading heartbeat response");
-                        capture_self.start_with_retry(retry_interval).await;
+                        capture_self
+                            .start_with_retry(Duration::from_millis(retry_interval))
+                            .await;

                        break;
                    }
@@ -179,13 +177,12 @@ impl HeartbeatTask {
                        if let Some(message) = message {
                            Self::new_heartbeat_request(&heartbeat_request, Some(message))
                        } else {
-                            warn!("Sender has been dropped, exiting the heartbeat loop");
                            // Receives None that means Sender was dropped, we need to break the current loop
                            break
                        }
                    }
                    _ = &mut sleep => {
-                        sleep.as_mut().reset(Instant::now() + report_interval);
+                        sleep.as_mut().reset(Instant::now() + Duration::from_millis(report_interval));
                       Self::new_heartbeat_request(&heartbeat_request, None)
                    }
                };
--- a/src/frontend/src/instance.rs
+++ b/src/frontend/src/instance.rs
@@ -237,13 +237,6 @@ impl Instance {

        let output = match stmt {
            Statement::Query(_) | Statement::Explain(_) | Statement::Delete(_) => {
-                // TODO: remove this when format is supported in datafusion
-                if let Statement::Explain(explain) = &stmt {
-                    if let Some(format) = explain.format() {
-                        query_ctx.set_explain_format(format.to_string());
-                    }
-                }
-
                let stmt = QueryStatement::Sql(stmt);
                let plan = self
                    .statement_executor
--- a/src/index/src/bloom_filter/applier.rs
+++ b/src/index/src/bloom_filter/applier.rs
@@ -42,16 +42,7 @@ impl BloomFilterApplier {
    ) -> Result<Vec<Range<usize>>> {
        let rows_per_segment = self.meta.rows_per_segment as usize;
        let start_seg = search_range.start / rows_per_segment;
-        let mut end_seg = search_range.end.div_ceil(rows_per_segment);
-
-        if end_seg == self.meta.segment_loc_indices.len() + 1 {
-            // In a previous version, there was a bug where if the last segment was all null,
-            // this segment would not be written into the index. This caused the slice
-            // `self.meta.segment_loc_indices[start_seg..end_seg]` to go out of bounds due to
-            // the missing segment. Since the `search` function does not search for nulls,
-            // we can simply ignore the last segment in this buggy scenario.
-            end_seg -= 1;
-        }
+        let end_seg = search_range.end.div_ceil(rows_per_segment);

        let locs = &self.meta.segment_loc_indices[start_seg..end_seg];

--- a/src/index/src/bloom_filter/creator.rs
+++ b/src/index/src/bloom_filter/creator.rs
@@ -64,9 +64,6 @@ pub struct BloomFilterCreator {
    /// Storage for finalized Bloom filters.
    finalized_bloom_filters: FinalizedBloomFilterStorage,

-    /// Row count that finalized so far.
-    finalized_row_count: usize,
-
    /// Global memory usage of the bloom filter creator.
    global_memory_usage: Arc<AtomicUsize>,
 }
@@ -99,7 +96,6 @@ impl BloomFilterCreator {
                global_memory_usage,
                global_memory_usage_threshold,
            ),
-            finalized_row_count: 0,
        }
    }

@@ -140,7 +136,6 @@ impl BloomFilterCreator {

            if self.accumulated_row_count % self.rows_per_segment == 0 {
                self.finalize_segment().await?;
-                self.finalized_row_count = self.accumulated_row_count;
            }
        }

@@ -166,7 +161,6 @@ impl BloomFilterCreator {

        if self.accumulated_row_count % self.rows_per_segment == 0 {
            self.finalize_segment().await?;
-            self.finalized_row_count = self.accumulated_row_count;
        }

        Ok(())
@@ -174,7 +168,7 @@ impl BloomFilterCreator {

    /// Finalizes any remaining segments and writes the bloom filters and metadata to the provided writer.
    pub async fn finish(&mut self, mut writer: impl AsyncWrite + Unpin) -> Result<()> {
-        if self.accumulated_row_count > self.finalized_row_count {
+        if !self.cur_seg_distinct_elems.is_empty() {
            self.finalize_segment().await?;
        }

@@ -412,35 +406,4 @@ mod tests {
            assert!(bf.contains(&b"f"));
        }
    }
-
-    #[tokio::test]
-    async fn test_final_seg_all_null() {
-        let mut writer = Cursor::new(Vec::new());
-        let mut creator = BloomFilterCreator::new(
-            2,
-            Arc::new(MockExternalTempFileProvider::new()),
-            Arc::new(AtomicUsize::new(0)),
-            None,
-        );
-
-        creator
-            .push_n_row_elems(4, vec![b"a".to_vec(), b"b".to_vec()])
-            .await
-            .unwrap();
-        creator.push_row_elems(Vec::new()).await.unwrap();
-
-        creator.finish(&mut writer).await.unwrap();
-
-        let bytes = writer.into_inner();
-        let total_size = bytes.len();
-        let meta_size_offset = total_size - 4;
-        let meta_size = u32::from_le_bytes((&bytes[meta_size_offset..]).try_into().unwrap());
-
-        let meta_bytes = &bytes[total_size - meta_size as usize - 4..total_size - 4];
-        let meta = BloomFilterMeta::decode(meta_bytes).unwrap();
-
-        assert_eq!(meta.rows_per_segment, 2);
-        assert_eq!(meta.segment_count, 3);
-        assert_eq!(meta.row_count, 5);
-    }
 }
--- a/src/index/src/fulltext_index/tests.rs
+++ b/src/index/src/fulltext_index/tests.rs
@@ -25,7 +25,7 @@ use crate::fulltext_index::create::{FulltextIndexCreator, TantivyFulltextIndexCr
 use crate::fulltext_index::search::{FulltextIndexSearcher, RowId, TantivyFulltextIndexSearcher};
 use crate::fulltext_index::{Analyzer, Config};

-async fn new_bounded_stager(prefix: &str) -> (TempDir, Arc<BoundedStager<String>>) {
+async fn new_bounded_stager(prefix: &str) -> (TempDir, Arc<BoundedStager>) {
    let staging_dir = create_temp_dir(prefix);
    let path = staging_dir.path().to_path_buf();
    (
@@ -68,13 +68,13 @@ async fn test_search(
    let file_accessor = Arc::new(MockFileAccessor::new(prefix));
    let puffin_manager = FsPuffinManager::new(stager, file_accessor);

-    let file_name = "fulltext_index".to_string();
-    let blob_key = "fulltext_index".to_string();
-    let mut writer = puffin_manager.writer(&file_name).await.unwrap();
-    create_index(prefix, &mut writer, &blob_key, texts, config).await;
+    let file_name = "fulltext_index";
+    let blob_key = "fulltext_index";
+    let mut writer = puffin_manager.writer(file_name).await.unwrap();
+    create_index(prefix, &mut writer, blob_key, texts, config).await;

-    let reader = puffin_manager.reader(&file_name).await.unwrap();
-    let index_dir = reader.dir(&blob_key).await.unwrap();
+    let reader = puffin_manager.reader(file_name).await.unwrap();
+    let index_dir = reader.dir(blob_key).await.unwrap();
    let searcher = TantivyFulltextIndexSearcher::new(index_dir.path()).unwrap();
    let results = searcher.search(query).await.unwrap();

--- a/src/meta-client/src/client/heartbeat.rs
+++ b/src/meta-client/src/client/heartbeat.rs
@@ -198,13 +198,13 @@ impl Inner {
            }
        );

-        let leader_addr = self
+        let leader = self
            .ask_leader
            .as_ref()
            .unwrap()
            .get_leader()
            .context(error::NoLeaderSnafu)?;
-        let mut leader = self.make_client(&leader_addr)?;
+        let mut leader = self.make_client(leader)?;

        let (sender, receiver) = mpsc::channel::<HeartbeatRequest>(128);

@@ -236,11 +236,7 @@ impl Inner {
            .await
            .map_err(error::Error::from)?
            .context(error::CreateHeartbeatStreamSnafu)?;
-
-        info!(
-            "Success to create heartbeat stream to server: {}, response: {:#?}",
-            leader_addr, res
-        );
+        info!("Success to create heartbeat stream to server: {:#?}", res);

        Ok((
            HeartbeatSender::new(self.id, self.role, sender),
--- a/src/meta-srv/src/handler.rs
+++ b/src/meta-srv/src/handler.rs
@@ -44,7 +44,6 @@ use mailbox_handler::MailboxHandler;
 use on_leader_start_handler::OnLeaderStartHandler;
 use publish_heartbeat_handler::PublishHeartbeatHandler;
 use region_lease_handler::RegionLeaseHandler;
-use remap_flow_peer_handler::RemapFlowPeerHandler;
 use response_header_handler::ResponseHeaderHandler;
 use snafu::{OptionExt, ResultExt};
 use store_api::storage::RegionId;
@@ -72,7 +71,6 @@ pub mod mailbox_handler;
 pub mod on_leader_start_handler;
 pub mod publish_heartbeat_handler;
 pub mod region_lease_handler;
-pub mod remap_flow_peer_handler;
 pub mod response_header_handler;

 #[async_trait::async_trait]
@@ -575,7 +573,6 @@ impl HeartbeatHandlerGroupBuilder {
            self.add_handler_last(publish_heartbeat_handler);
        }
        self.add_handler_last(CollectStatsHandler::new(self.flush_stats_factor));
-        self.add_handler_last(RemapFlowPeerHandler::default());

        if let Some(flow_state_handler) = self.flow_state_handler.take() {
            self.add_handler_last(flow_state_handler);
@@ -856,7 +853,7 @@ mod tests {
            .unwrap();

        let handlers = group.handlers;
-        assert_eq!(13, handlers.len());
+        assert_eq!(12, handlers.len());

        let names = [
            "ResponseHeaderHandler",
@@ -871,7 +868,6 @@ mod tests {
            "MailboxHandler",
            "FilterInactiveRegionStatsHandler",
            "CollectStatsHandler",
-            "RemapFlowPeerHandler",
        ];

        for (handler, name) in handlers.iter().zip(names.into_iter()) {
@@ -892,7 +888,7 @@ mod tests {

        let group = builder.build().unwrap();
        let handlers = group.handlers;
-        assert_eq!(14, handlers.len());
+        assert_eq!(13, handlers.len());

        let names = [
            "ResponseHeaderHandler",
@@ -908,7 +904,6 @@ mod tests {
            "CollectStatsHandler",
            "FilterInactiveRegionStatsHandler",
            "CollectStatsHandler",
-            "RemapFlowPeerHandler",
        ];

        for (handler, name) in handlers.iter().zip(names.into_iter()) {
@@ -926,7 +921,7 @@ mod tests {

        let group = builder.build().unwrap();
        let handlers = group.handlers;
-        assert_eq!(14, handlers.len());
+        assert_eq!(13, handlers.len());

        let names = [
            "CollectStatsHandler",
@@ -942,7 +937,6 @@ mod tests {
            "MailboxHandler",
            "FilterInactiveRegionStatsHandler",
            "CollectStatsHandler",
-            "RemapFlowPeerHandler",
        ];

        for (handler, name) in handlers.iter().zip(names.into_iter()) {
@@ -960,7 +954,7 @@ mod tests {

        let group = builder.build().unwrap();
        let handlers = group.handlers;
-        assert_eq!(14, handlers.len());
+        assert_eq!(13, handlers.len());

        let names = [
            "ResponseHeaderHandler",
@@ -976,7 +970,6 @@ mod tests {
            "CollectStatsHandler",
            "FilterInactiveRegionStatsHandler",
            "CollectStatsHandler",
-            "RemapFlowPeerHandler",
        ];

        for (handler, name) in handlers.iter().zip(names.into_iter()) {
@@ -994,7 +987,7 @@ mod tests {

        let group = builder.build().unwrap();
        let handlers = group.handlers;
-        assert_eq!(14, handlers.len());
+        assert_eq!(13, handlers.len());

        let names = [
            "ResponseHeaderHandler",
@@ -1010,7 +1003,6 @@ mod tests {
            "FilterInactiveRegionStatsHandler",
            "CollectStatsHandler",
            "ResponseHeaderHandler",
-            "RemapFlowPeerHandler",
        ];

        for (handler, name) in handlers.iter().zip(names.into_iter()) {
@@ -1028,7 +1020,7 @@ mod tests {

        let group = builder.build().unwrap();
        let handlers = group.handlers;
-        assert_eq!(13, handlers.len());
+        assert_eq!(12, handlers.len());

        let names = [
            "ResponseHeaderHandler",
@@ -1043,7 +1035,6 @@ mod tests {
            "CollectStatsHandler",
            "FilterInactiveRegionStatsHandler",
            "CollectStatsHandler",
-            "RemapFlowPeerHandler",
        ];

        for (handler, name) in handlers.iter().zip(names.into_iter()) {
@@ -1061,7 +1052,7 @@ mod tests {

        let group = builder.build().unwrap();
        let handlers = group.handlers;
-        assert_eq!(13, handlers.len());
+        assert_eq!(12, handlers.len());

        let names = [
            "ResponseHeaderHandler",
@@ -1076,7 +1067,6 @@ mod tests {
            "MailboxHandler",
            "FilterInactiveRegionStatsHandler",
            "ResponseHeaderHandler",
-            "RemapFlowPeerHandler",
        ];

        for (handler, name) in handlers.iter().zip(names.into_iter()) {
@@ -1094,7 +1084,7 @@ mod tests {

        let group = builder.build().unwrap();
        let handlers = group.handlers;
-        assert_eq!(13, handlers.len());
+        assert_eq!(12, handlers.len());

        let names = [
            "CollectStatsHandler",
@@ -1109,7 +1099,6 @@ mod tests {
            "MailboxHandler",
            "FilterInactiveRegionStatsHandler",
            "CollectStatsHandler",
-            "RemapFlowPeerHandler",
        ];

        for (handler, name) in handlers.iter().zip(names.into_iter()) {
--- a/src/meta-srv/src/handler/check_leader_handler.rs
+++ b/src/meta-srv/src/handler/check_leader_handler.rs
@@ -23,8 +23,8 @@ pub struct CheckLeaderHandler;

 #[async_trait::async_trait]
 impl HeartbeatHandler for CheckLeaderHandler {
-    fn is_acceptable(&self, _role: Role) -> bool {
-        true
+    fn is_acceptable(&self, role: Role) -> bool {
+        role == Role::Datanode
    }

    async fn handle(
--- a/src/meta-srv/src/handler/collect_cluster_info_handler.rs
+++ b/src/meta-srv/src/handler/collect_cluster_info_handler.rs
@@ -157,7 +157,7 @@ fn extract_base_info(request: &HeartbeatRequest) -> Option<(NodeInfoKey, Peer, P
 }

 async fn put_into_memory_store(ctx: &mut Context, key: NodeInfoKey, value: NodeInfo) -> Result<()> {
-    let key = (&key).into();
+    let key = key.into();
    let value = value.try_into().context(InvalidClusterInfoFormatSnafu)?;
    let put_req = PutRequest {
        key,
--- a/src/meta-srv/src/handler/collect_stats_handler.rs
+++ b/src/meta-srv/src/handler/collect_stats_handler.rs
@@ -21,7 +21,7 @@ use common_meta::key::node_address::{NodeAddressKey, NodeAddressValue};
 use common_meta::key::{MetadataKey, MetadataValue};
 use common_meta::peer::Peer;
 use common_meta::rpc::store::PutRequest;
-use common_telemetry::{error, info, warn};
+use common_telemetry::{error, warn};
 use dashmap::DashMap;
 use snafu::ResultExt;

@@ -185,10 +185,6 @@ async fn rewrite_node_address(ctx: &mut Context, stat: &Stat) {

        match ctx.leader_cached_kv_backend.put(put).await {
            Ok(_) => {
-                info!(
-                    "Successfully updated datanode `NodeAddressValue`: {:?}",
-                    peer
-                );
                // broadcast invalidating cache
                let cache_idents = stat
                    .table_ids()
@@ -204,14 +200,11 @@ async fn rewrite_node_address(ctx: &mut Context, stat: &Stat) {
                }
            }
            Err(e) => {
-                error!(e; "Failed to update datanode `NodeAddressValue`: {:?}", peer);
+                error!(e; "Failed to update NodeAddressValue: {:?}", peer);
            }
        }
    } else {
-        warn!(
-            "Failed to serialize datanode `NodeAddressValue`: {:?}",
-            peer
-        );
+        warn!("Failed to serialize NodeAddressValue: {:?}", peer);
    }
 }

--- a/src/meta-srv/src/handler/remap_flow_peer_handler.rs
+++ b/src/meta-srv/src/handler/remap_flow_peer_handler.rs
@@ -1,92 +0,0 @@
-// Copyright 2023 Greptime Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-use api::v1::meta::{HeartbeatRequest, Peer, Role};
-use common_meta::key::node_address::{NodeAddressKey, NodeAddressValue};
-use common_meta::key::{MetadataKey, MetadataValue};
-use common_meta::rpc::store::PutRequest;
-use common_telemetry::{error, info, warn};
-use dashmap::DashMap;
-
-use crate::handler::{HandleControl, HeartbeatAccumulator, HeartbeatHandler};
-use crate::metasrv::Context;
-use crate::Result;
-
-#[derive(Debug, Default)]
-pub struct RemapFlowPeerHandler {
-    /// flow_node_id -> epoch
-    epoch_cache: DashMap<u64, u64>,
-}
-
-#[async_trait::async_trait]
-impl HeartbeatHandler for RemapFlowPeerHandler {
-    fn is_acceptable(&self, role: Role) -> bool {
-        role == Role::Flownode
-    }
-
-    async fn handle(
-        &self,
-        req: &HeartbeatRequest,
-        ctx: &mut Context,
-        _acc: &mut HeartbeatAccumulator,
-    ) -> Result<HandleControl> {
-        let Some(peer) = req.peer.as_ref() else {
-            return Ok(HandleControl::Continue);
-        };
-
-        let current_epoch = req.node_epoch;
-        let flow_node_id = peer.id;
-
-        let refresh = if let Some(mut epoch) = self.epoch_cache.get_mut(&flow_node_id) {
-            if current_epoch > *epoch.value() {
-                *epoch.value_mut() = current_epoch;
-                true
-            } else {
-                false
-            }
-        } else {
-            self.epoch_cache.insert(flow_node_id, current_epoch);
-            true
-        };
-
-        if refresh {
-            rewrite_node_address(ctx, peer).await;
-        }
-
-        Ok(HandleControl::Continue)
-    }
-}
-
-async fn rewrite_node_address(ctx: &mut Context, peer: &Peer) {
-    let key = NodeAddressKey::with_flownode(peer.id).to_bytes();
-    if let Ok(value) = NodeAddressValue::new(peer.clone().into()).try_as_raw_value() {
-        let put = PutRequest {
-            key,
-            value,
-            prev_kv: false,
-        };
-
-        match ctx.leader_cached_kv_backend.put(put).await {
-            Ok(_) => {
-                info!("Successfully updated flow `NodeAddressValue`: {:?}", peer);
-                // TODO(discord): broadcast invalidating cache to all frontends
-            }
-            Err(e) => {
-                error!(e; "Failed to update flow `NodeAddressValue`: {:?}", peer);
-            }
-        }
-    } else {
-        warn!("Failed to serialize flow `NodeAddressValue`: {:?}", peer);
-    }
-}
--- a/src/meta-srv/src/metasrv.rs
+++ b/src/meta-srv/src/metasrv.rs
@@ -32,7 +32,6 @@ use common_meta::kv_backend::{KvBackendRef, ResettableKvBackend, ResettableKvBac
 use common_meta::leadership_notifier::{
    LeadershipChangeNotifier, LeadershipChangeNotifierCustomizerRef,
 };
-use common_meta::node_expiry_listener::NodeExpiryListener;
 use common_meta::peer::Peer;
 use common_meta::region_keeper::MemoryRegionKeeperRef;
 use common_meta::wal_options_allocator::WalOptionsAllocatorRef;
@@ -152,8 +151,6 @@ pub struct MetasrvOptions {
    #[cfg(feature = "pg_kvbackend")]
    /// Lock id for meta kv election. Only effect when using pg_kvbackend.
    pub meta_election_lock_id: u64,
-    #[serde(with = "humantime_serde")]
-    pub node_max_idle_time: Duration,
 }

 const DEFAULT_METASRV_ADDR_PORT: &str = "3002";
@@ -195,7 +192,6 @@ impl Default for MetasrvOptions {
            meta_table_name: DEFAULT_META_TABLE_NAME.to_string(),
            #[cfg(feature = "pg_kvbackend")]
            meta_election_lock_id: DEFAULT_META_ELECTION_LOCK_ID,
-            node_max_idle_time: Duration::from_secs(24 * 60 * 60),
        }
    }
 }
@@ -446,10 +442,6 @@ impl Metasrv {
            leadership_change_notifier.add_listener(self.wal_options_allocator.clone());
            leadership_change_notifier
                .add_listener(Arc::new(ProcedureManagerListenerAdapter(procedure_manager)));
-            leadership_change_notifier.add_listener(Arc::new(NodeExpiryListener::new(
-                self.options.node_max_idle_time,
-                self.in_memory.clone(),
-            )));
            if let Some(region_supervisor_ticker) = &self.region_supervisor_ticker {
                leadership_change_notifier.add_listener(region_supervisor_ticker.clone() as _);
            }
--- a/src/meta-srv/src/service/heartbeat.rs
+++ b/src/meta-srv/src/service/heartbeat.rs
@@ -27,9 +27,10 @@ use snafu::OptionExt;
 use tokio::sync::mpsc;
 use tokio::sync::mpsc::Sender;
 use tokio_stream::wrappers::ReceiverStream;
-use tonic::{Request, Response, Status, Streaming};
+use tonic::{Request, Response, Streaming};

-use crate::error::{self, Result};
+use crate::error;
+use crate::error::Result;
 use crate::handler::{HeartbeatHandlerGroup, Pusher, PusherId};
 use crate::metasrv::{Context, Metasrv};
 use crate::metrics::METRIC_META_HEARTBEAT_RECV;
@@ -67,15 +68,13 @@ impl heartbeat_server::Heartbeat for Metasrv {
                        };

                        if pusher_id.is_none() {
-                            pusher_id =
-                                Some(register_pusher(&handler_group, header, tx.clone()).await);
+                            pusher_id = register_pusher(&handler_group, header, tx.clone()).await;
                        }
                        if let Some(k) = &pusher_id {
                            METRIC_META_HEARTBEAT_RECV.with_label_values(&[&k.to_string()]);
                        } else {
                            METRIC_META_HEARTBEAT_RECV.with_label_values(&["none"]);
                        }
-
                        let res = handler_group
                            .handle(req, ctx.clone())
                            .await
@@ -108,12 +107,6 @@ impl heartbeat_server::Heartbeat for Metasrv {

                if is_not_leader {
                    warn!("Quit because it is no longer the leader");
-                    let _ = tx
-                        .send(Err(Status::aborted(format!(
-                            "The requested metasrv node is not leader, node addr: {}",
-                            ctx.server_addr
-                        ))))
-                        .await;
                    break;
                }
            }
@@ -180,13 +173,13 @@ async fn register_pusher(
    handler_group: &HeartbeatHandlerGroup,
    header: &RequestHeader,
    sender: Sender<std::result::Result<HeartbeatResponse, tonic::Status>>,
-) -> PusherId {
+) -> Option<PusherId> {
    let role = header.role();
    let id = get_node_id(header);
    let pusher_id = PusherId::new(role, id);
    let pusher = Pusher::new(sender, header);
    handler_group.register_pusher(pusher_id, pusher).await;
-    pusher_id
+    Some(pusher_id)
 }

 #[cfg(test)]
--- a/src/meta-srv/src/service/procedure.rs
+++ b/src/meta-srv/src/service/procedure.rs
@@ -17,15 +17,13 @@ use std::time::Duration;

 use api::v1::meta::{
    procedure_service_server, DdlTaskRequest as PbDdlTaskRequest,
-    DdlTaskResponse as PbDdlTaskResponse, Error, MigrateRegionRequest, MigrateRegionResponse,
+    DdlTaskResponse as PbDdlTaskResponse, MigrateRegionRequest, MigrateRegionResponse,
    ProcedureDetailRequest, ProcedureDetailResponse, ProcedureStateResponse, QueryProcedureRequest,
-    ResponseHeader,
 };
 use common_meta::ddl::ExecutorContext;
 use common_meta::rpc::ddl::{DdlTask, SubmitDdlTaskRequest};
 use common_meta::rpc::procedure;
-use common_telemetry::warn;
-use snafu::{OptionExt, ResultExt};
+use snafu::{ensure, OptionExt, ResultExt};
 use tonic::{Request, Response};

 use super::GrpcResult;
@@ -39,16 +37,6 @@ impl procedure_service_server::ProcedureService for Metasrv {
        &self,
        request: Request<QueryProcedureRequest>,
    ) -> GrpcResult<ProcedureStateResponse> {
-        if !self.is_leader() {
-            let resp = ProcedureStateResponse {
-                header: Some(ResponseHeader::failed(0, Error::is_not_leader())),
-                ..Default::default()
-            };
-
-            warn!("The current meta is not leader, but a `query procedure state` request have reached the meta. Detail: {:?}.", request);
-            return Ok(Response::new(resp));
-        }
-
        let QueryProcedureRequest { header, pid, .. } = request.into_inner();
        let _header = header.context(error::MissingRequestHeaderSnafu)?;
        let pid = pid.context(error::MissingRequiredParameterSnafu { param: "pid" })?;
@@ -69,16 +57,6 @@ impl procedure_service_server::ProcedureService for Metasrv {
    }

    async fn ddl(&self, request: Request<PbDdlTaskRequest>) -> GrpcResult<PbDdlTaskResponse> {
-        if !self.is_leader() {
-            let resp = PbDdlTaskResponse {
-                header: Some(ResponseHeader::failed(0, Error::is_not_leader())),
-                ..Default::default()
-            };
-
-            warn!("The current meta is not leader, but a `ddl` request have reached the meta. Detail: {:?}.", request);
-            return Ok(Response::new(resp));
-        }
-
        let PbDdlTaskRequest {
            header,
            query_context,
@@ -121,15 +99,12 @@ impl procedure_service_server::ProcedureService for Metasrv {
        &self,
        request: Request<MigrateRegionRequest>,
    ) -> GrpcResult<MigrateRegionResponse> {
-        if !self.is_leader() {
-            let resp = MigrateRegionResponse {
-                header: Some(ResponseHeader::failed(0, Error::is_not_leader())),
-                ..Default::default()
-            };
-
-            warn!("The current meta is not leader, but a `migrate` request have reached the meta. Detail: {:?}.", request);
-            return Ok(Response::new(resp));
-        }
+        ensure!(
+            self.meta_peer_client().is_leader(),
+            error::UnexpectedSnafu {
+                violated: "Trying to submit a region migration procedure to non-leader meta server"
+            }
+        );

        let MigrateRegionRequest {
            header,
@@ -175,16 +150,6 @@ impl procedure_service_server::ProcedureService for Metasrv {
        &self,
        request: Request<ProcedureDetailRequest>,
    ) -> GrpcResult<ProcedureDetailResponse> {
-        if !self.is_leader() {
-            let resp = ProcedureDetailResponse {
-                header: Some(ResponseHeader::failed(0, Error::is_not_leader())),
-                ..Default::default()
-            };
-
-            warn!("The current meta is not leader, but a `procedure details` request have reached the meta. Detail: {:?}.", request);
-            return Ok(Response::new(resp));
-        }
-
        let ProcedureDetailRequest { header } = request.into_inner();
        let _header = header.context(error::MissingRequestHeaderSnafu)?;
        let metas = self
--- a/src/metric-engine/src/data_region.rs
+++ b/src/metric-engine/src/data_region.rs
@@ -142,7 +142,6 @@ impl DataRegion {
                c.column_id = new_column_id_start + delta as u32;
                c.column_schema.set_nullable();
                match index_options {
-                    IndexOptions::None => {}
                    IndexOptions::Inverted => {
                        c.column_schema.set_inverted_index(true);
                    }
--- a/src/metric-engine/src/engine/create.rs
+++ b/src/metric-engine/src/engine/create.rs
@@ -21,7 +21,7 @@ use api::v1::SemanticType;
 use common_telemetry::info;
 use common_time::{Timestamp, FOREVER};
 use datatypes::data_type::ConcreteDataType;
-use datatypes::schema::{ColumnSchema, SkippingIndexOptions};
+use datatypes::schema::ColumnSchema;
 use datatypes::value::Value;
 use mito2::engine::MITO_ENGINE_NAME;
 use object_store::util::join_dir;
@@ -55,8 +55,6 @@ use crate::error::{
 use crate::metrics::PHYSICAL_REGION_COUNT;
 use crate::utils::{self, to_data_region_id, to_metadata_region_id};

-const DEFAULT_TABLE_ID_SKIPPING_INDEX_GRANULARITY: u32 = 1024;
-
 impl MetricEngineInner {
    pub async fn create_regions(
        &self,
@@ -442,7 +440,6 @@ impl MetricEngineInner {
    ///
    /// Return `[table_id_col, tsid_col]`
    fn internal_column_metadata() -> [ColumnMetadata; 2] {
-        // Safety: BloomFilter is a valid skipping index type
        let metric_name_col = ColumnMetadata {
            column_id: ReservedColumnId::table_id(),
            semantic_type: SemanticType::Tag,
@@ -451,11 +448,7 @@ impl MetricEngineInner {
                ConcreteDataType::uint32_datatype(),
                false,
            )
-            .with_skipping_options(SkippingIndexOptions {
-                granularity: DEFAULT_TABLE_ID_SKIPPING_INDEX_GRANULARITY,
-                index_type: datatypes::schema::SkippingIndexType::BloomFilter,
-            })
-            .unwrap(),
+            .with_inverted_index(true),
        };
        let tsid_col = ColumnMetadata {
            column_id: ReservedColumnId::tsid(),
--- a/src/metric-engine/src/engine/drop.rs
+++ b/src/metric-engine/src/engine/drop.rs
@@ -30,10 +30,9 @@ impl MetricEngineInner {
    pub async fn drop_region(
        &self,
        region_id: RegionId,
-        req: RegionDropRequest,
+        _req: RegionDropRequest,
    ) -> Result<AffectedRows> {
        let data_region_id = utils::to_data_region_id(region_id);
-        let fast_path = req.fast_path;

        // enclose the guard in a block to prevent the guard from polluting the async context
        let (is_physical_region, is_physical_region_busy) = {
@@ -53,7 +52,7 @@ impl MetricEngineInner {

        if is_physical_region {
            // check if there is no logical region relates to this physical region
-            if is_physical_region_busy && !fast_path {
+            if is_physical_region_busy {
                // reject if there is any present logical region
                return Err(PhysicalRegionBusySnafu {
                    region_id: data_region_id,
@@ -61,21 +60,9 @@ impl MetricEngineInner {
                .build());
            }

-            return self.drop_physical_region(data_region_id).await;
-        }
-
-        if fast_path {
-            // for fast path, we don't delete the metadata in the metadata region.
-            // it only remove the logical region from the engine state.
-            //
-            // The drop database procedure will ensure the metadata region and data region are dropped eventually.
-            self.state
-                .write()
-                .unwrap()
-                .remove_logical_region(region_id)?;
-
-            Ok(0)
+            self.drop_physical_region(data_region_id).await
        } else {
+            // cannot merge these two `if` otherwise the stupid type checker will complain
            let metadata_region_id = self
                .state
                .read()
@@ -100,16 +87,13 @@ impl MetricEngineInner {
        // Since the physical regions are going to be dropped, we don't need to
        // update the contents in metadata region.
        self.mito
-            .handle_request(
-                data_region_id,
-                RegionRequest::Drop(RegionDropRequest { fast_path: false }),
-            )
+            .handle_request(data_region_id, RegionRequest::Drop(RegionDropRequest {}))
            .await
            .with_context(|_| CloseMitoRegionSnafu { region_id })?;
        self.mito
            .handle_request(
                metadata_region_id,
-                RegionRequest::Drop(RegionDropRequest { fast_path: false }),
+                RegionRequest::Drop(RegionDropRequest {}),
            )
            .await
            .with_context(|_| CloseMitoRegionSnafu { region_id })?;
--- a/src/metric-engine/src/engine/options.rs
+++ b/src/metric-engine/src/engine/options.rs
@@ -40,7 +40,6 @@ pub struct PhysicalRegionOptions {
 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
 pub enum IndexOptions {
    #[default]
-    None,
    Inverted,
    Skipping {
        granularity: u32,
--- a/src/mito2/src/access_layer.rs
+++ b/src/mito2/src/access_layer.rs
@@ -13,7 +13,6 @@
 // limitations under the License.

 use std::sync::Arc;
-use std::time::Duration;

 use object_store::services::Fs;
 use object_store::util::{join_dir, with_instrument_layers};
@@ -43,29 +42,6 @@ pub type AccessLayerRef = Arc<AccessLayer>;
 /// SST write results.
 pub type SstInfoArray = SmallVec<[SstInfo; 2]>;

-#[derive(Debug, Default)]
-pub struct Metrics {
-    pub read: Duration,
-    pub write: Duration,
-    pub convert: Duration,
-    pub index_update: Duration,
-    pub index_finish: Duration,
-    pub close: Duration,
-    pub num_series: usize,
-
-    // SST Opendal metrics.
-    pub opendal_create_cost: Duration,
-    pub opendal_num_writes: usize,
-    pub opendal_write_cost: Duration,
-    pub opendal_complete_cost: Duration,
-}
-
-impl Metrics {
-    pub fn sum(&self) -> Duration {
-        self.read + self.write + self.convert + self.index_update + self.index_finish + self.close
-    }
-}
-
 /// A layer to access SST files under the same directory.
 pub struct AccessLayer {
    region_dir: String,
@@ -145,11 +121,10 @@ impl AccessLayer {
    /// Writes a SST with specific `file_id` and `metadata` to the layer.
    ///
    /// Returns the info of the SST. If no data written, returns None.
-    pub async fn write_sst(
+    pub(crate) async fn write_sst(
        &self,
        request: SstWriteRequest,
        write_opts: &WriteOptions,
-        metrics: &mut Metrics,
    ) -> Result<SstInfoArray> {
        let region_id = request.metadata.region_id;
        let cache_manager = request.cache_manager.clone();
@@ -171,14 +146,11 @@ impl AccessLayer {
        } else {
            // Write cache is disabled.
            let store = self.object_store.clone();
-            let path_provider = RegionFilePathFactory::new(self.region_dir.clone());
            let indexer_builder = IndexerBuilderImpl {
                op_type: request.op_type,
                metadata: request.metadata.clone(),
                row_group_size: write_opts.row_group_size,
-                puffin_manager: self
-                    .puffin_manager_factory
-                    .build(store, path_provider.clone()),
+                puffin_manager: self.puffin_manager_factory.build(store),
                intermediate_manager: self.intermediate_manager.clone(),
                index_options: request.index_options,
                inverted_index_config: request.inverted_index_config,
@@ -189,19 +161,14 @@ impl AccessLayer {
                self.object_store.clone(),
                request.metadata,
                indexer_builder,
-                path_provider,
+                RegionFilePathFactory {
+                    region_dir: self.region_dir.clone(),
+                },
            )
            .await;
-            let sst_info = writer
-                .write_all(request.source, request.max_sequence, write_opts, metrics)
-                .await?;
-            let opendal_metrics = writer.opendal_metrics_val();
-            metrics.opendal_create_cost += opendal_metrics.create_cost;
-            metrics.opendal_num_writes += opendal_metrics.num_writes;
-            metrics.opendal_write_cost += opendal_metrics.write_cost;
-            metrics.opendal_complete_cost += opendal_metrics.complete_cost;
-
-            sst_info
+            writer
+                .write_all(request.source, request.max_sequence, write_opts)
+                .await?
        };

        // Put parquet metadata to cache manager.
@@ -221,53 +188,28 @@ impl AccessLayer {
    }
 }

-/// Helper to build an [AccessLayerRef] with internal index managers.
-///
-/// This is a convenience constructor intended for tooling that needs to
-/// interact with SSTs without wiring all indexing internals manually.
-pub async fn build_access_layer(
-    region_dir: &str,
-    object_store: ObjectStore,
-    config: &crate::config::MitoConfig,
-) -> Result<AccessLayerRef> {
-    let puffin_manager_factory = PuffinManagerFactory::new(
-        &config.index.aux_path,
-        config.index.staging_size.as_bytes(),
-        Some(config.index.write_buffer_size.as_bytes() as _),
-        config.index.staging_ttl,
-    )
-    .await?;
-    let intermediate_manager = IntermediateManager::init_fs(&config.index.aux_path).await?;
-    Ok(Arc::new(AccessLayer::new(
-        region_dir,
-        object_store,
-        puffin_manager_factory,
-        intermediate_manager,
-    )))
-}
-
 /// `OperationType` represents the origin of the `SstWriteRequest`.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub enum OperationType {
+pub(crate) enum OperationType {
    Flush,
    Compact,
 }

 /// Contents to build a SST.
-pub struct SstWriteRequest {
-    pub op_type: OperationType,
-    pub metadata: RegionMetadataRef,
-    pub source: Source,
-    pub cache_manager: CacheManagerRef,
+pub(crate) struct SstWriteRequest {
+    pub(crate) op_type: OperationType,
+    pub(crate) metadata: RegionMetadataRef,
+    pub(crate) source: Source,
+    pub(crate) cache_manager: CacheManagerRef,
    #[allow(dead_code)]
-    pub storage: Option<String>,
-    pub max_sequence: Option<SequenceNumber>,
+    pub(crate) storage: Option<String>,
+    pub(crate) max_sequence: Option<SequenceNumber>,

    /// Configs for index
-    pub index_options: IndexOptions,
-    pub inverted_index_config: InvertedIndexConfig,
-    pub fulltext_index_config: FulltextIndexConfig,
-    pub bloom_filter_index_config: BloomFilterConfig,
+    pub(crate) index_options: IndexOptions,
+    pub(crate) inverted_index_config: InvertedIndexConfig,
+    pub(crate) fulltext_index_config: FulltextIndexConfig,
+    pub(crate) bloom_filter_index_config: BloomFilterConfig,
 }

 pub(crate) async fn new_fs_cache_store(root: &str) -> Result<ObjectStore> {
@@ -306,18 +248,8 @@ pub trait FilePathProvider: Send + Sync {
 /// Path provider that builds paths in local write cache.
 #[derive(Clone)]
 pub(crate) struct WriteCachePathProvider {
-    region_id: RegionId,
-    file_cache: FileCacheRef,
-}
-
-impl WriteCachePathProvider {
-    /// Creates a new `WriteCachePathProvider` instance.
-    pub fn new(region_id: RegionId, file_cache: FileCacheRef) -> Self {
-        Self {
-            region_id,
-            file_cache,
-        }
-    }
+    pub(crate) region_id: RegionId,
+    pub(crate) file_cache: FileCacheRef,
 }

 impl FilePathProvider for WriteCachePathProvider {
@@ -335,14 +267,7 @@ impl FilePathProvider for WriteCachePathProvider {
 /// Path provider that builds paths in region storage path.
 #[derive(Clone, Debug)]
 pub(crate) struct RegionFilePathFactory {
-    region_dir: String,
-}
-
-impl RegionFilePathFactory {
-    /// Creates a new `RegionFilePathFactory` instance.
-    pub fn new(region_dir: String) -> Self {
-        Self { region_dir }
-    }
+    pub(crate) region_dir: String,
 }

 impl FilePathProvider for RegionFilePathFactory {
--- a/src/mito2/src/cache/file_cache.rs
+++ b/src/mito2/src/cache/file_cache.rs
@@ -187,12 +187,9 @@ impl FileCache {
    }

    /// Removes a file from the cache explicitly.
-    /// It always tries to remove the file from the local store because we may not have the file
-    /// in the memory index if upload is failed.
    pub(crate) async fn remove(&self, key: IndexKey) {
        let file_path = self.cache_file_path(key);
        self.memory_index.remove(&key).await;
-        // Always delete the file from the local store.
        if let Err(e) = self.local_store.delete(&file_path).await {
            warn!(e; "Failed to delete a cached file {}", file_path);
        }
--- a/src/mito2/src/cache/write_cache.rs
+++ b/src/mito2/src/cache/write_cache.rs
@@ -22,7 +22,6 @@ use common_telemetry::{debug, info};
 use futures::AsyncWriteExt;
 use object_store::ObjectStore;
 use snafu::ResultExt;
-use store_api::storage::RegionId;

 use crate::access_layer::{
    new_fs_cache_store, FilePathProvider, RegionFilePathFactory, SstInfoArray, SstWriteRequest,
@@ -40,7 +39,6 @@ use crate::sst::index::IndexerBuilderImpl;
 use crate::sst::parquet::writer::ParquetWriter;
 use crate::sst::parquet::WriteOptions;
 use crate::sst::{DEFAULT_WRITE_BUFFER_SIZE, DEFAULT_WRITE_CONCURRENCY};
-use crate::Metrics;

 /// A cache for uploading files to remote object stores.
 ///
@@ -116,14 +114,15 @@ impl WriteCache {
        let region_id = write_request.metadata.region_id;

        let store = self.file_cache.local_store();
-        let path_provider = WriteCachePathProvider::new(region_id, self.file_cache.clone());
+        let path_provider = WriteCachePathProvider {
+            file_cache: self.file_cache.clone(),
+            region_id,
+        };
        let indexer = IndexerBuilderImpl {
            op_type: write_request.op_type,
            metadata: write_request.metadata.clone(),
            row_group_size: write_opts.row_group_size,
-            puffin_manager: self
-                .puffin_manager_factory
-                .build(store, path_provider.clone()),
+            puffin_manager: self.puffin_manager_factory.build(store),
            intermediate_manager: self.intermediate_manager.clone(),
            index_options: write_request.index_options,
            inverted_index_config: write_request.inverted_index_config,
@@ -141,12 +140,7 @@ impl WriteCache {
        .await;

        let sst_info = writer
-            .write_all(
-                write_request.source,
-                write_request.max_sequence,
-                write_opts,
-                &mut Metrics::default(),
-            )
+            .write_all(write_request.source, write_request.max_sequence, write_opts)
            .await?;

        timer.stop_and_record();
@@ -156,41 +150,24 @@ impl WriteCache {
            return Ok(sst_info);
        }

-        let mut upload_tracker = UploadTracker::new(region_id);
-        let mut err = None;
        let remote_store = &upload_request.remote_store;
        for sst in &sst_info {
            let parquet_key = IndexKey::new(region_id, sst.file_id, FileType::Parquet);
            let parquet_path = upload_request
                .dest_path_provider
                .build_sst_file_path(sst.file_id);
-            if let Err(e) = self.upload(parquet_key, &parquet_path, remote_store).await {
-                err = Some(e);
-                break;
-            }
-            upload_tracker.push_uploaded_file(parquet_path);
+            self.upload(parquet_key, &parquet_path, remote_store)
+                .await?;

            if sst.index_metadata.file_size > 0 {
                let puffin_key = IndexKey::new(region_id, sst.file_id, FileType::Puffin);
-                let puffin_path = upload_request
+                let puffin_path = &upload_request
                    .dest_path_provider
                    .build_index_file_path(sst.file_id);
-                if let Err(e) = self.upload(puffin_key, &puffin_path, remote_store).await {
-                    err = Some(e);
-                    break;
-                }
-                upload_tracker.push_uploaded_file(puffin_path);
+                self.upload(puffin_key, puffin_path, remote_store).await?;
            }
        }

-        if let Some(err) = err {
-            // Cleans files on failure.
-            upload_tracker
-                .clean(&sst_info, &self.file_cache, remote_store)
-                .await;
-            return Err(err);
-        }
-
        Ok(sst_info)
    }

@@ -356,61 +333,6 @@ pub struct SstUploadRequest {
    pub remote_store: ObjectStore,
 }

-/// A structs to track files to upload and clean them if upload failed.
-struct UploadTracker {
-    /// Id of the region to track.
-    region_id: RegionId,
-    /// Paths of files uploaded successfully.
-    files_uploaded: Vec<String>,
-}
-
-impl UploadTracker {
-    /// Creates a new instance of `UploadTracker` for a given region.
-    fn new(region_id: RegionId) -> Self {
-        Self {
-            region_id,
-            files_uploaded: Vec::new(),
-        }
-    }
-
-    /// Add a file path to the list of uploaded files.
-    fn push_uploaded_file(&mut self, path: String) {
-        self.files_uploaded.push(path);
-    }
-
-    /// Cleans uploaded files and files in the file cache at best effort.
-    async fn clean(
-        &self,
-        sst_info: &SstInfoArray,
-        file_cache: &FileCacheRef,
-        remote_store: &ObjectStore,
-    ) {
-        common_telemetry::info!(
-            "Start cleaning files on upload failure, region: {}, num_ssts: {}",
-            self.region_id,
-            sst_info.len()
-        );
-
-        // Cleans files in the file cache first.
-        for sst in sst_info {
-            let parquet_key = IndexKey::new(self.region_id, sst.file_id, FileType::Parquet);
-            file_cache.remove(parquet_key).await;
-
-            if sst.index_metadata.file_size > 0 {
-                let puffin_key = IndexKey::new(self.region_id, sst.file_id, FileType::Puffin);
-                file_cache.remove(puffin_key).await;
-            }
-        }
-
-        // Cleans uploaded files.
-        for file_path in &self.files_uploaded {
-            if let Err(e) = remote_store.delete(file_path).await {
-                common_telemetry::error!(e; "Failed to delete file {}", file_path);
-            }
-        }
-    }
-}
-
 #[cfg(test)]
 mod tests {
    use common_test_util::temp_dir::create_temp_dir;
@@ -433,7 +355,9 @@ mod tests {
        // and now just use local file system to mock.
        let mut env = TestEnv::new();
        let mock_store = env.init_object_store_manager();
-        let path_provider = RegionFilePathFactory::new("test".to_string());
+        let path_provider = RegionFilePathFactory {
+            region_dir: "test".to_string(),
+        };

        let local_dir = create_temp_dir("");
        let local_store = new_fs_store(local_dir.path().to_str().unwrap());
@@ -564,7 +488,9 @@ mod tests {
            ..Default::default()
        };
        let upload_request = SstUploadRequest {
-            dest_path_provider: RegionFilePathFactory::new(data_home.clone()),
+            dest_path_provider: RegionFilePathFactory {
+                region_dir: data_home.clone(),
+            },
            remote_store: mock_store.clone(),
        };

--- a/src/mito2/src/compaction/compactor.rs
+++ b/src/mito2/src/compaction/compactor.rs
@@ -27,7 +27,7 @@ use snafu::{OptionExt, ResultExt};
 use store_api::metadata::RegionMetadataRef;
 use store_api::storage::RegionId;

-use crate::access_layer::{AccessLayer, AccessLayerRef, Metrics, OperationType, SstWriteRequest};
+use crate::access_layer::{AccessLayer, AccessLayerRef, OperationType, SstWriteRequest};
 use crate::cache::{CacheManager, CacheManagerRef};
 use crate::compaction::picker::{new_picker, PickerOutput};
 use crate::compaction::{find_ttl, CompactionSstReaderBuilder};
@@ -340,7 +340,6 @@ impl Compactor for DefaultCompactor {
                            bloom_filter_index_config,
                        },
                        &write_opts,
-                        &mut Metrics::default(),
                    )
                    .await?
                    .into_iter()
--- a/src/mito2/src/engine/drop_test.rs
+++ b/src/mito2/src/engine/drop_test.rs
@@ -56,10 +56,7 @@ async fn test_engine_drop_region() {

    // It's okay to drop a region doesn't exist.
    engine
-        .handle_request(
-            region_id,
-            RegionRequest::Drop(RegionDropRequest { fast_path: false }),
-        )
+        .handle_request(region_id, RegionRequest::Drop(RegionDropRequest {}))
        .await
        .unwrap_err();

@@ -89,10 +86,7 @@ async fn test_engine_drop_region() {

    // drop the created region.
    engine
-        .handle_request(
-            region_id,
-            RegionRequest::Drop(RegionDropRequest { fast_path: false }),
-        )
+        .handle_request(region_id, RegionRequest::Drop(RegionDropRequest {}))
        .await
        .unwrap();
    assert!(!engine.is_region_exists(region_id));
@@ -198,10 +192,7 @@ async fn test_engine_drop_region_for_custom_store() {

    // Drop the custom region.
    engine
-        .handle_request(
-            custom_region_id,
-            RegionRequest::Drop(RegionDropRequest { fast_path: false }),
-        )
+        .handle_request(custom_region_id, RegionRequest::Drop(RegionDropRequest {}))
        .await
        .unwrap();
    assert!(!engine.is_region_exists(custom_region_id));
--- a/src/mito2/src/flush.rs
+++ b/src/mito2/src/flush.rs
@@ -25,7 +25,7 @@ use store_api::storage::RegionId;
 use strum::IntoStaticStr;
 use tokio::sync::{mpsc, watch};

-use crate::access_layer::{AccessLayerRef, Metrics, OperationType, SstWriteRequest};
+use crate::access_layer::{AccessLayerRef, OperationType, SstWriteRequest};
 use crate::cache::CacheManagerRef;
 use crate::config::MitoConfig;
 use crate::error::{
@@ -366,7 +366,7 @@ impl RegionFlushTask {

            let ssts_written = self
                .access_layer
-                .write_sst(write_request, &write_opts, &mut Metrics::default())
+                .write_sst(write_request, &write_opts)
                .await?;
            if ssts_written.is_empty() {
                // No data written.
--- a/Show More
+++ b/Show More