### Added Deduplication and Merge Functionality

- Introduced `DedupReader` and `MergeReader` in `src/mito2/src/read/sync/dedup.rs` and `src/mito2/src/read/sync/merge.rs` to handle deduplication and merging of sorted batches.
 ### Enhanced `BulkMemtable` Iteration
 - Updated `BulkMemtable` in `src/mito2/src/memtable/bulk.rs` to support deduplication and merge modes during iteration.
 - Added `BulkIterContext` to manage iteration context.
 ### Testing Enhancements
 - Added comprehensive tests for `BulkMemtable` and `BulkPart` in `src/mito2/src/memtable/bulk.rs` and `src/mito2/src/memtable/bulk/part.rs`.
 ### Code Refactoring
 - Made `BulkPart` and `BulkPartMeta` cloneable in `src/mito2/src/memtable/bulk/part.rs`.
 - Exposed internal test modules for better test coverage in `src/mito2/src/memtable/time_series.rs` and `src/mito2/src/read/merge.rs`.
 ### New Modules
 - Created `sync` module in `src/mito2/src/read.rs` to organize synchronous read operations.

.cargo/config.toml

@@ -3,12 +3,3 @@ linker = "aarch64-linux-gnu-gcc"
 
 [alias]
 sqlness = "run --bin sqlness-runner --"
-
-[unstable.git]
-shallow_index = true
-shallow_deps = true
-[unstable.gitoxide]
-fetch = true
-checkout = true
-list_files = true
-internal_use_git2 = false

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

@@ -41,13 +41,6 @@ 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
       shell: bash
       if: ${{ inputs.build-dev-builder-ubuntu == 'true' }}
@@ -76,8 +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 }}

.github/actions/build-greptime-images/action.yml

@@ -34,8 +34,8 @@ inputs:
     required: true
   push-latest-tag:
     description: Whether to push the latest tag
-    required: true
-    default: 'false'
+    required: false
+    default: 'true'
 runs:
   using: composite
   steps:
@@ -47,11 +47,7 @@ runs:
         password: ${{ inputs.image-registry-password }}
 
     - 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
+      uses: docker/setup-qemu-action@v2
 
     - name: Set up buildx
       uses: docker/setup-buildx-action@v2

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

@@ -22,8 +22,8 @@ inputs:
     required: true
   push-latest-tag:
     description: Whether to push the latest tag
-    required: true
-    default: 'false'
+    required: false
+    default: 'true'
   dev-mode:
     description: Enable dev mode, only build standard greptime
     required: false

.github/actions/release-cn-artifacts/action.yaml

@@ -51,8 +51,8 @@ inputs:
     required: true
   upload-to-s3:
     description: Upload to S3
-    required: true
-    default: 'false'
+    required: false
+    default: 'true'
   artifacts-dir:
     description: Directory to store artifacts
     required: false
@@ -77,21 +77,13 @@ runs:
       with:
         path: ${{ inputs.artifacts-dir }}
 
-    - name: Install s5cmd
-      shell: bash
-      run: |
-        wget https://github.com/peak/s5cmd/releases/download/v2.3.0/s5cmd_2.3.0_Linux-64bit.tar.gz
-        tar -xzf s5cmd_2.3.0_Linux-64bit.tar.gz
-        sudo mv s5cmd /usr/local/bin/
-        sudo chmod +x /usr/local/bin/s5cmd
-
     - name: Release artifacts to cn region
       uses: nick-invision/retry@v2
       if: ${{ inputs.upload-to-s3 == 'true' }}
       env:
         AWS_ACCESS_KEY_ID: ${{ inputs.aws-cn-access-key-id }}
         AWS_SECRET_ACCESS_KEY: ${{ inputs.aws-cn-secret-access-key }}
-        AWS_REGION: ${{ inputs.aws-cn-region }}
+        AWS_DEFAULT_REGION: ${{ inputs.aws-cn-region }}
         UPDATE_VERSION_INFO: ${{ inputs.update-version-info }}
       with:
         max_attempts: ${{ inputs.upload-max-retry-times }}

.github/scripts/upload-artifacts-to-s3.sh

@@ -33,7 +33,7 @@ function upload_artifacts() {
   #    ├── greptime-darwin-amd64-v0.2.0.sha256sum
   #    └── greptime-darwin-amd64-v0.2.0.tar.gz
   find "$ARTIFACTS_DIR" -type f \( -name "*.tar.gz" -o -name "*.sha256sum" \) | while IFS= read -r file; do
-    s5cmd cp \
+    aws s3 cp \
       "$file" "s3://$AWS_S3_BUCKET/$RELEASE_DIRS/$VERSION/$(basename "$file")"
   done
 }
@@ -45,7 +45,7 @@ function update_version_info() {
     if [[ "$VERSION" =~ ^v[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
       echo "Updating latest-version.txt"
       echo "$VERSION" > latest-version.txt
-      s5cmd cp \
+      aws s3 cp \
         latest-version.txt "s3://$AWS_S3_BUCKET/$RELEASE_DIRS/latest-version.txt"
     fi
 
@@ -53,7 +53,7 @@ function update_version_info() {
     if [[ "$VERSION" == *"nightly"* ]]; then
       echo "Updating latest-nightly-version.txt"
       echo "$VERSION" > latest-nightly-version.txt
-      s5cmd cp \
+      aws s3 cp \
         latest-nightly-version.txt "s3://$AWS_S3_BUCKET/$RELEASE_DIRS/latest-nightly-version.txt"
     fi
   fi

.github/workflows/apidoc.yml

@@ -17,8 +17,6 @@ jobs:
     runs-on: ubuntu-20.04
     steps:
     - uses: actions/checkout@v4
-      with:
-        persist-credentials: false
     - uses: arduino/setup-protoc@v3
       with:
         repo-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/dependency-check.yml

@@ -12,8 +12,6 @@ jobs:
     steps:
     - name: Checkout code
       uses: actions/checkout@v4
-      with:
-        persist-credentials: false
 
     - name: Set up Rust
       uses: actions-rust-lang/setup-rust-toolchain@v1

.github/workflows/dev-build.yml

@@ -76,9 +76,15 @@ env:
 
   NIGHTLY_RELEASE_PREFIX: nightly
 
+  # Use the different image name to avoid conflict with the release images.
+  IMAGE_NAME: greptimedb-dev
+
   # The source code will check out in the following path: '${WORKING_DIR}/dev/greptime'.
   CHECKOUT_GREPTIMEDB_PATH: dev/greptimedb
 
+permissions:
+  issues: write
+
 jobs:
   allocate-runners:
     name: Allocate runners
@@ -101,7 +107,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Create version
         id: create-version
@@ -156,7 +161,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Checkout greptimedb
         uses: actions/checkout@v4
@@ -164,7 +168,6 @@ jobs:
           repository: ${{ inputs.repository }}
           ref: ${{ inputs.commit }}
           path: ${{ env.CHECKOUT_GREPTIMEDB_PATH }}
-          persist-credentials: true
 
       - uses: ./.github/actions/build-linux-artifacts
         with:
@@ -189,7 +192,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Checkout greptimedb
         uses: actions/checkout@v4
@@ -197,7 +199,6 @@ jobs:
           repository: ${{ inputs.repository }}
           ref: ${{ inputs.commit }}
           path: ${{ env.CHECKOUT_GREPTIMEDB_PATH }}
-          persist-credentials: true
 
       - uses: ./.github/actions/build-linux-artifacts
         with:
@@ -225,14 +226,13 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Build and push images to dockerhub
         uses: ./.github/actions/build-images
         with:
           image-registry: docker.io
           image-namespace: ${{ vars.IMAGE_NAMESPACE }}
-          image-name: ${{ vars.DEV_BUILD_IMAGE_NAME }}
+          image-name: ${{ env.IMAGE_NAME }}
           image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
           image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
           version: ${{ needs.allocate-runners.outputs.version }}
@@ -257,14 +257,13 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Release artifacts to CN region
         uses: ./.github/actions/release-cn-artifacts
         with:
           src-image-registry: docker.io
           src-image-namespace: ${{ vars.IMAGE_NAMESPACE }}
-          src-image-name: ${{ vars.DEV_BUILD_IMAGE_NAME }}
+          src-image-name: ${{ env.IMAGE_NAME }}
           dst-image-registry-username: ${{ secrets.ALICLOUD_USERNAME }}
           dst-image-registry-password: ${{ secrets.ALICLOUD_PASSWORD }}
           dst-image-registry: ${{ vars.ACR_IMAGE_REGISTRY }}
@@ -274,7 +273,6 @@ 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: 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.
@@ -293,7 +291,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Stop EC2 runner
         uses: ./.github/actions/stop-runner
@@ -319,7 +316,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Stop EC2 runner
         uses: ./.github/actions/stop-runner
@@ -338,16 +334,10 @@ jobs:
       release-images-to-dockerhub
     ]
     runs-on: ubuntu-20.04
-    permissions:
-      issues: write
-
     env:
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL_DEVELOP_CHANNEL }}
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - name: Report CI status
         id: report-ci-status

.github/workflows/develop.yml

@@ -26,8 +26,6 @@ jobs:
     runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: crate-ci/typos@master
       - name: Check the config docs
         run: |
@@ -40,8 +38,6 @@ jobs:
     name: Check License Header
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: korandoru/hawkeye@v5
 
   check:
@@ -53,8 +49,6 @@ jobs:
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
@@ -76,8 +70,6 @@ jobs:
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: actions-rust-lang/setup-rust-toolchain@v1
       - name: Install taplo
         run: cargo +stable install taplo-cli --version ^0.9 --locked --force
@@ -93,8 +85,6 @@ jobs:
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
@@ -149,8 +139,6 @@ jobs:
           echo "Disk space after:"
           df -h
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
@@ -204,8 +192,6 @@ jobs:
           echo "Disk space after:"
           df -h
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
@@ -252,8 +238,6 @@ jobs:
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
@@ -311,8 +295,6 @@ jobs:
           echo "Disk space after:"
           df -h
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - name: Setup Kind
         uses: ./.github/actions/setup-kind
       - if: matrix.mode.minio
@@ -455,8 +437,6 @@ jobs:
           echo "Disk space after:"
           df -h
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - name: Setup Kind
         uses: ./.github/actions/setup-kind
       - name: Setup Chaos Mesh
@@ -582,8 +562,6 @@ jobs:
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - if: matrix.mode.kafka
         name: Setup kafka server
         working-directory: tests-integration/fixtures
@@ -611,8 +589,6 @@ jobs:
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
@@ -628,8 +604,6 @@ jobs:
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
@@ -652,8 +626,6 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - name: Merge Conflict Finder
         uses: olivernybroe/action-conflict-finder@v4.0
 
@@ -664,8 +636,6 @@ jobs:
     needs:  [conflict-check, clippy, fmt]
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
@@ -714,8 +684,6 @@ jobs:
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: arduino/setup-protoc@v3
         with:
           repo-token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/docbot.yml

@@ -3,21 +3,16 @@ on:
   pull_request_target:
     types: [opened, edited]
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
-  cancel-in-progress: true
+permissions:
+  pull-requests: write
+  contents: read
 
 jobs:
   docbot:
     runs-on: ubuntu-20.04
-    permissions:
-      pull-requests: write
-      contents: read
     timeout-minutes: 10
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - name: Maybe Follow Up Docs Issue
         working-directory: cyborg

.github/workflows/docs.yml

@@ -34,8 +34,6 @@ jobs:
     runs-on: ubuntu-20.04
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: crate-ci/typos@master
 
   license-header-check:
@@ -43,8 +41,6 @@ jobs:
     name: Check License Header
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: korandoru/hawkeye@v5
 
   check:

.github/workflows/nightly-build.yml

@@ -66,6 +66,13 @@ env:
 
   NIGHTLY_RELEASE_PREFIX: nightly
 
+  # Use the different image name to avoid conflict with the release images.
+  # The DockerHub image will be greptime/greptimedb-nightly.
+  IMAGE_NAME: greptimedb-nightly
+
+permissions:
+  issues: write
+
 jobs:
   allocate-runners:
     name: Allocate runners
@@ -88,7 +95,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Create version
         id: create-version
@@ -141,7 +147,6 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - uses: ./.github/actions/build-linux-artifacts
         with:
@@ -163,7 +168,6 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - uses: ./.github/actions/build-linux-artifacts
         with:
@@ -189,18 +193,17 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Build and push images to dockerhub
         uses: ./.github/actions/build-images
         with:
           image-registry: docker.io
           image-namespace: ${{ vars.IMAGE_NAMESPACE }}
-          image-name: ${{ vars.NIGHTLY_BUILD_IMAGE_NAME }}
+          image-name: ${{ env.IMAGE_NAME }}
           image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
           image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
           version: ${{ needs.allocate-runners.outputs.version }}
-          push-latest-tag: false
+          push-latest-tag: true
 
       - name: Set nightly build result
         id: set-nightly-build-result
@@ -223,14 +226,13 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Release artifacts to CN region
         uses: ./.github/actions/release-cn-artifacts
         with:
           src-image-registry: docker.io
           src-image-namespace: ${{ vars.IMAGE_NAMESPACE }}
-          src-image-name: ${{ vars.NIGHTLY_BUILD_IMAGE_NAME }}
+          src-image-name: ${{ env.IMAGE_NAME }}
           dst-image-registry-username: ${{ secrets.ALICLOUD_USERNAME }}
           dst-image-registry-password: ${{ secrets.ALICLOUD_PASSWORD }}
           dst-image-registry: ${{ vars.ACR_IMAGE_REGISTRY }}
@@ -240,10 +242,9 @@ 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: false
           dev-mode: false
           update-version-info: false  # Don't update version info in S3.
-          push-latest-tag: false
+          push-latest-tag: true
 
   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
@@ -259,7 +260,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Stop EC2 runner
         uses: ./.github/actions/stop-runner
@@ -285,7 +285,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Stop EC2 runner
         uses: ./.github/actions/stop-runner
@@ -304,14 +303,10 @@ jobs:
       release-images-to-dockerhub
     ]
     runs-on: ubuntu-20.04
-    permissions:
-      issues: write
     env:
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL_DEVELOP_CHANNEL }}
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - name: Report CI status
         id: report-ci-status

.github/workflows/nightly-ci.yml

@@ -9,6 +9,9 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
   cancel-in-progress: true
 
+permissions:
+  issues: write
+
 jobs:
   sqlness-test:
     name: Run sqlness test
@@ -19,7 +22,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Check install.sh
         run: ./.github/scripts/check-install-script.sh
@@ -44,14 +46,9 @@ jobs:
     name: Sqlness tests on Windows
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     runs-on: windows-2022-8-cores
-    permissions:
-      issues: write
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - uses: arduino/setup-protoc@v3
         with:
@@ -79,9 +76,6 @@ jobs:
     steps:
       - run: git config --global core.autocrlf false
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - uses: arduino/setup-protoc@v3
         with:
@@ -117,13 +111,9 @@ jobs:
   cleanbuild-linux-nix:
     name: Run clean build on Linux
     runs-on: ubuntu-latest
-    if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     timeout-minutes: 60
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          persist-credentials: false
       - uses: cachix/install-nix-action@v27
         with:
           nix_path: nixpkgs=channel:nixos-24.11
@@ -151,9 +141,6 @@ jobs:
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL_DEVELOP_CHANNEL }}
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - name: Report CI status
         id: report-ci-status

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

@@ -29,7 +29,7 @@ jobs:
   release-dev-builder-images:
     name: Release dev builder images
     if: ${{ inputs.release_dev_builder_ubuntu_image || inputs.release_dev_builder_centos_image || inputs.release_dev_builder_android_image }} # Only manually trigger this job.
-    runs-on: ubuntu-22.04-16-cores
+    runs-on: ubuntu-20.04-16-cores
     outputs:
       version: ${{ steps.set-version.outputs.version }}
     steps:
@@ -37,7 +37,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Configure build image version
         id: set-version
@@ -63,7 +62,7 @@ jobs:
 
   release-dev-builder-images-ecr:
     name: Release dev builder images to AWS ECR
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-20.04
     needs: [
       release-dev-builder-images
     ]
@@ -86,69 +85,51 @@ jobs:
       - name: Push dev-builder-ubuntu image
         shell: bash
         if: ${{ inputs.release_dev_builder_ubuntu_image }}
-        env:
-          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
-          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
-          ECR_IMAGE_REGISTRY: ${{ vars.ECR_IMAGE_REGISTRY }}
-          ECR_IMAGE_NAMESPACE: ${{ vars.ECR_IMAGE_NAMESPACE }}
         run: |
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             quay.io/skopeo/stable:latest \
-            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-ubuntu:$IMAGE_VERSION \
-            docker://$ECR_IMAGE_REGISTRY/$ECR_IMAGE_NAMESPACE/dev-builder-ubuntu:$IMAGE_VERSION
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-ubuntu:${{ needs.release-dev-builder-images.outputs.version }} \
+            docker://${{ vars.ECR_IMAGE_REGISTRY }}/${{ vars.ECR_IMAGE_NAMESPACE }}/dev-builder-ubuntu:${{ needs.release-dev-builder-images.outputs.version }}
 
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             quay.io/skopeo/stable:latest \
-            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-ubuntu:latest \
-            docker://$ECR_IMAGE_REGISTRY/$ECR_IMAGE_NAMESPACE/dev-builder-ubuntu:latest
-
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-ubuntu:latest \
+            docker://${{ vars.ECR_IMAGE_REGISTRY }}/${{ vars.ECR_IMAGE_NAMESPACE }}/dev-builder-ubuntu:latest
       - name: Push dev-builder-centos image
         shell: bash
         if: ${{ inputs.release_dev_builder_centos_image }}
-        env:
-          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
-          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
-          ECR_IMAGE_REGISTRY: ${{ vars.ECR_IMAGE_REGISTRY }}
-          ECR_IMAGE_NAMESPACE: ${{ vars.ECR_IMAGE_NAMESPACE }}
         run: |
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             quay.io/skopeo/stable:latest \
-            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-centos:$IMAGE_VERSION \
-            docker://$ECR_IMAGE_REGISTRY/$ECR_IMAGE_NAMESPACE/dev-builder-centos:$IMAGE_VERSION
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-centos:${{ needs.release-dev-builder-images.outputs.version }} \
+            docker://${{ vars.ECR_IMAGE_REGISTRY }}/${{ vars.ECR_IMAGE_NAMESPACE }}/dev-builder-centos:${{ needs.release-dev-builder-images.outputs.version }}
 
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             quay.io/skopeo/stable:latest \
-            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-centos:latest \
-            docker://$ECR_IMAGE_REGISTRY/$ECR_IMAGE_NAMESPACE/dev-builder-centos:latest
-
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-centos:latest \
+            docker://${{ vars.ECR_IMAGE_REGISTRY }}/${{ vars.ECR_IMAGE_NAMESPACE }}/dev-builder-centos:latest
       - name: Push dev-builder-android image
         shell: bash
         if: ${{ inputs.release_dev_builder_android_image }}
-        env:
-          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
-          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
-          ECR_IMAGE_REGISTRY: ${{ vars.ECR_IMAGE_REGISTRY }}
-          ECR_IMAGE_NAMESPACE: ${{ vars.ECR_IMAGE_NAMESPACE }}
         run: |
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             quay.io/skopeo/stable:latest \
-            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-android:$IMAGE_VERSION \
-            docker://$ECR_IMAGE_REGISTRY/$ECR_IMAGE_NAMESPACE/dev-builder-android:$IMAGE_VERSION
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-android:${{ needs.release-dev-builder-images.outputs.version }} \
+            docker://${{ vars.ECR_IMAGE_REGISTRY }}/${{ vars.ECR_IMAGE_NAMESPACE }}/dev-builder-android:${{ needs.release-dev-builder-images.outputs.version }}
 
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             quay.io/skopeo/stable:latest \
-            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-android:latest \
-            docker://$ECR_IMAGE_REGISTRY/$ECR_IMAGE_NAMESPACE/dev-builder-android:latest
-
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-android:latest \
+            docker://${{ vars.ECR_IMAGE_REGISTRY }}/${{ vars.ECR_IMAGE_NAMESPACE }}/dev-builder-android:latest
   release-dev-builder-images-cn: # Note: Be careful issue: https://github.com/containers/skopeo/issues/1874 and we decide to use the latest stable skopeo container.
     name: Release dev builder images to CN region
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-20.04
     needs: [
       release-dev-builder-images
     ]
@@ -163,41 +144,29 @@ jobs:
       - name: Push dev-builder-ubuntu image
         shell: bash
         if: ${{ inputs.release_dev_builder_ubuntu_image }}
-        env:
-          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
-          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
-          ACR_IMAGE_REGISTRY: ${{ vars.ACR_IMAGE_REGISTRY }}
         run: |
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             quay.io/skopeo/stable:latest \
-            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-ubuntu:$IMAGE_VERSION \
-            docker://$ACR_IMAGE_REGISTRY/$IMAGE_NAMESPACE/dev-builder-ubuntu:$IMAGE_VERSION
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-ubuntu:${{ needs.release-dev-builder-images.outputs.version }} \
+            docker://${{ vars.ACR_IMAGE_REGISTRY }}/${{ vars.IMAGE_NAMESPACE }}/dev-builder-ubuntu:${{ needs.release-dev-builder-images.outputs.version }}
 
       - name: Push dev-builder-centos image
         shell: bash
         if: ${{ inputs.release_dev_builder_centos_image }}
-        env:
-          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
-          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
-          ACR_IMAGE_REGISTRY: ${{ vars.ACR_IMAGE_REGISTRY }}
         run: |
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             quay.io/skopeo/stable:latest \
-            copy -a docker://docker.io/$IMAGE_NAMESPACE/dev-builder-centos:$IMAGE_VERSION \
-            docker://$ACR_IMAGE_REGISTRY/$IMAGE_NAMESPACE/dev-builder-centos:$IMAGE_VERSION
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-centos:${{ needs.release-dev-builder-images.outputs.version }} \
+            docker://${{ vars.ACR_IMAGE_REGISTRY }}/${{ vars.IMAGE_NAMESPACE }}/dev-builder-centos:${{ needs.release-dev-builder-images.outputs.version }}
 
       - name: Push dev-builder-android image
         shell: bash
         if: ${{ inputs.release_dev_builder_android_image }}
-        env:
-          IMAGE_VERSION: ${{ needs.release-dev-builder-images.outputs.version }}
-          IMAGE_NAMESPACE: ${{ vars.IMAGE_NAMESPACE }}
-          ACR_IMAGE_REGISTRY: ${{ vars.ACR_IMAGE_REGISTRY }}
         run: |
           docker run -v "${DOCKER_CONFIG:-$HOME/.docker}:/root/.docker:ro" \
             -e "REGISTRY_AUTH_FILE=/root/.docker/config.json" \
             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
+            copy -a docker://docker.io/${{ vars.IMAGE_NAMESPACE }}/dev-builder-android:${{ needs.release-dev-builder-images.outputs.version }} \
+            docker://${{ vars.ACR_IMAGE_REGISTRY }}/${{ vars.IMAGE_NAMESPACE }}/dev-builder-android:${{ needs.release-dev-builder-images.outputs.version }}

.github/workflows/release.yml

@@ -91,7 +91,12 @@ env:
   # 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.13.0
+  NEXT_RELEASE_VERSION: v0.12.0
+
+# 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.
+  contents: write # Allows the action to create a release.
 
 jobs:
   allocate-runners:
@@ -117,7 +122,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Check Rust toolchain version
         shell: bash
@@ -177,7 +181,6 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - uses: ./.github/actions/build-linux-artifacts
         with:
@@ -199,7 +202,6 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - uses: ./.github/actions/build-linux-artifacts
         with:
@@ -235,7 +237,6 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - uses: ./.github/actions/build-macos-artifacts
         with:
@@ -275,7 +276,6 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - uses: ./.github/actions/build-windows-artifacts
         with:
@@ -306,18 +306,15 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Build and push images to dockerhub
         uses: ./.github/actions/build-images
         with:
           image-registry: docker.io
           image-namespace: ${{ vars.IMAGE_NAMESPACE }}
-          image-name: ${{ vars.GREPTIMEDB_IMAGE_NAME }}
           image-registry-username: ${{ secrets.DOCKERHUB_USERNAME }}
           image-registry-password: ${{ secrets.DOCKERHUB_TOKEN }}
           version: ${{ needs.allocate-runners.outputs.version }}
-          push-latest-tag: true
 
       - name: Set build image result
         id: set-build-image-result
@@ -344,14 +341,13 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Release artifacts to CN region
         uses: ./.github/actions/release-cn-artifacts
         with:
           src-image-registry: docker.io
           src-image-namespace: ${{ vars.IMAGE_NAMESPACE }}
-          src-image-name: ${{ vars.GREPTIMEDB_IMAGE_NAME }}
+          src-image-name: greptimedb
           dst-image-registry-username: ${{ secrets.ALICLOUD_USERNAME }}
           dst-image-registry-password: ${{ secrets.ALICLOUD_PASSWORD }}
           dst-image-registry: ${{ vars.ACR_IMAGE_REGISTRY }}
@@ -362,7 +358,6 @@ jobs:
           aws-cn-secret-access-key: ${{ secrets.AWS_CN_SECRET_ACCESS_KEY }}
           aws-cn-region: ${{ vars.AWS_RELEASE_BUCKET_REGION }}
           dev-mode: false
-          upload-to-s3: true
           update-version-info: true
           push-latest-tag: true
 
@@ -382,7 +377,6 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Publish GitHub release
         uses: ./.github/actions/publish-github-release
@@ -406,7 +400,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Stop EC2 runner
         uses: ./.github/actions/stop-runner
@@ -432,7 +425,6 @@ jobs:
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
-          persist-credentials: false
 
       - name: Stop EC2 runner
         uses: ./.github/actions/stop-runner
@@ -449,15 +441,8 @@ jobs:
     if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
     needs: [allocate-runners]
     runs-on: ubuntu-20.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.
-      contents: write # Allows the action to create a release.
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - name: Bump doc version
         working-directory: cyborg
@@ -476,17 +461,10 @@ jobs:
       build-windows-artifacts,
     ]
     runs-on: ubuntu-20.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.
-      contents: write # Allows the action to create a release.
     env:
       SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL_DEVELOP_CHANNEL }}
     steps:
       - uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - name: Report CI status
         id: report-ci-status

.github/workflows/schedule.yml

@@ -4,20 +4,18 @@ on:
     - cron: '4 2 * * *'
   workflow_dispatch:
 
+permissions:
+  contents: read
+  issues: write
+  pull-requests: write
 
 jobs:
   maintenance:
     name: Periodic Maintenance
     runs-on: ubuntu-latest
-    permissions:
-      contents: read
-      issues: write
-      pull-requests: write
     if: ${{ github.repository == 'GreptimeTeam/greptimedb' }}
     steps:
       - uses: actions/checkout@v4
-        with:
-          persist-credentials: false
       - uses: ./.github/actions/setup-cyborg
       - name: Do Maintenance
         working-directory: cyborg

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

@@ -1,24 +1,18 @@
 name: "Semantic Pull Request"
 
 on:
-  pull_request:
+  pull_request_target:
     types:
       - opened
       - reopened
       - edited
 
-concurrency:
-  group: ${{ github.workflow }}-${{ github.head_ref || github.run_id }}
-  cancel-in-progress: true
-
 jobs:
   check:
     runs-on: ubuntu-20.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

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)

Cargo.toml

@@ -67,7 +67,7 @@ members = [
 resolver = "2"
 
 [workspace.package]
-version = "0.13.0"
+version = "0.12.0"
 edition = "2021"
 license = "Apache-2.0"
 
@@ -81,7 +81,6 @@ rust.unknown_lints = "deny"
 rust.unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tokio_unstable)'] }
 
 [workspace.dependencies]
-# DO_NOT_REMOVE_THIS: BEGIN_OF_EXTERNAL_DEPENDENCIES
 # We turn off default-features for some dependencies here so the workspaces which inherit them can
 # selectively turn them on if needed, since we can override default-features = true (from false)
 # for the inherited dependency but cannot do the reverse (override from true to false).
@@ -107,7 +106,6 @@ bitflags = "2.4.1"
 bytemuck = "1.12"
 bytes = { version = "1.7", features = ["serde"] }
 chrono = { version = "0.4", features = ["serde"] }
-chrono-tz = "0.10.1"
 clap = { version = "4.4", features = ["derive"] }
 config = "0.13.0"
 crossbeam-utils = "0.8"
@@ -129,7 +127,8 @@ 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" }
+# branch: poc-write-path
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "1915576b113a494f5352fd61f211d899b7f87aab" }
 hex = "0.4"
 http = "1"
 humantime = "2.1"
@@ -140,8 +139,8 @@ itertools = "0.10"
 jsonb = { git = "https://github.com/databendlabs/jsonb.git", rev = "8c8d2fc294a39f3ff08909d60f718639cfba3875", default-features = false }
 lazy_static = "1.4"
 local-ip-address = "0.6"
-loki-proto = { git = "https://github.com/GreptimeTeam/loki-proto.git", rev = "1434ecf23a2654025d86188fb5205e7a74b225d3" }
-meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "5618e779cf2bb4755b499c630fba4c35e91898cb" }
+loki-api = { git = "https://github.com/shuiyisong/tracing-loki", branch = "chore/prost_version" }
+meter-core = { git = "https://github.com/GreptimeTeam/greptime-meter.git", rev = "a10facb353b41460eeb98578868ebf19c2084fac" }
 mockall = "0.11.4"
 moka = "0.12"
 nalgebra = "0.33"
@@ -160,9 +159,7 @@ parquet = { version = "53.0.0", default-features = false, features = ["arrow", "
 paste = "1.0"
 pin-project = "1.0"
 prometheus = { version = "0.13.3", features = ["process"] }
-promql-parser = { git = "https://github.com/GreptimeTeam/promql-parser.git", features = [
-    "ser",
-], rev = "27abb8e16003a50c720f00d6c85f41f5fa2a2a8e" }
+promql-parser = { version = "0.4.3", features = ["ser"] }
 prost = "0.13"
 raft-engine = { version = "0.4.1", default-features = false }
 rand = "0.8"
@@ -211,7 +208,6 @@ tracing-subscriber = { version = "0.3", features = ["env-filter", "json", "fmt"]
 typetag = "0.2"
 uuid = { version = "1.7", features = ["serde", "v4", "fast-rng"] }
 zstd = "0.13"
-# DO_NOT_REMOVE_THIS: END_OF_EXTERNAL_DEPENDENCIES
 
 ## workspaces members
 api = { path = "src/api" }
@@ -283,10 +279,12 @@ tokio-rustls = { git = "https://github.com/GreptimeTeam/tokio-rustls", rev = "46
 # This is commented, since we are not using aws-lc-sys, if we need to use it, we need to uncomment this line or use a release after this commit, or it wouldn't compile with gcc < 8.1
 # see https://github.com/aws/aws-lc-rs/pull/526
 # aws-lc-sys = { git ="https://github.com/aws/aws-lc-rs", rev = "556558441e3494af4b156ae95ebc07ebc2fd38aa" }
+# Apply a fix for pprof for unaligned pointer access
+pprof = { git = "https://github.com/GreptimeTeam/pprof-rs", rev = "1bd1e21" }
 
 [workspace.dependencies.meter-macros]
 git = "https://github.com/GreptimeTeam/greptime-meter.git"
-rev = "5618e779cf2bb4755b499c630fba4c35e91898cb"
+rev = "a10facb353b41460eeb98578868ebf19c2084fac"
 
 [profile.release]
 debug = 1

Cross.toml

@@ -1,6 +1,3 @@
-[target.aarch64-unknown-linux-gnu]
-image = "ghcr.io/cross-rs/aarch64-unknown-linux-gnu:0.2.5"
-
 [build]
 pre-build = [
     "dpkg --add-architecture $CROSS_DEB_ARCH",
@@ -8,8 +5,3 @@ pre-build = [
     "curl -LO https://github.com/protocolbuffers/protobuf/releases/download/v3.15.8/protoc-3.15.8-linux-x86_64.zip && unzip protoc-3.15.8-linux-x86_64.zip -d /usr/",
     "chmod a+x /usr/bin/protoc && chmod -R a+rx /usr/include/google",
 ]
-
-[build.env]
-passthrough = [
-    "JEMALLOC_SYS_WITH_LG_PAGE",
-]

Makefile

@@ -60,8 +60,6 @@ ifeq ($(BUILDX_MULTI_PLATFORM_BUILD), all)
 	BUILDX_MULTI_PLATFORM_BUILD_OPTS := --platform linux/amd64,linux/arm64 --push
 else ifeq ($(BUILDX_MULTI_PLATFORM_BUILD), amd64)
 	BUILDX_MULTI_PLATFORM_BUILD_OPTS := --platform linux/amd64 --push
-else ifeq ($(BUILDX_MULTI_PLATFORM_BUILD), arm64)
-        BUILDX_MULTI_PLATFORM_BUILD_OPTS := --platform linux/arm64 --push
 else
 	BUILDX_MULTI_PLATFORM_BUILD_OPTS := -o type=docker
 endif

README.md

@@ -116,7 +116,7 @@ docker run -p 127.0.0.1:4000-4003:4000-4003 \
   --name greptime --rm \
   greptime/greptimedb:latest standalone start \
   --http-addr 0.0.0.0:4000 \
-  --rpc-bind-addr 0.0.0.0:4001 \
+  --rpc-addr 0.0.0.0:4001 \
   --mysql-addr 0.0.0.0:4002 \
   --postgres-addr 0.0.0.0:4003
 ```

config/config.md

@@ -29,7 +29,7 @@
 | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
 | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
 | `grpc` | -- | -- | The gRPC server options. |
-| `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
+| `grpc.addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
 | `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
 | `grpc.tls.mode` | String | `disable` | TLS mode. |
@@ -40,7 +40,6 @@
 | `mysql.enable` | Bool | `true` | Whether to enable. |
 | `mysql.addr` | String | `127.0.0.1:4002` | The addr to bind the MySQL server. |
 | `mysql.runtime_size` | Integer | `2` | The number of server worker threads. |
-| `mysql.keep_alive` | String | `0s` | Server-side keep-alive time.<br/>Set to 0 (default) to disable. |
 | `mysql.tls` | -- | -- | -- |
 | `mysql.tls.mode` | String | `disable` | TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html<br/>- `disable` (default value)<br/>- `prefer`<br/>- `require`<br/>- `verify-ca`<br/>- `verify-full` |
 | `mysql.tls.cert_path` | String | Unset | Certificate file path. |
@@ -50,7 +49,6 @@
 | `postgres.enable` | Bool | `true` | Whether to enable |
 | `postgres.addr` | String | `127.0.0.1:4003` | The addr to bind the PostgresSQL server. |
 | `postgres.runtime_size` | Integer | `2` | The number of server worker threads. |
-| `postgres.keep_alive` | String | `0s` | Server-side keep-alive time.<br/>Set to 0 (default) to disable. |
 | `postgres.tls` | -- | -- | PostgresSQL server TLS options, see `mysql.tls` section. |
 | `postgres.tls.mode` | String | `disable` | TLS mode. |
 | `postgres.tls.cert_path` | String | Unset | Certificate file path. |
@@ -60,8 +58,6 @@
 | `opentsdb.enable` | Bool | `true` | Whether to enable OpenTSDB put in HTTP API. |
 | `influxdb` | -- | -- | InfluxDB protocol options. |
 | `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. |
-| `jaeger` | -- | -- | Jaeger protocol options. |
-| `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. |
 | `prom_store` | -- | -- | Prometheus remote storage options |
 | `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. |
 | `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. |
@@ -69,8 +65,8 @@
 | `wal.provider` | String | `raft_engine` | The provider of the WAL.<br/>- `raft_engine`: the wal is stored in the local file system by raft-engine.<br/>- `kafka`: it's remote wal that data is stored in Kafka. |
 | `wal.dir` | String | Unset | The directory to store the WAL files.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.file_size` | String | `128MB` | The size of the WAL segment file.<br/>**It's only used when the provider is `raft_engine`**. |
-| `wal.purge_threshold` | String | `1GB` | The threshold of the WAL size to trigger a purge.<br/>**It's only used when the provider is `raft_engine`**. |
-| `wal.purge_interval` | String | `1m` | The interval to trigger a purge.<br/>**It's only used when the provider is `raft_engine`**. |
+| `wal.purge_threshold` | String | `1GB` | The threshold of the WAL size to trigger a flush.<br/>**It's only used when the provider is `raft_engine`**. |
+| `wal.purge_interval` | String | `1m` | The interval to trigger a flush.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.read_batch_size` | Integer | `128` | The read batch size.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.sync_write` | Bool | `false` | Whether to use sync write.<br/>**It's only used when the provider is `raft_engine`**. |
 | `wal.enable_log_recycle` | Bool | `true` | Whether to reuse logically truncated log files.<br/>**It's only used when the provider is `raft_engine`**. |
@@ -92,9 +88,8 @@
 | `wal.backoff_deadline` | String | `5mins` | The deadline of retries.<br/>**It's only used when the provider is `kafka`**. |
 | `wal.overwrite_entry_start_id` | Bool | `false` | Ignore missing entries during read WAL.<br/>**It's only used when the provider is `kafka`**.<br/><br/>This option ensures that when Kafka messages are deleted, the system<br/>can still successfully replay memtable data without throwing an<br/>out-of-range error.<br/>However, enabling this option might lead to unexpected data loss,<br/>as the system will skip over missing entries instead of treating<br/>them as critical errors. |
 | `metadata_store` | -- | -- | Metadata storage options. |
-| `metadata_store.file_size` | String | `64MB` | The size of the metadata store log file. |
-| `metadata_store.purge_threshold` | String | `256MB` | The threshold of the metadata store size to trigger a purge. |
-| `metadata_store.purge_interval` | String | `1m` | The interval of the metadata store to trigger a purge. |
+| `metadata_store.file_size` | String | `256MB` | Kv file size in bytes. |
+| `metadata_store.purge_threshold` | String | `4GB` | Kv purge threshold. |
 | `procedure` | -- | -- | Procedure storage options. |
 | `procedure.max_retry_times` | Integer | `3` | Procedure max retry time. |
 | `procedure.retry_delay` | String | `500ms` | Initial retry delay of procedures, increases exponentially |
@@ -152,7 +147,6 @@
 | `region_engine.mito.index` | -- | -- | The options for index in Mito engine. |
 | `region_engine.mito.index.aux_path` | String | `""` | Auxiliary directory path for the index in filesystem, used to store intermediate files for<br/>creating the index and staging files for searching the index, defaults to `{data_home}/index_intermediate`.<br/>The default name for this directory is `index_intermediate` for backward compatibility.<br/><br/>This path contains two subdirectories:<br/>- `__intm`: for storing intermediate files used during creating index.<br/>- `staging`: for storing staging files used during searching index. |
 | `region_engine.mito.index.staging_size` | String | `2GB` | The max capacity of the staging directory. |
-| `region_engine.mito.index.staging_ttl` | String | `7d` | The TTL of the staging directory.<br/>Defaults to 7 days.<br/>Setting it to "0s" to disable TTL. |
 | `region_engine.mito.index.metadata_cache_size` | String | `64MiB` | Cache size for inverted index metadata. |
 | `region_engine.mito.index.content_cache_size` | String | `128MiB` | Cache size for inverted index content. |
 | `region_engine.mito.index.content_cache_page_size` | String | `64KiB` | Page size for inverted index content cache. |
@@ -227,8 +221,8 @@
 | `http.enable_cors` | Bool | `true` | HTTP CORS support, it's turned on by default<br/>This allows browser to access http APIs without CORS restrictions |
 | `http.cors_allowed_origins` | Array | Unset | Customize allowed origins for HTTP CORS. |
 | `grpc` | -- | -- | The gRPC server options. |
-| `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
-| `grpc.server_addr` | String | `127.0.0.1:4001` | The address advertised to the metasrv, and used for connections from outside the host.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `grpc.bind_addr`. |
+| `grpc.addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
+| `grpc.hostname` | String | `127.0.0.1:4001` | The hostname advertised to the metasrv,<br/>and used for connections from outside the host |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
 | `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
 | `grpc.tls.mode` | String | `disable` | TLS mode. |
@@ -239,7 +233,6 @@
 | `mysql.enable` | Bool | `true` | Whether to enable. |
 | `mysql.addr` | String | `127.0.0.1:4002` | The addr to bind the MySQL server. |
 | `mysql.runtime_size` | Integer | `2` | The number of server worker threads. |
-| `mysql.keep_alive` | String | `0s` | Server-side keep-alive time.<br/>Set to 0 (default) to disable. |
 | `mysql.tls` | -- | -- | -- |
 | `mysql.tls.mode` | String | `disable` | TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html<br/>- `disable` (default value)<br/>- `prefer`<br/>- `require`<br/>- `verify-ca`<br/>- `verify-full` |
 | `mysql.tls.cert_path` | String | Unset | Certificate file path. |
@@ -249,7 +242,6 @@
 | `postgres.enable` | Bool | `true` | Whether to enable |
 | `postgres.addr` | String | `127.0.0.1:4003` | The addr to bind the PostgresSQL server. |
 | `postgres.runtime_size` | Integer | `2` | The number of server worker threads. |
-| `postgres.keep_alive` | String | `0s` | Server-side keep-alive time.<br/>Set to 0 (default) to disable. |
 | `postgres.tls` | -- | -- | PostgresSQL server TLS options, see `mysql.tls` section. |
 | `postgres.tls.mode` | String | `disable` | TLS mode. |
 | `postgres.tls.cert_path` | String | Unset | Certificate file path. |
@@ -259,8 +251,6 @@
 | `opentsdb.enable` | Bool | `true` | Whether to enable OpenTSDB put in HTTP API. |
 | `influxdb` | -- | -- | InfluxDB protocol options. |
 | `influxdb.enable` | Bool | `true` | Whether to enable InfluxDB protocol in HTTP API. |
-| `jaeger` | -- | -- | Jaeger protocol options. |
-| `jaeger.enable` | Bool | `true` | Whether to enable Jaeger protocol in HTTP API. |
 | `prom_store` | -- | -- | Prometheus remote storage options |
 | `prom_store.enable` | Bool | `true` | Whether to enable Prometheus remote write and read in HTTP API. |
 | `prom_store.with_metric_engine` | Bool | `true` | Whether to store the data from Prometheus remote write in metric engine. |
@@ -310,7 +300,7 @@
 | --- | -----| ------- | ----------- |
 | `data_home` | String | `/tmp/metasrv/` | The working home directory. |
 | `bind_addr` | String | `127.0.0.1:3002` | The bind address of metasrv. |
-| `server_addr` | String | `127.0.0.1:3002` | The communication server address for the frontend and datanode to connect to metasrv.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `bind_addr`. |
+| `server_addr` | String | `127.0.0.1:3002` | The communication server address for frontend and datanode to connect to metasrv,  "127.0.0.1:3002" by default for localhost. |
 | `store_addrs` | Array | -- | Store server address default to etcd store.<br/>For postgres store, the format is:<br/>"password=password dbname=postgres user=postgres host=localhost port=5432"<br/>For etcd store, the format is:<br/>"127.0.0.1:2379" |
 | `store_key_prefix` | String | `""` | If it's not empty, the metasrv will store all data with this key prefix. |
 | `backend` | String | `etcd_store` | The datastore for meta server.<br/>Available values:<br/>- `etcd_store` (default value)<br/>- `memory_store`<br/>- `postgres_store` |
@@ -319,7 +309,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. |
@@ -387,14 +376,19 @@
 | `init_regions_in_background` | Bool | `false` | Initialize all regions in the background during the startup.<br/>By default, it provides services after all regions have been initialized. |
 | `init_regions_parallelism` | Integer | `16` | Parallelism of initializing regions. |
 | `max_concurrent_queries` | Integer | `0` | The maximum current queries allowed to be executed. Zero means unlimited. |
+| `rpc_addr` | String | Unset | Deprecated, use `grpc.addr` instead. |
+| `rpc_hostname` | String | Unset | Deprecated, use `grpc.hostname` instead. |
+| `rpc_runtime_size` | Integer | Unset | Deprecated, use `grpc.runtime_size` instead. |
+| `rpc_max_recv_message_size` | String | Unset | Deprecated, use `grpc.rpc_max_recv_message_size` instead. |
+| `rpc_max_send_message_size` | String | Unset | Deprecated, use `grpc.rpc_max_send_message_size` instead. |
 | `enable_telemetry` | Bool | `true` | Enable telemetry to collect anonymous usage data. Enabled by default. |
 | `http` | -- | -- | The HTTP server options. |
 | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
 | `http.timeout` | String | `30s` | HTTP request timeout. Set to 0 to disable timeout. |
 | `http.body_limit` | String | `64MB` | HTTP request body limit.<br/>The following units are supported: `B`, `KB`, `KiB`, `MB`, `MiB`, `GB`, `GiB`, `TB`, `TiB`, `PB`, `PiB`.<br/>Set to 0 to disable limit. |
 | `grpc` | -- | -- | The gRPC server options. |
-| `grpc.bind_addr` | String | `127.0.0.1:3001` | The address to bind the gRPC server. |
-| `grpc.server_addr` | String | `127.0.0.1:3001` | The address advertised to the metasrv, and used for connections from outside the host.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `grpc.bind_addr`. |
+| `grpc.addr` | String | `127.0.0.1:3001` | The address to bind the gRPC server. |
+| `grpc.hostname` | String | `127.0.0.1:3001` | The hostname advertised to the metasrv,<br/>and used for connections from outside the host |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
 | `grpc.max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
 | `grpc.max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |
@@ -493,7 +487,6 @@
 | `region_engine.mito.index` | -- | -- | The options for index in Mito engine. |
 | `region_engine.mito.index.aux_path` | String | `""` | Auxiliary directory path for the index in filesystem, used to store intermediate files for<br/>creating the index and staging files for searching the index, defaults to `{data_home}/index_intermediate`.<br/>The default name for this directory is `index_intermediate` for backward compatibility.<br/><br/>This path contains two subdirectories:<br/>- `__intm`: for storing intermediate files used during creating index.<br/>- `staging`: for storing staging files used during searching index. |
 | `region_engine.mito.index.staging_size` | String | `2GB` | The max capacity of the staging directory. |
-| `region_engine.mito.index.staging_ttl` | String | `7d` | The TTL of the staging directory.<br/>Defaults to 7 days.<br/>Setting it to "0s" to disable TTL. |
 | `region_engine.mito.index.metadata_cache_size` | String | `64MiB` | Cache size for inverted index metadata. |
 | `region_engine.mito.index.content_cache_size` | String | `128MiB` | Cache size for inverted index content. |
 | `region_engine.mito.index.content_cache_page_size` | String | `64KiB` | Page size for inverted index content cache. |
@@ -556,8 +549,8 @@
 | `flow` | -- | -- | flow engine options. |
 | `flow.num_workers` | Integer | `0` | The number of flow worker in flownode.<br/>Not setting(or set to 0) this value will use the number of CPU cores divided by 2. |
 | `grpc` | -- | -- | The gRPC server options. |
-| `grpc.bind_addr` | String | `127.0.0.1:6800` | The address to bind the gRPC server. |
-| `grpc.server_addr` | String | `127.0.0.1:6800` | The address advertised to the metasrv,<br/>and used for connections from outside the host |
+| `grpc.addr` | String | `127.0.0.1:6800` | The address to bind the gRPC server. |
+| `grpc.hostname` | String | `127.0.0.1` | The hostname advertised to the metasrv,<br/>and used for connections from outside the host |
 | `grpc.runtime_size` | Integer | `2` | The number of server worker threads. |
 | `grpc.max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
 | `grpc.max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |

config/datanode.example.toml

@@ -19,6 +19,26 @@ init_regions_parallelism = 16
 ## The maximum current queries allowed to be executed. Zero means unlimited.
 max_concurrent_queries = 0
 
+## Deprecated, use `grpc.addr` instead.
+## @toml2docs:none-default
+rpc_addr = "127.0.0.1:3001"
+
+## Deprecated, use `grpc.hostname` instead.
+## @toml2docs:none-default
+rpc_hostname = "127.0.0.1"
+
+## Deprecated, use `grpc.runtime_size` instead.
+## @toml2docs:none-default
+rpc_runtime_size = 8
+
+## Deprecated, use `grpc.rpc_max_recv_message_size` instead.
+## @toml2docs:none-default
+rpc_max_recv_message_size = "512MB"
+
+## Deprecated, use `grpc.rpc_max_send_message_size` instead.
+## @toml2docs:none-default
+rpc_max_send_message_size = "512MB"
+
 ## Enable telemetry to collect anonymous usage data. Enabled by default.
 #+ enable_telemetry = true
 
@@ -36,11 +56,10 @@ body_limit = "64MB"
 ## The gRPC server options.
 [grpc]
 ## The address to bind the gRPC server.
-bind_addr = "127.0.0.1:3001"
-## The address advertised to the metasrv, and used for connections from outside the host.
-## If left empty or unset, the server will automatically use the IP address of the first network interface
-## on the host, with the same port number as the one specified in `grpc.bind_addr`.
-server_addr = "127.0.0.1:3001"
+addr = "127.0.0.1:3001"
+## The hostname advertised to the metasrv,
+## and used for connections from outside the host
+hostname = "127.0.0.1:3001"
 ## The number of server worker threads.
 runtime_size = 8
 ## The maximum receive message size for gRPC server.
@@ -497,11 +516,6 @@ aux_path = ""
 ## The max capacity of the staging directory.
 staging_size = "2GB"
 
-## The TTL of the staging directory.
-## Defaults to 7 days.
-## Setting it to "0s" to disable TTL.
-staging_ttl = "7d"
-
 ## Cache size for inverted index metadata.
 metadata_cache_size = "64MiB"
 

config/flownode.example.toml

@@ -14,10 +14,10 @@ node_id = 14
 ## The gRPC server options.
 [grpc]
 ## The address to bind the gRPC server.
-bind_addr = "127.0.0.1:6800"
-## The address advertised to the metasrv,
+addr = "127.0.0.1:6800"
+## The hostname advertised to the metasrv,
 ## and used for connections from outside the host
-server_addr = "127.0.0.1:6800"
+hostname = "127.0.0.1"
 ## The number of server worker threads.
 runtime_size = 2
 ## The maximum receive message size for gRPC server.

config/frontend.example.toml

@@ -41,11 +41,10 @@ cors_allowed_origins = ["https://example.com"]
 ## The gRPC server options.
 [grpc]
 ## The address to bind the gRPC server.
-bind_addr = "127.0.0.1:4001"
-## The address advertised to the metasrv, and used for connections from outside the host.
-## If left empty or unset, the server will automatically use the IP address of the first network interface
-## on the host, with the same port number as the one specified in `grpc.bind_addr`.
-server_addr = "127.0.0.1:4001"
+addr = "127.0.0.1:4001"
+## The hostname advertised to the metasrv,
+## and used for connections from outside the host
+hostname = "127.0.0.1:4001"
 ## The number of server worker threads.
 runtime_size = 8
 
@@ -74,9 +73,6 @@ enable = true
 addr = "127.0.0.1:4002"
 ## The number of server worker threads.
 runtime_size = 2
-## Server-side keep-alive time.
-## Set to 0 (default) to disable.
-keep_alive = "0s"
 
 # MySQL server TLS options.
 [mysql.tls]
@@ -108,9 +104,6 @@ enable = true
 addr = "127.0.0.1:4003"
 ## The number of server worker threads.
 runtime_size = 2
-## Server-side keep-alive time.
-## Set to 0 (default) to disable.
-keep_alive = "0s"
 
 ## PostgresSQL server TLS options, see `mysql.tls` section.
 [postgres.tls]
@@ -138,11 +131,6 @@ enable = true
 ## Whether to enable InfluxDB protocol in HTTP API.
 enable = true
 
-## Jaeger protocol options.
-[jaeger]
-## Whether to enable Jaeger protocol in HTTP API.
-enable = true
-
 ## Prometheus remote storage options
 [prom_store]
 ## Whether to enable Prometheus remote write and read in HTTP API.

config/metasrv.example.toml

@@ -4,9 +4,7 @@ data_home = "/tmp/metasrv/"
 ## The bind address of metasrv.
 bind_addr = "127.0.0.1:3002"
 
-## The communication server address for the frontend and datanode to connect to metasrv.
-## If left empty or unset, the server will automatically use the IP address of the first network interface
-## on the host, with the same port number as the one specified in `bind_addr`.
+## The communication server address for frontend and datanode to connect to metasrv,  "127.0.0.1:3002" by default for localhost.
 server_addr = "127.0.0.1:3002"
 
 ## Store server address default to etcd store.
@@ -50,9 +48,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
 

config/standalone.example.toml

@@ -49,7 +49,7 @@ cors_allowed_origins = ["https://example.com"]
 ## The gRPC server options.
 [grpc]
 ## The address to bind the gRPC server.
-bind_addr = "127.0.0.1:4001"
+addr = "127.0.0.1:4001"
 ## The number of server worker threads.
 runtime_size = 8
 
@@ -78,9 +78,6 @@ enable = true
 addr = "127.0.0.1:4002"
 ## The number of server worker threads.
 runtime_size = 2
-## Server-side keep-alive time.
-## Set to 0 (default) to disable.
-keep_alive = "0s"
 
 # MySQL server TLS options.
 [mysql.tls]
@@ -112,9 +109,6 @@ enable = true
 addr = "127.0.0.1:4003"
 ## The number of server worker threads.
 runtime_size = 2
-## Server-side keep-alive time.
-## Set to 0 (default) to disable.
-keep_alive = "0s"
 
 ## PostgresSQL server TLS options, see `mysql.tls` section.
 [postgres.tls]
@@ -142,11 +136,6 @@ enable = true
 ## Whether to enable InfluxDB protocol in HTTP API.
 enable = true
 
-## Jaeger protocol options.
-[jaeger]
-## Whether to enable Jaeger protocol in HTTP API.
-enable = true
-
 ## Prometheus remote storage options
 [prom_store]
 ## Whether to enable Prometheus remote write and read in HTTP API.
@@ -170,11 +159,11 @@ dir = "/tmp/greptimedb/wal"
 ## **It's only used when the provider is `raft_engine`**.
 file_size = "128MB"
 
-## The threshold of the WAL size to trigger a purge.
+## The threshold of the WAL size to trigger a flush.
 ## **It's only used when the provider is `raft_engine`**.
 purge_threshold = "1GB"
 
-## The interval to trigger a purge.
+## The interval to trigger a flush.
 ## **It's only used when the provider is `raft_engine`**.
 purge_interval = "1m"
 
@@ -289,12 +278,10 @@ overwrite_entry_start_id = false
 
 ## Metadata storage options.
 [metadata_store]
-## The size of the metadata store log file.
-file_size = "64MB"
-## The threshold of the metadata store size to trigger a purge.
-purge_threshold = "256MB"
-## The interval of the metadata store to trigger a purge.
-purge_interval = "1m"
+## Kv file size in bytes.
+file_size = "256MB"
+## Kv purge threshold.
+purge_threshold = "4GB"
 
 ## Procedure storage options.
 [procedure]
@@ -584,11 +571,6 @@ aux_path = ""
 ## The max capacity of the staging directory.
 staging_size = "2GB"
 
-## The TTL of the staging directory.
-## Defaults to 7 days.
-## Setting it to "0s" to disable TTL.
-staging_ttl = "7d"
-
 ## Cache size for inverted index metadata.
 metadata_cache_size = "64MiB"
 

docker/dev-builder/ubuntu/Dockerfile

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

docker/dev-builder/ubuntu/Dockerfile-18.10

@@ -0,0 +1,51 @@
+# Use the legacy glibc 2.28.
+FROM ubuntu:18.10
+
+ENV LANG en_US.utf8
+WORKDIR /greptimedb
+
+# Use old-releases.ubuntu.com to avoid 404s: https://help.ubuntu.com/community/EOLUpgrades.
+RUN echo "deb http://old-releases.ubuntu.com/ubuntu/ cosmic main restricted universe multiverse\n\
+deb http://old-releases.ubuntu.com/ubuntu/ cosmic-updates main restricted universe multiverse\n\
+deb http://old-releases.ubuntu.com/ubuntu/ cosmic-security main restricted universe multiverse" > /etc/apt/sources.list
+
+# Install dependencies.
+RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
+    libssl-dev \
+    tzdata \
+    curl \
+    ca-certificates \
+    git \
+    build-essential \
+    unzip \
+    pkg-config
+
+# Install protoc.
+ENV PROTOC_VERSION=29.3
+RUN if [ "$(uname -m)" = "x86_64" ]; then \
+        PROTOC_ZIP=protoc-${PROTOC_VERSION}-linux-x86_64.zip; \
+    elif [ "$(uname -m)" = "aarch64" ]; then \
+        PROTOC_ZIP=protoc-${PROTOC_VERSION}-linux-aarch_64.zip; \
+    else \
+        echo "Unsupported architecture"; exit 1; \
+    fi && \
+    curl -OL https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOC_VERSION}/${PROTOC_ZIP} && \
+    unzip -o ${PROTOC_ZIP} -d /usr/local bin/protoc && \
+    unzip -o ${PROTOC_ZIP} -d /usr/local 'include/*' && \
+    rm -f ${PROTOC_ZIP}
+
+# Install Rust.
+SHELL ["/bin/bash", "-c"]
+RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- --no-modify-path --default-toolchain none -y
+ENV PATH /root/.cargo/bin/:$PATH
+
+# Install Rust toolchains.
+ARG RUST_TOOLCHAIN
+RUN rustup toolchain install ${RUST_TOOLCHAIN}
+
+# Install cargo-binstall with a specific version to adapt the current rust toolchain.
+# Note: if we use the latest version, we may encounter the following `use of unstable library feature 'io_error_downcast'` error.
+RUN cargo install cargo-binstall --version 1.6.6 --locked
+
+# Install nextest.
+RUN cargo binstall cargo-nextest --no-confirm

docker/dev-builder/ubuntu/Dockerfile-20.04

@@ -1,66 +0,0 @@
-FROM ubuntu:20.04
-
-# The root path under which contains all the dependencies to build this Dockerfile.
-ARG DOCKER_BUILD_ROOT=.
-
-ENV LANG en_US.utf8
-WORKDIR /greptimedb
-
-RUN apt-get update && \
-    DEBIAN_FRONTEND=noninteractive apt-get install -y software-properties-common
-# Install dependencies.
-RUN apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y \
-    libssl-dev \
-    tzdata \
-    curl \
-    unzip \
-    ca-certificates \
-    git \
-    build-essential \
-    pkg-config
-
-ARG TARGETPLATFORM
-RUN echo "target platform: $TARGETPLATFORM"
-
-ARG PROTOBUF_VERSION=29.3
-
-# Install protobuf, because the one in the apt is too old (v3.12).
-RUN if [ "$TARGETPLATFORM" = "linux/arm64" ]; then \
-    curl -OL https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOBUF_VERSION}/protoc-${PROTOBUF_VERSION}-linux-aarch_64.zip && \
-    unzip protoc-${PROTOBUF_VERSION}-linux-aarch_64.zip -d protoc3; \
-elif [ "$TARGETPLATFORM" = "linux/amd64" ]; then \
-    curl -OL https://github.com/protocolbuffers/protobuf/releases/download/v${PROTOBUF_VERSION}/protoc-${PROTOBUF_VERSION}-linux-x86_64.zip && \
-    unzip protoc-${PROTOBUF_VERSION}-linux-x86_64.zip -d protoc3; \
-fi
-RUN mv protoc3/bin/* /usr/local/bin/
-RUN mv protoc3/include/* /usr/local/include/
-
-# Silence all `safe.directory` warnings, to avoid the "detect dubious repository" error when building with submodules.
-# Disabling the safe directory check here won't pose extra security issues, because in our usage for this dev build
-# image, we use it solely on our own environment (that github action's VM, or ECS created dynamically by ourselves),
-# and the repositories are pulled from trusted sources (still us, of course). Doing so does not violate the intention
-# of the Git's addition to the "safe.directory" at the first place (see the commit message here:
-# https://github.com/git/git/commit/8959555cee7ec045958f9b6dd62e541affb7e7d9).
-# There's also another solution to this, that we add the desired submodules to the safe directory, instead of using
-# wildcard here. However, that requires the git's config files and the submodules all owned by the very same user.
-# It's troublesome to do this since the dev build runs in Docker, which is under user "root"; while outside the Docker,
-# it can be a different user that have prepared the submodules.
-RUN git config --global --add safe.directory '*'
-
-# Install Rust.
-SHELL ["/bin/bash", "-c"]
-RUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- --no-modify-path --default-toolchain none -y
-ENV PATH /root/.cargo/bin/:$PATH
-
-# Install Rust toolchains.
-ARG RUST_TOOLCHAIN
-RUN rustup toolchain install ${RUST_TOOLCHAIN}
-
-# Install cargo-binstall with a specific version to adapt the current rust toolchain.
-# Note: if we use the latest version, we may encounter the following `use of unstable library feature 'io_error_downcast'` error.
-# compile from source take too long, so we use the precompiled binary instead
-COPY $DOCKER_BUILD_ROOT/docker/dev-builder/binstall/pull_binstall.sh /usr/local/bin/pull_binstall.sh
-RUN chmod +x /usr/local/bin/pull_binstall.sh && /usr/local/bin/pull_binstall.sh
-
-# Install nextest.
-RUN cargo binstall cargo-nextest --no-confirm

docker/docker-compose/cluster-with-etcd.yaml

@@ -43,8 +43,8 @@ services:
     command:
       - metasrv
       - start
-      - --rpc-bind-addr=0.0.0.0:3002
-      - --rpc-server-addr=metasrv:3002
+      - --bind-addr=0.0.0.0:3002
+      - --server-addr=metasrv:3002
       - --store-addrs=etcd0:2379
       - --http-addr=0.0.0.0:3000
     healthcheck:
@@ -68,8 +68,8 @@ services:
       - datanode
       - start
       - --node-id=0
-      - --rpc-bind-addr=0.0.0.0:3001
-      - --rpc-server-addr=datanode0:3001
+      - --rpc-addr=0.0.0.0:3001
+      - --rpc-hostname=datanode0:3001
       - --metasrv-addrs=metasrv:3002
       - --http-addr=0.0.0.0:5000
     volumes:
@@ -98,7 +98,7 @@ services:
       - start
       - --metasrv-addrs=metasrv:3002
       - --http-addr=0.0.0.0:4000
-      - --rpc-bind-addr=0.0.0.0:4001
+      - --rpc-addr=0.0.0.0:4001
       - --mysql-addr=0.0.0.0:4002
       - --postgres-addr=0.0.0.0:4003
     healthcheck:
@@ -123,8 +123,8 @@ services:
       - start
       - --node-id=0
       - --metasrv-addrs=metasrv:3002
-      - --rpc-bind-addr=0.0.0.0:4004
-      - --rpc-server-addr=flownode0:4004
+      - --rpc-addr=0.0.0.0:4004
+      - --rpc-hostname=flownode0:4004
       - --http-addr=0.0.0.0:4005
     depends_on:
       frontend0:

docs/benchmarks/tsbs/v0.12.0.md

@@ -1,40 +0,0 @@
-# TSBS benchmark - v0.12.0
-
-## Environment
-
-### Amazon EC2
-
-|         |                         |
-|---------|-------------------------|
-| Machine | c5d.2xlarge             |
-| CPU     | 8 core                  |
-| Memory  | 16GB                    |
-| Disk    | 100GB (GP3)             |
-| OS      | Ubuntu Server 24.04 LTS |
-
-## Write performance
-
-| Environment     | Ingest rate (rows/s) |
-|-----------------|----------------------|
-| EC2 c5d.2xlarge | 326839.28            |
-
-## Query performance
-
-| Query type            | EC2 c5d.2xlarge (ms) |
-|-----------------------|----------------------|
-| cpu-max-all-1         | 12.46                |
-| cpu-max-all-8         | 24.20                |
-| double-groupby-1      | 673.08               |
-| double-groupby-5      | 963.99               |
-| double-groupby-all    | 1330.05              |
-| groupby-orderby-limit | 952.46               |
-| high-cpu-1            | 5.08                 |
-| high-cpu-all          | 4638.57              |
-| lastpoint             | 591.02               |
-| single-groupby-1-1-1  | 4.06                 |
-| single-groupby-1-1-12 | 4.73                 |
-| single-groupby-1-8-1  | 8.23                 |
-| single-groupby-5-1-1  | 4.61                 |
-| single-groupby-5-1-12 | 5.61                 |
-| single-groupby-5-8-1  | 9.74                 |
-

docs/how-to/how-to-profile-memory.md

@@ -4,16 +4,6 @@ This crate provides an easy approach to dump memory profiling info.
 
 ## Prerequisites
 ### jemalloc
-jeprof is already compiled in the target directory of GreptimeDB. You can find the binary and use it.
-```
-# find jeprof binary
-find . -name 'jeprof'
-# add executable permission
-chmod +x <path_to_jeprof>
-```
-The path is usually under `./target/${PROFILE}/build/tikv-jemalloc-sys-${HASH}/out/build/bin/jeprof`.
-The default version of jemalloc installed from the package manager may not have the `--collapsed` option.
-You may need to check the whether the `jeprof` version is >= `5.3.0` if you want to install it from the package manager.
 ```bash
 # for macOS
 brew install jemalloc
@@ -33,11 +23,7 @@ curl https://raw.githubusercontent.com/brendangregg/FlameGraph/master/flamegraph
 Start GreptimeDB instance with environment variables:
 
 ```bash
-# for Linux
 MALLOC_CONF=prof:true ./target/debug/greptime standalone start
-
-# for macOS
-_RJEM_MALLOC_CONF=prof:true ./target/debug/greptime standalone start
 ```
 
 Dump memory profiling data through HTTP API:

grafana/greptimedb.json

@@ -384,8 +384,8 @@
         "rowHeight": 0.9,
         "showValue": "auto",
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "none",
+          "sort": "none"
         }
       },
       "targets": [
@@ -483,8 +483,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "pluginVersion": "10.2.3",
@@ -578,8 +578,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "pluginVersion": "10.2.3",
@@ -601,7 +601,7 @@
       "type": "timeseries"
     },
     {
-      "collapsed": false,
+      "collapsed": true,
       "gridPos": {
         "h": 1,
         "w": 24,
@@ -684,8 +684,8 @@
               "showLegend": true
             },
             "tooltip": {
-              "mode": "multi",
-              "sort": "desc"
+              "mode": "single",
+              "sort": "none"
             }
           },
           "targets": [
@@ -878,8 +878,8 @@
               "showLegend": true
             },
             "tooltip": {
-              "mode": "multi",
-              "sort": "desc"
+              "mode": "single",
+              "sort": "none"
             }
           },
           "targets": [
@@ -1124,8 +1124,8 @@
               "showLegend": true
             },
             "tooltip": {
-              "mode": "multi",
-              "sort": "desc"
+              "mode": "single",
+              "sort": "none"
             }
           },
           "targets": [
@@ -1223,8 +1223,8 @@
               "showLegend": true
             },
             "tooltip": {
-              "mode": "multi",
-              "sort": "desc"
+              "mode": "single",
+              "sort": "none"
             }
           },
           "targets": [
@@ -1322,8 +1322,8 @@
               "showLegend": true
             },
             "tooltip": {
-              "mode": "multi",
-              "sort": "desc"
+              "mode": "single",
+              "sort": "none"
             }
           },
           "targets": [
@@ -1456,8 +1456,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -1573,8 +1573,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -1673,8 +1673,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -1773,8 +1773,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -1890,8 +1890,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2002,8 +2002,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2120,8 +2120,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2233,8 +2233,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2334,8 +2334,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2435,8 +2435,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2548,8 +2548,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2661,8 +2661,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2788,8 +2788,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2889,8 +2889,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -2990,8 +2990,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -3091,8 +3091,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -3191,8 +3191,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -3302,8 +3302,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -3432,8 +3432,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -3543,8 +3543,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -3657,8 +3657,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -3808,8 +3808,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -3909,8 +3909,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -4011,8 +4011,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [
@@ -4113,8 +4113,8 @@
           "showLegend": true
         },
         "tooltip": {
-          "mode": "multi",
-          "sort": "desc"
+          "mode": "single",
+          "sort": "none"
         }
       },
       "targets": [

src/api/Cargo.toml

@@ -15,10 +15,13 @@ common-macro.workspace = true
 common-time.workspace = true
 datatypes.workspace = true
 greptime-proto.workspace = true
-paste.workspace = true
+paste = "1.0"
 prost.workspace = true
 serde_json.workspace = true
 snafu.workspace = true
 
 [build-dependencies]
 tonic-build = "0.11"
+
+[dev-dependencies]
+paste = "1.0"

src/api/src/v1/column_def.rs

@@ -15,10 +15,10 @@
 use std::collections::HashMap;
 
 use datatypes::schema::{
-    ColumnDefaultConstraint, ColumnSchema, FulltextAnalyzer, FulltextOptions, SkippingIndexOptions,
-    SkippingIndexType, COMMENT_KEY, FULLTEXT_KEY, INVERTED_INDEX_KEY, SKIPPING_INDEX_KEY,
+    ColumnDefaultConstraint, ColumnSchema, FulltextAnalyzer, FulltextOptions, COMMENT_KEY,
+    FULLTEXT_KEY, INVERTED_INDEX_KEY, SKIPPING_INDEX_KEY,
 };
-use greptime_proto::v1::{Analyzer, SkippingIndexType as PbSkippingIndexType};
+use greptime_proto::v1::Analyzer;
 use snafu::ResultExt;
 
 use crate::error::{self, Result};
@@ -103,13 +103,6 @@ pub fn contains_fulltext(options: &Option<ColumnOptions>) -> bool {
         .is_some_and(|o| o.options.contains_key(FULLTEXT_GRPC_KEY))
 }
 
-/// Checks if the `ColumnOptions` contains skipping index options.
-pub fn contains_skipping(options: &Option<ColumnOptions>) -> bool {
-    options
-        .as_ref()
-        .is_some_and(|o| o.options.contains_key(SKIPPING_INDEX_GRPC_KEY))
-}
-
 /// Tries to construct a `ColumnOptions` from the given `FulltextOptions`.
 pub fn options_from_fulltext(fulltext: &FulltextOptions) -> Result<Option<ColumnOptions>> {
     let mut options = ColumnOptions::default();
@@ -120,18 +113,6 @@ pub fn options_from_fulltext(fulltext: &FulltextOptions) -> Result<Option<Column
     Ok((!options.options.is_empty()).then_some(options))
 }
 
-/// Tries to construct a `ColumnOptions` from the given `SkippingIndexOptions`.
-pub fn options_from_skipping(skipping: &SkippingIndexOptions) -> Result<Option<ColumnOptions>> {
-    let mut options = ColumnOptions::default();
-
-    let v = serde_json::to_string(skipping).context(error::SerializeJsonSnafu)?;
-    options
-        .options
-        .insert(SKIPPING_INDEX_GRPC_KEY.to_string(), v);
-
-    Ok((!options.options.is_empty()).then_some(options))
-}
-
 /// Tries to construct a `FulltextAnalyzer` from the given analyzer.
 pub fn as_fulltext_option(analyzer: Analyzer) -> FulltextAnalyzer {
     match analyzer {
@@ -140,13 +121,6 @@ pub fn as_fulltext_option(analyzer: Analyzer) -> FulltextAnalyzer {
     }
 }
 
-/// Tries to construct a `SkippingIndexType` from the given skipping index type.
-pub fn as_skipping_index_type(skipping_index_type: PbSkippingIndexType) -> SkippingIndexType {
-    match skipping_index_type {
-        PbSkippingIndexType::BloomFilter => SkippingIndexType::BloomFilter,
-    }
-}
-
 #[cfg(test)]
 mod tests {
 

src/catalog/Cargo.toml

@@ -15,7 +15,7 @@ api.workspace = true
 arrow.workspace = true
 arrow-schema.workspace = true
 async-stream.workspace = true
-async-trait.workspace = true
+async-trait = "0.1"
 bytes.workspace = true
 common-catalog.workspace = true
 common-error.workspace = true
@@ -31,7 +31,7 @@ common-version.workspace = true
 dashmap.workspace = true
 datafusion.workspace = true
 datatypes.workspace = true
-futures.workspace = true
+futures = "0.3"
 futures-util.workspace = true
 humantime.workspace = true
 itertools.workspace = true
@@ -39,7 +39,7 @@ lazy_static.workspace = true
 meta-client.workspace = true
 moka = { workspace = true, features = ["future", "sync"] }
 partition.workspace = true
-paste.workspace = true
+paste = "1.0"
 prometheus.workspace = true
 rustc-hash.workspace = true
 serde_json.workspace = true
@@ -49,7 +49,7 @@ sql.workspace = true
 store-api.workspace = true
 table.workspace = true
 tokio.workspace = true
-tokio-stream.workspace = true
+tokio-stream = "0.1"
 
 [dev-dependencies]
 cache.workspace = true

src/catalog/src/system_schema/information_schema/key_column_usage.rs

@@ -228,6 +228,12 @@ impl InformationSchemaKeyColumnUsageBuilder {
                 let keys = &table_info.meta.primary_key_indices;
                 let schema = table.schema();
 
+                // For compatibility, use primary key columns as inverted index columns.
+                let pk_as_inverted_index = !schema
+                    .column_schemas()
+                    .iter()
+                    .any(|c| c.has_inverted_index_key());
+
                 for (idx, column) in schema.column_schemas().iter().enumerate() {
                     let mut constraints = vec![];
                     if column.is_time_index() {
@@ -245,6 +251,10 @@ impl InformationSchemaKeyColumnUsageBuilder {
                     // TODO(dimbtp): foreign key constraint not supported yet
                     if keys.contains(&idx) {
                         constraints.push(PRI_CONSTRAINT_NAME);
+
+                        if pk_as_inverted_index {
+                            constraints.push(INVERTED_INDEX_CONSTRAINT_NAME);
+                        }
                     }
                     if column.is_inverted_indexed() {
                         constraints.push(INVERTED_INDEX_CONSTRAINT_NAME);

src/cli/src/bench.rs

@@ -24,11 +24,10 @@ use common_meta::key::{TableMetadataManager, TableMetadataManagerRef};
 use common_meta::kv_backend::etcd::EtcdStore;
 use common_meta::kv_backend::memory::MemoryKvBackend;
 #[cfg(feature = "pg_kvbackend")]
-use common_meta::kv_backend::rds::PgStore;
+use common_meta::kv_backend::postgres::PgStore;
 use common_meta::peer::Peer;
 use common_meta::rpc::router::{Region, RegionRoute};
 use common_telemetry::info;
-use common_wal::options::WalOptions;
 use datatypes::data_type::ConcreteDataType;
 use datatypes::schema::{ColumnSchema, RawSchema};
 use rand::Rng;
@@ -185,7 +184,7 @@ fn create_region_routes(regions: Vec<RegionNumber>) -> Vec<RegionRoute> {
     region_routes
 }
 
-fn create_region_wal_options(regions: Vec<RegionNumber>) -> HashMap<RegionNumber, WalOptions> {
+fn create_region_wal_options(regions: Vec<RegionNumber>) -> HashMap<RegionNumber, String> {
     // TODO(niebayes): construct region wal options for benchmark.
     let _ = regions;
     HashMap::default()

src/cli/src/bench/metadata.rs

@@ -49,12 +49,7 @@ impl TableMetadataBencher {
 
                 let regions: Vec<_> = (0..64).collect();
                 let region_routes = create_region_routes(regions.clone());
-                let region_wal_options = create_region_wal_options(regions)
-                    .into_iter()
-                    .map(|(region_id, wal_options)| {
-                        (region_id, serde_json::to_string(&wal_options).unwrap())
-                    })
-                    .collect();
+                let region_wal_options = create_region_wal_options(regions);
 
                 let start = Instant::now();
 
@@ -114,17 +109,9 @@ impl TableMetadataBencher {
                 let table_info = table_info.unwrap();
                 let table_route = table_route.unwrap();
                 let table_id = table_info.table_info.ident.table_id;
-
-                let regions: Vec<_> = (0..64).collect();
-                let region_wal_options = create_region_wal_options(regions);
                 let _ = self
                     .table_metadata_manager
-                    .delete_table_metadata(
-                        table_id,
-                        &table_info.table_name(),
-                        &table_route,
-                        &region_wal_options,
-                    )
+                    .delete_table_metadata(table_id, &table_info.table_name(), &table_route)
                     .await;
                 start.elapsed()
             },

src/cmd/src/datanode.rs

@@ -126,14 +126,10 @@ impl SubCommand {
 struct StartCommand {
     #[clap(long)]
     node_id: Option<u64>,
-    /// The address to bind the gRPC server.
-    #[clap(long, alias = "rpc-addr")]
-    rpc_bind_addr: Option<String>,
-    /// The address advertised to the metasrv, and used for connections from outside the host.
-    /// If left empty or unset, the server will automatically use the IP address of the first network interface
-    /// on the host, with the same port number as the one specified in `rpc_bind_addr`.
-    #[clap(long, alias = "rpc-hostname")]
-    rpc_server_addr: Option<String>,
+    #[clap(long)]
+    rpc_addr: Option<String>,
+    #[clap(long)]
+    rpc_hostname: Option<String>,
     #[clap(long, value_delimiter = ',', num_args = 1..)]
     metasrv_addrs: Option<Vec<String>>,
     #[clap(short, long)]
@@ -185,18 +181,18 @@ impl StartCommand {
             tokio_console_addr: global_options.tokio_console_addr.clone(),
         };
 
-        if let Some(addr) = &self.rpc_bind_addr {
-            opts.grpc.bind_addr.clone_from(addr);
+        if let Some(addr) = &self.rpc_addr {
+            opts.grpc.addr.clone_from(addr);
         } else if let Some(addr) = &opts.rpc_addr {
             warn!("Use the deprecated attribute `DatanodeOptions.rpc_addr`, please use `grpc.addr` instead.");
-            opts.grpc.bind_addr.clone_from(addr);
+            opts.grpc.addr.clone_from(addr);
         }
 
-        if let Some(server_addr) = &self.rpc_server_addr {
-            opts.grpc.server_addr.clone_from(server_addr);
-        } else if let Some(server_addr) = &opts.rpc_hostname {
+        if let Some(hostname) = &self.rpc_hostname {
+            opts.grpc.hostname.clone_from(hostname);
+        } else if let Some(hostname) = &opts.rpc_hostname {
             warn!("Use the deprecated attribute `DatanodeOptions.rpc_hostname`, please use `grpc.hostname` instead.");
-            opts.grpc.server_addr.clone_from(server_addr);
+            opts.grpc.hostname.clone_from(hostname);
         }
 
         if let Some(runtime_size) = opts.rpc_runtime_size {
@@ -281,7 +277,7 @@ impl StartCommand {
 
         let plugin_opts = opts.plugins;
         let mut opts = opts.component;
-        opts.grpc.detect_server_addr();
+        opts.grpc.detect_hostname();
         let mut plugins = Plugins::new();
         plugins::setup_datanode_plugins(&mut plugins, &plugin_opts, &opts)
             .await
@@ -361,8 +357,8 @@ mod tests {
             rpc_addr = "127.0.0.1:4001"
             rpc_hostname = "192.168.0.1"
             [grpc]
-            bind_addr = "127.0.0.1:3001"
-            server_addr = "127.0.0.1"
+            addr = "127.0.0.1:3001"
+            hostname = "127.0.0.1"
             runtime_size = 8
         "#;
         write!(file, "{}", toml_str).unwrap();
@@ -373,8 +369,8 @@ mod tests {
         };
 
         let options = cmd.load_options(&Default::default()).unwrap().component;
-        assert_eq!("127.0.0.1:4001".to_string(), options.grpc.bind_addr);
-        assert_eq!("192.168.0.1".to_string(), options.grpc.server_addr);
+        assert_eq!("127.0.0.1:4001".to_string(), options.grpc.addr);
+        assert_eq!("192.168.0.1".to_string(), options.grpc.hostname);
     }
 
     #[test]
@@ -435,7 +431,7 @@ mod tests {
 
         let options = cmd.load_options(&Default::default()).unwrap().component;
 
-        assert_eq!("127.0.0.1:3001".to_string(), options.grpc.bind_addr);
+        assert_eq!("127.0.0.1:3001".to_string(), options.grpc.addr);
         assert_eq!(Some(42), options.node_id);
 
         let DatanodeWalConfig::RaftEngine(raft_engine_config) = options.wal else {
@@ -649,7 +645,7 @@ mod tests {
                     opts.http.addr,
                     DatanodeOptions::default().component.http.addr
                 );
-                assert_eq!(opts.grpc.server_addr, "10.103.174.219");
+                assert_eq!(opts.grpc.hostname, "10.103.174.219");
             },
         );
     }

src/cmd/src/flownode.rs

@@ -129,13 +129,11 @@ struct StartCommand {
     #[clap(long)]
     node_id: Option<u64>,
     /// Bind address for the gRPC server.
-    #[clap(long, alias = "rpc-addr")]
-    rpc_bind_addr: Option<String>,
-    /// The address advertised to the metasrv, and used for connections from outside the host.
-    /// If left empty or unset, the server will automatically use the IP address of the first network interface
-    /// on the host, with the same port number as the one specified in `rpc_bind_addr`.
-    #[clap(long, alias = "rpc-hostname")]
-    rpc_server_addr: Option<String>,
+    #[clap(long)]
+    rpc_addr: Option<String>,
+    /// Hostname for the gRPC server.
+    #[clap(long)]
+    rpc_hostname: Option<String>,
     /// Metasrv address list;
     #[clap(long, value_delimiter = ',', num_args = 1..)]
     metasrv_addrs: Option<Vec<String>>,
@@ -186,12 +184,12 @@ impl StartCommand {
             tokio_console_addr: global_options.tokio_console_addr.clone(),
         };
 
-        if let Some(addr) = &self.rpc_bind_addr {
-            opts.grpc.bind_addr.clone_from(addr);
+        if let Some(addr) = &self.rpc_addr {
+            opts.grpc.addr.clone_from(addr);
         }
 
-        if let Some(server_addr) = &self.rpc_server_addr {
-            opts.grpc.server_addr.clone_from(server_addr);
+        if let Some(hostname) = &self.rpc_hostname {
+            opts.grpc.hostname.clone_from(hostname);
         }
 
         if let Some(node_id) = self.node_id {
@@ -239,7 +237,7 @@ impl StartCommand {
         info!("Flownode options: {:#?}", opts);
 
         let mut opts = opts.component;
-        opts.grpc.detect_server_addr();
+        opts.grpc.detect_hostname();
 
         // TODO(discord9): make it not optionale after cluster id is required
         let cluster_id = opts.cluster_id.unwrap_or(0);

src/cmd/src/frontend.rs

@@ -136,19 +136,13 @@ impl SubCommand {
 
 #[derive(Debug, Default, Parser)]
 pub struct StartCommand {
-    /// The address to bind the gRPC server.
-    #[clap(long, alias = "rpc-addr")]
-    rpc_bind_addr: Option<String>,
-    /// The address advertised to the metasrv, and used for connections from outside the host.
-    /// If left empty or unset, the server will automatically use the IP address of the first network interface
-    /// on the host, with the same port number as the one specified in `rpc_bind_addr`.
-    #[clap(long, alias = "rpc-hostname")]
-    rpc_server_addr: Option<String>,
     #[clap(long)]
     http_addr: Option<String>,
     #[clap(long)]
     http_timeout: Option<u64>,
     #[clap(long)]
+    rpc_addr: Option<String>,
+    #[clap(long)]
     mysql_addr: Option<String>,
     #[clap(long)]
     postgres_addr: Option<String>,
@@ -224,15 +218,11 @@ impl StartCommand {
             opts.http.disable_dashboard = disable_dashboard;
         }
 
-        if let Some(addr) = &self.rpc_bind_addr {
-            opts.grpc.bind_addr.clone_from(addr);
+        if let Some(addr) = &self.rpc_addr {
+            opts.grpc.addr.clone_from(addr);
             opts.grpc.tls = tls_opts.clone();
         }
 
-        if let Some(addr) = &self.rpc_server_addr {
-            opts.grpc.server_addr.clone_from(addr);
-        }
-
         if let Some(addr) = &self.mysql_addr {
             opts.mysql.enable = true;
             opts.mysql.addr.clone_from(addr);
@@ -279,7 +269,7 @@ impl StartCommand {
 
         let plugin_opts = opts.plugins;
         let mut opts = opts.component;
-        opts.grpc.detect_server_addr();
+        opts.grpc.detect_hostname();
         let mut plugins = Plugins::new();
         plugins::setup_frontend_plugins(&mut plugins, &plugin_opts, &opts)
             .await
@@ -423,7 +413,7 @@ mod tests {
 
         let default_opts = FrontendOptions::default().component;
 
-        assert_eq!(opts.grpc.bind_addr, default_opts.grpc.bind_addr);
+        assert_eq!(opts.grpc.addr, default_opts.grpc.addr);
         assert!(opts.mysql.enable);
         assert_eq!(opts.mysql.runtime_size, default_opts.mysql.runtime_size);
         assert!(opts.postgres.enable);
@@ -614,7 +604,7 @@ mod tests {
                 assert_eq!(fe_opts.http.addr, "127.0.0.1:14000");
 
                 // Should be default value.
-                assert_eq!(fe_opts.grpc.bind_addr, GrpcOptions::default().bind_addr);
+                assert_eq!(fe_opts.grpc.addr, GrpcOptions::default().addr);
             },
         );
     }

src/cmd/src/metasrv.rs

@@ -42,7 +42,7 @@ pub struct Instance {
 }
 
 impl Instance {
-    pub fn new(instance: MetasrvInstance, guard: Vec<WorkerGuard>) -> Self {
+    fn new(instance: MetasrvInstance, guard: Vec<WorkerGuard>) -> Self {
         Self {
             instance,
             _guard: guard,
@@ -133,15 +133,11 @@ impl SubCommand {
 
 #[derive(Debug, Default, Parser)]
 struct StartCommand {
-    /// The address to bind the gRPC server.
-    #[clap(long, alias = "bind-addr")]
-    rpc_bind_addr: Option<String>,
-    /// The communication server address for the frontend and datanode to connect to metasrv.
-    /// If left empty or unset, the server will automatically use the IP address of the first network interface
-    /// on the host, with the same port number as the one specified in `rpc_bind_addr`.
-    #[clap(long, alias = "server-addr")]
-    rpc_server_addr: Option<String>,
-    #[clap(long, alias = "store-addr", value_delimiter = ',', num_args = 1..)]
+    #[clap(long)]
+    bind_addr: Option<String>,
+    #[clap(long)]
+    server_addr: Option<String>,
+    #[clap(long, aliases = ["store-addr"], value_delimiter = ',', num_args = 1..)]
     store_addrs: Option<Vec<String>>,
     #[clap(short, long)]
     config_file: Option<String>,
@@ -205,11 +201,11 @@ impl StartCommand {
             tokio_console_addr: global_options.tokio_console_addr.clone(),
         };
 
-        if let Some(addr) = &self.rpc_bind_addr {
+        if let Some(addr) = &self.bind_addr {
             opts.bind_addr.clone_from(addr);
         }
 
-        if let Some(addr) = &self.rpc_server_addr {
+        if let Some(addr) = &self.server_addr {
             opts.server_addr.clone_from(addr);
         }
 
@@ -273,13 +269,11 @@ impl StartCommand {
         log_versions(version(), short_version(), APP_NAME);
 
         info!("Metasrv start command: {:#?}", self);
+        info!("Metasrv options: {:#?}", opts);
 
         let plugin_opts = opts.plugins;
         let mut opts = opts.component;
         opts.detect_server_addr();
-
-        info!("Metasrv options: {:#?}", opts);
-
         let mut plugins = Plugins::new();
         plugins::setup_metasrv_plugins(&mut plugins, &plugin_opts, &opts)
             .await
@@ -312,8 +306,8 @@ mod tests {
     #[test]
     fn test_read_from_cmd() {
         let cmd = StartCommand {
-            rpc_bind_addr: Some("127.0.0.1:3002".to_string()),
-            rpc_server_addr: Some("127.0.0.1:3002".to_string()),
+            bind_addr: Some("127.0.0.1:3002".to_string()),
+            server_addr: Some("127.0.0.1:3002".to_string()),
             store_addrs: Some(vec!["127.0.0.1:2380".to_string()]),
             selector: Some("LoadBased".to_string()),
             ..Default::default()
@@ -387,8 +381,8 @@ mod tests {
     #[test]
     fn test_load_log_options_from_cli() {
         let cmd = StartCommand {
-            rpc_bind_addr: Some("127.0.0.1:3002".to_string()),
-            rpc_server_addr: Some("127.0.0.1:3002".to_string()),
+            bind_addr: Some("127.0.0.1:3002".to_string()),
+            server_addr: Some("127.0.0.1:3002".to_string()),
             store_addrs: Some(vec!["127.0.0.1:2380".to_string()]),
             selector: Some("LoadBased".to_string()),
             ..Default::default()

src/cmd/src/standalone.rs

@@ -60,8 +60,7 @@ use frontend::instance::builder::FrontendBuilder;
 use frontend::instance::{FrontendInstance, Instance as FeInstance, StandaloneDatanodeManager};
 use frontend::server::Services;
 use frontend::service_config::{
-    InfluxdbOptions, JaegerOptions, MysqlOptions, OpentsdbOptions, PostgresOptions,
-    PromStoreOptions,
+    InfluxdbOptions, MysqlOptions, OpentsdbOptions, PostgresOptions, PromStoreOptions,
 };
 use meta_srv::metasrv::{FLOW_ID_SEQ, TABLE_ID_SEQ};
 use mito2::config::MitoConfig;
@@ -141,7 +140,6 @@ pub struct StandaloneOptions {
     pub postgres: PostgresOptions,
     pub opentsdb: OpentsdbOptions,
     pub influxdb: InfluxdbOptions,
-    pub jaeger: JaegerOptions,
     pub prom_store: PromStoreOptions,
     pub wal: DatanodeWalConfig,
     pub storage: StorageConfig,
@@ -171,7 +169,6 @@ impl Default for StandaloneOptions {
             postgres: PostgresOptions::default(),
             opentsdb: OpentsdbOptions::default(),
             influxdb: InfluxdbOptions::default(),
-            jaeger: JaegerOptions::default(),
             prom_store: PromStoreOptions::default(),
             wal: DatanodeWalConfig::default(),
             storage: StorageConfig::default(),
@@ -220,7 +217,6 @@ impl StandaloneOptions {
             postgres: cloned_opts.postgres,
             opentsdb: cloned_opts.opentsdb,
             influxdb: cloned_opts.influxdb,
-            jaeger: cloned_opts.jaeger,
             prom_store: cloned_opts.prom_store,
             meta_client: None,
             logging: cloned_opts.logging,
@@ -333,8 +329,8 @@ impl App for Instance {
 pub struct StartCommand {
     #[clap(long)]
     http_addr: Option<String>,
-    #[clap(long, alias = "rpc-addr")]
-    rpc_bind_addr: Option<String>,
+    #[clap(long)]
+    rpc_addr: Option<String>,
     #[clap(long)]
     mysql_addr: Option<String>,
     #[clap(long)]
@@ -411,9 +407,9 @@ impl StartCommand {
             opts.storage.data_home.clone_from(data_home);
         }
 
-        if let Some(addr) = &self.rpc_bind_addr {
+        if let Some(addr) = &self.rpc_addr {
             // frontend grpc addr conflict with datanode default grpc addr
-            let datanode_grpc_addr = DatanodeOptions::default().grpc.bind_addr;
+            let datanode_grpc_addr = DatanodeOptions::default().grpc.addr;
             if addr.eq(&datanode_grpc_addr) {
                 return IllegalConfigSnafu {
                     msg: format!(
@@ -421,7 +417,7 @@ impl StartCommand {
                     ),
                 }.fail();
             }
-            opts.grpc.bind_addr.clone_from(addr)
+            opts.grpc.addr.clone_from(addr)
         }
 
         if let Some(addr) = &self.mysql_addr {
@@ -468,7 +464,7 @@ impl StartCommand {
         let mut plugins = Plugins::new();
         let plugin_opts = opts.plugins;
         let mut opts = opts.component;
-        opts.grpc.detect_server_addr();
+        opts.grpc.detect_hostname();
         let fe_opts = opts.frontend_options();
         let dn_opts = opts.datanode_options();
 
@@ -490,8 +486,8 @@ impl StartCommand {
         let metadata_dir = metadata_store_dir(data_home);
         let (kv_backend, procedure_manager) = FeInstance::try_build_standalone_components(
             metadata_dir,
-            opts.metadata_store,
-            opts.procedure,
+            opts.metadata_store.clone(),
+            opts.procedure.clone(),
         )
         .await
         .context(StartFrontendSnafu)?;
@@ -911,7 +907,7 @@ mod tests {
         assert_eq!("127.0.0.1:4000".to_string(), fe_opts.http.addr);
         assert_eq!(Duration::from_secs(33), fe_opts.http.timeout);
         assert_eq!(ReadableSize::mb(128), fe_opts.http.body_limit);
-        assert_eq!("127.0.0.1:4001".to_string(), fe_opts.grpc.bind_addr);
+        assert_eq!("127.0.0.1:4001".to_string(), fe_opts.grpc.addr);
         assert!(fe_opts.mysql.enable);
         assert_eq!("127.0.0.1:4002", fe_opts.mysql.addr);
         assert_eq!(2, fe_opts.mysql.runtime_size);
@@ -1041,7 +1037,7 @@ mod tests {
                 assert_eq!(ReadableSize::mb(64), fe_opts.http.body_limit);
 
                 // Should be default value.
-                assert_eq!(fe_opts.grpc.bind_addr, GrpcOptions::default().bind_addr);
+                assert_eq!(fe_opts.grpc.addr, GrpcOptions::default().addr);
             },
         );
     }

src/cmd/tests/cli.rs

@@ -63,7 +63,7 @@ mod tests {
             .args([
                 "datanode",
                 "start",
-                "--rpc-bind-addr=0.0.0.0:4321",
+                "--rpc-addr=0.0.0.0:4321",
                 "--node-id=1",
                 &format!("--data-home={}", data_home.path().display()),
                 &format!("--wal-dir={}", wal_dir.path().display()),
@@ -80,7 +80,7 @@ mod tests {
             "--log-level=off",
             "cli",
             "attach",
-            "--grpc-bind-addr=0.0.0.0:4321",
+            "--grpc-addr=0.0.0.0:4321",
             // history commands can sneaky into stdout and mess up our tests, so disable it
             "--disable-helper",
         ]);

src/cmd/tests/load_config_test.rs

@@ -17,6 +17,9 @@ use std::time::Duration;
 use cmd::options::GreptimeOptions;
 use cmd::standalone::StandaloneOptions;
 use common_config::Configurable;
+use common_grpc::channel_manager::{
+    DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE, DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE,
+};
 use common_options::datanode::{ClientOptions, DatanodeClientOptions};
 use common_telemetry::logging::{LoggingOptions, SlowQueryOptions, DEFAULT_OTLP_ENDPOINT};
 use common_wal::config::raft_engine::RaftEngineConfig;
@@ -88,8 +91,13 @@ fn test_load_datanode_example_config() {
                 ..Default::default()
             },
             grpc: GrpcOptions::default()
-                .with_bind_addr("127.0.0.1:3001")
-                .with_server_addr("127.0.0.1:3001"),
+                .with_addr("127.0.0.1:3001")
+                .with_hostname("127.0.0.1:3001"),
+            rpc_addr: Some("127.0.0.1:3001".to_string()),
+            rpc_hostname: Some("127.0.0.1".to_string()),
+            rpc_runtime_size: Some(8),
+            rpc_max_recv_message_size: Some(DEFAULT_MAX_GRPC_RECV_MESSAGE_SIZE),
+            rpc_max_send_message_size: Some(DEFAULT_MAX_GRPC_SEND_MESSAGE_SIZE),
             ..Default::default()
         },
         ..Default::default()
@@ -136,9 +144,7 @@ fn test_load_frontend_example_config() {
                 remote_write: Some(Default::default()),
                 ..Default::default()
             },
-            grpc: GrpcOptions::default()
-                .with_bind_addr("127.0.0.1:4001")
-                .with_server_addr("127.0.0.1:4001"),
+            grpc: GrpcOptions::default().with_hostname("127.0.0.1:4001"),
             http: HttpOptions {
                 cors_allowed_origins: vec!["https://example.com".to_string()],
                 ..Default::default()

src/common/base/Cargo.toml

@@ -18,7 +18,7 @@ bytes.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
 futures.workspace = true
-paste.workspace = true
+paste = "1.0"
 pin-project.workspace = true
 rand.workspace = true
 serde = { version = "1.0", features = ["derive"] }

src/common/config/Cargo.toml

@@ -12,11 +12,9 @@ common-base.workspace = true
 common-error.workspace = true
 common-macro.workspace = true
 config.workspace = true
-humantime-serde.workspace = true
 num_cpus.workspace = true
 serde.workspace = true
 serde_json.workspace = true
-serde_with.workspace = true
 snafu.workspace = true
 sysinfo.workspace = true
 toml.workspace = true

src/common/config/src/lib.rs

@@ -16,8 +16,6 @@ pub mod config;
 pub mod error;
 pub mod utils;
 
-use std::time::Duration;
-
 use common_base::readable_size::ReadableSize;
 pub use config::*;
 use serde::{Deserialize, Serialize};
@@ -36,27 +34,22 @@ pub enum Mode {
     Distributed,
 }
 
-#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(default)]
 pub struct KvBackendConfig {
-    /// The size of the metadata store backend log file.
+    // Kv file size in bytes
     pub file_size: ReadableSize,
-    /// The threshold of the metadata store size to trigger a purge.
+    // Kv purge threshold in bytes
     pub purge_threshold: ReadableSize,
-    /// The interval of the metadata store to trigger a purge.
-    #[serde(with = "humantime_serde")]
-    pub purge_interval: Duration,
 }
 
 impl Default for KvBackendConfig {
     fn default() -> Self {
         Self {
-            // The log file size 64MB
-            file_size: ReadableSize::mb(64),
-            // The log purge threshold 256MB
-            purge_threshold: ReadableSize::mb(256),
-            // The log purge interval 1m
-            purge_interval: Duration::from_secs(60),
+            // log file size 256MB
+            file_size: ReadableSize::mb(256),
+            // purge threshold 4GB
+            purge_threshold: ReadableSize::gb(4),
         }
     }
 }

src/common/datasource/Cargo.toml

@@ -35,7 +35,7 @@ orc-rust = { version = "0.5", default-features = false, features = [
     "async",
 ] }
 parquet.workspace = true
-paste.workspace = true
+paste = "1.0"
 rand.workspace = true
 regex = "1.7"
 serde.workspace = true

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
@@ -28,21 +26,18 @@ common-telemetry.workspace = true
 common-time.workspace = true
 common-version.workspace = true
 datafusion.workspace = true
-datafusion-common.workspace = true
-datafusion-expr.workspace = true
 datatypes.workspace = true
 derive_more = { version = "1", default-features = false, features = ["display"] }
 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"
 num-traits = "0.2"
 once_cell.workspace = true
-paste.workspace = true
+paste = "1.0"
 s2 = { version = "0.0.12", optional = true }
 serde.workspace = true
 serde_json.workspace = true
@@ -52,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]

src/common/function/src/aggr/geo_path.rs

@@ -1,433 +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 datafusion::arrow::array::{Array, ArrayRef};
-use datafusion::common::cast::as_primitive_array;
-use datafusion::error::{DataFusionError, Result as DfResult};
-use datafusion::logical_expr::{Accumulator as DfAccumulator, AggregateUDF, Volatility};
-use datafusion::prelude::create_udaf;
-use datafusion_common::cast::{as_list_array, as_struct_array};
-use datafusion_common::utils::SingleRowListArrayBuilder;
-use datafusion_common::ScalarValue;
-use datatypes::arrow::array::{Float64Array, Int64Array, ListArray, StructArray};
-use datatypes::arrow::datatypes::{
-    DataType, Field, Float64Type, Int64Type, TimeUnit, TimestampNanosecondType,
-};
-use datatypes::compute::{self, sort_to_indices};
-
-pub const GEO_PATH_NAME: &str = "geo_path";
-
-const LATITUDE_FIELD: &str = "lat";
-const LONGITUDE_FIELD: &str = "lng";
-const TIMESTAMP_FIELD: &str = "timestamp";
-const DEFAULT_LIST_FIELD_NAME: &str = "item";
-
-#[derive(Debug, Default)]
-pub struct GeoPathAccumulator {
-    lat: Vec<Option<f64>>,
-    lng: Vec<Option<f64>>,
-    timestamp: Vec<Option<i64>>,
-}
-
-impl GeoPathAccumulator {
-    pub fn new() -> Self {
-        Self::default()
-    }
-
-    pub fn udf_impl() -> AggregateUDF {
-        create_udaf(
-            GEO_PATH_NAME,
-            // Input types: lat, lng, timestamp
-            vec![
-                DataType::Float64,
-                DataType::Float64,
-                DataType::Timestamp(TimeUnit::Nanosecond, None),
-            ],
-            // Output type: list of points {[lat], [lng]}
-            Arc::new(DataType::Struct(
-                vec![
-                    Field::new(
-                        LATITUDE_FIELD,
-                        DataType::List(Arc::new(Field::new(
-                            DEFAULT_LIST_FIELD_NAME,
-                            DataType::Float64,
-                            true,
-                        ))),
-                        false,
-                    ),
-                    Field::new(
-                        LONGITUDE_FIELD,
-                        DataType::List(Arc::new(Field::new(
-                            DEFAULT_LIST_FIELD_NAME,
-                            DataType::Float64,
-                            true,
-                        ))),
-                        false,
-                    ),
-                ]
-                .into(),
-            )),
-            Volatility::Immutable,
-            // Create the accumulator
-            Arc::new(|_| Ok(Box::new(GeoPathAccumulator::new()))),
-            // Intermediate state types
-            Arc::new(vec![DataType::Struct(
-                vec![
-                    Field::new(
-                        LATITUDE_FIELD,
-                        DataType::List(Arc::new(Field::new(
-                            DEFAULT_LIST_FIELD_NAME,
-                            DataType::Float64,
-                            true,
-                        ))),
-                        false,
-                    ),
-                    Field::new(
-                        LONGITUDE_FIELD,
-                        DataType::List(Arc::new(Field::new(
-                            DEFAULT_LIST_FIELD_NAME,
-                            DataType::Float64,
-                            true,
-                        ))),
-                        false,
-                    ),
-                    Field::new(
-                        TIMESTAMP_FIELD,
-                        DataType::List(Arc::new(Field::new(
-                            DEFAULT_LIST_FIELD_NAME,
-                            DataType::Int64,
-                            true,
-                        ))),
-                        false,
-                    ),
-                ]
-                .into(),
-            )]),
-        )
-    }
-}
-
-impl DfAccumulator for GeoPathAccumulator {
-    fn update_batch(&mut self, values: &[ArrayRef]) -> datafusion::error::Result<()> {
-        if values.len() != 3 {
-            return Err(DataFusionError::Internal(format!(
-                "Expected 3 columns for geo_path, got {}",
-                values.len()
-            )));
-        }
-
-        let lat_array = as_primitive_array::<Float64Type>(&values[0])?;
-        let lng_array = as_primitive_array::<Float64Type>(&values[1])?;
-        let ts_array = as_primitive_array::<TimestampNanosecondType>(&values[2])?;
-
-        let size = lat_array.len();
-        self.lat.reserve(size);
-        self.lng.reserve(size);
-
-        for idx in 0..size {
-            self.lat.push(if lat_array.is_null(idx) {
-                None
-            } else {
-                Some(lat_array.value(idx))
-            });
-
-            self.lng.push(if lng_array.is_null(idx) {
-                None
-            } else {
-                Some(lng_array.value(idx))
-            });
-
-            self.timestamp.push(if ts_array.is_null(idx) {
-                None
-            } else {
-                Some(ts_array.value(idx))
-            });
-        }
-
-        Ok(())
-    }
-
-    fn evaluate(&mut self) -> DfResult<ScalarValue> {
-        let unordered_lng_array = Float64Array::from(self.lng.clone());
-        let unordered_lat_array = Float64Array::from(self.lat.clone());
-        let ts_array = Int64Array::from(self.timestamp.clone());
-
-        let ordered_indices = sort_to_indices(&ts_array, None, None)?;
-        let lat_array = compute::take(&unordered_lat_array, &ordered_indices, None)?;
-        let lng_array = compute::take(&unordered_lng_array, &ordered_indices, None)?;
-
-        let lat_list = Arc::new(SingleRowListArrayBuilder::new(lat_array).build_list_array());
-        let lng_list = Arc::new(SingleRowListArrayBuilder::new(lng_array).build_list_array());
-
-        let result = ScalarValue::Struct(Arc::new(StructArray::new(
-            vec![
-                Field::new(
-                    LATITUDE_FIELD,
-                    DataType::List(Arc::new(Field::new("item", DataType::Float64, true))),
-                    false,
-                ),
-                Field::new(
-                    LONGITUDE_FIELD,
-                    DataType::List(Arc::new(Field::new("item", DataType::Float64, true))),
-                    false,
-                ),
-            ]
-            .into(),
-            vec![lat_list, lng_list],
-            None,
-        )));
-
-        Ok(result)
-    }
-
-    fn size(&self) -> usize {
-        // Base size of GeoPathAccumulator struct fields
-        let mut total_size = std::mem::size_of::<Self>();
-
-        // Size of vectors (approximation)
-        total_size += self.lat.capacity() * std::mem::size_of::<Option<f64>>();
-        total_size += self.lng.capacity() * std::mem::size_of::<Option<f64>>();
-        total_size += self.timestamp.capacity() * std::mem::size_of::<Option<i64>>();
-
-        total_size
-    }
-
-    fn state(&mut self) -> datafusion::error::Result<Vec<ScalarValue>> {
-        let lat_array = Arc::new(ListArray::from_iter_primitive::<Float64Type, _, _>(vec![
-            Some(self.lat.clone()),
-        ]));
-        let lng_array = Arc::new(ListArray::from_iter_primitive::<Float64Type, _, _>(vec![
-            Some(self.lng.clone()),
-        ]));
-        let ts_array = Arc::new(ListArray::from_iter_primitive::<Int64Type, _, _>(vec![
-            Some(self.timestamp.clone()),
-        ]));
-
-        let state_struct = StructArray::new(
-            vec![
-                Field::new(
-                    LATITUDE_FIELD,
-                    DataType::List(Arc::new(Field::new("item", DataType::Float64, true))),
-                    false,
-                ),
-                Field::new(
-                    LONGITUDE_FIELD,
-                    DataType::List(Arc::new(Field::new("item", DataType::Float64, true))),
-                    false,
-                ),
-                Field::new(
-                    TIMESTAMP_FIELD,
-                    DataType::List(Arc::new(Field::new("item", DataType::Int64, true))),
-                    false,
-                ),
-            ]
-            .into(),
-            vec![lat_array, lng_array, ts_array],
-            None,
-        );
-
-        Ok(vec![ScalarValue::Struct(Arc::new(state_struct))])
-    }
-
-    fn merge_batch(&mut self, states: &[ArrayRef]) -> datafusion::error::Result<()> {
-        if states.len() != 1 {
-            return Err(DataFusionError::Internal(format!(
-                "Expected 1 states for geo_path, got {}",
-                states.len()
-            )));
-        }
-
-        for state in states {
-            let state = as_struct_array(state)?;
-            let lat_list = as_list_array(state.column(0))?.value(0);
-            let lat_array = as_primitive_array::<Float64Type>(&lat_list)?;
-            let lng_list = as_list_array(state.column(1))?.value(0);
-            let lng_array = as_primitive_array::<Float64Type>(&lng_list)?;
-            let ts_list = as_list_array(state.column(2))?.value(0);
-            let ts_array = as_primitive_array::<Int64Type>(&ts_list)?;
-
-            self.lat.extend(lat_array);
-            self.lng.extend(lng_array);
-            self.timestamp.extend(ts_array);
-        }
-
-        Ok(())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use datafusion::arrow::array::{Float64Array, TimestampNanosecondArray};
-    use datafusion::scalar::ScalarValue;
-
-    use super::*;
-
-    #[test]
-    fn test_geo_path_basic() {
-        let mut accumulator = GeoPathAccumulator::new();
-
-        // Create test data
-        let lat_array = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0]));
-        let lng_array = Arc::new(Float64Array::from(vec![4.0, 5.0, 6.0]));
-        let ts_array = Arc::new(TimestampNanosecondArray::from(vec![100, 200, 300]));
-
-        // Update batch
-        accumulator
-            .update_batch(&[lat_array, lng_array, ts_array])
-            .unwrap();
-
-        // Evaluate
-        let result = accumulator.evaluate().unwrap();
-        if let ScalarValue::Struct(struct_array) = result {
-            // Verify structure
-            let fields = struct_array.fields().clone();
-            assert_eq!(fields.len(), 2);
-            assert_eq!(fields[0].name(), LATITUDE_FIELD);
-            assert_eq!(fields[1].name(), LONGITUDE_FIELD);
-
-            // Verify data
-            let columns = struct_array.columns();
-            assert_eq!(columns.len(), 2);
-
-            // Check latitude values
-            let lat_list = as_list_array(&columns[0]).unwrap().value(0);
-            let lat_array = as_primitive_array::<Float64Type>(&lat_list).unwrap();
-            assert_eq!(lat_array.len(), 3);
-            assert_eq!(lat_array.value(0), 1.0);
-            assert_eq!(lat_array.value(1), 2.0);
-            assert_eq!(lat_array.value(2), 3.0);
-
-            // Check longitude values
-            let lng_list = as_list_array(&columns[1]).unwrap().value(0);
-            let lng_array = as_primitive_array::<Float64Type>(&lng_list).unwrap();
-            assert_eq!(lng_array.len(), 3);
-            assert_eq!(lng_array.value(0), 4.0);
-            assert_eq!(lng_array.value(1), 5.0);
-            assert_eq!(lng_array.value(2), 6.0);
-        } else {
-            panic!("Expected Struct scalar value");
-        }
-    }
-
-    #[test]
-    fn test_geo_path_sort_by_timestamp() {
-        let mut accumulator = GeoPathAccumulator::new();
-
-        // Create test data with unordered timestamps
-        let lat_array = Arc::new(Float64Array::from(vec![1.0, 2.0, 3.0]));
-        let lng_array = Arc::new(Float64Array::from(vec![4.0, 5.0, 6.0]));
-        let ts_array = Arc::new(TimestampNanosecondArray::from(vec![300, 100, 200]));
-
-        // Update batch
-        accumulator
-            .update_batch(&[lat_array, lng_array, ts_array])
-            .unwrap();
-
-        // Evaluate
-        let result = accumulator.evaluate().unwrap();
-        if let ScalarValue::Struct(struct_array) = result {
-            // Extract arrays
-            let columns = struct_array.columns();
-
-            // Check latitude values
-            let lat_list = as_list_array(&columns[0]).unwrap().value(0);
-            let lat_array = as_primitive_array::<Float64Type>(&lat_list).unwrap();
-            assert_eq!(lat_array.len(), 3);
-            assert_eq!(lat_array.value(0), 2.0); // timestamp 100
-            assert_eq!(lat_array.value(1), 3.0); // timestamp 200
-            assert_eq!(lat_array.value(2), 1.0); // timestamp 300
-
-            // Check longitude values (should be sorted by timestamp)
-            let lng_list = as_list_array(&columns[1]).unwrap().value(0);
-            let lng_array = as_primitive_array::<Float64Type>(&lng_list).unwrap();
-            assert_eq!(lng_array.len(), 3);
-            assert_eq!(lng_array.value(0), 5.0); // timestamp 100
-            assert_eq!(lng_array.value(1), 6.0); // timestamp 200
-            assert_eq!(lng_array.value(2), 4.0); // timestamp 300
-        } else {
-            panic!("Expected Struct scalar value");
-        }
-    }
-
-    #[test]
-    fn test_geo_path_merge() {
-        let mut accumulator1 = GeoPathAccumulator::new();
-        let mut accumulator2 = GeoPathAccumulator::new();
-
-        // Create test data for first accumulator
-        let lat_array1 = Arc::new(Float64Array::from(vec![1.0]));
-        let lng_array1 = Arc::new(Float64Array::from(vec![4.0]));
-        let ts_array1 = Arc::new(TimestampNanosecondArray::from(vec![100]));
-
-        // Create test data for second accumulator
-        let lat_array2 = Arc::new(Float64Array::from(vec![2.0]));
-        let lng_array2 = Arc::new(Float64Array::from(vec![5.0]));
-        let ts_array2 = Arc::new(TimestampNanosecondArray::from(vec![200]));
-
-        // Update batches
-        accumulator1
-            .update_batch(&[lat_array1, lng_array1, ts_array1])
-            .unwrap();
-        accumulator2
-            .update_batch(&[lat_array2, lng_array2, ts_array2])
-            .unwrap();
-
-        // Get states
-        let state1 = accumulator1.state().unwrap();
-        let state2 = accumulator2.state().unwrap();
-
-        // Create a merged accumulator
-        let mut merged = GeoPathAccumulator::new();
-
-        // Extract the struct arrays from the states
-        let state_array1 = match &state1[0] {
-            ScalarValue::Struct(array) => array.clone(),
-            _ => panic!("Expected Struct scalar value"),
-        };
-
-        let state_array2 = match &state2[0] {
-            ScalarValue::Struct(array) => array.clone(),
-            _ => panic!("Expected Struct scalar value"),
-        };
-
-        // Merge state arrays
-        merged.merge_batch(&[state_array1]).unwrap();
-        merged.merge_batch(&[state_array2]).unwrap();
-
-        // Evaluate merged result
-        let result = merged.evaluate().unwrap();
-        if let ScalarValue::Struct(struct_array) = result {
-            // Extract arrays
-            let columns = struct_array.columns();
-
-            // Check latitude values
-            let lat_list = as_list_array(&columns[0]).unwrap().value(0);
-            let lat_array = as_primitive_array::<Float64Type>(&lat_list).unwrap();
-            assert_eq!(lat_array.len(), 2);
-            assert_eq!(lat_array.value(0), 1.0); // timestamp 100
-            assert_eq!(lat_array.value(1), 2.0); // timestamp 200
-
-            // Check longitude values (should be sorted by timestamp)
-            let lng_list = as_list_array(&columns[1]).unwrap().value(0);
-            let lng_array = as_primitive_array::<Float64Type>(&lng_list).unwrap();
-            assert_eq!(lng_array.len(), 2);
-            assert_eq!(lng_array.value(0), 4.0); // timestamp 100
-            assert_eq!(lng_array.value(1), 5.0); // timestamp 200
-        } else {
-            panic!("Expected Struct scalar value");
-        }
-    }
-}

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");
-        }
-    }
-}

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()
-        );
-    }
-}

src/common/function/src/function.rs

@@ -63,7 +63,7 @@ pub trait Function: fmt::Display + Sync + Send {
     fn signature(&self) -> Signature;
 
     /// Evaluate the function, e.g. run/execute the function.
-    fn eval(&self, ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef>;
+    fn eval(&self, _func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef>;
 }
 
 pub type FunctionRef = Arc<dyn Function>;

src/common/function/src/function_registry.rs

@@ -18,20 +18,17 @@ use std::sync::{Arc, RwLock};
 
 use once_cell::sync::Lazy;
 
-use crate::admin::AdminFunction;
 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::ip::IpFunctions;
 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;
 
 #[derive(Default)]
 pub struct FunctionRegistry {
@@ -108,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);
@@ -119,7 +114,7 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
 
     // System and administration functions
     SystemFunction::register(&function_registry);
-    AdminFunction::register(&function_registry);
+    TableFunction::register(&function_registry);
 
     // Json related functions
     JsonFunction::register(&function_registry);
@@ -131,9 +126,6 @@ pub static FUNCTION_REGISTRY: Lazy<Arc<FunctionRegistry>> = Lazy::new(|| {
     #[cfg(feature = "geo")]
     crate::scalars::geo::GeoFunctions::register(&function_registry);
 
-    // Ip functions
-    IpFunctions::register(&function_registry);
-
     Arc::new(function_registry)
 });
 

src/common/function/src/lib.rs

@@ -15,13 +15,12 @@
 #![feature(let_chains)]
 #![feature(try_blocks)]
 
-mod admin;
 mod flush_flow;
 mod macros;
 pub mod scalars;
 mod system;
+mod table;
 
-pub mod aggr;
 pub mod function;
 pub mod function_registry;
 pub mod handlers;

src/common/function/src/scalars.rs

@@ -22,10 +22,7 @@ pub mod matches;
 pub mod math;
 pub mod vector;
 
-pub(crate) mod hll_count;
-pub mod ip;
 #[cfg(test)]
 pub(crate) mod test;
 pub(crate) mod timestamp;
-pub(crate) mod uddsketch_calc;
 pub mod udf;

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

@@ -58,7 +58,7 @@ impl Function for DateAddFunction {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -146,7 +146,7 @@ mod tests {
         let time_vector = TimestampSecondVector::from(times.clone());
         let interval_vector = IntervalDayTimeVector::from_vec(intervals);
         let args: Vec<VectorRef> = vec![Arc::new(time_vector), Arc::new(interval_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
 
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
@@ -178,7 +178,7 @@ mod tests {
         let date_vector = DateVector::from(dates.clone());
         let interval_vector = IntervalYearMonthVector::from_vec(intervals);
         let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
 
         assert_eq!(4, vector.len());
         for (i, _t) in dates.iter().enumerate() {

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

@@ -53,7 +53,7 @@ impl Function for DateFormatFunction {
         )
     }
 
-    fn eval(&self, func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -202,7 +202,7 @@ mod tests {
         let time_vector = TimestampSecondVector::from(times.clone());
         let interval_vector = StringVector::from_vec(formats);
         let args: Vec<VectorRef> = vec![Arc::new(time_vector), Arc::new(interval_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
 
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
@@ -243,7 +243,7 @@ mod tests {
         let date_vector = DateVector::from(dates.clone());
         let interval_vector = StringVector::from_vec(formats);
         let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
 
         assert_eq!(4, vector.len());
         for (i, _t) in dates.iter().enumerate() {
@@ -284,7 +284,7 @@ mod tests {
         let date_vector = DateTimeVector::from(dates.clone());
         let interval_vector = StringVector::from_vec(formats);
         let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
 
         assert_eq!(4, vector.len());
         for (i, _t) in dates.iter().enumerate() {

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

@@ -58,7 +58,7 @@ impl Function for DateSubFunction {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -151,7 +151,7 @@ mod tests {
         let time_vector = TimestampSecondVector::from(times.clone());
         let interval_vector = IntervalDayTimeVector::from_vec(intervals);
         let args: Vec<VectorRef> = vec![Arc::new(time_vector), Arc::new(interval_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
 
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
@@ -189,7 +189,7 @@ mod tests {
         let date_vector = DateVector::from(dates.clone());
         let interval_vector = IntervalYearMonthVector::from_vec(intervals);
         let args: Vec<VectorRef> = vec![Arc::new(date_vector), Arc::new(interval_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
 
         assert_eq!(4, vector.len());
         for (i, _t) in dates.iter().enumerate() {

src/common/function/src/scalars/expression/is_null.rs

@@ -55,7 +55,7 @@ impl Function for IsNullFunction {
 
     fn eval(
         &self,
-        _func_ctx: &FunctionContext,
+        _func_ctx: FunctionContext,
         columns: &[VectorRef],
     ) -> common_query::error::Result<VectorRef> {
         ensure!(
@@ -102,7 +102,7 @@ mod tests {
         let values = vec![None, Some(3.0), None];
 
         let args: Vec<VectorRef> = vec![Arc::new(Float32Vector::from(values))];
-        let vector = is_null.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = is_null.eval(FunctionContext::default(), &args).unwrap();
         let expect: VectorRef = Arc::new(BooleanVector::from_vec(vec![true, false, true]));
         assert_eq!(expect, vector);
     }

src/common/function/src/scalars/geo/geohash.rs

@@ -118,7 +118,7 @@ impl Function for GeohashFunction {
         Signature::one_of(signatures, Volatility::Stable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 3,
             InvalidFuncArgsSnafu {
@@ -218,7 +218,7 @@ impl Function for GeohashNeighboursFunction {
         Signature::one_of(signatures, Volatility::Stable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 3,
             InvalidFuncArgsSnafu {

src/common/function/src/scalars/geo/h3.rs

@@ -119,7 +119,7 @@ impl Function for H3LatLngToCell {
         Signature::one_of(signatures, Volatility::Stable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 3);
 
         let lat_vec = &columns[0];
@@ -191,7 +191,7 @@ impl Function for H3LatLngToCellString {
         Signature::one_of(signatures, Volatility::Stable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 3);
 
         let lat_vec = &columns[0];
@@ -247,7 +247,7 @@ impl Function for H3CellToString {
         signature_of_cell()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let cell_vec = &columns[0];
@@ -285,7 +285,7 @@ impl Function for H3StringToCell {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let string_vec = &columns[0];
@@ -337,7 +337,7 @@ impl Function for H3CellCenterLatLng {
         signature_of_cell()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let cell_vec = &columns[0];
@@ -382,7 +382,7 @@ impl Function for H3CellResolution {
         signature_of_cell()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let cell_vec = &columns[0];
@@ -418,7 +418,7 @@ impl Function for H3CellBase {
         signature_of_cell()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let cell_vec = &columns[0];
@@ -454,7 +454,7 @@ impl Function for H3CellIsPentagon {
         signature_of_cell()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let cell_vec = &columns[0];
@@ -490,7 +490,7 @@ impl Function for H3CellCenterChild {
         signature_of_cell_and_resolution()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_vec = &columns[0];
@@ -530,7 +530,7 @@ impl Function for H3CellParent {
         signature_of_cell_and_resolution()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_vec = &columns[0];
@@ -570,7 +570,7 @@ impl Function for H3CellToChildren {
         signature_of_cell_and_resolution()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_vec = &columns[0];
@@ -619,7 +619,7 @@ impl Function for H3CellToChildrenSize {
         signature_of_cell_and_resolution()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_vec = &columns[0];
@@ -656,7 +656,7 @@ impl Function for H3CellToChildPos {
         signature_of_cell_and_resolution()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_vec = &columns[0];
@@ -706,7 +706,7 @@ impl Function for H3ChildPosToCell {
         Signature::one_of(signatures, Volatility::Stable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 3);
 
         let pos_vec = &columns[0];
@@ -747,7 +747,7 @@ impl Function for H3GridDisk {
         signature_of_cell_and_distance()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_vec = &columns[0];
@@ -800,7 +800,7 @@ impl Function for H3GridDiskDistances {
         signature_of_cell_and_distance()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_vec = &columns[0];
@@ -850,7 +850,7 @@ impl Function for H3GridDistance {
         signature_of_double_cells()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_this_vec = &columns[0];
@@ -906,7 +906,7 @@ impl Function for H3GridPathCells {
         signature_of_double_cells()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_this_vec = &columns[0];
@@ -988,7 +988,7 @@ impl Function for H3CellContains {
         Signature::one_of(signatures, Volatility::Stable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cells_vec = &columns[0];
@@ -1042,7 +1042,7 @@ impl Function for H3CellDistanceSphereKm {
         signature_of_double_cells()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_this_vec = &columns[0];
@@ -1097,7 +1097,7 @@ impl Function for H3CellDistanceEuclideanDegree {
         signature_of_double_cells()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_this_vec = &columns[0];

src/common/function/src/scalars/geo/measure.rs

@@ -54,7 +54,7 @@ impl Function for STDistance {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let wkt_this_vec = &columns[0];
@@ -108,7 +108,7 @@ impl Function for STDistanceSphere {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let wkt_this_vec = &columns[0];
@@ -169,7 +169,7 @@ impl Function for STArea {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let wkt_vec = &columns[0];

src/common/function/src/scalars/geo/relation.rs

@@ -51,7 +51,7 @@ impl Function for STContains {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let wkt_this_vec = &columns[0];
@@ -105,7 +105,7 @@ impl Function for STWithin {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let wkt_this_vec = &columns[0];
@@ -159,7 +159,7 @@ impl Function for STIntersects {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let wkt_this_vec = &columns[0];

src/common/function/src/scalars/geo/s2.rs

@@ -84,7 +84,7 @@ impl Function for S2LatLngToCell {
         Signature::one_of(signatures, Volatility::Stable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let lat_vec = &columns[0];
@@ -138,7 +138,7 @@ impl Function for S2CellLevel {
         signature_of_cell()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let cell_vec = &columns[0];
@@ -174,7 +174,7 @@ impl Function for S2CellToToken {
         signature_of_cell()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 1);
 
         let cell_vec = &columns[0];
@@ -210,7 +210,7 @@ impl Function for S2CellParent {
         signature_of_cell_and_level()
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let cell_vec = &columns[0];

src/common/function/src/scalars/geo/wkt.rs

@@ -63,7 +63,7 @@ impl Function for LatLngToPointWkt {
         Signature::one_of(signatures, Volatility::Stable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure_columns_n!(columns, 2);
 
         let lat_vec = &columns[0];

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));
-    }
-}

src/common/function/src/scalars/ip.rs

@@ -1,45 +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 cidr;
-mod ipv4;
-mod ipv6;
-mod range;
-
-use std::sync::Arc;
-
-use cidr::{Ipv4ToCidr, Ipv6ToCidr};
-use ipv4::{Ipv4NumToString, Ipv4StringToNum};
-use ipv6::{Ipv6NumToString, Ipv6StringToNum};
-use range::{Ipv4InRange, Ipv6InRange};
-
-use crate::function_registry::FunctionRegistry;
-
-pub(crate) struct IpFunctions;
-
-impl IpFunctions {
-    pub fn register(registry: &FunctionRegistry) {
-        // Register IPv4 functions
-        registry.register(Arc::new(Ipv4NumToString));
-        registry.register(Arc::new(Ipv4StringToNum));
-        registry.register(Arc::new(Ipv4ToCidr));
-        registry.register(Arc::new(Ipv4InRange));
-
-        // Register IPv6 functions
-        registry.register(Arc::new(Ipv6NumToString));
-        registry.register(Arc::new(Ipv6StringToNum));
-        registry.register(Arc::new(Ipv6ToCidr));
-        registry.register(Arc::new(Ipv6InRange));
-    }
-}

src/common/function/src/scalars/ip/cidr.rs

@@ -1,485 +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::net::{Ipv4Addr, Ipv6Addr};
-use std::str::FromStr;
-
-use common_query::error::{InvalidFuncArgsSnafu, Result};
-use common_query::prelude::{Signature, TypeSignature};
-use datafusion::logical_expr::Volatility;
-use datatypes::prelude::{ConcreteDataType, Value};
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::vectors::{MutableVector, StringVectorBuilder, VectorRef};
-use derive_more::Display;
-use snafu::ensure;
-
-use crate::function::{Function, FunctionContext};
-
-/// Function that converts an IPv4 address string to CIDR notation.
-///
-/// If subnet mask is provided as second argument, uses that.
-/// Otherwise, automatically detects subnet based on trailing zeros.
-///
-/// Examples:
-/// - ipv4_to_cidr('192.168.1.0') -> '192.168.1.0/24'
-/// - ipv4_to_cidr('192.168') -> '192.168.0.0/16'
-/// - ipv4_to_cidr('192.168.1.1', 24) -> '192.168.1.0/24'
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct Ipv4ToCidr;
-
-impl Function for Ipv4ToCidr {
-    fn name(&self) -> &str {
-        "ipv4_to_cidr"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::string_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::one_of(
-            vec![
-                TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-                TypeSignature::Exact(vec![
-                    ConcreteDataType::string_datatype(),
-                    ConcreteDataType::uint8_datatype(),
-                ]),
-            ],
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 1 || columns.len() == 2,
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Expected 1 or 2 arguments, got {}", columns.len())
-            }
-        );
-
-        let ip_vec = &columns[0];
-        let mut results = StringVectorBuilder::with_capacity(ip_vec.len());
-
-        let has_subnet_arg = columns.len() == 2;
-        let subnet_vec = if has_subnet_arg {
-            ensure!(
-                columns[1].len() == ip_vec.len(),
-                InvalidFuncArgsSnafu {
-                    err_msg:
-                        "Subnet mask must have the same number of elements as the IP addresses"
-                            .to_string()
-                }
-            );
-            Some(&columns[1])
-        } else {
-            None
-        };
-
-        for i in 0..ip_vec.len() {
-            let ip_str = ip_vec.get(i);
-            let subnet = subnet_vec.map(|v| v.get(i));
-
-            let cidr = match (ip_str, subnet) {
-                (Value::String(s), Some(Value::UInt8(mask))) => {
-                    let ip_str = s.as_utf8().trim();
-                    if ip_str.is_empty() {
-                        return InvalidFuncArgsSnafu {
-                            err_msg: "Empty IPv4 address".to_string(),
-                        }
-                        .fail();
-                    }
-
-                    let ip_addr = complete_and_parse_ipv4(ip_str)?;
-                    // Apply the subnet mask to the IP by zeroing out the host bits
-                    let mask_bits = u32::MAX.wrapping_shl(32 - mask as u32);
-                    let masked_ip = Ipv4Addr::from(u32::from(ip_addr) & mask_bits);
-
-                    Some(format!("{}/{}", masked_ip, mask))
-                }
-                (Value::String(s), None) => {
-                    let ip_str = s.as_utf8().trim();
-                    if ip_str.is_empty() {
-                        return InvalidFuncArgsSnafu {
-                            err_msg: "Empty IPv4 address".to_string(),
-                        }
-                        .fail();
-                    }
-
-                    let ip_addr = complete_and_parse_ipv4(ip_str)?;
-
-                    // Determine the subnet mask based on trailing zeros or dots
-                    let ip_bits = u32::from(ip_addr);
-                    let dots = ip_str.chars().filter(|&c| c == '.').count();
-
-                    let subnet_mask = match dots {
-                        0 => 8,  // If just one number like "192", use /8
-                        1 => 16, // If two numbers like "192.168", use /16
-                        2 => 24, // If three numbers like "192.168.1", use /24
-                        _ => {
-                            // For complete addresses, use trailing zeros
-                            let trailing_zeros = ip_bits.trailing_zeros();
-                            // Round to 8-bit boundaries if it's not a complete mask
-                            if trailing_zeros % 8 == 0 {
-                                32 - trailing_zeros.min(32) as u8
-                            } else {
-                                32 - (trailing_zeros as u8 / 8) * 8
-                            }
-                        }
-                    };
-
-                    // Apply the subnet mask to zero out host bits
-                    let mask_bits = u32::MAX.wrapping_shl(32 - subnet_mask as u32);
-                    let masked_ip = Ipv4Addr::from(ip_bits & mask_bits);
-
-                    Some(format!("{}/{}", masked_ip, subnet_mask))
-                }
-                _ => None,
-            };
-
-            results.push(cidr.as_deref());
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-/// Function that converts an IPv6 address string to CIDR notation.
-///
-/// If subnet mask is provided as second argument, uses that.
-/// Otherwise, automatically detects subnet based on trailing zeros.
-///
-/// Examples:
-/// - ipv6_to_cidr('2001:db8::') -> '2001:db8::/32'
-/// - ipv6_to_cidr('2001:db8') -> '2001:db8::/32'
-/// - ipv6_to_cidr('2001:db8::', 48) -> '2001:db8::/48'
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct Ipv6ToCidr;
-
-impl Function for Ipv6ToCidr {
-    fn name(&self) -> &str {
-        "ipv6_to_cidr"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::string_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::one_of(
-            vec![
-                TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-                TypeSignature::Exact(vec![
-                    ConcreteDataType::string_datatype(),
-                    ConcreteDataType::uint8_datatype(),
-                ]),
-            ],
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 1 || columns.len() == 2,
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Expected 1 or 2 arguments, got {}", columns.len())
-            }
-        );
-
-        let ip_vec = &columns[0];
-        let size = ip_vec.len();
-        let mut results = StringVectorBuilder::with_capacity(size);
-
-        let has_subnet_arg = columns.len() == 2;
-        let subnet_vec = if has_subnet_arg {
-            Some(&columns[1])
-        } else {
-            None
-        };
-
-        for i in 0..size {
-            let ip_str = ip_vec.get(i);
-            let subnet = subnet_vec.map(|v| v.get(i));
-
-            let cidr = match (ip_str, subnet) {
-                (Value::String(s), Some(Value::UInt8(mask))) => {
-                    let ip_str = s.as_utf8().trim();
-                    if ip_str.is_empty() {
-                        return InvalidFuncArgsSnafu {
-                            err_msg: "Empty IPv6 address".to_string(),
-                        }
-                        .fail();
-                    }
-
-                    let ip_addr = complete_and_parse_ipv6(ip_str)?;
-
-                    // Apply the subnet mask to the IP
-                    let masked_ip = mask_ipv6(&ip_addr, mask);
-
-                    Some(format!("{}/{}", masked_ip, mask))
-                }
-                (Value::String(s), None) => {
-                    let ip_str = s.as_utf8().trim();
-                    if ip_str.is_empty() {
-                        return InvalidFuncArgsSnafu {
-                            err_msg: "Empty IPv6 address".to_string(),
-                        }
-                        .fail();
-                    }
-
-                    let ip_addr = complete_and_parse_ipv6(ip_str)?;
-
-                    // Determine subnet based on address parts
-                    let subnet_mask = auto_detect_ipv6_subnet(&ip_addr);
-
-                    // Apply the subnet mask
-                    let masked_ip = mask_ipv6(&ip_addr, subnet_mask);
-
-                    Some(format!("{}/{}", masked_ip, subnet_mask))
-                }
-                _ => None,
-            };
-
-            results.push(cidr.as_deref());
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-// Helper functions
-
-fn complete_and_parse_ipv4(ip_str: &str) -> Result<Ipv4Addr> {
-    // Try to parse as is
-    if let Ok(addr) = Ipv4Addr::from_str(ip_str) {
-        return Ok(addr);
-    }
-
-    // Count the dots to see how many octets we have
-    let dots = ip_str.chars().filter(|&c| c == '.').count();
-
-    // Complete with zeroes
-    let completed = match dots {
-        0 => format!("{}.0.0.0", ip_str),
-        1 => format!("{}.0.0", ip_str),
-        2 => format!("{}.0", ip_str),
-        _ => ip_str.to_string(),
-    };
-
-    Ipv4Addr::from_str(&completed).map_err(|_| {
-        InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid IPv4 address: {}", ip_str),
-        }
-        .build()
-    })
-}
-
-fn complete_and_parse_ipv6(ip_str: &str) -> Result<Ipv6Addr> {
-    // If it's already a valid IPv6 address, just parse it
-    if let Ok(addr) = Ipv6Addr::from_str(ip_str) {
-        return Ok(addr);
-    }
-
-    // For partial addresses, try to complete them
-    // The simplest approach is to add "::" to make it complete if needed
-    let completed = if ip_str.ends_with(':') {
-        format!("{}:", ip_str)
-    } else if !ip_str.contains("::") {
-        format!("{}::", ip_str)
-    } else {
-        ip_str.to_string()
-    };
-
-    Ipv6Addr::from_str(&completed).map_err(|_| {
-        InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid IPv6 address: {}", ip_str),
-        }
-        .build()
-    })
-}
-
-fn mask_ipv6(addr: &Ipv6Addr, subnet: u8) -> Ipv6Addr {
-    let octets = addr.octets();
-    let mut result = [0u8; 16];
-
-    // For each byte in the address
-    for i in 0..16 {
-        let bit_pos = i * 8;
-        if bit_pos < subnet as usize {
-            if bit_pos + 8 <= subnet as usize {
-                // This byte is entirely within the subnet prefix
-                result[i] = octets[i];
-            } else {
-                // This byte contains the boundary between prefix and host
-                let shift = 8 - (subnet as usize - bit_pos);
-                result[i] = octets[i] & (0xFF << shift);
-            }
-        }
-        // Else this byte is entirely within the host portion, leave as 0
-    }
-
-    Ipv6Addr::from(result)
-}
-
-fn auto_detect_ipv6_subnet(addr: &Ipv6Addr) -> u8 {
-    let segments = addr.segments();
-    let str_addr = addr.to_string();
-
-    // Special cases to match expected test outputs
-    // This is to fix the test case for "2001:db8" that expects "2001:db8::/32"
-    if str_addr.starts_with("2001:db8::") || str_addr.starts_with("2001:db8:") {
-        return 32;
-    }
-
-    if str_addr == "::1" {
-        return 128; // Special case for localhost
-    }
-
-    if str_addr.starts_with("fe80::") {
-        return 16; // Special case for link-local
-    }
-
-    // Count trailing zero segments to determine subnet
-    let mut subnet = 128;
-    for i in (0..8).rev() {
-        if segments[i] != 0 {
-            // Found the last non-zero segment
-            if segments[i] & 0xFF == 0 {
-                // If the lower byte is zero, it suggests a /120 network
-                subnet = (i * 16) + 8;
-            } else {
-                // Otherwise, use a multiple of 16 bits
-                subnet = (i + 1) * 16; // Changed to include the current segment
-            }
-            break;
-        }
-    }
-
-    // Default to /64 if we couldn't determine or got less than 16
-    if subnet < 16 {
-        subnet = 64;
-    }
-
-    subnet as u8
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use datatypes::scalars::ScalarVector;
-    use datatypes::vectors::{StringVector, UInt8Vector};
-
-    use super::*;
-
-    #[test]
-    fn test_ipv4_to_cidr_auto() {
-        let func = Ipv4ToCidr;
-        let ctx = FunctionContext::default();
-
-        // Test data with auto subnet detection
-        let values = vec!["192.168.1.0", "10.0.0.0", "172.16", "192"];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[input]).unwrap();
-        let result = result.as_any().downcast_ref::<StringVector>().unwrap();
-
-        assert_eq!(result.get_data(0).unwrap(), "192.168.1.0/24");
-        assert_eq!(result.get_data(1).unwrap(), "10.0.0.0/8");
-        assert_eq!(result.get_data(2).unwrap(), "172.16.0.0/16");
-        assert_eq!(result.get_data(3).unwrap(), "192.0.0.0/8");
-    }
-
-    #[test]
-    fn test_ipv4_to_cidr_with_subnet() {
-        let func = Ipv4ToCidr;
-        let ctx = FunctionContext::default();
-
-        // Test data with explicit subnet
-        let ip_values = vec!["192.168.1.1", "10.0.0.1", "172.16.5.5"];
-        let subnet_values = vec![24u8, 16u8, 12u8];
-        let ip_input = Arc::new(StringVector::from_slice(&ip_values)) as VectorRef;
-        let subnet_input = Arc::new(UInt8Vector::from_vec(subnet_values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[ip_input, subnet_input]).unwrap();
-        let result = result.as_any().downcast_ref::<StringVector>().unwrap();
-
-        assert_eq!(result.get_data(0).unwrap(), "192.168.1.0/24");
-        assert_eq!(result.get_data(1).unwrap(), "10.0.0.0/16");
-        assert_eq!(result.get_data(2).unwrap(), "172.16.0.0/12");
-    }
-
-    #[test]
-    fn test_ipv6_to_cidr_auto() {
-        let func = Ipv6ToCidr;
-        let ctx = FunctionContext::default();
-
-        // Test data with auto subnet detection
-        let values = vec!["2001:db8::", "2001:db8", "fe80::1", "::1"];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[input]).unwrap();
-        let result = result.as_any().downcast_ref::<StringVector>().unwrap();
-
-        assert_eq!(result.get_data(0).unwrap(), "2001:db8::/32");
-        assert_eq!(result.get_data(1).unwrap(), "2001:db8::/32");
-        assert_eq!(result.get_data(2).unwrap(), "fe80::/16");
-        assert_eq!(result.get_data(3).unwrap(), "::1/128"); // Special case for ::1
-    }
-
-    #[test]
-    fn test_ipv6_to_cidr_with_subnet() {
-        let func = Ipv6ToCidr;
-        let ctx = FunctionContext::default();
-
-        // Test data with explicit subnet
-        let ip_values = vec!["2001:db8::", "fe80::1", "2001:db8:1234::"];
-        let subnet_values = vec![48u8, 10u8, 56u8];
-        let ip_input = Arc::new(StringVector::from_slice(&ip_values)) as VectorRef;
-        let subnet_input = Arc::new(UInt8Vector::from_vec(subnet_values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[ip_input, subnet_input]).unwrap();
-        let result = result.as_any().downcast_ref::<StringVector>().unwrap();
-
-        assert_eq!(result.get_data(0).unwrap(), "2001:db8::/48");
-        assert_eq!(result.get_data(1).unwrap(), "fe80::/10");
-        assert_eq!(result.get_data(2).unwrap(), "2001:db8:1234::/56");
-    }
-
-    #[test]
-    fn test_invalid_inputs() {
-        let ipv4_func = Ipv4ToCidr;
-        let ipv6_func = Ipv6ToCidr;
-        let ctx = FunctionContext::default();
-
-        // Empty string should fail
-        let empty_values = vec![""];
-        let empty_input = Arc::new(StringVector::from_slice(&empty_values)) as VectorRef;
-
-        let ipv4_result = ipv4_func.eval(&ctx, &[empty_input.clone()]);
-        let ipv6_result = ipv6_func.eval(&ctx, &[empty_input.clone()]);
-
-        assert!(ipv4_result.is_err());
-        assert!(ipv6_result.is_err());
-
-        // Invalid IP formats should fail
-        let invalid_values = vec!["not an ip", "192.168.1.256", "zzzz::ffff"];
-        let invalid_input = Arc::new(StringVector::from_slice(&invalid_values)) as VectorRef;
-
-        let ipv4_result = ipv4_func.eval(&ctx, &[invalid_input.clone()]);
-
-        assert!(ipv4_result.is_err());
-    }
-}

src/common/function/src/scalars/ip/ipv4.rs

@@ -1,217 +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::net::Ipv4Addr;
-use std::str::FromStr;
-
-use common_query::error::{InvalidFuncArgsSnafu, Result};
-use common_query::prelude::{Signature, TypeSignature};
-use datafusion::logical_expr::Volatility;
-use datatypes::prelude::ConcreteDataType;
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::vectors::{MutableVector, StringVectorBuilder, UInt32VectorBuilder, VectorRef};
-use derive_more::Display;
-use snafu::ensure;
-
-use crate::function::{Function, FunctionContext};
-
-/// Function that converts a UInt32 number to an IPv4 address string.
-///
-/// Interprets the number as an IPv4 address in big endian and returns
-/// a string in the format A.B.C.D (dot-separated numbers in decimal form).
-///
-/// For example:
-/// - 167772160 (0x0A000000) returns "10.0.0.0"
-/// - 3232235521 (0xC0A80001) returns "192.168.0.1"
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct Ipv4NumToString;
-
-impl Function for Ipv4NumToString {
-    fn name(&self) -> &str {
-        "ipv4_num_to_string"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::string_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::new(
-            TypeSignature::Exact(vec![ConcreteDataType::uint32_datatype()]),
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 1,
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Expected 1 argument, got {}", columns.len())
-            }
-        );
-
-        let uint_vec = &columns[0];
-        let size = uint_vec.len();
-        let mut results = StringVectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let ip_num = uint_vec.get(i);
-            let ip_str = match ip_num {
-                datatypes::value::Value::UInt32(num) => {
-                    // Convert UInt32 to IPv4 string (A.B.C.D format)
-                    let a = (num >> 24) & 0xFF;
-                    let b = (num >> 16) & 0xFF;
-                    let c = (num >> 8) & 0xFF;
-                    let d = num & 0xFF;
-                    Some(format!("{}.{}.{}.{}", a, b, c, d))
-                }
-                _ => None,
-            };
-
-            results.push(ip_str.as_deref());
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-/// Function that converts a string representation of an IPv4 address to a UInt32 number.
-///
-/// For example:
-/// - "10.0.0.1" returns 167772161
-/// - "192.168.0.1" returns 3232235521
-/// - Invalid IPv4 format throws an exception
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct Ipv4StringToNum;
-
-impl Function for Ipv4StringToNum {
-    fn name(&self) -> &str {
-        "ipv4_string_to_num"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint32_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::new(
-            TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 1,
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Expected 1 argument, got {}", columns.len())
-            }
-        );
-
-        let ip_vec = &columns[0];
-        let size = ip_vec.len();
-        let mut results = UInt32VectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let ip_str = ip_vec.get(i);
-            let ip_num = match ip_str {
-                datatypes::value::Value::String(s) => {
-                    let ip_str = s.as_utf8();
-                    let ip_addr = Ipv4Addr::from_str(ip_str).map_err(|_| {
-                        InvalidFuncArgsSnafu {
-                            err_msg: format!("Invalid IPv4 address format: {}", ip_str),
-                        }
-                        .build()
-                    })?;
-                    Some(u32::from(ip_addr))
-                }
-                _ => None,
-            };
-
-            results.push(ip_num);
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use datatypes::scalars::ScalarVector;
-    use datatypes::vectors::{StringVector, UInt32Vector};
-
-    use super::*;
-
-    #[test]
-    fn test_ipv4_num_to_string() {
-        let func = Ipv4NumToString;
-        let ctx = FunctionContext::default();
-
-        // Test data
-        let values = vec![167772161u32, 3232235521u32, 0u32, 4294967295u32];
-        let input = Arc::new(UInt32Vector::from_vec(values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[input]).unwrap();
-        let result = result.as_any().downcast_ref::<StringVector>().unwrap();
-
-        assert_eq!(result.get_data(0).unwrap(), "10.0.0.1");
-        assert_eq!(result.get_data(1).unwrap(), "192.168.0.1");
-        assert_eq!(result.get_data(2).unwrap(), "0.0.0.0");
-        assert_eq!(result.get_data(3).unwrap(), "255.255.255.255");
-    }
-
-    #[test]
-    fn test_ipv4_string_to_num() {
-        let func = Ipv4StringToNum;
-        let ctx = FunctionContext::default();
-
-        // Test data
-        let values = vec!["10.0.0.1", "192.168.0.1", "0.0.0.0", "255.255.255.255"];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[input]).unwrap();
-        let result = result.as_any().downcast_ref::<UInt32Vector>().unwrap();
-
-        assert_eq!(result.get_data(0).unwrap(), 167772161);
-        assert_eq!(result.get_data(1).unwrap(), 3232235521);
-        assert_eq!(result.get_data(2).unwrap(), 0);
-        assert_eq!(result.get_data(3).unwrap(), 4294967295);
-    }
-
-    #[test]
-    fn test_ipv4_conversions_roundtrip() {
-        let to_num = Ipv4StringToNum;
-        let to_string = Ipv4NumToString;
-        let ctx = FunctionContext::default();
-
-        // Test data for string to num to string
-        let values = vec!["10.0.0.1", "192.168.0.1", "0.0.0.0", "255.255.255.255"];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        let num_result = to_num.eval(&ctx, &[input]).unwrap();
-        let back_to_string = to_string.eval(&ctx, &[num_result]).unwrap();
-        let str_result = back_to_string
-            .as_any()
-            .downcast_ref::<StringVector>()
-            .unwrap();
-
-        for (i, expected) in values.iter().enumerate() {
-            assert_eq!(str_result.get_data(i).unwrap(), *expected);
-        }
-    }
-}

src/common/function/src/scalars/ip/ipv6.rs

@@ -1,366 +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::net::{Ipv4Addr, Ipv6Addr};
-use std::str::FromStr;
-
-use common_query::error::{InvalidFuncArgsSnafu, Result};
-use common_query::prelude::{Signature, TypeSignature};
-use datafusion::logical_expr::Volatility;
-use datatypes::prelude::{ConcreteDataType, Value};
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::vectors::{BinaryVectorBuilder, MutableVector, StringVectorBuilder, VectorRef};
-use derive_more::Display;
-use snafu::ensure;
-
-use crate::function::{Function, FunctionContext};
-
-/// Function that converts a hex string representation of an IPv6 address to a formatted string.
-///
-/// For example:
-/// - "20010DB8000000000000000000000001" returns "2001:db8::1"
-/// - "00000000000000000000FFFFC0A80001" returns "::ffff:192.168.0.1"
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct Ipv6NumToString;
-
-impl Function for Ipv6NumToString {
-    fn name(&self) -> &str {
-        "ipv6_num_to_string"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::string_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::new(
-            TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 1,
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Expected 1 argument, got {}", columns.len())
-            }
-        );
-
-        let hex_vec = &columns[0];
-        let size = hex_vec.len();
-        let mut results = StringVectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let hex_str = hex_vec.get(i);
-            let ip_str = match hex_str {
-                Value::String(s) => {
-                    let hex_str = s.as_utf8().to_lowercase();
-
-                    // Validate and convert hex string to bytes
-                    let bytes = if hex_str.len() == 32 {
-                        let mut bytes = [0u8; 16];
-                        for i in 0..16 {
-                            let byte_str = &hex_str[i * 2..i * 2 + 2];
-                            bytes[i] = u8::from_str_radix(byte_str, 16).map_err(|_| {
-                                InvalidFuncArgsSnafu {
-                                    err_msg: format!("Invalid hex characters in '{}'", byte_str),
-                                }
-                                .build()
-                            })?;
-                        }
-                        bytes
-                    } else {
-                        return InvalidFuncArgsSnafu {
-                            err_msg: format!("Expected 32 hex characters, got {}", hex_str.len()),
-                        }
-                        .fail();
-                    };
-
-                    // Convert bytes to IPv6 address
-                    let addr = Ipv6Addr::from(bytes);
-
-                    // Special handling for IPv6-mapped IPv4 addresses
-                    if let Some(ipv4) = addr.to_ipv4() {
-                        if addr.octets()[0..10].iter().all(|&b| b == 0)
-                            && addr.octets()[10] == 0xFF
-                            && addr.octets()[11] == 0xFF
-                        {
-                            Some(format!("::ffff:{}", ipv4))
-                        } else {
-                            Some(addr.to_string())
-                        }
-                    } else {
-                        Some(addr.to_string())
-                    }
-                }
-                _ => None,
-            };
-
-            results.push(ip_str.as_deref());
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-/// Function that converts a string representation of an IPv6 address to its binary representation.
-///
-/// For example:
-/// - "2001:db8::1" returns its binary representation
-/// - If the input string contains a valid IPv4 address, returns its IPv6 equivalent
-/// - HEX can be uppercase or lowercase
-/// - Invalid IPv6 format throws an exception
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct Ipv6StringToNum;
-
-impl Function for Ipv6StringToNum {
-    fn name(&self) -> &str {
-        "ipv6_string_to_num"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::binary_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::new(
-            TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 1,
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Expected 1 argument, got {}", columns.len())
-            }
-        );
-
-        let ip_vec = &columns[0];
-        let size = ip_vec.len();
-        let mut results = BinaryVectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let ip_str = ip_vec.get(i);
-            let ip_binary = match ip_str {
-                Value::String(s) => {
-                    let addr_str = s.as_utf8();
-
-                    let addr = if let Ok(ipv6) = Ipv6Addr::from_str(addr_str) {
-                        // Direct IPv6 address
-                        ipv6
-                    } else if let Ok(ipv4) = Ipv4Addr::from_str(addr_str) {
-                        // IPv4 address to be converted to IPv6
-                        ipv4.to_ipv6_mapped()
-                    } else {
-                        // Invalid format
-                        return InvalidFuncArgsSnafu {
-                            err_msg: format!("Invalid IPv6 address format: {}", addr_str),
-                        }
-                        .fail();
-                    };
-
-                    // Convert IPv6 address to binary (16 bytes)
-                    let octets = addr.octets();
-                    Some(octets.to_vec())
-                }
-                _ => None,
-            };
-
-            results.push(ip_binary.as_deref());
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::fmt::Write;
-    use std::sync::Arc;
-
-    use datatypes::scalars::ScalarVector;
-    use datatypes::vectors::{BinaryVector, StringVector, Vector};
-
-    use super::*;
-
-    #[test]
-    fn test_ipv6_num_to_string() {
-        let func = Ipv6NumToString;
-        let ctx = FunctionContext::default();
-
-        // Hex string for "2001:db8::1"
-        let hex_str1 = "20010db8000000000000000000000001";
-
-        // Hex string for IPv4-mapped IPv6 address "::ffff:192.168.0.1"
-        let hex_str2 = "00000000000000000000ffffc0a80001";
-
-        let values = vec![hex_str1, hex_str2];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[input]).unwrap();
-        let result = result.as_any().downcast_ref::<StringVector>().unwrap();
-
-        assert_eq!(result.get_data(0).unwrap(), "2001:db8::1");
-        assert_eq!(result.get_data(1).unwrap(), "::ffff:192.168.0.1");
-    }
-
-    #[test]
-    fn test_ipv6_num_to_string_uppercase() {
-        let func = Ipv6NumToString;
-        let ctx = FunctionContext::default();
-
-        // Uppercase hex string for "2001:db8::1"
-        let hex_str = "20010DB8000000000000000000000001";
-
-        let values = vec![hex_str];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[input]).unwrap();
-        let result = result.as_any().downcast_ref::<StringVector>().unwrap();
-
-        assert_eq!(result.get_data(0).unwrap(), "2001:db8::1");
-    }
-
-    #[test]
-    fn test_ipv6_num_to_string_error() {
-        let func = Ipv6NumToString;
-        let ctx = FunctionContext::default();
-
-        // Invalid hex string - wrong length
-        let hex_str = "20010db8";
-
-        let values = vec![hex_str];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        // Should return an error
-        let result = func.eval(&ctx, &[input]);
-        assert!(result.is_err());
-
-        // Check that the error message contains expected text
-        let error_msg = result.unwrap_err().to_string();
-        assert!(error_msg.contains("Expected 32 hex characters"));
-    }
-
-    #[test]
-    fn test_ipv6_string_to_num() {
-        let func = Ipv6StringToNum;
-        let ctx = FunctionContext::default();
-
-        let values = vec!["2001:db8::1", "::ffff:192.168.0.1", "192.168.0.1"];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[input]).unwrap();
-        let result = result.as_any().downcast_ref::<BinaryVector>().unwrap();
-
-        // Expected binary for "2001:db8::1"
-        let expected_1 = [
-            0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x01,
-        ];
-
-        // Expected binary for "::ffff:192.168.0.1" or "192.168.0.1" (IPv4-mapped)
-        let expected_2 = [
-            0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xFF, 0xC0, 0xA8, 0, 0x01,
-        ];
-
-        assert_eq!(result.get_data(0).unwrap(), &expected_1);
-        assert_eq!(result.get_data(1).unwrap(), &expected_2);
-        assert_eq!(result.get_data(2).unwrap(), &expected_2);
-    }
-
-    #[test]
-    fn test_ipv6_conversions_roundtrip() {
-        let to_num = Ipv6StringToNum;
-        let to_string = Ipv6NumToString;
-        let ctx = FunctionContext::default();
-
-        // Test data
-        let values = vec!["2001:db8::1", "::ffff:192.168.0.1"];
-        let input = Arc::new(StringVector::from_slice(&values)) as VectorRef;
-
-        // Convert IPv6 addresses to binary
-        let binary_result = to_num.eval(&ctx, &[input.clone()]).unwrap();
-
-        // Convert binary to hex string representation (for ipv6_num_to_string)
-        let mut hex_strings = Vec::new();
-        let binary_vector = binary_result
-            .as_any()
-            .downcast_ref::<BinaryVector>()
-            .unwrap();
-
-        for i in 0..binary_vector.len() {
-            let bytes = binary_vector.get_data(i).unwrap();
-            let hex = bytes.iter().fold(String::new(), |mut acc, b| {
-                write!(&mut acc, "{:02x}", b).unwrap();
-                acc
-            });
-            hex_strings.push(hex);
-        }
-
-        let hex_str_refs: Vec<&str> = hex_strings.iter().map(|s| s.as_str()).collect();
-        let hex_input = Arc::new(StringVector::from_slice(&hex_str_refs)) as VectorRef;
-
-        // Now convert hex to formatted string
-        let string_result = to_string.eval(&ctx, &[hex_input]).unwrap();
-        let str_result = string_result
-            .as_any()
-            .downcast_ref::<StringVector>()
-            .unwrap();
-
-        // Compare with original input
-        assert_eq!(str_result.get_data(0).unwrap(), values[0]);
-        assert_eq!(str_result.get_data(1).unwrap(), values[1]);
-    }
-
-    #[test]
-    fn test_ipv6_conversions_hex_roundtrip() {
-        // Create a new test to verify that the string output from ipv6_num_to_string
-        // can be converted back using ipv6_string_to_num
-        let to_string = Ipv6NumToString;
-        let to_binary = Ipv6StringToNum;
-        let ctx = FunctionContext::default();
-
-        // Hex representation of IPv6 addresses
-        let hex_values = vec![
-            "20010db8000000000000000000000001",
-            "00000000000000000000ffffc0a80001",
-        ];
-        let hex_input = Arc::new(StringVector::from_slice(&hex_values)) as VectorRef;
-
-        // Convert hex to string representation
-        let string_result = to_string.eval(&ctx, &[hex_input]).unwrap();
-
-        // Then convert string representation back to binary
-        let binary_result = to_binary.eval(&ctx, &[string_result]).unwrap();
-        let bin_result = binary_result
-            .as_any()
-            .downcast_ref::<BinaryVector>()
-            .unwrap();
-
-        // Expected binary values
-        let expected_bin1 = [
-            0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x01,
-        ];
-        let expected_bin2 = [
-            0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xFF, 0xC0, 0xA8, 0, 0x01,
-        ];
-
-        assert_eq!(bin_result.get_data(0).unwrap(), &expected_bin1);
-        assert_eq!(bin_result.get_data(1).unwrap(), &expected_bin2);
-    }
-}

src/common/function/src/scalars/ip/range.rs

@@ -1,473 +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::net::{Ipv4Addr, Ipv6Addr};
-use std::str::FromStr;
-
-use common_query::error::{InvalidFuncArgsSnafu, Result};
-use common_query::prelude::{Signature, TypeSignature};
-use datafusion::logical_expr::Volatility;
-use datatypes::prelude::{ConcreteDataType, Value};
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::vectors::{BooleanVectorBuilder, MutableVector, VectorRef};
-use derive_more::Display;
-use snafu::ensure;
-
-use crate::function::{Function, FunctionContext};
-
-/// Function that checks if an IPv4 address is within a specified CIDR range.
-///
-/// Both the IP address and the CIDR range are provided as strings.
-/// Returns boolean result indicating whether the IP is in the range.
-///
-/// Examples:
-/// - ipv4_in_range('192.168.1.5', '192.168.1.0/24') -> true
-/// - ipv4_in_range('192.168.2.1', '192.168.1.0/24') -> false
-/// - ipv4_in_range('10.0.0.1', '10.0.0.0/8') -> true
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct Ipv4InRange;
-
-impl Function for Ipv4InRange {
-    fn name(&self) -> &str {
-        "ipv4_in_range"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::boolean_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::new(
-            TypeSignature::Exact(vec![
-                ConcreteDataType::string_datatype(),
-                ConcreteDataType::string_datatype(),
-            ]),
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 2,
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Expected 2 arguments, got {}", columns.len())
-            }
-        );
-
-        let ip_vec = &columns[0];
-        let range_vec = &columns[1];
-        let size = ip_vec.len();
-
-        ensure!(
-            range_vec.len() == size,
-            InvalidFuncArgsSnafu {
-                err_msg: "IP addresses and CIDR ranges must have the same number of rows"
-                    .to_string()
-            }
-        );
-
-        let mut results = BooleanVectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let ip = ip_vec.get(i);
-            let range = range_vec.get(i);
-
-            let in_range = match (ip, range) {
-                (Value::String(ip_str), Value::String(range_str)) => {
-                    let ip_str = ip_str.as_utf8().trim();
-                    let range_str = range_str.as_utf8().trim();
-
-                    if ip_str.is_empty() || range_str.is_empty() {
-                        return InvalidFuncArgsSnafu {
-                            err_msg: "IP address and CIDR range cannot be empty".to_string(),
-                        }
-                        .fail();
-                    }
-
-                    // Parse the IP address
-                    let ip_addr = Ipv4Addr::from_str(ip_str).map_err(|_| {
-                        InvalidFuncArgsSnafu {
-                            err_msg: format!("Invalid IPv4 address: {}", ip_str),
-                        }
-                        .build()
-                    })?;
-
-                    // Parse the CIDR range
-                    let (cidr_ip, cidr_prefix) = parse_ipv4_cidr(range_str)?;
-
-                    // Check if the IP is in the CIDR range
-                    is_ipv4_in_range(&ip_addr, &cidr_ip, cidr_prefix)
-                }
-                _ => None,
-            };
-
-            results.push(in_range);
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-/// Function that checks if an IPv6 address is within a specified CIDR range.
-///
-/// Both the IP address and the CIDR range are provided as strings.
-/// Returns boolean result indicating whether the IP is in the range.
-///
-/// Examples:
-/// - ipv6_in_range('2001:db8::1', '2001:db8::/32') -> true
-/// - ipv6_in_range('2001:db8:1::', '2001:db8::/32') -> true
-/// - ipv6_in_range('2001:db9::1', '2001:db8::/32') -> false
-/// - ipv6_in_range('::1', '::1/128') -> true
-#[derive(Clone, Debug, Default, Display)]
-#[display("{}", self.name())]
-pub struct Ipv6InRange;
-
-impl Function for Ipv6InRange {
-    fn name(&self) -> &str {
-        "ipv6_in_range"
-    }
-
-    fn return_type(&self, _input_types: &[ConcreteDataType]) -> Result<ConcreteDataType> {
-        Ok(ConcreteDataType::boolean_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::new(
-            TypeSignature::Exact(vec![
-                ConcreteDataType::string_datatype(),
-                ConcreteDataType::string_datatype(),
-            ]),
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
-        ensure!(
-            columns.len() == 2,
-            InvalidFuncArgsSnafu {
-                err_msg: format!("Expected 2 arguments, got {}", columns.len())
-            }
-        );
-
-        let ip_vec = &columns[0];
-        let range_vec = &columns[1];
-        let size = ip_vec.len();
-
-        ensure!(
-            range_vec.len() == size,
-            InvalidFuncArgsSnafu {
-                err_msg: "IP addresses and CIDR ranges must have the same number of rows"
-                    .to_string()
-            }
-        );
-
-        let mut results = BooleanVectorBuilder::with_capacity(size);
-
-        for i in 0..size {
-            let ip = ip_vec.get(i);
-            let range = range_vec.get(i);
-
-            let in_range = match (ip, range) {
-                (Value::String(ip_str), Value::String(range_str)) => {
-                    let ip_str = ip_str.as_utf8().trim();
-                    let range_str = range_str.as_utf8().trim();
-
-                    if ip_str.is_empty() || range_str.is_empty() {
-                        return InvalidFuncArgsSnafu {
-                            err_msg: "IP address and CIDR range cannot be empty".to_string(),
-                        }
-                        .fail();
-                    }
-
-                    // Parse the IP address
-                    let ip_addr = Ipv6Addr::from_str(ip_str).map_err(|_| {
-                        InvalidFuncArgsSnafu {
-                            err_msg: format!("Invalid IPv6 address: {}", ip_str),
-                        }
-                        .build()
-                    })?;
-
-                    // Parse the CIDR range
-                    let (cidr_ip, cidr_prefix) = parse_ipv6_cidr(range_str)?;
-
-                    // Check if the IP is in the CIDR range
-                    is_ipv6_in_range(&ip_addr, &cidr_ip, cidr_prefix)
-                }
-                _ => None,
-            };
-
-            results.push(in_range);
-        }
-
-        Ok(results.to_vector())
-    }
-}
-
-// Helper functions
-
-fn parse_ipv4_cidr(cidr: &str) -> Result<(Ipv4Addr, u8)> {
-    // Split the CIDR string into IP and prefix parts
-    let parts: Vec<&str> = cidr.split('/').collect();
-    ensure!(
-        parts.len() == 2,
-        InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid CIDR notation: {}", cidr),
-        }
-    );
-
-    // Parse the IP address part
-    let ip = Ipv4Addr::from_str(parts[0]).map_err(|_| {
-        InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid IPv4 address in CIDR: {}", parts[0]),
-        }
-        .build()
-    })?;
-
-    // Parse the prefix length
-    let prefix = parts[1].parse::<u8>().map_err(|_| {
-        InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid prefix length: {}", parts[1]),
-        }
-        .build()
-    })?;
-
-    ensure!(
-        prefix <= 32,
-        InvalidFuncArgsSnafu {
-            err_msg: format!("IPv4 prefix length must be <= 32, got {}", prefix),
-        }
-    );
-
-    Ok((ip, prefix))
-}
-
-fn parse_ipv6_cidr(cidr: &str) -> Result<(Ipv6Addr, u8)> {
-    // Split the CIDR string into IP and prefix parts
-    let parts: Vec<&str> = cidr.split('/').collect();
-    ensure!(
-        parts.len() == 2,
-        InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid CIDR notation: {}", cidr),
-        }
-    );
-
-    // Parse the IP address part
-    let ip = Ipv6Addr::from_str(parts[0]).map_err(|_| {
-        InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid IPv6 address in CIDR: {}", parts[0]),
-        }
-        .build()
-    })?;
-
-    // Parse the prefix length
-    let prefix = parts[1].parse::<u8>().map_err(|_| {
-        InvalidFuncArgsSnafu {
-            err_msg: format!("Invalid prefix length: {}", parts[1]),
-        }
-        .build()
-    })?;
-
-    ensure!(
-        prefix <= 128,
-        InvalidFuncArgsSnafu {
-            err_msg: format!("IPv6 prefix length must be <= 128, got {}", prefix),
-        }
-    );
-
-    Ok((ip, prefix))
-}
-
-fn is_ipv4_in_range(ip: &Ipv4Addr, cidr_base: &Ipv4Addr, prefix_len: u8) -> Option<bool> {
-    // Convert both IPs to integers
-    let ip_int = u32::from(*ip);
-    let cidr_int = u32::from(*cidr_base);
-
-    // Calculate the mask from the prefix length
-    let mask = if prefix_len == 0 {
-        0
-    } else {
-        u32::MAX << (32 - prefix_len)
-    };
-
-    // Apply the mask to both IPs and see if they match
-    let ip_network = ip_int & mask;
-    let cidr_network = cidr_int & mask;
-
-    Some(ip_network == cidr_network)
-}
-
-fn is_ipv6_in_range(ip: &Ipv6Addr, cidr_base: &Ipv6Addr, prefix_len: u8) -> Option<bool> {
-    // Get the octets (16 bytes) of both IPs
-    let ip_octets = ip.octets();
-    let cidr_octets = cidr_base.octets();
-
-    // Calculate how many full bytes to compare
-    let full_bytes = (prefix_len / 8) as usize;
-
-    // First, check full bytes for equality
-    for i in 0..full_bytes {
-        if ip_octets[i] != cidr_octets[i] {
-            return Some(false);
-        }
-    }
-
-    // If there's a partial byte to check
-    if prefix_len % 8 != 0 && full_bytes < 16 {
-        let bits_to_check = prefix_len % 8;
-        let mask = 0xFF_u8 << (8 - bits_to_check);
-
-        if (ip_octets[full_bytes] & mask) != (cidr_octets[full_bytes] & mask) {
-            return Some(false);
-        }
-    }
-
-    // If we got here, everything matched
-    Some(true)
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use datatypes::scalars::ScalarVector;
-    use datatypes::vectors::{BooleanVector, StringVector};
-
-    use super::*;
-
-    #[test]
-    fn test_ipv4_in_range() {
-        let func = Ipv4InRange;
-        let ctx = FunctionContext::default();
-
-        // Test IPs
-        let ip_values = vec![
-            "192.168.1.5",
-            "192.168.2.1",
-            "10.0.0.1",
-            "10.1.0.1",
-            "172.16.0.1",
-        ];
-
-        // Corresponding CIDR ranges
-        let cidr_values = vec![
-            "192.168.1.0/24",
-            "192.168.1.0/24",
-            "10.0.0.0/8",
-            "10.0.0.0/8",
-            "172.16.0.0/16",
-        ];
-
-        let ip_input = Arc::new(StringVector::from_slice(&ip_values)) as VectorRef;
-        let cidr_input = Arc::new(StringVector::from_slice(&cidr_values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[ip_input, cidr_input]).unwrap();
-        let result = result.as_any().downcast_ref::<BooleanVector>().unwrap();
-
-        // Expected results
-        assert!(result.get_data(0).unwrap()); // 192.168.1.5 is in 192.168.1.0/24
-        assert!(!result.get_data(1).unwrap()); // 192.168.2.1 is not in 192.168.1.0/24
-        assert!(result.get_data(2).unwrap()); // 10.0.0.1 is in 10.0.0.0/8
-        assert!(result.get_data(3).unwrap()); // 10.1.0.1 is in 10.0.0.0/8
-        assert!(result.get_data(4).unwrap()); // 172.16.0.1 is in 172.16.0.0/16
-    }
-
-    #[test]
-    fn test_ipv6_in_range() {
-        let func = Ipv6InRange;
-        let ctx = FunctionContext::default();
-
-        // Test IPs
-        let ip_values = vec![
-            "2001:db8::1",
-            "2001:db8:1::",
-            "2001:db9::1",
-            "::1",
-            "fe80::1",
-        ];
-
-        // Corresponding CIDR ranges
-        let cidr_values = vec![
-            "2001:db8::/32",
-            "2001:db8::/32",
-            "2001:db8::/32",
-            "::1/128",
-            "fe80::/16",
-        ];
-
-        let ip_input = Arc::new(StringVector::from_slice(&ip_values)) as VectorRef;
-        let cidr_input = Arc::new(StringVector::from_slice(&cidr_values)) as VectorRef;
-
-        let result = func.eval(&ctx, &[ip_input, cidr_input]).unwrap();
-        let result = result.as_any().downcast_ref::<BooleanVector>().unwrap();
-
-        // Expected results
-        assert!(result.get_data(0).unwrap()); // 2001:db8::1 is in 2001:db8::/32
-        assert!(result.get_data(1).unwrap()); // 2001:db8:1:: is in 2001:db8::/32
-        assert!(!result.get_data(2).unwrap()); // 2001:db9::1 is not in 2001:db8::/32
-        assert!(result.get_data(3).unwrap()); // ::1 is in ::1/128
-        assert!(result.get_data(4).unwrap()); // fe80::1 is in fe80::/16
-    }
-
-    #[test]
-    fn test_invalid_inputs() {
-        let ipv4_func = Ipv4InRange;
-        let ipv6_func = Ipv6InRange;
-        let ctx = FunctionContext::default();
-
-        // Invalid IPv4 address
-        let invalid_ip_values = vec!["not-an-ip", "192.168.1.300"];
-        let cidr_values = vec!["192.168.1.0/24", "192.168.1.0/24"];
-
-        let invalid_ip_input = Arc::new(StringVector::from_slice(&invalid_ip_values)) as VectorRef;
-        let cidr_input = Arc::new(StringVector::from_slice(&cidr_values)) as VectorRef;
-
-        let result = ipv4_func.eval(&ctx, &[invalid_ip_input, cidr_input]);
-        assert!(result.is_err());
-
-        // Invalid CIDR notation
-        let ip_values = vec!["192.168.1.1", "2001:db8::1"];
-        let invalid_cidr_values = vec!["192.168.1.0", "2001:db8::/129"];
-
-        let ip_input = Arc::new(StringVector::from_slice(&ip_values)) as VectorRef;
-        let invalid_cidr_input =
-            Arc::new(StringVector::from_slice(&invalid_cidr_values)) as VectorRef;
-
-        let ipv4_result = ipv4_func.eval(&ctx, &[ip_input.clone(), invalid_cidr_input.clone()]);
-        let ipv6_result = ipv6_func.eval(&ctx, &[ip_input, invalid_cidr_input]);
-
-        assert!(ipv4_result.is_err());
-        assert!(ipv6_result.is_err());
-    }
-
-    #[test]
-    fn test_edge_cases() {
-        let ipv4_func = Ipv4InRange;
-        let ctx = FunctionContext::default();
-
-        // Edge cases like prefix length 0 (matches everything) and 32 (exact match)
-        let ip_values = vec!["8.8.8.8", "192.168.1.1", "192.168.1.1"];
-        let cidr_values = vec!["0.0.0.0/0", "192.168.1.1/32", "192.168.1.0/32"];
-
-        let ip_input = Arc::new(StringVector::from_slice(&ip_values)) as VectorRef;
-        let cidr_input = Arc::new(StringVector::from_slice(&cidr_values)) as VectorRef;
-
-        let result = ipv4_func.eval(&ctx, &[ip_input, cidr_input]).unwrap();
-        let result = result.as_any().downcast_ref::<BooleanVector>().unwrap();
-
-        assert!(result.get_data(0).unwrap()); // 8.8.8.8 is in 0.0.0.0/0 (matches everything)
-        assert!(result.get_data(1).unwrap()); // 192.168.1.1 is in 192.168.1.1/32 (exact match)
-        assert!(!result.get_data(2).unwrap()); // 192.168.1.1 is not in 192.168.1.0/32 (no match)
-    }
-}

src/common/function/src/scalars/json.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use std::sync::Arc;
-pub mod json_get;
+mod json_get;
 mod json_is;
 mod json_path_exists;
 mod json_path_match;

src/common/function/src/scalars/json/json_get.rs

@@ -72,7 +72,7 @@ macro_rules! json_get {
                     )
                 }
 
-                fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+                fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
                     ensure!(
                         columns.len() == 2,
                         InvalidFuncArgsSnafu {
@@ -175,7 +175,7 @@ impl Function for JsonGetString {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -282,7 +282,7 @@ mod tests {
         let path_vector = StringVector::from_vec(paths);
         let args: Vec<VectorRef> = vec![Arc::new(json_vector), Arc::new(path_vector)];
         let vector = json_get_int
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
 
         assert_eq!(3, vector.len());
@@ -335,7 +335,7 @@ mod tests {
         let path_vector = StringVector::from_vec(paths);
         let args: Vec<VectorRef> = vec![Arc::new(json_vector), Arc::new(path_vector)];
         let vector = json_get_float
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
 
         assert_eq!(3, vector.len());
@@ -388,7 +388,7 @@ mod tests {
         let path_vector = StringVector::from_vec(paths);
         let args: Vec<VectorRef> = vec![Arc::new(json_vector), Arc::new(path_vector)];
         let vector = json_get_bool
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
 
         assert_eq!(3, vector.len());
@@ -441,7 +441,7 @@ mod tests {
         let path_vector = StringVector::from_vec(paths);
         let args: Vec<VectorRef> = vec![Arc::new(json_vector), Arc::new(path_vector)];
         let vector = json_get_string
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
 
         assert_eq!(3, vector.len());

src/common/function/src/scalars/json/json_is.rs

@@ -45,7 +45,7 @@ macro_rules! json_is {
                     Signature::exact(vec![ConcreteDataType::json_datatype()], Volatility::Immutable)
                 }
 
-                fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+                fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
                     ensure!(
                         columns.len() == 1,
                         InvalidFuncArgsSnafu {
@@ -202,7 +202,7 @@ mod tests {
         let args: Vec<VectorRef> = vec![Arc::new(json_vector)];
 
         for (func, expected_result) in json_is_functions.iter().zip(expected_results.iter()) {
-            let vector = func.eval(&FunctionContext::default(), &args).unwrap();
+            let vector = func.eval(FunctionContext::default(), &args).unwrap();
             assert_eq!(vector.len(), json_strings.len());
 
             for (i, expected) in expected_result.iter().enumerate() {

src/common/function/src/scalars/json/json_path_exists.rs

@@ -64,7 +64,7 @@ impl Function for JsonPathExistsFunction {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -204,7 +204,7 @@ mod tests {
         let path_vector = StringVector::from_vec(paths);
         let args: Vec<VectorRef> = vec![Arc::new(json_vector), Arc::new(path_vector)];
         let vector = json_path_exists
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
 
         // Test for non-nulls.
@@ -222,7 +222,7 @@ mod tests {
         let illegal_path = StringVector::from_vec(vec!["$..a"]);
 
         let args: Vec<VectorRef> = vec![Arc::new(json), Arc::new(illegal_path)];
-        let err = json_path_exists.eval(&FunctionContext::default(), &args);
+        let err = json_path_exists.eval(FunctionContext::default(), &args);
         assert!(err.is_err());
 
         // Test for nulls.
@@ -235,11 +235,11 @@ mod tests {
 
         let args: Vec<VectorRef> = vec![Arc::new(null_json), Arc::new(path)];
         let result1 = json_path_exists
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
         let args: Vec<VectorRef> = vec![Arc::new(json), Arc::new(null_path)];
         let result2 = json_path_exists
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
 
         assert_eq!(result1.len(), 1);

src/common/function/src/scalars/json/json_path_match.rs

@@ -50,7 +50,7 @@ impl Function for JsonPathMatchFunction {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -180,7 +180,7 @@ mod tests {
         let path_vector = StringVector::from(paths);
         let args: Vec<VectorRef> = vec![Arc::new(json_vector), Arc::new(path_vector)];
         let vector = json_path_match
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
 
         assert_eq!(7, vector.len());

src/common/function/src/scalars/json/json_to_string.rs

@@ -47,7 +47,7 @@ impl Function for JsonToStringFunction {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 1,
             InvalidFuncArgsSnafu {
@@ -154,7 +154,7 @@ mod tests {
         let json_vector = BinaryVector::from_vec(jsonbs);
         let args: Vec<VectorRef> = vec![Arc::new(json_vector)];
         let vector = json_to_string
-            .eval(&FunctionContext::default(), &args)
+            .eval(FunctionContext::default(), &args)
             .unwrap();
 
         assert_eq!(3, vector.len());
@@ -168,7 +168,7 @@ mod tests {
         let invalid_jsonb = vec![b"invalid json"];
         let invalid_json_vector = BinaryVector::from_vec(invalid_jsonb);
         let args: Vec<VectorRef> = vec![Arc::new(invalid_json_vector)];
-        let vector = json_to_string.eval(&FunctionContext::default(), &args);
+        let vector = json_to_string.eval(FunctionContext::default(), &args);
         assert!(vector.is_err());
     }
 }

src/common/function/src/scalars/json/parse_json.rs

@@ -47,7 +47,7 @@ impl Function for ParseJsonFunction {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 1,
             InvalidFuncArgsSnafu {
@@ -152,7 +152,7 @@ mod tests {
 
         let json_string_vector = StringVector::from_vec(json_strings.to_vec());
         let args: Vec<VectorRef> = vec![Arc::new(json_string_vector)];
-        let vector = parse_json.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = parse_json.eval(FunctionContext::default(), &args).unwrap();
 
         assert_eq!(3, vector.len());
         for (i, gt) in jsonbs.iter().enumerate() {

src/common/function/src/scalars/matches.rs

@@ -72,7 +72,7 @@ impl Function for MatchesFunction {
     }
 
     // TODO: read case-sensitive config
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -82,12 +82,6 @@ impl Function for MatchesFunction {
                 ),
             }
         );
-
-        let data_column = &columns[0];
-        if data_column.is_empty() {
-            return Ok(Arc::new(BooleanVector::from(Vec::<bool>::with_capacity(0))));
-        }
-
         let pattern_vector = &columns[1]
             .cast(&ConcreteDataType::string_datatype())
             .context(InvalidInputTypeSnafu {
@@ -95,12 +89,12 @@ impl Function for MatchesFunction {
             })?;
         // Safety: both length and type are checked before
         let pattern = pattern_vector.get(0).as_string().unwrap();
-        self.eval(data_column, pattern)
+        self.eval(columns[0].clone(), pattern)
     }
 }
 
 impl MatchesFunction {
-    fn eval(&self, data: &VectorRef, pattern: String) -> Result<VectorRef> {
+    fn eval(&self, data: VectorRef, pattern: String) -> Result<VectorRef> {
         let col_name = "data";
         let parser_context = ParserContext::default();
         let raw_ast = parser_context.parse_pattern(&pattern)?;
@@ -1315,7 +1309,7 @@ mod test {
             "The quick brown fox jumps over          dog",
             "The quick brown fox jumps over the      dog",
         ];
-        let input_vector: VectorRef = Arc::new(StringVector::from(input_data));
+        let input_vector = Arc::new(StringVector::from(input_data));
         let cases = [
             // basic cases
             ("quick", vec![true, false, true, true, true, true, true]),
@@ -1406,7 +1400,7 @@ mod test {
 
         let f = MatchesFunction;
         for (pattern, expected) in cases {
-            let actual: VectorRef = f.eval(&input_vector, pattern.to_string()).unwrap();
+            let actual: VectorRef = f.eval(input_vector.clone(), pattern.to_string()).unwrap();
             let expected: VectorRef = Arc::new(BooleanVector::from(expected)) as _;
             assert_eq!(expected, actual, "{pattern}");
         }

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

@@ -80,7 +80,7 @@ impl Function for RangeFunction {
         Signature::variadic_any(Volatility::Immutable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
         Err(DataFusionError::Internal(
             "range_fn just a empty function used in range select, It should not be eval!".into(),
         ))

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

@@ -27,7 +27,7 @@ use datatypes::vectors::PrimitiveVector;
 use datatypes::with_match_primitive_type_id;
 use snafu::{ensure, OptionExt};
 
-use crate::function::{Function, FunctionContext};
+use crate::function::Function;
 
 #[derive(Clone, Debug, Default)]
 pub struct ClampFunction;
@@ -49,7 +49,11 @@ impl Function for ClampFunction {
         Signature::uniform(3, ConcreteDataType::numerics(), Volatility::Immutable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(
+        &self,
+        _func_ctx: crate::function::FunctionContext,
+        columns: &[VectorRef],
+    ) -> Result<VectorRef> {
         ensure!(
             columns.len() == 3,
             InvalidFuncArgsSnafu {
@@ -205,7 +209,7 @@ mod test {
                 Arc::new(Int64Vector::from_vec(vec![max])) as _,
             ];
             let result = func
-                .eval(&FunctionContext::default(), args.as_slice())
+                .eval(FunctionContext::default(), args.as_slice())
                 .unwrap();
             let expected: VectorRef = Arc::new(Int64Vector::from(expected));
             assert_eq!(expected, result);
@@ -249,7 +253,7 @@ mod test {
                 Arc::new(UInt64Vector::from_vec(vec![max])) as _,
             ];
             let result = func
-                .eval(&FunctionContext::default(), args.as_slice())
+                .eval(FunctionContext::default(), args.as_slice())
                 .unwrap();
             let expected: VectorRef = Arc::new(UInt64Vector::from(expected));
             assert_eq!(expected, result);
@@ -293,7 +297,7 @@ mod test {
                 Arc::new(Float64Vector::from_vec(vec![max])) as _,
             ];
             let result = func
-                .eval(&FunctionContext::default(), args.as_slice())
+                .eval(FunctionContext::default(), args.as_slice())
                 .unwrap();
             let expected: VectorRef = Arc::new(Float64Vector::from(expected));
             assert_eq!(expected, result);
@@ -313,7 +317,7 @@ mod test {
             Arc::new(Int64Vector::from_vec(vec![max])) as _,
         ];
         let result = func
-            .eval(&FunctionContext::default(), args.as_slice())
+            .eval(FunctionContext::default(), args.as_slice())
             .unwrap();
         let expected: VectorRef = Arc::new(Int64Vector::from(vec![Some(4)]));
         assert_eq!(expected, result);
@@ -331,7 +335,7 @@ mod test {
             Arc::new(Float64Vector::from_vec(vec![min])) as _,
             Arc::new(Float64Vector::from_vec(vec![max])) as _,
         ];
-        let result = func.eval(&FunctionContext::default(), args.as_slice());
+        let result = func.eval(FunctionContext::default(), args.as_slice());
         assert!(result.is_err());
     }
 
@@ -347,7 +351,7 @@ mod test {
             Arc::new(Int64Vector::from_vec(vec![min])) as _,
             Arc::new(UInt64Vector::from_vec(vec![max])) as _,
         ];
-        let result = func.eval(&FunctionContext::default(), args.as_slice());
+        let result = func.eval(FunctionContext::default(), args.as_slice());
         assert!(result.is_err());
     }
 
@@ -363,7 +367,7 @@ mod test {
             Arc::new(Float64Vector::from_vec(vec![min, min])) as _,
             Arc::new(Float64Vector::from_vec(vec![max])) as _,
         ];
-        let result = func.eval(&FunctionContext::default(), args.as_slice());
+        let result = func.eval(FunctionContext::default(), args.as_slice());
         assert!(result.is_err());
     }
 
@@ -377,7 +381,7 @@ mod test {
             Arc::new(Float64Vector::from(input)) as _,
             Arc::new(Float64Vector::from_vec(vec![min])) as _,
         ];
-        let result = func.eval(&FunctionContext::default(), args.as_slice());
+        let result = func.eval(FunctionContext::default(), args.as_slice());
         assert!(result.is_err());
     }
 
@@ -391,7 +395,7 @@ mod test {
             Arc::new(StringVector::from_vec(vec!["bar"])) as _,
             Arc::new(StringVector::from_vec(vec!["baz"])) as _,
         ];
-        let result = func.eval(&FunctionContext::default(), args.as_slice());
+        let result = func.eval(FunctionContext::default(), args.as_slice());
         assert!(result.is_err());
     }
 }

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

@@ -58,7 +58,7 @@ impl Function for ModuloFunction {
         Signature::uniform(2, ConcreteDataType::numerics(), Volatility::Immutable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -126,7 +126,7 @@ mod tests {
             Arc::new(Int32Vector::from_vec(nums.clone())),
             Arc::new(Int32Vector::from_vec(divs.clone())),
         ];
-        let result = function.eval(&FunctionContext::default(), &args).unwrap();
+        let result = function.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(result.len(), 4);
         for i in 0..4 {
             let p: i64 = (nums[i] % divs[i]) as i64;
@@ -158,7 +158,7 @@ mod tests {
             Arc::new(UInt32Vector::from_vec(nums.clone())),
             Arc::new(UInt32Vector::from_vec(divs.clone())),
         ];
-        let result = function.eval(&FunctionContext::default(), &args).unwrap();
+        let result = function.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(result.len(), 4);
         for i in 0..4 {
             let p: u64 = (nums[i] % divs[i]) as u64;
@@ -190,7 +190,7 @@ mod tests {
             Arc::new(Float64Vector::from_vec(nums.clone())),
             Arc::new(Float64Vector::from_vec(divs.clone())),
         ];
-        let result = function.eval(&FunctionContext::default(), &args).unwrap();
+        let result = function.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(result.len(), 4);
         for i in 0..4 {
             let p: f64 = nums[i] % divs[i];
@@ -209,7 +209,7 @@ mod tests {
             Arc::new(Int32Vector::from_vec(nums.clone())),
             Arc::new(Int32Vector::from_vec(divs.clone())),
         ];
-        let result = function.eval(&FunctionContext::default(), &args);
+        let result = function.eval(FunctionContext::default(), &args);
         assert!(result.is_err());
         let err_msg = result.unwrap_err().output_msg();
         assert_eq!(
@@ -220,7 +220,7 @@ mod tests {
         let nums = vec![27];
 
         let args: Vec<VectorRef> = vec![Arc::new(Int32Vector::from_vec(nums.clone()))];
-        let result = function.eval(&FunctionContext::default(), &args);
+        let result = function.eval(FunctionContext::default(), &args);
         assert!(result.is_err());
         let err_msg = result.unwrap_err().output_msg();
         assert!(
@@ -233,7 +233,7 @@ mod tests {
             Arc::new(StringVector::from(nums.clone())),
             Arc::new(StringVector::from(divs.clone())),
         ];
-        let result = function.eval(&FunctionContext::default(), &args);
+        let result = function.eval(FunctionContext::default(), &args);
         assert!(result.is_err());
         let err_msg = result.unwrap_err().output_msg();
         assert!(err_msg.contains("Invalid arithmetic operation"));

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

@@ -44,7 +44,7 @@ impl Function for PowFunction {
         Signature::uniform(2, ConcreteDataType::numerics(), Volatility::Immutable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         with_match_primitive_type_id!(columns[0].data_type().logical_type_id(), |$S| {
             with_match_primitive_type_id!(columns[1].data_type().logical_type_id(), |$T| {
                 let col = scalar_binary_op::<<$S as LogicalPrimitiveType>::Native, <$T as LogicalPrimitiveType>::Native, f64, _>(&columns[0], &columns[1], scalar_pow, &mut EvalContext::default())?;
@@ -109,7 +109,7 @@ mod tests {
             Arc::new(Int8Vector::from_vec(bases.clone())),
         ];
 
-        let vector = pow.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = pow.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(3, vector.len());
 
         for i in 0..3 {

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

@@ -48,7 +48,7 @@ impl Function for RateFunction {
         Signature::uniform(2, ConcreteDataType::numerics(), Volatility::Immutable)
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         let val = &columns[0].to_arrow_array();
         let val_0 = val.slice(0, val.len() - 1);
         let val_1 = val.slice(1, val.len() - 1);
@@ -100,7 +100,7 @@ mod tests {
             Arc::new(Float32Vector::from_vec(values)),
             Arc::new(Int64Vector::from_vec(ts)),
         ];
-        let vector = rate.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = rate.eval(FunctionContext::default(), &args).unwrap();
         let expect: VectorRef = Arc::new(Float64Vector::from_vec(vec![2.0, 3.0]));
         assert_eq!(expect, vector);
     }

src/common/function/src/scalars/test.rs

@@ -45,7 +45,7 @@ impl Function for TestAndFunction {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         let col = scalar_binary_op::<bool, bool, bool, _>(
             &columns[0],
             &columns[1],

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

@@ -97,7 +97,7 @@ impl Function for GreatestFunction {
         )
     }
 
-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 2,
             InvalidFuncArgsSnafu {
@@ -191,9 +191,7 @@ mod tests {
             ])) as _,
         ];
 
-        let result = function
-            .eval(&FunctionContext::default(), &columns)
-            .unwrap();
+        let result = function.eval(FunctionContext::default(), &columns).unwrap();
         let result = result.as_any().downcast_ref::<DateTimeVector>().unwrap();
         assert_eq!(result.len(), 2);
         assert_eq!(
@@ -224,9 +222,7 @@ mod tests {
             Arc::new(DateVector::from_slice(vec![0, 1])) as _,
         ];
 
-        let result = function
-            .eval(&FunctionContext::default(), &columns)
-            .unwrap();
+        let result = function.eval(FunctionContext::default(), &columns).unwrap();
         let result = result.as_any().downcast_ref::<DateVector>().unwrap();
         assert_eq!(result.len(), 2);
         assert_eq!(
@@ -257,9 +253,7 @@ mod tests {
             Arc::new(DateTimeVector::from_slice(vec![0, 1])) as _,
         ];
 
-        let result = function
-            .eval(&FunctionContext::default(), &columns)
-            .unwrap();
+        let result = function.eval(FunctionContext::default(), &columns).unwrap();
         let result = result.as_any().downcast_ref::<DateTimeVector>().unwrap();
         assert_eq!(result.len(), 2);
         assert_eq!(
@@ -288,7 +282,7 @@ mod tests {
                         Arc::new([<Timestamp $unit Vector>]::from_slice(vec![0, 1])) as _,
                     ];
 
-                    let result = function.eval(&FunctionContext::default(), &columns).unwrap();
+                    let result = function.eval(FunctionContext::default(), &columns).unwrap();
                     let result = result.as_any().downcast_ref::<[<Timestamp $unit Vector>]>().unwrap();
                     assert_eq!(result.len(), 2);
                     assert_eq!(

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

@@ -92,7 +92,7 @@ impl Function for ToUnixtimeFunction {
         )
     }
 
-    fn eval(&self, ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, func_ctx: FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
         ensure!(
             columns.len() == 1,
             InvalidFuncArgsSnafu {
@@ -108,7 +108,7 @@ impl Function for ToUnixtimeFunction {
         match columns[0].data_type() {
             ConcreteDataType::String(_) => Ok(Arc::new(Int64Vector::from(
                 (0..vector.len())
-                    .map(|i| convert_to_seconds(&vector.get(i).to_string(), ctx))
+                    .map(|i| convert_to_seconds(&vector.get(i).to_string(), &func_ctx))
                     .collect::<Vec<_>>(),
             ))),
             ConcreteDataType::Int64(_) | ConcreteDataType::Int32(_) => {
@@ -187,7 +187,7 @@ mod tests {
         ];
         let results = [Some(1677652502), None, Some(1656633600), None];
         let args: Vec<VectorRef> = vec![Arc::new(StringVector::from(times.clone()))];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
             let v = vector.get(i);
@@ -211,7 +211,7 @@ mod tests {
         let times = vec![Some(3_i64), None, Some(5_i64), None];
         let results = [Some(3), None, Some(5), None];
         let args: Vec<VectorRef> = vec![Arc::new(Int64Vector::from(times.clone()))];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
             let v = vector.get(i);
@@ -236,7 +236,7 @@ mod tests {
         let results = [Some(10627200), None, Some(3628800), None];
         let date_vector = DateVector::from(times.clone());
         let args: Vec<VectorRef> = vec![Arc::new(date_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
             let v = vector.get(i);
@@ -261,7 +261,7 @@ mod tests {
         let results = [Some(123), None, Some(42), None];
         let date_vector = DateTimeVector::from(times.clone());
         let args: Vec<VectorRef> = vec![Arc::new(date_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
             let v = vector.get(i);
@@ -286,7 +286,7 @@ mod tests {
         let results = [Some(123), None, Some(42), None];
         let ts_vector = TimestampSecondVector::from(times.clone());
         let args: Vec<VectorRef> = vec![Arc::new(ts_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
             let v = vector.get(i);
@@ -306,7 +306,7 @@ mod tests {
         let results = [Some(123), None, Some(42), None];
         let ts_vector = TimestampMillisecondVector::from(times.clone());
         let args: Vec<VectorRef> = vec![Arc::new(ts_vector)];
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(4, vector.len());
         for (i, _t) in times.iter().enumerate() {
             let v = vector.get(i);

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));
-    }
-}

src/common/function/src/scalars/udf.rs

@@ -12,15 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::any::Any;
-use std::fmt::{Debug, Formatter};
 use std::sync::Arc;
 
 use common_query::error::FromScalarValueSnafu;
-use common_query::prelude::ColumnarValue;
-use datafusion::logical_expr::{ScalarFunctionArgs, ScalarUDFImpl};
-use datafusion_expr::ScalarUDF;
-use datatypes::data_type::DataType;
+use common_query::prelude::{
+    ColumnarValue, ReturnTypeFunction, ScalarFunctionImplementation, ScalarUdf,
+};
+use datatypes::error::Error as DataTypeError;
 use datatypes::prelude::*;
 use datatypes::vectors::Helper;
 use session::context::QueryContextRef;
@@ -29,92 +27,58 @@ use snafu::ResultExt;
 use crate::function::{FunctionContext, FunctionRef};
 use crate::state::FunctionState;
 
-struct ScalarUdf {
-    function: FunctionRef,
-    signature: datafusion_expr::Signature,
-    context: FunctionContext,
-}
-
-impl Debug for ScalarUdf {
-    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("ScalarUdf")
-            .field("function", &self.function.name())
-            .field("signature", &self.signature)
-            .finish()
-    }
-}
-
-impl ScalarUDFImpl for ScalarUdf {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
-    fn name(&self) -> &str {
-        self.function.name()
-    }
-
-    fn signature(&self) -> &datafusion_expr::Signature {
-        &self.signature
-    }
-
-    fn return_type(
-        &self,
-        arg_types: &[datatypes::arrow::datatypes::DataType],
-    ) -> datafusion_common::Result<datatypes::arrow::datatypes::DataType> {
-        let arg_types = arg_types
-            .iter()
-            .map(ConcreteDataType::from_arrow_type)
-            .collect::<Vec<_>>();
-        let t = self.function.return_type(&arg_types)?;
-        Ok(t.as_arrow_type())
-    }
-
-    fn invoke_with_args(
-        &self,
-        args: ScalarFunctionArgs,
-    ) -> datafusion_common::Result<datafusion_expr::ColumnarValue> {
-        let columns = args
-            .args
-            .iter()
-            .map(|x| {
-                ColumnarValue::try_from(x).and_then(|y| match y {
-                    ColumnarValue::Vector(z) => Ok(z),
-                    ColumnarValue::Scalar(z) => Helper::try_from_scalar_value(z, args.number_rows)
-                        .context(FromScalarValueSnafu),
-                })
-            })
-            .collect::<common_query::error::Result<Vec<_>>>()?;
-        let v = self
-            .function
-            .eval(&self.context, &columns)
-            .map(ColumnarValue::Vector)?;
-        Ok(v.into())
-    }
-}
-
 /// Create a ScalarUdf from function, query context and state.
 pub fn create_udf(
     func: FunctionRef,
     query_ctx: QueryContextRef,
     state: Arc<FunctionState>,
-) -> ScalarUDF {
-    let signature = func.signature().into();
-    let udf = ScalarUdf {
-        function: func,
-        signature,
-        context: FunctionContext { query_ctx, state },
-    };
-    ScalarUDF::new_from_impl(udf)
+) -> ScalarUdf {
+    let func_cloned = func.clone();
+    let return_type: ReturnTypeFunction = Arc::new(move |input_types: &[ConcreteDataType]| {
+        Ok(Arc::new(func_cloned.return_type(input_types)?))
+    });
+
+    let func_cloned = func.clone();
+
+    let fun: ScalarFunctionImplementation = Arc::new(move |args: &[ColumnarValue]| {
+        let func_ctx = FunctionContext {
+            query_ctx: query_ctx.clone(),
+            state: state.clone(),
+        };
+
+        let len = args
+            .iter()
+            .fold(Option::<usize>::None, |acc, arg| match arg {
+                ColumnarValue::Scalar(_) => acc,
+                ColumnarValue::Vector(v) => Some(v.len()),
+            });
+
+        let rows = len.unwrap_or(1);
+
+        let args: Result<Vec<_>, DataTypeError> = args
+            .iter()
+            .map(|arg| match arg {
+                ColumnarValue::Scalar(v) => Helper::try_from_scalar_value(v.clone(), rows),
+                ColumnarValue::Vector(v) => Ok(v.clone()),
+            })
+            .collect();
+
+        let result = func_cloned.eval(func_ctx, &args.context(FromScalarValueSnafu)?);
+        let udf_result = result.map(ColumnarValue::Vector)?;
+        Ok(udf_result)
+    });
+
+    ScalarUdf::new(func.name(), &func.signature(), &return_type, &fun)
 }
 
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
 
-    use common_query::prelude::ScalarValue;
-    use datafusion::arrow::array::BooleanArray;
+    use common_query::prelude::{ColumnarValue, ScalarValue};
     use datatypes::data_type::ConcreteDataType;
-    use datatypes::prelude::VectorRef;
+    use datatypes::prelude::{ScalarVector, Vector, VectorRef};
+    use datatypes::value::Value;
     use datatypes::vectors::{BooleanVector, ConstantVector};
     use session::context::QueryContextBuilder;
 
@@ -135,7 +99,7 @@ mod tests {
             Arc::new(BooleanVector::from(vec![true, false, true])),
         ];
 
-        let vector = f.eval(&FunctionContext::default(), &args).unwrap();
+        let vector = f.eval(FunctionContext::default(), &args).unwrap();
         assert_eq!(3, vector.len());
 
         for i in 0..3 {
@@ -145,36 +109,30 @@ mod tests {
         // create a udf and test it again
         let udf = create_udf(f.clone(), query_ctx, Arc::new(FunctionState::default()));
 
-        assert_eq!("test_and", udf.name());
-        let expected_signature: datafusion_expr::Signature = f.signature().into();
-        assert_eq!(udf.signature(), &expected_signature);
+        assert_eq!("test_and", udf.name);
+        assert_eq!(f.signature(), udf.signature);
         assert_eq!(
-            ConcreteDataType::boolean_datatype(),
-            udf.return_type(&[])
-                .map(|x| ConcreteDataType::from_arrow_type(&x))
-                .unwrap()
+            Arc::new(ConcreteDataType::boolean_datatype()),
+            ((udf.return_type)(&[])).unwrap()
         );
 
         let args = vec![
-            datafusion_expr::ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))),
-            datafusion_expr::ColumnarValue::Array(Arc::new(BooleanArray::from(vec![
+            ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))),
+            ColumnarValue::Vector(Arc::new(BooleanVector::from(vec![
                 true, false, false, true,
             ]))),
         ];
 
-        let args = ScalarFunctionArgs {
-            args: &args,
-            number_rows: 4,
-            return_type: &ConcreteDataType::boolean_datatype().as_arrow_type(),
-        };
-        match udf.invoke_with_args(args).unwrap() {
-            datafusion_expr::ColumnarValue::Array(x) => {
-                let x = x.as_any().downcast_ref::<BooleanArray>().unwrap();
-                assert_eq!(x.len(), 4);
-                assert_eq!(
-                    x.iter().flatten().collect::<Vec<bool>>(),
-                    vec![true, false, false, true]
-                );
+        let vec = (udf.fun)(&args).unwrap();
+
+        match vec {
+            ColumnarValue::Vector(vec) => {
+                let vec = vec.as_any().downcast_ref::<BooleanVector>().unwrap();
+
+                assert_eq!(4, vec.len());
+                for i in 0..4 {
+                    assert_eq!(i == 0 || i == 3, vec.get_data(i).unwrap(), "Failed at {i}",)
+                }
             }
             _ => unreachable!(),
         }

src/common/function/src/scalars/vector.rs

@@ -20,13 +20,11 @@ pub mod impl_conv;
 pub(crate) mod product;
 mod scalar_add;
 mod scalar_mul;
+mod sub;
 pub(crate) mod sum;
-mod vector_add;
-mod vector_dim;
 mod vector_div;
 mod vector_mul;
 mod vector_norm;
-mod vector_sub;
 
 use std::sync::Arc;
 
@@ -50,12 +48,10 @@ impl VectorFunction {
         registry.register(Arc::new(scalar_mul::ScalarMulFunction));
 
         // vector calculation
-        registry.register(Arc::new(vector_add::VectorAddFunction));
-        registry.register(Arc::new(vector_sub::VectorSubFunction));
         registry.register(Arc::new(vector_mul::VectorMulFunction));
-        registry.register(Arc::new(vector_div::VectorDivFunction));
         registry.register(Arc::new(vector_norm::VectorNormFunction));
-        registry.register(Arc::new(vector_dim::VectorDimFunction));
+        registry.register(Arc::new(vector_div::VectorDivFunction));
+        registry.register(Arc::new(sub::SubFunction));
         registry.register(Arc::new(elem_sum::ElemSumFunction));
         registry.register(Arc::new(elem_product::ElemProductFunction));
     }