fix: conn pool leak & placeholder feature so ci can compile

fix: placeholder feature so ci can compile
fix: time window filter expr use OR
2025-12-25 07:30:02 +00:00 · 2025-04-10 15:01:07 +08:00 · 2025-04-08 14:37:55 +08:00 · 2025-04-07 16:50:17 +08:00 · 2025-04-07 16:50:17 +08:00 · 2025-04-07 16:50:17 +08:00
164 changed files with 3498 additions and 4998 deletions
--- a/.cargo/config.toml
+++ b/.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
--- a/.github/actions/build-dev-builder-images/action.yml
+++ b/.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 }}
--- a/.github/workflows/release-dev-builder-images.yaml
+++ b/.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:
@@ -63,7 +63,7 @@ jobs:

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

  release-dev-builder-images-cn: # Note: Be careful issue: https://github.com/containers/skopeo/issues/1874 and we decide to use the latest stable skopeo container.
    name: Release dev builder images to CN region
-    runs-on: ubuntu-22.04
+    runs-on: ubuntu-20.04
    needs: [
      release-dev-builder-images
    ]
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -91,7 +91,7 @@ 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

 jobs:
  allocate-runners:
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -185,7 +185,7 @@ checksum = "d301b3b94cb4b2f23d7917810addbbaff90738e0ca2be692bd027e70d7e0330c"

 [[package]]
 name = "api"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "common-base",
 "common-decimal",
@@ -710,7 +710,7 @@ dependencies = [

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

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

 [[package]]
 name = "catalog"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "arrow",
@@ -1661,7 +1661,7 @@ checksum = "1462739cb27611015575c0c11df5df7601141071f07518d56fcc1be504cbec97"

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

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

 [[package]]
 name = "cmd"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-trait",
 "auth",
@@ -1841,7 +1841,7 @@ dependencies = [
 "similar-asserts",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.13.0",
+ "substrait 0.12.0",
 "table",
 "temp-env",
 "tempfile",
@@ -1887,7 +1887,7 @@ checksum = "55b672471b4e9f9e95499ea597ff64941a309b2cdbffcc46f2cc5e2d971fd335"

 [[package]]
 name = "common-base"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "anymap2",
 "async-trait",
@@ -1909,11 +1909,11 @@ dependencies = [

 [[package]]
 name = "common-catalog"
-version = "0.13.0"
+version = "0.12.0"

 [[package]]
 name = "common-config"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "common-base",
 "common-error",
@@ -1938,7 +1938,7 @@ dependencies = [

 [[package]]
 name = "common-datasource"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "arrow",
 "arrow-schema",
@@ -1974,7 +1974,7 @@ dependencies = [

 [[package]]
 name = "common-decimal"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "bigdecimal 0.4.5",
 "common-error",
@@ -1987,7 +1987,7 @@ dependencies = [

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

 [[package]]
 name = "common-frontend"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-trait",
 "common-error",
@@ -2007,7 +2007,7 @@ dependencies = [

 [[package]]
 name = "common-function"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -2026,8 +2026,6 @@ dependencies = [
 "common-time",
 "common-version",
 "datafusion",
- "datafusion-common",
- "datafusion-expr",
 "datatypes",
 "derive_more",
 "geo",
@@ -2057,7 +2055,7 @@ dependencies = [

 [[package]]
 name = "common-greptimedb-telemetry"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-trait",
 "common-runtime",
@@ -2074,7 +2072,7 @@ dependencies = [

 [[package]]
 name = "common-grpc"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -2102,7 +2100,7 @@ dependencies = [

 [[package]]
 name = "common-grpc-expr"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "common-base",
@@ -2121,7 +2119,7 @@ dependencies = [

 [[package]]
 name = "common-macro"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "arc-swap",
 "common-query",
@@ -2135,7 +2133,7 @@ dependencies = [

 [[package]]
 name = "common-mem-prof"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2148,7 +2146,7 @@ dependencies = [

 [[package]]
 name = "common-meta"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "anymap2",
 "api",
@@ -2208,7 +2206,7 @@ dependencies = [

 [[package]]
 name = "common-options"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "common-grpc",
 "humantime-serde",
@@ -2217,11 +2215,11 @@ dependencies = [

 [[package]]
 name = "common-plugins"
-version = "0.13.0"
+version = "0.12.0"

 [[package]]
 name = "common-pprof"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "common-error",
 "common-macro",
@@ -2233,7 +2231,7 @@ dependencies = [

 [[package]]
 name = "common-procedure"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -2260,7 +2258,7 @@ dependencies = [

 [[package]]
 name = "common-procedure-test"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-trait",
 "common-procedure",
@@ -2268,7 +2266,7 @@ dependencies = [

 [[package]]
 name = "common-query"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "async-trait",
@@ -2294,7 +2292,7 @@ dependencies = [

 [[package]]
 name = "common-recordbatch"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "arc-swap",
 "common-error",
@@ -2313,7 +2311,7 @@ dependencies = [

 [[package]]
 name = "common-runtime"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -2343,7 +2341,7 @@ dependencies = [

 [[package]]
 name = "common-telemetry"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "atty",
 "backtrace",
@@ -2371,7 +2369,7 @@ dependencies = [

 [[package]]
 name = "common-test-util"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "client",
 "common-query",
@@ -2383,7 +2381,7 @@ dependencies = [

 [[package]]
 name = "common-time"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "arrow",
 "chrono",
@@ -2401,7 +2399,7 @@ dependencies = [

 [[package]]
 name = "common-version"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "build-data",
 "const_format",
@@ -2411,7 +2409,7 @@ dependencies = [

 [[package]]
 name = "common-wal"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "common-base",
 "common-error",
@@ -3342,7 +3340,7 @@ dependencies = [

 [[package]]
 name = "datanode"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -3394,7 +3392,7 @@ dependencies = [
 "session",
 "snafu 0.8.5",
 "store-api",
- "substrait 0.13.0",
+ "substrait 0.12.0",
 "table",
 "tokio",
 "toml 0.8.19",
@@ -3403,7 +3401,7 @@ dependencies = [

 [[package]]
 name = "datatypes"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "arrow",
 "arrow-array",
@@ -4047,7 +4045,7 @@ dependencies = [

 [[package]]
 name = "file-engine"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "async-trait",
@@ -4157,7 +4155,7 @@ checksum = "8bf7cc16383c4b8d58b9905a8509f02926ce3058053c056376248d958c9df1e8"

 [[package]]
 name = "flow"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "arrow",
@@ -4167,6 +4165,7 @@ dependencies = [
 "bytes",
 "cache",
 "catalog",
+ "chrono",
 "client",
 "common-base",
 "common-catalog",
@@ -4218,7 +4217,7 @@ dependencies = [
 "snafu 0.8.5",
 "store-api",
 "strum 0.25.0",
- "substrait 0.13.0",
+ "substrait 0.12.0",
 "table",
 "tokio",
 "tonic 0.12.3",
@@ -4273,7 +4272,7 @@ checksum = "6c2141d6d6c8512188a7891b4b01590a45f6dac67afb4f255c4124dbb86d4eaa"

 [[package]]
 name = "frontend"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "arc-swap",
@@ -5541,7 +5540,7 @@ dependencies = [

 [[package]]
 name = "index"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-trait",
 "asynchronous-codec",
@@ -6333,7 +6332,7 @@ checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"

 [[package]]
 name = "log-query"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "chrono",
 "common-error",
@@ -6345,7 +6344,7 @@ dependencies = [

 [[package]]
 name = "log-store"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-stream",
 "async-trait",
@@ -6638,7 +6637,7 @@ dependencies = [

 [[package]]
 name = "meta-client"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "async-trait",
@@ -6665,7 +6664,7 @@ dependencies = [

 [[package]]
 name = "meta-srv"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "async-trait",
@@ -6751,7 +6750,7 @@ dependencies = [

 [[package]]
 name = "metric-engine"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -6849,7 +6848,7 @@ dependencies = [

 [[package]]
 name = "mito2"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -7546,7 +7545,7 @@ dependencies = [

 [[package]]
 name = "object-store"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "anyhow",
 "bytes",
@@ -7795,7 +7794,7 @@ dependencies = [

 [[package]]
 name = "operator"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -7843,7 +7842,7 @@ dependencies = [
 "sql",
 "sqlparser 0.52.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=71dd86058d2af97b9925093d40c4e03360403170)",
 "store-api",
- "substrait 0.13.0",
+ "substrait 0.12.0",
 "table",
 "tokio",
 "tokio-util",
@@ -8080,7 +8079,7 @@ dependencies = [

 [[package]]
 name = "partition"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "async-trait",
@@ -8348,7 +8347,7 @@ checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"

 [[package]]
 name = "pipeline"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -8488,7 +8487,7 @@ dependencies = [

 [[package]]
 name = "plugins"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "auth",
 "clap 4.5.19",
@@ -8750,7 +8749,7 @@ dependencies = [

 [[package]]
 name = "promql"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "ahash 0.8.11",
 "async-trait",
@@ -8995,7 +8994,7 @@ dependencies = [

 [[package]]
 name = "puffin"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-compression 0.4.13",
 "async-trait",
@@ -9036,7 +9035,7 @@ dependencies = [

 [[package]]
 name = "query"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -9101,7 +9100,7 @@ dependencies = [
 "sqlparser 0.52.0 (git+https://github.com/GreptimeTeam/sqlparser-rs.git?rev=71dd86058d2af97b9925093d40c4e03360403170)",
 "statrs",
 "store-api",
- "substrait 0.13.0",
+ "substrait 0.12.0",
 "table",
 "tokio",
 "tokio-stream",
@@ -10446,7 +10445,7 @@ dependencies = [

 [[package]]
 name = "servers"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "ahash 0.8.11",
 "api",
@@ -10563,7 +10562,7 @@ dependencies = [

 [[package]]
 name = "session"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "arc-swap",
@@ -10872,7 +10871,7 @@ dependencies = [

 [[package]]
 name = "sql"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "chrono",
@@ -10926,7 +10925,7 @@ dependencies = [

 [[package]]
 name = "sqlness-runner"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-trait",
 "clap 4.5.19",
@@ -11243,7 +11242,7 @@ dependencies = [

 [[package]]
 name = "store-api"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "aquamarine",
@@ -11373,7 +11372,7 @@ dependencies = [

 [[package]]
 name = "substrait"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "async-trait",
 "bytes",
@@ -11554,7 +11553,7 @@ dependencies = [

 [[package]]
 name = "table"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "async-trait",
@@ -11805,7 +11804,7 @@ checksum = "3369f5ac52d5eb6ab48c6b4ffdc8efbcad6b89c765749064ba298f2c68a16a76"

 [[package]]
 name = "tests-fuzz"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "arbitrary",
 "async-trait",
@@ -11849,7 +11848,7 @@ dependencies = [

 [[package]]
 name = "tests-integration"
-version = "0.13.0"
+version = "0.12.0"
 dependencies = [
 "api",
 "arrow-flight",
@@ -11915,7 +11914,7 @@ dependencies = [
 "sql",
 "sqlx",
 "store-api",
- "substrait 0.13.0",
+ "substrait 0.12.0",
 "table",
 "tempfile",
 "time",
--- a/Cargo.toml
+++ b/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"

--- a/2
+++ b/2
@@ -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
--- a/docker/dev-builder/ubuntu/Dockerfile
+++ b/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.
--- a/docker/dev-builder/ubuntu/Dockerfile-18.10
+++ b/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
--- a/docker/dev-builder/ubuntu/Dockerfile-20.04
+++ b/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
--- a/docs/benchmarks/tsbs/v0.12.0.md
+++ b/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                 |
-
--- a/src/api/src/v1/column_def.rs
+++ b/src/api/src/v1/column_def.rs
@@ -15,8 +15,8 @@
 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, SkippingIndexType,
+    COMMENT_KEY, FULLTEXT_KEY, INVERTED_INDEX_KEY, SKIPPING_INDEX_KEY,
 };
 use greptime_proto::v1::{Analyzer, SkippingIndexType as PbSkippingIndexType};
 use snafu::ResultExt;
@@ -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 {
--- a/src/client/src/lib.rs
+++ b/src/client/src/lib.rs
@@ -16,7 +16,6 @@

 mod client;
 pub mod client_manager;
-#[cfg(feature = "testing")]
 mod database;
 pub mod error;
 pub mod flow;
@@ -34,7 +33,6 @@ pub use common_recordbatch::{RecordBatches, SendableRecordBatchStream};
 use snafu::OptionExt;

 pub use self::client::Client;
-#[cfg(feature = "testing")]
 pub use self::database::Database;
 pub use self::error::{Error, Result};
 use crate::error::{IllegalDatabaseResponseSnafu, ServerSnafu};
--- a/src/cmd/src/flownode.rs
+++ b/src/cmd/src/flownode.rs
@@ -32,7 +32,7 @@ use common_meta::key::TableMetadataManager;
 use common_telemetry::info;
 use common_telemetry::logging::TracingOptions;
 use common_version::{short_version, version};
-use flow::{FlownodeBuilder, FlownodeInstance, FrontendInvoker};
+use flow::{FlownodeBuilder, FlownodeInstance, FrontendClient, FrontendInvoker};
 use meta_client::{MetaClientOptions, MetaClientType};
 use servers::Mode;
 use snafu::{OptionExt, ResultExt};
@@ -317,6 +317,8 @@ impl StartCommand {
            Arc::new(executor),
        );

+        let frontend_client = FrontendClient::from_meta_client(meta_client.clone());
+
        let flow_metadata_manager = Arc::new(FlowMetadataManager::new(cached_meta_backend.clone()));
        let flownode_builder = FlownodeBuilder::new(
            opts,
@@ -324,6 +326,7 @@ impl StartCommand {
            table_metadata_manager,
            catalog_manager.clone(),
            flow_metadata_manager,
+            Arc::new(frontend_client),
        )
        .with_heartbeat_task(heartbeat_task);

--- a/src/cmd/src/standalone.rs
+++ b/src/cmd/src/standalone.rs
@@ -54,7 +54,10 @@ use datanode::config::{DatanodeOptions, ProcedureConfig, RegionEngineConfig, Sto
 use datanode::datanode::{Datanode, DatanodeBuilder};
 use datanode::region_server::RegionServer;
 use file_engine::config::EngineConfig as FileEngineConfig;
-use flow::{FlowConfig, FlowWorkerManager, FlownodeBuilder, FlownodeOptions, FrontendInvoker};
+use flow::{
+    FlowConfig, FlowWorkerManager, FlownodeBuilder, FlownodeOptions, FrontendClient,
+    FrontendInvoker,
+};
 use frontend::frontend::FrontendOptions;
 use frontend::instance::builder::FrontendBuilder;
 use frontend::instance::{FrontendInstance, Instance as FeInstance, StandaloneDatanodeManager};
@@ -533,12 +536,16 @@ impl StartCommand {
            flow: opts.flow.clone(),
            ..Default::default()
        };
+
+        let fe_server_addr = fe_opts.grpc.bind_addr.clone();
+        let frontend_client = FrontendClient::from_static_grpc_addr(fe_server_addr);
        let flow_builder = FlownodeBuilder::new(
            flownode_options,
            plugins.clone(),
            table_metadata_manager.clone(),
            catalog_manager.clone(),
            flow_metadata_manager.clone(),
+            Arc::new(frontend_client),
        );
        let flownode = Arc::new(
            flow_builder
--- a/src/common/function/Cargo.toml
+++ b/src/common/function/Cargo.toml
@@ -28,8 +28,6 @@ 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 }
--- a/src/common/function/src/aggr.rs
+++ b/src/common/function/src/aggr.rs
@@ -12,11 +12,9 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-mod geo_path;
 mod hll;
 mod uddsketch_state;

-pub use geo_path::{GeoPathAccumulator, GEO_PATH_NAME};
 pub(crate) use hll::HllStateType;
 pub use hll::{HllState, HLL_MERGE_NAME, HLL_NAME};
 pub use uddsketch_state::{UddSketchState, UDDSKETCH_STATE_NAME};
--- a/src/common/function/src/aggr/geo_path.rs
+++ b/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");
-        }
-    }
-}
--- a/src/common/function/src/function.rs
+++ b/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>;
--- a/src/common/function/src/function_registry.rs
+++ b/src/common/function/src/function_registry.rs
@@ -18,13 +18,11 @@ 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;
@@ -32,6 +30,7 @@ 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 {
@@ -119,7 +118,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 +130,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)
 });

--- a/src/common/function/src/lib.rs
+++ b/src/common/function/src/lib.rs
@@ -15,11 +15,11 @@
 #![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;
--- a/src/common/function/src/scalars.rs
+++ b/src/common/function/src/scalars.rs
@@ -23,7 +23,6 @@ pub mod math;
 pub mod vector;

 pub(crate) mod hll_count;
-pub mod ip;
 #[cfg(test)]
 pub(crate) mod test;
 pub(crate) mod timestamp;
--- a/src/common/function/src/scalars/date/date_add.rs
+++ b/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() {
--- a/src/common/function/src/scalars/date/date_format.rs
+++ b/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() {
--- a/src/common/function/src/scalars/date/date_sub.rs
+++ b/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() {
--- a/src/common/function/src/scalars/expression/is_null.rs
+++ b/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);
    }
--- a/src/common/function/src/scalars/geo/geohash.rs
+++ b/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 {
--- a/src/common/function/src/scalars/geo/h3.rs
+++ b/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];
--- a/src/common/function/src/scalars/geo/measure.rs
+++ b/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];
--- a/src/common/function/src/scalars/geo/relation.rs
+++ b/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];
--- a/src/common/function/src/scalars/geo/s2.rs
+++ b/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];
--- a/src/common/function/src/scalars/geo/wkt.rs
+++ b/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];
--- a/src/common/function/src/scalars/hll_count.rs
+++ b/src/common/function/src/scalars/hll_count.rs
@@ -71,7 +71,7 @@ impl Function for HllCalcFunction {
        )
    }

-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    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()),
@@ -142,7 +142,7 @@ mod tests {
        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();
+        let result = function.eval(FunctionContext::default(), &args).unwrap();
        assert_eq!(result.len(), 1);

        // Test cardinality estimate
@@ -159,7 +159,7 @@ mod tests {

        // Test with invalid number of arguments
        let args: Vec<VectorRef> = vec![];
-        let result = function.eval(&FunctionContext::default(), &args);
+        let result = function.eval(FunctionContext::default(), &args);
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
@@ -168,7 +168,7 @@ mod tests {

        // 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();
+        let result = function.eval(FunctionContext::default(), &args).unwrap();
        assert_eq!(result.len(), 1);
        assert!(matches!(result.get(0), datatypes::value::Value::Null));
    }
--- a/src/common/function/src/scalars/ip.rs
+++ b/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));
-    }
-}
--- a/src/common/function/src/scalars/ip/cidr.rs
+++ b/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());
-    }
-}
--- a/src/common/function/src/scalars/ip/ipv4.rs
+++ b/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);
-        }
-    }
-}
--- a/src/common/function/src/scalars/ip/ipv6.rs
+++ b/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);
-    }
-}
--- a/src/common/function/src/scalars/ip/range.rs
+++ b/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)
-    }
-}
--- a/src/common/function/src/scalars/json/json_get.rs
+++ b/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());
--- a/src/common/function/src/scalars/json/json_is.rs
+++ b/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() {
--- a/src/common/function/src/scalars/json/json_path_exists.rs
+++ b/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);
--- a/src/common/function/src/scalars/json/json_path_match.rs
+++ b/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());
--- a/src/common/function/src/scalars/json/json_to_string.rs
+++ b/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());
    }
 }
--- a/src/common/function/src/scalars/json/parse_json.rs
+++ b/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() {
--- a/src/common/function/src/scalars/matches.rs
+++ b/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}");
        }
--- a/src/common/function/src/scalars/math.rs
+++ b/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(),
        ))
--- a/src/common/function/src/scalars/math/clamp.rs
+++ b/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());
    }
 }
--- a/src/common/function/src/scalars/math/modulo.rs
+++ b/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"));
--- a/src/common/function/src/scalars/math/pow.rs
+++ b/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 {
--- a/src/common/function/src/scalars/math/rate.rs
+++ b/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);
    }
--- a/src/common/function/src/scalars/test.rs
+++ b/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],
--- a/src/common/function/src/scalars/timestamp/greatest.rs
+++ b/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!(
--- a/src/common/function/src/scalars/timestamp/to_unixtime.rs
+++ b/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);
--- a/src/common/function/src/scalars/uddsketch_calc.rs
+++ b/src/common/function/src/scalars/uddsketch_calc.rs
@@ -75,7 +75,7 @@ impl Function for UddSketchCalcFunction {
        )
    }

-    fn eval(&self, _func_ctx: &FunctionContext, columns: &[VectorRef]) -> Result<VectorRef> {
+    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()),
@@ -169,7 +169,7 @@ mod tests {
            Arc::new(BinaryVector::from(vec![Some(serialized.clone()); 3])),
        ];

-        let result = function.eval(&FunctionContext::default(), &args).unwrap();
+        let result = function.eval(FunctionContext::default(), &args).unwrap();
        assert_eq!(result.len(), 3);

        // Test median (p50)
@@ -192,7 +192,7 @@ mod tests {

        // 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);
+        let result = function.eval(FunctionContext::default(), &args);
        assert!(result.is_err());
        assert!(result
            .unwrap_err()
@@ -204,7 +204,7 @@ mod tests {
            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();
+        let result = function.eval(FunctionContext::default(), &args).unwrap();
        assert_eq!(result.len(), 1);
        assert!(matches!(result.get(0), datatypes::value::Value::Null));
    }
--- a/src/common/function/src/scalars/udf.rs
+++ b/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!(),
        }
--- a/src/common/function/src/scalars/vector.rs
+++ b/src/common/function/src/scalars/vector.rs
@@ -22,7 +22,6 @@ mod scalar_add;
 mod scalar_mul;
 pub(crate) mod sum;
 mod vector_add;
-mod vector_dim;
 mod vector_div;
 mod vector_mul;
 mod vector_norm;
@@ -55,7 +54,6 @@ impl VectorFunction {
        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(elem_sum::ElemSumFunction));
        registry.register(Arc::new(elem_product::ElemProductFunction));
    }
--- a/src/common/function/src/scalars/vector/convert/parse_vector.rs
+++ b/src/common/function/src/scalars/vector/convert/parse_vector.rs
@@ -45,7 +45,7 @@ impl Function for ParseVectorFunction {
        )
    }

-    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 {
@@ -101,7 +101,7 @@ mod tests {
            None,
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input]).unwrap();
+        let result = func.eval(FunctionContext::default(), &[input]).unwrap();

        let result = result.as_ref();
        assert_eq!(result.len(), 3);
@@ -136,7 +136,7 @@ mod tests {
            Some("[7.0,8.0,9.0".to_string()),
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input]);
+        let result = func.eval(FunctionContext::default(), &[input]);
        assert!(result.is_err());

        let input = Arc::new(StringVector::from(vec![
@@ -145,7 +145,7 @@ mod tests {
            Some("7.0,8.0,9.0]".to_string()),
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input]);
+        let result = func.eval(FunctionContext::default(), &[input]);
        assert!(result.is_err());

        let input = Arc::new(StringVector::from(vec![
@@ -154,7 +154,7 @@ mod tests {
            Some("[7.0,hello,9.0]".to_string()),
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input]);
+        let result = func.eval(FunctionContext::default(), &[input]);
        assert!(result.is_err());
    }
 }
--- a/src/common/function/src/scalars/vector/convert/vector_to_string.rs
+++ b/src/common/function/src/scalars/vector/convert/vector_to_string.rs
@@ -46,7 +46,7 @@ impl Function for VectorToStringFunction {
        )
    }

-    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 {
@@ -129,7 +129,7 @@ mod tests {
        builder.push_null();
        let vector = builder.to_vector();

-        let result = func.eval(&FunctionContext::default(), &[vector]).unwrap();
+        let result = func.eval(FunctionContext::default(), &[vector]).unwrap();

        assert_eq!(result.len(), 3);
        assert_eq!(result.get(0), Value::String("[1,2,3]".to_string().into()));
--- a/src/common/function/src/scalars/vector/distance.rs
+++ b/src/common/function/src/scalars/vector/distance.rs
@@ -60,7 +60,7 @@ macro_rules! define_distance_function {
                )
            }

-            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 {
@@ -159,7 +159,7 @@ mod tests {
            ])) as VectorRef;

            let result = func
-                .eval(&FunctionContext::default(), &[vec1.clone(), vec2.clone()])
+                .eval(FunctionContext::default(), &[vec1.clone(), vec2.clone()])
                .unwrap();

            assert!(!result.get(0).is_null());
@@ -168,7 +168,7 @@ mod tests {
            assert!(result.get(3).is_null());

            let result = func
-                .eval(&FunctionContext::default(), &[vec2, vec1])
+                .eval(FunctionContext::default(), &[vec2, vec1])
                .unwrap();

            assert!(!result.get(0).is_null());
@@ -202,7 +202,7 @@ mod tests {
            ])) as VectorRef;

            let result = func
-                .eval(&FunctionContext::default(), &[vec1.clone(), vec2.clone()])
+                .eval(FunctionContext::default(), &[vec1.clone(), vec2.clone()])
                .unwrap();

            assert!(!result.get(0).is_null());
@@ -211,7 +211,7 @@ mod tests {
            assert!(result.get(3).is_null());

            let result = func
-                .eval(&FunctionContext::default(), &[vec2, vec1])
+                .eval(FunctionContext::default(), &[vec2, vec1])
                .unwrap();

            assert!(!result.get(0).is_null());
@@ -245,7 +245,7 @@ mod tests {
            ])) as VectorRef;

            let result = func
-                .eval(&FunctionContext::default(), &[vec1.clone(), vec2.clone()])
+                .eval(FunctionContext::default(), &[vec1.clone(), vec2.clone()])
                .unwrap();

            assert!(!result.get(0).is_null());
@@ -254,7 +254,7 @@ mod tests {
            assert!(result.get(3).is_null());

            let result = func
-                .eval(&FunctionContext::default(), &[vec2, vec1])
+                .eval(FunctionContext::default(), &[vec2, vec1])
                .unwrap();

            assert!(!result.get(0).is_null());
@@ -294,7 +294,7 @@ mod tests {

            let result = func
                .eval(
-                    &FunctionContext::default(),
+                    FunctionContext::default(),
                    &[const_str.clone(), vec1.clone()],
                )
                .unwrap();
@@ -306,7 +306,7 @@ mod tests {

            let result = func
                .eval(
-                    &FunctionContext::default(),
+                    FunctionContext::default(),
                    &[vec1.clone(), const_str.clone()],
                )
                .unwrap();
@@ -318,7 +318,7 @@ mod tests {

            let result = func
                .eval(
-                    &FunctionContext::default(),
+                    FunctionContext::default(),
                    &[const_str.clone(), vec2.clone()],
                )
                .unwrap();
@@ -330,7 +330,7 @@ mod tests {

            let result = func
                .eval(
-                    &FunctionContext::default(),
+                    FunctionContext::default(),
                    &[vec2.clone(), const_str.clone()],
                )
                .unwrap();
@@ -353,13 +353,13 @@ mod tests {
        for func in funcs {
            let vec1 = Arc::new(StringVector::from(vec!["[1.0]"])) as VectorRef;
            let vec2 = Arc::new(StringVector::from(vec!["[1.0, 1.0]"])) as VectorRef;
-            let result = func.eval(&FunctionContext::default(), &[vec1, vec2]);
+            let result = func.eval(FunctionContext::default(), &[vec1, vec2]);
            assert!(result.is_err());

            let vec1 = Arc::new(BinaryVector::from(vec![vec![0, 0, 128, 63]])) as VectorRef;
            let vec2 =
                Arc::new(BinaryVector::from(vec![vec![0, 0, 128, 63, 0, 0, 0, 64]])) as VectorRef;
-            let result = func.eval(&FunctionContext::default(), &[vec1, vec2]);
+            let result = func.eval(FunctionContext::default(), &[vec1, vec2]);
            assert!(result.is_err());
        }
    }
--- a/src/common/function/src/scalars/vector/elem_product.rs
+++ b/src/common/function/src/scalars/vector/elem_product.rs
@@ -68,7 +68,7 @@ impl Function for ElemProductFunction {

    fn eval(
        &self,
-        _func_ctx: &FunctionContext,
+        _func_ctx: FunctionContext,
        columns: &[VectorRef],
    ) -> common_query::error::Result<VectorRef> {
        ensure!(
@@ -131,7 +131,7 @@ mod tests {
            None,
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input0]).unwrap();
+        let result = func.eval(FunctionContext::default(), &[input0]).unwrap();

        let result = result.as_ref();
        assert_eq!(result.len(), 3);
--- a/src/common/function/src/scalars/vector/elem_sum.rs
+++ b/src/common/function/src/scalars/vector/elem_sum.rs
@@ -55,7 +55,7 @@ impl Function for ElemSumFunction {

    fn eval(
        &self,
-        _func_ctx: &FunctionContext,
+        _func_ctx: FunctionContext,
        columns: &[VectorRef],
    ) -> common_query::error::Result<VectorRef> {
        ensure!(
@@ -118,7 +118,7 @@ mod tests {
            None,
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input0]).unwrap();
+        let result = func.eval(FunctionContext::default(), &[input0]).unwrap();

        let result = result.as_ref();
        assert_eq!(result.len(), 3);
--- a/src/common/function/src/scalars/vector/scalar_add.rs
+++ b/src/common/function/src/scalars/vector/scalar_add.rs
@@ -73,7 +73,7 @@ impl Function for ScalarAddFunction {
        )
    }

-    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 {
@@ -154,7 +154,7 @@ mod tests {
        ]));

        let result = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap();

        let result = result.as_ref();
--- a/src/common/function/src/scalars/vector/scalar_mul.rs
+++ b/src/common/function/src/scalars/vector/scalar_mul.rs
@@ -73,7 +73,7 @@ impl Function for ScalarMulFunction {
        )
    }

-    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 {
@@ -154,7 +154,7 @@ mod tests {
        ]));

        let result = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap();

        let result = result.as_ref();
--- a/src/common/function/src/scalars/vector/vector_add.rs
+++ b/src/common/function/src/scalars/vector/vector_add.rs
@@ -72,7 +72,7 @@ impl Function for VectorAddFunction {

    fn eval(
        &self,
-        _func_ctx: &FunctionContext,
+        _func_ctx: FunctionContext,
        columns: &[VectorRef],
    ) -> common_query::error::Result<VectorRef> {
        ensure!(
@@ -166,7 +166,7 @@ mod tests {
        ]));

        let result = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap();

        let result = result.as_ref();
@@ -199,7 +199,7 @@ mod tests {
            Some("[3.0,2.0,2.0]".to_string()),
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input0, input1]);
+        let result = func.eval(FunctionContext::default(), &[input0, input1]);

        match result {
            Err(Error::InvalidFuncArgs { err_msg, .. }) => {
--- a/src/common/function/src/scalars/vector/vector_dim.rs
+++ b/src/common/function/src/scalars/vector/vector_dim.rs
@@ -1,172 +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::borrow::Cow;
-use std::fmt::Display;
-
-use common_query::error::InvalidFuncArgsSnafu;
-use common_query::prelude::{Signature, TypeSignature, Volatility};
-use datatypes::prelude::ConcreteDataType;
-use datatypes::scalars::ScalarVectorBuilder;
-use datatypes::vectors::{MutableVector, UInt64VectorBuilder, VectorRef};
-use snafu::ensure;
-
-use crate::function::{Function, FunctionContext};
-use crate::scalars::vector::impl_conv::{as_veclit, as_veclit_if_const};
-
-const NAME: &str = "vec_dim";
-
-/// Returns the dimension of the vector.
-///
-/// # Example
-///
-/// ```sql
-/// SELECT vec_dim('[7.0, 8.0, 9.0, 10.0]');
-///
-/// +---------------------------------------------------------------+
-/// | vec_dim(Utf8("[7.0, 8.0, 9.0, 10.0]"))                        |
-/// +---------------------------------------------------------------+
-/// | 4                                                             |
-/// +---------------------------------------------------------------+
-///
-#[derive(Debug, Clone, Default)]
-pub struct VectorDimFunction;
-
-impl Function for VectorDimFunction {
-    fn name(&self) -> &str {
-        NAME
-    }
-
-    fn return_type(
-        &self,
-        _input_types: &[ConcreteDataType],
-    ) -> common_query::error::Result<ConcreteDataType> {
-        Ok(ConcreteDataType::uint64_datatype())
-    }
-
-    fn signature(&self) -> Signature {
-        Signature::one_of(
-            vec![
-                TypeSignature::Exact(vec![ConcreteDataType::string_datatype()]),
-                TypeSignature::Exact(vec![ConcreteDataType::binary_datatype()]),
-            ],
-            Volatility::Immutable,
-        )
-    }
-
-    fn eval(
-        &self,
-        _func_ctx: &FunctionContext,
-        columns: &[VectorRef],
-    ) -> common_query::error::Result<VectorRef> {
-        ensure!(
-            columns.len() == 1,
-            InvalidFuncArgsSnafu {
-                err_msg: format!(
-                    "The length of the args is not correct, expect exactly one, have: {}",
-                    columns.len()
-                )
-            }
-        );
-        let arg0 = &columns[0];
-
-        let len = arg0.len();
-        let mut result = UInt64VectorBuilder::with_capacity(len);
-        if len == 0 {
-            return Ok(result.to_vector());
-        }
-
-        let arg0_const = as_veclit_if_const(arg0)?;
-
-        for i in 0..len {
-            let arg0 = match arg0_const.as_ref() {
-                Some(arg0) => Some(Cow::Borrowed(arg0.as_ref())),
-                None => as_veclit(arg0.get_ref(i))?,
-            };
-            let Some(arg0) = arg0 else {
-                result.push_null();
-                continue;
-            };
-            result.push(Some(arg0.len() as u64));
-        }
-
-        Ok(result.to_vector())
-    }
-}
-
-impl Display for VectorDimFunction {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", NAME.to_ascii_uppercase())
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use common_query::error::Error;
-    use datatypes::vectors::StringVector;
-
-    use super::*;
-
-    #[test]
-    fn test_vec_dim() {
-        let func = VectorDimFunction;
-
-        let input0 = Arc::new(StringVector::from(vec![
-            Some("[0.0,2.0,3.0]".to_string()),
-            Some("[1.0,2.0,3.0,4.0]".to_string()),
-            None,
-            Some("[5.0]".to_string()),
-        ]));
-
-        let result = func.eval(&FunctionContext::default(), &[input0]).unwrap();
-
-        let result = result.as_ref();
-        assert_eq!(result.len(), 4);
-        assert_eq!(result.get_ref(0).as_u64().unwrap(), Some(3));
-        assert_eq!(result.get_ref(1).as_u64().unwrap(), Some(4));
-        assert!(result.get_ref(2).is_null());
-        assert_eq!(result.get_ref(3).as_u64().unwrap(), Some(1));
-    }
-
-    #[test]
-    fn test_dim_error() {
-        let func = VectorDimFunction;
-
-        let input0 = Arc::new(StringVector::from(vec![
-            Some("[1.0,2.0,3.0]".to_string()),
-            Some("[4.0,5.0,6.0]".to_string()),
-            None,
-            Some("[2.0,3.0,3.0]".to_string()),
-        ]));
-        let input1 = Arc::new(StringVector::from(vec![
-            Some("[1.0,1.0,1.0]".to_string()),
-            Some("[6.0,5.0,4.0]".to_string()),
-            Some("[3.0,2.0,2.0]".to_string()),
-        ]));
-
-        let result = func.eval(&FunctionContext::default(), &[input0, input1]);
-
-        match result {
-            Err(Error::InvalidFuncArgs { err_msg, .. }) => {
-                assert_eq!(
-                    err_msg,
-                    "The length of the args is not correct, expect exactly one, have: 2"
-                )
-            }
-            _ => unreachable!(),
-        }
-    }
-}
--- a/src/common/function/src/scalars/vector/vector_div.rs
+++ b/src/common/function/src/scalars/vector/vector_div.rs
@@ -68,7 +68,7 @@ impl Function for VectorDivFunction {
        )
    }

-    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 {
@@ -155,7 +155,7 @@ mod tests {
        let input1 = Arc::new(StringVector::from(vec![Some(format!("{vec1:?}"))]));

        let err = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap_err();

        match err {
@@ -186,7 +186,7 @@ mod tests {
        ]));

        let result = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap();

        let result = result.as_ref();
@@ -206,7 +206,7 @@ mod tests {
        let input1 = Arc::new(StringVector::from(vec![Some("[0.0,0.0]".to_string())]));

        let result = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap();

        let result = result.as_ref();
--- a/src/common/function/src/scalars/vector/vector_mul.rs
+++ b/src/common/function/src/scalars/vector/vector_mul.rs
@@ -68,7 +68,7 @@ impl Function for VectorMulFunction {
        )
    }

-    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 {
@@ -155,7 +155,7 @@ mod tests {
        let input1 = Arc::new(StringVector::from(vec![Some(format!("{vec1:?}"))]));

        let err = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap_err();

        match err {
@@ -186,7 +186,7 @@ mod tests {
        ]));

        let result = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap();

        let result = result.as_ref();
--- a/src/common/function/src/scalars/vector/vector_norm.rs
+++ b/src/common/function/src/scalars/vector/vector_norm.rs
@@ -67,7 +67,7 @@ impl Function for VectorNormFunction {

    fn eval(
        &self,
-        _func_ctx: &FunctionContext,
+        _func_ctx: FunctionContext,
        columns: &[VectorRef],
    ) -> common_query::error::Result<VectorRef> {
        ensure!(
@@ -143,7 +143,7 @@ mod tests {
            None,
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input0]).unwrap();
+        let result = func.eval(FunctionContext::default(), &[input0]).unwrap();

        let result = result.as_ref();
        assert_eq!(result.len(), 5);
--- a/src/common/function/src/scalars/vector/vector_sub.rs
+++ b/src/common/function/src/scalars/vector/vector_sub.rs
@@ -72,7 +72,7 @@ impl Function for VectorSubFunction {

    fn eval(
        &self,
-        _func_ctx: &FunctionContext,
+        _func_ctx: FunctionContext,
        columns: &[VectorRef],
    ) -> common_query::error::Result<VectorRef> {
        ensure!(
@@ -166,7 +166,7 @@ mod tests {
        ]));

        let result = func
-            .eval(&FunctionContext::default(), &[input0, input1])
+            .eval(FunctionContext::default(), &[input0, input1])
            .unwrap();

        let result = result.as_ref();
@@ -199,7 +199,7 @@ mod tests {
            Some("[3.0,2.0,2.0]".to_string()),
        ]));

-        let result = func.eval(&FunctionContext::default(), &[input0, input1]);
+        let result = func.eval(FunctionContext::default(), &[input0, input1]);

        match result {
            Err(Error::InvalidFuncArgs { err_msg, .. }) => {
--- a/src/common/function/src/system/build.rs
+++ b/src/common/function/src/system/build.rs
@@ -45,7 +45,7 @@ impl Function for BuildFunction {
        Signature::nullary(Volatility::Immutable)
    }

-    fn eval(&self, _func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
        let build_info = common_version::build_info().to_string();
        let v = Arc::new(StringVector::from(vec![build_info]));
        Ok(v)
@@ -67,7 +67,7 @@ mod tests {
        );
        assert_eq!(build.signature(), Signature::nullary(Volatility::Immutable));
        let build_info = common_version::build_info().to_string();
-        let vector = build.eval(&FunctionContext::default(), &[]).unwrap();
+        let vector = build.eval(FunctionContext::default(), &[]).unwrap();
        let expect: VectorRef = Arc::new(StringVector::from(vec![build_info]));
        assert_eq!(expect, vector);
    }
--- a/src/common/function/src/system/database.rs
+++ b/src/common/function/src/system/database.rs
@@ -47,7 +47,7 @@ impl Function for DatabaseFunction {
        Signature::nullary(Volatility::Immutable)
    }

-    fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
        let db = func_ctx.query_ctx.current_schema();

        Ok(Arc::new(StringVector::from_slice(&[&db])) as _)
@@ -67,7 +67,7 @@ impl Function for CurrentSchemaFunction {
        Signature::uniform(0, vec![], Volatility::Immutable)
    }

-    fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
        let db = func_ctx.query_ctx.current_schema();

        Ok(Arc::new(StringVector::from_slice(&[&db])) as _)
@@ -87,7 +87,7 @@ impl Function for SessionUserFunction {
        Signature::uniform(0, vec![], Volatility::Immutable)
    }

-    fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
        let user = func_ctx.query_ctx.current_user();

        Ok(Arc::new(StringVector::from_slice(&[user.username()])) as _)
@@ -138,7 +138,7 @@ mod tests {
            query_ctx,
            ..Default::default()
        };
-        let vector = build.eval(&func_ctx, &[]).unwrap();
+        let vector = build.eval(func_ctx, &[]).unwrap();
        let expect: VectorRef = Arc::new(StringVector::from(vec!["test_db"]));
        assert_eq!(expect, vector);
    }
--- a/src/common/function/src/system/pg_catalog/pg_get_userbyid.rs
+++ b/src/common/function/src/system/pg_catalog/pg_get_userbyid.rs
@@ -53,7 +53,7 @@ impl Function for PGGetUserByIdFunction {
        )
    }

-    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(), |$T| {
            let col = scalar_unary_op::<<$T as LogicalPrimitiveType>::Native, String, _>(&columns[0], pg_get_user_by_id, &mut EvalContext::default())?;
            Ok(Arc::new(col))
--- a/src/common/function/src/system/pg_catalog/table_is_visible.rs
+++ b/src/common/function/src/system/pg_catalog/table_is_visible.rs
@@ -53,7 +53,7 @@ impl Function for PGTableIsVisibleFunction {
        )
    }

-    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(), |$T| {
            let col = scalar_unary_op::<<$T as LogicalPrimitiveType>::Native, bool, _>(&columns[0], pg_table_is_visible, &mut EvalContext::default())?;
            Ok(Arc::new(col))
--- a/src/common/function/src/system/pg_catalog/version.rs
+++ b/src/common/function/src/system/pg_catalog/version.rs
@@ -44,7 +44,7 @@ impl Function for PGVersionFunction {
        Signature::exact(vec![], Volatility::Immutable)
    }

-    fn eval(&self, _func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, _func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
        let result = StringVector::from(vec![format!(
            "PostgreSQL 16.3 GreptimeDB {}",
            env!("CARGO_PKG_VERSION")
--- a/src/common/function/src/system/timezone.rs
+++ b/src/common/function/src/system/timezone.rs
@@ -41,7 +41,7 @@ impl Function for TimezoneFunction {
        Signature::nullary(Volatility::Immutable)
    }

-    fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
        let tz = func_ctx.query_ctx.timezone().to_string();

        Ok(Arc::new(StringVector::from_slice(&[&tz])) as _)
@@ -77,7 +77,7 @@ mod tests {
            query_ctx,
            ..Default::default()
        };
-        let vector = build.eval(&func_ctx, &[]).unwrap();
+        let vector = build.eval(func_ctx, &[]).unwrap();
        let expect: VectorRef = Arc::new(StringVector::from(vec!["UTC"]));
        assert_eq!(expect, vector);
    }
--- a/src/common/function/src/system/version.rs
+++ b/src/common/function/src/system/version.rs
@@ -45,7 +45,7 @@ impl Function for VersionFunction {
        Signature::nullary(Volatility::Immutable)
    }

-    fn eval(&self, func_ctx: &FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
+    fn eval(&self, func_ctx: FunctionContext, _columns: &[VectorRef]) -> Result<VectorRef> {
        let version = match func_ctx.query_ctx.channel() {
            Channel::Mysql => {
                format!(
--- a/src/common/function/src/table.rs
+++ b/src/common/function/src/table.rs
@@ -26,9 +26,9 @@ use crate::flush_flow::FlushFlowFunction;
 use crate::function_registry::FunctionRegistry;

 /// Table functions
-pub(crate) struct AdminFunction;
+pub(crate) struct TableFunction;

-impl AdminFunction {
+impl TableFunction {
    /// Register all table functions to [`FunctionRegistry`].
    pub fn register(registry: &FunctionRegistry) {
        registry.register_async(Arc::new(MigrateRegionFunction));
--- a/src/common/function/src/table/flush_compact_region.rs
+++ b/src/common/function/src/table/flush_compact_region.rs
--- a/src/common/function/src/table/flush_compact_table.rs
+++ b/src/common/function/src/table/flush_compact_table.rs
--- a/src/common/function/src/table/migrate_region.rs
+++ b/src/common/function/src/table/migrate_region.rs
--- a/src/common/grpc-expr/src/error.rs
+++ b/src/common/grpc-expr/src/error.rs
@@ -111,9 +111,9 @@ pub enum Error {
    },

    #[snafu(display(
-        "Fulltext or Skipping index only supports string type, column: {column_name}, unexpected type: {column_type:?}"
+        "Fulltext index only supports string type, column: {column_name}, unexpected type: {column_type:?}"
    ))]
-    InvalidStringIndexColumnType {
+    InvalidFulltextColumnType {
        column_name: String,
        column_type: ColumnDataType,
        #[snafu(implicit)]
@@ -173,7 +173,7 @@ impl ErrorExt for Error {
                StatusCode::InvalidArguments
            }

-            Error::UnknownColumnDataType { .. } | Error::InvalidStringIndexColumnType { .. } => {
+            Error::UnknownColumnDataType { .. } | Error::InvalidFulltextColumnType { .. } => {
                StatusCode::InvalidArguments
            }
            Error::InvalidSetTableOptionRequest { .. }
--- a/src/common/grpc-expr/src/util.rs
+++ b/src/common/grpc-expr/src/util.rs
@@ -15,7 +15,7 @@
 use std::collections::HashSet;

 use api::v1::column_data_type_extension::TypeExt;
-use api::v1::column_def::{contains_fulltext, contains_skipping};
+use api::v1::column_def::contains_fulltext;
 use api::v1::{
    AddColumn, AddColumns, Column, ColumnDataType, ColumnDataTypeExtension, ColumnDef,
    ColumnOptions, ColumnSchema, CreateTableExpr, JsonTypeExtension, SemanticType,
@@ -27,7 +27,7 @@ use table::table_reference::TableReference;

 use crate::error::{
    self, DuplicatedColumnNameSnafu, DuplicatedTimestampColumnSnafu,
-    InvalidStringIndexColumnTypeSnafu, MissingTimestampColumnSnafu, Result,
+    InvalidFulltextColumnTypeSnafu, MissingTimestampColumnSnafu, Result,
    UnknownColumnDataTypeSnafu,
 };
 pub struct ColumnExpr<'a> {
@@ -152,9 +152,8 @@ pub fn build_create_table_expr(
        let column_type = infer_column_datatype(datatype, datatype_extension)?;

        ensure!(
-            (!contains_fulltext(options) && !contains_skipping(options))
-                || column_type == ColumnDataType::String,
-            InvalidStringIndexColumnTypeSnafu {
+            !contains_fulltext(options) || column_type == ColumnDataType::String,
+            InvalidFulltextColumnTypeSnafu {
                column_name,
                column_type,
            }
--- a/src/common/grpc/src/channel_manager.rs
+++ b/src/common/grpc/src/channel_manager.rs
@@ -445,10 +445,16 @@ impl Pool {

 async fn recycle_channel_in_loop(pool: Arc<Pool>, interval_secs: u64) {
    let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
-
+    // use weak ref here to prevent pool being leaked
+    let pool_weak = Arc::downgrade(&pool);
    loop {
        let _ = interval.tick().await;
-        pool.retain_channel(|_, c| c.access.swap(0, Ordering::Relaxed) != 0)
+        if let Some(pool) = pool_weak.upgrade() {
+            pool.retain_channel(|_, c| c.access.swap(0, Ordering::Relaxed) != 0)
+        } else {
+            // no one is using this pool, so we can also let go
+            break;
+        }
    }
 }

--- a/src/common/meta/src/ddl/create_flow.rs
+++ b/src/common/meta/src/ddl/create_flow.rs
@@ -343,6 +343,7 @@ pub enum FlowType {
 impl FlowType {
    pub const RECORDING_RULE: &str = "recording_rule";
    pub const STREAMING: &str = "streaming";
+    pub const FLOW_TYPE_KEY: &str = "flow_type";
 }

 impl Default for FlowType {
@@ -398,7 +399,8 @@ impl From<&CreateFlowData> for CreateRequest {
        };

        let flow_type = value.flow_type.unwrap_or_default().to_string();
-        req.flow_options.insert("flow_type".to_string(), flow_type);
+        req.flow_options
+            .insert(FlowType::FLOW_TYPE_KEY.to_string(), flow_type);
        req
    }
 }
@@ -430,7 +432,7 @@ impl From<&CreateFlowData> for (FlowInfoValue, Vec<(FlowPartitionId, FlowRouteVa
            .collect::<Vec<_>>();

        let flow_type = value.flow_type.unwrap_or_default().to_string();
-        options.insert("flow_type".to_string(), flow_type);
+        options.insert(FlowType::FLOW_TYPE_KEY.to_string(), flow_type);

        let flow_info = FlowInfoValue {
            source_table_ids: value.source_table_ids.clone(),
--- a/src/common/query/src/error.rs
+++ b/src/common/query/src/error.rs
@@ -30,6 +30,14 @@ use statrs::StatsError;
 #[snafu(visibility(pub))]
 #[stack_trace_debug]
 pub enum Error {
+    #[snafu(display("Failed to execute function"))]
+    ExecuteFunction {
+        #[snafu(source)]
+        error: DataFusionError,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Unsupported input datatypes {:?} in function {}", datatypes, function))]
    UnsupportedInputDataType {
        function: String,
@@ -256,7 +264,9 @@ impl ErrorExt for Error {
            | Error::ArrowCompute { .. }
            | Error::FlownodeNotFound { .. } => StatusCode::EngineExecuteQuery,

-            Error::GeneralDataFusion { error, .. } => datafusion_status_code::<Self>(error, None),
+            Error::ExecuteFunction { error, .. } | Error::GeneralDataFusion { error, .. } => {
+                datafusion_status_code::<Self>(error, None)
+            }

            Error::InvalidInputType { source, .. }
            | Error::IntoVector { source, .. }
--- a/src/common/query/src/function.rs
+++ b/src/common/query/src/function.rs
@@ -17,9 +17,23 @@ use std::sync::Arc;
 use datafusion_expr::ReturnTypeFunction as DfReturnTypeFunction;
 use datatypes::arrow::datatypes::DataType as ArrowDataType;
 use datatypes::prelude::{ConcreteDataType, DataType};
+use datatypes::vectors::VectorRef;
+use snafu::ResultExt;

-use crate::error::Result;
+use crate::error::{ExecuteFunctionSnafu, Result};
 use crate::logical_plan::Accumulator;
+use crate::prelude::{ColumnarValue, ScalarValue};
+
+/// Scalar function
+///
+/// The Fn param is the wrapped function but be aware that the function will
+/// be passed with the slice / vec of columnar values (either scalar or array)
+/// with the exception of zero param function, where a singular element vec
+/// will be passed. In that case the single element is a null array to indicate
+/// the batch's row count (so that the generative zero-argument function can know
+/// the result array size).
+pub type ScalarFunctionImplementation =
+    Arc<dyn Fn(&[ColumnarValue]) -> Result<ColumnarValue> + Send + Sync>;

 /// A function's return type
 pub type ReturnTypeFunction =
@@ -37,6 +51,48 @@ pub type AccumulatorCreatorFunction =
 pub type StateTypeFunction =
    Arc<dyn Fn(&ConcreteDataType) -> Result<Arc<Vec<ConcreteDataType>>> + Send + Sync>;

+/// decorates a function to handle [`ScalarValue`]s by converting them to arrays before calling the function
+/// and vice-versa after evaluation.
+pub fn make_scalar_function<F>(inner: F) -> ScalarFunctionImplementation
+where
+    F: Fn(&[VectorRef]) -> Result<VectorRef> + Sync + Send + 'static,
+{
+    Arc::new(move |args: &[ColumnarValue]| {
+        // first, identify if any of the arguments is an vector. If yes, store its `len`,
+        // as any scalar will need to be converted to an vector of len `len`.
+        let len = args
+            .iter()
+            .fold(Option::<usize>::None, |acc, arg| match arg {
+                ColumnarValue::Scalar(_) => acc,
+                ColumnarValue::Vector(v) => Some(v.len()),
+            });
+
+        // to array
+        // TODO(dennis):  we create new vectors from Scalar on each call,
+        //                            should be optimized in the future.
+        let args: Result<Vec<_>> = if let Some(len) = len {
+            args.iter()
+                .map(|arg| arg.clone().try_into_vector(len))
+                .collect()
+        } else {
+            args.iter()
+                .map(|arg| arg.clone().try_into_vector(1))
+                .collect()
+        };
+
+        let result = (inner)(&args?);
+
+        // maybe back to scalar
+        if len.is_some() {
+            result.map(ColumnarValue::Vector)
+        } else {
+            Ok(ScalarValue::try_from_array(&result?.to_arrow_array(), 0)
+                .map(ColumnarValue::Scalar)
+                .context(ExecuteFunctionSnafu)?)
+        }
+    })
+}
+
 pub fn to_df_return_type(func: ReturnTypeFunction) -> DfReturnTypeFunction {
    let df_func = move |data_types: &[ArrowDataType]| {
        // DataFusion DataType -> ConcreteDataType
@@ -55,3 +111,60 @@ pub fn to_df_return_type(func: ReturnTypeFunction) -> DfReturnTypeFunction {
    };
    Arc::new(df_func)
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use datatypes::prelude::{ScalarVector, Vector};
+    use datatypes::vectors::BooleanVector;
+
+    use super::*;
+
+    #[test]
+    fn test_make_scalar_function() {
+        let and_fun = |args: &[VectorRef]| -> Result<VectorRef> {
+            let left = &args[0]
+                .as_any()
+                .downcast_ref::<BooleanVector>()
+                .expect("cast failed");
+            let right = &args[1]
+                .as_any()
+                .downcast_ref::<BooleanVector>()
+                .expect("cast failed");
+
+            let result = left
+                .iter_data()
+                .zip(right.iter_data())
+                .map(|(left, right)| match (left, right) {
+                    (Some(left), Some(right)) => Some(left && right),
+                    _ => None,
+                })
+                .collect::<BooleanVector>();
+            Ok(Arc::new(result) as VectorRef)
+        };
+
+        let and_fun = make_scalar_function(and_fun);
+
+        let args = vec![
+            ColumnarValue::Scalar(ScalarValue::Boolean(Some(true))),
+            ColumnarValue::Vector(Arc::new(BooleanVector::from(vec![
+                true, false, false, true,
+            ]))),
+        ];
+
+        let vec = (and_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!(),
+        }
+    }
+}
--- a/src/common/query/src/logical_plan.rs
+++ b/src/common/query/src/logical_plan.rs
@@ -15,6 +15,7 @@
 pub mod accumulator;
 mod expr;
 mod udaf;
+mod udf;

 use std::sync::Arc;

@@ -23,14 +24,38 @@ use datafusion::error::Result as DatafusionResult;
 use datafusion::logical_expr::{LogicalPlan, LogicalPlanBuilder};
 use datafusion_common::Column;
 use datafusion_expr::col;
+use datatypes::prelude::ConcreteDataType;
 pub use expr::{build_filter_from_timestamp, build_same_type_ts_filter};

 pub use self::accumulator::{Accumulator, AggregateFunctionCreator, AggregateFunctionCreatorRef};
 pub use self::udaf::AggregateFunction;
+pub use self::udf::ScalarUdf;
 use crate::error::Result;
+use crate::function::{ReturnTypeFunction, ScalarFunctionImplementation};
 use crate::logical_plan::accumulator::*;
 use crate::signature::{Signature, Volatility};

+/// Creates a new UDF with a specific signature and specific return type.
+/// This is a helper function to create a new UDF.
+/// The function `create_udf` returns a subset of all possible `ScalarFunction`:
+/// * the UDF has a fixed return type
+/// * the UDF has a fixed signature (e.g. [f64, f64])
+pub fn create_udf(
+    name: &str,
+    input_types: Vec<ConcreteDataType>,
+    return_type: Arc<ConcreteDataType>,
+    volatility: Volatility,
+    fun: ScalarFunctionImplementation,
+) -> ScalarUdf {
+    let return_type: ReturnTypeFunction = Arc::new(move |_| Ok(return_type.clone()));
+    ScalarUdf::new(
+        name,
+        &Signature::exact(input_types, volatility),
+        &return_type,
+        &fun,
+    )
+}
+
 pub fn create_aggregate_function(
    name: String,
    args_count: u8,
@@ -102,17 +127,102 @@ pub type SubstraitPlanDecoderRef = Arc<dyn SubstraitPlanDecoder + Send + Sync>;
 mod tests {
    use std::sync::Arc;

+    use datafusion_common::DFSchema;
    use datafusion_expr::builder::LogicalTableSource;
-    use datafusion_expr::lit;
+    use datafusion_expr::{
+        lit, ColumnarValue as DfColumnarValue, ScalarUDF as DfScalarUDF,
+        TypeSignature as DfTypeSignature,
+    };
+    use datatypes::arrow::array::BooleanArray;
    use datatypes::arrow::datatypes::{DataType, Field, Schema, SchemaRef};
    use datatypes::prelude::*;
-    use datatypes::vectors::VectorRef;
+    use datatypes::vectors::{BooleanVector, VectorRef};

    use super::*;
    use crate::error::Result;
-    use crate::function::AccumulatorCreatorFunction;
+    use crate::function::{make_scalar_function, AccumulatorCreatorFunction};
+    use crate::prelude::ScalarValue;
    use crate::signature::TypeSignature;

+    #[test]
+    fn test_create_udf() {
+        let and_fun = |args: &[VectorRef]| -> Result<VectorRef> {
+            let left = &args[0]
+                .as_any()
+                .downcast_ref::<BooleanVector>()
+                .expect("cast failed");
+            let right = &args[1]
+                .as_any()
+                .downcast_ref::<BooleanVector>()
+                .expect("cast failed");
+
+            let result = left
+                .iter_data()
+                .zip(right.iter_data())
+                .map(|(left, right)| match (left, right) {
+                    (Some(left), Some(right)) => Some(left && right),
+                    _ => None,
+                })
+                .collect::<BooleanVector>();
+            Ok(Arc::new(result) as VectorRef)
+        };
+
+        let and_fun = make_scalar_function(and_fun);
+
+        let input_types = vec![
+            ConcreteDataType::boolean_datatype(),
+            ConcreteDataType::boolean_datatype(),
+        ];
+
+        let return_type = Arc::new(ConcreteDataType::boolean_datatype());
+
+        let udf = create_udf(
+            "and",
+            input_types.clone(),
+            return_type.clone(),
+            Volatility::Immutable,
+            and_fun.clone(),
+        );
+
+        assert_eq!("and", udf.name);
+        assert!(
+            matches!(&udf.signature.type_signature, TypeSignature::Exact(ts) if ts.clone() == input_types)
+        );
+        assert_eq!(return_type, (udf.return_type)(&[]).unwrap());
+
+        // test into_df_udf
+        let df_udf: DfScalarUDF = udf.into();
+        assert_eq!("and", df_udf.name());
+
+        let types = vec![DataType::Boolean, DataType::Boolean];
+        assert!(
+            matches!(&df_udf.signature().type_signature, DfTypeSignature::Exact(ts) if ts.clone() == types)
+        );
+        assert_eq!(
+            DataType::Boolean,
+            df_udf
+                .return_type_from_exprs(&[], &DFSchema::empty(), &[])
+                .unwrap()
+        );
+
+        let args = vec![
+            DfColumnarValue::Scalar(ScalarValue::Boolean(Some(true))),
+            DfColumnarValue::Array(Arc::new(BooleanArray::from(vec![true, false, false, true]))),
+        ];
+
+        // call the function
+        let result = df_udf.invoke_batch(&args, 4).unwrap();
+        match result {
+            DfColumnarValue::Array(arr) => {
+                let arr = arr.as_any().downcast_ref::<BooleanArray>().unwrap();
+                for i in 0..4 {
+                    assert_eq!(i == 0 || i == 3, arr.value(i));
+                }
+            }
+            _ => unreachable!(),
+        }
+    }
+
    #[derive(Debug)]
    struct DummyAccumulator;

--- a/src/common/query/src/logical_plan/udf.rs
+++ b/src/common/query/src/logical_plan/udf.rs
@@ -0,0 +1,134 @@
+// 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.
+
+//! Udf module contains foundational types that are used to represent UDFs.
+//! It's modified from datafusion.
+use std::any::Any;
+use std::fmt;
+use std::fmt::{Debug, Formatter};
+use std::sync::Arc;
+
+use datafusion_expr::{
+    ColumnarValue as DfColumnarValue,
+    ScalarFunctionImplementation as DfScalarFunctionImplementation, ScalarUDF as DfScalarUDF,
+    ScalarUDFImpl,
+};
+use datatypes::arrow::datatypes::DataType;
+
+use crate::error::Result;
+use crate::function::{ReturnTypeFunction, ScalarFunctionImplementation};
+use crate::prelude::to_df_return_type;
+use crate::signature::Signature;
+
+/// Logical representation of a UDF.
+#[derive(Clone)]
+pub struct ScalarUdf {
+    /// name
+    pub name: String,
+    /// signature
+    pub signature: Signature,
+    /// Return type
+    pub return_type: ReturnTypeFunction,
+    /// actual implementation
+    pub fun: ScalarFunctionImplementation,
+}
+
+impl Debug for ScalarUdf {
+    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+        f.debug_struct("ScalarUdf")
+            .field("name", &self.name)
+            .field("signature", &self.signature)
+            .field("fun", &"<FUNC>")
+            .finish()
+    }
+}
+
+impl ScalarUdf {
+    /// Create a new ScalarUdf
+    pub fn new(
+        name: &str,
+        signature: &Signature,
+        return_type: &ReturnTypeFunction,
+        fun: &ScalarFunctionImplementation,
+    ) -> Self {
+        Self {
+            name: name.to_owned(),
+            signature: signature.clone(),
+            return_type: return_type.clone(),
+            fun: fun.clone(),
+        }
+    }
+}
+
+#[derive(Clone)]
+struct DfUdfAdapter {
+    name: String,
+    signature: datafusion_expr::Signature,
+    return_type: datafusion_expr::ReturnTypeFunction,
+    fun: DfScalarFunctionImplementation,
+}
+
+impl Debug for DfUdfAdapter {
+    fn fmt(&self, f: &mut Formatter) -> fmt::Result {
+        f.debug_struct("DfUdfAdapter")
+            .field("name", &self.name)
+            .field("signature", &self.signature)
+            .finish()
+    }
+}
+
+impl ScalarUDFImpl for DfUdfAdapter {
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn name(&self) -> &str {
+        &self.name
+    }
+
+    fn signature(&self) -> &datafusion_expr::Signature {
+        &self.signature
+    }
+
+    fn return_type(&self, arg_types: &[DataType]) -> datafusion_common::Result<DataType> {
+        (self.return_type)(arg_types).map(|ty| ty.as_ref().clone())
+    }
+
+    fn invoke(&self, args: &[DfColumnarValue]) -> datafusion_common::Result<DfColumnarValue> {
+        (self.fun)(args)
+    }
+
+    fn invoke_no_args(&self, number_rows: usize) -> datafusion_common::Result<DfColumnarValue> {
+        Ok((self.fun)(&[])?.into_array(number_rows)?.into())
+    }
+}
+
+impl From<ScalarUdf> for DfScalarUDF {
+    fn from(udf: ScalarUdf) -> Self {
+        DfScalarUDF::new_from_impl(DfUdfAdapter {
+            name: udf.name,
+            signature: udf.signature.into(),
+            return_type: to_df_return_type(udf.return_type),
+            fun: to_df_scalar_func(udf.fun),
+        })
+    }
+}
+
+fn to_df_scalar_func(fun: ScalarFunctionImplementation) -> DfScalarFunctionImplementation {
+    Arc::new(move |args: &[DfColumnarValue]| {
+        let args: Result<Vec<_>> = args.iter().map(TryFrom::try_from).collect();
+        let result = fun(&args?);
+        result.map(From::from).map_err(|e| e.into())
+    })
+}
--- a/src/common/query/src/prelude.rs
+++ b/src/common/query/src/prelude.rs
@@ -16,7 +16,7 @@ pub use datafusion_common::ScalarValue;

 pub use crate::columnar_value::ColumnarValue;
 pub use crate::function::*;
-pub use crate::logical_plan::AggregateFunction;
+pub use crate::logical_plan::{create_udf, AggregateFunction, ScalarUdf};
 pub use crate::signature::{Signature, TypeSignature, Volatility};

 /// Default timestamp column name for Prometheus metrics.
--- a/src/datanode/src/tests.rs
+++ b/src/datanode/src/tests.rs
@@ -21,6 +21,7 @@ use async_trait::async_trait;
 use common_error::ext::BoxedError;
 use common_function::function::FunctionRef;
 use common_function::scalars::aggregate::AggregateFunctionMetaRef;
+use common_query::prelude::ScalarUdf;
 use common_query::Output;
 use common_runtime::runtime::{BuilderBuild, RuntimeTrait};
 use common_runtime::Runtime;
@@ -76,6 +77,8 @@ impl QueryEngine for MockQueryEngine {
        unimplemented!()
    }

+    fn register_udf(&self, _udf: ScalarUdf) {}
+
    fn register_aggregate_function(&self, _func: AggregateFunctionMetaRef) {}

    fn register_function(&self, _func: FunctionRef) {}
--- a/src/flow/Cargo.toml
+++ b/src/flow/Cargo.toml
@@ -16,6 +16,7 @@ async-trait.workspace = true
 bytes.workspace = true
 cache.workspace = true
 catalog.workspace = true
+chrono.workspace = true
 client.workspace = true
 common-base.workspace = true
 common-config.workspace = true
--- a/src/flow/src/adapter.rs
+++ b/src/flow/src/adapter.rs
@@ -49,12 +49,13 @@ pub(crate) use crate::adapter::node_context::FlownodeContext;
 use crate::adapter::refill::RefillTask;
 use crate::adapter::table_source::ManagedTableSource;
 use crate::adapter::util::relation_desc_to_column_schemas_with_fallback;
-pub(crate) use crate::adapter::worker::{create_worker, Worker, WorkerHandle};
+pub(crate) use crate::adapter::worker::{create_worker, WorkerHandle};
 use crate::compute::ErrCollector;
 use crate::df_optimizer::sql_to_flow_plan;
 use crate::error::{EvalSnafu, ExternalSnafu, InternalSnafu, InvalidQuerySnafu, UnexpectedSnafu};
 use crate::expr::Batch;
 use crate::metrics::{METRIC_FLOW_INSERT_ELAPSED, METRIC_FLOW_ROWS, METRIC_FLOW_RUN_INTERVAL_MS};
+use crate::recording_rules::RecordingRuleEngine;
 use crate::repr::{self, DiffRow, RelationDesc, Row, BATCH_SIZE};

 mod flownode_impl;
@@ -63,7 +64,7 @@ pub(crate) mod refill;
 mod stat;
 #[cfg(test)]
 mod tests;
-mod util;
+pub(crate) mod util;
 mod worker;

 pub(crate) mod node_context;
@@ -171,6 +172,8 @@ pub struct FlowWorkerManager {
    flush_lock: RwLock<()>,
    /// receive a oneshot sender to send state size report
    state_report_handler: RwLock<Option<StateReportHandler>>,
+    /// engine for recording rule
+    rule_engine: RecordingRuleEngine,
 }

 /// Building FlownodeManager
@@ -185,6 +188,7 @@ impl FlowWorkerManager {
        node_id: Option<u32>,
        query_engine: Arc<dyn QueryEngine>,
        table_meta: TableMetadataManagerRef,
+        rule_engine: RecordingRuleEngine,
    ) -> Self {
        let srv_map = ManagedTableSource::new(
            table_meta.table_info_manager().clone(),
@@ -207,6 +211,7 @@ impl FlowWorkerManager {
            node_id,
            flush_lock: RwLock::new(()),
            state_report_handler: RwLock::new(None),
+            rule_engine,
        }
    }

@@ -215,25 +220,6 @@ impl FlowWorkerManager {
        self
    }

-    /// Create a flownode manager with one worker
-    pub fn new_with_workers<'s>(
-        node_id: Option<u32>,
-        query_engine: Arc<dyn QueryEngine>,
-        table_meta: TableMetadataManagerRef,
-        num_workers: usize,
-    ) -> (Self, Vec<Worker<'s>>) {
-        let mut zelf = Self::new(node_id, query_engine, table_meta);
-
-        let workers: Vec<_> = (0..num_workers)
-            .map(|_| {
-                let (handle, worker) = create_worker();
-                zelf.add_worker_handle(handle);
-                worker
-            })
-            .collect();
-        (zelf, workers)
-    }
-
    /// add a worker handler to manager, meaning this corresponding worker is under it's manage
    pub fn add_worker_handle(&mut self, handle: WorkerHandle) {
        self.worker_handles.push(handle);
@@ -751,7 +737,11 @@ pub struct CreateFlowArgs {
 /// Create&Remove flow
 impl FlowWorkerManager {
    /// remove a flow by it's id
+    #[allow(unreachable_code)]
    pub async fn remove_flow(&self, flow_id: FlowId) -> Result<(), Error> {
+        // TODO(discord9): reroute some back to streaming engine later
+        return self.rule_engine.remove_flow(flow_id).await;
+
        for handle in self.worker_handles.iter() {
            if handle.contains_flow(flow_id).await? {
                handle.remove_flow(flow_id).await?;
@@ -767,8 +757,10 @@ impl FlowWorkerManager {
    /// steps to create task:
    /// 1. parse query into typed plan(and optional parse expire_after expr)
    /// 2. render source/sink with output table id and used input table id
-    #[allow(clippy::too_many_arguments)]
+    #[allow(clippy::too_many_arguments, unreachable_code)]
    pub async fn create_flow(&self, args: CreateFlowArgs) -> Result<Option<FlowId>, Error> {
+        // TODO(discord9): reroute some back to streaming engine later
+        return self.rule_engine.create_flow(args).await;
        let CreateFlowArgs {
            flow_id,
            sink_table_name,
--- a/src/flow/src/adapter/flownode_impl.rs
+++ b/src/flow/src/adapter/flownode_impl.rs
@@ -153,7 +153,13 @@ impl Flownode for FlowWorkerManager {
        }
    }

+    #[allow(unreachable_code, unused)]
    async fn handle_inserts(&self, request: InsertRequests) -> Result<FlowResponse> {
+        return self
+            .rule_engine
+            .handle_inserts(request)
+            .await
+            .map_err(to_meta_err(snafu::location!()));
        // using try_read to ensure two things:
        // 1. flush wouldn't happen until inserts before it is inserted
        // 2. inserts happening concurrently with flush wouldn't be block by flush
@@ -206,15 +212,15 @@ impl Flownode for FlowWorkerManager {
                    .collect_vec();
                let table_col_names = table_schema.relation_desc.names;
                let table_col_names = table_col_names
-                    .iter().enumerate()
-                    .map(|(idx,name)| match name {
-                        Some(name) => Ok(name.clone()),
-                        None => InternalSnafu {
-                            reason: format!("Expect column {idx} of table id={table_id} to have name in table schema, found None"),
-                        }
-                        .fail().map_err(BoxedError::new).context(ExternalSnafu),
-                    })
-                    .collect::<Result<Vec<_>>>()?;
+                        .iter().enumerate()
+                        .map(|(idx,name)| match name {
+                            Some(name) => Ok(name.clone()),
+                            None => InternalSnafu {
+                                reason: format!("Expect column {idx} of table id={table_id} to have name in table schema, found None"),
+                            }
+                            .fail().map_err(BoxedError::new).context(ExternalSnafu),
+                        })
+                        .collect::<Result<Vec<_>>>()?;
                let name_to_col = HashMap::<_, _>::from_iter(
                    insert_schema
                        .iter()
--- a/src/flow/src/adapter/util.rs
+++ b/src/flow/src/adapter/util.rs
@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

+//! Some utility functions
+
 use std::sync::Arc;

 use api::helper::ColumnDataTypeWrapper;
--- a/src/flow/src/error.rs
+++ b/src/flow/src/error.rs
@@ -16,6 +16,7 @@

 use std::any::Any;

+use arrow_schema::ArrowError;
 use common_error::ext::BoxedError;
 use common_error::{define_into_tonic_status, from_err_code_msg_to_header};
 use common_macro::stack_trace_debug;
@@ -53,6 +54,13 @@ pub enum Error {
        location: Location,
    },

+    #[snafu(display("Time error"))]
+    Time {
+        source: common_time::error::Error,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("External error"))]
    External {
        source: BoxedError,
@@ -156,6 +164,15 @@ pub enum Error {
        location: Location,
    },

+    #[snafu(display("Arrow error: {raw:?} in context: {context}"))]
+    Arrow {
+        #[snafu(source)]
+        raw: ArrowError,
+        context: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
    #[snafu(display("Datafusion error: {raw:?} in context: {context}"))]
    Datafusion {
        #[snafu(source)]
@@ -230,6 +247,7 @@ impl ErrorExt for Error {
        match self {
            Self::Eval { .. }
            | Self::JoinTask { .. }
+            | Self::Arrow { .. }
            | Self::Datafusion { .. }
            | Self::InsertIntoFlow { .. } => StatusCode::Internal,
            Self::FlowAlreadyExist { .. } => StatusCode::TableAlreadyExists,
@@ -238,7 +256,9 @@ impl ErrorExt for Error {
            | Self::FlowNotFound { .. }
            | Self::ListFlows { .. } => StatusCode::TableNotFound,
            Self::Plan { .. } | Self::Datatypes { .. } => StatusCode::PlanQuery,
-            Self::InvalidQuery { .. } | Self::CreateFlow { .. } => StatusCode::EngineExecuteQuery,
+            Self::InvalidQuery { .. } | Self::CreateFlow { .. } | Self::Time { .. } => {
+                StatusCode::EngineExecuteQuery
+            }
            Self::Unexpected { .. } => StatusCode::Unexpected,
            Self::NotImplemented { .. } | Self::UnsupportedTemporalFilter { .. } => {
                StatusCode::Unsupported
--- a/src/flow/src/expr/utils.rs
+++ b/src/flow/src/expr/utils.rs
@@ -238,6 +238,7 @@ mod test {

        for (sql, current, expected) in &testcases {
            let plan = sql_to_substrait(engine.clone(), sql).await;
+
            let mut ctx = create_test_ctx();
            let flow_plan = TypedPlan::from_substrait_plan(&mut ctx, &plan)
                .await
--- a/src/flow/src/heartbeat.rs
+++ b/src/flow/src/heartbeat.rs
@@ -130,13 +130,6 @@ impl HeartbeatTask {

    pub fn shutdown(&self) {
        info!("Close heartbeat task for flownode");
-        if self
-            .running
-            .compare_exchange(true, false, Ordering::AcqRel, Ordering::Acquire)
-            .is_err()
-        {
-            warn!("Call close heartbeat task multiple times");
-        }
    }

    fn new_heartbeat_request(
--- a/src/flow/src/lib.rs
+++ b/src/flow/src/lib.rs
@@ -33,6 +33,7 @@ mod expr;
 pub mod heartbeat;
 mod metrics;
 mod plan;
+mod recording_rules;
 mod repr;
 mod server;
 mod transform;
@@ -43,4 +44,5 @@ mod test_utils;

 pub use adapter::{FlowConfig, FlowWorkerManager, FlowWorkerManagerRef, FlownodeOptions};
 pub use error::{Error, Result};
+pub use recording_rules::FrontendClient;
 pub use server::{FlownodeBuilder, FlownodeInstance, FlownodeServer, FrontendInvoker};
--- a/src/flow/src/metrics.rs
+++ b/src/flow/src/metrics.rs
@@ -28,6 +28,32 @@ lazy_static! {
        &["table_id"]
    )
    .unwrap();
+    pub static ref METRIC_FLOW_RULE_ENGINE_QUERY_TIME: HistogramVec = register_histogram_vec!(
+        "greptime_flow_rule_engine_query_time",
+        "flow rule engine query time",
+        &["flow_id"],
+        vec![
+            0.0,
+            1.,
+            3.,
+            5.,
+            10.,
+            20.,
+            30.,
+            60.,
+            2. * 60.,
+            5. * 60.,
+            10. * 60.
+        ]
+    )
+    .unwrap();
+    pub static ref METRIC_FLOW_RULE_ENGINE_SLOW_QUERY: HistogramVec = register_histogram_vec!(
+        "greptime_flow_rule_engine_slow_query",
+        "flow rule engine slow query",
+        &["flow_id", "sql", "peer"],
+        vec![60., 2. * 60., 3. * 60., 5. * 60., 10. * 60.]
+    )
+    .unwrap();
    pub static ref METRIC_FLOW_RUN_INTERVAL_MS: IntGauge =
        register_int_gauge!("greptime_flow_run_interval_ms", "flow run interval in ms").unwrap();
    pub static ref METRIC_FLOW_ROWS: IntCounterVec = register_int_counter_vec!(
--- a/src/flow/src/recording_rules.rs
+++ b/src/flow/src/recording_rules.rs
@@ -0,0 +1,940 @@
+// 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.
+
+//! Run flow as recording rule which is time-window-aware normal query triggered every tick set by user
+
+mod engine;
+mod frontend_client;
+
+use std::collections::BTreeSet;
+use std::sync::Arc;
+
+use api::helper::pb_value_to_value_ref;
+use catalog::CatalogManagerRef;
+use common_error::ext::BoxedError;
+use common_recordbatch::DfRecordBatch;
+use common_telemetry::warn;
+use common_time::timestamp::TimeUnit;
+use common_time::Timestamp;
+use datafusion::error::Result as DfResult;
+use datafusion::logical_expr::Expr;
+use datafusion::physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner};
+use datafusion::prelude::SessionContext;
+use datafusion::sql::unparser::Unparser;
+use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion, TreeNodeRewriter};
+use datafusion_common::{DFSchema, TableReference};
+use datafusion_expr::{ColumnarValue, LogicalPlan};
+use datafusion_physical_expr::PhysicalExprRef;
+use datatypes::prelude::{ConcreteDataType, DataType};
+use datatypes::scalars::ScalarVector;
+use datatypes::schema::TIME_INDEX_KEY;
+use datatypes::value::Value;
+use datatypes::vectors::{
+    TimestampMicrosecondVector, TimestampMillisecondVector, TimestampNanosecondVector,
+    TimestampSecondVector, Vector,
+};
+pub use engine::RecordingRuleEngine;
+pub use frontend_client::FrontendClient;
+use itertools::Itertools;
+use query::parser::QueryLanguageParser;
+use query::QueryEngineRef;
+use session::context::QueryContextRef;
+use snafu::{ensure, OptionExt, ResultExt};
+
+use crate::adapter::util::from_proto_to_data_type;
+use crate::df_optimizer::apply_df_optimizer;
+use crate::error::{ArrowSnafu, DatafusionSnafu, DatatypesSnafu, ExternalSnafu, UnexpectedSnafu};
+use crate::expr::error::DataTypeSnafu;
+use crate::Error;
+
+#[derive(Debug, Clone)]
+pub struct TimeWindowExpr {
+    phy_expr: PhysicalExprRef,
+    column_name: String,
+    logical_expr: Expr,
+    df_schema: DFSchema,
+}
+
+impl TimeWindowExpr {
+    pub fn from_expr(expr: &Expr, column_name: &str, df_schema: &DFSchema) -> Result<Self, Error> {
+        let phy_planner = DefaultPhysicalPlanner::default();
+
+        let phy_expr: PhysicalExprRef = phy_planner
+            .create_physical_expr(expr, df_schema, &SessionContext::new().state())
+            .with_context(|_e| DatafusionSnafu {
+                context: format!(
+                    "Failed to create physical expression from {expr:?} using {df_schema:?}"
+                ),
+            })?;
+        Ok(Self {
+            phy_expr,
+            column_name: column_name.to_string(),
+            logical_expr: expr.clone(),
+            df_schema: df_schema.clone(),
+        })
+    }
+
+    pub fn eval(
+        &self,
+        current: Timestamp,
+    ) -> Result<(Option<Timestamp>, Option<Timestamp>), Error> {
+        let lower_bound =
+            find_expr_time_window_lower_bound(&self.logical_expr, &self.df_schema, current)?;
+        let upper_bound =
+            find_expr_time_window_upper_bound(&self.logical_expr, &self.df_schema, current)?;
+        Ok((lower_bound, upper_bound))
+    }
+
+    /// Find timestamps from rows using time window expr
+    pub async fn handle_rows(
+        &self,
+        rows_list: Vec<api::v1::Rows>,
+    ) -> Result<BTreeSet<Timestamp>, Error> {
+        let mut time_windows = BTreeSet::new();
+
+        for rows in rows_list {
+            // pick the time index column and use it to eval on `self.expr`
+            let ts_col_index = rows
+                .schema
+                .iter()
+                .map(|col| col.column_name.clone())
+                .position(|name| name == self.column_name);
+            let Some(ts_col_index) = ts_col_index else {
+                warn!("can't found time index column in schema: {:?}", rows.schema);
+                continue;
+            };
+            let col_schema = &rows.schema[ts_col_index];
+            let cdt = from_proto_to_data_type(col_schema)?;
+
+            let column_values = rows
+                .rows
+                .iter()
+                .map(|row| &row.values[ts_col_index])
+                .collect_vec();
+
+            let mut vector = cdt.create_mutable_vector(column_values.len());
+            for value in column_values {
+                let value = pb_value_to_value_ref(value, &None);
+                vector.try_push_value_ref(value).context(DataTypeSnafu {
+                    msg: "Failed to convert rows to columns",
+                })?;
+            }
+            let vector = vector.to_vector();
+
+            let df_schema = create_df_schema_for_ts_column(&self.column_name, cdt)?;
+
+            let rb =
+                DfRecordBatch::try_new(df_schema.inner().clone(), vec![vector.to_arrow_array()])
+                    .with_context(|_e| ArrowSnafu {
+                        context: format!(
+                            "Failed to create record batch from {df_schema:?} and {vector:?}"
+                        ),
+                    })?;
+
+            let eval_res = self
+                .phy_expr
+                .evaluate(&rb)
+                .with_context(|_| DatafusionSnafu {
+                    context: format!(
+                        "Failed to evaluate physical expression {:?} on {rb:?}",
+                        self.phy_expr
+                    ),
+                })?;
+
+            let res = columnar_to_ts_vector(&eval_res)?;
+
+            for ts in res.into_iter().flatten() {
+                time_windows.insert(ts);
+            }
+        }
+
+        Ok(time_windows)
+    }
+}
+
+fn create_df_schema_for_ts_column(name: &str, cdt: ConcreteDataType) -> Result<DFSchema, Error> {
+    let arrow_schema = Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+        name,
+        cdt.as_arrow_type(),
+        false,
+    )]));
+
+    let df_schema = DFSchema::from_field_specific_qualified_schema(
+        vec![Some(TableReference::bare("TimeIndexOnlyTable"))],
+        &arrow_schema,
+    )
+    .with_context(|_e| DatafusionSnafu {
+        context: format!("Failed to create DFSchema from arrow schema {arrow_schema:?}"),
+    })?;
+
+    Ok(df_schema)
+}
+
+/// Convert `ColumnarValue` to `Vec<Option<Timestamp>>`
+fn columnar_to_ts_vector(columnar: &ColumnarValue) -> Result<Vec<Option<Timestamp>>, Error> {
+    let val = match columnar {
+        datafusion_expr::ColumnarValue::Array(array) => {
+            let ty = array.data_type();
+            let ty = ConcreteDataType::from_arrow_type(ty);
+            let time_unit = if let ConcreteDataType::Timestamp(ty) = ty {
+                ty.unit()
+            } else {
+                return UnexpectedSnafu {
+                    reason: format!("Non-timestamp type: {ty:?}"),
+                }
+                .fail();
+            };
+
+            match time_unit {
+                TimeUnit::Second => TimestampSecondVector::try_from_arrow_array(array.clone())
+                    .with_context(|_| DatatypesSnafu {
+                        extra: format!("Failed to create vector from arrow array {array:?}"),
+                    })?
+                    .iter_data()
+                    .map(|d| d.map(|d| d.0))
+                    .collect_vec(),
+                TimeUnit::Millisecond => {
+                    TimestampMillisecondVector::try_from_arrow_array(array.clone())
+                        .with_context(|_| DatatypesSnafu {
+                            extra: format!("Failed to create vector from arrow array {array:?}"),
+                        })?
+                        .iter_data()
+                        .map(|d| d.map(|d| d.0))
+                        .collect_vec()
+                }
+                TimeUnit::Microsecond => {
+                    TimestampMicrosecondVector::try_from_arrow_array(array.clone())
+                        .with_context(|_| DatatypesSnafu {
+                            extra: format!("Failed to create vector from arrow array {array:?}"),
+                        })?
+                        .iter_data()
+                        .map(|d| d.map(|d| d.0))
+                        .collect_vec()
+                }
+                TimeUnit::Nanosecond => {
+                    TimestampNanosecondVector::try_from_arrow_array(array.clone())
+                        .with_context(|_| DatatypesSnafu {
+                            extra: format!("Failed to create vector from arrow array {array:?}"),
+                        })?
+                        .iter_data()
+                        .map(|d| d.map(|d| d.0))
+                        .collect_vec()
+                }
+            }
+        }
+        datafusion_expr::ColumnarValue::Scalar(scalar) => {
+            let value = Value::try_from(scalar.clone()).with_context(|_| DatatypesSnafu {
+                extra: format!("Failed to convert scalar {scalar:?} to value"),
+            })?;
+            let ts = value.as_timestamp().context(UnexpectedSnafu {
+                reason: format!("Expect Timestamp, found {:?}", value),
+            })?;
+            vec![Some(ts)]
+        }
+    };
+    Ok(val)
+}
+
+/// Convert sql to datafusion logical plan
+pub async fn sql_to_df_plan(
+    query_ctx: QueryContextRef,
+    engine: QueryEngineRef,
+    sql: &str,
+    optimize: bool,
+) -> Result<LogicalPlan, Error> {
+    let stmt = QueryLanguageParser::parse_sql(sql, &query_ctx)
+        .map_err(BoxedError::new)
+        .context(ExternalSnafu)?;
+    let plan = engine
+        .planner()
+        .plan(&stmt, query_ctx)
+        .await
+        .map_err(BoxedError::new)
+        .context(ExternalSnafu)?;
+    let plan = if optimize {
+        apply_df_optimizer(plan).await?
+    } else {
+        plan
+    };
+    Ok(plan)
+}
+
+/// Return (the column name of time index column, the time window expr, the expected time unit of time index column, the expr's schema for evaluating the time window)
+async fn find_time_window_expr(
+    plan: &LogicalPlan,
+    catalog_man: CatalogManagerRef,
+    query_ctx: QueryContextRef,
+) -> Result<(String, Option<datafusion_expr::Expr>, TimeUnit, DFSchema), Error> {
+    // TODO(discord9): find the expr that do time window
+
+    let mut table_name = None;
+
+    // first find the table source in the logical plan
+    plan.apply(|plan| {
+        let LogicalPlan::TableScan(table_scan) = plan else {
+            return Ok(TreeNodeRecursion::Continue);
+        };
+        table_name = Some(table_scan.table_name.clone());
+        Ok(TreeNodeRecursion::Stop)
+    })
+    .with_context(|_| DatafusionSnafu {
+        context: format!("Can't find table source in plan {plan:?}"),
+    })?;
+    let Some(table_name) = table_name else {
+        UnexpectedSnafu {
+            reason: format!("Can't find table source in plan {plan:?}"),
+        }
+        .fail()?
+    };
+
+    let current_schema = query_ctx.current_schema();
+
+    let catalog_name = table_name.catalog().unwrap_or(query_ctx.current_catalog());
+    let schema_name = table_name.schema().unwrap_or(&current_schema);
+    let table_name = table_name.table();
+
+    let Some(table_ref) = catalog_man
+        .table(catalog_name, schema_name, table_name, Some(&query_ctx))
+        .await
+        .map_err(BoxedError::new)
+        .context(ExternalSnafu)?
+    else {
+        UnexpectedSnafu {
+            reason: format!(
+                "Can't find table {table_name:?} in catalog {catalog_name:?}/{schema_name:?}"
+            ),
+        }
+        .fail()?
+    };
+
+    let schema = &table_ref.table_info().meta.schema;
+
+    let ts_index = schema.timestamp_column().context(UnexpectedSnafu {
+        reason: format!("Can't find timestamp column in table {table_name:?}"),
+    })?;
+
+    let ts_col_name = ts_index.name.clone();
+
+    let expected_time_unit = ts_index.data_type.as_timestamp().with_context(|| UnexpectedSnafu {
+        reason: format!(
+            "Expected timestamp column {ts_col_name:?} in table {table_name:?} to be timestamp, but got {ts_index:?}"
+        ),
+    })?.unit();
+
+    let arrow_schema = Arc::new(arrow_schema::Schema::new(vec![arrow_schema::Field::new(
+        ts_col_name.clone(),
+        ts_index.data_type.as_arrow_type(),
+        false,
+    )]));
+
+    let df_schema = DFSchema::from_field_specific_qualified_schema(
+        vec![Some(TableReference::bare(table_name))],
+        &arrow_schema,
+    )
+    .with_context(|_e| DatafusionSnafu {
+        context: format!("Failed to create DFSchema from arrow schema {arrow_schema:?}"),
+    })?;
+
+    // find the time window expr which refers to the time index column
+    let mut aggr_expr = None;
+    let mut time_window_expr: Option<Expr> = None;
+
+    let find_inner_aggr_expr = |plan: &LogicalPlan| {
+        if let LogicalPlan::Aggregate(aggregate) = plan {
+            aggr_expr = Some(aggregate.clone());
+        };
+
+        Ok(TreeNodeRecursion::Continue)
+    };
+    plan.apply(find_inner_aggr_expr)
+        .with_context(|_| DatafusionSnafu {
+            context: format!("Can't find aggr expr in plan {plan:?}"),
+        })?;
+
+    if let Some(aggregate) = aggr_expr {
+        for group_expr in &aggregate.group_expr {
+            let refs = group_expr.column_refs();
+            if refs.len() != 1 {
+                continue;
+            }
+            let ref_col = refs.iter().next().unwrap();
+
+            let index = aggregate.input.schema().maybe_index_of_column(ref_col);
+            let Some(index) = index else {
+                continue;
+            };
+            let field = aggregate.input.schema().field(index);
+
+            let is_time_index = field.metadata().get(TIME_INDEX_KEY) == Some(&"true".to_string());
+
+            if is_time_index {
+                let rewrite_column = group_expr.clone();
+                let rewritten = rewrite_column
+                    .rewrite(&mut RewriteColumn {
+                        table_name: table_name.to_string(),
+                    })
+                    .with_context(|_| DatafusionSnafu {
+                        context: format!("Rewrite expr failed, expr={:?}", group_expr),
+                    })?
+                    .data;
+                struct RewriteColumn {
+                    table_name: String,
+                }
+
+                impl TreeNodeRewriter for RewriteColumn {
+                    type Node = Expr;
+                    fn f_down(&mut self, node: Self::Node) -> DfResult<Transformed<Self::Node>> {
+                        let Expr::Column(mut column) = node else {
+                            return Ok(Transformed::no(node));
+                        };
+
+                        column.relation = Some(TableReference::bare(self.table_name.clone()));
+
+                        Ok(Transformed::yes(Expr::Column(column)))
+                    }
+                }
+
+                time_window_expr = Some(rewritten);
+                break;
+            }
+        }
+        Ok((ts_col_name, time_window_expr, expected_time_unit, df_schema))
+    } else {
+        // can't found time window expr, return None
+        Ok((ts_col_name, None, expected_time_unit, df_schema))
+    }
+}
+
+/// Find nearest lower bound for time `current` in given `plan` for the time window expr.
+/// i.e. for time window expr being `date_bin(INTERVAL '5 minutes', ts) as time_window` and `current="2021-07-01 00:01:01.000"`,
+/// return `Some("2021-07-01 00:00:00.000")`
+/// if `plan` doesn't contain a `TIME INDEX` column, return `None`
+///
+/// Time window expr is a expr that:
+/// 1. ref only to a time index column
+/// 2. is monotonic increasing
+/// 3. show up in GROUP BY clause
+///
+/// note this plan should only contain one TableScan
+pub async fn find_plan_time_window_bound(
+    plan: &LogicalPlan,
+    current: Timestamp,
+    query_ctx: QueryContextRef,
+    engine: QueryEngineRef,
+) -> Result<(String, Option<Timestamp>, Option<Timestamp>), Error> {
+    // TODO(discord9): find the expr that do time window
+    let catalog_man = engine.engine_state().catalog_manager();
+
+    let (ts_col_name, time_window_expr, expected_time_unit, df_schema) =
+        find_time_window_expr(plan, catalog_man.clone(), query_ctx).await?;
+    // cast current to ts_index's type
+    let new_current = current
+        .convert_to(expected_time_unit)
+        .with_context(|| UnexpectedSnafu {
+            reason: format!("Failed to cast current timestamp {current:?} to {expected_time_unit}"),
+        })?;
+
+    // if no time_window_expr is found, return None
+    if let Some(time_window_expr) = time_window_expr {
+        let lower_bound =
+            find_expr_time_window_lower_bound(&time_window_expr, &df_schema, new_current)?;
+        let upper_bound =
+            find_expr_time_window_upper_bound(&time_window_expr, &df_schema, new_current)?;
+        Ok((ts_col_name, lower_bound, upper_bound))
+    } else {
+        Ok((ts_col_name, None, None))
+    }
+}
+
+/// Find the lower bound of time window in given `expr` and `current` timestamp.
+///
+/// i.e. for `current="2021-07-01 00:01:01.000"` and `expr=date_bin(INTERVAL '5 minutes', ts) as time_window` and `ts_col=ts`,
+/// return `Some("2021-07-01 00:00:00.000")` since it's the lower bound
+/// return `Some("2021-07-01 00:00:00.000")` since it's the lower bound
+/// of current time window given the current timestamp
+///
+/// if return None, meaning this time window have no lower bound
+fn find_expr_time_window_lower_bound(
+    expr: &Expr,
+    df_schema: &DFSchema,
+    current: Timestamp,
+) -> Result<Option<Timestamp>, Error> {
+    let phy_planner = DefaultPhysicalPlanner::default();
+
+    let phy_expr: PhysicalExprRef = phy_planner
+        .create_physical_expr(expr, df_schema, &SessionContext::new().state())
+        .with_context(|_e| DatafusionSnafu {
+            context: format!(
+                "Failed to create physical expression from {expr:?} using {df_schema:?}"
+            ),
+        })?;
+
+    let cur_time_window = eval_ts_to_ts(&phy_expr, df_schema, current)?;
+    let input_time_unit = cur_time_window.unit();
+    Ok(cur_time_window.convert_to(input_time_unit))
+}
+
+/// Find the upper bound for time window expression
+fn find_expr_time_window_upper_bound(
+    expr: &Expr,
+    df_schema: &DFSchema,
+    current: Timestamp,
+) -> Result<Option<Timestamp>, Error> {
+    use std::cmp::Ordering;
+
+    let phy_planner = DefaultPhysicalPlanner::default();
+
+    let phy_expr: PhysicalExprRef = phy_planner
+        .create_physical_expr(expr, df_schema, &SessionContext::new().state())
+        .with_context(|_e| DatafusionSnafu {
+            context: format!(
+                "Failed to create physical expression from {expr:?} using {df_schema:?}"
+            ),
+        })?;
+
+    let cur_time_window = eval_ts_to_ts(&phy_expr, df_schema, current)?;
+
+    // search to find the lower bound
+    let mut offset: i64 = 1;
+    let mut lower_bound = Some(current);
+    let upper_bound;
+    // first expontial probe to found a range for binary search
+    loop {
+        let Some(next_val) = current.value().checked_add(offset) else {
+            // no upper bound if overflow
+            return Ok(None);
+        };
+
+        let next_time_probe = common_time::Timestamp::new(next_val, current.unit());
+
+        let next_time_window = eval_ts_to_ts(&phy_expr, df_schema, next_time_probe)?;
+
+        match next_time_window.cmp(&cur_time_window) {
+            Ordering::Less => {UnexpectedSnafu {
+                reason: format!(
+                    "Unsupported time window expression, expect monotonic increasing for time window expression {expr:?}"
+                ),
+            }
+            .fail()?
+            }
+            Ordering::Equal => {
+                lower_bound = Some(next_time_probe);
+            }
+            Ordering::Greater => {
+                upper_bound = Some(next_time_probe);
+                break
+            }
+        }
+
+        let Some(new_offset) = offset.checked_mul(2) else {
+            // no upper bound if overflow
+            return Ok(None);
+        };
+        offset = new_offset;
+    }
+
+    // binary search for the exact upper bound
+
+    ensure!(lower_bound.map(|v|v.unit())==upper_bound.map(|v|v.unit()), UnexpectedSnafu{
+        reason: format!(" unit mismatch for time window expression {expr:?}, found {lower_bound:?} and {upper_bound:?}"),
+    });
+
+    let output_unit = upper_bound
+        .context(UnexpectedSnafu {
+            reason: "should have lower bound",
+        })?
+        .unit();
+
+    let mut low = lower_bound
+        .context(UnexpectedSnafu {
+            reason: "should have lower bound",
+        })?
+        .value();
+    let mut high = upper_bound
+        .context(UnexpectedSnafu {
+            reason: "should have upper bound",
+        })?
+        .value();
+    while low < high {
+        let mid = (low + high) / 2;
+        let mid_probe = common_time::Timestamp::new(mid, output_unit);
+        let mid_time_window = eval_ts_to_ts(&phy_expr, df_schema, mid_probe)?;
+
+        match mid_time_window.cmp(&cur_time_window) {
+            Ordering::Less => UnexpectedSnafu {
+                reason: format!("Binary search failed for time window expression {expr:?}"),
+            }
+            .fail()?,
+            Ordering::Equal => low = mid + 1,
+            Ordering::Greater => high = mid,
+        }
+    }
+
+    let final_upper_bound_for_time_window = common_time::Timestamp::new(high, output_unit);
+
+    Ok(Some(final_upper_bound_for_time_window))
+}
+
+fn eval_ts_to_ts(
+    phy: &PhysicalExprRef,
+    df_schema: &DFSchema,
+    input_value: Timestamp,
+) -> Result<Timestamp, Error> {
+    let schema_ty = df_schema.field(0).data_type();
+    let schema_cdt = ConcreteDataType::from_arrow_type(schema_ty);
+    let schema_unit = if let ConcreteDataType::Timestamp(ts) = schema_cdt {
+        ts.unit()
+    } else {
+        return UnexpectedSnafu {
+            reason: format!("Expect Timestamp, found {:?}", schema_cdt),
+        }
+        .fail();
+    };
+    let input_value = input_value
+        .convert_to(schema_unit)
+        .with_context(|| UnexpectedSnafu {
+            reason: format!("Failed to convert timestamp {input_value:?} to {schema_unit}"),
+        })?;
+    let ts_vector = match schema_unit {
+        TimeUnit::Second => {
+            TimestampSecondVector::from_vec(vec![input_value.value()]).to_arrow_array()
+        }
+        TimeUnit::Millisecond => {
+            TimestampMillisecondVector::from_vec(vec![input_value.value()]).to_arrow_array()
+        }
+        TimeUnit::Microsecond => {
+            TimestampMicrosecondVector::from_vec(vec![input_value.value()]).to_arrow_array()
+        }
+        TimeUnit::Nanosecond => {
+            TimestampNanosecondVector::from_vec(vec![input_value.value()]).to_arrow_array()
+        }
+    };
+
+    let rb = DfRecordBatch::try_new(df_schema.inner().clone(), vec![ts_vector.clone()])
+        .with_context(|_| ArrowSnafu {
+            context: format!("Failed to create record batch from {df_schema:?} and {ts_vector:?}"),
+        })?;
+
+    let eval_res = phy.evaluate(&rb).with_context(|_| DatafusionSnafu {
+        context: format!("Failed to evaluate physical expression {phy:?} on {rb:?}"),
+    })?;
+
+    if let Some(Some(ts)) = columnar_to_ts_vector(&eval_res)?.first() {
+        Ok(*ts)
+    } else {
+        UnexpectedSnafu {
+            reason: format!(
+                "Expected timestamp in expression {phy:?} but got {:?}",
+                eval_res
+            ),
+        }
+        .fail()?
+    }
+}
+
+// TODO(discord9): a method to found out the precise time window
+
+/// Find out the `Filter` Node corresponding to outermost `WHERE` and add a new filter expr to it
+#[derive(Debug)]
+pub struct AddFilterRewriter {
+    extra_filter: Expr,
+    is_rewritten: bool,
+}
+
+impl AddFilterRewriter {
+    fn new(filter: Expr) -> Self {
+        Self {
+            extra_filter: filter,
+            is_rewritten: false,
+        }
+    }
+}
+
+impl TreeNodeRewriter for AddFilterRewriter {
+    type Node = LogicalPlan;
+    fn f_up(&mut self, node: Self::Node) -> DfResult<Transformed<Self::Node>> {
+        if self.is_rewritten {
+            return Ok(Transformed::no(node));
+        }
+        match node {
+            LogicalPlan::Filter(mut filter) if !filter.having => {
+                filter.predicate = filter.predicate.and(self.extra_filter.clone());
+                self.is_rewritten = true;
+                Ok(Transformed::yes(LogicalPlan::Filter(filter)))
+            }
+            LogicalPlan::TableScan(_) => {
+                // add a new filter
+                let filter =
+                    datafusion_expr::Filter::try_new(self.extra_filter.clone(), Arc::new(node))?;
+                self.is_rewritten = true;
+                Ok(Transformed::yes(LogicalPlan::Filter(filter)))
+            }
+            _ => Ok(Transformed::no(node)),
+        }
+    }
+}
+
+fn df_plan_to_sql(plan: &LogicalPlan) -> Result<String, Error> {
+    /// A dialect that forces all identifiers to be quoted
+    struct ForceQuoteIdentifiers;
+    impl datafusion::sql::unparser::dialect::Dialect for ForceQuoteIdentifiers {
+        fn identifier_quote_style(&self, identifier: &str) -> Option<char> {
+            if identifier.to_lowercase() != identifier {
+                Some('"')
+            } else {
+                None
+            }
+        }
+    }
+    let unparser = Unparser::new(&ForceQuoteIdentifiers);
+    // first make all column qualified
+    let sql = unparser
+        .plan_to_sql(plan)
+        .with_context(|_e| DatafusionSnafu {
+            context: format!("Failed to unparse logical plan {plan:?}"),
+        })?;
+    Ok(sql.to_string())
+}
+
+#[cfg(test)]
+mod test {
+    use datafusion_common::tree_node::TreeNode;
+    use pretty_assertions::assert_eq;
+    use session::context::QueryContext;
+
+    use super::{sql_to_df_plan, *};
+    use crate::recording_rules::{df_plan_to_sql, AddFilterRewriter};
+    use crate::test_utils::create_test_query_engine;
+
+    #[tokio::test]
+    async fn test_sql_plan_convert() {
+        let query_engine = create_test_query_engine();
+        let ctx = QueryContext::arc();
+        let old = r#"SELECT "NUMBER" FROM "UPPERCASE_NUMBERS_WITH_TS""#;
+        let new = sql_to_df_plan(ctx.clone(), query_engine.clone(), old, false)
+            .await
+            .unwrap();
+        let new_sql = df_plan_to_sql(&new).unwrap();
+
+        assert_eq!(
+            r#"SELECT "UPPERCASE_NUMBERS_WITH_TS"."NUMBER" FROM "UPPERCASE_NUMBERS_WITH_TS""#,
+            new_sql
+        );
+    }
+
+    #[tokio::test]
+    async fn test_add_filter() {
+        let testcases = vec![
+            (
+                "SELECT number FROM numbers_with_ts GROUP BY number","SELECT numbers_with_ts.number FROM numbers_with_ts WHERE (number > 4) GROUP BY numbers_with_ts.number"
+            ),
+            (
+                "SELECT number FROM numbers_with_ts WHERE number < 2 OR number >10",
+                "SELECT numbers_with_ts.number FROM numbers_with_ts WHERE ((numbers_with_ts.number < 2) OR (numbers_with_ts.number > 10)) AND (number > 4)"
+            ),
+            (
+                "SELECT date_bin('5 minutes', ts) as time_window FROM numbers_with_ts GROUP BY time_window",
+                "SELECT date_bin('5 minutes', numbers_with_ts.ts) AS time_window FROM numbers_with_ts WHERE (number > 4) GROUP BY date_bin('5 minutes', numbers_with_ts.ts)"
+            )
+        ];
+        use datafusion_expr::{col, lit};
+        let query_engine = create_test_query_engine();
+        let ctx = QueryContext::arc();
+
+        for (before, after) in testcases {
+            let sql = before;
+            let plan = sql_to_df_plan(ctx.clone(), query_engine.clone(), sql, false)
+                .await
+                .unwrap();
+
+            let mut add_filter = AddFilterRewriter::new(col("number").gt(lit(4u32)));
+            let plan = plan.rewrite(&mut add_filter).unwrap().data;
+            let new_sql = df_plan_to_sql(&plan).unwrap();
+            assert_eq!(after, new_sql);
+        }
+    }
+
+    #[tokio::test]
+    async fn test_plan_time_window_lower_bound() {
+        use datafusion_expr::{col, lit};
+        let query_engine = create_test_query_engine();
+        let ctx = QueryContext::arc();
+
+        let testcases = [
+            // same alias is not same column
+            (
+                "SELECT arrow_cast(date_bin(INTERVAL '1 MINS', numbers_with_ts.ts), 'Timestamp(Second, None)') AS ts FROM numbers_with_ts GROUP BY ts;",
+                Timestamp::new(1740394109, TimeUnit::Second),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(1740394109000, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(1740394109001, TimeUnit::Millisecond)),
+                ),
+                r#"SELECT arrow_cast(date_bin(INTERVAL '1 MINS', numbers_with_ts.ts), 'Timestamp(Second, None)') AS ts FROM numbers_with_ts WHERE ((ts >= CAST('2025-02-24 10:48:29' AS TIMESTAMP)) AND (ts <= CAST('2025-02-24 10:48:29.001' AS TIMESTAMP))) GROUP BY numbers_with_ts.ts"#
+            ),
+            // complex time window index
+            (
+                "SELECT arrow_cast(date_bin(INTERVAL '1 MINS', numbers_with_ts.ts), 'Timestamp(Second, None)') AS time_window FROM numbers_with_ts GROUP BY time_window;",
+                Timestamp::new(1740394109, TimeUnit::Second),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(1740394080, TimeUnit::Second)),
+                    Some(Timestamp::new(1740394140, TimeUnit::Second)),
+                ),
+                "SELECT arrow_cast(date_bin(INTERVAL '1 MINS', numbers_with_ts.ts), 'Timestamp(Second, None)') AS time_window FROM numbers_with_ts WHERE ((ts >= CAST('2025-02-24 10:48:00' AS TIMESTAMP)) AND (ts <= CAST('2025-02-24 10:49:00' AS TIMESTAMP))) GROUP BY arrow_cast(date_bin(INTERVAL '1 MINS', numbers_with_ts.ts), 'Timestamp(Second, None)')"
+            ),
+            // no time index
+            (
+                "SELECT date_bin('5 minutes', ts) FROM numbers_with_ts;",
+                Timestamp::new(23, TimeUnit::Millisecond),
+                ("ts".to_string(), None, None),
+                "SELECT date_bin('5 minutes', ts) FROM numbers_with_ts;"
+            ),
+            // time index
+            (
+                "SELECT date_bin('5 minutes', ts) as time_window FROM numbers_with_ts GROUP BY time_window;",
+                Timestamp::new(23, TimeUnit::Nanosecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT date_bin('5 minutes', numbers_with_ts.ts) AS time_window FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP))) GROUP BY date_bin('5 minutes', numbers_with_ts.ts)"
+            ),
+            // on spot
+            (
+                "SELECT date_bin('5 minutes', ts) as time_window FROM numbers_with_ts GROUP BY time_window;",
+                Timestamp::new(0, TimeUnit::Nanosecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT date_bin('5 minutes', numbers_with_ts.ts) AS time_window FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP))) GROUP BY date_bin('5 minutes', numbers_with_ts.ts)"
+            ),
+            // different time unit
+            (
+                "SELECT date_bin('5 minutes', ts) as time_window FROM numbers_with_ts GROUP BY time_window;",
+                Timestamp::new(23_000_000, TimeUnit::Nanosecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT date_bin('5 minutes', numbers_with_ts.ts) AS time_window FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP))) GROUP BY date_bin('5 minutes', numbers_with_ts.ts)"
+            ),
+            // time index with other fields
+            (
+                "SELECT sum(number) as sum_up, date_bin('5 minutes', ts) as time_window FROM numbers_with_ts GROUP BY time_window;",
+                Timestamp::new(23, TimeUnit::Millisecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT sum(numbers_with_ts.number) AS sum_up, date_bin('5 minutes', numbers_with_ts.ts) AS time_window FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP))) GROUP BY date_bin('5 minutes', numbers_with_ts.ts)"
+            ),
+            // time index with other pks
+            (
+                "SELECT number, date_bin('5 minutes', ts) as time_window FROM numbers_with_ts GROUP BY time_window, number;",
+                Timestamp::new(23, TimeUnit::Millisecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT numbers_with_ts.number, date_bin('5 minutes', numbers_with_ts.ts) AS time_window FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP))) GROUP BY date_bin('5 minutes', numbers_with_ts.ts), numbers_with_ts.number"
+            ),
+            // subquery
+            (
+                "SELECT number, time_window FROM (SELECT number, date_bin('5 minutes', ts) as time_window FROM numbers_with_ts GROUP BY time_window, number);",
+                Timestamp::new(23, TimeUnit::Millisecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT numbers_with_ts.number, time_window FROM (SELECT numbers_with_ts.number, date_bin('5 minutes', numbers_with_ts.ts) AS time_window FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP))) GROUP BY date_bin('5 minutes', numbers_with_ts.ts), numbers_with_ts.number)"
+            ),
+            // cte
+            (
+                "with cte as (select number, date_bin('5 minutes', ts) as time_window from numbers_with_ts GROUP BY time_window, number) select number, time_window from cte;",
+                Timestamp::new(23, TimeUnit::Millisecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT cte.number, cte.time_window FROM (SELECT numbers_with_ts.number, date_bin('5 minutes', numbers_with_ts.ts) AS time_window FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP))) GROUP BY date_bin('5 minutes', numbers_with_ts.ts), numbers_with_ts.number) AS cte"
+            ),
+            // complex subquery without alias
+            (
+                "SELECT sum(number), number, date_bin('5 minutes', ts) as time_window, bucket_name FROM (SELECT number, ts, case when number < 5 THEN 'bucket_0_5' when number >= 5 THEN 'bucket_5_inf' END as bucket_name FROM numbers_with_ts) GROUP BY number, time_window, bucket_name;",
+                Timestamp::new(23, TimeUnit::Millisecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT sum(numbers_with_ts.number), numbers_with_ts.number, date_bin('5 minutes', numbers_with_ts.ts) AS time_window, bucket_name FROM (SELECT numbers_with_ts.number, numbers_with_ts.ts, CASE WHEN (numbers_with_ts.number < 5) THEN 'bucket_0_5' WHEN (numbers_with_ts.number >= 5) THEN 'bucket_5_inf' END AS bucket_name FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP)))) GROUP BY numbers_with_ts.number, date_bin('5 minutes', numbers_with_ts.ts), bucket_name"
+            ),
+            // complex subquery alias
+            (
+                "SELECT sum(number), number, date_bin('5 minutes', ts) as time_window, bucket_name FROM (SELECT number, ts, case when number < 5 THEN 'bucket_0_5' when number >= 5 THEN 'bucket_5_inf' END as bucket_name FROM numbers_with_ts) as cte GROUP BY number, time_window, bucket_name;",
+                Timestamp::new(23, TimeUnit::Millisecond),
+                (
+                    "ts".to_string(),
+                    Some(Timestamp::new(0, TimeUnit::Millisecond)),
+                    Some(Timestamp::new(300000, TimeUnit::Millisecond)),
+                ),
+                "SELECT sum(cte.number), cte.number, date_bin('5 minutes', cte.ts) AS time_window, cte.bucket_name FROM (SELECT numbers_with_ts.number, numbers_with_ts.ts, CASE WHEN (numbers_with_ts.number < 5) THEN 'bucket_0_5' WHEN (numbers_with_ts.number >= 5) THEN 'bucket_5_inf' END AS bucket_name FROM numbers_with_ts WHERE ((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts <= CAST('1970-01-01 00:05:00' AS TIMESTAMP)))) AS cte GROUP BY cte.number, date_bin('5 minutes', cte.ts), cte.bucket_name"
+            ),
+        ];
+
+        for (sql, current, expected, expected_unparsed) in testcases {
+            let plan = sql_to_df_plan(ctx.clone(), query_engine.clone(), sql, true)
+                .await
+                .unwrap();
+
+            let real =
+                find_plan_time_window_bound(&plan, current, ctx.clone(), query_engine.clone())
+                    .await
+                    .unwrap();
+            assert_eq!(expected, real);
+
+            let plan = sql_to_df_plan(ctx.clone(), query_engine.clone(), sql, false)
+                .await
+                .unwrap();
+            let (col_name, lower, upper) = real;
+            let new_sql = if lower.is_some() {
+                let to_df_literal = |value| {
+                    let value = Value::from(value);
+
+                    value.try_to_scalar_value(&value.data_type()).unwrap()
+                };
+                let lower = to_df_literal(lower.unwrap());
+                let upper = to_df_literal(upper.unwrap());
+                let expr = col(&col_name)
+                    .gt_eq(lit(lower))
+                    .and(col(&col_name).lt_eq(lit(upper)));
+                let mut add_filter = AddFilterRewriter::new(expr);
+                let plan = plan.rewrite(&mut add_filter).unwrap().data;
+                df_plan_to_sql(&plan).unwrap()
+            } else {
+                sql.to_string()
+            };
+            assert_eq!(expected_unparsed, new_sql);
+        }
+    }
+}
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
discord9	7f7b974e8a	fix: conn pool leak & placeholder feature so ci can compile	2025-04-10 15:01:07 +08:00
discord9	4875ace0d0	fix: placeholder feature so ci can compile	2025-04-08 14:37:55 +08:00
discord9	a847d96649	fix: time window filter expr use OR	2025-04-07 16:50:17 +08:00
discord9	23a0a54e18	fix: convert timestamp unit too	2025-04-07 16:50:17 +08:00
discord9	78eb8b53f6	fix: quote&more info when time window too many chore: even more warning fix: filter first warn later	2025-04-07 16:50:17 +08:00
discord9	2455f39e8e	fix: subquery&cte time window expr	2025-04-07 16:46:53 +08:00
discord9	7fe0074202	refactor: even finer&limit time window num	2025-04-07 16:46:53 +08:00
discord9	e16bc203d0	feat: basic time window aware	2025-04-07 16:46:53 +08:00
discord9	9a3c26bb0a	metrics: better bucket&longer timeout	2025-04-07 16:46:53 +08:00
discord9	e1ff398c32	fix: timeout	2025-04-07 16:46:53 +08:00
discord9	780e3000de	fix: heartbeat&expire_after unit	2025-04-07 16:46:53 +08:00
discord9	2b5ddf8427	feat: time window in df plan WIP test: found out time window expr chore: pub tests: also unparsed tests: rm dup code feat: frontend client for recording rule fix: bound edgecase WIP WIP feat: rule engine feat: add init options& tmp rerounte to rule fix: dist client get fix: also not handle mirror write in flownode chore: clippy	2025-04-07 16:46:47 +08:00